refactor gather_vpn_signals to use VpnContext dataclass

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",

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(
+def gather_vpn_signals(ctx: VpnContext) -> list[Signal]:
-    local: LocalNet,
+    iface = ctx.local.iface_name or ""
-    geo: IpGeo,
+    anomaly = mtu_anomaly(ctx.local.iface_mtu)
-    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 [])
+        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("cf_warp", bool(ctx.cf and (ctx.cf.warp or "").lower() == "on"), (ctx.cf.warp if ctx.cf else "") or ""),
-        _signal("provider_proxy", bool(provider_flags.get("proxy"))),
+        _signal("provider_proxy", bool(ctx.provider_flags.get("proxy"))),
-        _signal("provider_hosting", bool(provider_flags.get("hosting"))),
+        _signal("provider_hosting", bool(ctx.provider_flags.get("hosting"))),
-        _signal("provider_mobile", bool(provider_flags.get("mobile"))),
+        _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)),
+            bool(ctx.local.default_gateway_v4 and is_tunnel_iface(iface)),
-            local.default_gateway_v4 or "",
+            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),
+            bool(tz_country and ctx.geo.country_code and tz_country != ctx.geo.country_code),
-            f"{os_timezone} vs {geo.country_code}" if tz_country else "",
+            f"{ctx.os_timezone} vs {ctx.geo.country_code}" if tz_country else "",
         ),
     ]
 

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"

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