feat(computer): a browser alone on the open web: every connection logged, every wait bounded (FD-4969, M8.21)

src/agent_env/computers/README.md

@@ -14,9 +14,10 @@ building it. Name it from another bundle as `ubuntu-24.04`, the alias `aliases.t
 (`display = "none"`), for one website browser, which runs headless there, and the `page` driver, whose screen is the
 browser's page: a browser computer, which agentenv's `chromium` deploys alone as. supervisord runs every program as the
 session user, among them a stand-in site proxy on `127.0.0.1:3128`, the browser's only route out when the computer
-routes no sites, which forwards every request to wherever it names, refusing nothing: the sandbox's network policy is
-all that limits it. A computer that routes sites names the shared site proxy's forward port, `127.0.0.1:3129`, to the
-browser instead: it serves the run's sites, refuses every other host and runs out of the agent's reach. It has a
+routes no sites. It forwards a request or a `CONNECT` tunnel to any host, never decrypting a tunnel and refusing
+nothing: the sandbox's network policy is all that limits it. Every line it writes to `/var/log/agentenv/site-proxy.log`
+is JSON, one per connection and one per connection that fails, in the same shapes as the shared site proxy's. A computer that routes sites names the shared site proxy's forward port,
+`127.0.0.1:3129`, to the browser instead: it serves the run's sites, refuses every other host and runs out of the agent's reach. It has a
 container form only. Name it from another bundle as `ubuntu-24.04-headless`, or by its id,
 `@local/agentenv-framework/computers/ubuntu-24.04-headless`.
 

src/agent_env/computers/artifacts/ubuntu-24.04-headless/site-proxy.py

@@ -1,68 +1,150 @@
-"""A stand-in for the shared site proxy in this base: it forwards every request the browser sends it, absolute-form
-HTTP and CONNECT tunnels alike, to wherever it names, refusing nothing, on 127.0.0.1:3128."""
+"""A stand-in for the shared site proxy in this base, on 127.0.0.1:3128: it forwards every request the browser sends
+it, absolute-form HTTP and CONNECT tunnels alike, to the host it names, refusing nothing and never decrypting a tunnel.
+The sandbox's own network policy still decides where a connection can go. Every line it logs is JSON, in the same
+shapes as the shared site proxy's: ``connection`` (host, port, address, bytes each way, start, duration) when one
+ends, and ``failed`` (method, host, port, ``reason`` "doesn't resolve" or "no answer", and the error) for one that
+couldn't open."""
 import http.server
-import select
+import json
+import selectors
 import socket
 import socketserver
+import time
 import urllib.parse
 
 PORT = 3128
-IDLE_S = 60
+CONNECT_TIMEOUT_S = 30
+IDLE_S = 300
 HOP = {"connection", "keep-alive", "proxy-connection", "proxy-authorization"}
 
 
+def log(**fields) -> None:
+    print(json.dumps(fields), flush=True)
+
+
+def _error(error: BaseException) -> str:
+    return f"{type(error).__name__}: {error}".removesuffix(": ")
+
+
 class Proxy(http.server.BaseHTTPRequestHandler):
     protocol_version = "HTTP/1.1"
 
     def do_CONNECT(self):
         host, _, port = self.path.rpartition(":")
         host = host.removeprefix("[").removesuffix("]")
-        try:
-            upstream = socket.create_connection((host, int(port)), timeout=10)
-        except (OSError, ValueError):
-            self.send_error(502)
+        if not port.isdigit():
+            self.send_error(400)
+            return
+        if (opened := self._open(host, int(port), "CONNECT")) is None:
             return
         self.send_response(200, "Connection established")
         self.end_headers()
         self.close_connection = True
-        _pipe(self.connection, upstream)
+        self._relay(*opened, host, int(port), b"")
 
     def forward(self):
         url = urllib.parse.urlsplit(self.path)
-        try:
-            upstream = socket.create_connection((url.hostname, url.port or 80), timeout=10)
-        except (OSError, TypeError):
-            self.send_error(502)
+        if not url.hostname:
+            self.send_error(400)
+            return
+        port = url.port or 80
+        if (opened := self._open(url.hostname, port, self.command)) is None:
             return
         target = (url.path or "/") + (f"?{url.query}" if url.query else "")
         headers = "".join(f"{k}: {v}\r\n" for k, v in self.headers.items() if k.lower() not in HOP)
-        upstream.sendall(f"{self.command} {target} HTTP/1.1\r\n{headers}Connection: close\r\n\r\n".encode("latin-1"))
-        if length := int(self.headers.get("Content-Length") or 0):
-            upstream.sendall(self.rfile.read(length))
+        head = f"{self.command} {target} HTTP/1.1\r\n{headers}Connection: close\r\n\r\n".encode("latin-1")
+        body = self.rfile.read(length) if (length := int(self.headers.get("Content-Length") or 0)) else b""
         self.close_connection = True
-        _pipe(self.connection, upstream)
+        self._relay(*opened, url.hostname, port, head + body)
 
     do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = do_HEAD = do_OPTIONS = forward
 
+    def _open(self, host: str, port: int, method: str) -> tuple[socket.socket, str] | None:
+        """A connection to an address ``host`` resolves to, tried in turn within one ``CONNECT_TIMEOUT_S`` for them
+        all, and that address; or None once the browser has its answer, a 502 for a host that doesn't resolve or
+        answer. The connection keeps an ``IDLE_S`` timeout, so a write to a peer that stopped reading ends too."""
+        deadline = time.monotonic() + CONNECT_TIMEOUT_S
+        try:
+            found = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+        except OSError as error:
+            log(failed={"method": method, "host": host, "port": port, "reason": "doesn't resolve",
+                        "error": _error(error)})
+            self._answer(502, f"{host} doesn't resolve\n")
+            return None
+        last = None
+        for address in dict.fromkeys(info[4][0] for info in found):
+            if (left := deadline - time.monotonic()) <= 0:
+                break
+            try:
+                upstream = socket.create_connection((address, port), timeout=left)
+            except OSError as error:
+                last = error
+                continue
+            upstream.settimeout(IDLE_S)
+            return upstream, address
+        log(failed={"method": method, "host": host, "port": port, "reason": "no answer",
+                    "error": _error(last) if last else f"none within {CONNECT_TIMEOUT_S} s"})
+        self._answer(502, f"{host}:{port} didn't answer\n")
+        return None
+
+    def _answer(self, status: int, body: str) -> None:
+        data = body.encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "text/plain; charset=utf-8")
+        self.send_header("Content-Length", str(len(data)))
+        self.send_header("X-Agentenv-Site-Proxy", "failed")
+        self.send_header("Connection", "close")
+        self.end_headers()
+        self.wfile.write(data)
+        self.close_connection = True
+
+    def _relay(self, upstream: socket.socket, address: str, host: str, port: int, first: bytes) -> None:
+        started, began = time.time(), time.monotonic()
+        sent, received = len(first), 0
+        self.connection.settimeout(IDLE_S)
+        try:
+            if first:
+                upstream.sendall(first)
+            more, received = _pipe(self.connection, upstream)
+            sent += more
+        finally:
+            log(connection={"host": host, "port": port, "address": address, "bytes_sent": sent,
+                            "bytes_received": received,
+                            "started": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(started)),
+                            "duration_s": round(time.monotonic() - began, 3)})
+
     def log_message(self, *args):
         pass
 
 
-def _pipe(a: socket.socket, b: socket.socket) -> None:
+def _pipe(client: socket.socket, upstream: socket.socket) -> tuple[int, int]:
+    """Relay both ways until either side closes or both are idle ``IDLE_S``; the bytes sent upstream and received.
+    It waits with a selector (epoll on Linux), not ``select()``, which refuses a socket numbered past 1023."""
+    sent = received = 0
+    waiting = selectors.DefaultSelector()
     try:
+        waiting.register(client, selectors.EVENT_READ)
+        waiting.register(upstream, selectors.EVENT_READ)
         while True:
-            ready, _, _ = select.select([a, b], [], [], IDLE_S)
+            ready = waiting.select(IDLE_S)
             if not ready:
-                return
-            for s in ready:
+                break
+            for key, _ in ready:
+                s = key.fileobj
                 data = s.recv(65536)
                 if not data:
-                    return
-                (b if s is a else a).sendall(data)
+                    return sent, received
+                (upstream if s is client else client).sendall(data)
+                if s is client:
+                    sent += len(data)
+                else:
+                    received += len(data)
     except OSError:
-        return
+        pass
     finally:
-        b.close()
+        waiting.close()
+        upstream.close()
+    return sent, received
 
 
 class Server(socketserver.ThreadingMixIn, http.server.HTTPServer):
@@ -71,4 +153,5 @@ class Server(socketserver.ThreadingMixIn, http.server.HTTPServer):
 
 
 if __name__ == "__main__":
+    log(listening={"address": "127.0.0.1", "port": PORT})
     Server(("127.0.0.1", PORT), Proxy).serve_forever()

src/agent_env/env/envs/computer/utils/site_proxy.py

@@ -106,6 +106,26 @@ def matches(pattern: str, host: str) -> bool:
     return host == pattern
 
 
+def log_event(**fields) -> None:
+    """One JSON line in the proxy's log, every line it writes being one: ``connection`` (host, port, address, bytes each
+    way, start, duration) when a connection to a live site's host ends, ``failed`` (method, host, port, ``reason``
+    "doesn't resolve" or "no answer", and the error) for one that couldn't open, both in the same shapes as the
+    headless base's stand-in proxy's; ``refused`` for a host that's no site, ``error`` for a connection that failed
+    unexpectedly, and ``restarted`` when ``keep`` starts the proxy again."""
+    print(json.dumps(fields), file=sys.stderr, flush=True)
+
+
+def described(error: BaseException) -> str:
+    return f"{type(error).__name__}: {error}".removesuffix(": ")
+
+
+class Count:
+    """The bytes a relay has carried one way, counted as it goes, so a connection that ends on an error keeps them."""
+
+    def __init__(self, start: int = 0):
+        self.bytes = start
+
+
 @dataclass
 class Site:
     """One site as the proxy serves it: its names, the route on the gateway it forwards to, and its certificate and key
@@ -361,10 +381,13 @@ async def relay_chunked(reader: asyncio.StreamReader, writer: asyncio.StreamWrit
         await relay_exact(reader, writer, size + 2)
 
 
-async def relay_until_eof(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+async def relay_until_eof(reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
+                          count: Optional[Count] = None) -> None:
     while data := await reader.read(CHUNK):
         writer.write(data)
         await writer.drain()
+        if count is not None:
+            count.bytes += len(data)
 
 
 async def relay_body(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, headers: list[tuple[str, str]],
@@ -380,10 +403,11 @@ async def relay_body(reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
         await relay_until_eof(reader, writer)
 
 
-async def splice(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
-    """One direction of an upgraded connection, closing the other side when it ends, which ends the other direction."""
+async def splice(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, count: Optional[Count] = None) -> None:
+    """One direction of an upgraded connection, closing the other side when it ends, which ends the other direction;
+    the bytes it carries counted in ``count`` when given."""
     try:
-        await relay_until_eof(reader, writer)
+        await relay_until_eof(reader, writer, count)
     except (ConnectionError, OSError, ssl.SSLError):
         pass
     finally:
@@ -455,22 +479,37 @@ async def _tunnel(self, reader: asyncio.StreamReader, writer: asyncio.StreamWrit
     async def _direct(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, host: str, port: int,
                       head: Optional[bytes] = None) -> None:
         """``host``, a live site's, reached straight at ``port``: a tunnel, answered as established, for a ``CONNECT``,
-        else ``head`` sent on; then both ends relayed until either closes."""
+        else ``head`` sent on; then both ends relayed until either closes, and the connection logged (``log_event``).
+        One that doesn't resolve or answer within ``UPSTREAM_TIMEOUT_S``, the lookup and every address's attempt
+        together, is a 502, logged too."""
         try:
             up_reader, up_writer = await asyncio.wait_for(asyncio.open_connection(host, port), UPSTREAM_TIMEOUT_S)
         except (OSError, asyncio.TimeoutError) as e:
-            writer.write(simple_response("502 Bad Gateway", f"{host}:{port} didn't answer: {e}\n", FAILED))
+            unresolved = isinstance(e, socket.gaierror)
+            log_event(failed={"method": "CONNECT" if head is None else head.split(b" ", 1)[0].decode("latin-1"),
+                              "host": host, "port": port, "reason": "doesn't resolve" if unresolved else "no answer",
+                              "error": described(e)})
+            writer.write(simple_response("502 Bad Gateway", f"{host} doesn't resolve\n" if unresolved else
+                                         f"{host}:{port} didn't answer\n", FAILED))
             return
+        address = (up_writer.get_extra_info("peername") or ("",))[0]
+        started, began, sent, received = time.time(), time.monotonic(), Count(len(head or b"")), Count()
         if head is None:
             writer.write(b"HTTP/1.1 200 Connection established\r\n\r\n")
         else:
             up_writer.write(head)
-        await asyncio.gather(writer.drain(), up_writer.drain())
-        await asyncio.gather(splice(reader, up_writer), splice(up_reader, writer))
+        try:
+            await asyncio.gather(writer.drain(), up_writer.drain())
+            await asyncio.gather(splice(reader, up_writer, sent), splice(up_reader, writer, received))
+        finally:
+            log_event(connection={"host": host, "port": port, "address": address, "bytes_sent": sent.bytes,
+                                  "bytes_received": received.bytes,
+                                  "started": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(started)),
+                                  "duration_s": round(time.monotonic() - began, 3)})
 
     @staticmethod
     def _refuse(writer: asyncio.StreamWriter, method: str, host: str) -> None:
-        print(f"site proxy: refused {method} {host}: no site of this computer", file=sys.stderr, flush=True)
+        log_event(refused={"method": method, "host": host, "reason": "no site of this computer"})
         writer.write(simple_response("403 Forbidden", f"{host} is no site of this computer, and the browser reaches "
                                      "its sites alone\n", FAILED))
 
@@ -483,7 +522,7 @@ async def _serving(writer: asyncio.StreamWriter, label: str, serving: Awaitable[
         except (ConnectionError, asyncio.IncompleteReadError, ssl.SSLError, OSError):
             pass
         except Exception as e:
-            print(f"site proxy: a {label} connection failed: {type(e).__name__}: {e}", file=sys.stderr, flush=True)
+            log_event(error={"connection": label, "error": described(e)})
         finally:
             try:
                 await writer.drain()
@@ -1101,7 +1140,7 @@ def keep(serve_path: str) -> None:
     while True:
         ended = subprocess.run([sys.executable, os.path.abspath(__file__), "serve", serve_path],
                                stdin=subprocess.DEVNULL).returncode
-        print(f"the site proxy ended ({ended}); starting it again", flush=True)
+        print(json.dumps({"restarted": {"exit_code": ended}}), flush=True)
         time.sleep(RESTART_DELAY_S)