terok_dbus

terok_dbus

D-Bus desktop notification package for terok.

CLEARANCE_BUS_NAME = 'org.terok.Clearance' module-attribute

Well-known bus name for the clearance daemon.

CLEARANCE_INTERFACE_NAME = 'org.terok.Clearance1' module-attribute

Versioned interface name for clearance signals and methods.

CLEARANCE_OBJECT_PATH = '/org/terok/Clearance' module-attribute

Object path for the Clearance1 interface.

CLEARANCE_XML = '<node>\n <interface name="org.terok.Clearance1">\n <signal name="RequestReceived">\n <arg type="s" name="request_id" direction="out"/>\n <arg type="s" name="project" direction="out"/>\n <arg type="s" name="task" direction="out"/>\n <arg type="s" name="dest" direction="out"/>\n <arg type="q" name="port" direction="out"/>\n <arg type="s" name="reason" direction="out"/>\n </signal>\n <method name="Resolve">\n <arg type="s" name="request_id" direction="in"/>\n <arg type="s" name="action" direction="in"/>\n <arg type="b" name="ok" direction="out"/>\n </method>\n <method name="ListPending">\n <arg type="a(ssssqs)" name="requests" direction="out"/>\n </method>\n <signal name="RequestResolved">\n <arg type="s" name="request_id" direction="out"/>\n <arg type="s" name="action" direction="out"/>\n <arg type="as" name="ips" direction="out"/>\n </signal>\n </interface>\n</node>' module-attribute

Introspection XML for org.terok.Clearance1.

SHIELD_BUS_NAME = 'org.terok.Shield' module-attribute

Well-known bus name for the shield D-Bus bridge.

SHIELD_INTERFACE_NAME = 'org.terok.Shield1' module-attribute

Versioned interface name for shield signals and methods.

SHIELD_OBJECT_PATH = '/org/terok/Shield' module-attribute

Object path for the Shield1 interface.

SHIELD_XML = '<node>\n <interface name="org.terok.Shield1">\n <signal name="ConnectionBlocked">\n <arg type="s" name="container" direction="out"/>\n <arg type="s" name="dest" direction="out"/>\n <arg type="q" name="port" direction="out"/>\n <arg type="q" name="proto" direction="out"/>\n <arg type="s" name="domain" direction="out"/>\n <arg type="s" name="request_id" direction="out"/>\n </signal>\n <method name="Verdict">\n <arg type="s" name="request_id" direction="in"/>\n <arg type="s" name="action" direction="in"/>\n <arg type="b" name="ok" direction="out"/>\n </method>\n <signal name="VerdictApplied">\n <arg type="s" name="container" direction="out"/>\n <arg type="s" name="dest" direction="out"/>\n <arg type="s" name="request_id" direction="out"/>\n <arg type="s" name="action" direction="out"/>\n <arg type="b" name="ok" direction="out"/>\n </signal>\n </interface>\n</node>' module-attribute

Introspection XML for org.terok.Shield1.

DbusNotifier(app_name='terok')

Send desktop notifications over the D-Bus session bus.

The connection is established lazily on the first notify call. Action callbacks are dispatched from the ActionInvoked signal; stale callbacks are cleaned up automatically on NotificationClosed.

Parameters:

Name	Type	Description	Default
app_name	str	Application name sent with every notification.	'terok'

Initialise with the given application name.

Source code in src/terok_dbus/_notifier.py

def __init__(self, app_name: str = "terok") -> None:
    """Initialise with the given application name."""
    self._app_name = app_name
    self._bus: MessageBus | None = None
    self._interface: object | None = None
    self._callbacks: dict[int, Callable[[str], None]] = {}
    self._connect_lock = asyncio.Lock()

notify(summary, body='', *, actions=(), timeout_ms=-1, hints=None, replaces_id=0, app_icon='') async

Send a desktop notification.

Parameters:

