signer

signer

SSH signer — signs with vault-stored private keys on behalf of clients.

Implements the SSH agent protocol_ in two deployment flavours that share one connection handler:

• Container-facing (start_ssh_signer) — TCP or Unix, guarded by a phantom-token handshake so a compromised container can't impersonate another. Scope is resolved from the token.
• Host-local, per-scope (start_ssh_signer_local) — one Unix socket per scope at mode 0600. Scope is fixed at bind time and the UID-gated filesystem permissions are the whole access control — host processes don't cross a trust boundary.

Private keys live in credentials.db (ssh_keys table) as unencrypted PKCS#8 DER and never touch the filesystem; the _DBKeyCache reloads them only when the DB version counter advances.

.. _SSH agent protocol: https://datatracker.ietf.org/doc/html/draft-miller-ssh-agent

SSH_AGENT_FAILURE = 5 module-attribute

SSH_AGENTC_REQUEST_IDENTITIES = 11 module-attribute

SSH_AGENT_IDENTITIES_ANSWER = 12 module-attribute

SSH_AGENTC_SIGN_REQUEST = 13 module-attribute

SSH_AGENT_SIGN_RESPONSE = 14 module-attribute

SSH_AGENT_RSA_SHA2_256 = 2 module-attribute

SSH_AGENT_RSA_SHA2_512 = 4 module-attribute

start_ssh_signer(db_path, host=None, port=None, socket_path=None) async

Start the container-facing SSH signer (token-gated).

Parameters:

Name	Type	Description	Default
db_path	str	Path to the vault sqlite3 database (phantom tokens + SSH keys).	required
host	str | None	Bind address for TCP (typically "127.0.0.1").	None
port	int | None	TCP port to listen on.	None
socket_path	str | None	Unix socket path to listen on.	None

Returns:

Type	Description
Server	The running asyncio.Server — caller is responsible for closing it.

Raises:

Type	Description
ValueError	If neither TCP (host+port) nor socket_path is provided.

Source code in src/terok_sandbox/vault/ssh/signer.py

async def start_ssh_signer(
    db_path: str,
    host: str | None = None,
    port: int | None = None,
    socket_path: str | None = None,
) -> asyncio.Server:
    """Start the container-facing SSH signer (token-gated).

    Args:
        db_path: Path to the vault sqlite3 database (phantom tokens + SSH keys).
        host: Bind address for TCP (typically ``"127.0.0.1"``).
        port: TCP port to listen on.
        socket_path: Unix socket path to listen on.

    Returns:
        The running [`asyncio.Server`][asyncio.Server] — caller is responsible for closing it.

    Raises:
        ValueError: If neither TCP (host+port) nor socket_path is provided.
    """
    from ..daemon.token_broker import _TokenDB

    token_db = _TokenDB(db_path)
    key_cache = _DBKeyCache(token_db)

    async def _on_connect(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        """Handle an incoming SSH agent connection (token-gated)."""
        await _handle_container_connection(reader, writer, token_db, key_cache)

    if socket_path:
        server = await _bind_hardened_unix(socket_path, _on_connect)
        _logger.info("SSH signer listening on %s", socket_path)
    elif host is not None and port is not None:
        server = await asyncio.start_server(_on_connect, host, port)
        _logger.info("SSH signer listening on %s:%d", host, port)
    else:
        raise ValueError("Either socket_path or host+port must be provided")
    return server

start_ssh_signer_local(*, scope, socket_path, db_path, passphrase=None) async

Start a host-local SSH signer bound to a single scope.

The returned server listens on a mode-0600 Unix socket; same-UID filesystem permissions are the whole access control. No token handshake — every accepted connection immediately enters the agent message loop with scope as its fixed identity source.

A pre-resolved passphrase opens the vault directly, keeping the bind off any slow keystore tier; None walks the resolution chain.

Source code in src/terok_sandbox/vault/ssh/signer.py

async def start_ssh_signer_local(
    *, scope: str, socket_path: Path, db_path: str, passphrase: str | None = None
) -> asyncio.Server:
    """Start a host-local SSH signer bound to a single scope.

    The returned server listens on a mode-0600 Unix socket; same-UID
    filesystem permissions are the whole access control.  No token
    handshake — every accepted connection immediately enters the agent
    message loop with *scope* as its fixed identity source.

    A pre-resolved *passphrase* opens the vault directly, keeping the bind
    off any slow keystore tier; ``None`` walks the resolution chain.
    """
    from ..daemon.token_broker import _TokenDB

    token_db = _TokenDB(db_path, passphrase=passphrase)
    key_cache = _DBKeyCache(token_db)

    async def _on_connect(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        """Handle an incoming host-local SSH agent connection."""
        await _handle_local_connection(reader, writer, scope, key_cache)

    server = await _bind_hardened_unix(str(socket_path), _on_connect, mode=_LOCAL_SOCKET_MODE)
    _logger.info("SSH signer listening on %s for scope %r", socket_path, scope)
    return server