mirror of
https://github.com/Horizontal42/netsleuth.git
synced 2026-11-10 15:42:10 +03:00
247 lines
7.2 KiB
Python
247 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
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():
|
|
with pytest.raises(ValueError, match="unknown ProbeError kind"):
|
|
ProbeError(source="ip-api", kind="exploded", message="boom")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"kind",
|
|
[
|
|
"timeout",
|
|
"http_error",
|
|
"rate_limited",
|
|
"blocked",
|
|
"parse_error",
|
|
"unavailable",
|
|
"no_privilege",
|
|
"not_applicable",
|
|
],
|
|
)
|
|
def test_probe_error_accepts_documented_kinds(kind):
|
|
assert ProbeError(source="s", kind=kind, message="m").kind == kind
|
|
|
|
|
|
def test_module_result_rejects_unknown_status():
|
|
with pytest.raises(ValueError, match="unknown ModuleResult status"):
|
|
ModuleResult(name="bgp", status="borked")
|
|
|
|
|
|
def test_module_result_serializes_to_strict_json():
|
|
result = ModuleResult(
|
|
name="reputation",
|
|
status="partial",
|
|
data={"firehol_hits": ["firehol_level1"]},
|
|
errors=[ProbeError(source="internetdb", kind="timeout", message="8s", retryable=True)],
|
|
warnings=["abuseipdb key missing"],
|
|
started_at="2026-08-08T19:12:00Z",
|
|
duration_ms=1234,
|
|
)
|
|
text = json.dumps(to_jsonable(result), allow_nan=False)
|
|
back = json.loads(text)
|
|
assert back["name"] == "reputation"
|
|
assert back["status"] == "partial"
|
|
assert back["data"]["firehol_hits"] == ["firehol_level1"]
|
|
assert back["errors"][0] == {
|
|
"source": "internetdb",
|
|
"kind": "timeout",
|
|
"message": "8s",
|
|
"retryable": True,
|
|
}
|
|
assert back["warnings"] == ["abuseipdb key missing"]
|
|
assert back["duration_ms"] == 1234
|
|
|
|
|
|
def test_to_jsonable_coerces_non_finite_numbers_to_null():
|
|
payload = {"a": float("inf"), "b": float("-inf"), "c": float("nan"), "d": 1.5}
|
|
out = to_jsonable(payload)
|
|
assert out == {"a": None, "b": None, "c": None, "d": 1.5}
|
|
json.dumps(out, allow_nan=False)
|
|
|
|
|
|
def test_to_jsonable_handles_nested_dataclasses_and_sets():
|
|
finding = Finding(
|
|
id="latency.high",
|
|
severity="warn",
|
|
title="Latency above target",
|
|
detail="avg 130 ms to 1.1.1.1",
|
|
metric="avg_ms",
|
|
value=130.0,
|
|
threshold=100.0,
|
|
advice="Check for a saturated uplink.",
|
|
)
|
|
signal = Signal(name="tunnel_iface", observed=True, weight=0.35, direction="vpn", note="wg0")
|
|
out = to_jsonable({"findings": [finding], "signals": (signal,), "tags": {"a"}})
|
|
assert out["findings"][0]["severity"] == "warn"
|
|
assert out["signals"][0]["weight"] == 0.35
|
|
assert out["tags"] == ["a"]
|
|
json.dumps(out, allow_nan=False)
|
|
|
|
|
|
def test_finding_rejects_unknown_severity():
|
|
with pytest.raises(ValueError, match="unknown Finding severity"):
|
|
Finding(
|
|
id="x",
|
|
severity="apocalyptic",
|
|
title="t",
|
|
detail="d",
|
|
metric=None,
|
|
value=None,
|
|
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"
|