dns_leak: per-adapter asn mismatch and ecs leak detection

src/netcheck/probes/dns_leak.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import ipaddress
+
+from netcheck.models import AdapterLeakResult, DnsLeak
+
+_LEAK_NOTE = (
+    "This test only sees resolvers configured at the OS level. A browser using DoH "
+    "or DoT bypasses them entirely and is not covered here."
+)
+
+
+def _unquote(text: str) -> str:
+    return text.strip().strip('"')
+
+
+def parse_akahelp(records: list[str]) -> dict[str, str]:
+    mapping: dict[str, str] = {}
+    for record in records:
+        parts = [_unquote(part) for part in record.replace('" "', '"\t"').split("\t")]
+        if len(parts) != 2:
+            parts = [_unquote(p) for p in record.split(None, 1)]
+        if len(parts) != 2 or not parts[0]:
+            continue
+        mapping[parts[0]] = parts[1]
+    return mapping
+
+
+def parse_myaddr(records: list[str]) -> str | None:
+    for record in records:
+        candidate = _unquote(record)
+        try:
+            ipaddress.ip_address(candidate)
+        except ValueError:
+            continue
+        return candidate
+    return None
+
+
+def detect_ecs_leak(akahelp: dict[str, str]) -> bool:
+    ecs = akahelp.get("ecs", "").strip()
+    if not ecs or ecs.startswith("0.0.0.0/0") or ecs.startswith("::/0"):
+        return False
+    return True
+
+
+def build_adapter_result(
+    adapter: str,
+    resolvers: list[str],
+    echoed_ip: str | None,
+    echoed_asn: str | None,
+    egress_asn: str | None,
+) -> AdapterLeakResult:
+    matches: bool | None = None
+    if echoed_asn and egress_asn:
+        matches = echoed_asn.upper() == egress_asn.upper()
+    return AdapterLeakResult(
+        adapter=adapter,
+        configured_resolvers=list(resolvers),
+        echoed_ip=echoed_ip,
+        echoed_asn=echoed_asn,
+        matches_egress_asn=matches,
+    )
+
+
+def build_dns_leak(results: list[AdapterLeakResult], ecs_leaked: bool) -> DnsLeak:
+    leaking = [r.adapter for r in results if r.matches_egress_asn is False]
+    parts: list[str] = []
+    if leaking:
+        parts.append(
+            f"DNS queries from {', '.join(leaking)} resolve through a network outside the egress ASN."
+        )
+    else:
+        parts.append("No adapter resolves DNS outside the egress ASN.")
+    if ecs_leaked:
+        parts.append("The resolver forwards your EDNS Client Subnet, exposing your network to authoritative servers.")
+    parts.append(_LEAK_NOTE)
+    return DnsLeak(per_adapter=results, ecs_leaked=ecs_leaked, note=" ".join(parts))

tests/test_dns_leak.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+from netcheck.probes.dns_leak import (
+    build_adapter_result,
+    build_dns_leak,
+    detect_ecs_leak,
+    parse_akahelp,
+    parse_myaddr,
+)
+
+
+def test_akahelp_txt_records_become_a_flat_mapping():
+    records = ['"ns" "203.0.113.9"', '"ecs" "198.51.100.0/24"', '"cip" "203.0.113.44"']
+    assert parse_akahelp(records) == {
+        "ns": "203.0.113.9",
+        "ecs": "198.51.100.0/24",
+        "cip": "203.0.113.44",
+    }
+
+
+def test_akahelp_records_without_quotes_are_handled_too():
+    assert parse_akahelp(["ns 203.0.113.9"]) == {"ns": "203.0.113.9"}
+
+
+def test_akahelp_ignores_records_it_cannot_split():
+    assert parse_akahelp(["garbage", '"ns" "1.2.3.4"']) == {"ns": "1.2.3.4"}
+
+
+def test_myaddr_returns_the_echoed_resolver_address():
+    assert parse_myaddr(['"203.0.113.9"']) == "203.0.113.9"
+    assert parse_myaddr(["203.0.113.9"]) == "203.0.113.9"
+
+
+def test_myaddr_of_no_answer_is_none():
+    assert parse_myaddr([]) is None
+    assert parse_myaddr(['"not an ip"']) is None
+
+
+def test_ecs_leak_is_detected_when_a_client_subnet_is_echoed():
+    assert detect_ecs_leak({"ecs": "198.51.100.0/24"}) is True
+
+
+def test_no_ecs_leak_when_the_field_is_absent_or_a_wildcard():
+    assert detect_ecs_leak({}) is False
+    assert detect_ecs_leak({"ecs": ""}) is False
+    assert detect_ecs_leak({"ecs": "0.0.0.0/0"}) is False
+
+
+def test_adapter_result_flags_a_resolver_in_a_different_asn():
+    result = build_adapter_result(
+        adapter="Wi-Fi",
+        resolvers=["192.168.1.1"],
+        echoed_ip="203.0.113.9",
+        echoed_asn="AS64501",
+        egress_asn="AS64500",
+    )
+    assert result.matches_egress_asn is False
+    assert result.adapter == "Wi-Fi"
+    assert result.configured_resolvers == ["192.168.1.1"]
+
+
+def test_adapter_result_is_clean_when_the_asns_agree():
+    result = build_adapter_result("wg0", ["10.7.0.1"], "203.0.113.44", "AS64500", "AS64500")
+    assert result.matches_egress_asn is True
+
+
+def test_adapter_result_is_unknown_when_either_asn_is_missing():
+    assert build_adapter_result("eth0", ["1.1.1.1"], "1.1.1.1", None, "AS64500").matches_egress_asn is None
+    assert build_adapter_result("eth0", ["1.1.1.1"], "1.1.1.1", "AS13335", None).matches_egress_asn is None
+
+
+def test_adapter_result_asn_comparison_is_case_insensitive():
+    assert build_adapter_result("eth0", ["1.1.1.1"], "1.1.1.1", "as64500", "AS64500").matches_egress_asn is True
+
+
+def test_dns_leak_note_names_the_leaking_adapter():
+    leaking = build_adapter_result("Wi-Fi", ["192.168.1.1"], "203.0.113.9", "AS64501", "AS64500")
+    clean = build_adapter_result("wg0", ["10.7.0.1"], "203.0.113.44", "AS64500", "AS64500")
+    leak = build_dns_leak([clean, leaking], ecs_leaked=False)
+    assert leak.ecs_leaked is False
+    assert "Wi-Fi" in leak.note
+    assert "browser" in leak.note.lower()
+
+
+def test_dns_leak_note_is_clean_when_every_adapter_agrees():
+    clean = build_adapter_result("wg0", ["10.7.0.1"], "203.0.113.44", "AS64500", "AS64500")
+    leak = build_dns_leak([clean], ecs_leaked=False)
+    assert "no adapter" in leak.note.lower()
+
+
+def test_dns_leak_note_mentions_ecs_when_the_subnet_leaked():
+    clean = build_adapter_result("wg0", ["10.7.0.1"], "203.0.113.44", "AS64500", "AS64500")
+    leak = build_dns_leak([clean], ecs_leaked=True)
+    assert leak.ecs_leaked is True
+    assert "client subnet" in leak.note.lower()