add system DNS vs DoH comparison and transparent proxy detection

src/netsleuth/probes/dns_advanced.py

@@ -0,0 +1,184 @@
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import time
+
+import dns.asyncquery
+import dns.asyncresolver
+import dns.exception
+import dns.message
+import httpx
+
+from netsleuth.models import DnsAdvanced, ResolverProbe
+
+_A_RECORD_TYPE = 1
+_AAAA_RECORD_TYPE = 28
+
+
+def parse_doh_json(payload: dict) -> list[str]:
+    if payload.get("Status") != 0:
+        return []
+    answers = payload.get("Answer") or []
+    return [
+        record["data"]
+        for record in answers
+        if record.get("type") in (_A_RECORD_TYPE, _AAAA_RECORD_TYPE)
+    ]
+
+
+def is_suspicious_answer(addr: str) -> bool:
+    if addr == "0.0.0.0":
+        return True
+    try:
+        ip = ipaddress.ip_address(addr)
+    except ValueError:
+        return True
+    return ip.is_private or ip.is_loopback or ip.is_reserved
+
+
+def compare_answers(system_probe: ResolverProbe, doh_probes: list[ResolverProbe]) -> list[str]:
+    divergences: list[str] = []
+    system_set = set(system_probe.answers)
+    for doh_probe in doh_probes:
+        doh_set = set(doh_probe.answers)
+        if system_set == doh_set:
+            continue
+        suspicious = detect_poisoning(system_probe, [doh_probe])
+        divergences.append(
+            f"{system_probe.query_name}: system={sorted(system_set)} "
+            f"vs {doh_probe.name}={sorted(doh_set)} (suspicious={suspicious})"
+        )
+    return divergences
+
+
+def detect_poisoning(system_probe: ResolverProbe, doh_probes: list[ResolverProbe]) -> bool:
+    system_suspicious = any(is_suspicious_answer(a) for a in system_probe.answers)
+    if not system_suspicious:
+        return False
+    return any(
+        any(not is_suspicious_answer(a) for a in doh_probe.answers) for doh_probe in doh_probes
+    )
+
+
+def detect_transparent_proxy(bogus_probe: ResolverProbe) -> tuple[bool | None, str]:
+    if bogus_probe.error is None and bogus_probe.answers:
+        return True, "A response was received from the bogus resolver IP: transparent DNS proxy detected."
+    return False, "No response from the bogus resolver IP (timeout): transparent DNS proxy not detected."
+
+
+async def resolve_system(name: str, timeout: float) -> ResolverProbe:
+    resolver = dns.asyncresolver.Resolver()
+    resolver.lifetime = timeout
+    start = time.monotonic()
+    try:
+        answer = await resolver.resolve(name, "A")
+        elapsed_ms = (time.monotonic() - start) * 1000
+        answers = [rdata.address for rdata in answer]
+        return ResolverProbe(
+            name="system",
+            kind="system",
+            query_name=name,
+            answers=answers,
+            elapsed_ms=elapsed_ms,
+        )
+    except dns.exception.DNSException as exc:
+        return ResolverProbe(name="system", kind="system", query_name=name, error=str(exc))
+
+
+async def resolve_doh(
+    client: httpx.AsyncClient,
+    name: str,
+    url: str,
+    query_name: str,
+    timeout: float,
+) -> ResolverProbe:
+    start = time.monotonic()
+    try:
+        response = await client.get(
+            url,
+            params={"name": query_name, "type": "A"},
+            headers={"accept": "application/dns-json"},
+            timeout=timeout,
+        )
+        response.raise_for_status()
+        elapsed_ms = (time.monotonic() - start) * 1000
+        answers = parse_doh_json(response.json())
+        return ResolverProbe(
+            name=name,
+            kind="doh",
+            endpoint=url,
+            query_name=query_name,
+            answers=answers,
+            elapsed_ms=elapsed_ms,
+        )
+    except httpx.HTTPError as exc:
+        return ResolverProbe(name=name, kind="doh", endpoint=url, query_name=query_name, error=str(exc))
+
+
+async def probe_bogus_resolver(ip: str, name: str, timeout: float) -> ResolverProbe:
+    query = dns.message.make_query(name, "A")
+    try:
+        response = await dns.asyncquery.udp(query, ip, timeout=timeout)
+        answers = [
+            rdata.address
+            for rrset in response.answer
+            for rdata in rrset
+            if rdata.rdtype == 1
+        ]
+        return ResolverProbe(name="bogus", kind="system", query_name=name, answers=answers)
+    except dns.exception.Timeout:
+        return ResolverProbe(name="bogus", kind="system", query_name=name, error="timeout")
+    except (dns.exception.DNSException, OSError) as exc:
+        return ResolverProbe(name="bogus", kind="system", query_name=name, error=str(exc))
+
+
+async def collect_dns_advanced(
+    client: httpx.AsyncClient,
+    control_names: list[str],
+    doh_endpoints,
+    bogus_resolver_ip: str,
+    bogus_probe_name: str,
+    timeout: float,
+) -> DnsAdvanced:
+    system_probes = [await resolve_system(name, timeout) for name in control_names]
+    doh_probes: list[ResolverProbe] = []
+    for name in control_names:
+        doh_probes.extend(
+            await asyncio.gather(
+                *(resolve_doh(client, ep.name, ep.url, name, timeout) for ep in doh_endpoints)
+            )
+        )
+
+    divergences: list[str] = []
+    for system_probe in system_probes:
+        matching_doh = [p for p in doh_probes if p.query_name == system_probe.query_name]
+        divergences.extend(compare_answers(system_probe, matching_doh))
+
+    system_times = [p.elapsed_ms for p in system_probes if p.elapsed_ms is not None]
+    doh_times = [p.elapsed_ms for p in doh_probes if p.elapsed_ms is not None]
+    system_avg_ms = sum(system_times) / len(system_times) if system_times else None
+    doh_avg_ms = sum(doh_times) / len(doh_times) if doh_times else None
+
+    bogus_probe = await probe_bogus_resolver(bogus_resolver_ip, bogus_probe_name, timeout)
+    transparent_proxy, transparent_proxy_detail = detect_transparent_proxy(bogus_probe)
+
+    note = (
+        "Compared system resolver answers and timing against DoH resolvers on control names, "
+        "and probed a bogus resolver IP to check for a transparent DNS proxy."
+    )
+    note_ru = (
+        "Сравнили ответы и время системного резолвера с DoH-резолверами на контрольных именах, "
+        "а также проверили фиктивный IP резолвера на признаки прозрачного DNS-прокси."
+    )
+
+    return DnsAdvanced(
+        probes=system_probes + doh_probes,
+        system_avg_ms=system_avg_ms,
+        doh_avg_ms=doh_avg_ms,
+        divergences=divergences,
+        transparent_proxy=transparent_proxy,
+        transparent_proxy_detail=transparent_proxy_detail,
+        note=note,
+        note_ru=note_ru,
+    )

