Serve SFTP through the gateway

docs/deploy.md

@@ -203,8 +203,12 @@ 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 serves an interactive shell, command execution, and SFTP.
+It refuses scp (the legacy protocol) and port forwarding. SFTP runs the
+sandbox's own SFTP server over one persistent session for each SSH
+connection, thus repeated file operations on one connection start no new
+session. The session closes after a short idle period, so an inactive
+sandbox still sleeps.
 
 The gateway is a requirement for gateway providers: `POST /hosts` for
 `docker-sbx` fails without `GATEWAY_SSH_HOST`. Set it to the address

src/gateway/backend.py

@@ -0,0 +1,181 @@
+# The SFTP client handler wants asyncssh's own reader and writer; the exec
+# adapters here duck-type the small surface it uses, so the argument-type
+# check does not apply.
+# pyright: reportArgumentType=false
+import asyncio
+import contextlib
+import logging
+from collections.abc import AsyncIterator
+
+from asyncssh.sftp import SFTPClientHandler
+
+from providers.base import SandboxProcess
+
+logger = logging.getLogger(__name__)
+
+_SFTP_VERSION = 3
+
+# The command that starts the OpenSSH SFTP server in the sandbox. The image
+# installs it at the standard Debian path.
+_SFTP_SERVER_COMMAND = "exec /usr/lib/openssh/sftp-server"
+
+# One `sbx exec` costs seconds of CLI startup. Thus the server stays open
+# between operations. It closes after this idle time, so an unused sandbox
+# sleeps again, and it opens again on the next operation.
+IDLE_CLOSE_SECONDS = 30.0
+
+
+class _ExecReader:
+    """Give the SFTP client the reader it expects on the process byte
+    stream: exact reads, connection info, and a child logger."""
+
+    def __init__(self, process: SandboxProcess) -> None:
+        self._process = process
+        self._buffer = bytearray()
+        self.logger = _NullLogger()
+
+    async def readexactly(self, count: int) -> bytes:
+        while len(self._buffer) < count:
+            chunk = await self._process.receive(count - len(self._buffer))
+            if not chunk:
+                raise asyncio.IncompleteReadError(bytes(self._buffer), count)
+            self._buffer.extend(chunk)
+        taken = bytes(self._buffer[:count])
+        del self._buffer[:count]
+        return taken
+
+    def get_extra_info(self, name: str, default=None):
+        return default
+
+
+class _ExecWriter:
+    def __init__(self, process: SandboxProcess) -> None:
+        self._process = process
+
+    def write(self, data: bytes) -> None:
+        self._process.send(data)
+
+    def close(self) -> None:
+        with contextlib.suppress(Exception):
+            self._process.send_eof()
+
+
+class _NullLogger:
+    def get_child(self, *args, **kwargs) -> "_NullLogger":
+        return self
+
+    def __getattr__(self, name):
+        return lambda *args, **kwargs: None
+
+
+class SandboxSftpBackend:
+    """The one SFTP server behind every SFTP session on one SSH connection.
+
+    An SFTP client drives the sandbox's own OpenSSH SFTP server over a
+    persistent `sbx exec`. The sessions share it, so a per-session poll
+    starts no new process. It closes after an idle time and opens again on
+    the next use, so an inactive sandbox still sleeps.
+    """
+
+    def __init__(
+        self,
+        process_class: type[SandboxProcess],
+        host_name: str,
+        *,
+        idle_close_seconds: float = IDLE_CLOSE_SECONDS,
+    ) -> None:
+        self._process_class = process_class
+        self._host_name = host_name
+        self._idle_close_seconds = idle_close_seconds
+        self._lock = asyncio.Lock()
+        self._process: SandboxProcess | None = None
+        self._handler: SFTPClientHandler | None = None
+        self._recv_task: asyncio.Task[None] | None = None
+        self._in_flight = 0
+        self._idle_closer: asyncio.Task[None] | None = None
+        self._closed = False
+
+    @contextlib.asynccontextmanager
+    async def session(self) -> AsyncIterator[SFTPClientHandler]:
+        """Give the live SFTP handler for one operation. It opens the
+        process on first use. The idle timer runs only when no operation
+        is in flight."""
+        handler = await self._acquire()
+        try:
+            yield handler
+        finally:
+            await self._release()
+
+    async def _acquire(self) -> SFTPClientHandler:
+        async with self._lock:
+            if self._closed:
+                raise ConnectionError("the backend is closed")
+            if self._idle_closer:
+                self._idle_closer.cancel()
+                self._idle_closer = None
+            if self._handler is None:
+                await self._open()
+            assert self._handler is not None
+            self._in_flight += 1
+            return self._handler
+
+    async def _release(self) -> None:
+        async with self._lock:
+            self._in_flight -= 1
+            if self._in_flight == 0 and not self._closed:
+                self._idle_closer = asyncio.create_task(self._close_after_idle())
+
+    async def aclose(self) -> None:
+        async with self._lock:
+            self._closed = True
+            if self._idle_closer:
+                self._idle_closer.cancel()
+            await self._teardown()
+
+    async def _open(self) -> None:
+        # Store the process before the handshake. A failed handshake then
+        # closes the process, and does not leak a live exec that would keep
+        # the sandbox awake.
+        self._process = await self._process_class.open(
+            self._host_name,
+            command=_SFTP_SERVER_COMMAND,
+            terminal=None,
+        )
+        # The client handler reads exact byte counts and writes framed
+        # packets. The exec adapters give that surface.
+        handler = SFTPClientHandler(
+            asyncio.get_running_loop(),
+            "strict",
+            _ExecReader(self._process),
+            _ExecWriter(self._process),
+            _SFTP_VERSION,
+        )
+        try:
+            await handler.start()
+        except Exception:
+            await self._teardown()
+            raise
+        self._handler = handler
+        self._recv_task = asyncio.create_task(handler.recv_packets())
+        logger.info("gateway: sftp backend open host=%s", self._host_name)
+
+    async def _close_after_idle(self) -> None:
+        await asyncio.sleep(self._idle_close_seconds)
+        async with self._lock:
+            if self._in_flight == 0 and not self._closed:
+                await self._teardown()
+                logger.info("gateway: sftp backend idle-closed host=%s", self._host_name)
+
+    async def _teardown(self) -> None:
+        if self._recv_task:
+            self._recv_task.cancel()
+            # The task can already have ended with a pipe error. Teardown
+            # needs it to be done. How it ended does not matter.
+            with contextlib.suppress(BaseException):
+                await self._recv_task
+            self._recv_task = None
+        if self._process:
+            with contextlib.suppress(Exception):
+                await self._process.aclose()
+            self._process = None
+        self._handler = None

