diff --git a/src/netcheck/models.py b/src/netcheck/models.py index 1004cc2..3a9348b 100644 --- a/src/netcheck/models.py +++ b/src/netcheck/models.py @@ -78,6 +78,233 @@ class Signal: raise ValueError(f"unknown Signal direction: {self.direction!r}") +@dataclass +class Capabilities: + os_name: str = "" + is_elevated: bool = False + icmp_dgram: bool = False + icmp_raw: bool = False + icmp_win_api: bool = False + mtr_binary: str | None = None + traceroute_binary: str | None = None + chosen_latency_backend: str = "none" + chosen_trace_backend: str = "none" + notes: list[str] = field(default_factory=list) + + +@dataclass +class LocalNet: + iface_name: str | None = None + local_ipv4: str | None = None + local_ipv6: str | None = None + iface_mtu: int | None = None + default_gateway_v4: str | None = None + default_gateway_v6: str | None = None + dns_servers_per_adapter: dict[str, list[str]] = field(default_factory=dict) + is_dual_stack: bool = False + + +@dataclass +class IpGeo: + ip: str | None = None + ip_version: int | None = None + reverse_dns: str | None = None + asn: str | None = None + as_name: str | None = None + org: str | None = None + country: str | None = None + country_code: str | None = None + city: str | None = None + lat: float | None = None + lon: float | None = None + timezone: str | None = None + ip_type: str = "unknown" + sources: dict[str, str] = field(default_factory=dict) + + +@dataclass +class CfTrace: + ip: str | None = None + colo: str | None = None + loc: str | None = None + warp: str | None = None + gateway: str | None = None + rbi: str | None = None + raw: dict[str, str] = field(default_factory=dict) + + +@dataclass +class AdapterLeakResult: + adapter: str + configured_resolvers: list[str] = field(default_factory=list) + echoed_ip: str | None = None + echoed_asn: str | None = None + matches_egress_asn: bool | None = None + + +@dataclass +class DnsLeak: + per_adapter: list[AdapterLeakResult] = field(default_factory=list) + ecs_leaked: bool = False + note: str = "" + + +@dataclass +class VpnAssessment: + verdict: str = "none" + confidence: float = 0.0 + signals: list[Signal] = field(default_factory=list) + tunnel_iface: str | None = None + dns_leak: DnsLeak | None = None + + +@dataclass +class BgpEvent: + timestamp: str + type: str + prefix: str | None = None + path: list[int] = field(default_factory=list) + + +@dataclass +class IxpPresence: + name: str + city: str | None = None + country: str | None = None + speed_mbps: int | None = None + + +@dataclass +class BgpIntel: + asn: str | None = None + holder: str | None = None + registry: str | None = None + allocated_at: str | None = None + upstreams: list[str] = field(default_factory=list) + peers: list[str] = field(default_factory=list) + downstreams: list[str] = field(default_factory=list) + announced_prefixes: list[str] = field(default_factory=list) + prefix_count_v4: int = 0 + prefix_count_v6: int = 0 + flaps: list[BgpEvent] = field(default_factory=list) + stability: str = "unknown" + ixps: list[IxpPresence] = field(default_factory=list) + pdb_info_type: str | None = None + pdb_traffic: str | None = None + asrank: int | None = None + cone_asns: int | None = None + cone_prefixes: int | None = None + + +@dataclass +class InternetDbResult: + ip: str | None = None + ports: list[int] = field(default_factory=list) + hostnames: list[str] = field(default_factory=list) + tags: list[str] = field(default_factory=list) + cpes: list[str] = field(default_factory=list) + vulns: list[str] = field(default_factory=list) + + +@dataclass +class DnsblHit: + zone: str + codes: list[str] = field(default_factory=list) + meaning: str = "listed" + + +@dataclass +class Reputation: + internetdb: InternetDbResult | None = None + firehol_hits: list[str] = field(default_factory=list) + dnsbl_hits: list[DnsblHit] | None = None + dnsbl_query_blocked: bool = False + abuseipdb_score: int | None = None + abuseipdb_reports: int | None = None + captcha_risk: str = "low" + rationale: str = "" + + +@dataclass +class PingResult: + label: str + host: str + resolved_ip: str | None = None + method: str = "none" + sent: int = 0 + received: int = 0 + loss_pct: float = 0.0 + min_ms: float | None = None + avg_ms: float | None = None + max_ms: float | None = None + mdev_ms: float | None = None + jitter_ms: float | None = None + samples: list[float | None] = field(default_factory=list) + + +@dataclass +class TraceHop: + ttl: int + ip: str | None = None + reverse_dns: str | None = None + asn: str | None = None + as_name: str | None = None + probes: list[float | None] = field(default_factory=list) + loss_pct: float = 0.0 + min_ms: float | None = None + avg_ms: float | None = None + max_ms: float | None = None + jitter_ms: float | None = None + annotations: list[str] = field(default_factory=list) + + +@dataclass +class TraceResult: + target: str | None = None + resolved_ip: str | None = None + backend: str = "none" + hops: list[TraceHop] = field(default_factory=list) + cycles: int = 0 + completed: bool = False + max_hops_reached: bool = False + + +@dataclass +class TierAttempt: + tier: str + ok: bool + reason: str | None = None + duration_ms: int = 0 + + +@dataclass +class CfL4Stats: + rtt_ms: float | None = None + min_rtt_ms: float | None = None + rtt_var_ms: float | None = None + delivery_rate_bps: int | None = None + cwnd: int | None = None + unsent_bytes: int | None = None + recv_bytes: int | None = None + + +@dataclass +class SpeedResult: + method: str = "none" + tier_attempts: list[TierAttempt] = field(default_factory=list) + download_mbps: float | None = None + upload_mbps: float | None = None + server: str | None = None + idle_rtt_ms: float | None = None + loaded_rtt_down_ms: float | None = None + loaded_rtt_up_ms: float | None = None + bufferbloat_down_ms: float | None = None + bufferbloat_up_ms: float | None = None + bufferbloat_grade: str | None = None + cfL4_stats: CfL4Stats | None = None + netflix_oca_onnet: bool | None = None + + def to_jsonable(obj: Any) -> Any: if obj is None or isinstance(obj, (str, bool, int)): return obj diff --git a/tests/test_models.py b/tests/test_models.py index 6b9cec3..73740aa 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -4,7 +4,32 @@ import json import pytest -from netcheck.models import Finding, ModuleResult, ProbeError, Signal, to_jsonable +from netcheck.models import ( + AdapterLeakResult, + BgpEvent, + BgpIntel, + Capabilities, + CfL4Stats, + CfTrace, + DnsLeak, + DnsblHit, + Finding, + InternetDbResult, + IpGeo, + IxpPresence, + LocalNet, + ModuleResult, + PingResult, + ProbeError, + Reputation, + Signal, + SpeedResult, + TierAttempt, + TraceHop, + TraceResult, + VpnAssessment, + to_jsonable, +) def test_probe_error_rejects_unknown_kind(): @@ -97,3 +122,126 @@ def test_finding_rejects_unknown_severity(): threshold=None, advice=None, ) + + +def test_every_domain_shape_defaults_to_constructible_with_no_arguments(): + for cls in ( + Capabilities, + LocalNet, + IpGeo, + CfTrace, + VpnAssessment, + DnsLeak, + BgpIntel, + Reputation, + SpeedResult, + TraceResult, + ): + instance = cls() + json.dumps(to_jsonable(instance), allow_nan=False) + + +def test_ping_result_round_trips_through_json(): + ping = PingResult( + label="cloudflare-dns", + host="1.1.1.1", + resolved_ip="1.1.1.1", + method="icmp_win", + sent=20, + received=20, + loss_pct=0.0, + min_ms=11.0, + avg_ms=12.4, + max_ms=15.1, + mdev_ms=0.9, + jitter_ms=1.9, + samples=[11.0, 12.0, None, 15.1], + ) + out = json.loads(json.dumps(to_jsonable(ping), allow_nan=False)) + assert out["method"] == "icmp_win" + assert out["samples"][2] is None + + +def test_trace_result_nests_hops(): + trace = TraceResult( + target="1.1.1.1", + resolved_ip="1.1.1.1", + backend="icmp_win", + hops=[TraceHop(ttl=1, ip="192.168.1.1", probes=[1.2, 1.1, None], annotations=["!H"])], + cycles=1, + completed=True, + ) + out = to_jsonable(trace) + assert out["hops"][0]["ttl"] == 1 + assert out["hops"][0]["annotations"] == ["!H"] + assert out["backend"] == "icmp_win" + + +def test_reputation_composes_optional_sub_results(): + rep = Reputation( + internetdb=InternetDbResult(ip="203.0.113.44", ports=[80, 443], tags=["cdn"]), + firehol_hits=["firehol_level1"], + dnsbl_hits=[DnsblHit(zone="zen.spamhaus.org", codes=["127.0.0.2"], meaning="listed")], + dnsbl_query_blocked=False, + captcha_risk="medium", + rationale="Listed on one blocklist.", + ) + out = to_jsonable(rep) + assert out["internetdb"]["ports"] == [80, 443] + assert out["dnsbl_hits"][0]["zone"] == "zen.spamhaus.org" + + +def test_bgp_intel_nests_events_and_ixps(): + bgp = BgpIntel( + asn="AS64500", + holder="Example Telecom", + flaps=[BgpEvent(timestamp="2026-08-01T00:00:00Z", type="A", prefix="203.0.113.0/24")], + ixps=[IxpPresence(name="AMS-IX", city="Amsterdam", country="NL", speed_mbps=100000)], + stability="stable", + ) + out = to_jsonable(bgp) + assert out["flaps"][0]["prefix"] == "203.0.113.0/24" + assert out["ixps"][0]["speed_mbps"] == 100000 + + +def test_speed_result_carries_tier_attempts_and_cfl4(): + speed = SpeedResult( + method="cloudflare", + tier_attempts=[ + TierAttempt(tier="ookla_bin", ok=False, reason="binary not on PATH"), + TierAttempt(tier="cloudflare", ok=True, reason=None), + ], + download_mbps=284.3, + upload_mbps=41.7, + cfL4_stats=CfL4Stats(rtt_ms=12.0, min_rtt_ms=11.0, rtt_var_ms=1.5, delivery_rate_bps=35000000, cwnd=42, unsent_bytes=0, recv_bytes=1048576), + ) + out = to_jsonable(speed) + assert out["tier_attempts"][0]["ok"] is False + assert out["cfL4_stats"]["rtt_ms"] == 12.0 + + +def test_adapter_leak_result_and_dns_leak_compose(): + 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, + ) + ], + ecs_leaked=True, + note="ISP resolver still active on Wi-Fi adapter.", + ) + out = to_jsonable(leak) + assert out["per_adapter"][0]["matches_egress_asn"] is False + assert out["ecs_leaked"] is True + + +def test_ip_geo_and_vpn_assessment_defaults_are_json_safe(): + geo = IpGeo(ip="203.0.113.44", ip_version=4, asn="AS64500", sources={"asn": "ip-api"}) + vpn = VpnAssessment(verdict="likely", confidence=0.55, signals=[Signal("warp", True, 0.5, "vpn")]) + out = to_jsonable({"geo": geo, "vpn": vpn}) + assert out["geo"]["ip_type"] == "unknown" + assert out["vpn"]["signals"][0]["name"] == "warp"