Name	Type	Description	Default
summary	str	Notification title.	required
body	str	Optional body text.	''
actions	Sequence[tuple[str, str]]	(action_id, label) pairs rendered as buttons.	()
timeout_ms	int	Expiration hint in milliseconds (-1 = server default).	-1
hints	Mapping[str, Any] | None	Freedesktop hint dict (values should be dbus_fast.Variant).	None
replaces_id	int	Replace an existing notification in-place.	0
app_icon	str	Icon name or file:/// URI.	''

Returns:

Type	Description
int	Server-assigned notification ID.

Source code in src/terok_dbus/_notifier.py

async def notify(
    self,
    summary: str,
    body: str = "",
    *,
    actions: Sequence[tuple[str, str]] = (),
    timeout_ms: int = -1,
    hints: Mapping[str, Any] | None = None,
    replaces_id: int = 0,
    app_icon: str = "",
) -> int:
    """Send a desktop notification.

    Args:
        summary: Notification title.
        body: Optional body text.
        actions: ``(action_id, label)`` pairs rendered as buttons.
        timeout_ms: Expiration hint in milliseconds (``-1`` = server default).
        hints: Freedesktop hint dict (values should be ``dbus_fast.Variant``).
        replaces_id: Replace an existing notification in-place.
        app_icon: Icon name or ``file:///`` URI.

    Returns:
        Server-assigned notification ID.
    """
    if self._interface is None:
        async with self._connect_lock:
            if self._interface is None:
                await self._connect()

    actions_flat: list[str] = []
    for action_id, label in actions:
        actions_flat.extend((action_id, label))

    return await self._interface.call_notify(  # type: ignore[union-attr]
        self._app_name,
        replaces_id,
        app_icon,
        summary,
        body,
        actions_flat,
        dict(hints) if hints is not None else {},
        timeout_ms,
    )

on_action(notification_id, callback) async

Register a callback for when the user clicks an action button.

Parameters:

Name	Type	Description	Default
notification_id	int	ID returned by notify.	required
callback	Callable[[str], None]	Called with the action_id string when invoked.	required

Source code in src/terok_dbus/_notifier.py

async def on_action(
    self,
    notification_id: int,
    callback: Callable[[str], None],
) -> None:
    """Register a callback for when the user clicks an action button.

    Args:
        notification_id: ID returned by ``notify``.
        callback: Called with the ``action_id`` string when invoked.
    """
    self._callbacks[notification_id] = callback

close(notification_id) async

Close an active notification.

Parameters:

Name	Type	Description	Default
notification_id	int	ID returned by notify.	required

Source code in src/terok_dbus/_notifier.py

async def close(self, notification_id: int) -> None:
    """Close an active notification.

    Args:
        notification_id: ID returned by ``notify``.
    """
    self._callbacks.pop(notification_id, None)
    if self._interface is not None:
        await self._interface.call_close_notification(notification_id)  # type: ignore[union-attr]

disconnect() async

Tear down the session-bus connection.

Source code in src/terok_dbus/_notifier.py

async def disconnect(self) -> None:
    """Tear down the session-bus connection."""
    if self._interface is not None:
        if hasattr(self._interface, "off_action_invoked"):
            self._interface.off_action_invoked(self._handle_action)
        if hasattr(self._interface, "off_notification_closed"):
            self._interface.off_notification_closed(self._handle_closed)
    if self._bus is not None:
        self._bus.disconnect()
    self._bus = None
    self._interface = None
    self._callbacks.clear()

NullNotifier

Silent fallback that satisfies the Notifier protocol.

Every method is a no-op. notify always returns 0.

notify(summary, body='', *, actions=(), timeout_ms=-1, hints=None, replaces_id=0, app_icon='') async

Accept and discard a notification, returning 0.

Source code in src/terok_dbus/_null.py

async def notify(
    self,
    summary: str,
    body: str = "",
    *,
    actions: Sequence[tuple[str, str]] = (),
    timeout_ms: int = -1,
    hints: Mapping[str, Any] | None = None,
    replaces_id: int = 0,
    app_icon: str = "",
) -> int:
    """Accept and discard a notification, returning ``0``."""
    return 0

on_action(notification_id, callback) async

Accept and discard an action callback registration.

Source code in src/terok_dbus/_null.py

