From a000a12b193be2f3951c758b9e6eb98950c4e7f7 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 17:45:27 +0300 Subject: [PATCH] add single-host DPI/TCP-RST self-diagnostic probe --- src/netsleuth/probes/dpi_check.py | 140 ++++++++++++++++++++++++++ tests/test_dpi_check.py | 161 ++++++++++++++++++++++++++++++ 2 files changed, 301 insertions(+) create mode 100644 src/netsleuth/probes/dpi_check.py create mode 100644 tests/test_dpi_check.py diff --git a/src/netsleuth/probes/dpi_check.py b/src/netsleuth/probes/dpi_check.py new file mode 100644 index 0000000..c7734a1 --- /dev/null +++ b/src/netsleuth/probes/dpi_check.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +import asyncio +import time + +from netsleuth.models import DpiCheckResult, PortProbe + + +def classify_connect_outcome(exc: BaseException | None) -> str: + if exc is None: + return "open" + if isinstance(exc, ConnectionResetError): + return "reset" + if isinstance(exc, ConnectionRefusedError): + return "closed" + if isinstance(exc, (asyncio.TimeoutError, TimeoutError)): + return "filtered" + return "error" + + +def dpi_verdict(ports: list[PortProbe]) -> tuple[str, str, str]: + states = [p.state for p in ports] + + if not states or all(s == "filtered" for s in states): + rationale = ( + "All probed ports were silently dropped (or the list of ports was empty). " + "This could mean the server is offline, or that a middlebox is filtering " + "traffic to it — the two are indistinguishable from the client side, so no " + "DPI verdict can be asserted." + ) + rationale_ru = ( + "Все проверенные порты не ответили (пакеты тихо потерялись), или список " + "портов был пуст. Это может означать, что сервер выключен, или что " + "провайдер/middlebox фильтрует трафик к нему — со стороны клиента отличить " + "одно от другого нельзя, поэтому уверенный вывод о DPI сделать нельзя." + ) + return "unreachable", rationale, rationale_ru + + has_open = "open" in states + has_reset = "reset" in states + has_filtered = "filtered" in states + + if has_reset and has_open: + rationale = ( + "At least one port responded normally (open) while another had its " + "connection actively reset. Active RST injection alongside working ports " + "is a characteristic signature of DPI-based interference." + ) + rationale_ru = ( + "Хотя бы один порт ответил нормально (открыт), а на другом соединение было " + "принудительно сброшено (RST). Активная инъекция RST на фоне работающих " + "портов — характерный признак вмешательства DPI." + ) + return "reset_injection", rationale, rationale_ru + + if has_open and has_filtered: + rationale = ( + "Some ports responded (open) while others were silently dropped (filtered). " + "This mixed pattern suggests selective filtering of specific ports/services " + "rather than a fully offline host." + ) + rationale_ru = ( + "Часть портов ответила (открыты), а часть тихо не ответила (отфильтрованы). " + "Такая смешанная картина указывает на выборочную фильтрацию отдельных " + "портов/сервисов, а не на полностью выключенный хост." + ) + return "partial_filtering", rationale, rationale_ru + + rationale = ( + "Every probed port returned a clean, honest TCP outcome (open or refused) with " + "no silent drops or connection resets observed." + ) + rationale_ru = ( + "Все проверенные порты дали честный TCP-ответ (открыт или отказ), без тихих " + "потерь пакетов или сбросов соединения." + ) + return "clean", rationale, rationale_ru + + +async def probe_port( + ip: str, port: int, *, timeout: float, source_ip: str | None = None +) -> PortProbe: + local_addr = (source_ip, 0) if source_ip else None + began = time.perf_counter() + writer = None + try: + _reader, writer = await asyncio.wait_for( + asyncio.open_connection(ip, port, local_addr=local_addr), timeout=timeout + ) + rtt_ms = (time.perf_counter() - began) * 1000.0 + return PortProbe(port=port, state="open", rtt_ms=rtt_ms) + except asyncio.CancelledError: + raise + except BaseException as exc: # noqa: BLE001 - a failed probe is a value here, not control flow + state = classify_connect_outcome(exc) + detail = str(exc) or exc.__class__.__name__ + return PortProbe(port=port, state=state, detail=detail) + finally: + if writer is not None: + writer.close() + try: + await writer.wait_closed() + except OSError: + pass + + +async def check_dpi( + target: str, + resolved_ip: str, + ports: list[int], + *, + timeout: float, + delay: float, + concurrency: int, + source_ip: str | None = None, +) -> DpiCheckResult: + semaphore = asyncio.Semaphore(concurrency) + + async def _bounded(port: int) -> PortProbe: + async with semaphore: + return await probe_port(resolved_ip, port, timeout=timeout, source_ip=source_ip) + + tasks = [] + for index, port in enumerate(ports): + if index: + await asyncio.sleep(delay) + tasks.append(asyncio.ensure_future(_bounded(port))) + + probes = list(await asyncio.gather(*tasks)) if tasks else [] + + verdict, rationale, rationale_ru = dpi_verdict(probes) + return DpiCheckResult( + target=target, + resolved_ip=resolved_ip, + consented=True, + ports=probes, + verdict=verdict, + rationale=rationale, + rationale_ru=rationale_ru, + ) diff --git a/tests/test_dpi_check.py b/tests/test_dpi_check.py new file mode 100644 index 0000000..84991de --- /dev/null +++ b/tests/test_dpi_check.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import asyncio +import inspect + +import pytest + +from netsleuth.models import PortProbe +from netsleuth.probes.dpi_check import ( + check_dpi, + classify_connect_outcome, + dpi_verdict, + probe_port, +) + +_CYRILLIC = set("абвгдеёжзийклмнопрстуфхцчшщъыьэюя") + + +def has_cyrillic(text: str) -> bool: + return any(ch in _CYRILLIC for ch in text.lower()) + + +@pytest.mark.parametrize( + ("exc", "expected"), + [ + (None, "open"), + (ConnectionResetError(), "reset"), + (ConnectionRefusedError(), "closed"), + (TimeoutError(), "filtered"), + (OSError(), "error"), + ], +) +def test_classify_connect_outcome(exc, expected): + assert classify_connect_outcome(exc) == expected + + +def test_dpi_verdict_clean_when_all_open_or_closed(): + ports = [ + PortProbe(port=80, state="open"), + PortProbe(port=443, state="closed"), + ] + verdict, rationale, rationale_ru = dpi_verdict(ports) + assert verdict == "clean" + assert rationale + assert has_cyrillic(rationale_ru) + + +def test_dpi_verdict_reset_injection_when_reset_seen_among_open(): + ports = [ + PortProbe(port=80, state="open"), + PortProbe(port=443, state="reset"), + ] + verdict, rationale, rationale_ru = dpi_verdict(ports) + assert verdict == "reset_injection" + assert has_cyrillic(rationale_ru) + + +def test_dpi_verdict_partial_filtering_for_mixed_open_and_filtered(): + ports = [ + PortProbe(port=80, state="open"), + PortProbe(port=443, state="filtered"), + ] + verdict, rationale, rationale_ru = dpi_verdict(ports) + assert verdict == "partial_filtering" + assert has_cyrillic(rationale_ru) + + +def test_dpi_verdict_unreachable_when_all_filtered_is_cautious_about_cause(): + ports = [ + PortProbe(port=80, state="filtered"), + PortProbe(port=443, state="filtered"), + ] + verdict, rationale, rationale_ru = dpi_verdict(ports) + assert verdict == "unreachable" + assert "or" in rationale.lower() + assert "или" in rationale_ru + + +def test_dpi_verdict_unreachable_when_ports_empty(): + verdict, rationale, rationale_ru = dpi_verdict([]) + assert verdict == "unreachable" + assert "or" in rationale.lower() + assert "или" in rationale_ru + + +@pytest.mark.parametrize( + "verdict_ports", + [ + [PortProbe(port=80, state="open")], + [PortProbe(port=80, state="reset"), PortProbe(port=443, state="open")], + [PortProbe(port=80, state="filtered"), PortProbe(port=443, state="open")], + [PortProbe(port=80, state="filtered")], + [], + ], +) +def test_dpi_verdict_rationale_ru_always_nonempty_and_cyrillic(verdict_ports): + _verdict, _rationale, rationale_ru = dpi_verdict(verdict_ports) + assert rationale_ru + assert has_cyrillic(rationale_ru) + + +def test_port_probe_rejects_unknown_state(): + with pytest.raises(ValueError): + PortProbe(port=443, state="bogus") + + +def test_check_dpi_refuses_a_target_that_was_not_explicitly_provided(): + sig = inspect.signature(check_dpi) + assert sig.parameters["target"].default is inspect.Parameter.empty + assert sig.parameters["resolved_ip"].default is inspect.Parameter.empty + + +@pytest.mark.asyncio +async def test_probe_port_reports_open_then_closed_on_same_loopback_port(): + server = await asyncio.start_server(lambda r, w: None, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + + open_probe = await probe_port("127.0.0.1", port, timeout=2.0) + assert open_probe.state == "open" + + server.close() + await server.wait_closed() + + # Windows loopback quirk: the first SYN to a just-closed port is dropped and + # ECONNREFUSED only surfaces after the ~2s retransmit timer fires, so this + # needs headroom above 2s to stay deterministic (Linux resolves near-instantly). + closed_probe = await probe_port("127.0.0.1", port, timeout=5.0) + assert closed_probe.state == "closed" + + +@pytest.mark.asyncio +async def test_check_dpi_produces_a_probe_per_port_with_a_consistent_verdict(): + server = await asyncio.start_server(lambda r, w: None, "127.0.0.1", 0) + open_port = server.sockets[0].getsockname()[1] + + closed_server = await asyncio.start_server(lambda r, w: None, "127.0.0.1", 0) + closed_port = closed_server.sockets[0].getsockname()[1] + closed_server.close() + await closed_server.wait_closed() + + result = await check_dpi( + "example.test", + "127.0.0.1", + [open_port, closed_port], + timeout=5.0, + delay=0.0, + concurrency=2, + ) + + server.close() + await server.wait_closed() + + assert result.target == "example.test" + assert result.resolved_ip == "127.0.0.1" + assert result.consented is True + assert len(result.ports) == 2 + + expected_verdict, expected_rationale, expected_rationale_ru = dpi_verdict(result.ports) + assert result.verdict == expected_verdict + assert result.rationale == expected_rationale + assert result.rationale_ru == expected_rationale_ru