tests/fixtures/api/doh_cloudflare.json

@@ -0,0 +1,8 @@
+{
+  "Status": 0,
+  "Question": [{"name": "cloudflare.com.", "type": 1}],
+  "Answer": [
+    {"name": "cloudflare.com.", "type": 1, "TTL": 300, "data": "104.16.132.229"},
+    {"name": "cloudflare.com.", "type": 1, "TTL": 300, "data": "104.16.133.229"}
+  ]
+}

tests/fixtures/api/doh_cname_then_a.json

@@ -0,0 +1,8 @@
+{
+  "Status": 0,
+  "Question": [{"name": "www.example.com.", "type": 1}],
+  "Answer": [
+    {"name": "www.example.com.", "type": 5, "TTL": 300, "data": "example.com."},
+    {"name": "example.com.", "type": 1, "TTL": 300, "data": "93.184.216.34"}
+  ]
+}

tests/fixtures/api/doh_nxdomain.json

@@ -0,0 +1,4 @@
+{
+  "Status": 3,
+  "Question": [{"name": "nonexistent.invalid.", "type": 1}]
+}

tests/test_dns_advanced.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from netsleuth.models import ResolverProbe
+from netsleuth.probes.dns_advanced import (
+    compare_answers,
+    detect_poisoning,
+    detect_transparent_proxy,
+    is_suspicious_answer,
+    parse_doh_json,
+    resolve_doh,
+)
+
+
+def test_parse_doh_json_extracts_addresses_from_a_successful_answer(api_fixture):
+    payload = api_fixture("doh_cloudflare.json")
+    assert parse_doh_json(payload) == ["104.16.132.229", "104.16.133.229"]
+
+
+def test_parse_doh_json_returns_empty_list_on_nxdomain_status(api_fixture):
+    payload = api_fixture("doh_nxdomain.json")
+    assert parse_doh_json(payload) == []
+
+
+def test_parse_doh_json_ignores_cname_and_keeps_only_a_records(api_fixture):
+    payload = api_fixture("doh_cname_then_a.json")
+    assert parse_doh_json(payload) == ["93.184.216.34"]
+
+
+def test_parse_doh_json_returns_empty_list_on_empty_payload():
+    assert parse_doh_json({}) == []
+
+
+@pytest.mark.parametrize(
+    "addr,expected",
+    [
+        ("0.0.0.0", True),
+        ("127.0.0.1", True),
+        ("10.1.1.1", True),
+        ("192.0.2.1", True),
+        ("1.1.1.1", False),
+        ("104.16.132.229", False),
+    ],
+)
+def test_is_suspicious_answer_table(addr, expected):
+    assert is_suspicious_answer(addr) is expected
+
+
+def test_cdn_answers_that_merely_differ_are_not_flagged_as_poisoning():
+    system_probe = ResolverProbe(name="system", kind="system", answers=["1.1.1.1"])
+    doh_probes = [ResolverProbe(name="cloudflare", kind="doh", answers=["104.16.132.229"])]
+    assert detect_poisoning(system_probe, doh_probes) is False
+
+
+def test_a_bogon_system_answer_against_a_clean_doh_answer_is_flagged_as_poisoning():
+    system_probe = ResolverProbe(name="system", kind="system", answers=["192.0.2.1"])
+    doh_probes = [ResolverProbe(name="cloudflare", kind="doh", answers=["1.1.1.1"])]
+    assert detect_poisoning(system_probe, doh_probes) is True
+
+
+def test_detect_poisoning_is_false_when_both_sides_are_suspicious():
+    system_probe = ResolverProbe(name="system", kind="system", answers=["0.0.0.0"])
+    doh_probes = [ResolverProbe(name="cloudflare", kind="doh", answers=["127.0.0.1"])]
+    assert detect_poisoning(system_probe, doh_probes) is False
+
+
+def test_detect_poisoning_is_false_when_both_sides_are_empty():
+    system_probe = ResolverProbe(name="system", kind="system", answers=[])
+    doh_probes = [ResolverProbe(name="cloudflare", kind="doh", answers=[])]
+    assert detect_poisoning(system_probe, doh_probes) is False
+
+
+def test_compare_answers_reports_a_divergence_when_addresses_differ():
+    system_probe = ResolverProbe(name="system", kind="system", query_name="cloudflare.com", answers=["1.1.1.1"])
+    doh_probes = [
+        ResolverProbe(name="cloudflare", kind="doh", query_name="cloudflare.com", answers=["104.16.132.229"])
+    ]
+    divergences = compare_answers(system_probe, doh_probes)
+    assert len(divergences) == 1
+    assert "cloudflare.com" in divergences[0]
+
+
+def test_compare_answers_returns_empty_list_when_answers_match():
+    system_probe = ResolverProbe(name="system", kind="system", query_name="cloudflare.com", answers=["1.1.1.1"])
+    doh_probes = [ResolverProbe(name="cloudflare", kind="doh", query_name="cloudflare.com", answers=["1.1.1.1"])]
+    assert compare_answers(system_probe, doh_probes) == []
+
+
+def test_detect_transparent_proxy_true_when_bogus_ip_answers():
+    bogus_probe = ResolverProbe(name="bogus", kind="system", answers=["93.184.216.34"], error=None)
+    detected, detail = detect_transparent_proxy(bogus_probe)
+    assert detected is True
+    assert "detect" in detail.lower()
+
+
+def test_detect_transparent_proxy_false_and_says_not_detected_not_absent_on_timeout():
+    bogus_probe = ResolverProbe(name="bogus", kind="system", answers=[], error="timeout")
+    detected, detail = detect_transparent_proxy(bogus_probe)
+    assert detected is False
+    lowered = detail.lower()
+    assert "not detected" in lowered or "не обнаружен" in lowered
+    assert "absent" not in lowered
+    assert "none" not in lowered
+
+
+@pytest.mark.asyncio
+async def test_resolve_doh_populates_answers_from_a_mocked_response(httpx_mock, api_fixture):
+    httpx_mock.add_response(
+        url="https://cloudflare-dns.com/dns-query?name=cloudflare.com&type=A",
+        json=api_fixture("doh_cloudflare.json"),
+    )
+    async with httpx.AsyncClient() as client:
+        probe = await resolve_doh(client, "cloudflare", "https://cloudflare-dns.com/dns-query", "cloudflare.com", 4.0)
+    assert probe.kind == "doh"
+    assert probe.answers == ["104.16.132.229", "104.16.133.229"]
+    assert probe.error is None
+
+
+@pytest.mark.asyncio
+async def test_resolve_doh_sets_error_on_http_error_status_without_raising(httpx_mock):
+    httpx_mock.add_response(
+        url="https://cloudflare-dns.com/dns-query?name=cloudflare.com&type=A",
+        status_code=500,
+    )
+    async with httpx.AsyncClient() as client:
+        probe = await resolve_doh(client, "cloudflare", "https://cloudflare-dns.com/dns-query", "cloudflare.com", 4.0)
+    assert probe.error is not None
+    assert probe.answers == []
+
+
+@pytest.mark.asyncio
+async def test_resolve_doh_sets_error_on_timeout_without_raising(httpx_mock):
+    httpx_mock.add_exception(httpx.TimeoutException("timed out"))
+    async with httpx.AsyncClient() as client:
+        probe = await resolve_doh(client, "cloudflare", "https://cloudflare-dns.com/dns-query", "cloudflare.com", 4.0)
+    assert probe.error is not None
+    assert probe.answers == []