"""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"])