diff --git a/src/netcheck/probes/dns_leak.py b/src/netcheck/probes/dns_leak.py new file mode 100644 index 0000000..68a40fe --- /dev/null +++ b/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)) diff --git a/tests/test_dns_leak.py b/tests/test_dns_leak.py new file mode 100644 index 0000000..5754034 --- /dev/null +++ b/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()