feat(prober): add IcmpErrorProbe — ICMP error-leakage fingerprint

Sends four crafted stimuli (UDP/closed-port, TTL=1, DF+oversized,
bad IP option) and records which ICMP error classes come back, the
per-error RTT, and the bytes echoed in each ICMP body. Absence is
as informative as a reply — Linux rate-limiting is a fingerprint signal.

Returns None when no packets could be sent (no CAP_NET_RAW), so the
probe is a no-op in non-root test environments. Port-free ActiveProbe
subclass (priority=850), metaclass auto-registered in the registry.

Also fixes three sets of stale tests left over from the TlsCertProbe
migration (4b2759e0):
- test_active_probe_registry: closed name/order sets updated for
  tls_certificate and icmp_error
- test_prober_rotation: dead patches on worker.fetch_leaf_cert removed
- test_prober_worker (TestProbeCycleTLSCert): rewritten to test
  TlsCertProbe as an independent registry probe, patch target updated
  from worker.fetch_leaf_cert to probes.tlscert_probe.fetch_leaf_cert
This commit is contained in:
2026-05-21 14:52:49 -04:00
parent 4b2759e0fc
commit 56229a272b
7 changed files with 781 additions and 84 deletions

View File

@@ -23,12 +23,15 @@ class TestRegistryContents:
def test_all_probes_registered(self):
names = {cls.probe_name for cls in ActiveProbeMeta.all()}
assert names == {"jarm", "hassh", "tcpfp", "ipv6_leak"}
assert names == {"jarm", "hassh", "tcpfp", "ipv6_leak", "tls_certificate", "icmp_error"}
def test_sorted_by_priority_then_name(self):
order = [cls.probe_name for cls in ActiveProbeMeta.all()]
# hassh/jarm/tcpfp all priority=100 (alphabetical), ipv6_leak priority=999 last
assert order == ["hassh", "jarm", "tcpfp", "ipv6_leak"]
# priority=100: hassh/jarm/tcpfp (alphabetical)
# priority=200: tls_certificate
# priority=850: icmp_error
# priority=999: ipv6_leak
assert order == ["hassh", "jarm", "tcpfp", "tls_certificate", "icmp_error", "ipv6_leak"]
def test_priority10_probe_sorts_first(self):
class _FastProbe(ActiveProbe):
@@ -48,7 +51,7 @@ class TestRegistryContents:
order = [cls.probe_name for cls in ActiveProbeMeta.all()]
assert order[0] == "_fast_test_probe"
assert set(order[1:]) == {"hassh", "jarm", "tcpfp", "ipv6_leak"}
assert set(order[1:]) == {"hassh", "jarm", "tcpfp", "ipv6_leak", "tls_certificate", "icmp_error"}
def test_port_none_probe_dispatched_with_none_port(self):
"""_run_probe must call run(ip, None, timeout) for a port-free probe."""