882 lines
40 KiB
Python
882 lines
40 KiB
Python
"""Unit tests for the framework-independent analysis core.
|
|
|
|
The tests use synthetic capture contexts and therefore require NO
|
|
tshark or pyshark. Run with:
|
|
|
|
cd django_project
|
|
python -m unittest analyzer.tests.test_analyzer -v
|
|
"""
|
|
|
|
import base64
|
|
import unittest
|
|
|
|
from analyzer import AnalysisEngine, render, FORMATS
|
|
from analyzer.context import (
|
|
CaptureContext, Conversation, DnsQuery, HttpRequest, HttpResponse,
|
|
TlsSession, ArpEvent, PlaintextAuth, AuthMaterial, Host,
|
|
)
|
|
from analyzer.findings import Finding, Category, Severity, Confidence
|
|
from analyzer import scoring
|
|
|
|
|
|
def run(ctx):
|
|
return AnalysisEngine().analyze_context(ctx)
|
|
|
|
|
|
def categories(result):
|
|
return {f.category for f in result.findings}
|
|
|
|
|
|
class PlaintextCredentialTests(unittest.TestCase):
|
|
def test_ftp_plaintext(self):
|
|
ctx = CaptureContext()
|
|
ctx.plaintext_auth.append(PlaintextAuth(
|
|
protocol="ftp", username="admin", password="secret",
|
|
src_ip="192.168.1.10", dst_ip="192.168.1.1", dst_port=21))
|
|
res = run(ctx)
|
|
self.assertIn(Category.PLAINTEXT_CREDENTIALS, categories(res))
|
|
f = next(f for f in res.findings if f.category == Category.PLAINTEXT_CREDENTIALS)
|
|
self.assertEqual(f.severity, Severity.CRITICAL)
|
|
self.assertEqual(f.evidence["username"], "admin")
|
|
self.assertNotEqual(f.evidence["password"], "secret") # Masked in evidence.
|
|
|
|
def test_http_basic_auth(self):
|
|
token = base64.b64encode(b"bob:hunter2").decode()
|
|
ctx = CaptureContext()
|
|
ctx.http_requests.append(HttpRequest(
|
|
method="GET", host="example.com", uri="/admin",
|
|
authorization=f"Basic {token}", src_ip="10.0.0.5", dst_ip="93.184.216.34",
|
|
dst_port=80))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if "Basic authentication over HTTP" in f.title)
|
|
self.assertEqual(f.evidence["username"], "bob")
|
|
|
|
|
|
class TokenSecretTests(unittest.TestCase):
|
|
def test_jwt_detected(self):
|
|
jwt = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NSJ9.abcDEF123_-xyz"
|
|
ctx = CaptureContext()
|
|
ctx.text_payloads.append({"text": f"Authorization: Bearer {jwt}",
|
|
"protocol": "http", "src_ip": "10.0.0.1",
|
|
"dst_ip": "1.2.3.4", "packet_number": 7})
|
|
res = run(ctx)
|
|
self.assertIn(Category.TOKENS_SECRETS, categories(res))
|
|
|
|
def test_session_cookie_over_http(self):
|
|
ctx = CaptureContext()
|
|
ctx.text_payloads.append({"text": "Cookie: sessionid=ABCDEF1234567890",
|
|
"protocol": "http", "src_ip": "10.0.0.1",
|
|
"dst_ip": "1.2.3.4"})
|
|
res = run(ctx)
|
|
self.assertTrue(any("Session cookie" in f.title for f in res.findings))
|
|
|
|
|
|
class InsecureProtocolTests(unittest.TestCase):
|
|
def test_telnet_flagged_critical(self):
|
|
ctx = CaptureContext()
|
|
ctx.conversations.append(Conversation(
|
|
src_ip="192.168.0.2", dst_ip="192.168.0.3", src_port=50000, dst_port=23,
|
|
protocol="tcp", packets=10, bytes=500, payload_bytes=200, synack_count=1,
|
|
app_protocols={"telnet"}))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if f.category == Category.INSECURE_PROTOCOL)
|
|
self.assertEqual(f.severity, Severity.CRITICAL)
|
|
self.assertEqual(f.port, 23)
|
|
|
|
|
|
class TlsWeaknessTests(unittest.TestCase):
|
|
def test_old_tls_version(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.0", cipher="TLS_RSA_WITH_RC4_128_SHA", sni="legacy.example",
|
|
server_hello_seen=True))
|
|
res = run(ctx)
|
|
titles = [f.title for f in res.findings if f.category == Category.TLS_WEAKNESS]
|
|
self.assertTrue(any("Obsolete TLS" in t for t in titles))
|
|
self.assertTrue(any("cipher" in t.lower() for t in titles))
|
|
|
|
|
|
class DnsAnomalyTests(unittest.TestCase):
|
|
def test_dns_tunneling_many_subdomains(self):
|
|
ctx = CaptureContext()
|
|
for i in range(40):
|
|
ctx.dns_queries.append(DnsQuery(
|
|
query_name=f"data{i}chunk.tunnel.example.com", rcode="NOERROR",
|
|
src_ip="10.0.0.9", dst_ip="8.8.8.8"))
|
|
res = run(ctx)
|
|
self.assertIn(Category.DNS_ANOMALY, categories(res))
|
|
|
|
def test_high_entropy_domain(self):
|
|
ctx = CaptureContext()
|
|
ctx.dns_queries.append(DnsQuery(
|
|
query_name="a8f3kd92jfh3kd0fj2k3.malware-c2.net", rcode="NOERROR"))
|
|
res = run(ctx)
|
|
self.assertIn(Category.DNS_ANOMALY, categories(res))
|
|
|
|
|
|
class MitmTests(unittest.TestCase):
|
|
def test_arp_spoof_multiple_macs(self):
|
|
ctx = CaptureContext()
|
|
ctx.arp_events.append(ArpEvent(sender_ip="192.168.1.1", sender_mac="aa:aa:aa:aa:aa:aa", opcode=2))
|
|
ctx.arp_events.append(ArpEvent(sender_ip="192.168.1.1", sender_mac="bb:bb:bb:bb:bb:bb", opcode=2))
|
|
res = run(ctx)
|
|
self.assertIn(Category.MITM, categories(res))
|
|
|
|
|
|
class BeaconingTests(unittest.TestCase):
|
|
def test_regular_beaconing(self):
|
|
ctx = CaptureContext()
|
|
conv = Conversation(src_ip="192.168.1.50", dst_ip="8.8.8.8", src_port=44000,
|
|
dst_port=443, protocol="tcp")
|
|
# Exact 60-second intervals produce a strong periodicity signal.
|
|
conv.timestamps = [float(t * 60) for t in range(8)]
|
|
conv.first_seen, conv.last_seen = conv.timestamps[0], conv.timestamps[-1]
|
|
ctx.conversations.append(conv)
|
|
res = run(ctx)
|
|
self.assertIn(Category.MALWARE_BEACONING, categories(res))
|
|
|
|
|
|
class ExfiltrationTests(unittest.TestCase):
|
|
def test_large_upload(self):
|
|
ctx = CaptureContext()
|
|
ctx.conversations.append(Conversation(
|
|
src_ip="192.168.1.20", dst_ip="8.8.8.8", src_port=40000, dst_port=443,
|
|
protocol="tcp", packets=100, bytes=10 * 1024 * 1024,
|
|
bytes_src_to_dst=10 * 1024 * 1024))
|
|
res = run(ctx)
|
|
self.assertIn(Category.EXFILTRATION, categories(res))
|
|
|
|
|
|
class BruteforceScanTests(unittest.TestCase):
|
|
def test_portscan(self):
|
|
ctx = CaptureContext()
|
|
for port in range(1000, 1030):
|
|
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))
|
|
res = run(ctx)
|
|
self.assertTrue(any("Port scan" in f.title for f in res.findings))
|
|
|
|
def test_malware_download(self):
|
|
ctx = CaptureContext()
|
|
ctx.http_requests.append(HttpRequest(
|
|
method="GET", host="evil.example", uri="/payload.exe",
|
|
src_ip="10.0.0.1", dst_ip="1.2.3.4", dst_port=80))
|
|
res = run(ctx)
|
|
self.assertTrue(any(".exe" in f.title for f in res.findings))
|
|
|
|
|
|
class PasswordHashTests(unittest.TestCase):
|
|
def test_ntlm_material(self):
|
|
ctx = CaptureContext()
|
|
ctx.auth_material.append(AuthMaterial(
|
|
kind="ntlmssp", account="user1", domain="CORP",
|
|
src_ip="10.0.0.1", dst_ip="10.0.0.2", protocol="ntlm"))
|
|
res = run(ctx)
|
|
self.assertIn(Category.PASSWORD_HASHES, categories(res))
|
|
|
|
|
|
class ScoringTests(unittest.TestCase):
|
|
def _f(self, sev, conf=Confidence.HIGH, key="x"):
|
|
return Finding(title="t", category=Category.EXFILTRATION, severity=sev,
|
|
confidence=conf, risk_description="r", recommendation="rec",
|
|
evidence={"dedup_key": key})
|
|
|
|
def test_dedupe_merges_identical(self):
|
|
a = self._f(Severity.HIGH, Confidence.LOW, key="same")
|
|
b = self._f(Severity.HIGH, Confidence.HIGH, key="same")
|
|
a.related_packets = [1]
|
|
b.related_packets = [2]
|
|
merged = scoring.dedupe([a, b])
|
|
self.assertEqual(len(merged), 1)
|
|
self.assertEqual(merged[0].confidence, Confidence.HIGH)
|
|
self.assertEqual(merged[0].related_packets, [1, 2])
|
|
|
|
def test_sort_by_priority(self):
|
|
low = self._f(Severity.LOW, key="a")
|
|
crit = self._f(Severity.CRITICAL, key="b")
|
|
med = self._f(Severity.MEDIUM, key="c")
|
|
ordered = scoring.sort_by_priority([low, crit, med])
|
|
self.assertEqual([f.severity for f in ordered],
|
|
[Severity.CRITICAL, Severity.MEDIUM, Severity.LOW])
|
|
|
|
|
|
class ReportRenderingTests(unittest.TestCase):
|
|
def _result(self):
|
|
ctx = CaptureContext(path="x.pcapng", packet_count=10)
|
|
ctx.plaintext_auth.append(PlaintextAuth(
|
|
protocol="ftp", username="admin", password="secret",
|
|
src_ip="192.168.1.10", dst_ip="192.168.1.1", dst_port=21))
|
|
return run(ctx)
|
|
|
|
def test_all_formats_render(self):
|
|
res = self._result()
|
|
for fmt in FORMATS:
|
|
out = render(res, fmt)
|
|
self.assertIsInstance(out, str)
|
|
self.assertTrue(len(out) > 0)
|
|
|
|
def test_json_is_valid(self):
|
|
import json
|
|
res = self._result()
|
|
data = json.loads(render(res, "json"))
|
|
self.assertIn("findings", data)
|
|
self.assertIn("summary", data)
|
|
self.assertGreaterEqual(len(data["findings"]), 1)
|
|
|
|
def test_extra_cloud_results_in_report(self):
|
|
import json
|
|
res = self._result()
|
|
extra = {
|
|
"urls": [{"url": "api.example.com", "analyzed": True, "ports": [443],
|
|
"tls_versions": ["TLS 1.2"], "cipher_suites": ["AES_256_GCM"]}],
|
|
"hashes": [{"hash_value": "5f4dcc3b5aa765d61d8327deb882cf99", "analyzed": True,
|
|
"cracked": True, "cracked_value": "password", "crack_time": "0.10s",
|
|
"hash_type": ["md5"]}],
|
|
}
|
|
md = render(res, "md", extra)
|
|
self.assertIn("api.example.com", md)
|
|
self.assertIn("password", md)
|
|
txt = render(res, "txt", extra)
|
|
self.assertIn("CRACKED", txt)
|
|
data = json.loads(render(res, "json", extra))
|
|
self.assertIn("cloud_results", data)
|
|
self.assertEqual(data["cloud_results"]["hashes"][0]["cracked_value"], "password")
|
|
|
|
|
|
class ReconstructionTests(unittest.TestCase):
|
|
def test_roundtrip_from_dict_renders(self):
|
|
import json
|
|
from analyzer import result_from_dict, render
|
|
ctx = CaptureContext(path="x.pcapng", packet_count=5)
|
|
ctx.plaintext_auth.append(PlaintextAuth(
|
|
protocol="ftp", username="admin", password="secret",
|
|
src_ip="192.168.1.10", dst_ip="192.168.1.1", dst_port=21))
|
|
res = run(ctx)
|
|
# Serialize and reconstruct a stored result.
|
|
stored = json.dumps(res.to_dict())
|
|
rebuilt = result_from_dict(json.loads(stored))
|
|
self.assertEqual(len(rebuilt.findings), len(res.findings))
|
|
self.assertEqual(rebuilt.findings[0].finding_id, res.findings[0].finding_id)
|
|
# Every renderer must work with the reconstructed result.
|
|
for fmt in FORMATS:
|
|
out = render(rebuilt, fmt)
|
|
self.assertTrue(len(out) > 0)
|
|
|
|
|
|
class EngineRobustnessTests(unittest.TestCase):
|
|
def test_empty_capture(self):
|
|
res = run(CaptureContext(path="empty.pcap"))
|
|
self.assertEqual(res.findings, [])
|
|
self.assertEqual(res.summary["total"], 0)
|
|
|
|
def test_all_ten_detectors_registered(self):
|
|
res = run(CaptureContext())
|
|
self.assertEqual(len(res.meta["detectors_run"]), 10)
|
|
self.assertEqual(res.meta["detector_errors"], {})
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Detector regression tests
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class BeaconingConnectionTests(unittest.TestCase):
|
|
def test_beaconing_via_distinct_connections(self):
|
|
"""Detect TCP beaconing spread across multiple short connections."""
|
|
ctx = CaptureContext()
|
|
for i in range(8):
|
|
t = float(i * 30) # One new connection every 30 seconds.
|
|
ctx.conversations.append(Conversation(
|
|
src_ip="192.168.1.77", dst_ip="8.8.8.8",
|
|
src_port=40000 + i, dst_port=443, 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, "Beaconing across separate connections was not detected")
|
|
self.assertEqual(beacons[0].evidence.get("signal"), "distinct_connections")
|
|
self.assertEqual(beacons[0].destination_ip, "8.8.8.8")
|
|
|
|
def test_irregular_connections_not_flagged(self):
|
|
ctx = CaptureContext()
|
|
for i, t in enumerate([0.0, 5.0, 90.0, 91.0, 400.0, 402.0]):
|
|
ctx.conversations.append(Conversation(
|
|
src_ip="192.168.1.77", dst_ip="8.8.8.8",
|
|
src_port=40000 + i, dst_port=443, protocol="tcp",
|
|
first_seen=t, last_seen=t, timestamps=[t]))
|
|
res = run(ctx)
|
|
self.assertNotIn(Category.MALWARE_BEACONING, categories(res))
|
|
|
|
|
|
class ProtocolStatsCaseTests(unittest.TestCase):
|
|
def test_eapol_handshake_real_casing(self):
|
|
"""Protocol statistics remain case-insensitive for tshark layer names."""
|
|
ctx = CaptureContext()
|
|
ctx.protocol_stats["EAPOL"] = 4
|
|
res = run(ctx)
|
|
self.assertTrue(any("EAPOL" in f.title for f in res.findings))
|
|
|
|
def test_smbv1_real_casing(self):
|
|
ctx = CaptureContext()
|
|
ctx.protocol_stats["SMB"] = 20
|
|
res = run(ctx)
|
|
self.assertTrue(any("SMBv1" in f.title for f in res.findings))
|
|
|
|
def test_smbv1_suppressed_when_smb2_present(self):
|
|
ctx = CaptureContext()
|
|
ctx.protocol_stats["SMB"] = 20
|
|
ctx.protocol_stats["SMB2"] = 100
|
|
res = run(ctx)
|
|
self.assertFalse(any("SMBv1" in f.title for f in res.findings))
|
|
|
|
|
|
class TlsCertTests(unittest.TestCase):
|
|
def test_self_signed_certificate_flagged(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", sni="iot.local",
|
|
cert_subject_cn="iot.local", cert_issuer_cn="iot.local",
|
|
cert_self_signed=True, cert_seen=True))
|
|
res = run(ctx)
|
|
self.assertTrue(any("self-signed" in f.title for f in res.findings))
|
|
|
|
def test_expired_certificate_flagged(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", cert_expired=True, cert_seen=True))
|
|
res = run(ctx)
|
|
self.assertTrue(any("expired" in f.title for f in res.findings))
|
|
|
|
|
|
class CertHelperTests(unittest.TestCase):
|
|
def test_self_signed_helper(self):
|
|
from analyzer.extract_tshark import _cert_is_self_signed
|
|
self.assertTrue(_cert_is_self_signed("device.local", "device.local"))
|
|
self.assertFalse(_cert_is_self_signed("device.local", "DigiCert CA"))
|
|
self.assertIsNone(_cert_is_self_signed("device.local", None))
|
|
|
|
def test_expired_helper(self):
|
|
from analyzer.extract_tshark import _cert_is_expired
|
|
# The certificate expired before the 2025 capture timestamp.
|
|
now = 1735689600.0 # 2025-01-01 UTC
|
|
self.assertTrue(_cert_is_expired("2020-01-01 00:00:00", now))
|
|
self.assertFalse(_cert_is_expired("2030-01-01 00:00:00", now))
|
|
self.assertIsNone(_cert_is_expired(None, now))
|
|
|
|
def test_parse_cert_time_formats(self):
|
|
from analyzer.extract_tshark import _parse_cert_time
|
|
self.assertIsNotNone(_parse_cert_time("2020-01-01 00:00:00"))
|
|
self.assertIsNotNone(_parse_cert_time("Jan 1 00:00:00 2020 GMT"))
|
|
self.assertIsNotNone(_parse_cert_time("200101000000Z"))
|
|
self.assertIsNone(_parse_cert_time("nonsense"))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Authentication material reconstruction
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class CrackFormatTests(unittest.TestCase):
|
|
def test_netntlmv2(self):
|
|
from analyzer.crackformats import build_netntlmv2
|
|
nt = "a" * 32 + "b" * 40 # NTProof(16B) + Blob
|
|
out = build_netntlmv2("alice", "CORP", "1122334455667788", nt)
|
|
self.assertIsNotNone(out)
|
|
self.assertEqual(out["hashcat_mode"], 5600)
|
|
self.assertEqual(out["hashcat"],
|
|
"alice::CORP:1122334455667788:" + "a" * 32 + ":" + "b" * 40)
|
|
|
|
def test_netntlmv2_rejects_bad_challenge(self):
|
|
from analyzer.crackformats import build_netntlmv2
|
|
self.assertIsNone(build_netntlmv2("alice", "CORP", "1122", "a" * 40))
|
|
|
|
def test_krb5_asrep(self):
|
|
from analyzer.crackformats import build_krb5_asrep
|
|
cipher = "c" * 32 + "d" * 64
|
|
out = build_krb5_asrep("bob", "corp.local", 23, cipher)
|
|
self.assertEqual(out["hashcat_mode"], 18200)
|
|
self.assertTrue(out["hashcat"].startswith("$krb5asrep$23$bob@CORP.LOCAL:"))
|
|
|
|
def test_krb5_tgsrep(self):
|
|
from analyzer.crackformats import build_krb5_tgsrep
|
|
cipher = "e" * 32 + "f" * 64
|
|
out = build_krb5_tgsrep("svc", "corp.local", "MSSQL/db", 23, cipher)
|
|
self.assertEqual(out["hashcat_mode"], 13100)
|
|
self.assertIn("$krb5tgs$23$*svc$CORP.LOCAL$MSSQL/db*$", out["hashcat"])
|
|
|
|
def test_smtp_auth_plain(self):
|
|
import base64
|
|
from analyzer.crackformats import decode_smtp_auth_plain
|
|
token = base64.b64encode(b"\x00joe\x00s3cret").decode()
|
|
self.assertEqual(decode_smtp_auth_plain(token), ("joe", "s3cret"))
|
|
|
|
def test_smtp_auth_login(self):
|
|
import base64
|
|
from analyzer.crackformats import decode_smtp_auth_login
|
|
u = base64.b64encode(b"joe").decode()
|
|
p = base64.b64encode(b"s3cret").decode()
|
|
self.assertEqual(decode_smtp_auth_login(u, p), ("joe", "s3cret"))
|
|
|
|
|
|
class CrackableAuthMaterialTests(unittest.TestCase):
|
|
def test_hashcat_material_surfaced_and_high_confidence(self):
|
|
ctx = CaptureContext()
|
|
ctx.auth_material.append(AuthMaterial(
|
|
kind="netntlmv2", account="alice", domain="CORP",
|
|
src_ip="10.0.0.5", dst_ip="10.0.0.1", protocol="ntlm",
|
|
hashcat="alice::CORP:1122334455667788:" + "a" * 32 + ":" + "b" * 40,
|
|
hashcat_mode=5600, hash_format="NetNTLMv2"))
|
|
res = run(ctx)
|
|
ph = [f for f in res.findings if f.category == Category.PASSWORD_HASHES]
|
|
self.assertTrue(ph)
|
|
f = ph[0]
|
|
self.assertEqual(f.confidence, Confidence.HIGH)
|
|
self.assertIn("crackable", f.title)
|
|
self.assertEqual(f.evidence.get("hashcat_mode"), 5600)
|
|
self.assertTrue(f.evidence.get("hashcat", "").startswith("alice::CORP:"))
|
|
|
|
|
|
class MailPlaintextAuthTests(unittest.TestCase):
|
|
def test_pop3_plaintext(self):
|
|
ctx = CaptureContext()
|
|
ctx.plaintext_auth.append(PlaintextAuth(
|
|
protocol="pop3", username="joe", password="pw123",
|
|
src_ip="10.0.0.5", dst_ip="10.0.0.2", dst_port=110))
|
|
res = run(ctx)
|
|
self.assertTrue(any("POP3" in f.title for f in res.findings))
|
|
|
|
def test_smtp_plaintext(self):
|
|
ctx = CaptureContext()
|
|
ctx.plaintext_auth.append(PlaintextAuth(
|
|
protocol="smtp", username="joe", password="pw123",
|
|
src_ip="10.0.0.5", dst_ip="10.0.0.2", dst_port=25))
|
|
res = run(ctx)
|
|
self.assertTrue(any("SMTP" in f.title for f in res.findings))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# False-positive regression tests
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class UserAgentTests(unittest.TestCase):
|
|
def _ua(self, ua):
|
|
ctx = CaptureContext()
|
|
ctx.http_requests.append(HttpRequest(
|
|
method="GET", host="x.example", uri="/", user_agent=ua,
|
|
src_ip="10.0.0.5", dst_ip="1.2.3.4", dst_port=80))
|
|
return run(ctx)
|
|
|
|
def test_benign_iot_ua_not_flagged(self):
|
|
for ua in ("okhttp/4.9.0", "Java/1.8.0", "Go-http-client/1.1"):
|
|
res = self._ua(ua)
|
|
self.assertFalse(
|
|
any(f.category == Category.MALWARE_BEACONING for f in res.findings),
|
|
f"{ua} sollte kein Finding erzeugen")
|
|
|
|
def test_scanner_ua_flagged_high(self):
|
|
res = self._ua("Mozilla/5.00 (Nikto/2.1.6)")
|
|
ua = [f for f in res.findings if f.category == Category.MALWARE_BEACONING]
|
|
self.assertTrue(ua)
|
|
self.assertEqual(ua[0].severity, Severity.HIGH)
|
|
self.assertEqual(ua[0].evidence.get("tier"), "scanner")
|
|
|
|
def test_scripting_ua_flagged_low(self):
|
|
res = self._ua("curl/8.1.2")
|
|
ua = [f for f in res.findings if f.category == Category.MALWARE_BEACONING]
|
|
self.assertTrue(ua)
|
|
self.assertEqual(ua[0].severity, Severity.LOW)
|
|
|
|
|
|
class DnsQtypeTests(unittest.TestCase):
|
|
def test_qtype_name_mapping(self):
|
|
from analyzer.extract_tshark import _dns_qtype_name
|
|
self.assertEqual(_dns_qtype_name("16"), "TXT")
|
|
self.assertEqual(_dns_qtype_name("1"), "A")
|
|
self.assertEqual(_dns_qtype_name("txt"), "TXT")
|
|
self.assertIsNone(_dns_qtype_name(None))
|
|
|
|
def test_txt_tunneling_flagged(self):
|
|
ctx = CaptureContext()
|
|
for i in range(25):
|
|
ctx.dns_queries.append(DnsQuery(
|
|
query_name=f"chunk{i}.exfil.example.com", qtype="TXT",
|
|
rcode="NOERROR", src_ip="10.0.0.9", dst_ip="8.8.8.8"))
|
|
res = run(ctx)
|
|
self.assertTrue(any("TXT or NULL" in f.title for f in res.findings))
|
|
|
|
def test_normal_a_queries_no_txt_finding(self):
|
|
ctx = CaptureContext()
|
|
for i in range(25):
|
|
ctx.dns_queries.append(DnsQuery(
|
|
query_name=f"host{i}.example.com", qtype="A", rcode="NOERROR"))
|
|
res = run(ctx)
|
|
self.assertFalse(any("TXT or NULL" in f.title for f in res.findings))
|
|
|
|
|
|
class DownloadHelperTests(unittest.TestCase):
|
|
def test_is_download_content_type(self):
|
|
from analyzer.extract_tshark import _is_download_content_type
|
|
self.assertTrue(_is_download_content_type("application/octet-stream"))
|
|
self.assertTrue(_is_download_content_type("application/x-msdownload"))
|
|
self.assertFalse(_is_download_content_type("text/html; charset=utf-8"))
|
|
self.assertFalse(_is_download_content_type(None))
|
|
|
|
def test_filename_from_disposition(self):
|
|
from analyzer.extract_tshark import _filename_from_disposition
|
|
self.assertEqual(_filename_from_disposition('attachment; filename="update.exe"'), "update.exe")
|
|
self.assertEqual(_filename_from_disposition("attachment; filename=payload.bin"), "payload.bin")
|
|
self.assertIsNone(_filename_from_disposition("inline"))
|
|
|
|
|
|
class FileTransferDownloadTests(unittest.TestCase):
|
|
def test_executable_content_type_flagged(self):
|
|
from analyzer.context import FileTransfer
|
|
ctx = CaptureContext()
|
|
ctx.file_transfers.append(FileTransfer(
|
|
filename=None, extension=None, content_type="application/x-msdownload",
|
|
protocol="http", src_ip="1.2.3.4", dst_ip="10.0.0.5", direction="download"))
|
|
res = run(ctx)
|
|
self.assertTrue(any("Executable" in f.title for f in res.findings))
|
|
|
|
def test_disposition_filename_extension_flagged(self):
|
|
from analyzer.context import FileTransfer
|
|
ctx = CaptureContext()
|
|
ctx.file_transfers.append(FileTransfer(
|
|
filename="update.exe", extension=".exe", content_type="application/octet-stream",
|
|
protocol="http", src_ip="1.2.3.4", dst_ip="10.0.0.5", direction="download"))
|
|
res = run(ctx)
|
|
self.assertTrue(any(".exe" in f.title for f in res.findings))
|
|
|
|
def test_benign_download_not_flagged(self):
|
|
from analyzer.context import FileTransfer
|
|
ctx = CaptureContext()
|
|
ctx.file_transfers.append(FileTransfer(
|
|
filename="report.pdf", extension=".pdf", content_type="application/pdf",
|
|
protocol="http", src_ip="1.2.3.4", dst_ip="10.0.0.5", direction="download"))
|
|
res = run(ctx)
|
|
self.assertFalse(any(f.category == Category.BRUTEFORCE_SCAN_FILETRANSFER
|
|
for f in res.findings))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Packet-level artifact provenance
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class ProvenanceHelperTests(unittest.TestCase):
|
|
def test_derive_artifact_credentials(self):
|
|
from analyzer.provenance import derive_artifact
|
|
art = derive_artifact({"username": "joe", "password": "pw"})
|
|
self.assertEqual(art["label"], "Credentials")
|
|
self.assertIn("joe", art["value"])
|
|
|
|
def test_derive_artifact_hashcat(self):
|
|
from analyzer.provenance import derive_artifact
|
|
art = derive_artifact({"hashcat": "alice::CORP:...:...:..."})
|
|
self.assertEqual(art["label"], "Hash (hashcat)")
|
|
|
|
def test_derive_artifact_none_for_behavioral(self):
|
|
from analyzer.provenance import derive_artifact
|
|
self.assertIsNone(derive_artifact({"distinct_ports": 30}))
|
|
|
|
def test_build_sources_single_packet_has_timestamp(self):
|
|
from analyzer.provenance import build_sources
|
|
srcs = build_sources(related_packets=[42], related_streams=[3],
|
|
src_ip="10.0.0.5", dst_ip="10.0.0.2", protocol="ftp",
|
|
port=21, timestamp_first_seen=1577880000.0, capture="c.pcapng")
|
|
self.assertEqual(len(srcs), 1)
|
|
self.assertEqual(srcs[0]["packet"], 42)
|
|
self.assertEqual(srcs[0]["stream"], 3)
|
|
self.assertEqual(srcs[0]["timestamp"], 1577880000.0)
|
|
self.assertEqual(srcs[0]["capture"], "c.pcapng")
|
|
|
|
def test_build_sources_multi_packet_capped_no_per_packet_ts(self):
|
|
from analyzer.provenance import build_sources, MAX_SOURCE_PACKETS
|
|
srcs = build_sources(related_packets=list(range(1, 20)), related_streams=[1, 2],
|
|
src_ip="10.0.0.5", dst_ip="8.8.8.8", protocol="tcp",
|
|
port=443, timestamp_first_seen=100.0, capture="c.pcapng")
|
|
self.assertEqual(len(srcs), MAX_SOURCE_PACKETS)
|
|
self.assertNotIn("timestamp", srcs[0]) # Never infer per-packet timestamps.
|
|
self.assertNotIn("stream", srcs[0]) # Omit an ambiguous stream reference.
|
|
|
|
def test_build_sources_no_packet_still_has_reference(self):
|
|
from analyzer.provenance import build_sources
|
|
srcs = build_sources(related_packets=[], related_streams=[],
|
|
src_ip="10.0.0.9", dst_ip=None, protocol="dns", port=53,
|
|
timestamp_first_seen=None, capture="c.pcapng")
|
|
self.assertEqual(len(srcs), 1)
|
|
self.assertIn("note", srcs[0])
|
|
|
|
|
|
class EngineProvenanceTests(unittest.TestCase):
|
|
def test_finding_gets_capture_artifact_sources(self):
|
|
ctx = CaptureContext(path="/data/demo_capture.pcapng")
|
|
ctx.plaintext_auth.append(PlaintextAuth(
|
|
protocol="ftp", username="admin", password="s3cret",
|
|
src_ip="10.0.0.5", dst_ip="10.0.0.2", dst_port=21,
|
|
timestamp=1577880000.0, packet_number=42, stream_id=3))
|
|
res = run(ctx)
|
|
f = next(f for f in res.findings if f.category == Category.PLAINTEXT_CREDENTIALS)
|
|
# Store only the capture basename.
|
|
self.assertEqual(f.evidence["capture_file"], "demo_capture.pcapng")
|
|
self.assertEqual(f.evidence["artifact"]["label"], "Credentials")
|
|
self.assertTrue(f.evidence["sources"])
|
|
self.assertEqual(f.evidence["sources"][0]["packet"], 42)
|
|
self.assertEqual(f.evidence["sources"][0]["capture"], "demo_capture.pcapng")
|
|
|
|
def test_aggregated_findings_have_packets(self):
|
|
# Aggregated scan findings retain representative packet numbers.
|
|
ctx = CaptureContext(path="cap.pcapng")
|
|
for p in range(1000, 1020):
|
|
ctx.conversations.append(Conversation(
|
|
src_ip="10.0.0.66", dst_ip="10.0.0.5", src_port=50000 + p,
|
|
dst_port=p, protocol="tcp", packets=1, bytes=60,
|
|
packet_numbers=[p]))
|
|
res = run(ctx)
|
|
scan = next(f for f in res.findings if "Port scan" in f.title)
|
|
self.assertTrue(scan.related_packets, "Port scan has no packet evidence")
|
|
self.assertTrue(scan.evidence["sources"])
|
|
self.assertIn("packet", scan.evidence["sources"][0])
|
|
|
|
def test_beaconing_distinct_connections_have_packets(self):
|
|
ctx = CaptureContext(path="cap.pcapng")
|
|
for i in range(8):
|
|
t = float(i * 30)
|
|
ctx.conversations.append(Conversation(
|
|
src_ip="192.168.1.77", dst_ip="8.8.8.8", src_port=40000 + i,
|
|
dst_port=443, protocol="tcp", first_seen=t, last_seen=t,
|
|
timestamps=[t], stream_id=i, packet_numbers=[500 + i]))
|
|
res = run(ctx)
|
|
beacon = next(f for f in res.findings if f.category == Category.MALWARE_BEACONING)
|
|
self.assertTrue(beacon.related_packets)
|
|
|
|
|
|
class ReportProvenanceTests(unittest.TestCase):
|
|
def test_report_shows_artifact_and_packet(self):
|
|
ctx = CaptureContext(path="/data/evidence.pcapng")
|
|
ctx.plaintext_auth.append(PlaintextAuth(
|
|
protocol="ftp", username="admin", password="s3cret",
|
|
src_ip="10.0.0.5", dst_ip="10.0.0.2", dst_port=21,
|
|
timestamp=1577880000.0, packet_number=42, stream_id=3))
|
|
res = run(ctx)
|
|
md = render(res, "md")
|
|
self.assertIn("Observed artifact", md)
|
|
self.assertIn("Packet #42", md)
|
|
self.assertIn("evidence.pcapng", md)
|
|
txt = render(res, "txt")
|
|
self.assertIn("ARTIFACT", txt)
|
|
self.assertIn("Packet #42", txt)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Artifact extraction and enrichment integration
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class HashModeTests(unittest.TestCase):
|
|
def test_length_based_modes(self):
|
|
from analyzer.artifacts import identify_hash_mode
|
|
self.assertEqual(identify_hash_mode("a" * 32)[0], 0) # md5
|
|
self.assertEqual(identify_hash_mode("a" * 40)[0], 100) # sha1
|
|
self.assertEqual(identify_hash_mode("a" * 64)[0], 1400) # sha256
|
|
self.assertEqual(identify_hash_mode("a" * 128)[0], 1700) # sha512
|
|
|
|
def test_structural_modes(self):
|
|
from analyzer.artifacts import identify_hash_mode
|
|
netntlmv2 = "alice::CORP:" + "1" * 16 + ":" + "a" * 32 + ":" + "b" * 40
|
|
self.assertEqual(identify_hash_mode(netntlmv2)[0], 5600)
|
|
self.assertEqual(identify_hash_mode("$krb5asrep$23$bob@CORP:aaa$bbb")[0], 18200)
|
|
self.assertEqual(identify_hash_mode("$krb5tgs$23$*svc$CORP$spn*$a$b")[0], 13100)
|
|
self.assertEqual(identify_hash_mode("wpa*01*...")[0], 22000)
|
|
|
|
def test_hint_format_wins(self):
|
|
from analyzer.artifacts import identify_hash_mode
|
|
self.assertEqual(identify_hash_mode("whatever", hint_format="NetNTLMv2")[0], 5600)
|
|
self.assertEqual(identify_hash_mode("x", hint_format="krb5asrep(etype=23)")[0], 18200)
|
|
|
|
def test_unknown_returns_none(self):
|
|
from analyzer.artifacts import identify_hash_mode
|
|
self.assertEqual(identify_hash_mode("nothash"), (None, None))
|
|
self.assertEqual(identify_hash_mode("a" * 33), (None, None))
|
|
|
|
|
|
class ArtifactExtractionTests(unittest.TestCase):
|
|
def test_extract_urls_dedupes_and_observes(self):
|
|
from analyzer import extract_urls
|
|
ctx = CaptureContext()
|
|
ctx.dns_queries.append(DnsQuery(query_name="api.example.com.", src_ip="10.0.0.5"))
|
|
ctx.tls_sessions.append(TlsSession(src_ip="10.0.0.5", dst_ip="1.2.3.4",
|
|
dst_port=443, version="TLS 1.2", sni="api.example.com"))
|
|
ctx.http_requests.append(HttpRequest(method="GET", host="cdn.example.net",
|
|
uri="/x", src_ip="10.0.0.5", dst_ip="1.2.3.4", dst_port=80))
|
|
urls = extract_urls(ctx)
|
|
vals = {u.value for u in urls}
|
|
self.assertIn("api.example.com", vals)
|
|
self.assertIn("cdn.example.net", vals)
|
|
# DNS and SNI observations merge into one endpoint artifact.
|
|
self.assertEqual(len([u for u in urls if u.value == "api.example.com"]), 1)
|
|
api = next(u for u in urls if u.value == "api.example.com")
|
|
self.assertIn(443, api.observed_ports)
|
|
self.assertIn("TLS 1.2", api.observed_tls)
|
|
|
|
def test_extract_hashes_from_auth_material_and_payload(self):
|
|
from analyzer import extract_hashes
|
|
ctx = CaptureContext()
|
|
ntlm = "alice::CORP:" + "1" * 16 + ":" + "a" * 32 + ":" + "b" * 40
|
|
ctx.auth_material.append(AuthMaterial(
|
|
kind="netntlmv2", account="alice", hashcat=ntlm, hashcat_mode=5600,
|
|
hash_format="NetNTLMv2", src_ip="10.0.0.5", packet_number=7))
|
|
ctx.text_payloads.append({"text": "token=" + "d" * 64, "protocol": "http",
|
|
"src_ip": "10.0.0.5", "packet_number": 9})
|
|
hashes = extract_hashes(ctx)
|
|
modes = {h.hashcat_mode for h in hashes}
|
|
self.assertIn(5600, modes) # NTLM
|
|
self.assertIn(1400, modes) # SHA-256 extracted from payload text.
|
|
ntlm_art = next(h for h in hashes if h.hashcat_mode == 5600)
|
|
self.assertEqual(ntlm_art.account, "alice")
|
|
self.assertEqual(ntlm_art.kind, "netntlmv2")
|
|
|
|
|
|
class DemoPayloadRegressionTests(unittest.TestCase):
|
|
"""Extract hashes and URLs from generic HTTP body payloads."""
|
|
|
|
def test_hashes_and_urls_from_body_payload(self):
|
|
from analyzer import extract_urls, extract_hashes
|
|
ctx = CaptureContext()
|
|
ctx.text_payloads.append({"text": (
|
|
"POST /hashes HTTP/1.1\r\nHost: testserver\r\n\r\n"
|
|
"5f4dcc3b5aa765d61d8327deb882cf99\n" # md5
|
|
"e27f400cc2fe611bbb129ae0b4b4be6b32debf12\n" # sha1
|
|
"0e2cfcd348361a44a6f3e646552cb56d8d4d9555b4bcd89770f96983d8f321ad" # sha256
|
|
), "protocol": "http", "packet_number": 1})
|
|
ctx.text_payloads.append({"text": (
|
|
"GET /allowed_urls HTTP/1.1\r\nHost: testserver\r\n\r\n"
|
|
"eu.iot.dreame.tech\n"
|
|
"https://app.mt.eu.iot.dreame.tech\n"
|
|
"http://40.101.70.178/\n"
|
|
), "protocol": "http", "packet_number": 2})
|
|
|
|
hashes = extract_hashes(ctx)
|
|
self.assertEqual(len(hashes), 3)
|
|
self.assertEqual({h.hashcat_mode for h in hashes}, {0, 100, 1400})
|
|
|
|
urls = {u.value for u in extract_urls(ctx)}
|
|
self.assertIn("eu.iot.dreame.tech", urls)
|
|
self.assertIn("https://app.mt.eu.iot.dreame.tech", urls)
|
|
self.assertTrue(any("40.101.70.178" in u for u in urls))
|
|
|
|
def test_hexish_payload_decoder(self):
|
|
from analyzer.extract_tshark import _bytes_from_hexish
|
|
# pyshark commonly exposes payload bytes as colon-separated hex.
|
|
self.assertEqual(_bytes_from_hexish("47:45:54"), "GET")
|
|
self.assertEqual(_bytes_from_hexish("already text"), "already text")
|
|
|
|
|
|
class HashCrackTests(unittest.TestCase):
|
|
def _runner_factory(self, show_stdout):
|
|
import types
|
|
calls = {"n": 0}
|
|
def runner(cmd, **kw):
|
|
calls["n"] += 1
|
|
# The second invocation reads the cracked result.
|
|
out = show_stdout if "--show" in cmd else ""
|
|
return types.SimpleNamespace(stdout=out, stderr="", returncode=0)
|
|
return runner, calls
|
|
|
|
def test_command_builders_and_parser(self):
|
|
from analyzer.enrich.hash_crack import build_crack_command, build_show_command, parse_show_output
|
|
cmd = build_crack_command("hashcat", 5600, "/tmp/h.txt", ["/wl/rockyou.txt"], "/tmp/pot")
|
|
self.assertEqual(cmd[:3], ["hashcat", "-m", "5600"])
|
|
self.assertIn("--potfile-path", cmd)
|
|
self.assertIn("--show", build_show_command("hashcat", 0, "/tmp/h.txt", "/tmp/pot"))
|
|
self.assertEqual(parse_show_output("abc123:hunter2"), "hunter2")
|
|
self.assertIsNone(parse_show_output(""))
|
|
|
|
def test_crack_success(self):
|
|
from analyzer.artifacts import HashArtifact
|
|
from analyzer.enrich.hash_crack import DEFAULT_WORDLISTS, crack_hash
|
|
runner, _ = self._runner_factory(show_stdout="deadbeef" * 8 + ":hunter2\n")
|
|
art = HashArtifact(value="d" * 64, hashcat_mode=1400, hash_format="sha256")
|
|
res = crack_hash(art, wordlists=DEFAULT_WORDLISTS, hashcat_path="hashcat",
|
|
runner=runner, which=lambda p: "/usr/bin/hashcat")
|
|
self.assertTrue(res.cracked)
|
|
self.assertEqual(res.plaintext, "hunter2")
|
|
self.assertEqual(res.state, "cracked")
|
|
|
|
def test_crack_not_found(self):
|
|
from analyzer.artifacts import HashArtifact
|
|
from analyzer.enrich.hash_crack import DEFAULT_WORDLISTS, crack_hash
|
|
runner, _ = self._runner_factory(show_stdout="")
|
|
art = HashArtifact(value="d" * 64, hashcat_mode=1400)
|
|
res = crack_hash(art, wordlists=DEFAULT_WORDLISTS, runner=runner,
|
|
which=lambda p: "/usr/bin/hashcat")
|
|
self.assertFalse(res.cracked)
|
|
self.assertEqual(res.state, "not_cracked")
|
|
|
|
def test_crack_unsupported_mode(self):
|
|
from analyzer.enrich.hash_crack import crack_hash
|
|
res = crack_hash("not-a-hash", which=lambda p: "/usr/bin/hashcat")
|
|
self.assertEqual(res.state, "unsupported")
|
|
|
|
def test_crack_hashcat_missing(self):
|
|
from analyzer.artifacts import HashArtifact
|
|
from analyzer.enrich.hash_crack import crack_hash
|
|
art = HashArtifact(value="d" * 64, hashcat_mode=1400)
|
|
res = crack_hash(art, wordlists=["/etc/hostname"], which=lambda p: None)
|
|
self.assertEqual(res.state, "error")
|
|
|
|
|
|
class UrlProbeTests(unittest.TestCase):
|
|
def test_host_of(self):
|
|
from analyzer.enrich.url_probe import host_of
|
|
self.assertEqual(host_of("https://api.example.com/x?y=1"), "api.example.com")
|
|
self.assertEqual(host_of("example.net:8080/path"), "example.net")
|
|
self.assertEqual(host_of("plain.host"), "plain.host")
|
|
|
|
def test_passive_returns_observed(self):
|
|
from analyzer.artifacts import UrlArtifact
|
|
from analyzer.enrich.url_probe import probe_url
|
|
art = UrlArtifact(value="api.example.com", kind="sni",
|
|
observed_ports={443}, observed_tls={"TLS 1.2"})
|
|
probe = probe_url(art, active=False)
|
|
self.assertFalse(probe.active)
|
|
self.assertEqual(probe.open_ports, [443])
|
|
self.assertEqual(probe.tls_versions, ["TLS 1.2"])
|
|
|
|
def test_active_merges_injected_scanners(self):
|
|
from analyzer.artifacts import UrlArtifact
|
|
from analyzer.enrich.url_probe import probe_url
|
|
art = UrlArtifact(value="api.example.com", observed_ports={443})
|
|
probe = probe_url(
|
|
art, active=True,
|
|
dns_resolver=lambda h: ["93.184.216.34"],
|
|
port_scanner=lambda h, ports: [80, 443],
|
|
tls_scanner=lambda h: {"tls_version": "TLS 1.3", "cipher": "TLS_AES_128_GCM_SHA256",
|
|
"certificate": "Present"},
|
|
)
|
|
self.assertTrue(probe.active)
|
|
self.assertEqual(probe.resolved_ips, ["93.184.216.34"])
|
|
self.assertEqual(probe.open_ports, [80, 443])
|
|
self.assertIn("TLS 1.3", probe.tls_versions)
|
|
self.assertEqual(probe.certificate, "Present")
|
|
|
|
|
|
class EnrichmentEngineTests(unittest.TestCase):
|
|
def test_engine_progress_and_results(self):
|
|
from analyzer.artifacts import HashArtifact
|
|
from analyzer.enrich import EnrichmentEngine
|
|
seen = []
|
|
arts = [HashArtifact(value="d" * 64, hashcat_mode=1400),
|
|
HashArtifact(value="nope", hashcat_mode=None)]
|
|
res = EnrichmentEngine().crack_hashes(
|
|
arts, progress=lambda i, t, item: seen.append((i, t)),
|
|
which=lambda p: None) # Avoid a real subprocess in this unit test.
|
|
self.assertEqual(len(res), 2)
|
|
self.assertEqual(seen, [(1, 2), (2, 2)])
|
|
self.assertEqual(res[1].state, "unsupported")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|