src/gateway/server.py

@@ -3,10 +3,13 @@
 import logging
 
 import asyncssh
+import asyncssh.sftp
 from sqlalchemy import select
 
 from core.database import async_session_factory
+from gateway.backend import SandboxSftpBackend
 from gateway.settings import GatewaySettings
+from gateway.sftp import GatewaySFTPServer
 from hosts.models import Host, HostStatus
 from providers.base import SandboxProcess, TerminalSize
 from providers.exceptions import ProviderError
@@ -16,13 +19,50 @@
 
 _RECEIVE_CHUNK_BYTES = 32768
 
+# The gateway forwards each file operation to the sandbox by an opaque
+# handle. Two SFTP extensions ask the server to seek inside an open file:
+# server-side copy and sparse-range detection. The gateway cannot serve
+# them on an opaque handle. asyncssh advertises them from this class list
+# and gives no per-server control, so the gateway removes them from the
+# list. The filter is idempotent, thus it can run on each SFTP session.
+_UNSUPPORTED_SFTP_EXTENSIONS = (b"copy-data", b"ranges@asyncssh.com")
+
+
+def _disable_unsupported_sftp_extensions() -> None:
+    asyncssh.sftp.SFTPServerHandler._extensions = [
+        extension
+        for extension in asyncssh.sftp.SFTPServerHandler._extensions
+        if extension[0] not in _UNSUPPORTED_SFTP_EXTENSIONS
+    ]
+
 
 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
+        self._sftp_backend: SandboxSftpBackend | None = None
+        self._cleanup: asyncio.Task[None] | None = None
+
+    def sftp_backend(self) -> SandboxSftpBackend:
+        """Return the connection's one SFTP backend, shared by every SFTP
+        session. It is made on first use; its process opens lazily."""
+        assert self.host is not None
+        if self._sftp_backend is None:
+            provider = get_vm_provider(self.host.provider)
+            if not provider.gateway_process_class:
+                raise asyncssh.SFTPOpUnsupported("cannot open a session for this host")
+            self._sftp_backend = SandboxSftpBackend(provider.gateway_process_class, self.host.name)
+        return self._sftp_backend
+
+    def connection_lost(self, exc: Exception | None) -> None:
+        # Close the backend, so its exec process ends and the sandbox can
+        # sleep. Keep the task until it finishes: the loop can otherwise
+        # collect it during the cleanup.
+        if self._sftp_backend:
+            with contextlib.suppress(RuntimeError):
+                self._cleanup = asyncio.get_running_loop().create_task(self._sftp_backend.aclose())
 
     def begin_auth(self, username: str) -> bool:
         return True
