terok_clearance

`terok_clearance` ¶

Clearance hub + desktop notification library for terok.

The operator-UI plane for terok-shield: turns shield's blocked- connection events into Allow/Deny prompts and routes the operator's verdict back to shield for enforcement. Two axes of pluggability apply:

Producer (event source) — closed. Shield is the only producer. The wire vocabulary (shield_up, connection_blocked, …) names shield's state machine, and the verdict path execs terok-shield allow|deny. A non-shield "clearance" wouldn't work end-to-end; the package is shield's UI plane, not a generic firewall console.
Operator UI (consumer) — open. Anything that subscribes to the hub's varlink stream and implements the Notifier protocol on the verdict-routing side is a valid UI: today the D-Bus desktop notifier (DbusNotifier), the standalone Textual terok clearance app, and the embedded terok-tui screen all ride on this seam.

Container-runtime inspection is no longer a clearance concern: the shield reader resolves the orchestrator-supplied dossier at emit time and ships it on the wire (ClearanceEvent.dossier), so clearance has no Python-level coupling to any runtime.

Two unrelated wire formats live under this one package as a result:

org.terok.Clearance1 over a unix-socket varlink transport — the hub (ClearanceHub) and the client library (ClearanceClient, EventSubscriber) that drive the per-container block / verdict / lifecycle flow.
org.freedesktop.Notifications over D-Bus — the DbusNotifier wrapper that renders those events as desktop popups. Kept because that's the OS API; the in-package transport is varlink.

The supervisor (in terok-sandbox) composes one ClearanceHub and one VerdictServer in-process per container. Each container has its own hub socket, so operator UIs multiplex across the per-container sockets via MultiSocketSubscriber.

`all = ['ALL_NOTIFY_CATEGORIES', 'COMMANDS', 'CallbackNotifier', 'ClearanceClient', 'ClearanceEvent', 'ClearanceHub', 'EventSubscriber', 'MultiSocketSubscriber', 'NOTIFY_BLOCKED', 'NOTIFY_VERDICT', 'Notification', 'VerdictClient', 'VerdictServer', 'create_notifier', 'default_clearance_socket_path']` `module-attribute` ¶

`ALL_NOTIFY_CATEGORIES = frozenset({NOTIFY_BLOCKED, NOTIFY_VERDICT, NOTIFY_SHIELD_UP, NOTIFY_SHIELD_DOWN, NOTIFY_CONTAINER_STARTED, NOTIFY_CONTAINER_EXITED})` `module-attribute` ¶

Every category the subscriber recognises — the default opt-in set.

`NOTIFY_BLOCKED = 'blocked'` `module-attribute` ¶

Category for connection_blocked Allow/Deny prompts (interactive).

`NOTIFY_VERDICT = 'verdict'` `module-attribute` ¶

Category for verdict_applied resolution popups (ack of a verdict).

COMMANDS = (CommandDef(name='notify', help='Send a one-shot desktop notification', source='terok_clearance.cli.verbs.notify:COMMAND'), CommandDef(name='serve', help='Run the clearance hub (serves org.terok.Clearance1 varlink on a unix socket)', source='terok_clearance.cli.verbs.serve:COMMAND'), CommandDef(name='serve-verdict', help='Run the verdict helper (serves org.terok.ClearanceVerdict1 for shield exec)', source='terok_clearance.cli.verbs.serve_verdict:COMMAND'), CommandDef(name='clearance', help='Interactive terminal tool for shield clearance verdicts', source='terok_clearance.cli.verbs.clearance:COMMAND')) `module-attribute` ¶

`version = '0.0.0'` `module-attribute` ¶

`ClearanceClient(*, socket_path=None)` ¶

Thin async client for the Clearance1 varlink service.

Two async coroutines to drive:

start — open the subscribe + RPC connections and begin relaying events to the user-supplied callback. Returns once both channels are live; events arrive via on_event from then on.
verdict — RPC call; returns True if terok-shield applied the action, False on any refusal or shield failure. The refusal reason is logged at WARNING.

The callback runs on the same event loop as the rest of the client; exceptions it raises are logged and swallowed so one bad handler can't kill the stream for every subsequent event.

Remember the target socket; defaults to default_clearance_socket_path.

Source code in src/terok_clearance/client/client.py

