From fc3c1d51b1be9db6ba6ad286584fbc34bb305f38 Mon Sep 17 00:00:00 2001 From: Andrew Guschin Date: Fri, 30 Oct 2026 03:32:00 +0400 Subject: Fix protected activation ordering and diagnostic review findings --- tests/vm_guard_acceptance.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) (limited to 'tests/vm_guard_acceptance.py') diff --git a/tests/vm_guard_acceptance.py b/tests/vm_guard_acceptance.py index ff99c33..8886d7b 100644 --- a/tests/vm_guard_acceptance.py +++ b/tests/vm_guard_acceptance.py @@ -80,8 +80,13 @@ def serve(host,port): c.sendall(data) while True: c,_=s.accept();threading.Thread(target=handle,args=(c,),daemon=True).start() +def udp(host): + s=socket.socket(socket.AF_INET6 if ':' in host else socket.AF_INET,socket.SOCK_DGRAM);s.bind((host,53)) + while True: + data,peer=s.recvfrom(1024);s.sendto(data,peer) for host in ['198.51.100.2','10.254.0.2','2001:db8:726f::2','fded:726f::2']: for port in [53,18080]:threading.Thread(target=serve,args=(host,port),daemon=True).start() + threading.Thread(target=udp,args=(host,),daemon=True).start() time.sleep(120) ''' children.append(subprocess.Popen(['ip', 'netns', 'exec', 'roentgen-peer', 'python3', '-u', '-c', server], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)) @@ -123,6 +128,17 @@ for line in sys.stdin: else: raise AssertionError(f'Public direct fallback leaked to {host}:{port}') results.append('public IPv4/IPv6 destinations are blocked on ports 53 and 18080') + for host, allowed in [('10.254.0.2', True), ('fded:726f::2', True), ('198.51.100.2', False), ('2001:db8:726f::2', False)]: + with socket.socket(socket.AF_INET6 if ':' in host else socket.AF_INET, socket.SOCK_DGRAM) as udp: + udp.settimeout(1) + try: + udp.sendto(b'dns-port-observation', (host, 53)) + data, _ = udp.recvfrom(1024) + except OSError: + assert not allowed, f'Private UDP DNS was blocked: {host}' + else: + assert allowed and data == b'dns-port-observation', f'Public UDP DNS leaked: {host}' + results.append('UDP DNS follows the same IPv4/IPv6 private/public destination rule') incoming.stdin.write('enabled\n'); incoming.stdin.flush() assert incoming.stdout.readline().strip() == 'ok' guard(False) -- cgit v1.2.3