shield

shield

Adapter for terok-shield egress firewall.

Two classes carry the sandbox-side policy layer over terok-shield:

• ShieldManager — per-task wrapper around Shield. Caches the underlying instance. Kill-switch-aware methods (pre_start, up, down, check_environment) short-circuit when shield_disabled is set; always-on methods (quarantine, state) always hit the live shield because panic overrides the kill-switch and state probes report what nft actually sees. status is config-level only and surfaces the kill-switch flag in its dict rather than short-circuiting.
• ShieldHooks — the host-wide OCI hooks installer, scoped to the root/user dual-scope flag pair the terok setup and terok-sandbox CLIs expose. Delegates to terok-shield's HooksInstaller for the actual file writes; terok-shield owns the on-disk install layout, so sandbox carries no private mirror of it.

ShieldManager(task_dir, cfg=None, *, runtime=ShieldRuntime.DEFAULT, loopback_ports_override=None)

Per-task wrapper around Shield.

Holds the (task_dir, cfg, runtime) tuple a Shield is built from and caches the constructed instance — the previous free-function surface rebuilt a Shield on every call, which paid the ShieldConfig + collaborator-wiring cost twice for every transition pair (pre_start → up, up → down, …).

Kill-switch-aware methods (pre_start, up, down) short-circuit when shield_disabled is set on the configuration. Always-on methods (quarantine, state) always run — panic overrides the kill-switch, and state probes report what nft actually sees regardless of operator intent.

Bind the manager to a task directory and shield configuration.

runtime selects the container runtime category — DEFAULT for crun/runc/youki (dnsmasq on netns 127.0.0.1), KRUN for the libkrun microVM path (dnsmasq on a link-local address the guest can reach via passt). Callers that drive the launch path map their runtime string (RunSpec.runtime) to the enum.

loopback_ports_override replaces the cfg-derived (gate_port, token_broker_port, ssh_signer_port) triple — the per-container launch path passes the freshly-allocated broker and signer ports so shield's nft rules allow the actual host ports the supervisor binds.

Source code in src/terok_sandbox/integrations/shield.py

def __init__(
    self,
    task_dir: Path,
    cfg: SandboxConfig | None = None,
    *,
    runtime: ShieldRuntime = ShieldRuntime.DEFAULT,
    loopback_ports_override: tuple[int, ...] | None = None,
) -> None:
    """Bind the manager to a task directory and shield configuration.

    *runtime* selects the container runtime category — ``DEFAULT``
    for crun/runc/youki (dnsmasq on netns ``127.0.0.1``), ``KRUN``
    for the libkrun microVM path (dnsmasq on a link-local address
    the guest can reach via passt).  Callers that drive the launch
    path map their runtime string (``RunSpec.runtime``) to the
    enum.

    *loopback_ports_override* replaces the cfg-derived
    ``(gate_port, token_broker_port, ssh_signer_port)`` triple — the
    per-container launch path passes the freshly-allocated broker
    and signer ports so shield's nft rules allow the actual host
    ports the supervisor binds.
    """
    self._task_dir = task_dir
    self._cfg = cfg or SandboxConfig()
    self._runtime = runtime
    self._loopback_ports_override = loopback_ports_override

state_dir property

Per-task shield state directory: {task_dir}/shield.

disabled property

True when shield_disabled is set on the sandbox configuration.

shield cached property

Lazily constructed Shield instance.

