add single-host DPI/TCP-RST self-diagnostic probe

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,
+    )

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