terok_clearance

`terok_clearance` ¶

Clearance hub + desktop notification library for terok.

Two unrelated wire formats live under this one package:

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; every other D-Bus path in this package (org.terok.Shield1) was removed in favour of the varlink transport.

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

all = ['CLEARANCE_INTERFACE_NAME', 'CallbackNotifier', 'Clearance1Interface', 'ClearanceClient', 'ClearanceEvent', 'ClearanceHub', 'ContainerIdentity', 'ContainerInfo', 'ContainerInspector', 'DbusNotifier', 'EventSubscriber', 'IdentityResolver', 'InvalidAction', 'Notification', 'Notifier', 'NullInspector', 'NullNotifier', 'ShieldCliFailed', 'UnknownRequest', 'VerdictTupleMismatch', 'check_units_outdated', 'configure_logging', 'create_notifier', 'default_clearance_socket_path', 'install_notifier_service', 'read_installed_unit_version', 'serve', 'uninstall_notifier_service', 'uninstall_service', 'wait_for_shutdown_signal'] `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: object | None = None
    self._rpc_transport: object | None = None
    self._sub_proxy: object | None = None
    self._rpc_proxy: object | 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))

`IdentityResolver(inspector)` ¶

Compose a ContainerInspector lookup + task-meta YAML into an identity.

Callable: resolver(container_id) -> ContainerIdentity. Four soft-fail paths, all returning a degraded identity that keeps the notification pipeline usable:

The inspector failed → empty ContainerIdentity; the subscriber falls back to the raw container ID.
Container carries no terok annotations (a standalone container that happened to hit the firewall) → container-name-only.
ai.terok.task_meta_path annotation absent → identity without task_name (project + task_id still present).
task_meta_path YAML unreadable / missing / malformed → same as above; the name field is left empty.

Configure the resolver with a ContainerInspector implementation.

The inspector is required (no default) so the caller owns the runtime-selection decision — clearance is runtime-neutral and must not reach for a specific backend itself. The notifier entry point picks an appropriate implementation at startup (terok-sandbox's create_container_inspector when available, NullInspector otherwise).

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

def __init__(self, inspector: ContainerInspector) -> None:
    """Configure the resolver with a [`ContainerInspector`][terok_clearance.client.identity_resolver.ContainerInspector] implementation.

    The inspector is required (no default) so the caller owns the
    runtime-selection decision — clearance is runtime-neutral and
    must not reach for a specific backend itself.  The notifier
    entry point picks an appropriate implementation at startup
    (terok-sandbox's ``create_container_inspector`` when available,
    [`NullInspector`][terok_clearance.NullInspector] otherwise).
    """
    self._inspector = inspector

`call(container_id)` ¶

Return the task-aware identity for container_id.

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

def __call__(self, container_id: str) -> ContainerIdentity:
    """Return the task-aware identity for *container_id*."""
    try:
        info = self._inspector(container_id)
    except Exception:
        # The contract says inspectors soft-fail by returning an
        # empty ``ContainerInfo``, but a runtime-side race or an
        # unexpected error path in a third-party backend can still
        # raise.  Clamp it here so the caller (notifier / TUI)
        # never takes a crash from identity resolution.
        _log.debug("ContainerInspector raised for %s", container_id, exc_info=True)
        return ContainerIdentity()
    if not info.container_id:
        return ContainerIdentity()
    project = info.annotations.get(ANNOTATION_PROJECT, "")
    task_id = info.annotations.get(ANNOTATION_TASK, "")
    if not (project and task_id):
        return ContainerIdentity(container_name=info.name, project=project, task_id=task_id)
    meta_path = info.annotations.get(ANNOTATION_TASK_META_PATH, "")
    return ContainerIdentity(
        container_name=info.name,
        project=project,
        task_id=task_id,
        task_name=_read_task_name(meta_path) if meta_path else "",
    )

`EventSubscriber(notifier, client=None, *, identity_resolver=None, socket_path=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`
`identity_resolver`	`Callable[[str], ContainerIdentity] \| None`	Turns a short container ID into a `ContainerIdentity` so terok task annotations surface as "Task: project/task_id · name" bodies. Called from a worker thread so a slow `podman inspect` doesn't stall the event loop. `None` renders the raw container ID.	`None`
`socket_path`	`Path \| None`	Clearance-socket override when client isn't supplied (tests).	`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,
    *,
    identity_resolver: Callable[[str], ContainerIdentity] | None = None,
    socket_path: Path | None = None,
) -> None:
    """Initialise the subscriber with a notifier and transport."""
    self._notifier = notifier
    self._client = client or ClearanceClient(socket_path=socket_path)
    self._identity_resolver = identity_resolver
    # 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()

