server

server

The verdict helper — a minimal varlink server wrapping terok-shield.

One process, one socket, one method (Apply). In the per-container-supervisor model the helper runs inside the Sandbox supervisor alongside the hub, on a per-container socket; the standalone terok-clearance-hub serve-verdict entry point is kept for integration testing.

Stateless: no authz decisions, no request-id binding, no fan-out. The hub already validated the verdict triple before forwarding; the helper exists solely to isolate the hostile exec path from the hardened receive path.

VerdictServer(*, socket_path=None, shield_binary=None, socket_context=None)

Per-process wrapper around the Apply varlink interface.

The hub is the only legitimate client; SO_PEERCRED on the unix socket rejects peers with a different UID, and bind_hardened leaves the socket mode 0600 for the lifetime of the server.

Configure the socket + shield executable path.

socket_context — see ClearanceHub; forwarded to bind_hardened so the verdict socket can carry a custom SELinux type.

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

def __init__(
    self,
    *,
    socket_path: Path | None = None,
    shield_binary: str | None = None,
    socket_context: Callable[[], AbstractContextManager[None]] | None = None,
) -> None:
    """Configure the socket + shield executable path.

    *socket_context* — see
    [`ClearanceHub`][terok_clearance.ClearanceHub];
    forwarded to
    [`bind_hardened`][terok_clearance.wire.socket.bind_hardened] so
    the verdict socket can carry a custom SELinux type.
    """
    self._socket_path = socket_path or default_verdict_socket_path()
    self._shield_binary = shield_binary or find_shield_binary()
    self._socket_context = socket_context
    self._server: VarlinkUnixServer | None = None

start() async

Bind the varlink server and start accepting hub verdict calls.

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

async def start(self) -> None:
    """Bind the varlink server and start accepting hub verdict calls."""
    registry = VarlinkInterfaceRegistry()
    registry.register_interface(Verdict1Interface(apply_verdict=self._apply))
    registry.register_interface(
        VarlinkServiceInterface(
            vendor="terok",
            product="terok-clearance-verdict",
            version=_own_version(),
            url="https://github.com/terok-ai/terok-clearance",
            registry=registry,
        )
    )

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

    self._server = await bind_hardened(
        _factory,
        self._socket_path,
        "verdict",
        socket_context=self._socket_context,
    )
    _log.info("verdict helper online at %s", self._socket_path)

stop() async

Close the varlink server; existing in-flight Apply calls finish first.

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

async def stop(self) -> None:
    """Close the varlink server; existing in-flight Apply calls finish first."""
    if self._server is None:
        return
    self._server.close()
    with contextlib.suppress(AttributeError):
        self._server.close_clients()
    with contextlib.suppress(TimeoutError, Exception):
        await asyncio.wait_for(self._server.wait_closed(), timeout=1.0)
    self._server = None

run_shield(shield_binary, container, dest, action) async

Invoke terok-shield <action> <container> <dest>; return (ok, snippet).

Bounded by _SHIELD_CLI_TIMEOUT_S. Spawn errors, non-zero exit, and timeouts all fold into (False, reason) so callers see one shape regardless of how shield misbehaved. snippet is capped at _STDERR_CAP_BYTES.

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

async def run_shield(
    shield_binary: str | None, container: str, dest: str, action: str
) -> tuple[bool, str]:
    """Invoke ``terok-shield <action> <container> <dest>``; return ``(ok, snippet)``.

    Bounded by `_SHIELD_CLI_TIMEOUT_S`.  Spawn errors, non-zero
    exit, and timeouts all fold into ``(False, reason)`` so callers
    see one shape regardless of how shield misbehaved.  ``snippet``
    is capped at `_STDERR_CAP_BYTES`.
    """
    if not shield_binary:
        return False, "terok-shield not found on PATH"
    try:
        proc = await asyncio.create_subprocess_exec(
            shield_binary,
            action,
            container,
            dest,
            stdout=asyncio.subprocess.DEVNULL,
            stderr=asyncio.subprocess.PIPE,
        )
    except OSError as exc:
        _log.exception("failed to spawn terok-shield")
        return False, f"spawn failed: {exc}"
    try:
        _, stderr_bytes = await asyncio.wait_for(proc.communicate(), timeout=_SHIELD_CLI_TIMEOUT_S)
    except TimeoutError:
        proc.kill()
        with contextlib.suppress(Exception):
            await proc.communicate()
        _log.warning("shield %s timed out after %gs", action, _SHIELD_CLI_TIMEOUT_S)
        return False, f"timed out after {_SHIELD_CLI_TIMEOUT_S}s"
    snippet = (stderr_bytes[:_STDERR_CAP_BYTES] or b"").decode(errors="replace").strip()
    ok = proc.returncode == 0
    if not ok:
        _log.warning("shield %s failed: %s", action, snippet)
    return ok, snippet

find_shield_binary()

Locate terok-shield — sibling venv first, then PATH, then None.

The sibling check handles the pipx / poetry case where terok-shield ships in the same venv as terok-clearance; we prefer it over PATH so a shell-rc PATH shim can't redirect verdicts through a different installation. is_file alone would happily return a non-executable artifact, so the exec-bit check prevents a broken install from failing every verdict instead of falling through to PATH's working copy.

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

def find_shield_binary() -> str | None:
    """Locate ``terok-shield`` — sibling venv first, then PATH, then ``None``.

    The sibling check handles the pipx / poetry case where terok-shield
    ships in the same venv as terok-clearance; we prefer it over PATH
    so a shell-rc ``PATH`` shim can't redirect verdicts through a
    different installation.  ``is_file`` alone would happily return a
    non-executable artifact, so the exec-bit check prevents a broken
    install from failing every verdict instead of falling through to
    PATH's working copy.
    """
    sibling = Path(sys.executable).parent / "terok-shield"
    if sibling.is_file() and os.access(sibling, os.X_OK):
        return str(sibling)
    return shutil.which("terok-shield")

serve() async

Bring the verdict helper online and stay up until SIGINT/SIGTERM.

Mirrors terok_clearance.hub.server.serve so the CLI layer can dispatch both entrypoints through the same asyncio.run pattern.

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

async def serve() -> None:
    """Bring the verdict helper online and stay up until SIGINT/SIGTERM.

    Mirrors [`terok_clearance.hub.server.serve`][terok_clearance.hub.server.serve] so the CLI layer
    can dispatch both entrypoints through the same ``asyncio.run``
    pattern.
    """
    from terok_clearance.runtime.service import configure_logging, wait_for_shutdown_signal

    configure_logging()
    server = VerdictServer()
    await server.start()
    try:
        await wait_for_shutdown_signal()
    finally:
        await server.stop()