def __init__(self, *, socket_path: Path | None = None) -> None:
    """Remember the target socket; defaults to [`default_clearance_socket_path`][terok_clearance.client.client.default_clearance_socket_path]."""
    self._socket_path = socket_path or default_clearance_socket_path()
    self._on_event: EventCallback | None = None
    self._sub_transport: VarlinkTransport | None = None
    self._rpc_transport: VarlinkTransport | None = None
    self._sub_proxy: VarlinkInterfaceProxy | None = None
    self._rpc_proxy: VarlinkInterfaceProxy | None = None
    self._stream_task: asyncio.Task[None] | None = None
    self._stopping = False
    # Set by [`poke_reconnect`][terok_clearance.client.client.ClearanceClient.poke_reconnect]; awaited inside the back-off
    # window.  Constructed here (not lazily) so a focus-gain poke
    # that lands between ``start()`` and the first ``_run_stream``
    # iteration isn't silently dropped.
    self._reconnect_poke = asyncio.Event()

`start(on_event)` `async` ¶

Open both connections and begin relaying events to on_event.

The initial connect is awaited synchronously so callers see start() return only after the subscription is live — a hub that's down at startup still propagates as an exception. Subsequent drops are handled by _run_stream's internal reconnect loop so long-running consumers (TUI, notifier) survive a systemctl restart terok-clearance without restarting themselves.

Source code in src/terok_clearance/client/client.py

async def start(self, on_event: EventCallback) -> None:
    """Open both connections and begin relaying events to *on_event*.

    The initial connect is awaited synchronously so callers see
    ``start()`` return only after the subscription is live — a
    hub that's down at startup still propagates as an exception.
    Subsequent drops are handled by `_run_stream`'s internal
    reconnect loop so long-running consumers (TUI, notifier)
    survive a ``systemctl restart terok-clearance`` without
    restarting themselves.
    """
    self._on_event = on_event
    self._stopping = False
    await self._connect()
    self._stream_task = asyncio.create_task(self._run_stream())

`stop()` `async` ¶

Close both connections and await the stream task.

Source code in src/terok_clearance/client/client.py

async def stop(self) -> None:
    """Close both connections and await the stream task."""
    self._stopping = True
    if self._stream_task is not None:
        self._stream_task.cancel()
        with contextlib.suppress(asyncio.CancelledError, Exception):
            await self._stream_task
        self._stream_task = None
    self._close_transports()

`poke_reconnect()` ¶

Skip any in-flight reconnect back-off and retry immediately.

Idempotent; a no-op when the stream is healthy because the event is only awaited inside _run_stream's back-off window.

Source code in src/terok_clearance/client/client.py

def poke_reconnect(self) -> None:
    """Skip any in-flight reconnect back-off and retry immediately.

    Idempotent; a no-op when the stream is healthy because the
    event is only awaited inside `_run_stream`'s back-off
    window.
    """
    self._reconnect_poke.set()

`verdict(container, request_id, dest, action)` `async` ¶

Apply action (allow / deny) to dest via the hub's Verdict RPC.

Returns True when the hub accepted and applied the verdict, False for any refusal (unknown request_id, tuple mismatch, invalid action, shield-exec failure). Callers typically ignore the return value and let the subsequent verdict_applied event drive UI updates; refusal reasons are logged at WARNING.

Source code in src/terok_clearance/client/client.py

async def verdict(self, container: str, request_id: str, dest: str, action: str) -> bool:
    """Apply *action* (``allow`` / ``deny``) to *dest* via the hub's ``Verdict`` RPC.

    Returns ``True`` when the hub accepted and applied the verdict,
    ``False`` for any refusal (unknown request_id, tuple mismatch,
    invalid action, shield-exec failure).  Callers typically ignore
    the return value and let the subsequent ``verdict_applied``
    event drive UI updates; refusal reasons are logged at WARNING.
    """
    if self._rpc_proxy is None:
        _log.error("verdict() called before start()")
        return False
    try:
        reply = await self._rpc_proxy.Verdict(
            container=container,
            request_id=request_id,
            dest=dest,
            action=action,
        )
    except VarlinkErrorReply as err:
        _log.warning(
            "Verdict refused for %s (%s → %s): %s",
            container,
            request_id,
            action,
            err,
        )
        return False
    # reply is {"ok": bool} per the return_parameter wrapper.
    return bool(reply.get("ok", False))

`EventSubscriber(notifier, client=None, *, socket_path=None, enabled_categories=None)` ¶

Bridge clearance-hub events into desktop notifications.

Owns the presentation-layer state a rendering client needs: live-block dedup keyed on (container, target), the tracked ShieldDown popup per container so ShieldUp can retire it, and verdict routing through notifier action callbacks.

