sandbox

sandbox

High-level sandbox facade composing shield, gate, runtime, and SSH.

Convenience composition layer — delegates container lifecycle to the injected ContainerRuntime, plus convenience wrappers for gate and shield services. The launch path (Sandbox.run, Sandbox.create) is still podman-specific and invokes the podman CLI directly; Phase 3 will factor that through the runtime as well.

READY_MARKER = '>> init complete' module-attribute

Default log line emitted by init-ssh-and-repo.sh when the container is ready.

SAFE_RUNTIMES = frozenset({'crun', 'krun'}) module-attribute

OCI runtimes the sandbox will pass to podman --runtime.

Allowlist enforced at command-assembly time. Podman's --runtime accepts either a runtime name (crun, krun) or a path to a binary — passing a path would let a caller who controls RunSpec.runtime make podman execute an arbitrary host binary as part of container creation. By rejecting anything outside this set (and anything that looks path-shaped) we keep the runtime selection a known-isolation choice rather than an arbitrary-code-execution surface.

SAFE_ANNOTATION_KEYS = frozenset({'dossier.meta_path', 'krun.cpus', 'terok.sandbox.sidecar'}) module-attribute

OCI annotation keys allowed on RunSpec.annotations.

Annotations are privileged config — they bind a running container to host-side state the shield (or other readers) consult on every event. The allowlist prevents a caller-controlled RunSpec from smuggling an unrecognised key past the sandbox.

LifecycleHooks(pre_start=None, post_start=None, post_ready=None, post_stop=None) dataclass

Optional callbacks fired at container lifecycle transitions.

All slots are None by default. Sandbox.run() fires pre_start before podman run and post_start after a successful launch. post_ready and post_stop are available for callers to invoke at the appropriate time (e.g. after log streaming or container exit).

pre_start = None class-attribute instance-attribute

Fired before podman run.

post_start = None class-attribute instance-attribute

Fired after a successful podman run.

post_ready = None class-attribute instance-attribute

Fired when the container reports ready (caller responsibility).

post_stop = None class-attribute instance-attribute

Fired after the container exits (caller responsibility).

Sharing

Directory sharing semantics — expresses intent, not backend details.

The sandbox translates these into backend-specific flags (e.g. SELinux relabel :z / :Z for Podman) and uses them to drive sealed-mode decisions (private dirs are injected, shared dirs may be skipped).

PRIVATE = 'private' class-attribute instance-attribute

Exclusive to one container — no other container accesses this directory.

SHARED = 'shared' class-attribute instance-attribute

Shared across multiple containers (e.g. agent auth/config directories).

VolumeSpec(host_path, container_path, sharing=Sharing.SHARED, read_only=False, live=False) dataclass

Typed description of a host↔container directory binding.

Replaces raw volume strings ("host:container:z") with structured data so the sandbox can decide how to materialise each binding — as a bind mount (shared mode) or a podman cp injection (sealed mode).

sharing expresses the caller's intent (private vs shared); the sandbox translates that into backend-specific flags (e.g. SELinux relabeling for Podman). In sealed mode, sharing semantics can also drive whether a directory is injected (private) or skipped (shared config that the vault replaces).

host_path instance-attribute

Absolute host-side path to mount or copy in.

container_path instance-attribute

Absolute path inside the container (e.g. "/workspace").

sharing = Sharing.SHARED class-attribute instance-attribute

Sharing semantics: Sharing.PRIVATE or Sharing.SHARED.

read_only = False class-attribute instance-attribute

When True, mount the volume read-only inside the container.

Used to layer immutable views on top of writable directory mounts — e.g. exposing a credential file to the agent while preventing it from overwriting the host-side phantom token.

live = False class-attribute instance-attribute

When True, this volume is bind-mounted even in sealed mode.

Service plumbing (per-container vault/ssh-agent socket dir, gate socket, sourced-at-runtime bridge scripts) must be live: sealed-mode podman cp would snapshot an empty dir on the container side and the supervisor's later-bound sockets would never appear inside. Operator state (workspace, agent config) leaves this False so sealed mode gets fresh copies as designed.