async def on_action(
    self,
    notification_id: int,
    callback: Callable[[str], None],
) -> None:
    """Accept and discard an action callback registration."""

close(notification_id) async

Accept and discard a close request.

Source code in src/terok_dbus/_null.py

async def close(self, notification_id: int) -> None:
    """Accept and discard a close request."""

disconnect() async

Accept and discard a teardown request.

Source code in src/terok_dbus/_null.py

async def disconnect(self) -> None:
    """Accept and discard a teardown request."""

Notifier

Bases: Protocol

Structural type for desktop notification backends.

Implementations must provide notify, on_action, close, and disconnect. DbusNotifier talks to a real session bus; NullNotifier silently discards everything for headless environments.

notify(summary, body='', *, actions=(), timeout_ms=-1, hints=None, replaces_id=0, app_icon='') async

Send a desktop notification.

Parameters:

Name	Type	Description	Default
summary	str	Notification title.	required
body	str	Optional body text.	''
actions	Sequence[tuple[str, str]]	(action_id, label) pairs rendered as buttons.	()
timeout_ms	int	Expiration hint in milliseconds (-1 = server default).	-1
hints	Mapping[str, Any] | None	Freedesktop hint dict (values are dbus_fast.Variant for DbusNotifier, ignored by NullNotifier).	None
replaces_id	int	Replace an existing notification in-place.	0
app_icon	str	Icon name or file:/// URI.	''

Returns:

Type	Description
int	Server-assigned notification ID (0 for null implementations).

Source code in src/terok_dbus/_protocol.py

async def notify(
    self,
    summary: str,
    body: str = "",
    *,
    actions: Sequence[tuple[str, str]] = (),
    timeout_ms: int = -1,
    hints: Mapping[str, Any] | None = None,
    replaces_id: int = 0,
    app_icon: str = "",
) -> int:
    """Send a desktop notification.

    Args:
        summary: Notification title.
        body: Optional body text.
        actions: ``(action_id, label)`` pairs rendered as buttons.
        timeout_ms: Expiration hint in milliseconds (``-1`` = server default).
        hints: Freedesktop hint dict (values are ``dbus_fast.Variant`` for
            ``DbusNotifier``, ignored by ``NullNotifier``).
        replaces_id: Replace an existing notification in-place.
        app_icon: Icon name or ``file:///`` URI.

    Returns:
        Server-assigned notification ID (``0`` for null implementations).
    """
    ...

on_action(notification_id, callback) async

Register a callback for when the user clicks an action button.

Parameters:

Name	Type	Description	Default
notification_id	int	ID returned by notify.	required
callback	Callable[[str], None]	Called with the action_id string when invoked.	required

Source code in src/terok_dbus/_protocol.py

async def on_action(
    self,
    notification_id: int,
    callback: Callable[[str], None],
) -> None:
    """Register a callback for when the user clicks an action button.

    Args:
        notification_id: ID returned by ``notify``.
        callback: Called with the ``action_id`` string when invoked.
    """
    ...

close(notification_id) async

Close an active notification.

Parameters:

Name	Type	Description	Default
notification_id	int	ID returned by notify.	required

Source code in src/terok_dbus/_protocol.py

async def close(self, notification_id: int) -> None:
    """Close an active notification.

    Args:
        notification_id: ID returned by ``notify``.
    """
    ...

disconnect() async

Release backend resources (no-op for null backends).

Source code in src/terok_dbus/_protocol.py

async def disconnect(self) -> None:
    """Release backend resources (no-op for null backends)."""
    ...

EventSubscriber(notifier, bus=None)

Subscribe to Shield1 and Clearance1 D-Bus signals and present desktop notifications.

Creates desktop notifications with Allow/Deny action buttons for blocked connections (Shield) and clearance requests (Clearance). Operator actions are routed back as Verdict / Resolve D-Bus method calls.

Parameters:

Name	Type	Description	Default
notifier	Notifier	Desktop notification backend.	required
bus	MessageBus | None	Optional pre-connected MessageBus (for testing). If None, a new session-bus connection is created on start().	None

Initialise the subscriber with a notifier and optional bus.