@@ -142,6 +182,13 @@ def _load_host_key(settings: GatewaySettings) -> asyncssh.SSHKey:
     return key
 
 
+def _open_sftp(channel: asyncssh.SSHServerChannel) -> GatewaySFTPServer:
+    _disable_unsupported_sftp_extensions()
+    server = channel.get_connection().get_owner()
+    assert isinstance(server, GatewayConnection)
+    return GatewaySFTPServer(channel, server.sftp_backend())
+
+
 async def start(settings: GatewaySettings) -> asyncssh.SSHAcceptor:
     server = await asyncssh.listen(
         host=settings.bind_host,
@@ -151,7 +198,7 @@ async def start(settings: GatewaySettings) -> asyncssh.SSHAcceptor:
         process_factory=_bridge,
         encoding=None,
         allow_scp=False,
-        sftp_factory=None,
+        sftp_factory=_open_sftp,
         agent_forwarding=False,
         x11_forwarding=False,
     )

src/gateway/sftp.py

@@ -0,0 +1,87 @@
+# asyncssh types SFTP file handles as bytes on the client side and as an
+# opaque object on the server side. This module bridges the two by handing
+# each server handle straight to the client, so the handle-argument and
+# override checks do not apply here.
+# pyright: reportArgumentType=false, reportIncompatibleMethodOverride=false
+import asyncssh
+
+from gateway.backend import SandboxSftpBackend
+
+# The SFTP protocol version drukbox speaks upstream and to callers. A stat
+# with these flags asks the server for every attribute it knows.
+_STAT_ALL = 0x8000_01FD
+
+
+class GatewaySFTPServer(asyncssh.SFTPServer):
+    """One caller SFTP session, forwarded to the sandbox's own SFTP server.
+
+    The connection's backend holds the real upstream client. Every operation
+    delegates to it. Thus the sandbox's OpenSSH server does the file work,
+    and its errors — already SFTP errors — go straight back to the caller.
+    Many sessions share one backend, so a per-session poll starts no new
+    process.
+    """
+
+    def __init__(self, chan, backend: SandboxSftpBackend):
+        super().__init__(chan)
+        self._backend = backend
+
+    async def open(self, path, pflags, attrs):
+        async with self._backend.session() as handler:
+            return await handler.open(path, pflags, attrs)
+
+    async def close(self, file_obj):
+        async with self._backend.session() as handler:
+            await handler.close(file_obj)
+
+    async def read(self, file_obj, offset, size):
+        async with self._backend.session() as handler:
+            data, _ = await handler.read(file_obj, offset, size)
+            return data
+
+    async def write(self, file_obj, offset, data):
+        async with self._backend.session() as handler:
+            return await handler.write(file_obj, offset, data)
+
+    async def fstat(self, file_obj):
+        async with self._backend.session() as handler:
+            return await handler.fstat(file_obj, _STAT_ALL)
+
+    async def stat(self, path):
+        async with self._backend.session() as handler:
+            return await handler.stat(path, _STAT_ALL)
+
+    async def lstat(self, path):
+        async with self._backend.session() as handler:
+            return await handler.lstat(path, _STAT_ALL)
+
+    async def setstat(self, path, attrs):
+        async with self._backend.session() as handler:
+            await handler.setstat(path, attrs)
+
+    async def mkdir(self, path, attrs):
+        async with self._backend.session() as handler:
+            await handler.mkdir(path, attrs)
+
+    async def remove(self, path):
+        async with self._backend.session() as handler:
+            await handler.remove(path)
+
+    async def realpath(self, path):
+        async with self._backend.session() as handler:
+            names, _ = await handler.realpath(path)
+            return names[0].filename
+
+    def _unsupported(self, *args, **kwargs):
+        raise asyncssh.SFTPOpUnsupported("not served by this gateway")
+
+    fsetstat = _unsupported
+    scandir = _unsupported
+    rmdir = _unsupported
+    rename = _unsupported
+    posix_rename = _unsupported
+    readlink = _unsupported
+    symlink = _unsupported
+    link = _unsupported
+    lock = _unsupported
+    unlock = _unsupported

src/gateway/tests/conftest.py

@@ -0,0 +1,10 @@
+import pytest
+
+from gateway.tests.localprocess import SFTP_SERVER_COMMAND
+
+
+@pytest.fixture(autouse=True)
+def _local_sftp_server_command(monkeypatch):
+    # The tests back SFTP with the host's own sftp-server, which sits at a
+    # different path than the sandbox image's. Point the backend at it.
+    monkeypatch.setattr("gateway.backend._SFTP_SERVER_COMMAND", SFTP_SERVER_COMMAND)