From fc502f1756fb8fe5a4342d7aa1ff167586d55699 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 17:49:30 +0300 Subject: [PATCH] add finding rules for TLS, prefix bench, DPI, DNS advanced, path diversity --- src/netsleuth/interpret.py | 251 ++++++++++++++++++++++++++++++++++ tests/test_interpret_l7.py | 271 +++++++++++++++++++++++++++++++++++++ 2 files changed, 522 insertions(+) create mode 100644 tests/test_interpret_l7.py diff --git a/src/netsleuth/interpret.py b/src/netsleuth/interpret.py index c3fc2a0..3946d30 100644 --- a/src/netsleuth/interpret.py +++ b/src/netsleuth/interpret.py @@ -4,13 +4,18 @@ from __future__ import annotations from netsleuth.config import Band, BufferbloatBands, Thresholds, VpnBands from netsleuth.models import ( CfTrace, + DnsAdvanced, DnsLeak, + DpiCheckResult, Finding, IpGeo, LocalNet, + PathDiversity, PingResult, + PrefixBenchmark, Signal, SpeedResult, + TlsResult, TraceResult, VpnAssessment, VpnContext, @@ -334,6 +339,252 @@ def speed_findings(speed: SpeedResult, bands: BufferbloatBands) -> list[Finding] return findings +def _cpu_bound_ratio(tls_handshake_ms: float | None, tcp_rtt_ms: float | None) -> float | None: + if tls_handshake_ms is None or not tcp_rtt_ms: + return None + return tls_handshake_ms / tcp_rtt_ms + + +def tls_findings(results: list[TlsResult], t: Thresholds) -> list[Finding]: + findings: list[Finding] = [] + for r in results: + if r.error: + findings.append( + Finding( + id=f"tls.unreachable.{r.label}", + severity="warn", + title=f"TLS handshake to {r.host} failed", + detail=f"Could not complete a TLS connection to {r.host}:{r.port}: {r.error}", + metric="error", + value=r.error, + advice="The service may be down, or the port may be blocked between here and the server.", + title_ru=f"TLS-рукопожатие с {r.host} не удалось", + detail_ru=f"Не удалось установить TLS-соединение с {r.host}:{r.port}: {r.error}", + advice_ru="Сервис может быть недоступен, либо порт заблокирован на пути к серверу.", + ) + ) + continue + ratio = _cpu_bound_ratio(r.tls_handshake_ms, r.tcp_rtt_ms) + if ratio is not None and ratio > t.tls_cpu_bound_ratio and (r.tls_handshake_ms or 0.0) >= 20.0: + findings.append( + Finding( + id=f"tls.server_cpu_bound.{r.label}", + severity="warn", + title=f"TLS handshake to {r.host} is disproportionately slow", + detail=f"TLS handshake took {r.tls_handshake_ms} ms, {ratio:.1f}x the {r.tcp_rtt_ms} ms TCP RTT.", + metric="tls_handshake_ms", + value=r.tls_handshake_ms, + threshold=t.tls_cpu_bound_ratio, + advice="Slow handshake relative to network RTT usually points at a CPU-bound TLS terminator, not the network.", + title_ru=f"TLS-рукопожатие с {r.host} непропорционально медленное", + detail_ru=f"TLS-рукопожатие заняло {r.tls_handshake_ms} мс — в {ratio:.1f} раза больше TCP RTT ({r.tcp_rtt_ms} мс).", + advice_ru="Медленное рукопожатие относительно сетевого RTT обычно указывает на упирающийся в CPU TLS-терминатор, а не на сеть.", + ) + ) + severity = severity_for(r.tls_handshake_ms, t.tls_handshake_ms) + if severity not in ("ok", "info"): + findings.append( + Finding( + id=f"tls.handshake_slow.{r.label}", + severity=severity, + title=f"TLS handshake to {r.host} above target", + detail=f"TLS handshake took {r.tls_handshake_ms} ms.", + metric="tls_handshake_ms", + value=r.tls_handshake_ms, + threshold=t.tls_handshake_ms.warn, + advice="Check server load and TLS session resumption settings.", + title_ru=f"TLS-рукопожатие с {r.host} выше целевого", + detail_ru=f"TLS-рукопожатие заняло {r.tls_handshake_ms} мс.", + advice_ru="Проверьте нагрузку на сервер и настройки возобновления TLS-сессий.", + ) + ) + severity = severity_for(r.ttfb_ms, t.ttfb_ms) + if severity not in ("ok", "info"): + findings.append( + Finding( + id=f"tls.ttfb_slow.{r.label}", + severity=severity, + title=f"Time to first byte from {r.host} above target", + detail=f"TTFB was {r.ttfb_ms} ms.", + metric="ttfb_ms", + value=r.ttfb_ms, + threshold=t.ttfb_ms.warn, + advice="Slow TTFB with a fast handshake points at server-side processing, not the network.", + title_ru=f"Время до первого байта от {r.host} выше целевого", + detail_ru=f"TTFB составило {r.ttfb_ms} мс.", + advice_ru="Медленный TTFB при быстром рукопожатии указывает на обработку на стороне сервера, а не на сеть.", + ) + ) + return findings + + +def prefix_findings(bench: PrefixBenchmark, t: Thresholds) -> list[Finding]: + findings: list[Finding] = [] + if not bench.results: + return findings + severity = severity_for(bench.spread_ms, t.prefix_spread_ms) + if severity not in ("ok", "info"): + findings.append( + Finding( + id="prefix.spread", + severity=severity, + title="Latency spread across AS prefixes is high", + detail=f"Spread between best ({bench.best}) and worst ({bench.worst}) probed prefix is {bench.spread_ms} ms.", + metric="spread_ms", + value=bench.spread_ms, + threshold=t.prefix_spread_ms.warn, + advice="Pick the best-performing prefix's PoP as your entry point if your provider lets you choose.", + title_ru="Разброс задержки между префиксами AS велик", + detail_ru=f"Разброс между лучшим ({bench.best}) и худшим ({bench.worst}) префиксом составляет {bench.spread_ms} мс.", + advice_ru="Если провайдер позволяет выбрать точку входа, выбирайте PoP с лучшим префиксом.", + ) + ) + unreachable = sum(1 for p in bench.results if not p.reachable) + if unreachable / len(bench.results) > 0.7: + reachable_pct = round(100 * (1 - unreachable / len(bench.results)), 1) + findings.append( + Finding( + id="prefix.mostly_unreachable", + severity="info", + title="Most probed prefixes did not answer", + detail=f"{unreachable} of {len(bench.results)} probed prefixes gave no ICMP reply.", + metric="reachable_pct", + value=reachable_pct, + advice="Many networks filter ICMP to the first host of a subnet; this is common and not necessarily a fault.", + title_ru="Большинство проверенных префиксов не ответили", + detail_ru=f"{unreachable} из {len(bench.results)} проверенных префиксов не дали ответа по ICMP.", + advice_ru="Многие сети фильтруют ICMP к первому хосту подсети; это обычное явление, не обязательно неисправность.", + ) + ) + return findings + + +_DPI_TEXT = { + "reset_injection": ("crit", "Possible active DPI interference on {t}", "Возможное активное вмешательство DPI на {t}"), + "partial_filtering": ("warn", "Selective port filtering detected on {t}", "Обнаружена выборочная фильтрация портов на {t}"), + "unreachable": ("info", "{t} did not respond on any probed port", "{t} не ответил ни на одном проверенном порту"), +} + + +def dpi_findings(result: DpiCheckResult) -> list[Finding]: + spec = _DPI_TEXT.get(result.verdict) + if spec is None: + return [] + severity, title, title_ru = spec + title = title.format(t=result.target) + title_ru = title_ru.format(t=result.target) + return [ + Finding( + id=f"dpi.{result.verdict}", + severity=severity, + title=title, + detail=result.rationale, + metric="verdict", + value=result.verdict, + advice="Try the same ports over a VPN/proxy to confirm whether the block is on the path to this specific server.", + title_ru=title_ru, + detail_ru=result.rationale_ru or "", + advice_ru="Попробуйте те же порты через VPN/прокси, чтобы подтвердить, что блокировка именно на пути к этому серверу.", + ) + ] + + +def dns_advanced_findings(adv: DnsAdvanced, t: Thresholds) -> list[Finding]: + findings: list[Finding] = [] + if adv.transparent_proxy is True: + findings.append( + Finding( + id="dns.transparent_proxy", + severity="warn", + title="ISP appears to intercept DNS traffic on port 53", + detail=adv.transparent_proxy_detail or "", + advice="A transparent DNS proxy can silently redirect or rewrite lookups even when you configure a different resolver.", + title_ru="Провайдер, похоже, перехватывает DNS-трафик на 53 порту", + detail_ru=adv.transparent_proxy_detail or "", + advice_ru="Прозрачный DNS-прокси может незаметно перенаправлять или подменять запросы, даже если вы указали другой резолвер.", + ) + ) + if adv.system_avg_ms is not None and adv.doh_avg_ms is not None: + severity = severity_for(adv.system_avg_ms, t.dns_resolve_ms) + if severity not in ("ok", "info") and adv.system_avg_ms > adv.doh_avg_ms: + findings.append( + Finding( + id="dns.system_slow", + severity=severity, + title="System DNS resolver is slow compared to DoH", + detail=f"System resolver averaged {adv.system_avg_ms} ms vs {adv.doh_avg_ms} ms over DoH.", + metric="system_avg_ms", + value=adv.system_avg_ms, + threshold=t.dns_resolve_ms.warn, + advice="Switching the OS resolver to a faster public one (or using DoH) may cut page load latency.", + title_ru="Системный DNS-резолвер медленнее, чем DoH", + detail_ru=f"Системный резолвер в среднем {adv.system_avg_ms} мс против {adv.doh_avg_ms} мс через DoH.", + advice_ru="Переключение системного резолвера на более быстрый публичный (или использование DoH) может снизить задержку загрузки страниц.", + ) + ) + for divergence in adv.divergences: + if "suspicious=True" not in divergence: + continue + name = divergence.split(":", 1)[0].strip() + findings.append( + Finding( + id=f"dns.poisoned_answer.{name}", + severity="crit", + title=f"System resolver returned a bogus answer for {name}", + detail=divergence, + advice="Compare against a known-good resolver (1.1.1.1/8.8.8.8) or DoH; this can indicate ISP-level DNS injection.", + title_ru=f"Системный резолвер вернул некорректный ответ для {name}", + detail_ru=divergence, + advice_ru="Сравните с заведомо надёжным резолвером (1.1.1.1/8.8.8.8) или DoH; это может указывать на подмену DNS на стороне провайдера.", + ) + ) + return findings + + +_EDGE_FAR_MS = 50.0 + + +def path_diversity_findings(pd: PathDiversity) -> list[Finding]: + findings: list[Finding] = [] + if pd.international_loop: + countries = ", ".join(pd.detour_countries) + findings.append( + Finding( + id="anycast.international_loop", + severity="warn", + title="Anycast routed traffic through another country", + detail=pd.note or f"Traffic detoured through {countries}.", + metric="detour_countries", + value=countries, + advice="Ask your ISP about local peering/CDN caching for this provider, or use a resolver closer to the intended PoP.", + title_ru="Anycast завернул трафик через другую страну", + detail_ru=pd.note_ru or f"Трафик прошёл через {countries}.", + advice_ru="Спросите провайдера про локальный пиринг/кэширование CDN для этого сервиса, либо используйте резолвер ближе к нужному PoP.", + ) + ) + for hop in pd.hops: + if hop.client_rtt_ms is None or hop.edge_rtt_ms is None: + continue + delta = hop.client_rtt_ms - hop.edge_rtt_ms + if delta > _EDGE_FAR_MS: + findings.append( + Finding( + id=f"anycast.edge_far.{hop.target}", + severity="info", + title=f"Anycast edge for {hop.target} is far from the client", + detail=f"Client RTT {hop.client_rtt_ms} ms vs edge-observed RTT {hop.edge_rtt_ms} ms.", + metric="client_rtt_ms", + value=hop.client_rtt_ms, + threshold=_EDGE_FAR_MS, + advice="The last mile to the nearest PoP may be the bottleneck rather than the wider path.", + title_ru=f"Anycast-точка для {hop.target} далеко от клиента", + detail_ru=f"RTT клиента {hop.client_rtt_ms} мс против RTT со стороны edge {hop.edge_rtt_ms} мс.", + advice_ru="Узким местом может быть последняя миля до ближайшего PoP, а не весь остальной путь.", + ) + ) + return findings + + _SCORE_PENALTY = {"ok": 0, "info": 3, "warn": 10, "crit": 25} diff --git a/tests/test_interpret_l7.py b/tests/test_interpret_l7.py new file mode 100644 index 0000000..edb966a --- /dev/null +++ b/tests/test_interpret_l7.py @@ -0,0 +1,271 @@ +from __future__ import annotations + +import re + +import pytest + +from netsleuth.config import Thresholds +from netsleuth.interpret import ( + dns_advanced_findings, + dpi_findings, + path_diversity_findings, + prefix_findings, + tls_findings, +) +from netsleuth.models import ( + AnycastHop, + DnsAdvanced, + DpiCheckResult, + PathDiversity, + PortProbe, + PrefixBenchmark, + PrefixProbe, + ResolverProbe, + TlsResult, +) + +_CYRILLIC = re.compile(r"[а-яА-ЯёЁ]") + + +def has_cyrillic(text: str) -> bool: + return bool(_CYRILLIC.search(text)) + + +@pytest.fixture() +def thresholds() -> Thresholds: + return Thresholds() + + +def tls(**kw) -> TlsResult: + base = dict(label="cloudflare", host="cloudflare.com", port=443, tcp_rtt_ms=20.0, tls_handshake_ms=25.0, ttfb_ms=50.0) + base.update(kw) + return TlsResult(**base) + + +def test_healthy_tls_result_produces_no_findings(thresholds): + assert tls_findings([tls()], thresholds) == [] + + +def test_tls_error_produces_a_labeled_unreachable_finding(thresholds): + findings = tls_findings([tls(tcp_rtt_ms=None, tls_handshake_ms=None, ttfb_ms=None, error="timeout")], thresholds) + assert len(findings) == 1 + assert findings[0].id == "tls.unreachable.cloudflare" + assert findings[0].severity == "warn" + assert has_cyrillic(findings[0].title_ru) + assert has_cyrillic(findings[0].detail_ru) + + +def test_slow_handshake_above_threshold_produces_a_warn_finding(thresholds): + findings = tls_findings([tls(tls_handshake_ms=350.0)], thresholds) + ids = [f.id for f in findings] + assert "tls.handshake_slow.cloudflare" in ids + + +def test_fast_handshake_below_threshold_produces_no_handshake_finding(thresholds): + findings = tls_findings([tls(tls_handshake_ms=10.0)], thresholds) + ids = [f.id for f in findings] + assert "tls.handshake_slow.cloudflare" not in ids + + +def test_slow_ttfb_above_threshold_produces_a_finding(thresholds): + findings = tls_findings([tls(ttfb_ms=900.0)], thresholds) + ids = [f.id for f in findings] + assert "tls.ttfb_slow.cloudflare" in ids + + +def test_server_cpu_bound_fires_when_handshake_dwarfs_tcp_rtt(thresholds): + findings = tls_findings([tls(tcp_rtt_ms=20.0, tls_handshake_ms=80.0)], thresholds) + ids = [f.id for f in findings] + assert "tls.server_cpu_bound.cloudflare" in ids + + +def test_server_cpu_bound_does_not_fire_on_sub_millisecond_loopback_timings(thresholds): + # ratio is huge (10x) but absolute handshake time is trivial - not a real CPU-bound server. + findings = tls_findings([tls(tcp_rtt_ms=1.0, tls_handshake_ms=10.0)], thresholds) + ids = [f.id for f in findings] + assert "tls.server_cpu_bound.cloudflare" not in ids + + +def test_two_tls_targets_get_distinct_finding_ids_so_dedupe_keeps_both(thresholds): + findings = tls_findings( + [tls(label="a", host="a.test", tls_handshake_ms=350.0), tls(label="b", host="b.test", tls_handshake_ms=350.0)], + thresholds, + ) + ids = {f.id for f in findings if f.id.startswith("tls.handshake_slow")} + assert ids == {"tls.handshake_slow.a", "tls.handshake_slow.b"} + + +def test_tls_findings_do_not_crash_on_a_result_with_no_measurements(thresholds): + assert tls_findings([tls(tcp_rtt_ms=None, tls_handshake_ms=None, ttfb_ms=None)], thresholds) == [] + + +def prefix_probe(**kw) -> PrefixProbe: + base = dict(prefix="203.0.113.0/24", probe_ip="203.0.113.1", avg_ms=10.0, reachable=True) + base.update(kw) + return PrefixProbe(**base) + + +def bench(**kw) -> PrefixBenchmark: + base = dict(asn="AS64500", prefixes_announced=10, prefixes_probed=2, method="icmp") + base.update(kw) + return PrefixBenchmark(**base) + + +def test_empty_prefix_benchmark_produces_no_findings(thresholds): + assert prefix_findings(bench(results=[]), thresholds) == [] + + +def test_high_spread_produces_a_finding(thresholds): + results = [prefix_probe(prefix="a/24", avg_ms=5.0, reachable=True), prefix_probe(prefix="b/24", avg_ms=200.0, reachable=True)] + findings = prefix_findings(bench(results=results, best="a/24", worst="b/24", spread_ms=195.0), thresholds) + ids = [f.id for f in findings] + assert "prefix.spread" in ids + assert findings[[f.id for f in findings].index("prefix.spread")].severity == "crit" + + +def test_low_spread_produces_no_spread_finding(thresholds): + results = [prefix_probe(prefix="a/24", avg_ms=10.0), prefix_probe(prefix="b/24", avg_ms=15.0)] + findings = prefix_findings(bench(results=results, spread_ms=5.0), thresholds) + assert "prefix.spread" not in [f.id for f in findings] + + +def test_mostly_unreachable_prefixes_produce_an_info_finding_not_a_warning(thresholds): + results = [prefix_probe(reachable=False, avg_ms=None) for _ in range(8)] + [prefix_probe(reachable=True)] * 2 + findings = prefix_findings(bench(results=results), thresholds) + hit = [f for f in findings if f.id == "prefix.mostly_unreachable"] + assert len(hit) == 1 + assert hit[0].severity == "info" + + +def test_mostly_reachable_prefixes_produce_no_unreachable_finding(thresholds): + results = [prefix_probe(reachable=True)] * 10 + findings = prefix_findings(bench(results=results, spread_ms=0.0), thresholds) + assert "prefix.mostly_unreachable" not in [f.id for f in findings] + + +def dpi_result(**kw) -> DpiCheckResult: + base = dict(target="203.0.113.9", resolved_ip="203.0.113.9", consented=True, verdict="clean", rationale="clean", rationale_ru="чисто") + base.update(kw) + return DpiCheckResult(**base) + + +def test_clean_dpi_verdict_produces_no_findings(): + assert dpi_findings(dpi_result(verdict="clean")) == [] + + +def test_reset_injection_verdict_produces_a_crit_finding(): + findings = dpi_findings(dpi_result(verdict="reset_injection", rationale="reset seen", rationale_ru="сброс обнаружен")) + assert len(findings) == 1 + assert findings[0].id == "dpi.reset_injection" + assert findings[0].severity == "crit" + assert has_cyrillic(findings[0].detail_ru) + + +def test_partial_filtering_verdict_produces_a_warn_finding(): + findings = dpi_findings(dpi_result(verdict="partial_filtering", rationale="mixed", rationale_ru="смешанно")) + assert findings[0].severity == "warn" + assert findings[0].id == "dpi.partial_filtering" + + +def test_unreachable_verdict_produces_only_an_info_finding_not_an_accusation(): + findings = dpi_findings(dpi_result(verdict="unreachable", rationale="or offline", rationale_ru="или выключен")) + assert findings[0].severity == "info" + assert findings[0].id == "dpi.unreachable" + + +def resolver_probe(**kw) -> ResolverProbe: + base = dict(name="system", kind="system", query_name="example.com", answers=["1.2.3.4"], elapsed_ms=20.0) + base.update(kw) + return ResolverProbe(**base) + + +def dns_adv(**kw) -> DnsAdvanced: + base = dict(probes=[], note="ok", note_ru="ок") + base.update(kw) + return DnsAdvanced(**base) + + +def test_no_transparent_proxy_and_normal_timing_produce_no_findings(thresholds): + adv = dns_adv(transparent_proxy=False, system_avg_ms=20.0, doh_avg_ms=25.0) + assert dns_advanced_findings(adv, thresholds) == [] + + +def test_transparent_proxy_detected_produces_a_warn_finding(thresholds): + adv = dns_adv(transparent_proxy=True, transparent_proxy_detail="responded") + findings = dns_advanced_findings(adv, thresholds) + ids = [f.id for f in findings] + assert "dns.transparent_proxy" in ids + assert [f for f in findings if f.id == "dns.transparent_proxy"][0].severity == "warn" + + +def test_slow_system_resolver_compared_to_doh_produces_a_finding(thresholds): + adv = dns_adv(transparent_proxy=False, system_avg_ms=200.0, doh_avg_ms=20.0) + findings = dns_advanced_findings(adv, thresholds) + assert "dns.system_slow" in [f.id for f in findings] + + +def test_system_resolver_slower_measurement_but_still_under_threshold_produces_nothing(thresholds): + adv = dns_adv(transparent_proxy=False, system_avg_ms=25.0, doh_avg_ms=20.0) + assert dns_advanced_findings(adv, thresholds) == [] + + +def test_cdn_style_divergence_without_a_suspicious_flag_produces_no_poisoning_finding(thresholds): + adv = dns_adv(divergences=["example.com: system=['1.1.1.1'] vs cloudflare=['104.16.1.1'] (suspicious=False)"]) + findings = dns_advanced_findings(adv, thresholds) + assert not any(f.id.startswith("dns.poisoned_answer") for f in findings) + + +def test_suspicious_divergence_produces_a_crit_poisoning_finding_named_after_the_domain(thresholds): + adv = dns_adv(divergences=["example.com: system=['192.0.2.1'] vs cloudflare=['1.1.1.1'] (suspicious=True)"]) + findings = dns_advanced_findings(adv, thresholds) + hit = [f for f in findings if f.id == "dns.poisoned_answer.example.com"] + assert len(hit) == 1 + assert hit[0].severity == "crit" + assert has_cyrillic(hit[0].title_ru) + + +def test_dns_advanced_findings_do_not_crash_on_all_none_fields(thresholds): + assert dns_advanced_findings(dns_adv(), thresholds) == [] + + +def anycast_hop(**kw) -> AnycastHop: + base = dict(target="cloudflare.com", source="cf_ray") + base.update(kw) + return AnycastHop(**base) + + +def test_no_international_loop_produces_no_loop_finding(): + pd = PathDiversity(client_country="RU", hops=[anycast_hop()], international_loop=False) + assert "anycast.international_loop" not in [f.id for f in path_diversity_findings(pd)] + + +def test_international_loop_produces_a_warn_finding(): + pd = PathDiversity( + client_country="RU", + hops=[anycast_hop(edge_country="DE")], + international_loop=True, + detour_countries=["DE"], + note="detour", + note_ru="крюк через Германию", + ) + findings = path_diversity_findings(pd) + hit = [f for f in findings if f.id == "anycast.international_loop"] + assert len(hit) == 1 + assert hit[0].severity == "warn" + assert has_cyrillic(hit[0].detail_ru) + + +def test_edge_far_from_client_produces_an_info_finding_named_after_the_target(): + pd = PathDiversity(client_country="RU", hops=[anycast_hop(target="slow.test", client_rtt_ms=150.0, edge_rtt_ms=10.0)]) + findings = path_diversity_findings(pd) + assert "anycast.edge_far.slow.test" in [f.id for f in findings] + + +def test_edge_close_to_client_produces_no_edge_far_finding(): + pd = PathDiversity(client_country="RU", hops=[anycast_hop(target="fast.test", client_rtt_ms=12.0, edge_rtt_ms=10.0)]) + findings = path_diversity_findings(pd) + assert not any(f.id.startswith("anycast.edge_far") for f in findings) + + +def test_path_diversity_findings_do_not_crash_on_empty_hops(): + assert path_diversity_findings(PathDiversity()) == []