Parameters:

Name	Type	Description	Default
`notifier`	`Notifier`	Desktop notification backend (any `Notifier` works).	required
`client`	`ClearanceClient \| None`	Pre-configured `ClearanceClient`. When omitted, one is created on `start` pointing at socket_path (defaulting to `default_clearance_socket_path`).	`None`
`socket_path`	`Path \| None`	Clearance-socket override when client isn't supplied (tests).	`None`
`enabled_categories`	`Set[str] \| None`	Subset of `ALL_NOTIFY_CATEGORIES` whose popups should reach the notifier. `None` (the default) enables every category and matches the historical "render everything" behaviour — appropriate for the in-TUI subscriber that builds its own UI on top of the full event stream. The desktop notifier daemon narrows this to the interactive subset `{blocked, verdict}` so passive shield/container popups don't overrun the operator's tray. Lifecycle hook dispatch (`_dispatch_lifecycle`) and pending-state cleanup remain unconditional — the gate fires only on the `notify()` call itself, so an embedded consumer's lifecycle callbacks keep firing regardless of which categories are silenced.	`None`

Initialise the subscriber with a notifier and transport.

Source code in src/terok_clearance/client/subscriber.py

def __init__(
    self,
    notifier: Notifier,
    client: ClearanceClient | None = None,
    *,
    socket_path: Path | None = None,
    enabled_categories: AbstractSet[str] | None = None,
) -> None:
    """Initialise the subscriber with a notifier and transport."""
    self._notifier = notifier
    self._client = client or ClearanceClient(socket_path=socket_path)
    self._enabled_categories: frozenset[str] = (
        ALL_NOTIFY_CATEGORIES
        if enabled_categories is None
        else frozenset(enabled_categories) & ALL_NOTIFY_CATEGORIES
    )
    # request_id → pending block + its notification.
    self._pending: dict[str, _PendingBlock] = {}
    # container → notification_id of the active ShieldDown popup, so
    # ShieldUp can close the matching one before firing its brief
    # confirmation.  A stale "Shield DOWN" popup after shield is back
    # is a security hazard, not a benign leftover.
    self._shield_down_notifs: dict[str, int] = {}
    # Background action / lifecycle tasks we spawn.
    self._tasks: set[asyncio.Task[None]] = set()

`start()` `async` ¶

Connect to the clearance hub and begin rendering its event stream.

Source code in src/terok_clearance/client/subscriber.py

async def start(self) -> None:
    """Connect to the clearance hub and begin rendering its event stream."""
    await self._client.start(self._on_event)
    _log.info("clearance subscriber online")

`stop()` `async` ¶

Drain pending tasks and close the transport.

Closes the client first so no new handler tasks are scheduled, then awaits the currently-tracked tasks to settle (with their own CancelledError suppressed). A bare sleep(0) would yield only one loop turn — not enough for cancellation to propagate through chained awaits — and tasks.clear() on its own would drop references to tasks still writing to handles we then close.

Source code in src/terok_clearance/client/subscriber.py

async def stop(self) -> None:
    """Drain pending tasks and close the transport.

    Closes the client first so no new handler tasks are scheduled,
    then awaits the currently-tracked tasks to settle (with their
    own ``CancelledError`` suppressed).  A bare ``sleep(0)`` would
    yield only one loop turn — not enough for cancellation to
    propagate through chained awaits — and ``tasks.clear()`` on its
    own would drop references to tasks still writing to handles we
    then close.
    """
    tasks = list(self._tasks)
    for task in tasks:
        task.cancel()
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)
    self._tasks.clear()
    await self._client.stop()
    self._pending.clear()
    self._shield_down_notifs.clear()

`poke_reconnect()` ¶

Cut short any in-flight reconnect back-off — forwards to the client.

Source code in src/terok_clearance/client/subscriber.py

def poke_reconnect(self) -> None:
    """Cut short any in-flight reconnect back-off — forwards to the client."""
    self._client.poke_reconnect()

`MultiSocketSubscriber(notifier, *, socket_glob=None, enabled_categories=None, rescan_interval_s=2.0)` ¶

Subscribe to every per-container clearance hub socket under a glob.

The per-container-supervisor model gives every container its own hub socket at $XDG_RUNTIME_DIR/terok/clearance/<container_id>.sock, so an operator UI that wants the union of every supervisor's event stream must multiplex across the set. This class watches socket_glob (default $XDG_RUNTIME_DIR/terok/clearance/*.sock), opens an EventSubscriber against each matching path on start, re-scans every rescan_interval_s seconds to pick up sockets that appeared after start, and tears down child subscribers whose sockets vanish. All events fan into the same notifier.

