server

server

Credential proxy — HTTP-over-Unix-socket reverse proxy with secret injection.

This module has zero terok imports. It is a self-contained security component: a standalone daemon that listens on a Unix socket, validates phantom tokens against a sqlite3 database, injects real credentials from the same database, and forwards requests to upstream API servers.

Task containers see only phantom API keys (worthless outside the proxy). Real secrets never enter the container filesystem or environment.

Startup::

terok-credential-proxy \
    --socket-path /run/terok/credential-proxy.sock \
    --db-path ~/.local/share/terok/proxy/credentials.db \
    --routes-file ~/.local/share/terok/proxy/routes.json

Route config (JSON, generated by terok-agent from YAML registry)::

{
  "claude": {
    "upstream": "https://api.anthropic.com",
    "auth_header": "Authorization",
    "auth_prefix": "Bearer "
  }
}

main()

Parse CLI args and run the credential proxy.

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

def main() -> None:
    """Parse CLI args and run the credential proxy."""
    import asyncio

    parser = argparse.ArgumentParser(
        prog="terok-credential-proxy",
        description="Credential injection reverse proxy for terok containers",
    )
    parser.add_argument("--socket-path", required=True, help="Unix socket path to listen on")
    parser.add_argument("--db-path", required=True, help="Path to the credential sqlite3 database")
    parser.add_argument("--routes-file", required=True, help="Path to the route config JSON")

    def _tcp_port(value: str) -> int:
        port = int(value)
        if not 1 <= port <= 65535:
            raise argparse.ArgumentTypeError("--port must be between 1 and 65535")
        return port

    parser.add_argument(
        "--port",
        type=_tcp_port,
        default=None,
        help="TCP port for container access (in addition to the Unix socket)",
    )
    parser.add_argument(
        "--ssh-agent-port",
        type=_tcp_port,
        default=None,
        help="TCP port for the SSH agent proxy (signs with host-side keys)",
    )
    parser.add_argument(
        "--ssh-keys-file",
        default=None,
        help="Path to ssh-keys.json mapping project IDs to key file paths",
    )
    parser.add_argument(
        "--pid-file", default=None, help="Write PID to this file (for lifecycle management)"
    )
    parser.add_argument("--log-level", default="INFO", help="Logging level (default: INFO)")
    parser.add_argument("--log-file", default=None, help="Append log output to this file")
    args = parser.parse_args()

    if bool(args.ssh_agent_port) != bool(args.ssh_keys_file):
        parser.error("--ssh-agent-port and --ssh-keys-file must be provided together")

    log_level = getattr(logging, args.log_level.upper(), logging.INFO)
    log_fmt = "%(asctime)s %(name)s %(levelname)s %(message)s"
    handlers: list[logging.Handler] = []
    if args.log_file:
        handlers.append(logging.FileHandler(args.log_file))
    else:
        handlers.append(logging.StreamHandler())
    logging.basicConfig(level=log_level, format=log_fmt, handlers=handlers)

    if args.pid_file:
        import os

        Path(args.pid_file).write_text(str(os.getpid()), encoding="utf-8")

    app = _build_app(args.db_path, args.routes_file)

    # Graceful shutdown on SIGTERM
    def _handle_sigterm(*_: object) -> None:
        _logger.info("SIGTERM received, shutting down")
        raise SystemExit(0)

    signal.signal(signal.SIGTERM, _handle_sigterm)

    try:
        asyncio.run(
            _run_multi(
                app,
                sock_path=args.socket_path,
                port=args.port,
                ssh_agent_port=args.ssh_agent_port,
                ssh_keys_file=args.ssh_keys_file,
                db_path=args.db_path,
            )
        )
    except (KeyboardInterrupt, SystemExit):
        pass