n0mad1k
3394c72814
- bigbrother.py: Click CLI with start/stop/status/activate/deactivate/ kill/config/selftest/modules commands plus Rich interactive menu. Module discovery scans modules/stealth/ for BaseModule subclasses. Supports --daemon mode with PID file and --passive-only flag. - tests/conftest.py: Shared fixtures (tmp_dir, mock_config, mock_bus, mock_state, hardware_tier_override, project_root) - tests/test_bus.py: 6 tests for EventBus pub/sub, filtering, wildcards - tests/test_engine.py: 5 tests for Engine lifecycle, deps, tier, phase - tests/test_crypto.py: 5 tests for AES-256-GCM, file encryption, KDF - tests/test_resource.py: 7 tests for hardware detection, memory, disk - tests/test_tool_manager.py: 6 tests for subprocess management - tests/test_stealth_modules.py: 4 parametrized test classes covering all 12 stealth modules (import, inheritance, attributes, instantiation)
135 lines
4.2 KiB
Python
135 lines
4.2 KiB
Python
"""Tests for core.bus — EventBus pub/sub system."""
|
|
|
|
import time
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from core.bus import EventBus, Event
|
|
|
|
|
|
class TestEventBus:
|
|
|
|
def test_subscribe_and_publish(self, mock_bus):
|
|
"""A subscriber receives an event that was published."""
|
|
received = []
|
|
|
|
def handler(event):
|
|
received.append(event)
|
|
|
|
mock_bus.subscribe(handler, event_type="HOST_DISCOVERED")
|
|
mock_bus.publish(Event(
|
|
event_type="HOST_DISCOVERED",
|
|
payload={"ip": "10.0.0.1"},
|
|
source_module="test",
|
|
))
|
|
|
|
# Wait for dispatch
|
|
deadline = time.time() + 3.0
|
|
while not received and time.time() < deadline:
|
|
time.sleep(0.05)
|
|
|
|
assert len(received) == 1
|
|
assert received[0].event_type == "HOST_DISCOVERED"
|
|
assert received[0].payload["ip"] == "10.0.0.1"
|
|
|
|
def test_publish_no_subscribers(self, mock_bus):
|
|
"""Publishing with no subscribers does not raise an error."""
|
|
mock_bus.publish(Event(
|
|
event_type="CREDENTIAL_FOUND",
|
|
payload={"user": "admin"},
|
|
source_module="test",
|
|
))
|
|
# Give the dispatcher a moment to process
|
|
time.sleep(0.2)
|
|
# No crash = pass
|
|
|
|
def test_multiple_subscribers(self, mock_bus):
|
|
"""Multiple subscribers for the same event type all receive the event."""
|
|
results_a = []
|
|
results_b = []
|
|
|
|
mock_bus.subscribe(lambda e: results_a.append(e), event_type="PCAP_ROTATED")
|
|
mock_bus.subscribe(lambda e: results_b.append(e), event_type="PCAP_ROTATED")
|
|
|
|
mock_bus.publish(Event(
|
|
event_type="PCAP_ROTATED",
|
|
payload={"file": "/tmp/test.pcap"},
|
|
source_module="test",
|
|
))
|
|
|
|
deadline = time.time() + 3.0
|
|
while (not results_a or not results_b) and time.time() < deadline:
|
|
time.sleep(0.05)
|
|
|
|
assert len(results_a) == 1
|
|
assert len(results_b) == 1
|
|
|
|
def test_event_type_filtering(self, mock_bus):
|
|
"""A subscriber only receives events matching its registered type."""
|
|
received = []
|
|
|
|
mock_bus.subscribe(lambda e: received.append(e), event_type="CREDENTIAL_FOUND")
|
|
|
|
# Publish a different event type
|
|
mock_bus.publish(Event(
|
|
event_type="HOST_DISCOVERED",
|
|
payload={"ip": "10.0.0.2"},
|
|
source_module="test",
|
|
))
|
|
|
|
# And the matching type
|
|
mock_bus.publish(Event(
|
|
event_type="CREDENTIAL_FOUND",
|
|
payload={"user": "root"},
|
|
source_module="test",
|
|
))
|
|
|
|
deadline = time.time() + 3.0
|
|
while len(received) < 1 and time.time() < deadline:
|
|
time.sleep(0.05)
|
|
|
|
# Allow a little extra time for any misdelivered events
|
|
time.sleep(0.2)
|
|
|
|
assert len(received) == 1
|
|
assert received[0].event_type == "CREDENTIAL_FOUND"
|
|
|
|
def test_wildcard_subscriber(self, mock_bus):
|
|
"""A wildcard subscriber (event_type=None) receives all events."""
|
|
received = []
|
|
|
|
mock_bus.subscribe(lambda e: received.append(e), event_type=None)
|
|
|
|
mock_bus.publish(Event(event_type="HOST_DISCOVERED", payload={}, source_module="test"))
|
|
mock_bus.publish(Event(event_type="CREDENTIAL_FOUND", payload={}, source_module="test"))
|
|
|
|
deadline = time.time() + 3.0
|
|
while len(received) < 2 and time.time() < deadline:
|
|
time.sleep(0.05)
|
|
|
|
assert len(received) == 2
|
|
types = {e.event_type for e in received}
|
|
assert "HOST_DISCOVERED" in types
|
|
assert "CREDENTIAL_FOUND" in types
|
|
|
|
def test_event_dataclass(self):
|
|
"""Event dataclass fields and to_dict() work correctly."""
|
|
before = time.time()
|
|
event = Event(
|
|
event_type="MODULE_STARTED",
|
|
payload={"module": "dns_logger", "pid": 1234},
|
|
source_module="engine",
|
|
)
|
|
after = time.time()
|
|
|
|
assert event.event_type == "MODULE_STARTED"
|
|
assert event.payload["module"] == "dns_logger"
|
|
assert event.source_module == "engine"
|
|
assert before <= event.timestamp <= after
|
|
|
|
d = event.to_dict()
|
|
assert isinstance(d, dict)
|
|
assert d["event_type"] == "MODULE_STARTED"
|
|
assert d["payload"]["pid"] == 1234
|