to_mount_arg()

Format as a -v flag value for podman run.

Source code in src/terok_sandbox/sandbox.py

def to_mount_arg(self) -> str:
    """Format as a ``-v`` flag value for ``podman run``."""
    try:
        relabel = _SHARING_TO_RELABEL[self.sharing]
    except KeyError:
        raise ValueError(f"Unknown sharing mode: {self.sharing!r}") from None
    opts = relabel + (",ro" if self.read_only else "")
    return f"{self.host_path}:{self.container_path}:{opts}"

RunSpec(container_name, image, env, volumes, command, task_dir, gpu_enabled=False, memory=None, cpus=None, extra_args=(), unrestricted=True, sealed=False, hostname=None, runtime=None, annotations=(lambda: MappingProxyType({}))(), loopback_ports=()) dataclass

Everything needed for a single podman run invocation.

container_name instance-attribute

Unique container name.

image instance-attribute

Image tag to run (e.g. terok-l1-cli:ubuntu-24.04).

env instance-attribute

Environment variables injected into the container.

volumes instance-attribute

Host↔container directory bindings (mounted or injected per sealed).

command instance-attribute

Command to execute inside the container.

task_dir instance-attribute

Host-side task directory (for shield state, logs, etc.).

gpu_enabled = False class-attribute instance-attribute

Whether to pass GPU device args to podman.

memory = None class-attribute instance-attribute

Podman --memory value (e.g. "4g", "512m"). None = unlimited.

cpus = None class-attribute instance-attribute

Podman --cpus value (e.g. "2.0", "0.5"). None = unlimited.

extra_args = () class-attribute instance-attribute

Additional podman run arguments (e.g. port publishing).

unrestricted = True class-attribute instance-attribute

When False, adds --security-opt no-new-privileges.

sealed = False class-attribute instance-attribute

When True, volumes are injected via podman cp instead of bind-mounted.

hostname = None class-attribute instance-attribute

Override the in-container hostname (podman --hostname).

When None (default), podman assigns the short container ID as the hostname. Orchestrators may set this to a value that correlates with their own task/container identity — e.g. so a shell prompt inside the container matches the name the operator sees in task lists. Must be a valid DNS hostname (letters/digits/hyphens, ≤253 chars); podman enforces the rule when parsing the flag.

runtime = None class-attribute instance-attribute

OCI runtime to use (podman --runtime).

None (default) lets podman pick — its built-in default is crun. Set to "krun" to launch the task inside a KVM microVM (Phase 3 KrunRuntime). Backend-neutral here; the runtime string is passed through verbatim and any compatibility decisions live higher up (e.g. orchestrator config validation).

annotations = field(default_factory=(lambda: MappingProxyType({}))) class-attribute instance-attribute

OCI annotations forwarded as podman --annotation k=v entries.

Keys must be on SAFE_ANNOTATION_KEYS. Declared as Mapping so callers can pass plain dicts; __post_init__ snapshots into a MappingProxyType so the frozen-dataclass guarantee holds against caller mutation.

loopback_ports = () class-attribute instance-attribute

Per-container host ports shield's nft rules must allow.

Empty falls back to the cfg-resolved (gate_port, token_broker_port, ssh_signer_port) triple (legacy / single-daemon shape). The per-container launch path passes (gate_port, per_container.token_broker_port, per_container.ssh_signer_port) so shield allows the actual ports the supervisor binds — without this override, shield blocks the per-container broker/signer with "No route to host".

__post_init__()

Snapshot annotations so a caller-owned dict can't mutate the spec.

Callers may legitimately pass a plain dict (Pydantic, JSON-load, tests) — we'd lose the frozen guarantee if we kept the live reference. Take a copy, wrap it in a MappingProxyType, and write it back through object.__setattr__ since the dataclass itself is frozen=True.

Source code in src/terok_sandbox/sandbox.py

