From 63e7597520b5eaa4dcb50ab41c5eb73d4a901f0e Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 17:45:33 +0300 Subject: [PATCH] add system DNS vs DoH comparison and transparent proxy detection --- src/netsleuth/probes/dns_advanced.py | 184 +++++++++++++++++++++++ tests/fixtures/api/doh_cloudflare.json | 8 + tests/fixtures/api/doh_cname_then_a.json | 8 + tests/fixtures/api/doh_nxdomain.json | 4 + tests/test_dns_advanced.py | 139 +++++++++++++++++ 5 files changed, 343 insertions(+) create mode 100644 src/netsleuth/probes/dns_advanced.py create mode 100644 tests/fixtures/api/doh_cloudflare.json create mode 100644 tests/fixtures/api/doh_cname_then_a.json create mode 100644 tests/fixtures/api/doh_nxdomain.json create mode 100644 tests/test_dns_advanced.py diff --git a/src/netsleuth/probes/dns_advanced.py b/src/netsleuth/probes/dns_advanced.py new file mode 100644 index 0000000..e1f2c22 --- /dev/null +++ b/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, + ) diff --git a/tests/fixtures/api/doh_cloudflare.json b/tests/fixtures/api/doh_cloudflare.json new file mode 100644 index 0000000..5118608 --- /dev/null +++ b/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"} + ] +} diff --git a/tests/fixtures/api/doh_cname_then_a.json b/tests/fixtures/api/doh_cname_then_a.json new file mode 100644 index 0000000..d4d27d8 --- /dev/null +++ b/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"} + ] +} diff --git a/tests/fixtures/api/doh_nxdomain.json b/tests/fixtures/api/doh_nxdomain.json new file mode 100644 index 0000000..f1c774a --- /dev/null +++ b/tests/fixtures/api/doh_nxdomain.json @@ -0,0 +1,4 @@ +{ + "Status": 3, + "Question": [{"name": "nonexistent.invalid.", "type": 1}] +} diff --git a/tests/test_dns_advanced.py b/tests/test_dns_advanced.py new file mode 100644 index 0000000..4247af6 --- /dev/null +++ b/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 == []