n0mad1k
4ca9d3cdf1
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.
330 lines
12 KiB
Python
330 lines
12 KiB
Python
"""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
|