Vault commands

vault_commands

Executor-level vault helpers: route generation + credential-leak scan.

The vault is served per container: the supervisor spawns on container start via the terok-sandbox OCI hook and reads the per-container sidecar to bind its proxy. Sandbox owns the unlock / lock / passphrase verbs (passphrase-tier CRUD on the DB).

What lives here:

• routes — regenerate routes.json from the YAML agent roster.
• clean — remove leaked credential files from shared config mounts.
• scan_leaked_credentials + the _BENIGN_CREDENTIAL_CHECKS registry of per-provider recognizers for credential files that are legitimately non-empty — primitives the scan + clean verbs share.

Both verbs operate on host-side files only.

SANDBOX_TREE = _build_sandbox_tree() module-attribute

SANDBOX_GROUP = CommandDef(name='sandbox', help='Sandbox subsystem (full deep tree — same verbs as terok-sandbox)', children=(SANDBOX_TREE.roots)) module-attribute

VAULT_GROUP = SANDBOX_TREE.find_at(('vault',)) module-attribute

VAULT_COMMANDS = (VAULT_GROUP,) module-attribute

scan_leaked_credentials(mounts_base)

Return (provider, host_path) for credential files found in shared mounts.

When the vault is active, real secrets should only live in the vault's sqlite DB — not in the shared config directories that get mounted into containers. This function checks each routed provider's mount for credential files that would leak real tokens alongside phantom ones.

Non-empty files recognised as benign by the provider's _BENIGN_CREDENTIAL_CHECKS entry — terok-injected phantoms, or glab's settings-only config.yml — are skipped.

A credential file that does not exist is a definitive clean result, not a skipped check: an agent that ships in the image but was never authenticated on this host has nothing to leak. Only genuine read failures (permissions, I/O errors) warrant the skip warning.

Symlinks are rejected to prevent a container from tricking the scan into reading arbitrary host files via a crafted symlink in the shared mount.

Source code in src/terok_executor/credentials/vault_commands.py

def scan_leaked_credentials(mounts_base: Path) -> list[tuple[str, Path]]:
    """Return ``(provider, host_path)`` for credential files found in shared mounts.

    When the vault is active, real secrets should only live in the
    vault's sqlite DB — not in the shared config directories that get mounted
    into containers.  This function checks each routed provider's mount for
    credential files that would leak real tokens alongside phantom ones.

    Non-empty files recognised as benign by the provider's
    `_BENIGN_CREDENTIAL_CHECKS` entry — terok-injected phantoms, or
    glab's settings-only ``config.yml`` — are skipped.

    A credential file that does not exist is a definitive clean result,
    not a skipped check: an agent that ships in the image but was never
    authenticated on this host has nothing to leak.  Only genuine read
    failures (permissions, I/O errors) warrant the skip warning.

    Symlinks are rejected to prevent a container from tricking the scan into
    reading arbitrary host files via a crafted symlink in the shared mount.
    """
    import stat

    from terok_executor.roster import AgentRoster

    roster = AgentRoster.shared()
    base_resolved = mounts_base.resolve(strict=False)
    leaked: list[tuple[str, Path]] = []
    # Iterate the shared mounts directly: each MountDef already pairs the
    # agent's auth dir with the credential file its binding declares, so we
    # don't have to re-join provider-keyed routes against agent-keyed auth
    # providers.  ``mount.provider`` is the owning agent name (the label the
    # operator sees); mounts without a credential file (opencode state dirs,
    # explicit ``mounts:`` blocks) carry an empty string and are skipped.
    for mount in roster.mounts:
        if not mount.credential_file:
            continue
        try:
            path = mounts_base / mount.host_dir / mount.credential_file
            # lstat: do not follow symlinks — reject them outright
            st = path.lstat()
            if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode):
                continue
            # Ensure resolved path stays within the mounts base
            if base_resolved not in path.resolve(strict=True).parents:
                continue
            is_benign = _BENIGN_CREDENTIAL_CHECKS.get(mount.provider)
            if st.st_size > 0 and not (is_benign and is_benign(path)):
                leaked.append((mount.provider, path))
        except FileNotFoundError:
            # No credential file at all — the agent was never authenticated
            # on this host, so there is nothing to scan and nothing to leak.
            continue
        except (OSError, TypeError) as exc:
            # Silently skipping turns a real leak into a no-result: the
            # operator would believe the scan was clean.  Surface a
            # warning so it's obvious which mount was not checked and why;
            # the loop continues so other mounts still get scanned.
            print(
                f"Warning [vault]: credential leak scan skipped {mount.provider!r}: {exc}",
                file=sys.stderr,
            )
            continue
    return leaked