Cobra
ffd384f64b
Replace all sensor.* logger namespaces with __name__ (generic module identifiers instead of discoverable 'sensor.*' prefixes). Change hardcoded 'bb' API user to 'admin' in config and code defaults. Change hardcoded relay_user 'bb' to 'operator' — prevents network profiling from exposing tool identity via SSH config. Fixes #457, #458, #459
422 lines
16 KiB
Python
422 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Subprocess lifecycle manager for external tools.
|
|
|
|
Manages bettercap, tcpdump, Responder, mitmproxy, ntlmrelayx as
|
|
supervised subprocesses with:
|
|
- Start/stop/restart with configurable retry (max 3, exponential backoff)
|
|
- stdout/stderr capture
|
|
- PID tracking for kill_switch integration
|
|
- Resource monitoring via /proc/PID/stat
|
|
- Process disguise integration (prctl rename)
|
|
- Health check callbacks
|
|
- Graceful shutdown (SIGTERM -> 5s -> SIGKILL)
|
|
- Crash callbacks (e.g., ARP restoration on bettercap crash)
|
|
"""
|
|
|
|
import os
|
|
import time
|
|
import signal
|
|
import logging
|
|
import subprocess
|
|
import threading
|
|
from dataclasses import dataclass, field
|
|
from typing import Callable, Optional
|
|
|
|
from core.bus import EventBus
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ManagedTool:
|
|
"""Configuration and runtime state for a managed subprocess."""
|
|
name: str
|
|
binary: str
|
|
args: list = field(default_factory=list)
|
|
health_check: Optional[Callable] = None
|
|
crash_callback: Optional[Callable] = None
|
|
max_restarts: int = 3
|
|
disguise_name: str = ""
|
|
env: dict = field(default_factory=dict)
|
|
cwd: Optional[str] = None
|
|
|
|
# Runtime state
|
|
process: Optional[subprocess.Popen] = field(default=None, repr=False)
|
|
pid: Optional[int] = None
|
|
restart_count: int = 0
|
|
started_at: Optional[float] = None
|
|
_stdout_thread: Optional[threading.Thread] = field(default=None, repr=False)
|
|
_stderr_thread: Optional[threading.Thread] = field(default=None, repr=False)
|
|
stdout_lines: list = field(default_factory=list, repr=False)
|
|
stderr_lines: list = field(default_factory=list, repr=False)
|
|
_max_log_lines: int = 1000
|
|
|
|
|
|
class ToolManager:
|
|
"""Manages external tool subprocesses with supervision.
|
|
|
|
Usage:
|
|
tm = ToolManager(bus)
|
|
tm.register(name="bettercap", binary="/usr/local/bin/bettercap",
|
|
args=["-api-rest-address", "127.0.0.1"],
|
|
health_check=lambda: api.is_alive(),
|
|
crash_callback=on_crash, max_restarts=3,
|
|
disguise_name="networkd-dispatcher")
|
|
tm.start("bettercap")
|
|
tm.stop("bettercap")
|
|
"""
|
|
|
|
SIGTERM_TIMEOUT = 5.0 # seconds to wait after SIGTERM before SIGKILL
|
|
|
|
def __init__(self, bus: EventBus):
|
|
self.bus = bus
|
|
self._tools: dict[str, ManagedTool] = {}
|
|
self._lock = threading.Lock()
|
|
self._monitor_thread: Optional[threading.Thread] = None
|
|
self._running = False
|
|
|
|
# ------------------------------------------------------------------
|
|
# Lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
def start_monitoring(self) -> None:
|
|
"""Start the background health/crash monitor."""
|
|
if self._running:
|
|
return
|
|
self._running = True
|
|
self._monitor_thread = threading.Thread(
|
|
target=self._monitor_loop, daemon=True, name="sensor-tool-monitor"
|
|
)
|
|
self._monitor_thread.start()
|
|
|
|
def stop_monitoring(self) -> None:
|
|
"""Stop monitoring thread."""
|
|
self._running = False
|
|
if self._monitor_thread and self._monitor_thread.is_alive():
|
|
self._monitor_thread.join(timeout=3.0)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Registration
|
|
# ------------------------------------------------------------------
|
|
|
|
def register(self, name: str, binary: str, args: list = None,
|
|
health_check: Callable = None, crash_callback: Callable = None,
|
|
max_restarts: int = 3, disguise_name: str = "",
|
|
env: dict = None, cwd: str = None) -> None:
|
|
"""Register a tool for management."""
|
|
with self._lock:
|
|
self._tools[name] = ManagedTool(
|
|
name=name,
|
|
binary=binary,
|
|
args=args or [],
|
|
health_check=health_check,
|
|
crash_callback=crash_callback,
|
|
max_restarts=max_restarts,
|
|
disguise_name=disguise_name,
|
|
env=env or {},
|
|
cwd=cwd,
|
|
)
|
|
logger.info("Registered tool: %s (%s)", name, binary)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Start / Stop / Restart
|
|
# ------------------------------------------------------------------
|
|
|
|
def start(self, name: str) -> bool:
|
|
"""Start a registered tool subprocess."""
|
|
with self._lock:
|
|
tool = self._tools.get(name)
|
|
if tool is None:
|
|
logger.error("Tool %s not registered", name)
|
|
return False
|
|
|
|
if tool.process and tool.process.poll() is None:
|
|
logger.warning("Tool %s already running (pid=%d)", name, tool.pid)
|
|
return True
|
|
|
|
return self._start_tool(tool)
|
|
|
|
def _start_tool(self, tool: ManagedTool) -> bool:
|
|
"""Internal: launch the subprocess. Caller must hold _lock."""
|
|
cmd = [tool.binary] + tool.args
|
|
env = dict(os.environ)
|
|
env.update(tool.env)
|
|
|
|
try:
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=env,
|
|
cwd=tool.cwd,
|
|
preexec_fn=self._make_preexec(tool.disguise_name) if tool.disguise_name else None,
|
|
)
|
|
except FileNotFoundError:
|
|
logger.error("Binary not found: %s", tool.binary)
|
|
return False
|
|
except Exception as e:
|
|
logger.error("Failed to start %s: %s", tool.name, e)
|
|
return False
|
|
|
|
tool.process = proc
|
|
tool.pid = proc.pid
|
|
tool.started_at = time.time()
|
|
tool.stdout_lines.clear()
|
|
tool.stderr_lines.clear()
|
|
|
|
# Stdout/stderr reader threads
|
|
tool._stdout_thread = threading.Thread(
|
|
target=self._pipe_reader, args=(proc.stdout, tool.stdout_lines, tool._max_log_lines),
|
|
daemon=True, name=f"bb-{tool.name}-stdout",
|
|
)
|
|
tool._stderr_thread = threading.Thread(
|
|
target=self._pipe_reader, args=(proc.stderr, tool.stderr_lines, tool._max_log_lines),
|
|
daemon=True, name=f"bb-{tool.name}-stderr",
|
|
)
|
|
tool._stdout_thread.start()
|
|
tool._stderr_thread.start()
|
|
|
|
self.bus.emit("TOOL_RESTARTED" if tool.restart_count > 0 else "MODULE_STARTED",
|
|
{"tool": tool.name, "pid": proc.pid, "restart_count": tool.restart_count},
|
|
source_module="tool_manager")
|
|
logger.info("Started tool %s (pid=%d)", tool.name, proc.pid)
|
|
return True
|
|
|
|
def stop(self, name: str) -> bool:
|
|
"""Stop a tool. SIGTERM -> 5s timeout -> SIGKILL."""
|
|
with self._lock:
|
|
tool = self._tools.get(name)
|
|
if tool is None:
|
|
return True
|
|
return self._stop_tool(tool)
|
|
|
|
def _stop_tool(self, tool: ManagedTool) -> bool:
|
|
"""Internal: stop subprocess. Caller must hold _lock."""
|
|
proc = tool.process
|
|
if proc is None or proc.poll() is not None:
|
|
tool.process = None
|
|
tool.pid = None
|
|
return True
|
|
|
|
# Graceful: SIGTERM
|
|
try:
|
|
proc.terminate()
|
|
except OSError:
|
|
pass
|
|
|
|
try:
|
|
proc.wait(timeout=self.SIGTERM_TIMEOUT)
|
|
except subprocess.TimeoutExpired:
|
|
# Forceful: SIGKILL
|
|
logger.warning("Tool %s did not stop after SIGTERM, sending SIGKILL", tool.name)
|
|
try:
|
|
proc.kill()
|
|
proc.wait(timeout=2.0)
|
|
except Exception:
|
|
pass
|
|
|
|
tool.process = None
|
|
tool.pid = None
|
|
tool.started_at = None
|
|
logger.info("Stopped tool %s", tool.name)
|
|
return True
|
|
|
|
def restart(self, name: str) -> bool:
|
|
"""Stop then start a tool, resetting restart count."""
|
|
with self._lock:
|
|
tool = self._tools.get(name)
|
|
if tool is None:
|
|
return False
|
|
self._stop_tool(tool)
|
|
tool.restart_count = 0
|
|
return self._start_tool(tool)
|
|
|
|
def stop_all(self) -> None:
|
|
"""Stop all managed tools."""
|
|
with self._lock:
|
|
for tool in self._tools.values():
|
|
self._stop_tool(tool)
|
|
self.stop_monitoring()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Status / resources
|
|
# ------------------------------------------------------------------
|
|
|
|
def get_status(self, name: str) -> dict:
|
|
"""Get tool status including resource usage from /proc."""
|
|
with self._lock:
|
|
tool = self._tools.get(name)
|
|
if tool is None:
|
|
return {"error": "not registered"}
|
|
|
|
alive = tool.process is not None and tool.process.poll() is None
|
|
result = {
|
|
"name": tool.name,
|
|
"running": alive,
|
|
"pid": tool.pid,
|
|
"uptime": time.time() - tool.started_at if tool.started_at and alive else 0,
|
|
"restart_count": tool.restart_count,
|
|
}
|
|
|
|
if alive and tool.pid:
|
|
resources = self._read_proc_stats(tool.pid)
|
|
result.update(resources)
|
|
|
|
return result
|
|
|
|
def get_all_status(self) -> dict:
|
|
"""Get status for all registered tools."""
|
|
with self._lock:
|
|
names = list(self._tools.keys())
|
|
return {n: self.get_status(n) for n in names}
|
|
|
|
def get_all_pids(self) -> list[int]:
|
|
"""Return PIDs of all running tools (for kill_switch)."""
|
|
pids = []
|
|
with self._lock:
|
|
for tool in self._tools.values():
|
|
if tool.pid and tool.process and tool.process.poll() is None:
|
|
pids.append(tool.pid)
|
|
return pids
|
|
|
|
@staticmethod
|
|
def _read_proc_stats(pid: int) -> dict:
|
|
"""Read RAM/CPU from /proc/PID/stat and /proc/PID/statm."""
|
|
result = {"ram_mb": 0.0, "cpu_pct": 0.0}
|
|
try:
|
|
with open(f"/proc/{pid}/statm", "r") as f:
|
|
parts = f.read().split()
|
|
# statm: size resident shared text lib data dt (in pages)
|
|
page_size = os.sysconf("SC_PAGE_SIZE")
|
|
result["ram_mb"] = round(int(parts[1]) * page_size / (1024 * 1024), 1)
|
|
except (FileNotFoundError, PermissionError, IndexError, ValueError):
|
|
pass
|
|
|
|
try:
|
|
with open(f"/proc/{pid}/stat", "r") as f:
|
|
parts = f.read().split()
|
|
# Fields 14,15 = utime, stime in clock ticks
|
|
utime = int(parts[13])
|
|
stime = int(parts[14])
|
|
total_ticks = utime + stime
|
|
hz = os.sysconf("SC_CLK_TCK")
|
|
uptime_f = open("/proc/uptime", "r")
|
|
system_uptime = float(uptime_f.read().split()[0])
|
|
uptime_f.close()
|
|
start_ticks = int(parts[21])
|
|
proc_uptime = system_uptime - (start_ticks / hz)
|
|
if proc_uptime > 0:
|
|
result["cpu_pct"] = round((total_ticks / hz) / proc_uptime * 100, 1)
|
|
except (FileNotFoundError, PermissionError, IndexError, ValueError, ZeroDivisionError):
|
|
pass
|
|
|
|
return result
|
|
|
|
# ------------------------------------------------------------------
|
|
# Monitor loop
|
|
# ------------------------------------------------------------------
|
|
|
|
def _monitor_loop(self) -> None:
|
|
"""Background loop: detect crashes, run health checks, auto-restart."""
|
|
while self._running:
|
|
time.sleep(5.0)
|
|
with self._lock:
|
|
tools = list(self._tools.values())
|
|
|
|
for tool in tools:
|
|
if tool.process is None:
|
|
continue
|
|
|
|
rc = tool.process.poll()
|
|
if rc is not None:
|
|
# Process exited
|
|
logger.warning("Tool %s exited (rc=%s)", tool.name, rc)
|
|
self.bus.emit("TOOL_CRASHED",
|
|
{"tool": tool.name, "exit_code": rc,
|
|
"restart_count": tool.restart_count},
|
|
source_module="tool_manager")
|
|
|
|
# Call crash callback OUTSIDE lock to prevent deadlock
|
|
callback = tool.crash_callback
|
|
restart_needed = tool.restart_count < tool.max_restarts
|
|
|
|
# Release lock before callback
|
|
if callback:
|
|
try:
|
|
callback()
|
|
except Exception:
|
|
logger.exception("Crash callback failed for %s", tool.name)
|
|
|
|
# Re-acquire lock for restart decision
|
|
with self._lock:
|
|
# Auto-restart with exponential backoff
|
|
if tool.restart_count < tool.max_restarts:
|
|
backoff = min(2 ** tool.restart_count, 30)
|
|
logger.info("Restarting %s in %ds (attempt %d/%d)",
|
|
tool.name, backoff, tool.restart_count + 1,
|
|
tool.max_restarts)
|
|
time.sleep(backoff)
|
|
tool.restart_count += 1
|
|
self._start_tool(tool)
|
|
else:
|
|
logger.error("Tool %s exceeded max restarts (%d), giving up",
|
|
tool.name, tool.max_restarts)
|
|
tool.process = None
|
|
tool.pid = None
|
|
|
|
elif tool.health_check:
|
|
# Process alive — run health check
|
|
try:
|
|
if not tool.health_check():
|
|
logger.warning("Health check failed for %s", tool.name)
|
|
except Exception:
|
|
logger.exception("Health check error for %s", tool.name)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _pipe_reader(pipe, line_buffer: list, max_lines: int) -> None:
|
|
"""Read lines from a subprocess pipe into a bounded buffer."""
|
|
try:
|
|
for line in iter(pipe.readline, b""):
|
|
try:
|
|
decoded = line.decode("utf-8", errors="replace").rstrip()
|
|
except Exception:
|
|
decoded = repr(line)
|
|
line_buffer.append(decoded)
|
|
if len(line_buffer) > max_lines:
|
|
line_buffer.pop(0)
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
try:
|
|
pipe.close()
|
|
except Exception:
|
|
pass
|
|
|
|
@staticmethod
|
|
def _make_preexec(disguise_name: str):
|
|
"""Return a preexec_fn that renames the process via prctl."""
|
|
def _preexec():
|
|
try:
|
|
import ctypes
|
|
libc = ctypes.CDLL("libc.so.6", use_errno=True)
|
|
PR_SET_NAME = 15
|
|
name_bytes = disguise_name[:15].encode("utf-8")
|
|
libc.prctl(PR_SET_NAME, name_bytes, 0, 0, 0)
|
|
except Exception:
|
|
pass
|
|
return _preexec
|
|
|
|
def get_output(self, name: str, stream: str = "stderr",
|
|
lines: int = 50) -> list[str]:
|
|
"""Get recent stdout/stderr lines from a tool."""
|
|
with self._lock:
|
|
tool = self._tools.get(name)
|
|
if tool is None:
|
|
return []
|
|
buf = tool.stderr_lines if stream == "stderr" else tool.stdout_lines
|
|
return list(buf[-lines:])
|