Skip to content

server

server

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

One process, one socket, one method (Apply). Runs as its own systemd user unit (terok-clearance-verdict.service) so the companion hub unit can take full seccomp + mount-ns hardening without tripping the kernel's NNP requirement and SELinux's denial of the unconfined_t → container_runtime_t transition that rootless podman needs every time shield exec's podman unshare nsenter nft.

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)

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.

Source code in src/terok_clearance/verdict/server.py 
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    *,
    socket_path: Path | None = None,
    shield_binary: str | None = None,
) -> None:
    """Configure the socket + shield executable path."""
    self._socket_path = socket_path or default_verdict_socket_path()
    self._shield_binary = shield_binary or find_shield_binary()
    self._server: object | None = None

start() async

Bind the varlink server and start accepting hub verdict calls.

Source code in src/terok_clearance/verdict/server.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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")
    _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 
77
78
79
80
81
82
83
84
85
86
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

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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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()