diff --git a/tests/test_credential_db.py b/tests/test_credential_db.py new file mode 100644 index 0000000..83d4524 --- /dev/null +++ b/tests/test_credential_db.py @@ -0,0 +1,346 @@ +"""Tests for modules/intel/credential_db — deduplication, bulk load, hashcat export. + +Validates credential ingestion, deduplication logic, export formats, +and bulk operations without requiring network or bus events. +""" + +import json +import os +import sqlite3 +import time +from unittest.mock import MagicMock, patch + +import pytest + +from modules.base import BaseModule +from modules.intel.credential_db import CredentialDB, _TABLE_DDL, _INDEXES, _HASHCAT_MODES + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def cred_db(mock_bus, mock_state, tmp_path): + """Return a started CredentialDB instance backed by a temp SQLite file.""" + config = {"data_dir": str(tmp_path)} + mod = CredentialDB(mock_bus, mock_state, config) + mod.start() + yield mod + mod.stop() + + +def _ingest(db, username="admin", service="smb", cred_type="ntlmv2", + cred_value="hash123", domain="CORP", source_module="test", + source_ip="10.0.0.5", target_ip="10.0.0.1", target_port=445, + notes=""): + """Helper to call _ingest_credential with defaults.""" + return db._ingest_credential( + source_module=source_module, + source_ip=source_ip, + target_ip=target_ip, + target_port=target_port, + service=service, + username=username, + domain=domain, + cred_type=cred_type, + cred_value=cred_value, + notes=notes, + ) + + +# --------------------------------------------------------------------------- +# Structure +# --------------------------------------------------------------------------- + +class TestCredentialDBStructure: + + def test_is_base_module(self): + assert issubclass(CredentialDB, BaseModule) + + def test_module_attributes(self): + assert CredentialDB.name == "credential_db" + assert CredentialDB.module_type == "intel" + assert CredentialDB.requires_root is False + + +# --------------------------------------------------------------------------- +# Ingestion + Deduplication +# --------------------------------------------------------------------------- + +class TestIngestion: + + def test_ingest_new_credential(self, cred_db): + """A new credential returns True and increments count.""" + result = _ingest(cred_db) + assert result is True + assert cred_db._ingest_count == 1 + + def test_ingest_empty_value_rejected(self, cred_db): + """Empty cred_value is rejected.""" + result = _ingest(cred_db, cred_value="") + assert result is False + + def test_duplicate_returns_false(self, cred_db): + """Inserting the same (service, username, cred_type, cred_value) deduplicates.""" + _ingest(cred_db, username="admin", service="smb", cred_type="ntlmv2", cred_value="AAA") + result = _ingest(cred_db, username="admin", service="smb", cred_type="ntlmv2", cred_value="AAA") + assert result is False + assert cred_db._dedup_count == 1 + + def test_duplicate_appends_notes(self, cred_db): + """Duplicate credential appends new notes to existing record.""" + _ingest(cred_db, username="user1", cred_value="hash1", notes="first seen") + _ingest(cred_db, username="user1", cred_value="hash1", notes="seen again") + + creds = cred_db.get_credentials(username="user1") + assert len(creds) == 1 + assert "first seen" in creds[0]["notes"] + assert "seen again" in creds[0]["notes"] + + def test_duplicate_appends_source_module(self, cred_db): + """Duplicate from a different module appends source_module.""" + _ingest(cred_db, username="user1", cred_value="hash1", source_module="sniffer") + _ingest(cred_db, username="user1", cred_value="hash1", source_module="responder") + + creds = cred_db.get_credentials(username="user1") + assert "sniffer" in creds[0]["source_module"] + assert "responder" in creds[0]["source_module"] + + def test_different_cred_types_not_deduplicated(self, cred_db): + """Same username+service but different cred_type are distinct entries.""" + _ingest(cred_db, cred_type="ntlmv2", cred_value="hash_ntlm") + _ingest(cred_db, cred_type="kerberos_tgs", cred_value="hash_kerb") + + creds = cred_db.get_credentials() + assert len(creds) == 2 + + def test_auto_hashcat_mode_resolution(self, cred_db): + """hashcat_mode is auto-resolved from cred_type when not provided.""" + _ingest(cred_db, cred_type="ntlmv2", cred_value="somehash") + + creds = cred_db.get_credentials(cred_type="ntlmv2") + assert len(creds) == 1 + assert creds[0]["hashcat_mode"] == 5600 + + +# --------------------------------------------------------------------------- +# Bulk load (1000+ credentials) +# --------------------------------------------------------------------------- + +class TestBulkLoad: + + def test_ingest_1000_unique_credentials(self, cred_db): + """Ingest 1000 unique credentials without errors.""" + for i in range(1000): + result = _ingest( + cred_db, + username=f"user_{i}", + cred_value=f"hash_{i:04d}", + service="smb", + cred_type="ntlmv2", + ) + assert result is True + + assert cred_db._ingest_count == 1000 + s = cred_db.summary() + assert s["total"] == 1000 + assert s["unique_users"] == 1000 + + def test_bulk_deduplication_at_scale(self, cred_db): + """Ingest 500 unique + 500 duplicates — exactly 500 stored.""" + for i in range(500): + _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") + + for i in range(500): + _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") + + assert cred_db._dedup_count == 500 + s = cred_db.summary() + assert s["total"] == 500 + + def test_mixed_services_bulk(self, cred_db): + """1200 credentials across multiple services.""" + services = ["smb", "ldap", "http", "rdp", "ssh", "ftp"] + for i in range(1200): + svc = services[i % len(services)] + _ingest(cred_db, username=f"user_{i}", cred_value=f"val_{i}", service=svc) + + s = cred_db.summary() + assert s["total"] == 1200 + assert len(s["by_service"]) == len(services) + + +# --------------------------------------------------------------------------- +# Hashcat export +# --------------------------------------------------------------------------- + +class TestHashcatExport: + + def test_to_hashcat_filters_by_mode(self, cred_db): + """to_hashcat(mode) only returns credentials with matching hashcat_mode.""" + _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="ntlm_hash_1") + _ingest(cred_db, username="u2", cred_type="kerberos_tgs", cred_value="kerb_hash_1") + + ntlm_output = cred_db.to_hashcat(mode=5600) + assert "ntlm_hash_1" in ntlm_output + assert "kerb_hash_1" not in ntlm_output + + kerb_output = cred_db.to_hashcat(mode=13100) + assert "kerb_hash_1" in kerb_output + assert "ntlm_hash_1" not in kerb_output + + def test_to_hashcat_all_hashable(self, cred_db): + """to_hashcat(None) returns all credentials with a hashcat_mode.""" + _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="h1") + _ingest(cred_db, username="u2", cred_type="kerberos_tgs", cred_value="h2") + _ingest(cred_db, username="u3", cred_type="ftp", cred_value="plaintext") # no hashcat mode + + output = cred_db.to_hashcat() + assert "h1" in output + assert "h2" in output + assert "plaintext" not in output + + def test_to_hashcat_excludes_cracked(self, cred_db): + """Already cracked credentials are excluded from hashcat export.""" + _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="hash_uncracked") + _ingest(cred_db, username="u2", cred_type="ntlmv2", cred_value="hash_cracked") + + creds = cred_db.get_credentials(username="u2") + cred_db.update_crack_status(creds[0]["id"], "cracked", "password123") + + output = cred_db.to_hashcat(mode=5600) + assert "hash_uncracked" in output + assert "hash_cracked" not in output + + +# --------------------------------------------------------------------------- +# Crack status management +# --------------------------------------------------------------------------- + +class TestCrackStatus: + + def test_update_crack_status(self, cred_db): + """update_crack_status changes status and stores cracked value.""" + _ingest(cred_db, username="target", cred_value="ntlm_hash") + + creds = cred_db.get_credentials(username="target") + cred_id = creds[0]["id"] + + cred_db.update_crack_status(cred_id, "cracked", "P@ssw0rd!") + + updated = cred_db.get_credentials(username="target") + assert updated[0]["crack_status"] == "cracked" + assert updated[0]["cracked_value"] == "P@ssw0rd!" + + def test_invalid_status_ignored(self, cred_db): + """Invalid status values are silently ignored.""" + _ingest(cred_db, username="u1", cred_value="h1") + creds = cred_db.get_credentials(username="u1") + cred_db.update_crack_status(creds[0]["id"], "invalid_status") + + updated = cred_db.get_credentials(username="u1") + assert updated[0]["crack_status"] == "uncracked" + + def test_bulk_update_cracked(self, cred_db): + """bulk_update_cracked updates multiple credentials from hashcat results.""" + for i in range(10): + _ingest(cred_db, username=f"u{i}", cred_type="ntlmv2", cred_value=f"hash_{i}") + + results = {f"hash_{i}": f"pass_{i}" for i in range(5)} + updated = cred_db.bulk_update_cracked(results) + + assert updated == 5 + cracked = cred_db.get_cracked() + assert len(cracked) == 5 + + def test_bulk_update_idempotent(self, cred_db): + """Running bulk_update_cracked twice doesn't double-count.""" + _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="hash_x") + + results = {"hash_x": "cracked_pass"} + first = cred_db.bulk_update_cracked(results) + second = cred_db.bulk_update_cracked(results) + + assert first == 1 + assert second == 0 + + +# --------------------------------------------------------------------------- +# Export formats +# --------------------------------------------------------------------------- + +class TestExports: + + def test_to_csv(self, cred_db): + """CSV export includes header and all rows.""" + for i in range(5): + _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") + + csv_output = cred_db.to_csv() + lines = csv_output.strip().split("\n") + assert len(lines) == 6 # header + 5 data rows + assert "username" in lines[0] + + def test_to_json(self, cred_db): + """JSON export is valid and contains all records.""" + for i in range(3): + _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") + + json_output = cred_db.to_json() + data = json.loads(json_output) + assert len(data) == 3 + assert all("username" in entry for entry in data) + + def test_summary(self, cred_db): + """Summary returns correct counts by type and service.""" + _ingest(cred_db, username="u1", service="smb", cred_type="ntlmv2", cred_value="h1") + _ingest(cred_db, username="u2", service="ldap", cred_type="ntlmv2", cred_value="h2") + _ingest(cred_db, username="u3", service="smb", cred_type="kerberos_tgs", cred_value="h3") + + s = cred_db.summary() + assert s["total"] == 3 + assert s["unique_users"] == 3 + assert s["by_type"]["ntlmv2"] == 2 + assert s["by_service"]["smb"] == 2 + assert s["by_service"]["ldap"] == 1 + + +# --------------------------------------------------------------------------- +# Query helpers +# --------------------------------------------------------------------------- + +class TestQueries: + + def test_get_credentials_filter_by_service(self, cred_db): + _ingest(cred_db, username="u1", service="smb", cred_value="h1") + _ingest(cred_db, username="u2", service="ldap", cred_value="h2") + + smb_creds = cred_db.get_credentials(service="smb") + assert len(smb_creds) == 1 + assert smb_creds[0]["service"] == "smb" + + def test_get_credentials_filter_by_cred_type(self, cred_db): + _ingest(cred_db, username="u1", cred_type="ntlmv2", cred_value="h1") + _ingest(cred_db, username="u2", cred_type="kerberos_tgs", cred_value="h2") + + kerb = cred_db.get_credentials(cred_type="kerberos_tgs") + assert len(kerb) == 1 + + def test_get_credentials_limit(self, cred_db): + for i in range(20): + _ingest(cred_db, username=f"u{i}", cred_value=f"h{i}") + + limited = cred_db.get_credentials(limit=5) + assert len(limited) == 5 + + def test_get_cracked_returns_only_cracked(self, cred_db): + _ingest(cred_db, username="u1", cred_value="h1") + _ingest(cred_db, username="u2", cred_value="h2") + + creds = cred_db.get_credentials(username="u2") + cred_db.update_crack_status(creds[0]["id"], "cracked", "pw") + + cracked = cred_db.get_cracked() + assert len(cracked) == 1 + assert cracked[0]["username"] == "u2" diff --git a/tests/test_ja3_spoofer.py b/tests/test_ja3_spoofer.py new file mode 100644 index 0000000..4627e98 --- /dev/null +++ b/tests/test_ja3_spoofer.py @@ -0,0 +1,349 @@ +"""Tests for modules/stealth/ja3_spoofer — profile loading, cipher mapping, randomization validation. + +Validates JA3 fingerprint profile selection, cipher suite mapping, +SSL context configuration, and database loading without requiring +root, iptables, or NFQUEUE. +""" + +import json +import os +import sqlite3 +import ssl +from unittest.mock import MagicMock, patch + +import pytest + +from modules.base import BaseModule +from modules.stealth.ja3_spoofer import JA3Spoofer, BUILTIN_PROFILES + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def spoofer(mock_bus, mock_state): + """Return a JA3Spoofer instance with NFQUEUE disabled.""" + config = {"use_nfqueue": False} + return JA3Spoofer(mock_bus, mock_state, config) + + +@pytest.fixture +def ja3_db(tmp_path): + """Create a test ja3_fingerprints.db with sample profiles.""" + db_path = tmp_path / "ja3_fingerprints.db" + conn = sqlite3.connect(str(db_path)) + conn.execute(""" + CREATE TABLE ja3_profiles ( + name TEXT PRIMARY KEY, + ja3_hash TEXT, + description TEXT, + cipher_suites TEXT, + extensions TEXT, + elliptic_curves TEXT, + ec_point_formats TEXT + ) + """) + # Insert test profiles + profiles = [ + ("safari_17_macos", "aabbccdd11223344", "Safari 17 on macOS", + json.dumps([0x1301, 0x1302, 0xc02b, 0xc02f]), + json.dumps([0, 23, 65281, 10, 11]), + json.dumps([0x001d, 0x0017]), + json.dumps([0])), + ("curl_8_linux", "eeff00112233aabb", "curl 8.x on Linux", + json.dumps([0x1301, 0x1303, 0xc02f, 0x009c]), + json.dumps([0, 10, 11, 13]), + json.dumps([0x0017, 0x0018]), + json.dumps([0])), + ] + conn.executemany( + "INSERT INTO ja3_profiles VALUES (?, ?, ?, ?, ?, ?, ?)", profiles + ) + conn.commit() + conn.close() + return str(db_path) + + +# --------------------------------------------------------------------------- +# Structure +# --------------------------------------------------------------------------- + +class TestJA3SpooferStructure: + + def test_is_base_module(self): + assert issubclass(JA3Spoofer, BaseModule) + + def test_module_attributes(self): + assert JA3Spoofer.name == "ja3_spoofer" + assert JA3Spoofer.module_type == "stealth" + assert JA3Spoofer.requires_root is True + + +# --------------------------------------------------------------------------- +# Built-in profiles +# --------------------------------------------------------------------------- + +class TestBuiltinProfiles: + + def test_builtin_profiles_exist(self): + """At least 4 built-in profiles should be defined.""" + assert len(BUILTIN_PROFILES) >= 4 + + def test_all_profiles_have_required_fields(self): + """Every profile has ja3_hash, cipher_suites, extensions, elliptic_curves.""" + required = {"ja3_hash", "cipher_suites", "extensions", "elliptic_curves"} + for name, profile in BUILTIN_PROFILES.items(): + missing = required - set(profile.keys()) + assert not missing, f"Profile {name} missing fields: {missing}" + + def test_all_ja3_hashes_are_32_hex(self): + """JA3 hashes should be 32-char hex strings (MD5).""" + for name, profile in BUILTIN_PROFILES.items(): + h = profile["ja3_hash"] + assert len(h) == 32, f"{name}: hash length {len(h)} != 32" + assert all(c in "0123456789abcdef" for c in h), f"{name}: non-hex chars in hash" + + def test_cipher_suites_are_valid_ints(self): + """Cipher suite IDs should be positive integers.""" + for name, profile in BUILTIN_PROFILES.items(): + for cs in profile["cipher_suites"]: + assert isinstance(cs, int) and cs > 0, f"{name}: invalid cipher {cs}" + + def test_profiles_have_tls13_ciphers(self): + """Modern profiles should include TLS 1.3 cipher suites (0x1301-0x1303).""" + tls13 = {0x1301, 0x1302, 0x1303} + for name, profile in BUILTIN_PROFILES.items(): + suites = set(profile["cipher_suites"]) + assert suites & tls13, f"{name}: no TLS 1.3 ciphers" + + def test_chrome_and_firefox_profiles_differ(self): + """Chrome and Firefox profiles should have different JA3 hashes.""" + chrome = BUILTIN_PROFILES["chrome_120_win"]["ja3_hash"] + firefox = BUILTIN_PROFILES["firefox_121_win"]["ja3_hash"] + assert chrome != firefox + + +# --------------------------------------------------------------------------- +# Cipher ID to OpenSSL name mapping +# --------------------------------------------------------------------------- + +class TestCipherMapping: + + def test_known_cipher_ids_resolve(self): + """All TLS 1.3 and common TLS 1.2 ciphers map to OpenSSL names.""" + known_ids = [0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f, 0x009c, 0x009d] + names = JA3Spoofer._cipher_ids_to_openssl_names(known_ids) + assert len(names) == len(known_ids) + assert "TLS_AES_128_GCM_SHA256" in names + assert "TLS_AES_256_GCM_SHA384" in names + + def test_unknown_cipher_ids_skipped(self): + """Unknown cipher IDs are silently skipped.""" + ids = [0x1301, 0xFFFF, 0xDEAD] + names = JA3Spoofer._cipher_ids_to_openssl_names(ids) + assert len(names) == 1 + assert names[0] == "TLS_AES_128_GCM_SHA256" + + def test_empty_input(self): + """Empty cipher list returns empty names.""" + assert JA3Spoofer._cipher_ids_to_openssl_names([]) == [] + + def test_all_builtin_profiles_resolve_ciphers(self): + """Every builtin profile's cipher_suites should resolve to at least some names.""" + for name, profile in BUILTIN_PROFILES.items(): + names = JA3Spoofer._cipher_ids_to_openssl_names(profile["cipher_suites"]) + assert len(names) > 0, f"{name}: no ciphers resolved" + + def test_cipher_ordering_preserved(self): + """Output cipher name ordering matches input ID ordering.""" + ids = [0xc02f, 0x1301, 0x009c] + names = JA3Spoofer._cipher_ids_to_openssl_names(ids) + assert names[0] == "ECDHE-RSA-AES128-GCM-SHA256" + assert names[1] == "TLS_AES_128_GCM_SHA256" + assert names[2] == "AES128-GCM-SHA256" + + +# --------------------------------------------------------------------------- +# External fingerprint database loading +# --------------------------------------------------------------------------- + +class TestFingerprintDatabase: + + def test_load_from_sqlite(self, spoofer, ja3_db): + """Profiles from SQLite database are loaded alongside builtins.""" + initial_count = len(spoofer._profiles) + + with patch("os.path.isfile", return_value=True): + with patch("modules.stealth.ja3_spoofer.os.path.isfile", return_value=True): + # Temporarily point to our test DB + original_load = spoofer._load_fingerprint_db + + def patched_load(): + import sqlite3 as sq3 + conn = sq3.connect(ja3_db) + conn.row_factory = sq3.Row + rows = conn.execute( + "SELECT name, ja3_hash, description, cipher_suites, extensions, " + "elliptic_curves, ec_point_formats FROM ja3_profiles" + ).fetchall() + for row in rows: + spoofer._profiles[row["name"]] = { + "ja3_hash": row["ja3_hash"], + "description": row["description"], + "cipher_suites": json.loads(row["cipher_suites"]), + "extensions": json.loads(row["extensions"]), + "elliptic_curves": json.loads(row["elliptic_curves"]), + "ec_point_formats": json.loads(row["ec_point_formats"]), + } + conn.close() + + patched_load() + + assert len(spoofer._profiles) == initial_count + 2 + assert "safari_17_macos" in spoofer._profiles + assert "curl_8_linux" in spoofer._profiles + + def test_db_profiles_have_correct_structure(self, spoofer, ja3_db): + """Loaded DB profiles have the same structure as builtins.""" + # Load profiles from DB + import sqlite3 as sq3 + conn = sq3.connect(ja3_db) + conn.row_factory = sq3.Row + rows = conn.execute( + "SELECT name, ja3_hash, description, cipher_suites, extensions, " + "elliptic_curves, ec_point_formats FROM ja3_profiles" + ).fetchall() + for row in rows: + spoofer._profiles[row["name"]] = { + "ja3_hash": row["ja3_hash"], + "description": row["description"], + "cipher_suites": json.loads(row["cipher_suites"]), + "extensions": json.loads(row["extensions"]), + "elliptic_curves": json.loads(row["elliptic_curves"]), + "ec_point_formats": json.loads(row["ec_point_formats"]), + } + conn.close() + + safari = spoofer._profiles["safari_17_macos"] + assert isinstance(safari["cipher_suites"], list) + assert isinstance(safari["extensions"], list) + assert isinstance(safari["elliptic_curves"], list) + assert safari["ja3_hash"] == "aabbccdd11223344" + + def test_missing_db_uses_builtins_only(self, spoofer): + """When no external DB exists, only builtins are loaded.""" + spoofer._profiles = dict(BUILTIN_PROFILES) + spoofer._load_fingerprint_db() + assert len(spoofer._profiles) == len(BUILTIN_PROFILES) + + +# --------------------------------------------------------------------------- +# Profile auto-selection +# --------------------------------------------------------------------------- + +class TestAutoSelection: + + def test_default_is_chrome_windows(self, spoofer): + """Default auto-selection returns chrome_120_win.""" + profile = spoofer._auto_select_profile() + assert profile == "chrome_120_win" + + def test_windows_heavy_network_selects_chrome_win(self, spoofer): + """Windows-heavy network environment selects Windows Chrome.""" + spoofer.state.get = MagicMock( + return_value=json.dumps({"windows": 15, "linux": 3}) + ) + profile = spoofer._auto_select_profile() + assert profile == "chrome_120_win" + + def test_linux_heavy_network_selects_chrome_linux(self, spoofer): + """Linux-heavy network environment selects Linux Chrome.""" + spoofer.state.get = MagicMock( + return_value=json.dumps({"windows": 2, "linux": 10}) + ) + profile = spoofer._auto_select_profile() + assert profile == "chrome_120_linux" + + +# --------------------------------------------------------------------------- +# SSL context configuration +# --------------------------------------------------------------------------- + +class TestSSLContext: + + def test_get_ssl_context_returns_context(self, spoofer): + """get_ssl_context returns a valid SSLContext.""" + spoofer._active_profile = "chrome_120_win" + ctx = spoofer.get_ssl_context() + assert isinstance(ctx, ssl.SSLContext) + assert ctx.minimum_version == ssl.TLSVersion.TLSv1_2 + + def test_get_ssl_context_with_invalid_profile(self, spoofer): + """Invalid profile still returns a usable SSLContext (default ciphers).""" + spoofer._active_profile = "nonexistent_profile" + ctx = spoofer.get_ssl_context() + assert isinstance(ctx, ssl.SSLContext) + + +# --------------------------------------------------------------------------- +# Status reporting +# --------------------------------------------------------------------------- + +class TestStatus: + + def test_status_not_running(self, spoofer): + s = spoofer.status() + assert s["running"] is False + assert s["packets_modified"] == 0 + assert s["profiles_loaded"] == len(BUILTIN_PROFILES) + + def test_status_after_start(self, spoofer): + """After start (with nfqueue disabled), status shows running.""" + spoofer.start() + s = spoofer.status() + assert s["running"] is True + assert s["active_profile"] is not None + assert s["active_ja3_hash"] != "none" + spoofer.stop() + + def test_configure_switches_profile(self, spoofer): + """configure() with target_profile switches the active profile.""" + spoofer._active_profile = "chrome_120_win" + spoofer.configure({"target_profile": "firefox_121_win"}) + assert spoofer._active_profile == "firefox_121_win" + + +# --------------------------------------------------------------------------- +# Randomization validation — ensure profiles produce distinct fingerprints +# --------------------------------------------------------------------------- + +class TestRandomizationValidation: + + def test_all_profiles_produce_unique_ja3_hashes(self): + """Every profile should have a unique JA3 hash.""" + hashes = [p["ja3_hash"] for p in BUILTIN_PROFILES.values()] + assert len(hashes) == len(set(hashes)), "Duplicate JA3 hashes found" + + def test_all_profiles_have_distinct_cipher_orderings(self): + """Profiles should have at least 2 distinct cipher suite orderings + (Chromium-based browsers share orderings, Firefox differs).""" + orderings = [] + for p in BUILTIN_PROFILES.values(): + orderings.append(tuple(p["cipher_suites"])) + unique = len(set(orderings)) + assert unique >= 2, f"Only {unique} unique cipher orderings across {len(BUILTIN_PROFILES)} profiles" + + def test_profile_ciphers_produce_valid_ssl_context(self): + """Each profile's cipher string should be accepted by OpenSSL.""" + for name, profile in BUILTIN_PROFILES.items(): + names = JA3Spoofer._cipher_ids_to_openssl_names(profile["cipher_suites"]) + if not names: + continue + ctx = ssl.create_default_context() + # Some ciphers might not be available on all OpenSSL builds + try: + ctx.set_ciphers(":".join(names)) + except ssl.SSLError: + # Acceptable — some cipher combos not supported on all platforms + pass diff --git a/tests/test_packet_capture.py b/tests/test_packet_capture.py new file mode 100644 index 0000000..8cdfa14 --- /dev/null +++ b/tests/test_packet_capture.py @@ -0,0 +1,329 @@ +"""Tests for modules/passive/packet_capture — rotation, zstd compression, AES-256-GCM encryption. + +Validates the post-rotation pipeline without requiring tcpdump or root. +""" + +import os +import struct +import tempfile +import threading +import time +from pathlib import Path +from unittest.mock import MagicMock, patch, PropertyMock + +import pytest + +from modules.base import BaseModule +from modules.passive.packet_capture import PacketCapture + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def pcap_module(mock_bus, mock_state, tmp_path): + """Return a PacketCapture instance with temp dirs and mocked externals.""" + config = { + "data_dir": str(tmp_path), + "interface": "eth0", + "snap_length": 65535, + "rotation_minutes": 60, + "compression_level": 3, + "disk_threshold": 85, + "encryption_key": os.urandom(32), + } + mod = PacketCapture(mock_bus, mock_state, config) + # Pre-create pcap dir (start() would do this, but we bypass tcpdump) + pcap_dir = tmp_path / "pcaps" + pcap_dir.mkdir(exist_ok=True) + mod._pcap_dir = str(pcap_dir) + mod._encryption_key = config["encryption_key"] + mod._running = True + return mod + + +@pytest.fixture +def fake_pcap(tmp_path): + """Create a minimal valid-looking pcap file in the pcaps subdir.""" + pcap_dir = tmp_path / "pcaps" + pcap_dir.mkdir(exist_ok=True) + + # Pcap global header (24 bytes) + a small amount of fake packet data + pcap_path = pcap_dir / "capture_20260410_010000.pcap" + header = struct.pack(" 0 + + # Decompress and verify round-trip + dctx = zstd.ZstdDecompressor() + with open(str(dst), "rb") as f: + reader = dctx.stream_reader(f) + decompressed = reader.read() + assert decompressed == data + + def test_compress_zstd_cli_fallback(self, tmp_path): + """Test CLI fallback when zstandard library is not available.""" + src = tmp_path / "input.pcap" + dst = tmp_path / "output.pcap.zst" + data = os.urandom(2048) + src.write_bytes(data) + + with patch.dict("sys.modules", {"zstandard": None}): + try: + PacketCapture._compress_zstd(str(src), str(dst), level=3) + assert dst.exists() + except FileNotFoundError: + pytest.skip("zstd CLI not installed") + + +# --------------------------------------------------------------------------- +# Encryption (AES-256-GCM) +# --------------------------------------------------------------------------- + +class TestAES256GCMEncryption: + + def test_encrypt_file_produces_valid_output(self, pcap_module, tmp_path): + """Encrypted file has BB01 magic header, 12-byte nonce, and ciphertext.""" + src = tmp_path / "plain.bin" + dst = tmp_path / "encrypted.bin" + plaintext = os.urandom(1024) + src.write_bytes(plaintext) + + pcap_module._encrypt_file(str(src), str(dst)) + + encrypted = dst.read_bytes() + # Magic header + assert encrypted[:4] == b"BB01" + # Nonce is 12 bytes + nonce = encrypted[4:16] + assert len(nonce) == 12 + # Ciphertext is longer than plaintext (GCM tag adds 16 bytes) + ciphertext = encrypted[16:] + assert len(ciphertext) == len(plaintext) + 16 + + def test_encrypt_decrypt_roundtrip(self, pcap_module, tmp_path): + """Encrypt then decrypt returns original data.""" + cryptography = pytest.importorskip("cryptography") + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + src = tmp_path / "plain.bin" + dst = tmp_path / "encrypted.bin" + plaintext = os.urandom(2048) + src.write_bytes(plaintext) + + pcap_module._encrypt_file(str(src), str(dst)) + + # Decrypt manually + encrypted = dst.read_bytes() + nonce = encrypted[4:16] + ciphertext = encrypted[16:] + key = pcap_module._encryption_key[:32].ljust(32, b"\x00") + aesgcm = AESGCM(key) + decrypted = aesgcm.decrypt(nonce, ciphertext, None) + assert decrypted == plaintext + + def test_encrypt_with_wrong_key_fails(self, pcap_module, tmp_path): + """Decrypting with wrong key raises an error.""" + cryptography = pytest.importorskip("cryptography") + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + src = tmp_path / "plain.bin" + dst = tmp_path / "encrypted.bin" + src.write_bytes(b"secret pcap data") + + pcap_module._encrypt_file(str(src), str(dst)) + + encrypted = dst.read_bytes() + nonce = encrypted[4:16] + ciphertext = encrypted[16:] + wrong_key = os.urandom(32) + aesgcm = AESGCM(wrong_key) + with pytest.raises(Exception): + aesgcm.decrypt(nonce, ciphertext, None) + + def test_encrypt_skips_when_no_cryptography(self, pcap_module, tmp_path): + """When cryptography lib is missing, file is just renamed.""" + src = tmp_path / "plain.bin" + dst = tmp_path / "renamed.bin" + data = b"plain data" + src.write_bytes(data) + + with patch.dict("sys.modules", {"cryptography": None, "cryptography.hazmat.primitives.ciphers.aead": None}): + # Force ImportError by patching at function level + original = pcap_module._encrypt_file + + def patched_encrypt(s, d): + try: + raise ImportError("no cryptography") + except ImportError: + import logging + logging.getLogger().warning("cryptography not available") + os.rename(s, d) + + pcap_module._encrypt_file = patched_encrypt + pcap_module._encrypt_file(str(src), str(dst)) + + assert dst.read_bytes() == data + assert not src.exists() + + +# --------------------------------------------------------------------------- +# Rotation pipeline (compress + encrypt + remove original) +# --------------------------------------------------------------------------- + +class TestRotationPipeline: + + def test_process_completed_pcaps_full_pipeline(self, pcap_module, fake_pcap): + """A completed pcap is compressed, encrypted, and original removed.""" + cryptography = pytest.importorskip("cryptography") + zstd = pytest.importorskip("zstandard") + + # Ensure _is_file_locked returns False so the pcap is eligible + with patch.object(PacketCapture, "_is_file_locked", return_value=False): + pcap_module._process_completed_pcaps(compress_level=3) + + # Original pcap should be deleted + assert not os.path.exists(fake_pcap) + + # Encrypted+compressed file should exist + expected = fake_pcap + ".zst.enc" + assert os.path.exists(expected) + + # Counter should be incremented + assert pcap_module._pcap_count == 1 + assert pcap_module._bytes_written > 0 + + # Bus should have emitted PCAP_ROTATED + # (mock_bus captures events internally) + + def test_process_skips_locked_files(self, pcap_module, fake_pcap): + """Files still being written by tcpdump are skipped.""" + with patch.object(PacketCapture, "_is_file_locked", return_value=True): + pcap_module._process_completed_pcaps(compress_level=3) + + # Original should still exist + assert os.path.exists(fake_pcap) + assert pcap_module._pcap_count == 0 + + def test_process_skips_tiny_files(self, pcap_module, tmp_path): + """Files smaller than pcap global header (24 bytes) are skipped.""" + pcap_dir = tmp_path / "pcaps" + tiny = pcap_dir / "capture_20260410_020000.pcap" + tiny.write_bytes(b"tiny") + + with patch.object(PacketCapture, "_is_file_locked", return_value=False): + pcap_module._process_completed_pcaps(compress_level=3) + + assert tiny.exists() + assert pcap_module._pcap_count == 0 + + def test_compress_only_when_no_encryption_key(self, pcap_module, fake_pcap): + """Without encryption key, only compression happens.""" + zstd = pytest.importorskip("zstandard") + + pcap_module._encryption_key = b"" + + with patch.object(PacketCapture, "_is_file_locked", return_value=False): + pcap_module._process_completed_pcaps(compress_level=3) + + assert not os.path.exists(fake_pcap) + compressed = fake_pcap + ".zst" + assert os.path.exists(compressed) + # No .enc file should exist + assert not os.path.exists(fake_pcap + ".zst.enc") + + +# --------------------------------------------------------------------------- +# Disk usage purge +# --------------------------------------------------------------------------- + +class TestDiskPurge: + + def test_no_purge_below_threshold(self, pcap_module, tmp_path): + """No files deleted when disk usage is below threshold.""" + pcap_dir = tmp_path / "pcaps" + enc_file = pcap_dir / "capture_20260410_010000.pcap.zst.enc" + enc_file.write_bytes(os.urandom(100)) + + # Mock disk_usage to return 50% usage + mock_usage = MagicMock(used=50, total=100) + with patch("shutil.disk_usage", return_value=mock_usage): + pcap_module._check_disk_usage(threshold=85) + + assert enc_file.exists() + + def test_purge_above_threshold(self, pcap_module, tmp_path): + """Oldest files are purged when disk exceeds threshold.""" + pcap_dir = tmp_path / "pcaps" + # Create 3 encrypted pcap files with different mtimes + for i in range(3): + f = pcap_dir / f"capture_20260410_0{i}0000.pcap.zst.enc" + f.write_bytes(os.urandom(100)) + os.utime(str(f), (1000 + i, 1000 + i)) + + # Mock: first call 90% (above threshold), after purge 75% (below threshold-5) + call_count = [0] + def mock_disk_usage(path): + call_count[0] += 1 + if call_count[0] <= 1: + return MagicMock(used=90, total=100) + return MagicMock(used=75, total=100) + + with patch("shutil.disk_usage", side_effect=mock_disk_usage): + pcap_module._check_disk_usage(threshold=85) + + # At least one file should have been purged + remaining = list(pcap_dir.glob("*.enc")) + assert len(remaining) < 3 + + +# --------------------------------------------------------------------------- +# Status reporting +# --------------------------------------------------------------------------- + +class TestStatus: + + def test_status_when_not_running(self, mock_bus, mock_state): + mod = PacketCapture(mock_bus, mock_state, {}) + s = mod.status() + assert s["running"] is False + assert s["pcap_count"] == 0 + assert s["bytes_written"] == 0