fxroute/bluez_audio_agent.py at main · CobbyCode/fxroute · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
#!/usr/bin/env python3
# SPDX-License-Identifier: AGPL-3.0-only

"""Minimal BlueZ audio agent for headless/no-input Bluetooth audio pairing."""

from __future__ import annotations

import signal
import sys

import dbus
import dbus.mainloop.glib
import dbus.service
from gi.repository import GLib

BUS_NAME = "org.bluez"
AGENT_INTERFACE = "org.bluez.Agent1"
AGENT_MANAGER_INTERFACE = "org.bluez.AgentManager1"
PROPERTIES_INTERFACE = "org.freedesktop.DBus.Properties"
AGENT_PATH = "/fxroute/agent"
CAPABILITY = "DisplayYesNo"

AUDIO_UUID_PREFIXES = {
    "0000110a",  # Audio Source
    "0000110b",  # Audio Sink
    "0000110c",  # AVRCP Target
    "0000110e",  # AVRCP Controller
    "0000111e",  # Handsfree
    "0000111f",  # Handsfree Audio Gateway
}


class Rejected(dbus.DBusException):
    _dbus_error_name = "org.bluez.Error.Rejected"


class Agent(dbus.service.Object):
    @dbus.service.method(AGENT_INTERFACE, in_signature="", out_signature="")
    def Release(self):
        mainloop.quit()

    @dbus.service.method(AGENT_INTERFACE, in_signature="o", out_signature="s")
    def RequestPinCode(self, device):
        return "0000"

    @dbus.service.method(AGENT_INTERFACE, in_signature="o", out_signature="u")
    def RequestPasskey(self, device):
        return dbus.UInt32(0)

    @dbus.service.method(AGENT_INTERFACE, in_signature="ou", out_signature="")
    def DisplayPasskey(self, device, passkey):
        return

    @dbus.service.method(AGENT_INTERFACE, in_signature="os", out_signature="")
    def DisplayPinCode(self, device, pincode):
        return

    @dbus.service.method(AGENT_INTERFACE, in_signature="ou", out_signature="")
    def RequestConfirmation(self, device, passkey):
        return

    @dbus.service.method(AGENT_INTERFACE, in_signature="o", out_signature="")
    def RequestAuthorization(self, device):
        return

    @dbus.service.method(AGENT_INTERFACE, in_signature="os", out_signature="")
    def AuthorizeService(self, device, uuid):
        normalized = str(uuid).strip().lower()
        if any(normalized.startswith(prefix) for prefix in AUDIO_UUID_PREFIXES):
            return
        # Be permissive for already-paired device services while agent is active.
        return

    @dbus.service.method(AGENT_INTERFACE, in_signature="", out_signature="")
    def Cancel(self):
        return


mainloop: GLib.MainLoop


def _quit(*_args):
    try:
        mainloop.quit()
    except Exception:
        pass


def main() -> int:
    global mainloop
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    bus = dbus.SystemBus()

    agent = Agent(bus, AGENT_PATH)
    manager = dbus.Interface(bus.get_object(BUS_NAME, "/org/bluez"), AGENT_MANAGER_INTERFACE)
    manager.RegisterAgent(AGENT_PATH, CAPABILITY)
    manager.RequestDefaultAgent(AGENT_PATH)

    for sig in (signal.SIGTERM, signal.SIGINT):
        signal.signal(sig, _quit)

    mainloop = GLib.MainLoop()
    try:
        mainloop.run()
    finally:
        try:
            manager.UnregisterAgent(AGENT_PATH)
        except Exception:
            pass
        del agent
    return 0


if __name__ == "__main__":
    sys.exit(main())