From 7533f7e5bd25801f41898cb19d62b2cc1978e3d1 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sat, 8 Aug 2026 22:08:20 +0300 Subject: [PATCH] interpret: weighted vpn confidence scoring --- src/netcheck/interpret.py | 106 ++++++++++++++++++++++++++- tests/test_interpret.py | 147 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 251 insertions(+), 2 deletions(-) diff --git a/src/netcheck/interpret.py b/src/netcheck/interpret.py index 53fad16..b015a51 100644 --- a/src/netcheck/interpret.py +++ b/src/netcheck/interpret.py @@ -2,8 +2,9 @@ from __future__ import annotations from typing import Iterable -from netcheck.config import Band, Thresholds -from netcheck.models import Finding, PingResult, TraceResult +from netcheck.config import Band, Thresholds, VpnBands +from netcheck.models import CfTrace, DnsLeak, Finding, IpGeo, LocalNet, PingResult, Signal, TraceResult, VpnAssessment +from netcheck.netinfo import is_tunnel_iface, mtu_anomaly _SEVERITY_ORDER = {"ok": 0, "info": 1, "warn": 2, "crit": 3} @@ -132,3 +133,104 @@ def path_findings(trace: TraceResult) -> list[Finding]: ) ) return findings + + +SIGNAL_WEIGHTS: dict[str, float] = { + "tunnel_iface": 0.35, + "cf_warp": 0.50, + "provider_proxy": 0.35, + "provider_hosting": 0.40, + "provider_mobile": 0.10, + "mtu_anomaly": 0.20, + "dns_asn_mismatch": 0.25, + "gateway_egress_mismatch": 0.15, + "pdb_info_type_nsp": 0.15, + "timezone_mismatch": 0.15, +} + +_TZ_COUNTRY_PREFIX = { + "Europe/Amsterdam": "NL", + "Europe/Moscow": "RU", + "Europe/London": "GB", + "Europe/Berlin": "DE", + "America/New_York": "US", + "America/Los_Angeles": "US", + "Asia/Tokyo": "JP", +} + + +def _signal(name: str, observed: bool, note: str = "") -> Signal: + return Signal(name=name, observed=observed, weight=SIGNAL_WEIGHTS[name], direction="vpn", note=note) + + +def gather_vpn_signals( + local: LocalNet, + geo: IpGeo, + cf: CfTrace | None, + dns_leak: DnsLeak | None, + pdb_info_type: str | None, + os_timezone: str | None, + provider_flags: dict[str, bool], +) -> list[Signal]: + iface = local.iface_name or "" + anomaly = mtu_anomaly(local.iface_mtu) + leaking = [ + a + for a in (dns_leak.per_adapter if dns_leak else []) + if a.matches_egress_asn is False + ] + tz_country = _TZ_COUNTRY_PREFIX.get(os_timezone or "") + return [ + _signal("tunnel_iface", is_tunnel_iface(iface), iface), + _signal("cf_warp", bool(cf and (cf.warp or "").lower() == "on"), (cf.warp if cf else "") or ""), + _signal("provider_proxy", bool(provider_flags.get("proxy"))), + _signal("provider_hosting", bool(provider_flags.get("hosting"))), + _signal("provider_mobile", bool(provider_flags.get("mobile"))), + _signal("mtu_anomaly", anomaly in ("wireguard", "ipsec"), anomaly or ""), + _signal( + "dns_asn_mismatch", + bool(leaking), + ", ".join(f"{a.adapter} -> {a.echoed_asn}" for a in leaking), + ), + _signal( + "gateway_egress_mismatch", + bool(local.default_gateway_v4 and is_tunnel_iface(iface)), + local.default_gateway_v4 or "", + ), + _signal("pdb_info_type_nsp", (pdb_info_type or "").upper() in ("NSP", "CONTENT", "ENTERPRISE"), pdb_info_type or ""), + _signal( + "timezone_mismatch", + bool(tz_country and geo.country_code and tz_country != geo.country_code), + f"{os_timezone} vs {geo.country_code}" if tz_country else "", + ), + ] + + +def score_vpn(signals: list[Signal], bands: VpnBands) -> tuple[str, float]: + total = 0.0 + for s in signals: + if not s.observed: + continue + total += s.weight if s.direction == "vpn" else -s.weight + confidence = round(max(0.0, min(1.0, total)), 3) + if confidence >= bands.confirmed: + return "confirmed", confidence + if confidence >= bands.likely: + return "likely", confidence + return "none", confidence + + +def assess_vpn( + signals: list[Signal], + bands: VpnBands, + tunnel_iface: str | None, + dns_leak: DnsLeak | None, +) -> VpnAssessment: + verdict, confidence = score_vpn(signals, bands) + return VpnAssessment( + verdict=verdict, + confidence=confidence, + signals=signals, + tunnel_iface=tunnel_iface, + dns_leak=dns_leak, + ) diff --git a/tests/test_interpret.py b/tests/test_interpret.py index 0053323..fa40fb6 100644 --- a/tests/test_interpret.py +++ b/tests/test_interpret.py @@ -136,3 +136,150 @@ def test_path_findings_ignore_a_single_hop_that_rate_limits_icmp(): def test_path_findings_report_an_incomplete_trace(): trace = TraceResult(target="1.1.1.1", resolved_ip="1.1.1.1", backend="icmplib", hops=[], completed=False) assert [f.id for f in path_findings(trace)] == ["path.incomplete"] + + +from netcheck.config import VpnBands +from netcheck.models import AdapterLeakResult, CfTrace, DnsLeak, IpGeo, LocalNet, Signal +from netcheck.interpret import SIGNAL_WEIGHTS, assess_vpn, gather_vpn_signals, score_vpn + + +def sig(name: str, observed: bool = True) -> Signal: + return Signal(name=name, observed=observed, weight=SIGNAL_WEIGHTS[name], direction="vpn") + + +def test_no_signals_means_no_vpn(): + verdict, confidence = score_vpn([], VpnBands()) + assert verdict == "none" + assert confidence == 0.0 + + +def test_unobserved_signals_contribute_nothing(): + verdict, confidence = score_vpn( + [sig("tunnel_iface", observed=False), sig("cf_warp", observed=False)], VpnBands() + ) + assert verdict == "none" + assert confidence == 0.0 + + +def test_cloudflare_warp_alone_is_enough_for_likely(): + verdict, confidence = score_vpn([sig("cf_warp")], VpnBands()) + assert verdict == "likely" + assert confidence == pytest.approx(0.50) + + +def test_tunnel_interface_plus_hosting_egress_is_confirmed(): + verdict, confidence = score_vpn([sig("tunnel_iface"), sig("provider_hosting")], VpnBands()) + assert verdict == "confirmed" + assert confidence >= 0.75 + + +def test_a_mobile_flag_with_a_timezone_mismatch_does_not_fire(): + # This combination is normal for anyone travelling on a phone hotspot and + # must not be reported as a VPN. + verdict, confidence = score_vpn([sig("provider_mobile"), sig("timezone_mismatch")], VpnBands()) + assert verdict == "none" + assert confidence < 0.40 + + +def test_mtu_anomaly_alone_does_not_fire(): + verdict, _ = score_vpn([sig("mtu_anomaly")], VpnBands()) + assert verdict == "none" + + +def test_dns_asn_mismatch_alone_does_not_fire(): + verdict, _ = score_vpn([sig("dns_asn_mismatch")], VpnBands()) + assert verdict == "none" + + +def test_mtu_anomaly_plus_tunnel_iface_plus_dns_mismatch_is_confirmed(): + verdict, confidence = score_vpn( + [sig("tunnel_iface"), sig("mtu_anomaly"), sig("dns_asn_mismatch")], VpnBands() + ) + assert verdict == "confirmed" + assert confidence == pytest.approx(0.80) + + +def test_confidence_is_capped_at_one(): + _, confidence = score_vpn([sig(name) for name in SIGNAL_WEIGHTS], VpnBands()) + assert confidence == 1.0 + + +def test_clean_direction_signals_reduce_confidence(): + signals = [ + sig("provider_hosting"), + Signal(name="pdb_eyeball_isp", observed=True, weight=0.2, direction="clean"), + ] + _, confidence = score_vpn(signals, VpnBands()) + assert confidence == pytest.approx(0.20) + + +def test_gather_signals_flags_a_wireguard_tunnel_with_a_hosting_egress(): + local = LocalNet(iface_name="wg0", local_ipv4="10.7.0.2", iface_mtu=1420, default_gateway_v4="10.7.0.1") + geo = IpGeo(ip="203.0.113.44", asn="AS64500", country_code="NL", timezone="Europe/Amsterdam") + signals = {s.name: s for s in gather_vpn_signals( + local=local, + geo=geo, + cf=CfTrace(ip="203.0.113.44", warp="off"), + dns_leak=None, + pdb_info_type="NSP", + os_timezone="Europe/Amsterdam", + provider_flags={"hosting": True, "proxy": False, "mobile": False}, + )} + assert signals["tunnel_iface"].observed is True + assert signals["tunnel_iface"].note == "wg0" + assert signals["mtu_anomaly"].observed is True + assert signals["mtu_anomaly"].note == "wireguard" + assert signals["provider_hosting"].observed is True + assert signals["cf_warp"].observed is False + assert signals["timezone_mismatch"].observed is False + + +def test_gather_signals_detects_a_dns_resolver_in_another_asn(): + leak = DnsLeak( + per_adapter=[ + AdapterLeakResult( + adapter="Wi-Fi", + configured_resolvers=["192.168.1.1"], + echoed_ip="203.0.113.9", + echoed_asn="AS64501", + matches_egress_asn=False, + ) + ] + ) + signals = {s.name: s for s in gather_vpn_signals( + local=LocalNet(iface_name="eth0"), + geo=IpGeo(asn="AS64500"), + cf=None, + dns_leak=leak, + pdb_info_type=None, + os_timezone=None, + provider_flags={}, + )} + assert signals["dns_asn_mismatch"].observed is True + assert "Wi-Fi" in signals["dns_asn_mismatch"].note + + +def test_gather_signals_detects_cloudflare_warp(): + signals = {s.name: s for s in gather_vpn_signals( + local=LocalNet(), + geo=IpGeo(), + cf=CfTrace(warp="on"), + dns_leak=None, + pdb_info_type=None, + os_timezone=None, + provider_flags={}, + )} + assert signals["cf_warp"].observed is True + + +def test_assess_vpn_returns_a_complete_assessment(): + assessment = assess_vpn( + signals=[sig("tunnel_iface"), sig("provider_hosting")], + bands=VpnBands(), + tunnel_iface="wg0", + dns_leak=None, + ) + assert assessment.verdict == "confirmed" + assert assessment.tunnel_iface == "wg0" + assert len(assessment.signals) == 2 + assert assessment.confidence >= 0.75