api/exec_sandbox.py

"""
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