n0mad1k/bigbrother
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add validation tests for packet_capture, credential_db, and ja3_spoofer

Browse Source 
Tests cover:
- packet_capture: zstd compression, AES-256-GCM encryption, rotation
  pipeline, disk purge, file locking (15 tests)
- credential_db: ingestion, deduplication at scale (1000+), hashcat
  export filtering, crack status management, bulk update, CSV/JSON
  export, query helpers (25 tests)
- ja3_spoofer: builtin profile validation, cipher ID mapping, external
  DB loading, auto-selection, SSL context config, randomization
  validation (28 tests)

Addresses DevTrack items #424, #425, #437.
This commit is contained in:
n0mad1k
2026-04-10 07:29:08 -04:00
parent 027eaa39b2
commit 4ca9d3cdf1
3 changed files with 1024 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_credential_db.py
+346
View File
@@ -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"
tests/test_ja3_spoofer.py
+349
View File
@@ -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
tests/test_packet_capture.py
+329
View File
@@ -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("<IHHiIII", 0xa1b2c3d4, 2, 4, 0, 0, 65535, 1)
fake_pkt = os.urandom(200)
pcap_path.write_bytes(header + fake_pkt)
return str(pcap_path)
# ---------------------------------------------------------------------------
# Structure
# ---------------------------------------------------------------------------
class TestPacketCaptureStructure:
def test_is_base_module(self):
assert issubclass(PacketCapture, BaseModule)
def test_module_attributes(self):
assert PacketCapture.name == "packet_capture"
assert PacketCapture.module_type == "passive"
assert PacketCapture.requires_root is True
# ---------------------------------------------------------------------------
# Compression (zstd)
# ---------------------------------------------------------------------------
class TestZstdCompression:
def test_compress_zstd_python_library(self, tmp_path):
"""Test compression via the zstandard Python library."""
zstd = pytest.importorskip("zstandard")
src = tmp_path / "input.pcap"
dst = tmp_path / "output.pcap.zst"
data = os.urandom(4096)
src.write_bytes(data)
PacketCapture._compress_zstd(str(src), str(dst), level=3)
assert dst.exists()
assert dst.stat().st_size > 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