feat(sniffer): ISN sequence classifier (reuses seq_class helper)
Mirrors the IP-ID classifier for TCP ISN values: per-source-IP rolling deque (maxlen=8) populated from each inbound SYN's tcp.seq, classified on every emission. A 'random' verdict is the modern norm; 'incremental', 'zero', or 'constant' indicates legacy stacks or hand-rolled raw-socket tooling — a strong fingerprint signal. Active prober now also captures server_isn (single sample, not classified in-flight; downstream consumers correlating multi-probe results can apply seq_class.classify_sequence themselves). Profiler rollup carries the latest non-'unknown' label into attacker.tcp_fingerprint. Dedup key already covers isn_class from the previous commit, so transitions emit cleanly. UI surfaces ISN class as a colour-coded tag with a ⚠ glyph for non-random verdicts, since they're the genuinely interesting case.
This commit is contained in:
@@ -139,6 +139,10 @@ def _parse_synack(resp: Any) -> dict[str, Any]:
|
||||
|
||||
# TCP fields
|
||||
window_size = tcp_layer.window
|
||||
# Server ISN: single sample from one probe — not classified here, but
|
||||
# exported so a downstream consumer correlating multiple probes against
|
||||
# the same target can apply seq_class.classify_sequence().
|
||||
server_isn = int(getattr(tcp_layer, "seq", 0))
|
||||
|
||||
# Parse TCP options
|
||||
mss = 0
|
||||
@@ -165,6 +169,7 @@ def _parse_synack(resp: Any) -> dict[str, Any]:
|
||||
"tos": tos,
|
||||
"dscp": dscp,
|
||||
"ecn": ecn,
|
||||
"server_isn": server_isn,
|
||||
"mss": mss,
|
||||
"window_scale": window_scale,
|
||||
"sack_ok": sack_ok,
|
||||
|
||||
@@ -415,6 +415,7 @@ def _tcpfp_phase(
|
||||
tos=str(result["tos"]),
|
||||
dscp=str(result["dscp"]),
|
||||
ecn=str(result["ecn"]),
|
||||
server_isn=str(result["server_isn"]),
|
||||
msg=f"TCPFP {ip}:{port} = {result['tcpfp_hash']}",
|
||||
)
|
||||
logger.info("prober: TCPFP %s:%d = %s", ip, port, result["tcpfp_hash"])
|
||||
|
||||
@@ -144,6 +144,7 @@ def sniffer_rollup(events: list[LogEvent]) -> dict[str, Any]:
|
||||
hops: list[int] = []
|
||||
tcp_fp: dict[str, Any] | None = None
|
||||
ipid_latest: str | None = None
|
||||
isn_latest: str | None = None
|
||||
# Tracks which event set tcp_fp last — picks the provider "context"
|
||||
# (syn vs synack) when we feed the p0f-v2 matcher below.
|
||||
tcp_fp_context: str = "syn"
|
||||
@@ -193,6 +194,10 @@ def sniffer_rollup(events: list[LogEvent]) -> dict[str, Any]:
|
||||
if ipid_class and ipid_class != "unknown":
|
||||
ipid_latest = ipid_class
|
||||
tcp_fp["ipid_class"] = ipid_latest
|
||||
isn_class = e.fields.get("isn_class")
|
||||
if isn_class and isn_class != "unknown":
|
||||
isn_latest = isn_class
|
||||
tcp_fp["isn_class"] = isn_latest
|
||||
tcp_fp_context = "syn"
|
||||
|
||||
elif e.event_type == _SNIFFER_FLOW_EVENT:
|
||||
|
||||
@@ -752,6 +752,7 @@ class SnifferEngine:
|
||||
# we can label them random/incremental/zero/constant.
|
||||
self._SEQ_SAMPLE_SIZE = 8
|
||||
self._ipid_samples: dict[str, deque[int]] = {}
|
||||
self._isn_samples: dict[str, deque[int]] = {}
|
||||
|
||||
# Per-flow timing aggregator. Key: (src_ip, src_port, dst_ip, dst_port).
|
||||
# Flow direction is client→decky; reverse packets are associated back
|
||||
@@ -1052,6 +1053,12 @@ class SnifferEngine:
|
||||
)
|
||||
ipid_buf.append(int(ip.id))
|
||||
ipid_class = classify_sequence(list(ipid_buf))
|
||||
|
||||
isn_buf = self._isn_samples.setdefault(
|
||||
src_ip, deque(maxlen=self._SEQ_SAMPLE_SIZE)
|
||||
)
|
||||
isn_buf.append(int(tcp.seq))
|
||||
isn_class = classify_sequence(list(isn_buf))
|
||||
os_label = guess_os(
|
||||
ttl=ip.ttl,
|
||||
window=int(tcp.window),
|
||||
@@ -1082,6 +1089,8 @@ class SnifferEngine:
|
||||
ecn=str(int(getattr(ip, "tos", 0)) & 0x3),
|
||||
ipid_class=ipid_class,
|
||||
ipid_samples=str(len(ipid_buf)),
|
||||
isn_class=isn_class,
|
||||
isn_samples=str(len(isn_buf)),
|
||||
os_guess=os_label,
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user