383 lines
20 KiB
Python
383 lines
20 KiB
Python
"""Regression tests for detector precision, device attribution and the inventory.
|
|
|
|
Every case here mirrors a false positive or a gap that was observed on real
|
|
captures. Synthetic contexts keep the suite independent of tshark; the final
|
|
class exercises the real extractor when tshark and a capture are available.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import unittest
|
|
|
|
from analyzer import AnalysisEngine, render, FORMATS
|
|
from analyzer.context import (
|
|
CaptureContext, Conversation, DnsQuery, HttpRequest, HttpResponse, TlsSession,
|
|
ArpEvent, PlaintextAuth, AuthAttempt, FileTransfer, DhcpEvent, ServiceEvent,
|
|
)
|
|
from analyzer.findings import Category, Severity
|
|
from analyzer import tls_tables
|
|
|
|
|
|
def run(ctx):
|
|
return AnalysisEngine().analyze_context(ctx)
|
|
|
|
|
|
def titles(res):
|
|
return [f.title for f in res.findings]
|
|
|
|
|
|
class FalsePositiveRegressionTests(unittest.TestCase):
|
|
def test_bulk_flow_is_not_beaconing(self):
|
|
ctx = CaptureContext()
|
|
conv = Conversation(src_ip="192.168.1.50", dst_ip="8.8.4.4", src_port=40000,
|
|
dst_port=443, protocol="tcp", stream_id=3, packets=200,
|
|
bytes=200 * 1400)
|
|
conv.timestamps = [1000 + i * 0.01 for i in range(200)] # Line-rate pacing
|
|
ctx.conversations.append(conv)
|
|
res = run(ctx)
|
|
self.assertNotIn(Category.MALWARE_BEACONING, {f.category for f in res.findings})
|
|
|
|
def test_beaconing_to_known_provider_is_low(self):
|
|
ctx = CaptureContext()
|
|
ctx.dns_queries.append(DnsQuery(query_name="courier.push.apple.com", is_response=True,
|
|
response_ips=["17.57.146.22"], rcode="NOERROR"))
|
|
for i in range(8):
|
|
t = float(i * 60)
|
|
ctx.conversations.append(Conversation(
|
|
src_ip="192.168.1.77", dst_ip="17.57.146.22", src_port=40000 + i, dst_port=5223,
|
|
protocol="tcp", packets=2, bytes=200, first_seen=t, last_seen=t + 1,
|
|
stream_id=i, timestamps=[t, t + 1]))
|
|
res = run(ctx)
|
|
beacons = [f for f in res.findings if f.category == Category.MALWARE_BEACONING]
|
|
self.assertTrue(beacons)
|
|
self.assertEqual(beacons[0].severity, Severity.LOW)
|
|
|
|
def test_beaconing_to_unnamed_ip_is_high(self):
|
|
ctx = CaptureContext()
|
|
for i in range(8):
|
|
t = float(i * 60)
|
|
ctx.conversations.append(Conversation(
|
|
src_ip="192.168.1.77", dst_ip="45.33.32.156", src_port=40000 + i, dst_port=4444,
|
|
protocol="tcp", packets=2, bytes=200, first_seen=t, last_seen=t + 1,
|
|
stream_id=i, timestamps=[t, t + 1]))
|
|
res = run(ctx)
|
|
beacons = [f for f in res.findings if f.category == Category.MALWARE_BEACONING]
|
|
self.assertTrue(beacons)
|
|
self.assertEqual(beacons[0].severity, Severity.HIGH)
|
|
|
|
def test_dhcp_source_address_is_not_arp_spoofing(self):
|
|
ctx = CaptureContext()
|
|
h = ctx.host("0.0.0.0")
|
|
h.macs = {"aa:aa:aa:aa:aa:01", "aa:aa:aa:aa:aa:02"}
|
|
res = run(ctx)
|
|
self.assertNotIn(Category.MITM, {f.category for f in res.findings})
|
|
|
|
def test_javascript_asset_is_not_malware(self):
|
|
ctx = CaptureContext()
|
|
ctx.http_requests.append(HttpRequest(
|
|
method="GET", host="192.168.1.50", uri="/static/app.js", src_ip="192.168.1.10",
|
|
dst_ip="192.168.1.50", dst_port=80, packet_number=1))
|
|
res = run(ctx)
|
|
self.assertFalse(any("Executable" in t for t in titles(res)))
|
|
|
|
def test_normal_arp_replies_are_not_a_flood(self):
|
|
ctx = CaptureContext()
|
|
ctx.start_time, ctx.end_time = 0.0, 3600.0
|
|
for i in range(60):
|
|
ctx.arp_events.append(ArpEvent(sender_ip=f"192.168.1.{i % 20 + 1}",
|
|
sender_mac=f"bb:bb:bb:bb:bb:{i % 20:02x}",
|
|
opcode=2, timestamp=i * 60.0, packet_number=100 + i))
|
|
res = run(ctx)
|
|
self.assertFalse(any("ARP reply" in t for t in titles(res)))
|
|
|
|
def test_arp_flood_by_rate_is_flagged(self):
|
|
ctx = CaptureContext()
|
|
ctx.start_time, ctx.end_time = 0.0, 60.0
|
|
for i in range(400):
|
|
ctx.arp_events.append(ArpEvent(sender_ip="192.168.1.1", sender_mac="cc:cc:cc:cc:cc:cc",
|
|
opcode=2, timestamp=i * 0.1, packet_number=i + 1))
|
|
res = run(ctx)
|
|
self.assertTrue(any("ARP reply" in t for t in titles(res)))
|
|
|
|
def test_nxdomain_ratio_counts_responses_only(self):
|
|
ctx = CaptureContext()
|
|
for i in range(30):
|
|
ctx.dns_queries.append(DnsQuery(query_name=f"host{i}.example.net", is_response=False,
|
|
src_ip="10.0.0.9", dst_ip="10.0.0.1"))
|
|
ctx.dns_queries.append(DnsQuery(query_name=f"host{i}.example.net", is_response=True,
|
|
rcode="NXDOMAIN" if i < 15 else "NOERROR",
|
|
src_ip="10.0.0.1", dst_ip="10.0.0.9"))
|
|
res = run(ctx)
|
|
self.assertTrue(any("NXDOMAIN" in t for t in titles(res)))
|
|
|
|
def test_http_403_is_not_bruteforce(self):
|
|
ctx = CaptureContext()
|
|
for i in range(12):
|
|
ctx.http_responses.append(HttpResponse(status_code=403, src_ip="10.0.0.1",
|
|
dst_ip="10.0.0.9", packet_number=i + 1))
|
|
res = run(ctx)
|
|
self.assertFalse(any("failed HTTP" in t for t in titles(res)))
|
|
|
|
def test_cdn_with_many_named_subdomains_is_low(self):
|
|
ctx = CaptureContext()
|
|
for i in range(45):
|
|
ctx.dns_queries.append(DnsQuery(query_name=f"edge-{i}.cdn-vendor.example",
|
|
src_ip="10.0.0.9", dst_ip="10.0.0.1"))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if "subdomains" in f.title)
|
|
self.assertEqual(f.severity, Severity.LOW)
|
|
|
|
|
|
class TlsPrecisionTests(unittest.TestCase):
|
|
def test_cipher_name_mapping_from_tshark_values(self):
|
|
self.assertEqual(tls_tables.cipher_name("0xc02f"), "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256")
|
|
self.assertEqual(tls_tables.cipher_name("4866"), "TLS_AES_256_GCM_SHA384")
|
|
self.assertEqual(tls_tables.version_name("0x0304"), "TLS 1.3")
|
|
self.assertTrue(tls_tables.is_grease(0x9a9a))
|
|
self.assertFalse(tls_tables.is_grease(0xc02f))
|
|
|
|
def test_negotiated_weak_cipher_is_high(self):
|
|
ctx = CaptureContext()
|
|
ctx.tls_sessions.append(TlsSession(src_ip="10.0.0.1", dst_ip="1.2.3.4", dst_port=443,
|
|
version="TLS 1.2", cipher="TLS_RSA_WITH_RC4_128_MD5",
|
|
server_hello_seen=True, sni="x.example"))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if "Weak TLS cipher" in f.title)
|
|
self.assertEqual(f.severity, Severity.HIGH)
|
|
|
|
def test_client_offered_weak_cipher_is_medium_for_device(self):
|
|
ctx = CaptureContext(device_ip="10.0.0.1")
|
|
ctx.tls_sessions.append(TlsSession(src_ip="10.0.0.1", dst_ip="1.2.3.4", dst_port=8883,
|
|
client_hello_seen=True, client_version="TLS 1.2",
|
|
client_offered_ciphers=["TLS_RSA_WITH_3DES_EDE_CBC_SHA",
|
|
"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256"]))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if "offers broken cipher" in f.title)
|
|
self.assertEqual(f.severity, Severity.MEDIUM)
|
|
self.assertEqual(f.affected_device, "10.0.0.1")
|
|
|
|
def test_application_data_only_session_is_not_judged(self):
|
|
ctx = CaptureContext()
|
|
ctx.tls_sessions.append(TlsSession(src_ip="10.0.0.1", dst_ip="1.2.3.4", version="TLS 1.0"))
|
|
res = run(ctx)
|
|
self.assertFalse(any("Obsolete" in t for t in titles(res)))
|
|
|
|
def test_sha1_certificate_flagged(self):
|
|
ctx = CaptureContext()
|
|
ctx.tls_sessions.append(TlsSession(src_ip="10.0.0.1", dst_ip="1.2.3.4", version="TLS 1.2",
|
|
cert_seen=True, cert_sig_algorithm="sha1",
|
|
cert_key_type="RSA", cert_key_bits=1024))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if "certificate" in f.title.lower())
|
|
self.assertIn("SHA1", f.title)
|
|
self.assertIn("1024", f.title)
|
|
|
|
|
|
class NewSignalTests(unittest.TestCase):
|
|
def test_json_login_over_http(self):
|
|
ctx = CaptureContext()
|
|
ctx.http_requests.append(HttpRequest(
|
|
method="POST", host="192.168.1.50", uri="/api/login", content_type="application/json",
|
|
body='{"username":"admin","password":"12345678"}', src_ip="192.168.1.10",
|
|
dst_ip="192.168.1.50", dst_port=80, packet_number=4))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if "Login form" in f.title)
|
|
self.assertEqual(f.evidence["username"], "admin")
|
|
self.assertNotEqual(f.evidence["password"], "12345678")
|
|
|
|
def test_query_string_credentials(self):
|
|
ctx = CaptureContext()
|
|
ctx.http_requests.append(HttpRequest(
|
|
method="GET", host="cam.local", uri="/cgi-bin/login?user=root&pwd=toor",
|
|
src_ip="192.168.1.10", dst_ip="192.168.1.50", dst_port=80))
|
|
res = run(ctx)
|
|
self.assertTrue(any("Login form" in t for t in titles(res)))
|
|
|
|
def test_mqtt_credentials(self):
|
|
ctx = CaptureContext(device_ip="192.168.1.60")
|
|
ctx.plaintext_auth.append(PlaintextAuth(protocol="mqtt", username="dev01", password="pw",
|
|
src_ip="192.168.1.60", dst_ip="10.1.1.1", dst_port=1883))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if "MQTT" in f.title)
|
|
self.assertEqual(f.severity, Severity.CRITICAL)
|
|
self.assertEqual(f.affected_device, "192.168.1.60")
|
|
|
|
def test_snmp_community(self):
|
|
ctx = CaptureContext()
|
|
ctx.plaintext_auth.append(PlaintextAuth(protocol="snmp", password="public",
|
|
src_ip="10.0.0.2", dst_ip="10.0.0.3", dst_port=161))
|
|
res = run(ctx)
|
|
self.assertTrue(any("SNMP community" in t for t in titles(res)))
|
|
|
|
def test_failed_logins_from_auth_attempts(self):
|
|
ctx = CaptureContext()
|
|
for i in range(6):
|
|
ctx.auth_attempts.append(AuthAttempt(protocol="ftp", success=False, src_ip="10.0.0.9",
|
|
dst_ip="10.0.0.5", dst_port=21, username="admin",
|
|
packet_number=i + 1))
|
|
res = run(ctx)
|
|
self.assertTrue(any("failed FTP" in t for t in titles(res)))
|
|
|
|
def test_firmware_over_http_to_device(self):
|
|
ctx = CaptureContext(device_ip="192.168.1.60")
|
|
ctx.file_transfers.append(FileTransfer(filename="fw_v2.bin", extension=".bin",
|
|
content_type="application/octet-stream", protocol="http",
|
|
src_ip="1.2.3.4", dst_ip="192.168.1.60", direction="download"))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if "Firmware" in f.title)
|
|
self.assertEqual(f.severity, Severity.HIGH)
|
|
self.assertEqual(f.affected_device, "192.168.1.60")
|
|
|
|
def test_dissected_mqtt_flow_is_flagged_by_protocol_not_port(self):
|
|
ctx = CaptureContext()
|
|
ctx.conversations.append(Conversation(src_ip="192.168.1.60", dst_ip="10.1.1.1", src_port=5000,
|
|
dst_port=4711, protocol="tcp", packets=20, bytes=2000,
|
|
payload_bytes=800, synack_count=1, app_protocols={"mqtt"}))
|
|
res = run(ctx)
|
|
self.assertTrue(any("MQTT" in t and "4711" in t for t in titles(res)))
|
|
|
|
def test_encrypted_flow_on_port_80_not_flagged(self):
|
|
ctx = CaptureContext()
|
|
ctx.conversations.append(Conversation(src_ip="192.168.1.60", dst_ip="10.1.1.1", src_port=5000,
|
|
dst_port=80, protocol="tcp", packets=20, bytes=2000,
|
|
payload_bytes=800, synack_count=1, app_protocols={"tls"}))
|
|
res = run(ctx)
|
|
self.assertNotIn(Category.INSECURE_PROTOCOL, {f.category for f in res.findings})
|
|
|
|
def test_unanswered_syns_are_a_scan(self):
|
|
ctx = CaptureContext()
|
|
for port in range(20, 30):
|
|
ctx.conversations.append(Conversation(src_ip="10.0.0.66", dst_ip="10.0.0.5", src_port=50000 + port,
|
|
dst_port=port, protocol="tcp", packets=1, bytes=60,
|
|
syn_count=1))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if "Port scan" in f.title)
|
|
self.assertEqual(f.evidence["unanswered_ports"], 10)
|
|
|
|
def test_multiple_dhcp_servers(self):
|
|
ctx = CaptureContext()
|
|
ctx.dhcp_events.append(DhcpEvent(message_type="offer", server_id="192.168.1.1", src_ip="192.168.1.1"))
|
|
ctx.dhcp_events.append(DhcpEvent(message_type="offer", server_id="192.168.1.200", src_ip="192.168.1.200"))
|
|
res = run(ctx)
|
|
self.assertTrue(any("DHCP servers" in t for t in titles(res)))
|
|
|
|
def test_public_resolver_bypass(self):
|
|
ctx = CaptureContext(device_ip="192.168.1.60")
|
|
ctx.dns_queries.append(DnsQuery(query_name="api.vendor.example", src_ip="192.168.1.60", dst_ip="8.8.8.8"))
|
|
res = run(ctx)
|
|
self.assertTrue(any("external resolver" in t for t in titles(res)))
|
|
|
|
def test_device_services_listed(self):
|
|
ctx = CaptureContext(device_ip="192.168.1.60")
|
|
ctx.conversations.append(Conversation(src_ip="192.168.1.10", dst_ip="192.168.1.60", src_port=5000,
|
|
dst_port=8080, protocol="tcp", packets=5, bytes=500,
|
|
synack_count=1, app_protocols={"http"}, payload_bytes=100))
|
|
res = run(ctx)
|
|
self.assertTrue(any("offers a network service on port 8080" in t for t in titles(res)))
|
|
|
|
|
|
class DeviceAttributionTests(unittest.TestCase):
|
|
def test_device_findings_sorted_first_within_severity(self):
|
|
ctx = CaptureContext(device_ip="192.168.1.60")
|
|
ctx.plaintext_auth.append(PlaintextAuth(protocol="ftp", username="a", password="b",
|
|
src_ip="192.168.1.10", dst_ip="10.0.0.5", dst_port=21))
|
|
ctx.plaintext_auth.append(PlaintextAuth(protocol="ftp", username="c", password="d",
|
|
src_ip="192.168.1.60", dst_ip="10.0.0.5", dst_port=21))
|
|
res = run(ctx)
|
|
self.assertEqual(res.findings[0].affected_device, "192.168.1.60")
|
|
self.assertTrue(res.findings[0].evidence["device_related"])
|
|
self.assertFalse(res.findings[1].evidence["device_related"])
|
|
self.assertEqual(res.summary["device_related"], 1)
|
|
self.assertEqual(res.summary["other_hosts"], 1)
|
|
|
|
def test_no_device_leaves_attribution_unset(self):
|
|
ctx = CaptureContext()
|
|
ctx.plaintext_auth.append(PlaintextAuth(protocol="ftp", username="a", password="b",
|
|
src_ip="192.168.1.10", dst_ip="10.0.0.5", dst_port=21))
|
|
res = run(ctx)
|
|
self.assertNotIn("device_related", res.findings[0].evidence)
|
|
self.assertFalse(res.inventory["device"]["configured"])
|
|
|
|
|
|
class InventoryTests(unittest.TestCase):
|
|
def _ctx(self):
|
|
ctx = CaptureContext(path="x.pcap", device_ip="192.168.1.60")
|
|
ctx.start_time, ctx.end_time = 0.0, 120.0
|
|
h = ctx.host("192.168.1.60")
|
|
h.macs.add("00:11:22:33:44:55")
|
|
h.hostnames.add("sensor-1")
|
|
ctx.host("45.33.32.157")
|
|
ctx.conversations.append(Conversation(src_ip="192.168.1.60", dst_ip="45.33.32.157", src_port=5000,
|
|
dst_port=8883, protocol="tcp", packets=50, bytes=5000,
|
|
bytes_src_to_dst=3000, bytes_dst_to_src=2000, payload_bytes=4000,
|
|
synack_count=1, app_protocols={"tls"}, first_seen=1.0, last_seen=100.0))
|
|
ctx.dns_queries.append(DnsQuery(query_name="mqtt.vendor.example", src_ip="192.168.1.60", dst_ip="192.168.1.1"))
|
|
ctx.dns_queries.append(DnsQuery(query_name="mqtt.vendor.example", is_response=True, rcode="NOERROR",
|
|
response_ips=["45.33.32.157"], src_ip="192.168.1.1", dst_ip="192.168.1.60"))
|
|
ctx.tls_sessions.append(TlsSession(src_ip="192.168.1.60", dst_ip="45.33.32.157", dst_port=8883,
|
|
version="TLS 1.2", cipher="TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
|
|
sni="mqtt.vendor.example", server_hello_seen=True, client_hello_seen=True,
|
|
client_version="TLS 1.2", cert_seen=True, cert_subject_cn="*.vendor.example",
|
|
cert_issuer_cn="Example CA", client_cert_seen=True))
|
|
return ctx
|
|
|
|
def test_inventory_sections_and_device_focus(self):
|
|
res = run(self._ctx())
|
|
inv = res.inventory
|
|
for key in ("capture", "device", "hosts", "external_endpoints", "protocols", "services",
|
|
"tls", "dns", "http", "cleartext", "coverage"):
|
|
self.assertIn(key, inv)
|
|
self.assertTrue(inv["device"]["configured"])
|
|
self.assertEqual(inv["device"]["names"], ["sensor-1"])
|
|
self.assertEqual(inv["device"]["peers"][0]["names"], ["mqtt.vendor.example"])
|
|
self.assertTrue(inv["external_endpoints"][0]["device"])
|
|
self.assertEqual(inv["external_endpoints"][0]["names"], ["mqtt.vendor.example"])
|
|
self.assertTrue(inv["tls"]["sessions"][0]["mutual_tls"])
|
|
self.assertEqual(inv["services"][0]["service"], "mqtts")
|
|
self.assertTrue(inv["coverage"]["complete"])
|
|
|
|
def test_inventory_survives_roundtrip_and_renders(self):
|
|
import json
|
|
from analyzer import result_from_dict
|
|
res = run(self._ctx())
|
|
rebuilt = result_from_dict(json.loads(json.dumps(res.to_dict())))
|
|
self.assertEqual(rebuilt.inventory["device"]["ip"], "192.168.1.60")
|
|
for fmt in FORMATS:
|
|
out = render(rebuilt, fmt)
|
|
self.assertIn("192.168.1.60", out)
|
|
self.assertIn("mqtt.vendor.example", out)
|
|
|
|
|
|
@unittest.skipUnless(shutil.which("tshark"), "tshark is not installed")
|
|
class RealCaptureTests(unittest.TestCase):
|
|
"""Golden checks against the bundled demo capture through the real extractor."""
|
|
|
|
def _capture(self, name):
|
|
here = os.path.dirname(os.path.abspath(__file__))
|
|
for base in (os.path.join(here, "..", "..", "..", "demo_files"),
|
|
os.environ.get("ANALYZER_DEMO_DIR", "")):
|
|
p = os.path.join(base, name) if base else None
|
|
if p and os.path.exists(p):
|
|
return p
|
|
return None
|
|
|
|
def test_demo_capture_end_to_end(self):
|
|
path = self._capture("demo_capture.pcapng")
|
|
if not path:
|
|
self.skipTest("demo capture not available")
|
|
from analyzer.extract_tshark import build_context_from_pcap
|
|
from analyzer.artifacts import extract_urls, extract_hashes
|
|
ctx = build_context_from_pcap(path)
|
|
self.assertEqual(ctx.packet_count, 2)
|
|
self.assertFalse(ctx.truncated)
|
|
self.assertEqual(len(ctx.http_requests), 1)
|
|
self.assertTrue(ctx.http_requests[0].body.startswith("5f4dcc3b")) # Decoded, not hex
|
|
urls = {u.value for u in extract_urls(ctx)}
|
|
self.assertIn("eu.iot.dreame.tech", urls) # From the retransmitted, undissected segment
|
|
self.assertGreaterEqual(len(extract_hashes(ctx)), 10)
|
|
res = AnalysisEngine().analyze_context(ctx)
|
|
self.assertTrue(any("HTTP" in f.title for f in res.findings))
|
|
self.assertIn("http", res.inventory["cleartext"]["flows"][0]["protocol"])
|