def __post_init__(self) -> None:
    """Snapshot ``annotations`` so a caller-owned dict can't mutate the spec.

    Callers may legitimately pass a plain ``dict`` (Pydantic, JSON-load,
    tests) — we'd lose the frozen guarantee if we kept the live
    reference.  Take a copy, wrap it in a ``MappingProxyType``, and
    write it back through ``object.__setattr__`` since the dataclass
    itself is ``frozen=True``.
    """
    object.__setattr__(self, "annotations", MappingProxyType(dict(self.annotations)))

Sandbox(config=None, *, runtime=None)

Per-task orchestrator composing runtime + services.

Holds a ContainerRuntime (defaulting to PodmanRuntime) and a SandboxConfig, and exposes gate / shield / lifecycle verbs bundled in one place. Container lifecycle verbs delegate to the runtime; the launch path (run, create) still drives podman directly because shield / gate integration is podman-specific today.

Source code in src/terok_sandbox/sandbox.py

def __init__(
    self,
    config: SandboxConfig | None = None,
    *,
    runtime: ContainerRuntime | None = None,
) -> None:
    # ``Sandbox`` is the facade that launches containers + composes
    # gate/vault managers; resolve TCP ports here so the same
    # registry pass covers everyone downstream.  ``cfg`` itself
    # stays pure — only the cfg ``Sandbox`` carries is allocated.
    self._cfg = (config or SandboxConfig()).with_resolved_ports()
    self._runtime: ContainerRuntime = runtime or PodmanRuntime()

config property

Return the sandbox configuration.

runtime property

Return the injected container runtime.

mint_gate_token()

Mint a fresh per-container gate token.

The gate lives in each container's supervisor; the token travels to the container via the sidecar and is validated in-process, so there is nothing to persist.

Source code in src/terok_sandbox/sandbox.py

def mint_gate_token(self) -> str:
    """Mint a fresh per-container gate token.

    The gate lives in each container's supervisor; the token
    travels to the container via the sidecar and is validated
    in-process, so there is nothing to persist.
    """
    from .gate.tokens import mint_gate_token

    return mint_gate_token()

gate_url(repo_path, token)

Build the in-container HTTP URL for gate access to repo_path.

Always uses the fixed loopback bridge port (see _CONTAINER_GATE_PORT): the container reaches the per-container gate through the socat bridge in both transport modes, so the URL carries no host address (gate_port is None in socket mode).

Source code in src/terok_sandbox/sandbox.py

def gate_url(self, repo_path: Path, token: str) -> str:
    """Build the in-container HTTP URL for gate access to *repo_path*.

    Always uses the fixed loopback bridge port (see
    `_CONTAINER_GATE_PORT`): the container reaches the per-container
    gate through the socat bridge in both transport modes, so the URL
    carries no host address (``gate_port`` is ``None`` in socket mode).
    """
    rel = repo_path.relative_to(self._cfg.gate_base_path).as_posix()
    return f"http://{token}@localhost:{_CONTAINER_GATE_PORT}/{rel}"

pre_start_args(container, task_dir, *, runtime=None, loopback_ports=())

Return extra podman args for shield integration.

runtime is the podman --runtime selector — passed to ShieldRuntime.from_runtime_name so shield picks the right dnsmasq bind for the krun guest's isolated loopback.

loopback_ports overrides shield's cfg-derived allowlist with per-container ports (see RunSpec.loopback_ports).

Source code in src/terok_sandbox/sandbox.py

def pre_start_args(
    self,
    container: str,
    task_dir: Path,
    *,
    runtime: str | None = None,
    loopback_ports: tuple[int, ...] = (),
) -> list[str]:
    """Return extra podman args for shield integration.

    *runtime* is the podman ``--runtime`` selector — passed to
    [`ShieldRuntime.from_runtime_name`][terok_shield.ShieldRuntime.from_runtime_name]
    so shield picks the right dnsmasq bind for the krun guest's
    isolated loopback.

    *loopback_ports* overrides shield's cfg-derived allowlist
    with per-container ports (see ``RunSpec.loopback_ports``).
    """
    from .integrations.shield import ShieldManager, ShieldRuntime

    return ShieldManager(
        task_dir,
        self._cfg,
        runtime=ShieldRuntime.from_runtime_name(runtime),
        loopback_ports_override=loopback_ports or None,
    ).pre_start(container)

