tests: udploop, UDP services in one thread

udppm: 20 clients at once each answered with their own datagrams, -s
ending a session with its first answer, STRING_LONG of silence ending
it (92), maxconn leaving a client without a slot unanswered, and a
session adopted on a reload and ended when the new rules deny its
client. dnspr: a query sent on to the name server and its answer
relayed, and a query answered while another waits for a silent name
server, which ends at DNS_TO (821).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/cases/udploop.py

@@ -0,0 +1,217 @@
+"""UDP services in one thread: udppm sessions, many at once, their limits and
+deadlines, and their adoption by the new service on a reload; dnspr
+forwarding to a name server without holding up other queries."""
+
+import socket
+import struct
+import threading
+import time
+
+
+def exchange(sock, port, payload, timeout=2.0):
+    sock.settimeout(timeout)
+    sock.sendto(payload, ("127.0.0.1", port))
+    try:
+        return sock.recvfrom(65536)[0]
+    except OSError:
+        return None
+
+
+def dns_raw(name, qtype, qid=0x3B3B):
+    q = struct.pack("!HHHHHH", qid, 0x0100, 1, 0, 0, 0)
+    for label in name.split("."):
+        q += bytes([len(label)]) + label.encode()
+    return q + b"\x00" + struct.pack("!HH", qtype, 1)
+
+
+def fake_name_server(answer):
+    """A name server answering every query with one TXT record, the text
+    given; and one which answers nothing."""
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.bind(("127.0.0.1", 0))
+
+    def serve():
+        while True:
+            try:
+                data, peer = sock.recvfrom(65536)
+            except OSError:
+                return
+            if answer is None:
+                continue
+            reply = bytearray(data)
+            reply[2] = 0x81
+            reply[3] = 0x80
+            reply[6:8] = struct.pack("!H", 1)
+            txt = bytes([len(answer)]) + answer
+            reply += b"\xc0\x0c" + struct.pack("!HHIH", 16, 1, 60, len(txt)) + txt
+            sock.sendto(bytes(reply), peer)
+
+    threading.Thread(target=serve, daemon=True).start()
+    return sock, sock.getsockname()[1]
+
+
+def run(t):
+    echo = t.udp_echo()
+    adm = t.free_port()
+    many, single, idle, limited, acl = (t.free_port() for _ in range(5))
+
+    def config(acl_line):
+        return f"""
+            log
+            auth iponly
+            allow *
+            admin -p{adm}
+
+            flush
+            auth iponly
+            allow *
+            udppm {many} 127.0.0.1 {echo}
+            udppm -s {single} 127.0.0.1 {echo}
+
+            flush
+            timeouts 1 5 30 2 180 1800 15 60 15 5 5
+            auth iponly
+            allow *
+            udppm {idle} 127.0.0.1 {echo}
+
+            flush
+            maxconn 2
+            auth iponly
+            allow *
+            udppm {limited} 127.0.0.1 {echo}
+
+            flush
+            auth iponly
+            {acl_line}
+            udppm {acl} 127.0.0.1 {echo}
+        """
+
+    server = t.start("udploop", config("allow *"), ports=[adm])
+    for p in (many, single, idle, limited, acl):
+        t.wait_udp(p)
+
+    # --- many sessions at once, each answered with its own datagrams ------
+    clients = [socket.socket(socket.AF_INET, socket.SOCK_DGRAM) for _ in range(20)]
+    wrong = 0
+    for rnd in range(3):
+        for i, c in enumerate(clients):
+            c.sendto(f"c{i}r{rnd}".encode(), ("127.0.0.1", many))
+        for i, c in enumerate(clients):
+            c.settimeout(2)
+            try:
+                got = c.recvfrom(2048)[0]
+            except OSError:
+                got = None
+            if got != f"echo:c{i}r{rnd}".encode():
+                wrong += 1
+    for c in clients:
+        c.close()
+    t.eq(0, wrong, "20 clients at once each get their own answers, three rounds")
+
+    # --- -s: a session ends with its first answer -------------------------
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    t.eq(b"echo:one", exchange(s, single, b"one"), "with -s a datagram is answered")
+    t.eq(b"echo:two", exchange(s, single, b"two"), "and the next one too, in a new session")
+    cport = s.getsockname()[1]
+    s.close()
+    time.sleep(0.3)
+    ended = [l for l in server.output().splitlines()
+             if f" {single} 00000 " in l and f":{cport} " in l]
+    t.eq(2, len(ended), "each answer ended its session, logged")
+
+    # --- idle timeout: STRING_LONG of silence ends a session (92) ----------
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    t.eq(b"echo:x", exchange(s, idle, b"x"), "a session is answered")
+    cport = s.getsockname()[1]
+    def idle_ended():
+        return any(f" {idle} 00092 " in l and f":{cport} " in l for l in server.output().splitlines())
+
+    deadline = time.time() + 6
+    while time.time() < deadline and not idle_ended():
+        time.sleep(0.2)
+    t.ok("two seconds of silence end it, logged with 92") if idle_ended() else \
+        t.fail("two seconds of silence end it, logged with 92", "a 00092 line", "none")
+    s.close()
+
+    # --- maxconn: a client without a free slot is not answered -------------
+    held = [socket.socket(socket.AF_INET, socket.SOCK_DGRAM) for _ in range(2)]
+    t.eq([b"echo:a", b"echo:a"], [exchange(h, limited, b"a") for h in held],
+         "two clients take the two slots")
+    third = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    t.eq(None, exchange(third, limited, b"b", timeout=1),
+         "a third gets no answer while they hold them")
+    t.contains(server.output(), "too many connected clients",
+               "and that is logged")
+    for h in held + [third]:
+        h.close()
+
+    # --- reload: sessions are adopted and re-authenticated -----------------
+    kept = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    t.eq(b"echo:k1", exchange(kept, acl, b"k1"), "a session before the reload")
+    mark = len(server.output())
+    t.write_config("udploop", config("deny *"))
+    t.http(f"http://127.0.0.1:{adm}/R")
+    time.sleep(2.5)
+    t.eq(None, exchange(kept, acl, b"k2", timeout=1),
+         "denied by the new rules it is answered no more")
+    cport = kept.getsockname()[1]
+    t.ok("its session ended by the re-authentication") \
+        if any(f" {acl} " in l and f":{cport} " in l and " 00000 " not in l
+               for l in server.output()[mark:].splitlines()) else \
+        t.fail("its session ended by the re-authentication", "a line with a result", "none")
+    kept.close()
+    t.write_config("udploop", config("allow *"))
+    t.http(f"http://127.0.0.1:{adm}/R")
+    time.sleep(2.5)
+
+    # --- dnspr: forwarded queries wait in the loop, not in it ---------------
+    good, gport = fake_name_server(b"from-the-server")
+    mute, mport = fake_name_server(None)
+    fwd, slow, local = t.free_port(), t.free_port(), t.free_port()
+    t.start("udploop_dns", f"""
+        log
+        timeouts 1 5 30 60 180 1800 3 60 15 5 5
+        nserver 127.0.0.1:{gport}
+        nscache 256
+        nsrecord local.test 10.1.2.3
+        auth iponly
+        allow *
+        dnspr -p{fwd}
+
+        flush
+        auth iponly
+        allow *
+        dnspr -p{local}
+    """)
+    # a service of its own whose name server says nothing
+    dns2 = t.start("udploop_mute", f"""
+        log
+        timeouts 1 5 30 60 180 1800 3 60 15 5 5
+        nserver 127.0.0.1:{mport}
+        nscache 256
+        nsrecord local.test 10.1.2.3
+        auth iponly
+        allow *
+        dnspr -p{slow}
+    """)
+    t.wait_udp(fwd, payload=dns_raw("local.test", 1))
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    got = exchange(s, fwd, dns_raw("txt.test", 16))
+    t.ok("a TXT query is sent on to the name server and its answer relayed") \
+        if got and b"from-the-server" in got else \
+        t.fail("a TXT query is sent on to the name server and its answer relayed", "the TXT", got)
+
+    t.wait_udp(slow, payload=dns_raw("local.test", 1))
+    pending = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    pending.sendto(dns_raw("mute.test", 16), ("127.0.0.1", slow))
+    start = time.time()
+    other = exchange(s, slow, dns_raw("local.test", 1), timeout=2)
+    t.ok(f"while one query waits for a silent name server another is answered ({time.time() - start:.2f} s)") \
+        if other and b"\x0a\x01\x02\x03" in other else \
+        t.fail("while one query waits for a silent name server another is answered", "10.1.2.3", other)
+    deadline = time.time() + 6
+    while time.time() < deadline and " 00821 " not in dns2.output():
+        time.sleep(0.2)
+    t.contains(dns2.output(), " 00821 ", "the waiting one ends at DNS_TO (821)")
+    for x in (s, pending, good, mute):
+        x.close()