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

Initial public release

Browse Source 
Full BigBrother network implant - passive SOC + active exploitation.
Personal identifiers removed; all capabilities intact.
See README.md for setup and docs/deployment.md for detailed deployment.
This commit is contained in:
n0mad1k
2026-06-26 09:52:50 -04:00
commit ccc6b729de
175 changed files with 47941 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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