`ContainerInfo(container_id='', name='', state='', annotations=(lambda: _EMPTY_ANNOTATIONS)())` `dataclass` ¶

What podman inspect tells us about one container.

Empty instance (ContainerInfo()) represents "not found" or "lookup failed" — callers should treat missing fields as best-effort and fall back to the raw container ID when they don't have a better label.

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

The short ID podman reported back, or empty on failure.

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

The container's name without podman's leading / prefix.

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

Lifecycle state: running, exited, created, etc. Empty when unknown.

`annotations = field(default_factory=(lambda: _EMPTY_ANNOTATIONS))` `class-attribute` `instance-attribute` ¶

Every OCI annotation podman recorded for this container.

Exposed as a read-only Mapping — cached instances are shared across inspector callers, so mutating the underlying dict would poison future lookups. Build with types.MappingProxyType at construction time; callers (clearance's task-aware resolver, anything else that cares) pluck out the keys they know about.

`ClearanceEvent(type, container, request_id='', dest='', port=0, proto=0, domain='', action='', ok=False, reason='')` `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. Requires an operator verdict.
verdict_applied — request_id, action, ok.
container_started — no extras.
container_exited — reason.
shield_up / shield_down / shield_down_all — no extras.

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` ¶

`ContainerIdentity(container_name='', project='', task_id='', task_name='')` `dataclass` ¶

Host-side facts about a container, as much as the resolver found.

Terok-managed task containers carry project and task_id via OCI annotations set at podman run time; task_name is looked up live from terok's task metadata so a rename between block and verdict is reflected in the resolved popup. Standalone containers produce an instance with only container_name set (or empty everywhere when podman inspect itself failed).

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

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

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

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

`ContainerInspector` ¶

Bases: Protocol

Callable that maps a container id to a ContainerInfo.

The protocol intentionally covers only the notification-rendering use case — name + OCI annotations + lifecycle state. Broader runtime operations (exec, mount, signals) live on terok_sandbox.runtime.ContainerRuntime and are not part of this contract.

Implementations MUST soft-fail: an unreachable runtime / missing container / malformed metadata returns an empty ContainerInfo rather than raising, so notification pipelines keep their fallback label instead of crashing on a lookup hiccup.

`call(container_id)` ¶

Return the best-effort ContainerInfo for container_id.

Source code in src/terok_clearance/domain/inspector.py

def __call__(self, container_id: str) -> ContainerInfo:
    """Return the best-effort [`ContainerInfo`][terok_clearance.ContainerInfo] for *container_id*."""
    ...

`NullInspector` ¶

Always-empty ContainerInspector — the graceful-degradation default.

Installed when no runtime-aware package provides a concrete backend. Every lookup returns ContainerInfo() so the notifier still renders (raw container id, no enrichment).

`call(_container_id)` ¶

Return the universal empty ContainerInfo.

Source code in src/terok_clearance/domain/inspector.py

def __call__(self, _container_id: str) -> ContainerInfo:
    """Return the universal empty [`ContainerInfo`][terok_clearance.ContainerInfo]."""
    return ContainerInfo()

`ClearanceHub(*, clearance_socket=None, reader_socket=None, verdict_client=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 burn systemd's stop-sigterm deadline.

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.

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,
) -> 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.
    """
    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._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: object | None = None  # asyncvarlink's UnixServer

`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"
        )
    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_down_all=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_down_all`	`Callable[[str], None] \| None`	Called for every `ShieldDownAll` signal — unrestricted bypass. Split from `on_shield_down` so the consumer can render the two modes differently.	`None`

Bind optional notify and lifecycle callbacks.