anti/DECNET
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: JA4/JA4S/JA4L fingerprints, TLS session resumption, certificate extraction

Browse Source 
Extend the passive TLS sniffer with next-gen attacker fingerprinting:

- JA4 (ClientHello) and JA4S (ServerHello) computation with
  supported_versions, signature_algorithms, and ALPN parsing
- JA4L latency measurement via TCP SYN→SYN-ACK RTT tracking
- TLS session resumption detection (session tickets, PSK, 0-RTT early data)
- Certificate extraction for TLS ≤1.2 with minimal DER/ASN.1 parser
  (subject CN, issuer, SANs, validity period, self-signed flag)
- Ingester bounty extraction for all new fingerprint types
- 116 tests covering all new functionality (1255 total passing)
This commit is contained in:
anti
2026-04-13 23:20:37 -04:00
parent a022b4fed6
commit ea340065c6
4 changed files with 1621 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
tests/test_fingerprinting.py
View File

@@ -206,3 +206,199 @@ async def test_fields_missing_entirely_no_crash():
}
await _extract_bounty(repo, log_data)
repo.add_bounty.assert_not_awaited()
# ---------------------------------------------------------------------------
# JA4/JA4S extraction (sniffer)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_ja4_included_in_ja3_bounty():
repo = _make_repo()
log_data = {
"decky": "decky-05",
"service": "sniffer",
"attacker_ip": "10.0.0.20",
"event_type": "tls_session",
"fields": {
"ja3": "abc123",
"ja3s": "def456",
"ja4": "t13d0203h2_aabbccddee00_112233445566",
"ja4s": "t1302h2_ffeeddccbbaa",
"tls_version": "TLS 1.3",
"dst_port": "443",
},
}
await _extract_bounty(repo, log_data)
calls = repo.add_bounty.call_args_list
ja3_calls = [c for c in calls if c[0][0]["payload"].get("fingerprint_type") == "ja3"]
assert len(ja3_calls) == 1
payload = ja3_calls[0][0][0]["payload"]
assert payload["ja4"] == "t13d0203h2_aabbccddee00_112233445566"
assert payload["ja4s"] == "t1302h2_ffeeddccbbaa"
# ---------------------------------------------------------------------------
# JA4L latency extraction
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_ja4l_bounty_extracted():
repo = _make_repo()
log_data = {
"decky": "decky-05",
"service": "sniffer",
"attacker_ip": "10.0.0.21",
"event_type": "tls_session",
"fields": {
"ja4l_rtt_ms": "12.5",
"ja4l_client_ttl": "64",
},
}
await _extract_bounty(repo, log_data)
calls = repo.add_bounty.call_args_list
ja4l_calls = [c for c in calls if c[0][0]["payload"].get("fingerprint_type") == "ja4l"]
assert len(ja4l_calls) == 1
payload = ja4l_calls[0][0][0]["payload"]
assert payload["rtt_ms"] == "12.5"
assert payload["client_ttl"] == "64"
@pytest.mark.asyncio
async def test_ja4l_not_extracted_without_rtt():
repo = _make_repo()
log_data = {
"decky": "decky-05",
"service": "sniffer",
"attacker_ip": "10.0.0.22",
"event_type": "tls_session",
"fields": {
"ja4l_client_ttl": "64",
},
}
await _extract_bounty(repo, log_data)
calls = repo.add_bounty.call_args_list
ja4l_calls = [c for c in calls if c[0][0].get("payload", {}).get("fingerprint_type") == "ja4l"]
assert len(ja4l_calls) == 0
# ---------------------------------------------------------------------------
# TLS session resumption extraction
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_tls_resumption_bounty_extracted():
repo = _make_repo()
log_data = {
"decky": "decky-05",
"service": "sniffer",
"attacker_ip": "10.0.0.23",
"event_type": "tls_client_hello",
"fields": {
"resumption": "session_ticket,psk",
},
}
await _extract_bounty(repo, log_data)
calls = repo.add_bounty.call_args_list
res_calls = [c for c in calls if c[0][0]["payload"].get("fingerprint_type") == "tls_resumption"]
assert len(res_calls) == 1
assert res_calls[0][0][0]["payload"]["mechanisms"] == "session_ticket,psk"
@pytest.mark.asyncio
async def test_no_resumption_no_bounty():
repo = _make_repo()
log_data = {
"decky": "decky-05",
"service": "sniffer",
"attacker_ip": "10.0.0.24",
"event_type": "tls_client_hello",
"fields": {
"ja3": "abc123",
},
}
await _extract_bounty(repo, log_data)
calls = repo.add_bounty.call_args_list
res_calls = [c for c in calls if c[0][0]["payload"].get("fingerprint_type") == "tls_resumption"]
assert len(res_calls) == 0
# ---------------------------------------------------------------------------
# TLS certificate extraction
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_tls_certificate_bounty_extracted():
repo = _make_repo()
log_data = {
"decky": "decky-05",
"service": "sniffer",
"attacker_ip": "10.0.0.25",
"event_type": "tls_certificate",
"fields": {
"subject_cn": "evil.c2.local",
"issuer": "CN=Evil CA",
"self_signed": "true",
"not_before": "230101000000Z",
"not_after": "260101000000Z",
"sans": "evil.c2.local,*.evil.c2.local",
"sni": "evil.c2.local",
},
}
await _extract_bounty(repo, log_data)
calls = repo.add_bounty.call_args_list
cert_calls = [c for c in calls if c[0][0]["payload"].get("fingerprint_type") == "tls_certificate"]
assert len(cert_calls) == 1
payload = cert_calls[0][0][0]["payload"]
assert payload["subject_cn"] == "evil.c2.local"
assert payload["self_signed"] == "true"
assert payload["issuer"] == "CN=Evil CA"
@pytest.mark.asyncio
async def test_tls_certificate_not_extracted_from_non_sniffer():
repo = _make_repo()
log_data = {
"decky": "decky-05",
"service": "http",
"attacker_ip": "10.0.0.26",
"event_type": "tls_certificate",
"fields": {
"subject_cn": "not-from-sniffer.local",
},
}
await _extract_bounty(repo, log_data)
calls = repo.add_bounty.call_args_list
cert_calls = [c for c in calls if c[0][0].get("payload", {}).get("fingerprint_type") == "tls_certificate"]
assert len(cert_calls) == 0
# ---------------------------------------------------------------------------
# Multiple fingerprints from single sniffer log
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_sniffer_log_yields_multiple_fingerprint_types():
"""A complete TLS session log with JA3 + JA4L + resumption yields 3 bounties."""
repo = _make_repo()
log_data = {
"decky": "decky-05",
"service": "sniffer",
"attacker_ip": "10.0.0.30",
"event_type": "tls_session",
"fields": {
"ja3": "abc123",
"ja3s": "def456",
"ja4": "t13d0203h2_aabb_ccdd",
"ja4s": "t1302h2_eeff",
"ja4l_rtt_ms": "5.2",
"ja4l_client_ttl": "128",
"resumption": "session_ticket",
"tls_version": "TLS 1.3",
"dst_port": "443",
},
}
await _extract_bounty(repo, log_data)
assert repo.add_bounty.await_count == 3
types = {c[0][0]["payload"]["fingerprint_type"] for c in repo.add_bounty.call_args_list}
assert types == {"ja3", "ja4l", "tls_resumption"}