Source code in src/terok_dbus/_subscriber.py

def __init__(self, notifier: Notifier, bus: MessageBus | None = None) -> None:
    """Initialise the subscriber with a notifier and optional bus."""
    self._notifier = notifier
    self._bus = bus
    self._owns_bus = bus is None
    self._shield_iface: Any | None = None
    self._clearance_iface: Any | None = None
    self._pending: dict[int, str] = {}  # notification_id → request_id
    self._tasks: set[asyncio.Task[None]] = set()

start() async

Connect to the session bus and subscribe to Shield1 and Clearance1 signals.

Source code in src/terok_dbus/_subscriber.py

async def start(self) -> None:
    """Connect to the session bus and subscribe to Shield1 and Clearance1 signals."""
    if self._bus is None:
        self._bus = await MessageBus().connect()

    shield_node = Node.parse(SHIELD_XML)
    shield_proxy = self._bus.get_proxy_object(SHIELD_BUS_NAME, SHIELD_OBJECT_PATH, shield_node)
    self._shield_iface = shield_proxy.get_interface(SHIELD_INTERFACE_NAME)
    self._shield_iface.on_connection_blocked(self._on_connection_blocked)
    self._shield_iface.on_verdict_applied(self._on_verdict_applied)
    _log.info("Subscribed to %s", SHIELD_INTERFACE_NAME)

    clearance_node = Node.parse(CLEARANCE_XML)
    clearance_proxy = self._bus.get_proxy_object(
        CLEARANCE_BUS_NAME, CLEARANCE_OBJECT_PATH, clearance_node
    )
    self._clearance_iface = clearance_proxy.get_interface(CLEARANCE_INTERFACE_NAME)
    self._clearance_iface.on_request_received(self._on_request_received)
    self._clearance_iface.on_request_resolved(self._on_request_resolved)
    _log.info("Subscribed to %s", CLEARANCE_INTERFACE_NAME)

stop() async

Unsubscribe from signals and disconnect the bus if owned.

Source code in src/terok_dbus/_subscriber.py

async def stop(self) -> None:
    """Unsubscribe from signals and disconnect the bus if owned."""
    for task in self._tasks:
        task.cancel()
    await asyncio.sleep(0)  # yield to let cancellations propagate
    self._tasks.clear()

    if self._shield_iface is not None:
        if hasattr(self._shield_iface, "off_connection_blocked"):
            self._shield_iface.off_connection_blocked(self._on_connection_blocked)
        if hasattr(self._shield_iface, "off_verdict_applied"):
            self._shield_iface.off_verdict_applied(self._on_verdict_applied)
    if self._clearance_iface is not None:
        if hasattr(self._clearance_iface, "off_request_received"):
            self._clearance_iface.off_request_received(self._on_request_received)
        if hasattr(self._clearance_iface, "off_request_resolved"):
            self._clearance_iface.off_request_resolved(self._on_request_resolved)

    if self._owns_bus and self._bus is not None:
        self._bus.disconnect()

    self._shield_iface = None
    self._clearance_iface = None
    self._pending.clear()

create_notifier(app_name='terok') async

Return a connected DbusNotifier, or a NullNotifier on failure.

This is the primary entry point. Callers get a working notifier without caring whether a D-Bus session bus is available.

Parameters:

Name	Type	Description	Default
app_name	str	Application name sent with every notification.	'terok'

Returns:

Type	Description
Notifier	A Notifier-compatible instance.

Source code in src/terok_dbus/__init__.py

async def create_notifier(app_name: str = "terok") -> Notifier:
    """Return a connected ``DbusNotifier``, or a ``NullNotifier`` on failure.

    This is the primary entry point. Callers get a working notifier without
    caring whether a D-Bus session bus is available.

    Args:
        app_name: Application name sent with every notification.

    Returns:
        A ``Notifier``-compatible instance.
    """
    notifier = DbusNotifier(app_name)
    try:
        await notifier._connect()
    except (OSError, DBusError, ValueError) as exc:
        _log.debug("D-Bus session bus unavailable, falling back to NullNotifier: %s", exc)
        return NullNotifier()
    return notifier