feat(observatory): bridge local M49 worker container

src/k1link/observatory/m49_worker_container_main.py

@@ -0,0 +1,169 @@
+"""Container entrypoint for the fixed portable M4.9 Worker service.
+
+Docker Desktop reaches the Windows host through ``host.docker.internal``, while
+the authenticated Worker gateway deliberately accepts plaintext HTTP only on a
+loopback URL.  This wrapper supplies that missing transport seam without
+weakening the gateway: a process-local TCP bridge binds only
+``127.0.0.1:18080`` and forwards only to the fixed Worker-host endpoint.
+"""
+
+from __future__ import annotations
+
+import shutil
+import socket
+import socketserver
+import threading
+from collections.abc import Sequence
+from contextlib import suppress
+from typing import Final
+
+from k1link.observatory import m49_worker_service
+
+M49_CONTAINER_PROXY_LISTEN_HOST: Final = "127.0.0.1"
+M49_CONTAINER_PROXY_LISTEN_PORT: Final = 18080
+M49_CONTAINER_PROXY_UPSTREAM_HOST: Final = "host.docker.internal"
+M49_CONTAINER_PROXY_UPSTREAM_PORT: Final = 18080
+M49_CONTAINER_PROXY_CONNECT_TIMEOUT_SECONDS: Final = 10.0
+M49_CONTAINER_PROXY_COPY_BYTES: Final = 1024 * 1024
+
+
+class M49WorkerContainerProxyError(RuntimeError):
+    """The fixed container loopback bridge could not be started safely."""
+
+
+class _ThreadedTcpServer(socketserver.ThreadingTCPServer):
+    allow_reuse_address = True
+    daemon_threads = True
+
+
+class _FixedProxyHandler(socketserver.BaseRequestHandler):
+    server: _ThreadedTcpServer
+
+    def handle(self) -> None:
+        upstream_address = getattr(self.server, "upstream_address", None)
+        connect_timeout = getattr(self.server, "connect_timeout", None)
+        if (
+            not isinstance(upstream_address, tuple)
+            or len(upstream_address) != 2
+            or not isinstance(upstream_address[0], str)
+            or not isinstance(upstream_address[1], int)
+            or not isinstance(connect_timeout, float)
+        ):
+            return
+        try:
+            upstream = socket.create_connection(
+                upstream_address,
+                timeout=connect_timeout,
+            )
+        except OSError:
+            return
+        with upstream:
+            upstream.settimeout(None)
+            client = self.request
+            if not isinstance(client, socket.socket):
+                return
+            client.settimeout(None)
+            client_to_upstream = threading.Thread(
+                target=_copy_socket,
+                args=(client, upstream),
+                daemon=True,
+                name="m49-proxy-client-to-host",
+            )
+            upstream_to_client = threading.Thread(
+                target=_copy_socket,
+                args=(upstream, client),
+                daemon=True,
+                name="m49-proxy-host-to-client",
+            )
+            client_to_upstream.start()
+            upstream_to_client.start()
+            client_to_upstream.join()
+            upstream_to_client.join()
+
+
+class FixedM49ContainerLoopbackProxy:
+    """Own one bounded TCP bridge for the lifetime of the Worker process."""
+
+    def __init__(
+        self,
+        *,
+        listen_host: str = M49_CONTAINER_PROXY_LISTEN_HOST,
+        listen_port: int = M49_CONTAINER_PROXY_LISTEN_PORT,
+        upstream_host: str = M49_CONTAINER_PROXY_UPSTREAM_HOST,
+        upstream_port: int = M49_CONTAINER_PROXY_UPSTREAM_PORT,
+        connect_timeout: float = M49_CONTAINER_PROXY_CONNECT_TIMEOUT_SECONDS,
+    ) -> None:
+        if listen_host != M49_CONTAINER_PROXY_LISTEN_HOST:
+            raise ValueError("M4.9 container proxy must bind IPv4 loopback")
+        if not 0 <= listen_port <= 65_535:
+            raise ValueError("M4.9 container proxy listen port is invalid")
+        if not upstream_host or upstream_host != upstream_host.strip():
+            raise ValueError("M4.9 container proxy upstream host is invalid")
+        if not 1 <= upstream_port <= 65_535:
+            raise ValueError("M4.9 container proxy upstream port is invalid")
+        if not 0.05 <= connect_timeout <= 60.0:
+            raise ValueError("M4.9 container proxy timeout is invalid")
+        try:
+            server = _ThreadedTcpServer(
+                (listen_host, listen_port),
+                _FixedProxyHandler,
+                bind_and_activate=True,
+            )
+        except OSError as exc:
+            raise M49WorkerContainerProxyError(
+                "M4.9 container loopback proxy could not bind"
+            ) from exc
+        server.upstream_address = (upstream_host, upstream_port)  # type: ignore[attr-defined]
+        server.connect_timeout = float(connect_timeout)  # type: ignore[attr-defined]
+        self._server = server
+        self._thread = threading.Thread(
+            target=server.serve_forever,
+            kwargs={"poll_interval": 0.1},
+            daemon=True,
+            name="m49-container-loopback-proxy",
+        )
+
+    @property
+    def listen_port(self) -> int:
+        address = self._server.server_address
+        if not isinstance(address, tuple) or not isinstance(address[1], int):
+            raise M49WorkerContainerProxyError("M4.9 proxy address is invalid")
+        return address[1]
+
+    def __enter__(self) -> FixedM49ContainerLoopbackProxy:
+        self._thread.start()
+        return self
+
+    def __exit__(self, *_args: object) -> None:
+        self._server.shutdown()
+        self._server.server_close()
+        self._thread.join(timeout=5.0)
+        if self._thread.is_alive():
+            raise M49WorkerContainerProxyError(
+                "M4.9 container loopback proxy did not stop"
+            )
+
+
+def _copy_socket(source: socket.socket, destination: socket.socket) -> None:
+    try:
+        shutil.copyfileobj(
+            source.makefile("rb", buffering=0),
+            destination.makefile("wb", buffering=0),
+            length=M49_CONTAINER_PROXY_COPY_BYTES,
+        )
+    except OSError:
+        pass
+    finally:
+        with suppress(OSError):
+            destination.shutdown(socket.SHUT_WR)
+
+
+def main(arguments: Sequence[str] | None = None) -> int:
+    """Run the fixed bridge and the sealed M4.9 service in one process."""
+
+    with FixedM49ContainerLoopbackProxy():
+        return m49_worker_service.main(arguments)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_observatory_m49_worker_container_main.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import socket
+import socketserver
+import threading
+
+import pytest
+
+import k1link.observatory.m49_worker_container_main as container_main
+
+
+class _EchoHandler(socketserver.BaseRequestHandler):
+    def handle(self) -> None:
+        payload = self.request.recv(1024)
+        self.request.sendall(payload)
+
+
+def test_fixed_container_proxy_bridges_loopback_without_gateway_override() -> None:
+    upstream = socketserver.ThreadingTCPServer(("127.0.0.1", 0), _EchoHandler)
+    upstream_thread = threading.Thread(target=upstream.serve_forever, daemon=True)
+    upstream_thread.start()
+    try:
+        upstream_port = upstream.server_address[1]
+        assert isinstance(upstream_port, int)
+        with container_main.FixedM49ContainerLoopbackProxy(
+            listen_port=0,
+            upstream_host="127.0.0.1",
+            upstream_port=upstream_port,
+        ) as proxy, socket.create_connection(
+            ("127.0.0.1", proxy.listen_port)
+        ) as client:
+            client.sendall(b"fixed-m49-proxy")
+            client.shutdown(socket.SHUT_WR)
+            assert client.recv(1024) == b"fixed-m49-proxy"
+    finally:
+        upstream.shutdown()
+        upstream.server_close()
+        upstream_thread.join(timeout=5.0)
+
+
+def test_container_entrypoint_owns_proxy_around_worker(monkeypatch: pytest.MonkeyPatch) -> None:
+    lifecycle: list[str] = []
+
+    class _Proxy:
+        def __enter__(self) -> _Proxy:
+            lifecycle.append("proxy-started")
+            return self
+
+        def __exit__(self, *_args: object) -> None:
+            lifecycle.append("proxy-stopped")
+
+    def worker(arguments: object) -> int:
+        lifecycle.append(f"worker:{arguments!r}")
+        return 17
+
+    monkeypatch.setattr(container_main, "FixedM49ContainerLoopbackProxy", _Proxy)
+    monkeypatch.setattr(container_main.m49_worker_service, "main", worker)
+
+    assert container_main.main(("--once",)) == 17
+    assert lifecycle == ["proxy-started", "worker:('--once',)", "proxy-stopped"]
+
+
+def test_container_proxy_refuses_non_loopback_listener() -> None:
+    with pytest.raises(ValueError, match="IPv4 loopback"):
+        container_main.FixedM49ContainerLoopbackProxy(listen_host="0.0.0.0")
+
+
+def test_production_proxy_endpoints_are_fixed() -> None:
+    assert container_main.M49_CONTAINER_PROXY_LISTEN_HOST == "127.0.0.1"
+    assert container_main.M49_CONTAINER_PROXY_LISTEN_PORT == 18080
+    assert container_main.M49_CONTAINER_PROXY_UPSTREAM_HOST == "host.docker.internal"
+    assert container_main.M49_CONTAINER_PROXY_UPSTREAM_PORT == 18080