#!/usr/bin/env python3 """Independent audit checks, run INSIDE a container with the probe's exact flags (see audit_run.sh). Part 1 (before any listener exists): is there any route out? Interfaces, v4/v6 routes, and real connect/sendto attempts to public and host-side addresses; every one must fail with ENETUNREACH. Part 2: positive controls through the probe's OWN listeners (imported from /probe/kit/inside.py): - urllib HTTPS to the collector name: must show in the fake DNS log AND as an SNI in the TLS log - urllib HTTP to the collector name: must show as a Host header in the HTTP log - a raw connect to a hard-coded collector IP: invisible to DNS/TLS logs, must show in strace Prints one JSON object. Standard library only; talks to nothing but loopback. """ import errno import json import os import socket import subprocess import sys import time sys.path.insert(0, "/probe/kit") import inside # noqa: E402 (the probe's own listeners; the instrument under audit) COLLECTOR = inside.COLLECTOR OUT = {} def attempt_tcp(host, port, fam=socket.AF_INET): s = socket.socket(fam, socket.SOCK_STREAM) s.settimeout(3) try: s.connect((host, port)) return "CONNECTED" except OSError as e: return f"errno {e.errno} {errno.errorcode.get(e.errno, '?')}" finally: s.close() def attempt_udp(host, port, fam=socket.AF_INET): s = socket.socket(fam, socket.SOCK_DGRAM) try: s.sendto(b"\x00" * 12, (host, port)) return "SENT" except OSError as e: return f"errno {e.errno} {errno.errorcode.get(e.errno, '?')}" finally: s.close() def read(p): try: with open(p) as f: return f.read() except OSError as e: return repr(e) def part1(extra_ips): r = {"netns": os.readlink("/proc/self/ns/net"), "interfaces": sorted(os.listdir("/sys/class/net")), "proc_net_dev": read("/proc/net/dev"), "route_v4": read("/proc/net/route"), "route_v6": read("/proc/net/ipv6_route"), "if_inet6": read("/proc/net/if_inet6"), "resolv_conf": read("/etc/resolv.conf"), "proxy_env": {k: v for k, v in os.environ.items() if "proxy" in k.lower()}, "run_dirs": {d: (sorted(os.listdir(d)) if os.path.isdir(d) else None) for d in ("/run", "/var/run")}, "uid": os.getuid(), "cap_eff": [l for l in read("/proc/self/status").splitlines() if l.startswith("Cap")]} tcp = {f"{h}:{p}": attempt_tcp(h, p) for h, p in [("1.1.1.1", 443), ("8.8.8.8", 53), ("20.184.175.9", 443), ("20.184.175.13", 443), ("172.17.0.1", 80)] + [(ip, 443) for ip in extra_ips]} tcp6 = {f"[{h}]:{p}": attempt_tcp(h, p, socket.AF_INET6) for h, p in [("2606:4700:4700::1111", 443), ("2001:4860:4860::8888", 53)]} udp = {f"{h}:{p}/udp": attempt_udp(h, p) for h, p in [("1.1.1.1", 53), ("8.8.8.8", 53)]} udp6 = {f"[{h}]:{p}/udp": attempt_udp(h, p, socket.AF_INET6) for h, p in [("2606:4700:4700::1111", 53)]} r["attempts"] = {**tcp, **tcp6, **udp, **udp6} # the default resolver path: resolv.conf says 127.0.0.1, where nothing listens yet try: r["gai_before_listeners"] = socket.getaddrinfo(COLLECTOR, 443)[0][4][0] except OSError as e: r["gai_before_listeners"] = repr(e) r["no_route_out"] = (r["interfaces"] == ["lo"] and len(r["route_v4"].splitlines()) == 1 and all(v.startswith("errno 101") for v in r["attempts"].values()) and not r["gai_before_listeners"].startswith("127.") and not r["proxy_env"]) return r def run_traced(code, tag): """Run a python snippet under the bundled strace; return its rc, stderr tail and inet connects.""" sd = "/probe/work/strace" log = f"/tmp/strace-{tag}.log" py = "/probe/work/venvs/v1300/bin/python" cmd = [f"{sd}/ld-linux-x86-64.so.2", "--library-path", sd, f"{sd}/strace", "-f", "--seccomp-bpf", "-e", "trace=%network", "-ttt", "-s", "128", "-o", log, "--", py, "-c", code] env = {"PATH": "/usr/bin:/bin", "HOME": "/tmp/home", "LANG": "C.UTF-8", "PYTHONDONTWRITEBYTECODE": "1"} t0 = time.time() p = subprocess.run(cmd, env=env, capture_output=True, text=True, timeout=60) calls = inside.parse_strace(log, t0) or [] return {"rc": p.returncode, "stdout": p.stdout[-400:], "stderr": p.stderr[-400:], "t0": t0, "inet_calls": [c for c in calls if c["family"] != "AF_UNIX"]} def part2(): inside.start_listeners() time.sleep(0.3) r = {} n0 = len(inside.EVENTS) r["urllib_https"] = run_traced( "import urllib.request\n" "try:\n urllib.request.urlopen('https://" + COLLECTOR + "/OneCollector/1.0/', timeout=5)\n" "except Exception as e: print('client error:', repr(e))", "https") time.sleep(0.5) with inside.LOCK: ev = inside.EVENTS[n0:] r["https_dns"] = [(e["name"], e["qtype"]) for e in ev if e["kind"] == "dns"] r["https_tls"] = [(e.get("sni"), e.get("bytes"), e.get("alpn")) for e in ev if e["kind"] == "tls"] n1 = len(inside.EVENTS) r["urllib_http"] = run_traced( "import urllib.request\n" "try:\n urllib.request.urlopen('http://" + COLLECTOR + "/OneCollector/1.0/', timeout=5)\n" "except Exception as e: print('client error:', repr(e))", "http") time.sleep(0.5) with inside.LOCK: ev = inside.EVENTS[n1:] r["http_dns"] = [(e["name"], e["qtype"]) for e in ev if e["kind"] == "dns"] r["http_log"] = [(e.get("host"), e.get("request_line")) for e in ev if e["kind"] == "http"] n2 = len(inside.EVENTS) r["hardcoded_ip"] = run_traced( "import socket\n" "try:\n socket.create_connection(('20.184.175.9', 443), timeout=3)\n" "except Exception as e: print('client error:', repr(e))", "hardip") time.sleep(0.3) with inside.LOCK: ev = inside.EVENTS[n2:] r["hardcoded_ip_listener_events"] = len(ev) r["positive_controls_pass"] = ( (COLLECTOR, "A") in r["https_dns"] and any(t[0] == COLLECTOR for t in r["https_tls"]) and any(h == COLLECTOR for h, _ in r["http_log"]) and any(c["dst"] == "20.184.175.9:443" and "ENETUNREACH" in str(c["result"]) for c in r["hardcoded_ip"]["inet_calls"])) return r if __name__ == "__main__": OUT["part1_no_route"] = part1(sys.argv[1:]) OUT["part2_positive_controls"] = part2() print(json.dumps(OUT, indent=1, default=str))