A simple poll loop drives the rescan — no inotify dependency. The overhead is one glob.glob call per interval, which is cheap compared to the cost of carrying a fanotify watch fd for what is effectively a directory that mutates only at container start / stop.

Parameters:

Name	Type	Description	Default
`notifier`	`Notifier`	Desktop notification backend (any `Notifier` works). All child subscribers fan their events into this single notifier so the operator sees one merged stream.	required
`socket_glob`	`str \| None`	Filesystem glob of clearance hub sockets to track. `None` (the default) derives `$XDG_RUNTIME_DIR/terok/clearance/*.sock` from the current runtime dir.	`None`
`enabled_categories`	`Set[str] \| None`	Subset of `ALL_NOTIFY_CATEGORIES` propagated to every child `EventSubscriber`. `None` enables every category.	`None`
`rescan_interval_s`	`float`	Seconds between glob rescans. Must be strictly positive. The default (2.0) balances startup latency for a freshly-spawned supervisor against background polling cost when the host is idle.	`2.0`

Capture notifier + scan parameters; sockets are discovered in start.

Raises:

Type	Description
`ValueError`	If rescan_interval_s is not strictly positive. `0` would spin the rescan loop with no delay; a negative value would make `asyncio.sleep` raise.

Source code in src/terok_clearance/client/subscriber.py

def __init__(
    self,
    notifier: Notifier,
    *,
    socket_glob: str | None = None,
    enabled_categories: AbstractSet[str] | None = None,
    rescan_interval_s: float = 2.0,
) -> None:
    """Capture *notifier* + scan parameters; sockets are discovered in `start`.

    Raises:
        ValueError: If *rescan_interval_s* is not strictly positive.
            ``0`` would spin the rescan loop with no delay; a negative
            value would make ``asyncio.sleep`` raise.
    """
    if rescan_interval_s <= 0:
        raise ValueError(f"rescan_interval_s must be > 0, got {rescan_interval_s!r}")
    self._notifier = notifier
    self._socket_glob = socket_glob or _default_socket_glob()
    self._enabled_categories: frozenset[str] | None = (
        None if enabled_categories is None else frozenset(enabled_categories)
    )
    self._rescan_interval_s = rescan_interval_s
    self._subscribers: dict[str, EventSubscriber] = {}
    self._rescan_task: asyncio.Task[None] | None = None
    self._stopping = False

`start()` `async` ¶

Connect to every currently-matching socket and start the rescan loop.

Sockets that fail to connect at start are logged and skipped — a flaky supervisor doesn't stop the rest from coming online. The rescan loop runs as a background task so newly-spawned supervisors join the merged stream without a manual restart.

Idempotent: a second start() while the rescan loop is still live is a no-op (the alternative would orphan the existing task and leak a loop that keeps reconciling).

Source code in src/terok_clearance/client/subscriber.py

async def start(self) -> None:
    """Connect to every currently-matching socket and start the rescan loop.

    Sockets that fail to connect at start are logged and skipped —
    a flaky supervisor doesn't stop the rest from coming online.
    The rescan loop runs as a background task so newly-spawned
    supervisors join the merged stream without a manual restart.

    Idempotent: a second ``start()`` while the rescan loop is still
    live is a no-op (the alternative would orphan the existing task
    and leak a loop that keeps reconciling).
    """
    if self._rescan_task is not None and not self._rescan_task.done():
        return
    self._stopping = False
    await self._reconcile()
    self._rescan_task = asyncio.create_task(self._rescan_loop())
    _log.info(
        "multi-socket subscriber online (glob=%s, %d socket(s))",
        self._socket_glob,
        len(self._subscribers),
    )

`stop()` `async` ¶

Cancel the rescan loop and stop every child subscriber.

Stops are awaited in parallel because each child holds its own transports — serialising would extend shutdown latency by the sum of every disconnect timeout instead of the worst-case one.

Source code in src/terok_clearance/client/subscriber.py

async def stop(self) -> None:
    """Cancel the rescan loop and stop every child subscriber.

    Stops are awaited in parallel because each child holds its own
    transports — serialising would extend shutdown latency by the
    sum of every disconnect timeout instead of the worst-case one.
    """
    self._stopping = True
    if self._rescan_task is not None:
        self._rescan_task.cancel()
        with contextlib.suppress(asyncio.CancelledError, Exception):
            await self._rescan_task
        self._rescan_task = None
    children = list(self._subscribers.values())
    self._subscribers.clear()
    if children:
        await asyncio.gather(*(child.stop() for child in children), return_exceptions=True)

