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()