Built from a ShieldConfig whose loopback_ports reflect the actual gate/broker/signer ports — auto-allocated configs default those fields to None, which would otherwise silently produce an empty tuple and a shield ruleset with no tcp dport <p> ip daddr 169.254.1.2 accept rules, causing container→host TCP traffic to fall through to the private-range reject (#156 regression follow-up).

dns_tier property

The DNS tier this task launched with; None when none was recorded.

Reads only the recorded tier file — like status, it pays no Shield wire-up cost. The tier says what it provides: whether its allow sets are live, and a hint for the operator when not.

pre_start(container, *, security_deny=(), provider_allow=(), project_allow=(), override=())

Return extra podman run args for egress firewalling.

The four tier arguments are the orchestrator's generated policy tiers, which shield writes into the bundle so this layer only carries the data: security_deny → t20 (deny direct-to-vault-host), provider_allow → t30 (provider egress), project_allow → t40 (git remote + custom, merged with the composed profiles), override → t10 (break-glass allow above the deny). Shield owns each tier outright: an empty tuple (the default) clears that tier, so every call must carry the full current data — never rely on a previous launch's content surviving.

Returns an empty list (no firewall args) when the dangerous disable_firewall_no_protection override is active.

Propagates SetupRequiredError when the podman environment requires one-time hook installation.

Source code in src/terok_sandbox/integrations/shield.py

def pre_start(
    self,
    container: str,
    *,
    security_deny: tuple[str, ...] = (),
    provider_allow: tuple[str, ...] = (),
    project_allow: tuple[str, ...] = (),
    override: tuple[str, ...] = (),
) -> list[str]:
    """Return extra ``podman run`` args for egress firewalling.

    The four tier arguments are the orchestrator's generated policy tiers,
    which shield writes into the bundle so this layer only carries the data:
    *security_deny* → t20 (deny direct-to-vault-host), *provider_allow* → t30
    (provider egress), *project_allow* → t40 (git remote + custom, merged
    with the composed profiles), *override* → t10 (break-glass allow above
    the deny).  Shield owns each tier outright: an empty tuple (the
    default) *clears* that tier, so every call must carry the full
    current data — never rely on a previous launch's content surviving.

    Returns an empty list (no firewall args) when the dangerous
    ``disable_firewall_no_protection`` override is active.

    Propagates [`SetupRequiredError`][terok_util.SetupRequiredError] when
    the podman environment requires one-time hook installation.
    """
    if self.disabled:
        warnings.warn(_DISABLED_WARNING, stacklevel=2)
        return []
    return self.shield.pre_start(
        container,
        security_deny=security_deny,
        provider_allow=provider_allow,
        project_allow=project_allow,
        override=override,
    )

refresh(container, *, security_deny=(), provider_allow=(), project_allow=(), override=())

Recompute an existing container's policy bundle before a plain restart.

Same tier data and owns-and-clears semantics as pre_start, but for a container that already exists: shield rewrites the tiers and regenerates the pre-applied artifacts (ruleset.nft, dnsmasq config) so the next podman start enforces current policy instead of the bundle frozen at creation. No podman args are produced — the container keeps its launch-time configuration.

Source code in src/terok_sandbox/integrations/shield.py

def refresh(
    self,
    container: str,
    *,
    security_deny: tuple[str, ...] = (),
    provider_allow: tuple[str, ...] = (),
    project_allow: tuple[str, ...] = (),
    override: tuple[str, ...] = (),
) -> None:
    """Recompute an existing container's policy bundle before a plain restart.

    Same tier data and owns-and-clears semantics as
    [`pre_start`][terok_sandbox.integrations.shield.ShieldManager.pre_start],
    but for a container that already exists: shield rewrites the tiers and
    regenerates the pre-applied artifacts (``ruleset.nft``, dnsmasq
    config) so the next ``podman start`` enforces *current* policy
    instead of the bundle frozen at creation.  No podman args are
    produced — the container keeps its launch-time configuration.
    """
    if self.disabled:
        warnings.warn(_DISABLED_WARNING, stacklevel=2)
        return
    self.shield.refresh(
        container,
        security_deny=security_deny,
        provider_allow=provider_allow,
        project_allow=project_allow,
        override=override,
    )

up(container, container_id)

Set shield to deny-all mode for a running container.

container is the operator-facing podman name (audit-log key); container_id is the full podman UUID — terok-shield's per- container hub socket is keyed on it. Both are mandatory: terok-shield removed the global-hub fallback in feat/per-container-supervisor.

Source code in src/terok_sandbox/integrations/shield.py

def up(self, container: str, container_id: str) -> None:
    """Set shield to deny-all mode for a running container.

    *container* is the operator-facing podman name (audit-log key);
    *container_id* is the full podman UUID — terok-shield's per-
    container hub socket is keyed on it.  Both are mandatory:
    terok-shield removed the global-hub fallback in
    ``feat/per-container-supervisor``.
    """
    if self.disabled:
        return
    self.shield.up(container, container_id)

down(container, container_id, *, disengaged=False)

Switch shield to the DOWN posture (allow egress) for a running container.

container / container_id — see up. When disengaged is True, the container takes the DISENGAGED posture instead: nothing is enforced — no deny set, no private-range or hard-deny reject.

Source code in src/terok_sandbox/integrations/shield.py

def down(self, container: str, container_id: str, *, disengaged: bool = False) -> None:
    """Switch shield to the DOWN posture (allow egress) for a running container.

    *container* / *container_id* — see
    [`up`][terok_sandbox.integrations.shield.ShieldManager.up].  When
    *disengaged* is True, the container takes the DISENGAGED posture
    instead: nothing is enforced — no deny set, no private-range or
    hard-deny reject.
    """
    if self.disabled:
        return
    self.shield.down(container, container_id, disengaged=disengaged)

quarantine(container)

Total network blackout — drop all traffic, log dropped traffic.

Ignores shield_disabled because panic overrides the kill-switch.

Source code in src/terok_sandbox/integrations/shield.py

def quarantine(self, container: str) -> None:
    """Total network blackout — drop all traffic, log dropped traffic.

    Ignores ``shield_disabled`` because panic overrides the kill-switch.
    """
    self.shield.quarantine(container)

state(container)

Return the live shield state for a running container.

Queries actual nft state even when the kill-switch is set, because containers started before it was enabled may still have active rules.

Source code in src/terok_sandbox/integrations/shield.py

def state(self, container: str) -> ShieldState:
    """Return the live shield state for a running container.

    Queries actual nft state even when the kill-switch is set,
    because containers started *before* it was enabled may still
    have active rules.
    """
    return self.shield.state(container)

status()

Return shield status dict from the sandbox configuration.

Reads only the sandbox configuration — does not instantiate the underlying Shield, so callers that only want configuration-level shape don't pay the Shield wire-up cost.

Source code in src/terok_sandbox/integrations/shield.py

def status(self) -> dict:
    """Return shield status dict from the sandbox configuration.

    Reads only the sandbox configuration — does not instantiate
    the underlying Shield, so callers that only want
    configuration-level shape don't pay the Shield wire-up cost.
    """
    result: dict = {
        "mode": "hook",
        "profiles": list(self._cfg.shield_profiles),
        "audit_enabled": self._cfg.shield_audit,
    }
    if self.disabled:
        result["disable_firewall_no_protection"] = True
    return result

check_environment()

Check the podman environment for shield compatibility.

Returns a synthetic EnvironmentCheck flagging the kill-switch when the dangerous disable override is active.

Source code in src/terok_sandbox/integrations/shield.py

def check_environment(self) -> EnvironmentCheck:
    """Check the podman environment for shield compatibility.

    Returns a synthetic [`EnvironmentCheck`][terok_shield.EnvironmentCheck]
    flagging the kill-switch when the dangerous disable override is active.
    """
    if self.disabled:
        return EnvironmentCheck(
            ok=False,
            health="disabled",
            issues=["disable_firewall_no_protection is set — egress firewall disabled"],
        )
    return self.shield.check_environment()

interactive_session(container)

Run the terminal clearance fallback for this task's shield.

Thin wrapper that spares callers from reaching into terok_shield.simple_clearance and rebuilding the state_dir themselves. Refuses to run when the D-Bus clearance hub is already handling the session.

Source code in src/terok_sandbox/integrations/shield.py

def interactive_session(self, container: str) -> None:
    """Run the terminal clearance fallback for this task's shield.

    Thin wrapper that spares callers from reaching into
    [`terok_shield.simple_clearance`][terok_shield.simple_clearance]
    and rebuilding the ``state_dir`` themselves.  Refuses to run
    when the D-Bus clearance hub is already handling the session.
    """
    from terok_shield.simple_clearance import run_simple_clearance

    run_simple_clearance(self.state_dir, container)

watch_session(container)

Stream shield blocked-access events for this task as JSON lines.

Thin wrapper that spares callers from reaching into terok_shield.watch and rebuilding the state_dir themselves.

Source code in src/terok_sandbox/integrations/shield.py

def watch_session(self, container: str) -> None:
    """Stream shield blocked-access events for this task as JSON lines.

    Thin wrapper that spares callers from reaching into
    [`terok_shield.watch`][terok_shield.watch] and rebuilding the
    ``state_dir`` themselves.
    """
    from terok_shield.watch import run_watch

    run_watch(self.state_dir, container)

ShieldHooks

Host-wide OCI hooks installer — no task context.

Thin pass-through to terok-shield's HooksInstaller. Kept as a class so the sandbox setup aggregator can swap it out in tests without poking around terok-shield internals.

check_setup(*, live=False) staticmethod

Delegate readiness to Shield, which owns its installation.

Source code in src/terok_sandbox/integrations/shield.py

@staticmethod
def check_setup(*, live: bool = False) -> tuple[SetupCheck, ...]:
    """Delegate readiness to Shield, which owns its installation."""
    return HooksInstaller().check_setup(live=live)

install() staticmethod

Install global OCI hooks for shield egress firewalling.

Global hooks are required on all podman versions to survive container stop/start cycles (terok-shield#122). Single layout: scripts, ballast, and JSON descriptors all land in namespace_state_dir("shield") / "hooks"; containers.conf is patched to register that path.

Source code in src/terok_sandbox/integrations/shield.py

@staticmethod
def install() -> None:
    """Install global OCI hooks for shield egress firewalling.

    Global hooks are required on all podman versions to survive
    container stop/start cycles (terok-shield#122).  Single
    layout: scripts, ballast, and JSON descriptors all land in
    ``namespace_state_dir("shield") / "hooks"``;
    ``containers.conf`` is patched to register that path.
    """
    HooksInstaller().install()

uninstall() staticmethod

Remove the global OCI hooks install writes.

Idempotent — missing files are tolerated.

Source code in src/terok_sandbox/integrations/shield.py

@staticmethod
def uninstall() -> None:
    """Remove the global OCI hooks [`install`][terok_sandbox.integrations.shield.ShieldHooks.install] writes.

    Idempotent — missing files are tolerated.
    """
    HooksInstaller().uninstall()

check_environment(cfg=None)

Probe the podman environment with no task context.

Returns a synthetic EnvironmentCheck when shield_disabled is set; otherwise constructs a throwaway ShieldManager bound to a temp directory and delegates to its check_environment. Kept as a free function because the setup CLI runs before any task directory exists.

Source code in src/terok_sandbox/integrations/shield.py

def check_environment(cfg: SandboxConfig | None = None) -> EnvironmentCheck:
    """Probe the podman environment with no task context.

    Returns a synthetic [`EnvironmentCheck`][terok_shield.EnvironmentCheck]
    when ``shield_disabled`` is set; otherwise constructs a throwaway
    [`ShieldManager`][terok_sandbox.integrations.shield.ShieldManager]
    bound to a temp directory and delegates to its
    [`check_environment`][terok_sandbox.integrations.shield.ShieldManager.check_environment].
    Kept as a free function because the setup CLI runs before any
    task directory exists.
    """
    with tempfile.TemporaryDirectory() as tmp:
        return ShieldManager(Path(tmp), cfg).check_environment()