diff --git a/src/netsleuth/cli.py b/src/netsleuth/cli.py index 411221c..e84faa5 100644 --- a/src/netsleuth/cli.py +++ b/src/netsleuth/cli.py @@ -28,7 +28,7 @@ from netsleuth.interpret import ( speed_findings, ) from netsleuth.ip_geo import dual_stack_mismatch, gather_identity -from netsleuth.models import BindTarget, Finding, IpGeo, LocalNet, ModuleResult, SpeedResult +from netsleuth.models import BindTarget, Finding, IpGeo, LocalNet, ModuleResult, SpeedResult, VpnContext from netsleuth.netinfo import collect_local_net, detect_capabilities, iface_for_ip, is_tunnel_iface, primary_interface_ip from netsleuth.orchestration import gather_modules, run_module, utc_now_iso from netsleuth.probes.dns_leak import collect_dns_leak @@ -372,7 +372,7 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat ), ) modules["bgp"], modules["reputation"] = bgp_result, rep_result - signals = gather_vpn_signals( + ctx = VpnContext( local=local, geo=geo, cf=bundle.get("cf_trace"), @@ -381,6 +381,7 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat os_timezone=_os_timezone(), provider_flags=flags, ) + signals = gather_vpn_signals(ctx) modules["vpn_assessment"] = ModuleResult( name="vpn_assessment", status="ok" if dns_result.status == "ok" else "partial", diff --git a/src/netsleuth/interpret.py b/src/netsleuth/interpret.py index ee47ed9..f987358 100644 --- a/src/netsleuth/interpret.py +++ b/src/netsleuth/interpret.py @@ -14,6 +14,7 @@ from netsleuth.models import ( SpeedResult, TraceResult, VpnAssessment, + VpnContext, ) from netsleuth.netinfo import is_tunnel_iface, mtu_anomaly @@ -198,29 +199,21 @@ 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) +def gather_vpn_signals(ctx: VpnContext) -> list[Signal]: + iface = ctx.local.iface_name or "" + anomaly = mtu_anomaly(ctx.local.iface_mtu) leaking = [ a - for a in (dns_leak.per_adapter if dns_leak else []) + for a in (ctx.dns_leak.per_adapter if ctx.dns_leak else []) if a.matches_egress_asn is False ] - tz_country = _TZ_COUNTRY_PREFIX.get(os_timezone or "") + tz_country = _TZ_COUNTRY_PREFIX.get(ctx.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("cf_warp", bool(ctx.cf and (ctx.cf.warp or "").lower() == "on"), (ctx.cf.warp if ctx.cf else "") or ""), + _signal("provider_proxy", bool(ctx.provider_flags.get("proxy"))), + _signal("provider_hosting", bool(ctx.provider_flags.get("hosting"))), + _signal("provider_mobile", bool(ctx.provider_flags.get("mobile"))), _signal("mtu_anomaly", anomaly in ("wireguard", "ipsec"), anomaly or ""), _signal( "dns_asn_mismatch", @@ -229,14 +222,14 @@ def gather_vpn_signals( ), _signal( "gateway_egress_mismatch", - bool(local.default_gateway_v4 and is_tunnel_iface(iface)), - local.default_gateway_v4 or "", + bool(ctx.local.default_gateway_v4 and is_tunnel_iface(iface)), + ctx.local.default_gateway_v4 or "", ), - _signal("pdb_info_type_nsp", (pdb_info_type or "").upper() in ("NSP", "CONTENT", "ENTERPRISE"), pdb_info_type or ""), + _signal("pdb_info_type_nsp", (ctx.pdb_info_type or "").upper() in ("NSP", "CONTENT", "ENTERPRISE"), ctx.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 "", + bool(tz_country and ctx.geo.country_code and tz_country != ctx.geo.country_code), + f"{ctx.os_timezone} vs {ctx.geo.country_code}" if tz_country else "", ), ] diff --git a/src/netsleuth/models.py b/src/netsleuth/models.py index c03f361..7430741 100644 --- a/src/netsleuth/models.py +++ b/src/netsleuth/models.py @@ -163,6 +163,17 @@ class DnsLeak: note_ru: str | None = None +@dataclass +class VpnContext: + local: LocalNet + geo: IpGeo + cf: CfTrace | None = None + dns_leak: DnsLeak | None = None + pdb_info_type: str | None = None + os_timezone: str | None = None + provider_flags: dict[str, bool] = field(default_factory=dict) + + @dataclass class VpnAssessment: verdict: str = "none" diff --git a/tests/test_interpret.py b/tests/test_interpret.py index 51a7980..eaf65af 100644 --- a/tests/test_interpret.py +++ b/tests/test_interpret.py @@ -187,7 +187,7 @@ def test_path_findings_flag_loss_that_persists_all_the_way_to_the_last_hop(): from netsleuth.config import VpnBands -from netsleuth.models import AdapterLeakResult, CfTrace, DnsLeak, IpGeo, LocalNet, Signal +from netsleuth.models import AdapterLeakResult, CfTrace, DnsLeak, IpGeo, LocalNet, Signal, VpnContext from netsleuth.interpret import SIGNAL_WEIGHTS, assess_vpn, gather_vpn_signals, score_vpn @@ -264,7 +264,7 @@ def test_clean_direction_signals_reduce_confidence(): 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( + ctx = VpnContext( local=local, geo=geo, cf=CfTrace(ip="203.0.113.44", warp="off"), @@ -272,7 +272,8 @@ def test_gather_signals_flags_a_wireguard_tunnel_with_a_hosting_egress(): pdb_info_type="NSP", os_timezone="Europe/Amsterdam", provider_flags={"hosting": True, "proxy": False, "mobile": False}, - )} + ) + signals = {s.name: s for s in gather_vpn_signals(ctx)} assert signals["tunnel_iface"].observed is True assert signals["tunnel_iface"].note == "wg0" assert signals["mtu_anomaly"].observed is True @@ -294,7 +295,7 @@ def test_gather_signals_detects_a_dns_resolver_in_another_asn(): ) ] ) - signals = {s.name: s for s in gather_vpn_signals( + ctx = VpnContext( local=LocalNet(iface_name="eth0"), geo=IpGeo(asn="AS64500"), cf=None, @@ -302,13 +303,14 @@ def test_gather_signals_detects_a_dns_resolver_in_another_asn(): pdb_info_type=None, os_timezone=None, provider_flags={}, - )} + ) + signals = {s.name: s for s in gather_vpn_signals(ctx)} 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( + ctx = VpnContext( local=LocalNet(), geo=IpGeo(), cf=CfTrace(warp="on"), @@ -316,7 +318,8 @@ def test_gather_signals_detects_cloudflare_warp(): pdb_info_type=None, os_timezone=None, provider_flags={}, - )} + ) + signals = {s.name: s for s in gather_vpn_signals(ctx)} assert signals["cf_warp"].observed is True