`ClearanceEvent(type, container, request_id='', dest='', port=0, proto=0, domain='', action='', ok=False, reason='', dossier=dict())` `dataclass` ¶

One event fanned out to every Subscribe() caller.

type + container are always populated; the remaining fields are filled in per-kind and default to zero-values otherwise.

Known values of type (additional fields beyond container):

connection_blocked — request_id, dest, port, proto, domain, dossier. Requires an operator verdict.
verdict_applied — request_id, action, ok.
container_started — dossier.
container_exited — reason, dossier.
shield_up / shield_down / shield_disengaged — dossier.

Unknown values are forwarded unchanged so the wire format can grow without breaking clients pinned to older schemas.

`type` `instance-attribute` ¶

`container` `instance-attribute` ¶

`request_id = ''` `class-attribute` `instance-attribute` ¶

`dest = ''` `class-attribute` `instance-attribute` ¶

`port = 0` `class-attribute` `instance-attribute` ¶

`proto = 0` `class-attribute` `instance-attribute` ¶

`domain = ''` `class-attribute` `instance-attribute` ¶

`action = ''` `class-attribute` `instance-attribute` ¶

`ok = False` `class-attribute` `instance-attribute` ¶

`reason = ''` `class-attribute` `instance-attribute` ¶

`dossier = field(default_factory=dict)` `class-attribute` `instance-attribute` ¶

`ClearanceHub(*, clearance_socket=None, reader_socket=None, verdict_client=None, socket_context=None)` ¶

Server for the org.terok.Clearance1 interface.

Owns three pieces of state:

_subscribers — a set of bounded per-connection queues; the hub puts a ClearanceEvent on each one every time the reader ingester delivers an event. Slow clients see their oldest events dropped; fast clients aren't affected.
_live_verdicts — the request_id → (container, dest) map the Verdict method checks for the authz binding.
An EventIngester bound to the canonical reader socket.

Lifecycle: start brings everything up; stop tears it down under individual timeouts so a flaky bus or a stuck subscriber can't stall the per-container supervisor's teardown.

Configure the two sockets and the verdict-helper client.

verdict_client is injected so tests can stub out shield exec without spawning the helper process. Production callers leave it defaulted — a fresh VerdictClient pointing at the canonical helper socket.

socket_context — optional zero-arg callable returning a context manager around the varlink bind() (forwarded to bind_hardened). terok-sandbox passes its SELinux setsockcreatecon helper here so the hub socket is labelled terok_socket_t and confined containers can connectto it.

Source code in src/terok_clearance/hub/server.py

def __init__(
    self,
    *,
    clearance_socket: Path | None = None,
    reader_socket: Path | None = None,
    verdict_client: VerdictClient | None = None,
    socket_context: Callable[[], AbstractContextManager[None]] | None = None,
) -> None:
    """Configure the two sockets and the verdict-helper client.

    ``verdict_client`` is injected so tests can stub out shield exec
    without spawning the helper process.  Production callers leave
    it defaulted — a fresh [`VerdictClient`][terok_clearance.hub.server.VerdictClient] pointing at the
    canonical helper socket.

    *socket_context* — optional zero-arg callable returning a context
    manager around the varlink ``bind()`` (forwarded to
    [`bind_hardened`][terok_clearance.wire.socket.bind_hardened]).
    terok-sandbox passes its SELinux ``setsockcreatecon`` helper here
    so the hub socket is labelled ``terok_socket_t`` and confined
    containers can ``connectto`` it.
    """
    self._clearance_socket = clearance_socket or default_clearance_socket_path()
    self._reader_socket = reader_socket  # None → EventIngester picks its default.
    self._verdict_client = verdict_client or VerdictClient()
    self._socket_context = socket_context

    self._subscribers: set[asyncio.Queue[ClearanceEvent]] = set()
    # request_id → (container, dest) the hub emitted in the matching
    # ConnectionBlocked; Verdict calls must cite a triple that matches.
    self._live_verdicts: dict[str, tuple[str, str]] = {}

    self._ingester: EventIngester | None = None
    self._varlink_server: VarlinkUnixServer | None = None

`start()` `async` ¶

Bring the ingester + varlink server online and accept clients.

