models: domain shapes for geo, bgp, reputation, probes, speed

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

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"