From 5fb3db9fea460d046e26041714db4490611262a9 Mon Sep 17 00:00:00 2001 From: Harenome Ranaivoarivony-Razanajato Date: Thu, 5 Nov 2026 16:33:33 +0100 Subject: [PATCH] Resolve a licence server before its forwarder takes the server's name A licence server's forwarder joins the unit's internal network under the server's own name. It resolved that name for each connection, and the network's DNS then answered with the forwarder itself, so the agent got a closed connection. The forwarder now resolves the name once, before it listens, and keeps every address. Each connection tries the addresses in order and skips an address family this machine lacks. A name that does not resolve ends the forwarder with an "unresolved" event. The unit connects the forwarder to its network only after the forwarder listens. --- app/nullius/standard/container.py | 5 +- app/nullius/standard/forward.py | 41 +++++++-- app/tests/test_standard_container.py | 9 ++ app/tests/test_standard_forward.py | 121 +++++++++++++++++++++++++++ 4 files changed, 165 insertions(+), 11 deletions(-) create mode 100644 app/tests/test_standard_forward.py diff --git a/app/nullius/standard/container.py b/app/nullius/standard/container.py index 794f239..f4bfe26 100644 --- a/app/nullius/standard/container.py +++ b/app/nullius/standard/container.py @@ -684,14 +684,15 @@ def start(self) -> "Unit": def _forward(self, i: int, address: str) -> str: """A licence server's forwarder, started and on the unit's network - under the server's name: the address the agent reaches it by.""" + under the server's name: the address the agent reaches it by. It joins + that network once it listens, after it resolves the server's name.""" alias, host, port = server_alias(address, i) name = f"{self.names['agent']}-lic{i}" self.started.append(name) _docker(forward_argv(self.image, name=name, job=self.job, port=port)[1:], env=client_env({"NULLIUS_FORWARD_TO": f"{host}:{port}"})) - _docker(["network", "connect", "--alias", alias, self.names["network"], name]) self._wait(name, "the licence server's forwarder") + _docker(["network", "connect", "--alias", alias, self.names["network"], name]) return f"{alias}:{port}" def _door(self, i: int, alias: str, port: int, sock: str) -> None: diff --git a/app/nullius/standard/forward.py b/app/nullius/standard/forward.py index 0d82f8c..1aac8d4 100644 --- a/app/nullius/standard/forward.py +++ b/app/nullius/standard/forward.py @@ -7,7 +7,10 @@ beside the relay (``container.Unit``), on the default bridge and on the unit's network under the server's own name, listening on the server's port and passing every connection to that one address - and to nothing else, -whatever the agent sends. It reads nothing of what it passes. +whatever the agent sends. It reads nothing of what it passes. It resolves the +server's name once, before it listens: on the unit's network, that name is +the forwarder's own. A name that does not resolve ends it with an +``unresolved`` event. A benchmark's MCP server run on the machine (``mcp_machine``) is reached the same way: a forwarder on the unit's internal network only, the server's @@ -44,14 +47,26 @@ def _pump(src: socket.socket, dst: socket.socket) -> None: pass -def _serve(conn: socket.socket, host: str, port) -> None: - try: - if port is None: # a Unix socket: host is its path - up = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) +def _connect(targets) -> socket.socket: + """The first of ``targets`` (family, address) that accepts, or the last error.""" + error = OSError("no address to connect to") + for family, addr in targets: + up = None + try: + up = socket.socket(family, socket.SOCK_STREAM) # in the try: a family this kernel lacks is skipped up.settimeout(30) - up.connect(host) - else: - up = socket.create_connection((host, port), timeout=30) + up.connect(addr) + return up + except OSError as e: + error = e + if up is not None: + up.close() + raise error + + +def _serve(conn: socket.socket, targets) -> None: + try: + up = _connect(targets) up.settimeout(None) except OSError as e: print(json.dumps({"event": "unreachable", "error": str(e)[:200]}), flush=True) @@ -78,6 +93,14 @@ def main(argv=None) -> int: ap.error("--to, or NULLIUS_FORWARD_TO: where to forward") lhost, lport = _split(a.listen) host, port = _split(a.to) + if port is None: + targets = [(socket.AF_UNIX, host)] + else: + try: + targets = [(f, addr) for f, _, _, _, addr in socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)] + except OSError as e: + print(json.dumps({"event": "unresolved", "host": host, "error": str(e)[:200]}), flush=True) + return 1 srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind((lhost, lport)) @@ -85,7 +108,7 @@ def main(argv=None) -> int: print(json.dumps({"event": "listening", "port": lport}), flush=True) while True: conn, _ = srv.accept() - threading.Thread(target=_serve, args=(conn, host, port), daemon=True).start() + threading.Thread(target=_serve, args=(conn, targets), daemon=True).start() if __name__ == "__main__": diff --git a/app/tests/test_standard_container.py b/app/tests/test_standard_container.py index 490d186..93d4521 100644 --- a/app/tests/test_standard_container.py +++ b/app/tests/test_standard_container.py @@ -144,6 +144,15 @@ def test_wrap_passes_values_in_the_docker_programs_environment_and_leaves_host_p self.assertNotEqual(env["PATH"], "/i/bin:" + container.IMAGE_PATH) # the docker program's own self.assertEqual(u.inside("u0002"), "http://relay:8080/u0002/v1") + def test_a_licence_forwarder_listens_before_it_takes_the_servers_name(self): + u = container.Unit("j", "u0001", os.path.join(self.tmp, "rec"), image="img", app="/app", base="b", key="k") + calls = [] + with mock.patch.object(container, "_docker", side_effect=lambda argv, **kw: calls.append(argv[:2])), \ + mock.patch.object(container, "wait_listening", side_effect=lambda name, *a: calls.append(["wait", name])): + self.assertEqual(u._forward(1, "licence.example:27000"), "licence.example:27000") + name = u.names["agent"] + "-lic1" + self.assertEqual(calls, [["run", "-d"], ["wait", name], ["network", "connect"]]) + def test_a_long_job_name_gives_a_short_unique_slug(self): a, b = container.slug("x" * 200), container.slug("x" * 199 + "y") self.assertLessEqual(len(a), 80) diff --git a/app/tests/test_standard_forward.py b/app/tests/test_standard_forward.py new file mode 100644 index 0000000..3d45063 --- /dev/null +++ b/app/tests/test_standard_forward.py @@ -0,0 +1,121 @@ +"""The licence server's forwarder (``forward.py``), run in-process: it +resolves its target once, before it listens, tries each address in order, and +ends when the name does not resolve.""" +from __future__ import annotations + +import contextlib +import io +import json +import socket +import threading +import time +import unittest +from unittest import mock + +from nullius.standard import forward + + +def _greeter(greeting: bytes, host: str = "127.0.0.1") -> int: + family, _, _, _, addr = socket.getaddrinfo(host, 0, type=socket.SOCK_STREAM)[0] # where the name goes first + srv = socket.socket(family, socket.SOCK_STREAM) + srv.bind(addr[:2]) + srv.listen(8) + + def serve(): + while True: + conn, _ = srv.accept() + conn.sendall(greeting) + conn.close() + threading.Thread(target=serve, daemon=True).start() + return srv.getsockname()[1] + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _forwarding(to: str) -> int: + """A forwarder to ``to`` in a thread: the port it listens on, once it listens.""" + listen = _free_port() + threading.Thread(target=forward.main, args=(["--listen", f"127.0.0.1:{listen}", "--to", to],), daemon=True).start() + deadline = time.time() + 10 + while True: + try: + with socket.socket() as s: # by address: no lookup, which a test may have patched + s.settimeout(1) + s.connect(("127.0.0.1", listen)) + return listen + except OSError: + if time.time() > deadline: + raise + time.sleep(0.05) + + +def _greeting(port: int) -> bytes: + with socket.socket() as c: + c.settimeout(10) + c.connect(("127.0.0.1", port)) + return c.recv(64) + + +class ForwardTest(unittest.TestCase): + def test_a_named_server_is_resolved_once_so_a_later_lookup_cannot_redirect_it(self): + real = _greeter(b"LICENSE-OK", "localhost") + listen = _forwarding(f"localhost:{real}") + decoy = _greeter(b"SELF") # the name now answers as the forwarder's own + later = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", decoy))] + with mock.patch("socket.getaddrinfo", return_value=later): + self.assertEqual(_greeting(listen), b"LICENSE-OK") + + def test_each_connection_takes_the_names_first_address_that_accepts(self): + first, second = _greeter(b"FIRST"), _greeter(b"SECOND") + both = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", first)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", second))] + with mock.patch("socket.getaddrinfo", return_value=both): + listen = _forwarding("licence.example:27000") + self.assertEqual([_greeting(listen) for _ in range(3)], [b"FIRST"] * 3) + + def test_a_refused_address_is_passed_over_for_the_next(self): + real = _greeter(b"LICENSE-OK") + dead = socket.socket() # bound, never listening: refused, and no one else can take it + self.addCleanup(dead.close) + dead.bind(("127.0.0.1", 0)) + both = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", dead.getsockname()), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", real))] + with mock.patch("socket.getaddrinfo", return_value=both): + listen = _forwarding("licence.example:27000") + self.assertEqual(_greeting(listen), b"LICENSE-OK") + + def test_an_address_family_this_machine_lacks_is_skipped(self): + real, make = _greeter(b"LICENSE-OK"), socket.socket + + def no_ipv6(family=-1, *args, **kwargs): # the greeter's accept makes sockets too + if family == socket.AF_INET6: + raise OSError(97, "Address family not supported by protocol") + return make(family, *args, **kwargs) + with mock.patch("socket.socket", side_effect=no_ipv6): + up = forward._connect([(socket.AF_INET6, ("::1", real)), (socket.AF_INET, ("127.0.0.1", real))]) + with up: + self.assertEqual(up.recv(64), b"LICENSE-OK") + + def test_no_address_at_all_is_an_error_not_a_crash(self): + with self.assertRaises(OSError): + forward._connect([]) + + def test_a_name_that_does_not_resolve_ends_it_with_the_reason(self): + out, ended = io.StringIO(), [] + argv = ["--listen", f"127.0.0.1:{_free_port()}", "--to", "nowhere.invalid:27000"] + with mock.patch("socket.getaddrinfo", side_effect=socket.gaierror("no such name")), \ + contextlib.redirect_stdout(out): + t = threading.Thread(target=lambda: ended.append(forward.main(argv)), daemon=True) + t.start() + t.join(10) # a forwarder that listened anyway never returns + self.assertEqual(ended, [1]) + event = json.loads(next(line for line in out.getvalue().splitlines() if '"unresolved"' in line)) + self.assertEqual((event["event"], event["host"]), ("unresolved", "nowhere.invalid")) + + +if __name__ == "__main__": + unittest.main()