ssh_agent

ssh_agent

SSH agent proxy — signs with host-side private keys on behalf of containers.

Implements the SSH agent protocol_ over TCP with a phantom-token handshake. Containers connect via a socat bridge (UNIX-LISTEN → TCP) and set SSH_AUTH_SOCK to the local Unix socket. Private keys never enter the container — the proxy reads them from the host filesystem.

Like :mod:server, this module has zero terok imports. It is a self-contained security component that reads phantom tokens from the same sqlite3 database and key paths from a JSON sidecar file.

Wire format (per draft-miller-ssh-agent_):

[4-byte big-endian length][1-byte message type][payload]

Custom handshake (first bytes on each TCP connection):

[4-byte big-endian length][phantom-token UTF-8 bytes]

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

start_ssh_agent_server(db_path, keys_file, host, port) async

Start the SSH agent TCP server.

Parameters:

Name	Type	Description	Default
db_path	str	Path to the credential proxy sqlite3 database (for phantom token lookups).	required
keys_file	str	Path to ssh-keys.json mapping project IDs to key file paths.	required
host	str	Bind address (typically "127.0.0.1").	required
port	int	TCP port to listen on.	required

Returns:

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

Source code in src/terok_sandbox/credential_proxy/ssh_agent.py

async def start_ssh_agent_server(
    db_path: str, keys_file: str, host: str, port: int
) -> asyncio.Server:
    """Start the SSH agent TCP server.

    Args:
        db_path: Path to the credential proxy sqlite3 database (for phantom token lookups).
        keys_file: Path to ``ssh-keys.json`` mapping project IDs to key file paths.
        host: Bind address (typically ``"127.0.0.1"``).
        port: TCP port to listen on.

    Returns:
        The running :class:`asyncio.Server` — caller is responsible for closing it.
    """
    from .server import _TokenDB

    token_db = _TokenDB(db_path)
    key_cache = _KeyCache(keys_file)

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

    server = await asyncio.start_server(_on_connect, host, port)
    _logger.info("SSH agent proxy listening on %s:%d", host, port)
    return server