From e2de8188abb3d1a639435aba42b4a3056e72aaae Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Mon, 31 Aug 2026 16:58:03 +0300 Subject: [PATCH] feat(observatory): bridge local M49 worker container --- .../observatory/m49_worker_container_main.py | 169 ++++++++++++++++++ ...t_observatory_m49_worker_container_main.py | 72 ++++++++ 2 files changed, 241 insertions(+) create mode 100644 src/k1link/observatory/m49_worker_container_main.py create mode 100644 tests/test_observatory_m49_worker_container_main.py diff --git a/src/k1link/observatory/m49_worker_container_main.py b/src/k1link/observatory/m49_worker_container_main.py new file mode 100644 index 0000000..b40485a --- /dev/null +++ b/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()) diff --git a/tests/test_observatory_m49_worker_container_main.py b/tests/test_observatory_m49_worker_container_main.py new file mode 100644 index 0000000..79b0714 --- /dev/null +++ b/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