Something went wrong. Try again.
A Material 3 inspired desktop shell for Niri and Hyprland created with Ignis.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150from __future__ import annotations
import loggingfrom collections.abc import Callable
from gi.repository import Gio, GLib
from .model import PairingRequest, RequestKind
log = logging.getLogger("exo.bluetooth")
AGENT_PATH = "/sh/debuggyo/Exo/bluez/agent"AGENT_INTERFACE = "org.bluez.Agent1"CAPABILITY = "KeyboardDisplay"REJECTED = "org.bluez.Error.Rejected"CANCELED = "org.bluez.Error.Canceled"
AGENT_XML = f"""<node> <interface name="{AGENT_INTERFACE}"> <method name="Release" /> <method name="RequestPinCode"> <arg name="device" type="o" direction="in" /> <arg name="pincode" type="s" direction="out" /> </method> <method name="DisplayPinCode"> <arg name="device" type="o" direction="in" /> <arg name="pincode" type="s" direction="in" /> </method> <method name="RequestPasskey"> <arg name="device" type="o" direction="in" /> <arg name="passkey" type="u" direction="out" /> </method> <method name="DisplayPasskey"> <arg name="device" type="o" direction="in" /> <arg name="passkey" type="u" direction="in" /> <arg name="entered" type="q" direction="in" /> </method> <method name="RequestConfirmation"> <arg name="device" type="o" direction="in" /> <arg name="passkey" type="u" direction="in" /> </method> <method name="RequestAuthorization"> <arg name="device" type="o" direction="in" /> </method> <method name="AuthorizeService"> <arg name="device" type="o" direction="in" /> <arg name="uuid" type="s" direction="in" /> </method> <method name="Cancel" /> </interface></node>"""
class PairingAgent: def __init__( self, bus: Gio.DBusConnection, device_name: Callable[[str], str], on_change: Callable[[], None], ): self._device_name = device_name self._on_change = on_change self._pending: Gio.DBusMethodInvocation | None = None self._next_id = 0 self.request: PairingRequest | None = None
node = Gio.DBusNodeInfo.new_for_xml(AGENT_XML) bus.register_object( AGENT_PATH, node.lookup_interface(AGENT_INTERFACE), self._on_call, None, None )
def _raise(self, kind: RequestKind, device: str, **fields) -> PairingRequest: self._next_id += 1 self.request = PairingRequest( self._next_id, kind, device, self._device_name(device), **fields ) self._on_change() return self.request
def _on_call(self, _bus, _sender, _path, _interface, method, params, invocation): if method == "Release": self.clear() invocation.return_value(None) return if method == "Cancel": self.clear() invocation.return_value(None) return
if method == "DisplayPasskey": device, passkey, entered = params.unpack() self._raise( RequestKind.DISPLAY_PASSKEY, device, code=f"{passkey:06d}", entered=entered ) invocation.return_value(None) return if method == "DisplayPinCode": device, pin = params.unpack() self._raise(RequestKind.DISPLAY_PIN, device, code=pin) invocation.return_value(None) return
if self._pending is not None: invocation.return_dbus_error(REJECTED, "another request is in progress") return
device = params.unpack()[0] self._pending = invocation if method == "RequestPinCode": self._raise(RequestKind.ENTER_PIN, device) elif method == "RequestPasskey": self._raise(RequestKind.ENTER_PASSKEY, device) elif method == "RequestConfirmation": self._raise(RequestKind.CONFIRM_PASSKEY, device, code=f"{params.unpack()[1]:06d}") elif method == "RequestAuthorization": self._raise(RequestKind.AUTHORIZE_PAIRING, device) elif method == "AuthorizeService": self._raise(RequestKind.AUTHORIZE_SERVICE, device, service_uuid=params.unpack()[1]) else: self._pending = None invocation.return_dbus_error(REJECTED, f"unsupported method {method}")
def respond(self, request_id: int, accepted: bool, text: str = "") -> None: request = self.request invocation = self._pending if request is None or request.request_id != request_id or invocation is None: return self._pending = None self.request = None self._on_change()
if not accepted: invocation.return_dbus_error(REJECTED, "rejected by the user") elif request.kind is RequestKind.ENTER_PIN: invocation.return_value(GLib.Variant("(s)", (text,))) elif request.kind is RequestKind.ENTER_PASSKEY: invocation.return_value(GLib.Variant("(u)", (int(text),))) else: invocation.return_value(None)
def clear(self) -> None: invocation, self._pending = self._pending, None had_request, self.request = self.request is not None, None if invocation is not None: invocation.return_dbus_error(CANCELED, "request cancelled") if had_request: self._on_change()