From e2a69f801e885a381a62c47cac1f15db0f1db8a3 Mon Sep 17 00:00:00 2001 From: Paulo Alvarado Date: Thu, 20 Aug 2026 17:09:16 +0200 Subject: [PATCH] Add the SSH gateway for docker-sbx hosts Sandboxes have no dialable address: the daemon stops them when idle and owns the data plane. The gateway is the SSH path. A caller connects with the per-host key and the host name as the username; the gateway opens an sbx exec session, which wakes a stopped sandbox and keeps it awake while connected. A shell and command execution only. --- AGENTS.md | 2 + alembic/versions/0003_host_public_key.py | 23 ++ docs/deploy.md | 55 +++- pyproject.toml | 2 + src/gateway/__init__.py | 0 src/gateway/server.py | 173 ++++++++++ src/gateway/settings.py | 38 +++ src/gateway/tests/__init__.py | 0 src/gateway/tests/test_server.py | 298 ++++++++++++++++++ src/hosts/api.py | 8 +- src/hosts/janitor.py | 6 +- src/hosts/models.py | 3 + src/hosts/pool.py | 10 +- src/hosts/service.py | 37 ++- src/hosts/tests/test_gateway_path.py | 113 +++++++ src/http_proxies/api.py | 2 +- src/networking/tailscale.py | 4 +- src/providers/base.py | 59 +++- src/providers/docker_sbx/api.py | 19 -- src/providers/docker_sbx/process.py | 165 ++++++++++ src/providers/docker_sbx/provider.py | 20 +- src/providers/docker_sbx/tests/test_api.py | 44 --- .../docker_sbx/tests/test_process.py | 117 +++++++ .../docker_sbx/tests/test_provider.py | 35 +- uv.lock | 15 + 25 files changed, 1120 insertions(+), 128 deletions(-) create mode 100644 alembic/versions/0003_host_public_key.py create mode 100644 src/gateway/__init__.py create mode 100644 src/gateway/server.py create mode 100644 src/gateway/settings.py create mode 100644 src/gateway/tests/__init__.py create mode 100644 src/gateway/tests/test_server.py create mode 100644 src/hosts/tests/test_gateway_path.py create mode 100644 src/providers/docker_sbx/process.py create mode 100644 src/providers/docker_sbx/tests/test_process.py diff --git a/AGENTS.md b/AGENTS.md index 4cd0388..303b776 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,6 +15,7 @@ It owns: local Docker, Docker Sandboxes) - Tailscale auth key creation, device discovery, and cleanup - SSH host key scanning and `known_hosts` material +- An SSH gateway for hosts of gateway providers (`python -m gateway.server`) - Account-bound exe.dev HTTP proxy resources Periodic maintenance runs as cron jobs: `python -m hosts.janitor` reaps @@ -44,6 +45,7 @@ src/ api/ # FastAPI app and global handlers core/ # Settings, database, exception base hosts/ # Host API, models, schemas, service, janitor, pool, auth + gateway/ # SSH gateway for gateway-provider hosts http_proxies/ # HTTP proxy API, schemas, service, deps providers/ # VM provider ABC, capabilities, registry, adapters networking/ # Network provider framework and Tailscale adapter diff --git a/alembic/versions/0003_host_public_key.py b/alembic/versions/0003_host_public_key.py new file mode 100644 index 0000000..be3d359 --- /dev/null +++ b/alembic/versions/0003_host_public_key.py @@ -0,0 +1,23 @@ +"""per-host public key for the SSH gateway + +Revision ID: 0003_host_public_key +Revises: 0002_host_sizing +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + +revision: str = "0003_host_public_key" +down_revision: str | None = "0002_host_sizing" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column("hosts", sa.Column("public_key", sa.Text(), nullable=False, server_default="")) + + +def downgrade() -> None: + op.drop_column("hosts", "public_key") diff --git a/docs/deploy.md b/docs/deploy.md index f0499c4..d695876 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -155,13 +155,12 @@ docker run --rm --network host \ The daemon reads workspace paths on its own filesystem. Thus the workspace mount must have the same path on the host and in the container. The janitor and pool containers need the same mounts and -variables. Host networking is necessary: the daemon publishes sandbox -SSH ports on the host loopback interface only, and drukbox returns -`127.0.0.1` addresses, the same as the `docker` provider. +variables. -Only this machine can connect to the sandboxes. The key for each host -is the auth boundary. Sandboxes have no `SERVICE_LABEL` tag, because -`sbx create` has no label option. +Callers reach the sandboxes through +[the SSH gateway](#the-ssh-gateway); the provider requires it. The key +for each host is the auth boundary. Sandboxes have no `SERVICE_LABEL` +tag, because `sbx create` has no label option. The template image (`DOCKER_SBX_DEFAULT_IMAGE`, default `ghcr.io/czpython/drukbox/sbx-sandbox:latest`) must start sshd without @@ -187,6 +186,50 @@ cache, and more than 30 seconds at the first pull. Thus a warm pool `DOCKER_SBX_CPUS` and `DOCKER_SBX_MEMORY` sizes. Without them, the daemon gives one sandbox all host CPUs and half of the host memory. +## The SSH gateway + +The gateway gives remote callers SSH access to the hosts of gateway +providers such as `docker-sbx`, whose sandboxes have no dialable sshd +of their own. Callers connect with normal SSH: + +```bash +ssh -p 2222 -i @ +``` + +The gateway authenticates the key against the host's stored public key, +and the username must name the same host. It then opens a session +through the provider — `sbx exec` for `docker-sbx`. The daemon stops an +idle sandbox; a connection through the gateway wakes it (approximately +6 seconds) and keeps it awake while connected. The first data can +therefore come after a short delay. + +The gateway serves an interactive shell and command execution. It +refuses SFTP, scp, and port forwarding. + +The gateway is a requirement for gateway providers: `POST /hosts` for +`docker-sbx` fails without `GATEWAY_SSH_HOST`. Set it to the address +callers use. The response then carries the gateway coordinates: +`external_ssh_host` is the gateway, `external_ssh_port` is the gateway +port, and `ssh_username` is the host name. `known_hosts` carries the +gateway's host key. Hosts of the other providers are not affected. + +Run the gateway on the machine that runs sandboxd, as the same user, and +run the migrations first: + +```bash +uv run python -m gateway.server +``` + +A systemd unit follows the same pattern as the API service. The gateway +reads the same `drukbox.env` and connects to the same database. + +| Variable | Default | Purpose | +| --- | --- | --- | +| `GATEWAY_SSH_HOST` | — (required) | Address of the gateway. Gateway-provider hosts advertise it; creation fails without it. | +| `GATEWAY_SSH_PORT` | `2222` | Port the gateway listens on and advertises. | +| `GATEWAY_BIND_HOST` | `0.0.0.0` | Interface the gateway server binds. | +| `GATEWAY_HOST_KEY_PATH` | `~/.drukbox/gateway_host_key` | Private host key. The server makes one at start when the file does not exist. | + ## Choose a networking mode `TAILSCALE_ENABLED=false` (default): callers reach sandboxes over the diff --git a/pyproject.toml b/pyproject.toml index d12e7ee..500cce2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,6 +7,7 @@ packages = [ "src/api", "src/core", "src/diagnostics", + "src/gateway", "src/hosts", "src/http_proxies", "src/networking", @@ -50,6 +51,7 @@ classifiers = [ dependencies = [ "aiosqlite>=0.20", "alembic>=1.14", + "asyncssh>=2.14", "cryptography>=43", "fastapi>=0.115", "greenlet>=3.0", diff --git a/src/gateway/__init__.py b/src/gateway/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/gateway/server.py b/src/gateway/server.py new file mode 100644 index 0000000..1738f14 --- /dev/null +++ b/src/gateway/server.py @@ -0,0 +1,173 @@ +import asyncio +import contextlib +import logging + +import asyncssh +from sqlalchemy import select + +from core.database import async_session_factory +from gateway.settings import GatewaySettings +from hosts.models import Host, HostStatus +from providers.base import SandboxProcess, TerminalSize +from providers.exceptions import ProviderError +from providers.registry import get_vm_provider + +logger = logging.getLogger(__name__) + +_RECEIVE_CHUNK_BYTES = 32768 + + +class GatewayConnection(asyncssh.SSHServer): + """One caller connection. The key is the identity; the username must name + the same host, so one leaked key cannot probe other host names.""" + + def __init__(self) -> None: + self.host: Host | None = None + + def begin_auth(self, username: str) -> bool: + return True + + def public_key_auth_supported(self) -> bool: + return True + + async def validate_public_key(self, username: str, key: asyncssh.SSHKey) -> bool: + async with async_session_factory() as session: + result = await session.execute( + select(Host).where(Host.name == username, Host.status == HostStatus.ACTIVE.value) + ) + host = result.scalar_one_or_none() + if host and host.public_key: + try: + stored = asyncssh.import_public_key(host.public_key) + except asyncssh.KeyImportError: + logger.warning("gateway: host %s has an unreadable public key", host.name) + return False + if stored.public_data == key.public_data: + self.host = host + return True + logger.info("gateway: rejected key for username=%r", username) + return False + + +async def _bridge(process: asyncssh.SSHServerProcess) -> None: + connection = process.channel.get_connection() + server = connection.get_owner() + assert isinstance(server, GatewayConnection) and server.host is not None + host = server.host + + terminal: TerminalSize | None = None + if process.get_terminal_type(): + columns, rows, _, _ = process.get_terminal_size() + terminal = TerminalSize(columns=columns, rows=rows) + + process_class = get_vm_provider(host.provider).gateway_process_class + if not process_class: + logger.warning("gateway: host %s has a direct-dial provider", host.name) + process.stderr.write(b"cannot open a session for this host\n") + process.exit(255) + return + try: + sandbox_process = await process_class.open( + host.name, command=process.command, terminal=terminal + ) + except ProviderError as error: + logger.warning("gateway: open failed for host=%s: %s", host.name, error) + process.stderr.write(b"cannot open a session for this host\n") + process.exit(255) + return + + logger.info( + "gateway: session open host=%s command=%r terminal=%s", + host.name, + process.command, + terminal, + ) + # The input pump can outlive the session (a caller who sends no EOF), + # thus it gets a cancel, not a wait. + input_pump = asyncio.create_task(_pump_channel_to_process(process, sandbox_process)) + stderr_pump = asyncio.create_task(_pump_stderr_to_channel(process, sandbox_process)) + try: + while data := await sandbox_process.receive(_RECEIVE_CHUNK_BYTES): + process.stdout.write(data) + await process.stdout.drain() + await stderr_pump + process.exit(await sandbox_process.wait()) + finally: + input_pump.cancel() + stderr_pump.cancel() + with contextlib.suppress(asyncio.CancelledError): + await input_pump + with contextlib.suppress(asyncio.CancelledError): + await stderr_pump + await sandbox_process.aclose() + logger.info("gateway: session closed host=%s", host.name) + + +async def _pump_stderr_to_channel( + process: asyncssh.SSHServerProcess, + sandbox_process: SandboxProcess, +) -> None: + while data := await sandbox_process.receive_stderr(_RECEIVE_CHUNK_BYTES): + process.stderr.write(data) + await process.stderr.drain() + + +async def _pump_channel_to_process( + process: asyncssh.SSHServerProcess, + sandbox_process: SandboxProcess, +) -> None: + while True: + try: + data = await process.stdin.read(_RECEIVE_CHUNK_BYTES) + except asyncssh.TerminalSizeChanged as change: + sandbox_process.resize(TerminalSize(columns=change.width, rows=change.height)) + continue + except asyncssh.BreakReceived: + continue + if not data: + sandbox_process.send_eof() + return + sandbox_process.send(data) + + +def _load_host_key(settings: GatewaySettings) -> asyncssh.SSHKey: + path = settings.host_key_path + if path.exists(): + return asyncssh.read_private_key(path) + path.parent.mkdir(parents=True, exist_ok=True) + key = asyncssh.generate_private_key("ssh-ed25519") + path.touch(mode=0o600) + path.write_bytes(key.export_private_key("openssh")) + logger.info("gateway: made a new host key at %s", path) + return key + + +async def start(settings: GatewaySettings) -> asyncssh.SSHAcceptor: + server = await asyncssh.listen( + host=settings.bind_host, + port=settings.ssh_port, + server_host_keys=[_load_host_key(settings)], + server_factory=GatewayConnection, + process_factory=_bridge, + encoding=None, + allow_scp=False, + sftp_factory=None, + agent_forwarding=False, + x11_forwarding=False, + ) + logger.info("gateway: listening on %s:%d", settings.bind_host, settings.ssh_port) + return server + + +async def serve() -> None: + server = await start(GatewaySettings()) + await server.wait_closed() + + +if __name__ == "__main__": + # Service entry point: `python -m gateway.server`. + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + asyncio.run(serve()) diff --git a/src/gateway/settings.py b/src/gateway/settings.py new file mode 100644 index 0000000..e7ce423 --- /dev/null +++ b/src/gateway/settings.py @@ -0,0 +1,38 @@ +from pathlib import Path + +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class GatewaySettings(BaseSettings): + """SSH gateway configuration.""" + + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + env_prefix="GATEWAY_", + extra="ignore", + ) + + ssh_host: str = Field( + default="", + description=( + "Address of the gateway. Gateway-provider hosts advertise it; " + "the service refuses to provision them without it." + ), + ) + ssh_port: int = Field( + default=2222, + description="Port the gateway listens on and advertises.", + ) + bind_host: str = Field( + default="0.0.0.0", + description="Interface the gateway server binds.", + ) + host_key_path: Path = Field( + default_factory=lambda: Path.home() / ".drukbox" / "gateway_host_key", + description=( + "Private host key of the gateway server. The server makes one " + "at start when the file does not exist." + ), + ) diff --git a/src/gateway/tests/__init__.py b/src/gateway/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/gateway/tests/test_server.py b/src/gateway/tests/test_server.py new file mode 100644 index 0000000..31daed5 --- /dev/null +++ b/src/gateway/tests/test_server.py @@ -0,0 +1,298 @@ +import asyncio +from datetime import UTC, datetime +from types import SimpleNamespace +from typing import ClassVar + +import asyncssh +import pytest + +from core.database import async_session_factory +from gateway import server as gateway_server +from gateway.settings import GatewaySettings +from hosts.models import Host, HostStatus +from providers.base import SandboxProcess, TerminalSize +from providers.exceptions import ProviderTransportError + + +class FakeProcess(SandboxProcess): + """Consumes caller input until EOF, echoes one payload, exits with 7.""" + + opened: ClassVar[list["FakeProcess"]] = [] + + @classmethod + async def open(cls, name, *, command, terminal): + fake = cls(command=command, terminal=terminal) + cls.opened.append(fake) + return fake + + def __init__(self, *, command: str | None, terminal: TerminalSize | None) -> None: + self.command = command + self.terminal = terminal + self.sent = bytearray() + self._caller_done = asyncio.Event() + self._announced = False + + async def receive(self, max_bytes: int) -> bytes: + if self._announced: + await self._caller_done.wait() + return b"" + self._announced = True + return f"ran:{self.command or 'shell'}".encode() + + async def receive_stderr(self, max_bytes: int) -> bytes: + return b"" + + def send(self, data: bytes) -> None: + self.sent.extend(data) + # A newline ends the fake, the way "exit" ends a shell. PTY callers + # end sessions this way; they send no stdin EOF. + if b"\n" in data: + self._caller_done.set() + + def send_eof(self) -> None: + self._caller_done.set() + + def resize(self, size: TerminalSize) -> None: + self.terminal = size + + async def wait(self) -> int: + return 7 + + async def aclose(self) -> None: + self._caller_done.set() + + +async def _insert_active_host(name: str, public_key: str, status: str = "active") -> None: + now = datetime.now(UTC) + async with async_session_factory() as session: + session.add( + Host( + name=name, + provider="docker-sbx", + image="template", + status=status, + public_key=public_key, + created_at=now, + updated_at=now, + ) + ) + await session.commit() + + +@pytest.fixture +def gateway_settings(tmp_path): + return GatewaySettings( + ssh_host="127.0.0.1", + ssh_port=0, + bind_host="127.0.0.1", + host_key_path=tmp_path / "gateway_host_key", + ) + + +@pytest.fixture +def fake_provider(monkeypatch): + FakeProcess.opened.clear() + provider = SimpleNamespace(gateway_process_class=FakeProcess) + monkeypatch.setattr(gateway_server, "get_vm_provider", lambda name: provider) + return FakeProcess + + +async def test_gateway_runs_a_command_and_returns_the_exit_status(gateway_settings, fake_provider): + caller_key = asyncssh.generate_private_key("ssh-ed25519") + await _insert_active_host("sb-gwtest", caller_key.export_public_key().decode()) + + server = await gateway_server.start(gateway_settings) + try: + async with asyncssh.connect( + "127.0.0.1", + server.get_port(), + username="sb-gwtest", + client_keys=[caller_key], + known_hosts=None, + ) as connection: + result = await connection.run("uname -a", input="fed-to-session") + finally: + server.close() + + assert result.exit_status == 7 + assert result.stdout == "ran:uname -a" + assert bytes(fake_provider.opened[0].sent) == b"fed-to-session" + + +async def test_gateway_rejects_a_key_that_is_not_the_hosts_key(gateway_settings, fake_provider): + host_key = asyncssh.generate_private_key("ssh-ed25519") + await _insert_active_host("sb-gwtest", host_key.export_public_key().decode()) + attacker_key = asyncssh.generate_private_key("ssh-ed25519") + + server = await gateway_server.start(gateway_settings) + try: + with pytest.raises(asyncssh.PermissionDenied): + await asyncssh.connect( + "127.0.0.1", + server.get_port(), + username="sb-gwtest", + client_keys=[attacker_key], + known_hosts=None, + ) + finally: + server.close() + assert fake_provider.opened == [] + + +async def test_gateway_rejects_the_right_key_under_another_hosts_name( + gateway_settings, fake_provider +): + # One leaked key must not open sessions on other hosts: the username and + # the key must name the same host. + caller_key = asyncssh.generate_private_key("ssh-ed25519") + other_key = asyncssh.generate_private_key("ssh-ed25519") + await _insert_active_host("sb-mine", caller_key.export_public_key().decode()) + await _insert_active_host("sb-other", other_key.export_public_key().decode()) + + server = await gateway_server.start(gateway_settings) + try: + with pytest.raises(asyncssh.PermissionDenied): + await asyncssh.connect( + "127.0.0.1", + server.get_port(), + username="sb-other", + client_keys=[caller_key], + known_hosts=None, + ) + finally: + server.close() + + +async def test_gateway_rejects_hosts_that_are_not_active(gateway_settings, fake_provider): + caller_key = asyncssh.generate_private_key("ssh-ed25519") + await _insert_active_host( + "sb-gone", + caller_key.export_public_key().decode(), + status=HostStatus.ERROR.value, + ) + + server = await gateway_server.start(gateway_settings) + try: + with pytest.raises(asyncssh.PermissionDenied): + await asyncssh.connect( + "127.0.0.1", + server.get_port(), + username="sb-gone", + client_keys=[caller_key], + known_hosts=None, + ) + finally: + server.close() + + +async def test_gateway_passes_the_callers_terminal_to_the_session(gateway_settings, fake_provider): + caller_key = asyncssh.generate_private_key("ssh-ed25519") + await _insert_active_host("sb-pty", caller_key.export_public_key().decode()) + + server = await gateway_server.start(gateway_settings) + try: + async with asyncssh.connect( + "127.0.0.1", + server.get_port(), + username="sb-pty", + client_keys=[caller_key], + known_hosts=None, + ) as connection: + result = await connection.run( + term_type="xterm", term_size=(121, 43), command=None, input="exit\n" + ) + finally: + server.close() + + assert result.exit_status == 7 + session = fake_provider.opened[0] + assert session.command is None + assert session.terminal == TerminalSize(columns=121, rows=43) + + +async def test_gateway_routes_the_sessions_error_stream_to_ssh_stderr( + gateway_settings, monkeypatch +): + class NoisyProcess(FakeProcess): + opened: ClassVar[list[FakeProcess]] = [] + + def __init__(self, **kwargs) -> None: + super().__init__(**kwargs) + self._warned = False + + async def receive_stderr(self, max_bytes: int) -> bytes: + if self._warned: + return b"" + self._warned = True + return b"sandbox woke up\n" + + provider = SimpleNamespace(gateway_process_class=NoisyProcess) + monkeypatch.setattr(gateway_server, "get_vm_provider", lambda name: provider) + caller_key = asyncssh.generate_private_key("ssh-ed25519") + await _insert_active_host("sb-noisy", caller_key.export_public_key().decode()) + + server = await gateway_server.start(gateway_settings) + try: + async with asyncssh.connect( + "127.0.0.1", + server.get_port(), + username="sb-noisy", + client_keys=[caller_key], + known_hosts=None, + ) as connection: + result = await connection.run("true", input="done\n") + finally: + server.close() + + # Wake banners and warnings must not pollute the command's output. + assert result.stdout == "ran:true" + assert "sandbox woke up" in str(result.stderr) + + +async def test_gateway_reports_a_failed_dial_and_exits(gateway_settings, monkeypatch): + # A stopped daemon or a vanished sandbox must end the connection with a + # clear message and status 255, not a closed channel. + class FailingProcess(FakeProcess): + @classmethod + async def open(cls, name, *, command, terminal): + raise ProviderTransportError("daemon unavailable") + + provider = SimpleNamespace(gateway_process_class=FailingProcess) + monkeypatch.setattr(gateway_server, "get_vm_provider", lambda name: provider) + caller_key = asyncssh.generate_private_key("ssh-ed25519") + await _insert_active_host("sb-nodial", caller_key.export_public_key().decode()) + + server = await gateway_server.start(gateway_settings) + try: + async with asyncssh.connect( + "127.0.0.1", + server.get_port(), + username="sb-nodial", + client_keys=[caller_key], + known_hosts=None, + ) as connection: + result = await connection.run("uname -a") + finally: + server.close() + + assert result.exit_status == 255 + assert "cannot open a session" in str(result.stderr) + + +async def test_gateway_refuses_sftp(gateway_settings, fake_provider): + caller_key = asyncssh.generate_private_key("ssh-ed25519") + await _insert_active_host("sb-nosftp", caller_key.export_public_key().decode()) + + server = await gateway_server.start(gateway_settings) + try: + async with asyncssh.connect( + "127.0.0.1", + server.get_port(), + username="sb-nosftp", + client_keys=[caller_key], + known_hosts=None, + ) as connection: + with pytest.raises((asyncssh.SFTPError, asyncssh.ChannelOpenError)): + await connection.start_sftp_client() + finally: + server.close() diff --git a/src/hosts/api.py b/src/hosts/api.py index 8643520..f665cee 100644 --- a/src/hosts/api.py +++ b/src/hosts/api.py @@ -14,7 +14,7 @@ from networking.tailscale import NetworkError from providers.exceptions import ProviderError, UnknownProviderError, UnsupportedSizingError -log = logging.getLogger(__name__) +logger = logging.getLogger(__name__) router = APIRouter(prefix="/hosts", tags=["hosts"]) @@ -63,7 +63,7 @@ async def create_host( except (UnknownProviderError, UnsupportedSizingError) as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except SQLAlchemyError as exc: - log.exception("unexpected database error during host provisioning") + logger.exception("unexpected database error during host provisioning") raise HTTPException( status_code=503, detail="host provisioning could not be completed", @@ -113,9 +113,9 @@ async def delete_host(host_id: uuid.UUID, service: HostServiceDep) -> Response: try: await service.delete_host(host_id) except (ProviderError, NetworkError) as exc: - log.exception("unexpected error deleting sandbox host") + logger.exception("unexpected error deleting sandbox host") raise HostTeardownError("host teardown could not be completed") from exc except SQLAlchemyError as exc: - log.exception("unexpected database error during host teardown") + logger.exception("unexpected database error during host teardown") raise HostTeardownError("host teardown could not be completed") from exc return Response(status_code=status.HTTP_204_NO_CONTENT) diff --git a/src/hosts/janitor.py b/src/hosts/janitor.py index 8af560c..98649de 100644 --- a/src/hosts/janitor.py +++ b/src/hosts/janitor.py @@ -10,7 +10,7 @@ from hosts.service import HostService, utc_now from networking.tailscale import Tailscale -log = logging.getLogger(__name__) +logger = logging.getLogger(__name__) async def reap_expired_hosts() -> list[uuid.UUID]: @@ -48,13 +48,13 @@ async def reap_expired_hosts() -> list[uuid.UUID]: # Another janitor cycle, or an explicit DELETE, already removed it. continue except Exception: - log.exception("janitor: failed to reap expired host: host_id=%s", host_id) + logger.exception("janitor: failed to reap expired host: host_id=%s", host_id) continue else: if not deleted: # Renewed between selection and the locked delete — spared. continue - log.info("janitor: reaped expired host: host_id=%s", host_id) + logger.info("janitor: reaped expired host: host_id=%s", host_id) reaped.append(host_id) finally: if tailscale is not None: diff --git a/src/hosts/models.py b/src/hosts/models.py index 5bf6329..7be3aa0 100644 --- a/src/hosts/models.py +++ b/src/hosts/models.py @@ -87,6 +87,9 @@ class Host(Base): external_ssh_host: Mapped[str] = mapped_column(Text, default="") external_ssh_port: Mapped[int] = mapped_column(default=22) ssh_username: Mapped[str] = mapped_column(Text, default="") + # The public half of the per-host keypair. The gateway authenticates + # callers against it. The private half is returned once and never stored. + public_key: Mapped[str] = mapped_column(Text, default="") internal_ssh_host: Mapped[str | None] = mapped_column(Text, nullable=True, default=None) known_hosts: Mapped[str] = mapped_column(Text, default="") tailscale_device_id: Mapped[str | None] = mapped_column(Text, nullable=True, default=None) diff --git a/src/hosts/pool.py b/src/hosts/pool.py index 451bb93..7947911 100644 --- a/src/hosts/pool.py +++ b/src/hosts/pool.py @@ -12,7 +12,7 @@ from hosts.service import HostService, utc_now from networking.tailscale import Tailscale -log = logging.getLogger(__name__) +logger = logging.getLogger(__name__) # Bail out after this many consecutive create failures (broker outage scenario) # rather than logging the same failure once per deficit slot per tick. @@ -128,9 +128,9 @@ async def _maintain( consecutive_failures = 0 except Exception: consecutive_failures += 1 - log.exception("pool: failed to create pool host provider=%s", provider) + logger.exception("pool: failed to create pool host provider=%s", provider) if consecutive_failures >= _MAX_CONSECUTIVE_CREATE_FAILURES: - log.error( + logger.error( "pool: aborting top-up after %d consecutive failures", consecutive_failures, ) @@ -144,10 +144,10 @@ async def _maintain( if await service.delete_host(host_id, pool_shed=True): removed_excess += 1 except Exception: - log.exception("pool: failed to shed excess pool host_id=%s", host_id) + logger.exception("pool: failed to shed excess pool host_id=%s", host_id) if created or removed_excess: - log.info( + logger.info( "pool: maintained (created=%d, removed_excess=%d, targets=%s)", created, removed_excess, diff --git a/src/hosts/service.py b/src/hosts/service.py index aabdc4c..b9c5af1 100644 --- a/src/hosts/service.py +++ b/src/hosts/service.py @@ -14,6 +14,7 @@ from core.database import async_session_factory from core.exceptions import ResourceNotFoundError from core.settings import Settings, get_settings +from gateway.settings import GatewaySettings from hosts.exceptions import HostStateError, ProvisioningFailedError from hosts.models import Host, HostStatus, IdempotencyKey from networking.tailscale import ( @@ -30,7 +31,7 @@ ) from providers.registry import get_provider_names, get_vm_provider -log = logging.getLogger(__name__) +logger = logging.getLogger(__name__) _SANDBOX_BOOTSTRAP_SCRIPT = ( pathlib.Path(__file__).resolve().parent / "scripts" / "sandbox_bootstrap.sh" @@ -135,7 +136,7 @@ async def get_or_create_host( ) if idempotency_key and not await self._record_idempotency_key(idempotency_key, host): - log.info( + logger.info( "idempotency: lost race on key=%s host_id=%s claimed_at=%s", idempotency_key, host.id, @@ -190,7 +191,7 @@ async def _try_claim_pool_host( if not host: # Lost the race to another claimant; let the caller fall through. return - log.info("pool: claimed host_id=%s name=%s", host.id, host.name) + logger.info("pool: claimed host_id=%s name=%s", host.id, host.name) return host async def create_host( @@ -329,14 +330,14 @@ async def _release_idempotency_loser(self, host: Host) -> None: fresh.claimed_at = None fresh.expires_at = now + timedelta(hours=self.settings.pool_host_max_age_hours) fresh.updated_at = now - log.info( + logger.info( "idempotency: returned pool host_id=%s to pool after lost race", fresh.id, ) else: fresh.expires_at = now fresh.updated_at = now - log.info( + logger.info( "idempotency: marked host_id=%s for janitor reaping after lost race", fresh.id, ) @@ -421,7 +422,7 @@ async def delete_host( # it, or a previous delete partially succeeded. Treat as done # so we can clean up the DB row, but log so unexpected # evictions are visible. - log.warning( + logger.warning( "host VM already absent at provider during teardown: " "host_id=%s name=%s provider=%s", host.id, @@ -465,8 +466,21 @@ async def provision(self, host_id: str) -> None: host.updated_at = utc_now() await self.session.commit() + vm = get_vm_provider(host.provider) + gateway = GatewaySettings() + if vm.gateway_process_class and not gateway.ssh_host: + # A gateway-provider host is reachable only through the gateway; + # provisioning one without an address would hand out dead + # coordinates. + await self.mark_failed( + host, + ProvisioningFailedError( + f"provider {vm.name!r} requires the SSH gateway; set GATEWAY_SSH_HOST" + ), + ) + return try: - vm_result = await get_vm_provider(host.provider).create_vm( + vm_result = await vm.create_vm( name=host.name, image=host.image, env=environment, @@ -482,6 +496,13 @@ async def provision(self, host_id: str) -> None: host.external_ssh_host = vm_result.ssh_host host.external_ssh_port = vm_result.ssh_port host.ssh_username = vm_result.ssh_username + host.public_key = vm_result.public_key or "" + if vm.gateway_process_class: + # The gateway is the SSH path for hosts of a gateway provider. + # The username names the host; the per-host key is the credential. + host.external_ssh_host = gateway.ssh_host + host.external_ssh_port = gateway.ssh_port + host.ssh_username = host.name # Stamp the per-VM key onto this instance so the POST response # carries it. There's no column behind `private_key`, so a later # GET that loads a fresh row sees the class default (None) and @@ -520,7 +541,7 @@ async def provision(self, host_id: str) -> None: await self.session.commit() async def mark_failed(self, host: Host, exc: Exception) -> None: - log.exception( + logger.exception( "sandbox host failed: host_id=%s host_name=%s status=%s", host.id, host.name, diff --git a/src/hosts/tests/test_gateway_path.py b/src/hosts/tests/test_gateway_path.py new file mode 100644 index 0000000..92595b9 --- /dev/null +++ b/src/hosts/tests/test_gateway_path.py @@ -0,0 +1,113 @@ +"""Provisioning tests for the gateway path. + +Hosts of a gateway provider advertise the gateway's address. Hosts of +other providers keep their provider's own coordinates, with or without +a configured gateway. +""" + +from collections.abc import Generator +from unittest.mock import AsyncMock + +import pytest + +from core.database import async_session_factory +from core.settings import Settings, get_settings +from hosts.exceptions import ProvisioningFailedError +from hosts.models import HostStatus +from hosts.service import HostService +from providers.base import VMCreateResult + +MODULE_DOCKER_SBX = "providers.docker_sbx.provider.DockerSbxProvider" + + +@pytest.fixture +def gateway_configured_settings( + monkeypatch: pytest.MonkeyPatch, +) -> Generator[Settings, None, None]: + monkeypatch.setenv("TAILSCALE_ENABLED", "false") + monkeypatch.setenv("GATEWAY_SSH_HOST", "gateway.example.com") + monkeypatch.setenv("GATEWAY_SSH_PORT", "2222") + get_settings.cache_clear() + yield get_settings() + get_settings.cache_clear() + + +def _sbx_create_result() -> VMCreateResult: + return VMCreateResult( + provider_id="sb-x", + name="sb-x", + ssh_port=0, + ssh_username="root", + ssh_host="", + private_key="PRIVATE", + public_key="ssh-ed25519 AAAAPUB", + ) + + +async def test_gateway_provider_hosts_advertise_the_gateway( + gateway_configured_settings: Settings, monkeypatch: pytest.MonkeyPatch +) -> None: + create_vm = AsyncMock(return_value=_sbx_create_result()) + monkeypatch.setattr(f"{MODULE_DOCKER_SBX}.create_vm", create_vm) + scan = AsyncMock(return_value=b"gateway.example.com ssh-ed25519 AAAAGW\n") + monkeypatch.setattr("hosts.service.HostService.scan_known_hosts", scan) + + async with async_session_factory() as session: + service = HostService(session, settings=gateway_configured_settings) + host = await service.create_host(env={}, image=None, provider="docker-sbx") + + assert host.status == HostStatus.ACTIVE.value + assert host.external_ssh_host == "gateway.example.com" + assert host.external_ssh_port == 2222 + assert host.ssh_username == host.name + assert host.public_key == "ssh-ed25519 AAAAPUB" + # The caller still gets the private key exactly once. + assert host.private_key == "PRIVATE" + + +async def test_gateway_provider_hosts_fail_cleanly_without_an_address( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # A gateway-provider host is reachable only through the gateway. Without + # an address, provisioning must fail loudly, not hand out dead + # coordinates. + monkeypatch.setenv("TAILSCALE_ENABLED", "false") + monkeypatch.delenv("GATEWAY_SSH_HOST", raising=False) + get_settings.cache_clear() + + create_vm = AsyncMock(return_value=_sbx_create_result()) + monkeypatch.setattr(f"{MODULE_DOCKER_SBX}.create_vm", create_vm) + + async with async_session_factory() as session: + service = HostService(session, settings=get_settings()) + with pytest.raises(ProvisioningFailedError, match="GATEWAY_SSH_HOST"): + await service.create_host(env={}, image=None, provider="docker-sbx") + + get_settings.cache_clear() + create_vm.assert_not_awaited() + + +async def test_direct_dial_provider_hosts_ignore_the_gateway( + gateway_configured_settings: Settings, monkeypatch: pytest.MonkeyPatch +) -> None: + create_vm = AsyncMock( + return_value=VMCreateResult( + provider_id="vm-1", + name="vm-1", + ssh_port=22, + ssh_username="exedev", + ssh_host="vm-1.public.example.com", + ) + ) + monkeypatch.setattr("providers.exe.provider.ExeProvider.create_vm", create_vm) + scan = AsyncMock(return_value=b"vm-1.public.example.com ssh-ed25519 AAAATEST\n") + monkeypatch.setattr("hosts.service.HostService.scan_known_hosts", scan) + + async with async_session_factory() as session: + service = HostService(session, settings=gateway_configured_settings) + host = await service.create_host(env={}, image=None, provider="exe") + + assert host.external_ssh_host == "vm-1.public.example.com" + assert host.external_ssh_port == 22 + assert host.ssh_username == "exedev" + assert host.public_key == "" diff --git a/src/http_proxies/api.py b/src/http_proxies/api.py index 5f2de4e..10ff1d3 100644 --- a/src/http_proxies/api.py +++ b/src/http_proxies/api.py @@ -14,7 +14,7 @@ ) from http_proxies.service import HTTPProxyService -log = logging.getLogger(__name__) +logger = logging.getLogger(__name__) router = APIRouter(prefix="/http-proxies", tags=["http-proxies"]) diff --git a/src/networking/tailscale.py b/src/networking/tailscale.py index e66d89b..467b63f 100644 --- a/src/networking/tailscale.py +++ b/src/networking/tailscale.py @@ -11,7 +11,7 @@ from .tailscale_settings import TailscaleSettings -log = logging.getLogger(__name__) +logger = logging.getLogger(__name__) class NetworkError(RuntimeError): @@ -314,7 +314,7 @@ async def _poll(self) -> None: except Exception as exc: # Transient transport/listing errors: log and retry within the # caller's timeout budget. - log.warning("tailscale list_devices failed: %s", exc) + logger.warning("tailscale list_devices failed: %s", exc) else: # The set_result loop is sync — no awaits between checking # `not fut.done()` and resolving, so a concurrent timeout diff --git a/src/providers/base.py b/src/providers/base.py index 4f06e49..f690305 100644 --- a/src/providers/base.py +++ b/src/providers/base.py @@ -1,20 +1,75 @@ import abc from dataclasses import dataclass -from typing import ClassVar, Self +from typing import ClassVar, NamedTuple, Self @dataclass(frozen=True) class VMCreateResult: provider_id: str name: str - ssh_port: int ssh_username: str + # Unset ssh_host/ssh_port mean the VM has no directly dialable address. ssh_host: str = "" + ssh_port: int = 0 private_key: str | None = None + # The public half of a per-VM keypair. The service stores it so the SSH + # gateway can authenticate callers of gateway providers. + public_key: str | None = None + + +class TerminalSize(NamedTuple): + columns: int + rows: int + + def __str__(self): + return f"{self.columns}x{self.rows}" + + +class SandboxProcess(abc.ABC): + """One live process inside a sandbox. The gateway pumps bytes between an + SSH channel and this object; the receive methods return b"" at the end.""" + + @classmethod + @abc.abstractmethod + async def open( + cls, + name: str, + *, + command: str | None, + terminal: TerminalSize | None, + ) -> "SandboxProcess": + """Open a process in the sandbox: a shell when command is None, with + a PTY when terminal is not None. Raises neutral provider errors.""" + ... # pragma: no cover + + @abc.abstractmethod + async def receive(self, max_bytes: int) -> bytes: ... # pragma: no cover + + @abc.abstractmethod + async def receive_stderr(self, max_bytes: int) -> bytes: ... # pragma: no cover + + @abc.abstractmethod + def send(self, data: bytes) -> None: ... # pragma: no cover + + @abc.abstractmethod + def send_eof(self) -> None: ... # pragma: no cover + + @abc.abstractmethod + def resize(self, size: TerminalSize) -> None: ... # pragma: no cover + + @abc.abstractmethod + async def wait(self) -> int: ... # pragma: no cover + + @abc.abstractmethod + async def aclose(self) -> None: ... # pragma: no cover class VMProvider(abc.ABC): name: ClassVar[str] + # The process class that serves this provider's hosts through the SSH + # gateway. None means the hosts have their own dialable sshd and the + # gateway plays no part. + gateway_process_class: ClassVar[type[SandboxProcess] | None] = None # Remediation slug attached to a failed /doctor probe. Owned here because # the provider is what knows how its own dependency gets fixed. diagnose_hint: ClassVar[str] diff --git a/src/providers/docker_sbx/api.py b/src/providers/docker_sbx/api.py index 2b1e398..06f12ba 100644 --- a/src/providers/docker_sbx/api.py +++ b/src/providers/docker_sbx/api.py @@ -65,25 +65,6 @@ async def run_bootstrap(self, name: str, script: str) -> None: stdin=script, ) - async def publish_ssh_port(self, name: str) -> int: - # A bare sandbox port tells the daemon to select a free host port on - # the loopback interface. Thus sandboxes do not compete for port - # numbers. The daemon cannot select a free port on an explicit - # address: it rejects "IP::22" and port 0. - output = await self._run("ports", name, "--publish", "22") - # Each binding shows as ": -> 22/tcp". The output can - # have one line for each address family. The lines share the host port. - for line in output.splitlines(): - binding, arrow, target = line.partition("->") - if arrow and target.strip().startswith("22/"): - try: - return int(binding.strip().rsplit(":", 1)[1]) - except (IndexError, ValueError) as error: - raise DockerSbxTransportError( - f"sandbox {name!r} published an unparsable SSH port: {line.strip()!r}" - ) from error - raise DockerSbxTransportError(f"sandbox {name!r} published no SSH port") - async def remove_sandbox(self, name: str) -> None: # The --force flag stops the confirmation prompt. It also removes a # sandbox that has an open SSH session. diff --git a/src/providers/docker_sbx/process.py b/src/providers/docker_sbx/process.py new file mode 100644 index 0000000..63319e7 --- /dev/null +++ b/src/providers/docker_sbx/process.py @@ -0,0 +1,165 @@ +import asyncio +import contextlib +import fcntl +import os +import pty +import struct +import termios + +from providers.base import SandboxProcess, TerminalSize +from providers.exceptions import ProviderTransportError + + +def _set_terminal_size(descriptor: int, size: TerminalSize) -> None: + winsize = struct.pack("HHHH", size.rows, size.columns, 0, 0) + fcntl.ioctl(descriptor, termios.TIOCSWINSZ, winsize) + + +class SbxExecProcess(SandboxProcess): + """A live `sbx exec` process. A caller PTY request gets a local PTY pair, + because the CLI refuses `-t` on pipes. The exec is a daemon session: a + stopped sandbox wakes on open and stays awake while the process runs.""" + + def __init__( + self, + process: asyncio.subprocess.Process, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + pty_master: int | None, + stderr_reader: asyncio.StreamReader | None, + ) -> None: + self._process = process + self._reader = reader + self._writer = writer + self._pty_master = pty_master + self._stderr_reader = stderr_reader + + @classmethod + async def open( + cls, + name: str, + *, + command: str | None, + terminal: TerminalSize | None, + ) -> "SbxExecProcess": + argv = ["sbx", "exec", "--interactive"] + if terminal: + argv.append("--tty") + argv.extend([name, "bash", "-l"]) + if command is not None: + argv.extend(["-c", command]) + environment = {**os.environ, "SBX_NO_TELEMETRY": "1"} + + try: + if terminal: + return await cls._open_with_pty(argv, environment, terminal) + return await cls._open_with_pipes(argv, environment) + except OSError as error: + raise ProviderTransportError(f"sbx CLI could not be started: {error}") from error + + @classmethod + async def _open_with_pty( + cls, + argv: list[str], + environment: dict[str, str], + terminal: TerminalSize, + ) -> "SbxExecProcess": + master, slave = pty.openpty() + try: + _set_terminal_size(slave, terminal) + process = await asyncio.create_subprocess_exec( + *argv, + stdin=slave, + stdout=slave, + stderr=slave, + env=environment, + start_new_session=True, + ) + except OSError: + os.close(master) + raise + finally: + os.close(slave) + reader, writer = await _connect_pty_master(master) + return cls(process, reader, writer, master, stderr_reader=None) + + @classmethod + async def _open_with_pipes( + cls, + argv: list[str], + environment: dict[str, str], + ) -> "SbxExecProcess": + process = await asyncio.create_subprocess_exec( + *argv, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=environment, + start_new_session=True, + ) + assert process.stdout is not None and process.stdin is not None + return cls( + process, + process.stdout, + process.stdin, + pty_master=None, + stderr_reader=process.stderr, + ) + + async def receive(self, max_bytes: int) -> bytes: + try: + return await self._reader.read(max_bytes) + except OSError: + # A PTY master raises EIO at the normal end of the stream. + return b"" + + async def receive_stderr(self, max_bytes: int) -> bytes: + # A terminal merges every stream; pipe mode keeps stderr separate, + # and the sbx wake banner arrives there. + if self._stderr_reader is None: + return b"" + return await self._stderr_reader.read(max_bytes) + + def send(self, data: bytes) -> None: + self._writer.write(data) + + def send_eof(self) -> None: + # A PTY has no end-of-input signal. + if self._pty_master is None: + self._writer.write_eof() + + def resize(self, size: TerminalSize) -> None: + if self._pty_master is None: + return + _set_terminal_size(self._pty_master, size) + + async def wait(self) -> int: + return await self._process.wait() + + async def aclose(self) -> None: + if self._process.returncode is None: + with contextlib.suppress(ProcessLookupError): + self._process.terminate() + with contextlib.suppress(OSError): + self._writer.close() + # No stream owns the PTY master descriptor; it leaks without this. + if self._pty_master is not None: + with contextlib.suppress(OSError): + os.close(self._pty_master) + self._pty_master = None + await self._process.wait() + + +async def _connect_pty_master(master: int) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]: + loop = asyncio.get_running_loop() + reader = asyncio.StreamReader() + await loop.connect_read_pipe( + lambda: asyncio.StreamReaderProtocol(reader), + os.fdopen(master, "rb", buffering=0, closefd=False), + ) + transport, protocol = await loop.connect_write_pipe( + asyncio.streams.FlowControlMixin, + os.fdopen(os.dup(master), "wb", buffering=0), + ) + writer = asyncio.StreamWriter(transport, protocol, reader, loop) + return reader, writer diff --git a/src/providers/docker_sbx/provider.py b/src/providers/docker_sbx/provider.py index 37b42a8..7a553f0 100644 --- a/src/providers/docker_sbx/provider.py +++ b/src/providers/docker_sbx/provider.py @@ -14,6 +14,7 @@ from .api import SbxCLI from .exceptions import DockerSbxNotFoundError, DockerSbxProviderError +from .process import SbxExecProcess from .settings import DockerSbxSettings # The /etc/environment format has one entry on each line. A NUL or a newline @@ -44,9 +45,9 @@ def _bootstrap_script(*, public_key: str, env: dict[str, str], ssh_username: str class DockerSbxProvider(VMProvider): name: ClassVar[str] = "docker-sbx" diagnose_hint: ClassVar[str] = "check_sandboxd_is_running_and_logged_in" - # A local microVM cannot connect to the tailnet: the host proxies all - # sandbox egress, and the template has no init system. These hosts keep - # the published sshd port, also on a tailnet-mode service. + # Sandboxes have no dialable sshd; the gateway serves them, and there is + # no path onto the tailnet. + gateway_process_class = SbxExecProcess supports_tailnet: ClassVar[bool] = False # Each sbx invocation spends approximately 3 seconds on CLI startup work # before the command runs. The default 5-second probe budget fails on a @@ -129,32 +130,27 @@ async def create_vm( try: # The template starts sshd with an empty authorized_keys file. The - # sandbox accepts SSH only after this key is in the file. sshd - # reads the file for each authentication; a restart is not - # necessary. + # sandbox accepts SSH only after this key is in the file. script = _bootstrap_script( public_key=public_key, env=caller_env, ssh_username=self.settings.ssh_username, ) await self.api.run_bootstrap(name, script) - ssh_port = await self.api.publish_ssh_port(name) except DockerSbxProviderError as exc: with contextlib.suppress(DockerSbxProviderError): await self.api.remove_sandbox(name) self._remove_workspace(name) raise ProviderTransportError(str(exc)) from exc - # The daemon publishes sandbox ports on the host loopback interface - # only, the same as the docker provider. A drukbox container reaches - # them through host networking. + # A sandbox has no reachable address of its own: callers arrive + # through the gateway, and the service fills the coordinates in. return VMCreateResult( provider_id=name, name=name, - ssh_port=ssh_port, - ssh_host="127.0.0.1", ssh_username=self.settings.ssh_username, private_key=private_key, + public_key=public_key, ) async def delete_vm(self, name: str) -> None: diff --git a/src/providers/docker_sbx/tests/test_api.py b/src/providers/docker_sbx/tests/test_api.py index 7a77c9f..a41ac1d 100644 --- a/src/providers/docker_sbx/tests/test_api.py +++ b/src/providers/docker_sbx/tests/test_api.py @@ -69,50 +69,6 @@ async def fake_exec(*args, **kwargs): process.communicate.assert_awaited_once_with(script.encode()) -@pytest.mark.asyncio -async def test_publish_ssh_port_asks_for_an_ephemeral_port_and_parses_the_binding(monkeypatch): - captured: dict = {} - - async def fake_exec(*args, **kwargs): - captured["args"] = args - return _process(stdout=b"Published 172.17.0.1:49160 -> 22/tcp\n") - - monkeypatch.setattr("providers.docker_sbx.api.asyncio.create_subprocess_exec", fake_exec) - - port = await SbxCLI().publish_ssh_port("sb-test") - - assert port == 49160 - # A bare sandbox port tells the daemon to select a free loopback port. - assert captured["args"][-2:] == ("--publish", "22") - - -@pytest.mark.asyncio -async def test_publish_ssh_port_picks_the_ssh_binding_out_of_a_multi_line_listing(monkeypatch): - listing = b"Published 127.0.0.1:8080 -> 80/tcp\nPublished 127.0.0.1:49161 -> 22/tcp\n" - create = AsyncMock(return_value=_process(stdout=listing)) - monkeypatch.setattr("providers.docker_sbx.api.asyncio.create_subprocess_exec", create) - - assert await SbxCLI().publish_ssh_port("sb-test") == 49161 - - -@pytest.mark.asyncio -async def test_publish_ssh_port_raises_when_nothing_was_published(monkeypatch): - create = AsyncMock(return_value=_process(stdout=b"\n")) - monkeypatch.setattr("providers.docker_sbx.api.asyncio.create_subprocess_exec", create) - - with pytest.raises(DockerSbxTransportError, match="no SSH port"): - await SbxCLI().publish_ssh_port("sb-test") - - -@pytest.mark.asyncio -async def test_publish_ssh_port_raises_on_unparsable_output(monkeypatch): - create = AsyncMock(return_value=_process(stdout=b"Published 172.17.0.1:notaport -> 22/tcp\n")) - monkeypatch.setattr("providers.docker_sbx.api.asyncio.create_subprocess_exec", create) - - with pytest.raises(DockerSbxTransportError, match="unparsable"): - await SbxCLI().publish_ssh_port("sb-test") - - @pytest.mark.asyncio async def test_sandbox_count_reads_the_listing(monkeypatch): listing = b'{"sandboxes": [{"name": "sb-a"}, {"name": "sb-b"}]}' diff --git a/src/providers/docker_sbx/tests/test_process.py b/src/providers/docker_sbx/tests/test_process.py new file mode 100644 index 0000000..9be54df --- /dev/null +++ b/src/providers/docker_sbx/tests/test_process.py @@ -0,0 +1,117 @@ +import asyncio +import os + +import pytest + +from providers.base import TerminalSize +from providers.docker_sbx.process import SbxExecProcess +from providers.exceptions import ProviderTransportError + +_REAL_EXEC = asyncio.create_subprocess_exec + + +def _stub_sbx_with(script: str, captured: dict): + async def fake_exec(*argv, **kwargs): + captured["argv"] = argv + return await _REAL_EXEC("bash", "-c", script, **kwargs) + + return fake_exec + + +async def _drain(session: SbxExecProcess) -> bytes: + output = bytearray() + while data := await session.receive(4096): + output.extend(data) + return bytes(output) + + +async def test_exec_session_bridges_pipes_in_both_directions(monkeypatch): + captured: dict = {} + monkeypatch.setattr( + "providers.docker_sbx.process.asyncio.create_subprocess_exec", + _stub_sbx_with("cat; exit 5", captured), + ) + + session = await SbxExecProcess.open("sb-test", command="true", terminal=None) + session.send(b"round-trip") + session.send_eof() + output = await _drain(session) + status = await session.wait() + await session.aclose() + + assert captured["argv"] == ( + "sbx", + "exec", + "--interactive", + "sb-test", + "bash", + "-l", + "-c", + "true", + ) + assert output == b"round-trip" + assert status == 5 + + +async def test_exec_session_keeps_the_error_stream_separate(monkeypatch): + # The sbx wake banner goes to stderr; it must not arrive in the output. + monkeypatch.setattr( + "providers.docker_sbx.process.asyncio.create_subprocess_exec", + _stub_sbx_with("echo OUT; echo BANNER >&2; exit 0", {}), + ) + + session = await SbxExecProcess.open("sb-test", command="true", terminal=None) + output = await _drain(session) + stderr = await session.receive_stderr(4096) + await session.aclose() + + assert output == b"OUT\n" + assert stderr == b"BANNER\n" + + +async def test_terminal_request_allocates_a_real_pty_at_the_requested_size(monkeypatch): + captured: dict = {} + monkeypatch.setattr( + "providers.docker_sbx.process.asyncio.create_subprocess_exec", + _stub_sbx_with("test -t 0 && echo ISTTY; stty size; exit 3", captured), + ) + + session = await SbxExecProcess.open( + "sb-test", command=None, terminal=TerminalSize(columns=101, rows=42) + ) + output = await _drain(session) + status = await session.wait() + await session.aclose() + + assert "--tty" in captured["argv"] + assert captured["argv"][-2:] == ("bash", "-l") + assert b"ISTTY" in output + assert b"42 101" in output + assert status == 3 + + +async def test_aclose_releases_the_terminal_descriptor(monkeypatch): + monkeypatch.setattr( + "providers.docker_sbx.process.asyncio.create_subprocess_exec", + _stub_sbx_with("exit 0", {}), + ) + + session = await SbxExecProcess.open( + "sb-test", command=None, terminal=TerminalSize(columns=80, rows=24) + ) + master = session._pty_master + assert master is not None + await session.aclose() + + with pytest.raises(OSError): + os.fstat(master) + + +async def test_missing_sbx_binary_maps_to_transport_error(monkeypatch): + async def fail_exec(*argv, **kwargs): + raise FileNotFoundError("sbx") + + monkeypatch.setattr("providers.docker_sbx.process.asyncio.create_subprocess_exec", fail_exec) + + with pytest.raises(ProviderTransportError, match="could not be started"): + await SbxExecProcess.open("sb-test", command=None, terminal=None) diff --git a/src/providers/docker_sbx/tests/test_provider.py b/src/providers/docker_sbx/tests/test_provider.py index 733a199..f2a35f6 100644 --- a/src/providers/docker_sbx/tests/test_provider.py +++ b/src/providers/docker_sbx/tests/test_provider.py @@ -25,14 +25,13 @@ def _api_mock() -> MagicMock: api = MagicMock() api.create_sandbox = AsyncMock() api.run_bootstrap = AsyncMock() - api.publish_ssh_port = AsyncMock(return_value=49160) api.remove_sandbox = AsyncMock() api.sandbox_count = AsyncMock(return_value=2) return api @pytest.mark.asyncio -async def test_create_vm_creates_a_sized_sandbox_and_returns_published_coords(tmp_path): +async def test_create_vm_creates_a_sized_sandbox_and_returns_key_material(tmp_path): api = _api_mock() provider = DockerSbxProvider(api, _settings(tmp_path, cpus=4, memory="8g")) @@ -48,12 +47,15 @@ async def test_create_vm_creates_a_sized_sandbox_and_returns_published_coords(tm assert create_kwargs["workspace"] == str(tmp_path / "sb-test") assert (tmp_path / "sb-test").is_dir() - api.publish_ssh_port.assert_awaited_once_with("sb-test") - assert result.ssh_host == "127.0.0.1" - assert result.ssh_port == 49160 + # The sandbox has no reachable address of its own; the service fills + # the gateway coordinates in. + assert result.ssh_host == "" + assert result.ssh_port == 0 assert result.ssh_username == "root" assert result.private_key is not None assert "-----BEGIN OPENSSH PRIVATE KEY-----" in result.private_key + assert result.public_key is not None + assert result.public_key.startswith("ssh-ed25519 ") @pytest.mark.asyncio @@ -138,29 +140,18 @@ async def test_create_vm_translates_an_unwritable_workspace_root(tmp_path): @pytest.mark.asyncio -async def test_create_vm_tears_down_sandbox_and_workspace_when_publishing_fails(tmp_path): +async def test_create_vm_tears_down_after_a_failed_bootstrap_and_keeps_the_first_error(tmp_path): + # The cleanup of the sandbox can itself fail. The caller must get the + # first error, and the workspace must still go. api = _api_mock() - api.publish_ssh_port.side_effect = DockerSbxTransportError("no port") - provider = DockerSbxProvider(api, _settings(tmp_path)) - - with pytest.raises(ProviderTransportError): - await provider.create_vm(name="sb-test", image="img", env={}) - api.remove_sandbox.assert_awaited_once_with("sb-test") - assert not (tmp_path / "sb-test").exists() - - -@pytest.mark.asyncio -async def test_create_vm_publish_error_survives_failed_cleanup(tmp_path): - # The cleanup of the sandbox can also fail. The caller must get the - # first error, not the cleanup error. - api = _api_mock() - api.publish_ssh_port.side_effect = DockerSbxTransportError("no port") + api.run_bootstrap.side_effect = DockerSbxTransportError("exec failed") api.remove_sandbox.side_effect = DockerSbxTransportError("cleanup failed") provider = DockerSbxProvider(api, _settings(tmp_path)) - with pytest.raises(ProviderTransportError, match="no port"): + with pytest.raises(ProviderTransportError, match="exec failed"): await provider.create_vm(name="sb-test", image="img", env={}) api.remove_sandbox.assert_awaited_once_with("sb-test") + assert not (tmp_path / "sb-test").exists() @pytest.mark.asyncio diff --git a/uv.lock b/uv.lock index 1132740..a74176f 100644 --- a/uv.lock +++ b/uv.lock @@ -250,6 +250,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/da/42/e921fccf5015463e32a3cf6ee7f980a6ed0f395ceeaa45060b61d86486c2/anyio-4.13.0-py3-none-any.whl", hash = "sha256:08b310f9e24a9594186fd75b4f73f4a4152069e3853f1ed8bfbf58369f4ad708", size = 114353, upload-time = "2026-03-24T12:59:08.246Z" }, ] +[[package]] +name = "asyncssh" +version = "2.24.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cryptography" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/dd/7f/2d79247bacc562104f312d27efe541673aa177feac89de291bd61bca52be/asyncssh-2.24.0.tar.gz", hash = "sha256:4064c590e59ce2e8d82a2f66d35f3120d765828b4df5e3dbfb07b4a8c24686c9", size = 550148, upload-time = "2026-06-27T20:34:44.755Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3e/29/908ce0ca5e8cae76662e354a0f08df552d6d221844748b9e5ca06051cc44/asyncssh-2.24.0-py3-none-any.whl", hash = "sha256:9abd46300adcb6d4b73269b34c53cd0d17a138b9a22b5b38008ce7d5808734b7", size = 381237, upload-time = "2026-06-27T20:34:43.198Z" }, +] + [[package]] name = "attrs" version = "26.1.0" @@ -450,6 +463,7 @@ source = { editable = "." } dependencies = [ { name = "aiosqlite" }, { name = "alembic" }, + { name = "asyncssh" }, { name = "cryptography" }, { name = "fastapi" }, { name = "greenlet" }, @@ -481,6 +495,7 @@ requires-dist = [ { name = "aioboto3", marker = "extra == 'aws'", specifier = ">=13" }, { name = "aiosqlite", specifier = ">=0.20" }, { name = "alembic", specifier = ">=1.14" }, + { name = "asyncssh", specifier = ">=2.14" }, { name = "cryptography", specifier = ">=43" }, { name = "fastapi", specifier = ">=0.115" }, { name = "greenlet", specifier = ">=3.0" },