Transactional: if the varlink bind fails after the ingester is already listening, the ingester is stopped before the exception propagates so a half-started hub doesn't leak a live reader-side socket on systemd restart paths.

Source code in src/terok_clearance/hub/server.py

async def start(self) -> None:
    """Bring the ingester + varlink server online and accept clients.

    Transactional: if the varlink bind fails after the ingester is
    already listening, the ingester is stopped before the exception
    propagates so a half-started hub doesn't leak a live
    reader-side socket on systemd restart paths.
    """
    self._ingester = EventIngester(
        socket_path=self._reader_socket or _default_reader_socket(),
        on_event=self._relay_reader_event,
    )
    await self._ingester.start()
    try:
        registry = VarlinkInterfaceRegistry()
        registry.register_interface(
            Clearance1Interface(
                event_stream_factory=self._subscribe,
                apply_verdict=self._apply_verdict,
            )
        )
        registry.register_interface(
            VarlinkServiceInterface(
                vendor="terok",
                product="terok-clearance",
                version=_own_version(),
                url="https://github.com/terok-ai/terok-clearance",
                registry=registry,
            )
        )

        from terok_clearance.wire.socket import bind_hardened

        async def _factory(path: str) -> object:
            return await create_unix_server(registry.protocol_factory, path=path)

        self._varlink_server = await bind_hardened(
            _factory,
            self._clearance_socket,
            "clearance",
            socket_context=self._socket_context,
        )
    except BaseException:
        with contextlib.suppress(Exception):
            await self._ingester.stop()
        self._ingester = None
        raise
    _log.info("clearance hub online at %s", self._clearance_socket)

`stop()` `async` ¶

Close the varlink server + ingester; drain subscriber queues.

Source code in src/terok_clearance/hub/server.py

async def stop(self) -> None:
    """Close the varlink server + ingester; drain subscriber queues."""
    if self._varlink_server is not None:
        # ``close()`` on its own only stops accepting new connections;
        # existing subscribers would sit forever in ``queue.get()`` and
        # ``wait_closed`` would hang until the timeout fires.
        # ``close_clients()`` walks the live transports and closes them,
        # which makes the server-side ``_call_async_method_more``'s
        # next ``send_reply`` fail with OSError — that in turn calls
        # ``generator.aclose()`` on the subscriber, propagating cleanly
        # through to our ``finally`` block.  This avoids the
        # assertion asyncvarlink fires when a streaming generator
        # ends "normally" with ``continues=True`` on the last reply.
        self._varlink_server.close()
        with contextlib.suppress(AttributeError):
            self._varlink_server.close_clients()
        with contextlib.suppress(TimeoutError, Exception):
            await asyncio.wait_for(self._varlink_server.wait_closed(), timeout=1.0)
        self._varlink_server = None
    if self._ingester is not None:
        with contextlib.suppress(Exception):
            await self._ingester.stop()
        self._ingester = None
    with contextlib.suppress(Exception):
        await self._verdict_client.stop()
    self._subscribers.clear()
    self._live_verdicts.clear()

`CallbackNotifier(on_notify=None, *, on_container_started=None, on_container_exited=None, on_shield_up=None, on_shield_down=None, on_shield_disengaged=None)` ¶

Notifier backend that delegates rendering to caller-supplied hooks.

Parameters:

Name	Type	Description	Default
`on_notify`	`Callable[[Notification], None] \| None`	Called for every `notify()` with a `Notification`. Receives new notifications (`replaces_id == 0`) and in-place updates (`replaces_id > 0`, e.g. verdict results).	`None`
`on_container_started`	`Callable[[str], None] \| None`	Called for every `ContainerStarted` signal with the short container ID. Optional — consumers that don't care about container lifecycle skip the parameter.	`None`
`on_container_exited`	`Callable[[str, str], None] \| None`	Called for every `ContainerExited` signal with `(container, reason)`. Optional, same semantics.	`None`
`on_shield_up`	`Callable[[str], None] \| None`	Called for every `ShieldUp` signal with the container identifier. Lets the TUI flip a "shielded" badge on the per-container row without polling nft state.	`None`
`on_shield_down`	`Callable[[str], None] \| None`	Called for every `ShieldDown` signal — partial bypass (loopback-only traffic still allowed).	`None`
`on_shield_disengaged`	`Callable[[str], None] \| None`	Called for every `ShieldDisengaged` signal — unrestricted bypass. Split from `on_shield_down` so the consumer can render the two modes differently.	`None`

Bind optional notify and lifecycle callbacks.