""" exec_sandbox.py — Reusable secure execution sandbox (S365) Design invariants: - Isolated working dir per task_id (tmpfs, cleaned up guaranteed via finally) - Stripped environment: no API keys, tokens, secrets pass through - Hard ceiling _HARD_TIMEOUT_S — cannot be overridden by callers - Sync variant (run_in_sandbox) for quality_guardian thread pool usage - Async variant (run_in_sandbox_async) for FastAPI route usage - Never raises exceptions — returns error dict on all failure paths - RLIMIT_NOFILE/NPROC/FSIZE enforced via preexec_fn (GAP-SANDBOX-RLIMIT) Usage: from api.exec_sandbox import run_in_sandbox, run_in_sandbox_async """ from __future__ import annotations import asyncio import logging import os import signal as _signal import shutil import subprocess import sys import tempfile from pathlib import Path # ── Resource limits (GAP-SANDBOX-RLIMIT) ────────────────────────────────────── # Gap reale confermato: subprocess senza setrlimit → fd exhaustion + fork bomb. # Fix: preexec_fn con RLIMIT_NOFILE/NPROC/FSIZE — gira nel figlio pre-execve. try: import resource as _resource _RLIMIT_AVAILABLE = True except ImportError: _RLIMIT_AVAILABLE = False # Windows / ambienti senza resource module def _set_resource_limits() -> None: """ Imposta limiti hard nel subprocess figlio PRIMA di execve. Fail-safe per limite: se un tipo non è supportato, salta silenziosamente. NON applicato ai sandbox di sessione (stato persistente li renderebbe rotti). Protezioni: - RLIMIT_NOFILE 64: previene fd exhaustion (fork + open flooding) - RLIMIT_NPROC 32: previene fork bomb - RLIMIT_FSIZE 50MB: previene filesystem flooding """ if not _RLIMIT_AVAILABLE: return _limits = [ (_resource.RLIMIT_NOFILE, (64, 64)), (_resource.RLIMIT_NPROC, (32, 32)), (_resource.RLIMIT_FSIZE, (50 * 1024 * 1024, 50 * 1024 * 1024)), ] for limit_type, (soft, hard) in _limits: try: _resource.setrlimit(limit_type, (soft, hard)) except (ValueError, OSError): pass # container/OS potrebbe non supportare questo limite def _set_resource_limits_setsid() -> None: """GAP-PTY-FIX: setsid() + resource limits nel subprocess figlio. os.setsid() crea un nuovo process group leader → tutti i processi figli del subprocess appartengono al nuovo group. Su timeout, os.killpg() invia SIGKILL a tutto il process group, non solo al PID diretto. Questo elimina i processi orfani (bash shell → subcomandi → pipeline) che prima sopravvivevano dopo proc.kill() (solo PID diretto). """ os.setsid() # nuovo session leader → nuovo process group _set_resource_limits() # poi applica RLIMIT come prima # ── Constants ───────────────────────────────────────────────────────────────── _HARD_TIMEOUT_S: int = 12 # absolute ceiling per task generici — quality_guardian uses 12 _EXTENDED_TIMEOUT_S: int = 120 # Shadow Execution Kernel — build/test/install task # Pattern che attivano il timeout esteso (build, test, install — legittimamente lenti) _EXTENDED_CMD_PATTERNS: frozenset[str] = frozenset({ 'npm install', 'pnpm install', 'pip install', 'pip3 install', 'pnpm build', 'npm run build', 'yarn build', 'pytest', 'python -m pytest', 'npm test', 'cargo build', 'go build', 'mvn', 'gradle', 'make', 'cmake', }) def _resolve_timeout(code: str, requested: int) -> int: """Shadow Execution Kernel: usa _EXTENDED_TIMEOUT_S per build/test/install. Doppio token: il validatore B può autorizzare task con complexity=high. """; if any(p in code for p in _EXTENDED_CMD_PATTERNS): return min(requested, _EXTENDED_TIMEOUT_S) return min(requested, _HARD_TIMEOUT_S) # Env vars that are safe to pass into the sandbox (pure computation, no secrets) _SAFE_ENV_KEYS: frozenset[str] = frozenset({ "PATH", "LANG", "LC_ALL", "LC_CTYPE", "TERM", "PYTHONDONTWRITEBYTECODE", "PYTHONUNBUFFERED", "NODE_ENV", }) # ── Environment stripping ───────────────────────────────────────────────────── def _safe_env(sandbox_dir: str) -> dict[str, str]: """Return env with no secrets/tokens — only safe computation vars.""" env: dict[str, str] = {k: v for k, v in os.environ.items() if k in _SAFE_ENV_KEYS} env["HOME"] = sandbox_dir env["TMPDIR"] = sandbox_dir env["PWD"] = sandbox_dir env.setdefault("PATH", "/usr/local/bin:/usr/bin:/bin") return env # ── Venv Python helper (S756) ───────────────────────────────────────────────── _VENV_PYTHON_PATH = '/data/venv/bin/python3' _logger = logging.getLogger('agente_ai') if not os.path.exists(_VENV_PYTHON_PATH): # GAP-5: visibilità diagnostica _logger.warning('S756: /data/venv non trovato — pip install non persistente tra restart') def _venv_python() -> str: """Usa /data/venv/bin/python3 se il venv S755 esiste, altrimenti sys.executable.""" return _VENV_PYTHON_PATH if os.path.exists(_VENV_PYTHON_PATH) else sys.executable # ── Sandbox lifecycle ───────────────────────────────────────────────────────── def create_sandbox(task_id: str = "default") -> Path: """Create isolated working dir. Caller MUST call cleanup_sandbox() afterwards.""" prefix = f"agent_{task_id[:28].replace('/', '_')}_" return Path(tempfile.mkdtemp(prefix=prefix)) def cleanup_sandbox(sandbox_dir: "Path | str") -> None: """Remove sandbox dir. Silent on any failure.""" try: shutil.rmtree(str(sandbox_dir), ignore_errors=True) except Exception: pass # S365: cleanup is best-effort, never fatal # ── Sync execution (quality_guardian runs this in a thread-pool executor) ───── def run_in_sandbox( code: str, task_id: str = "default", timeout: int = _HARD_TIMEOUT_S, ) -> dict: """ Execute Python code string in an isolated sandbox. Returns: {returncode, stdout, stderr} Never raises. """ effective_timeout = _resolve_timeout(code, int(timeout)) if 'code' in dir() else min(int(timeout), _HARD_TIMEOUT_S) sandbox = create_sandbox(task_id) try: proc = subprocess.run( [_venv_python(), "-c", code], capture_output=True, text=True, timeout=effective_timeout, cwd=str(sandbox), env=_safe_env(str(sandbox)), preexec_fn=_set_resource_limits if _RLIMIT_AVAILABLE else None, # GAP-SANDBOX-RLIMIT ) return { "returncode": proc.returncode, "stdout": proc.stdout[:8000], # S568: allineato con async variant (era 800) "stderr": proc.stderr[:4000], # S568: allineato con async variant (era 500) } except subprocess.TimeoutExpired: return {"returncode": 1, "stdout": "", "stderr": f"timeout after {effective_timeout}s"} except Exception as e: return {"returncode": 1, "stdout": "", "stderr": str(e)[:300]} # S588: 200→300 finally: cleanup_sandbox(sandbox) # S365: guaranteed cleanup even on exception # ── Async execution (FastAPI routes / unified_loop) ─────────────────────────── async def run_in_sandbox_async( code: str, lang: str = "python", task_id: str = "default", timeout: float = float(_HARD_TIMEOUT_S), ) -> dict: """ Async sandbox execution for FastAPI context. Supports: python, js/javascript, node. Returns: {returncode, stdout, stderr} Never raises. """ effective_timeout = float(_resolve_timeout(code, int(timeout))) sandbox = create_sandbox(task_id) try: if lang in ("python", "py"): cmd = [_venv_python(), "-c", code] # S756: usa venv se disponibile elif lang in ("js", "javascript", "node"): cmd = ["node", "-e", code] elif lang in ("bash", "shell", "sh"): cmd = ["bash", "-c", code] # S679: bash support per execute_shell sandbox else: return {"returncode": 1, "stdout": "", "stderr": f"unsupported lang: {lang}"} proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=str(sandbox), env=_safe_env(str(sandbox)), preexec_fn=_set_resource_limits_setsid if _RLIMIT_AVAILABLE else os.setsid, # GAP-PTY-FIX ) try: stdout, stderr = await asyncio.wait_for( proc.communicate(), timeout=effective_timeout ) except asyncio.TimeoutError: # GAP-PTY-FIX: killpg per terminare tutto il process group (figli inclusi) try: os.killpg(os.getpgid(proc.pid), _signal.SIGKILL) except (ProcessLookupError, PermissionError): proc.kill() # fallback se il gruppo non è accessibile await proc.wait() return {"returncode": -1, "stdout": "", "stderr": f"timeout after {effective_timeout}s"} return { "returncode": proc.returncode, "stdout": stdout.decode("utf-8", errors="replace")[:8000], "stderr": stderr.decode("utf-8", errors="replace")[:4000], } except Exception as e: return {"returncode": 1, "stdout": "", "stderr": str(e)[:300]} # S589: 200→300 finally: cleanup_sandbox(sandbox) # S365: guaranteed cleanup # ── Session-based execution (S694: Persistenza sandbox) ─────────────────── import threading as _threading import time as _time class _SessionSandboxManager: """Singleton: gestisce sandbox persistenti per session_id. La sandbox sopravvive tra le chiamate — permette installare lib e riusarle. """ _MAX_SESSIONS = 30 _TTL_S = 1800 # 30 min inattività def __init__(self) -> None: self._sessions: dict[str, dict] = {} self._lock = _threading.Lock() def get_or_create(self, session_id: str) -> Path: with self._lock: if session_id in self._sessions: self._sessions[session_id]['last_used'] = _time.monotonic() return Path(self._sessions[session_id]['path']) if len(self._sessions) >= self._MAX_SESSIONS: oldest = min(self._sessions, key=lambda k: self._sessions[k]['last_used']) self._evict(oldest) path = tempfile.mkdtemp(prefix=f"sess_{session_id[:16].replace('/', '_')}_") self._sessions[session_id] = {'path': path, 'last_used': _time.monotonic()} return Path(path) def evict(self, session_id: str) -> None: with self._lock: self._evict(session_id) def _evict(self, session_id: str) -> None: entry = self._sessions.pop(session_id, None) if entry: shutil.rmtree(entry['path'], ignore_errors=True) def prune_stale(self) -> int: now = _time.monotonic() with self._lock: dead = [sid for sid, e in self._sessions.items() if now - e['last_used'] > self._TTL_S] for sid in dead: self._evict(sid) return len(dead) _session_manager = _SessionSandboxManager() async def run_in_sandbox_session( code: str, lang: str = "python", session_id: str = "default", timeout: float = float(_HARD_TIMEOUT_S), ) -> dict: """ Esegui codice in una sandbox persistente per session_id. La sandbox sopravvive tra le chiamate — stato, file e librerie installate restano disponibili per la stessa sessione (fino a TTL o evict manuale). Returns: {returncode, stdout, stderr, session_id} Never raises. """ effective_timeout = float(_resolve_timeout(code, int(timeout))) sandbox = _session_manager.get_or_create(session_id) try: if lang in ("python", "py"): cmd = [sys.executable, "-c", code] elif lang in ("js", "javascript", "node"): cmd = ["node", "-e", code] elif lang in ("bash", "shell", "sh"): cmd = ["bash", "-c", code] else: return {"returncode": 1, "stdout": "", "stderr": f"unsupported lang: {lang}"} proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=str(sandbox), env=_safe_env(str(sandbox)), ) try: stdout, stderr = await asyncio.wait_for( proc.communicate(), timeout=effective_timeout ) except asyncio.TimeoutError: proc.kill() await proc.wait() return {"returncode": -1, "stdout": "", "stderr": f"timeout after {effective_timeout}s", "session_id": session_id} return { "returncode": proc.returncode, "stdout": stdout.decode("utf-8", errors="replace")[:8000], "stderr": stderr.decode("utf-8", errors="replace")[:4000], "session_id": session_id, } except Exception as e: return {"returncode": 1, "stdout": "", "stderr": str(e)[:300], "session_id": session_id} # NON chiama cleanup_sandbox — la sandbox persiste per tutta la sessione