"""Process runner with safety controls and structured output."""
import subprocess
import shlex
import time
import re
import os
import signal
from dataclasses import dataclass, asdict
from typing import Optional
@dataclass
class ProcessResult:
"""Structured result from a process execution."""
success: bool
command: str
exit_code: int
stdout: str
stderr: str
duration_ms: int
timed_out: bool
pid: Optional[int] = None
blocked: bool = False
block_reason: Optional[str] = None
BLOCKED_PATTERNS = [
r"rm\s+-rf\s+/\s*$",
r"rm\s+-rf\s+/\*",
r"mkfs\.",
r"dd\s+if=.*of=/dev/sd",
r">\s*/dev/sd",
r"chmod\s+-R\s+777\s+/\s*$",
r":\(\)\{\s*:\|:&\s*\};:",
r"\bshutdown\b",
r"\breboot\b",
r"\binit\s+[06]\b",
r"systemctl\s+disable\s+firewalld",
r"iptables\s+-F",
]
def is_blocked(command: str) -> Optional[str]:
"""Check if a command matches any blocked pattern."""
for pattern in BLOCKED_PATTERNS:
if re.search(pattern, command):
return f"Command matches blocked pattern: {pattern}"
return None
def run(
command: str,
timeout: int = 60,
cwd: Optional[str] = None,
env: Optional[dict] = None,
shell: bool = False,
) -> ProcessResult:
"""
Execute a command with safety controls.
Args:
command: The command string to execute.
timeout: Maximum execution time in seconds.
cwd: Working directory for the command.
env: Environment variables (merged with current env).
shell: Whether to run through shell (default True).
Returns:
ProcessResult with exit code, stdout, stderr, and timing.
"""
block_reason = is_blocked(command)
if block_reason:
return ProcessResult(
success=False,
command=command,
exit_code=-1,
stdout="",
stderr=block_reason,
duration_ms=0,
timed_out=False,
blocked=True,
block_reason=block_reason,
)
run_env = os.environ.copy()
if env:
run_env.update(env)
start = time.monotonic()
timed_out = False
try:
proc = subprocess.run(
command if shell else shlex.split(command),
shell=shell,
capture_output=True,
text=True,
timeout=timeout,
cwd=cwd,
env=run_env,
)
exit_code = proc.returncode
stdout = proc.stdout
stderr = proc.stderr
except subprocess.TimeoutExpired as e:
timed_out = True
exit_code = -1
stdout = e.stdout or ""
stderr = e.stderr or ""
except Exception as e:
exit_code = -1
stdout = ""
stderr = str(e)
duration_ms = int((time.monotonic() - start) * 1000)
return ProcessResult(
success=exit_code == 0,
command=command,
exit_code=exit_code,
stdout=stdout if isinstance(stdout, str) else stdout.decode("utf-8", errors="replace"),
stderr=stderr if isinstance(stderr, str) else stderr.decode("utf-8", errors="replace"),
duration_ms=duration_ms,
timed_out=timed_out,
)
def run_background(
command: str,
cwd: Optional[str] = None,
log_file: Optional[str] = None,
) -> ProcessResult:
"""
Start a process in the background and return its PID.
Args:
command: The command to run.
cwd: Working directory.
log_file: File to redirect output to.
Returns:
ProcessResult with PID set.
"""
block_reason = is_blocked(command)
if block_reason:
return ProcessResult(
success=False, command=command, exit_code=-1,
stdout="", stderr=block_reason, duration_ms=0,
timed_out=False, blocked=True, block_reason=block_reason,
)
stdout_dest = subprocess.DEVNULL
if log_file:
stdout_dest = open(log_file, "w")
proc = subprocess.Popen(
command,
shell=True,
stdout=stdout_dest,
stderr=subprocess.STDOUT,
cwd=cwd,
start_new_session=True,
)
return ProcessResult(
success=True, command=command, exit_code=0,
stdout="", stderr="", duration_ms=0,
timed_out=False, pid=proc.pid,
)
def kill_process(pid: int, graceful_timeout: int = 5) -> bool:
"""
Kill a process, trying SIGTERM first then SIGKILL.
Args:
pid: Process ID to kill.
graceful_timeout: Seconds to wait after SIGTERM before SIGKILL.
Returns:
True if process was terminated.
"""
try:
os.kill(pid, signal.SIGTERM)
start = time.monotonic()
while time.monotonic() - start < graceful_timeout:
try:
os.kill(pid, 0)
time.sleep(0.1)
except ProcessLookupError:
return True
os.kill(pid, signal.SIGKILL)
return True
except ProcessLookupError:
return True
except PermissionError:
return False