shield_down(container, container_id, task_dir)

Remove shield rules for a container (allow all egress).

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.

Source code in src/terok_sandbox/sandbox.py

def shield_down(self, container: str, container_id: str, task_dir: Path) -> None:
    """Remove shield rules for a container (allow all egress).

    *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.
    """
    from .integrations.shield import ShieldManager

    ShieldManager(task_dir, self._cfg).down(container, container_id)

run(spec, *, hooks=None)

Launch a detached container from spec.

In shared mode (default), assembles and executes a single podman run -d with bind mounts.

In sealed mode (spec.sealed), splits into create → inject → start: the container is created without volumes, directories are copied in via podman cp, and the container is then started.

Fires hooks.pre_start before creation and hooks.post_start after a successful start. Raises GpuConfigError when the launch fails due to NVIDIA CDI misconfiguration.

Source code in src/terok_sandbox/sandbox.py

def run(self, spec: RunSpec, *, hooks: LifecycleHooks | None = None) -> None:
    """Launch a detached container from *spec*.

    In **shared** mode (default), assembles and executes a single
    ``podman run -d`` with bind mounts.

    In **sealed** mode (``spec.sealed``), splits into create → inject →
    start: the container is created without volumes, directories are
    copied in via ``podman cp``, and the container is then started.

    Fires *hooks.pre_start* before creation and *hooks.post_start*
    after a successful start.  Raises [`GpuConfigError`][terok_sandbox.GpuConfigError] when the
    launch fails due to NVIDIA CDI misconfiguration.
    """
    if spec.sealed:
        self.create(spec, hooks=hooks)
        # ``live`` volumes are bind-mounted (handled by _build_cmd);
        # only the rest get copied in here.
        present = tuple(v for v in spec.volumes if not v.live and v.host_path.exists())
        # Drop overlay file mounts (a file landing inside a sibling
        # dir mount); the dir-copy already wrote them, and podman cp
        # refuses to overwrite.
        dir_targets = tuple(v.container_path for v in present if v.host_path.is_dir())

        def _under_dir_mount(path: str) -> bool:
            return any(path == d or path.startswith(d.rstrip("/") + "/") for d in dir_targets)

        effective = tuple(
            v for v in present if v.host_path.is_dir() or not _under_dir_mount(v.container_path)
        )
        self._ensure_parents(spec.container_name, effective)
        for vol in effective:
            self.copy_to(spec.container_name, vol.host_path, vol.container_path)
        self.start(spec.container_name, hooks=hooks)
        return

    cmd = self._build_cmd(spec, verb="run")
    print("$", shlex.join(redact_env_args(cmd)))

    if hooks and hooks.pre_start:
        hooks.pre_start()

    self._exec_podman(cmd)

    if hooks and hooks.post_start:
        hooks.post_start()

create(spec, *, hooks=None)

Create a container without starting it.

Returns the container name. Fires hooks.pre_start before podman create. The container can then receive injected files via copy_to before being started with start.

Source code in src/terok_sandbox/sandbox.py

def create(self, spec: RunSpec, *, hooks: LifecycleHooks | None = None) -> str:
    """Create a container without starting it.

    Returns the container name.  Fires *hooks.pre_start* before
    ``podman create``.  The container can then receive injected files
    via [`copy_to`][terok_sandbox.sandbox.Sandbox.copy_to] before being started with [`start`][terok_sandbox.sandbox.Sandbox.start].
    """
    cmd = self._build_cmd(spec, verb="create")
    print("$", shlex.join(redact_env_args(cmd)))

    if hooks and hooks.pre_start:
        hooks.pre_start()

    self._exec_podman(cmd)
    return spec.container_name

start(container_name, *, hooks=None)

Start a previously created container via the runtime.

Fires hooks.post_start after a successful start.

Source code in src/terok_sandbox/sandbox.py

