Fix protected activation ordering and diagnostic review findings

1 files changed, 16 insertions, 0 deletions

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)