add L4-L7 diagnostic models (TLS, DPI, DNS, anycast, prefix bench)

src/netsleuth/models.py

@@ -20,6 +20,10 @@ ERROR_KINDS = (
 STATUSES = ("ok", "partial", "failed", "skipped")
 SEVERITIES = ("ok", "info", "warn", "crit")
 DIRECTIONS = ("vpn", "clean")
+PORT_STATES = ("open", "closed", "filtered", "reset", "error")
+DPI_VERDICTS = ("clean", "partial_filtering", "reset_injection", "unreachable", "unknown")
+RESOLVER_KINDS = ("system", "doh")
+EDGE_SOURCES = ("cf_ray", "cf_trace", "server_timing", "none")
 
 
 @dataclass
@@ -342,6 +346,127 @@ class SpeedResult:
     netflix_oca_onnet: bool | None = None
 
 
+@dataclass
+class TlsResult:
+    label: str = ""
+    host: str = ""
+    port: int = 443
+    resolved_ip: str | None = None
+    tcp_rtt_ms: float | None = None
+    tls_handshake_ms: float | None = None
+    ttfb_ms: float | None = None
+    tls_version: str | None = None
+    cipher: str | None = None
+    alpn: str | None = None
+    cert_verified: bool | None = None
+    error: str | None = None
+
+
+@dataclass
+class PrefixProbe:
+    prefix: str
+    probe_ip: str | None = None
+    avg_ms: float | None = None
+    loss_pct: float = 100.0
+    reachable: bool = False
+    reverse_dns: str | None = None
+
+
+@dataclass
+class PrefixBenchmark:
+    asn: str | None = None
+    prefixes_announced: int = 0
+    prefixes_probed: int = 0
+    method: str = "none"
+    results: list[PrefixProbe] = field(default_factory=list)
+    best: str | None = None
+    worst: str | None = None
+    spread_ms: float | None = None
+    skipped_reason: str | None = None
+
+
+@dataclass
+class PortProbe:
+    port: int
+    state: str
+    rtt_ms: float | None = None
+    detail: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.state not in PORT_STATES:
+            raise ValueError(f"unknown PortProbe state: {self.state!r}")
+
+
+@dataclass
+class DpiCheckResult:
+    target: str | None = None
+    resolved_ip: str | None = None
+    consented: bool = False
+    ports: list[PortProbe] = field(default_factory=list)
+    verdict: str = "unknown"
+    rationale: str = ""
+    rationale_ru: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.verdict not in DPI_VERDICTS:
+            raise ValueError(f"unknown DpiCheckResult verdict: {self.verdict!r}")
+
+
+@dataclass
+class ResolverProbe:
+    name: str
+    kind: str
+    endpoint: str | None = None
+    query_name: str = ""
+    answers: list[str] = field(default_factory=list)
+    elapsed_ms: float | None = None
+    error: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.kind not in RESOLVER_KINDS:
+            raise ValueError(f"unknown ResolverProbe kind: {self.kind!r}")
+
+
+@dataclass
+class DnsAdvanced:
+    probes: list[ResolverProbe] = field(default_factory=list)
+    system_avg_ms: float | None = None
+    doh_avg_ms: float | None = None
+    divergences: list[str] = field(default_factory=list)
+    transparent_proxy: bool | None = None
+    transparent_proxy_detail: str | None = None
+    note: str = ""
+    note_ru: str | None = None
+
+
+@dataclass
+class AnycastHop:
+    target: str
+    resolved_ip: str | None = None
+    ip_country: str | None = None
+    ip_city: str | None = None
+    edge_colo: str | None = None
+    edge_city: str | None = None
+    edge_country: str | None = None
+    edge_rtt_ms: float | None = None
+    client_rtt_ms: float | None = None
+    source: str = "none"
+
+    def __post_init__(self) -> None:
+        if self.source not in EDGE_SOURCES:
+            raise ValueError(f"unknown AnycastHop source: {self.source!r}")
+
+
+@dataclass
+class PathDiversity:
+    client_country: str | None = None
+    hops: list[AnycastHop] = field(default_factory=list)
+    international_loop: bool = False
+    detour_countries: list[str] = field(default_factory=list)
+    note: str = ""
+    note_ru: str | None = None
+
+
 def to_jsonable(obj: Any) -> Any:
     if obj is None or isinstance(obj, (str, bool, int)):
         return obj

tests/test_models.py

@@ -6,25 +6,34 @@ import pytest
 
 from netsleuth.models import (
     AdapterLeakResult,
+    AnycastHop,
     BgpEvent,
     BgpIntel,
     Capabilities,
     CfL4Stats,
     CfTrace,
+    DnsAdvanced,
     DnsLeak,
     DnsblHit,
+    DpiCheckResult,
     Finding,
     InternetDbResult,
     IpGeo,
     IxpPresence,
     LocalNet,
     ModuleResult,
+    PathDiversity,
     PingResult,
+    PortProbe,
+    PrefixBenchmark,
+    PrefixProbe,
     ProbeError,
     Reputation,
+    ResolverProbe,
     Signal,
     SpeedResult,
     TierAttempt,
+    TlsResult,
     TraceHop,
     TraceResult,
     VpnAssessment,
@@ -245,3 +254,122 @@ def test_ip_geo_and_vpn_assessment_defaults_are_json_safe():
     out = to_jsonable({"geo": geo, "vpn": vpn})
     assert out["geo"]["ip_type"] == "unknown"
     assert out["vpn"]["signals"][0]["name"] == "warp"
+
+
+def test_every_l7_domain_shape_defaults_to_constructible_with_no_arguments():
+    for cls in (TlsResult, PrefixBenchmark, DpiCheckResult, DnsAdvanced, PathDiversity):
+        instance = cls()
+        json.dumps(to_jsonable(instance), allow_nan=False)
+
+
+def test_tls_result_round_trips_through_json():
+    tls = TlsResult(
+        label="cloudflare",
+        host="cloudflare.com",
+        port=443,
+        resolved_ip="104.16.132.229",
+        tcp_rtt_ms=12.0,
+        tls_handshake_ms=45.0,
+        ttfb_ms=60.0,
+        tls_version="TLSv1.3",
+        cipher="TLS_AES_256_GCM_SHA384",
+        alpn="h2",
+        cert_verified=True,
+    )
+    out = json.loads(json.dumps(to_jsonable(tls), allow_nan=False))
+    assert out["host"] == "cloudflare.com"
+    assert out["tls_handshake_ms"] == 45.0
+
+
+def test_port_probe_rejects_unknown_state():
+    with pytest.raises(ValueError, match="unknown PortProbe state"):
+        PortProbe(port=443, state="bogus")
+
+
+@pytest.mark.parametrize("state", ["open", "closed", "filtered", "reset", "error"])
+def test_port_probe_accepts_documented_states(state):
+    assert PortProbe(port=443, state=state).state == state
+
+
+def test_dpi_check_result_rejects_unknown_verdict():
+    with pytest.raises(ValueError, match="unknown DpiCheckResult verdict"):
+        DpiCheckResult(verdict="bogus")
+
+
+@pytest.mark.parametrize(
+    "verdict", ["clean", "partial_filtering", "reset_injection", "unreachable", "unknown"]
+)
+def test_dpi_check_result_accepts_documented_verdicts(verdict):
+    assert DpiCheckResult(verdict=verdict).verdict == verdict
+
+
+def test_dpi_check_result_nests_port_probes():
+    result = DpiCheckResult(
+        target="203.0.113.9",
+        resolved_ip="203.0.113.9",
+        consented=True,
+        ports=[PortProbe(port=443, state="open", rtt_ms=20.0)],
+        verdict="clean",
+        rationale="all probed ports behave normally",
+    )
+    out = to_jsonable(result)
+    assert out["ports"][0]["state"] == "open"
+    assert out["consented"] is True
+
+
+def test_resolver_probe_rejects_unknown_kind():
+    with pytest.raises(ValueError, match="unknown ResolverProbe kind"):
+        ResolverProbe(name="system", kind="bogus", query_name="example.com")
+
+
+@pytest.mark.parametrize("kind", ["system", "doh"])
+def test_resolver_probe_accepts_documented_kinds(kind):
+    assert ResolverProbe(name="n", kind=kind, query_name="example.com").kind == kind
+
+
+def test_dns_advanced_nests_resolver_probes():
+    adv = DnsAdvanced(
+        probes=[ResolverProbe(name="system", kind="system", query_name="example.com", answers=["1.2.3.4"])],
+        system_avg_ms=12.0,
+        doh_avg_ms=30.0,
+        transparent_proxy=False,
+        note="system resolver answered directly",
+    )
+    out = to_jsonable(adv)
+    assert out["probes"][0]["answers"] == ["1.2.3.4"]
+    assert out["transparent_proxy"] is False
+
+
+def test_anycast_hop_rejects_unknown_source():
+    with pytest.raises(ValueError, match="unknown AnycastHop source"):
+        AnycastHop(target="example.com", source="bogus")
+
+
+@pytest.mark.parametrize("source", ["cf_ray", "cf_trace", "server_timing", "none"])
+def test_anycast_hop_accepts_documented_sources(source):
+    assert AnycastHop(target="example.com", source=source).source == source
+
+
+def test_path_diversity_nests_anycast_hops():
+    pd = PathDiversity(
+        client_country="RU",
+        hops=[AnycastHop(target="cloudflare.com", edge_colo="DME", source="cf_ray")],
+        international_loop=False,
+    )
+    out = to_jsonable(pd)
+    assert out["hops"][0]["edge_colo"] == "DME"
+    assert out["international_loop"] is False
+
+
+def test_prefix_benchmark_nests_prefix_probes_and_ranks():
+    bench = PrefixBenchmark(
+        asn="AS64500",
+        prefixes_announced=120,
+        prefixes_probed=32,
+        method="icmp",
+        results=[PrefixProbe(prefix="203.0.113.0/24", probe_ip="203.0.113.1", avg_ms=15.0, reachable=True)],
+        best="203.0.113.0/24",
+    )
+    out = to_jsonable(bench)
+    assert out["results"][0]["avg_ms"] == 15.0
+    assert out["best"] == "203.0.113.0/24"