def start(self, container_name: str, *, hooks: LifecycleHooks | None = None) -> None:
    """Start a previously created container via the runtime.

    Fires *hooks.post_start* after a successful start.
    """
    self._runtime.container(container_name).start()
    if hooks and hooks.post_start:
        hooks.post_start()

copy_to(container_name, src, dest)

Copy a host path into a stopped container via the runtime.

Source code in src/terok_sandbox/sandbox.py

def copy_to(self, container_name: str, src: Path, dest: str) -> None:
    """Copy a host path into a stopped container via the runtime."""
    self._runtime.container(container_name).copy_in(src, dest)

stream_logs(container, *, timeout=None, ready_check=None)

Stream container logs until ready_check matches or timeout.

Source code in src/terok_sandbox/sandbox.py

def stream_logs(
    self,
    container: str,
    *,
    timeout: float | None = None,
    ready_check: Callable[[str], bool] | None = None,
) -> bool:
    """Stream container logs until *ready_check* matches or timeout."""
    check = ready_check or (lambda line: READY_MARKER in line)
    return self._runtime.container(container).stream_initial_logs(check, timeout)

wait_for_exit(container, timeout=None)

Block until container exits; return its exit code.

Source code in src/terok_sandbox/sandbox.py

def wait_for_exit(self, container: str, timeout: float | None = None) -> int:
    """Block until *container* exits; return its exit code."""
    return self._runtime.container(container).wait(timeout)

stop(containers)

Best-effort stop and remove containers.

Returns one ContainerRemoveResult per entry.

Source code in src/terok_sandbox/sandbox.py

def stop(self, containers: list[str]) -> list[ContainerRemoveResult]:
    """Best-effort stop and remove *containers*.

    Returns one [`ContainerRemoveResult`][terok_sandbox.runtime.ContainerRemoveResult] per entry.
    """
    handles = [self._runtime.container(name) for name in containers]
    return self._runtime.force_remove(handles)

task_state_dir(container)

Per-container state directory used by the launch / cleanup verbs.

The path is consumed by the launch module: compose writes the plan + readiness markers under it, and launch.cleanup removes it on teardown. The facade owns the derivation — state_dir / "sandbox" / "runs" / {container} — so the runs subtree layout has a single canonical owner.

Source code in src/terok_sandbox/sandbox.py

def task_state_dir(self, container: str) -> Path:
    """Per-container state directory used by the launch / cleanup verbs.

    The path is consumed by the
    [`launch`][terok_sandbox.launch] module: ``compose`` writes
    the plan + readiness markers under it, and
    [`launch.cleanup`][terok_sandbox.launch.cleanup] removes it on
    teardown.  The facade owns the *derivation* — ``state_dir /
    "sandbox" / "runs" / {container}`` — so the runs subtree
    layout has a single canonical owner.
    """
    return self._cfg.state_dir / "sandbox" / "runs" / container

init_ssh(scope)

Create an SSH manager for scope that owns its own CredentialDB.

Callers receive an SSHManager whose DB connection is opened against SandboxConfig.db_path. Use it as a context manager (with sandbox.init_ssh(scope) as m: ...) or call SSHManager.close when done.

Source code in src/terok_sandbox/sandbox.py

def init_ssh(self, scope: str) -> SSHManager:
    """Create an SSH manager for *scope* that owns its own ``CredentialDB``.

    Callers receive an ``SSHManager`` whose DB connection is opened
    against [`SandboxConfig.db_path`][terok_sandbox.SandboxConfig.db_path].  Use it as a context
    manager (``with sandbox.init_ssh(scope) as m: ...``) or call
    [`SSHManager.close`][terok_sandbox.SSHManager.close] when done.
    """
    from .vault.ssh.manager import SSHManager

    # Library code never prompts: a locked vault raises rather than
    # spinning up a prompt_toolkit prompt (which cannot own a running
    # event loop).  The frontend unlocks before calling in.
    return SSHManager.open_for_config(scope=scope, cfg=self._cfg, prompt_on_tty=False)