Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions app/nullius/standard/container.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
41 changes: 32 additions & 9 deletions app/nullius/standard/forward.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -78,14 +93,22 @@ 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))
srv.listen(64)
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__":
Expand Down
9 changes: 9 additions & 0 deletions app/tests/test_standard_container.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
121 changes: 121 additions & 0 deletions app/tests/test_standard_forward.py
Original file line number Diff line number Diff line change
@@ -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()
Loading