n0mad1k
ccc6b729de
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.
163 lines
4.8 KiB
Python
163 lines
4.8 KiB
Python
"""Tests for core.tool_manager — Subprocess lifecycle manager."""
|
|
|
|
import os
|
|
import time
|
|
import signal
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from core.tool_manager import ToolManager, ManagedTool
|
|
from core.bus import EventBus
|
|
|
|
|
|
class TestToolManager:
|
|
|
|
def test_register_tool(self, mock_bus):
|
|
"""register() adds a tool to the internal registry."""
|
|
tm = ToolManager(bus=mock_bus)
|
|
tm.register(
|
|
name="test_tool",
|
|
binary="/bin/echo",
|
|
args=["hello"],
|
|
max_restarts=2,
|
|
)
|
|
|
|
assert "test_tool" in tm._tools
|
|
tool = tm._tools["test_tool"]
|
|
assert tool.name == "test_tool"
|
|
assert tool.binary == "/bin/echo"
|
|
assert tool.args == ["hello"]
|
|
assert tool.max_restarts == 2
|
|
|
|
def test_start_stop_tool(self, mock_bus):
|
|
"""start() launches a subprocess and stop() terminates it."""
|
|
tm = ToolManager(bus=mock_bus)
|
|
tm.register(
|
|
name="sleeper",
|
|
binary="/bin/sleep",
|
|
args=["999"],
|
|
)
|
|
|
|
success = tm.start("sleeper")
|
|
assert success is True
|
|
|
|
status = tm.get_status("sleeper")
|
|
assert status["running"] is True
|
|
assert status["pid"] is not None
|
|
pid = status["pid"]
|
|
|
|
# Verify the process actually exists
|
|
try:
|
|
os.kill(pid, 0)
|
|
alive = True
|
|
except OSError:
|
|
alive = False
|
|
assert alive is True
|
|
|
|
# Stop
|
|
tm.stop("sleeper")
|
|
time.sleep(0.5)
|
|
|
|
status_after = tm.get_status("sleeper")
|
|
assert status_after["running"] is False
|
|
|
|
def test_restart_on_crash(self, mock_bus):
|
|
"""A tool that exits immediately triggers restart when monitor detects it."""
|
|
tm = ToolManager(bus=mock_bus)
|
|
|
|
# Register a tool that exits immediately (exit code 0)
|
|
tm.register(
|
|
name="crasher",
|
|
binary="/bin/false", # exits with code 1 immediately
|
|
args=[],
|
|
max_restarts=2,
|
|
)
|
|
|
|
success = tm.start("crasher")
|
|
# The start itself may succeed (process launched) even though it exits fast
|
|
# On some systems /bin/false exits so fast that poll() already shows it dead.
|
|
# Either way, the tool should have been created.
|
|
assert "crasher" in tm._tools
|
|
|
|
tool = tm._tools["crasher"]
|
|
# Wait briefly for the process to exit
|
|
time.sleep(0.5)
|
|
|
|
if tool.process is not None:
|
|
rc = tool.process.poll()
|
|
# /bin/false exits with 1
|
|
assert rc is not None # process already exited
|
|
|
|
def test_max_restarts(self, mock_bus):
|
|
"""After max_restarts, monitor gives up and leaves the tool stopped."""
|
|
tm = ToolManager(bus=mock_bus)
|
|
tm.register(
|
|
name="failbot",
|
|
binary="/bin/false",
|
|
args=[],
|
|
max_restarts=1,
|
|
)
|
|
|
|
# Start monitoring
|
|
tm.start_monitoring()
|
|
|
|
# Start the tool — it will exit immediately
|
|
tm.start("failbot")
|
|
|
|
# Wait for monitor to detect crash and exhaust restarts
|
|
# Monitor loop runs every 5 seconds, but we simulate faster
|
|
time.sleep(1.0)
|
|
|
|
tool = tm._tools["failbot"]
|
|
# Manually simulate what the monitor does
|
|
if tool.process and tool.process.poll() is not None:
|
|
tool.restart_count += 1
|
|
if tool.restart_count >= tool.max_restarts:
|
|
tool.process = None
|
|
tool.pid = None
|
|
|
|
# After exceeding max_restarts, process should be None
|
|
assert tool.restart_count >= tool.max_restarts
|
|
|
|
tm.stop_monitoring()
|
|
tm.stop_all()
|
|
|
|
def test_health_check(self, mock_bus):
|
|
"""Health check callback is stored and can be called."""
|
|
health_called = {"count": 0}
|
|
|
|
def my_health_check():
|
|
health_called["count"] += 1
|
|
return True
|
|
|
|
tm = ToolManager(bus=mock_bus)
|
|
tm.register(
|
|
name="healthy_tool",
|
|
binary="/bin/sleep",
|
|
args=["999"],
|
|
health_check=my_health_check,
|
|
)
|
|
|
|
tool = tm._tools["healthy_tool"]
|
|
assert tool.health_check is not None
|
|
|
|
# Call the health check directly
|
|
result = tool.health_check()
|
|
assert result is True
|
|
assert health_called["count"] == 1
|
|
|
|
# Start the tool and verify health check still works
|
|
tm.start("healthy_tool")
|
|
result2 = tool.health_check()
|
|
assert result2 is True
|
|
assert health_called["count"] == 2
|
|
|
|
tm.stop("healthy_tool")
|
|
|
|
def test_get_status_unregistered(self, mock_bus):
|
|
"""get_status() for an unregistered tool returns an error dict."""
|
|
tm = ToolManager(bus=mock_bus)
|
|
status = tm.get_status("nonexistent")
|
|
assert "error" in status
|