"""backend/api/exec.py — Code execution, shell, pip-install, LLM fix (S354).""" import os, asyncio, sys, tempfile, time, resource as _resource, signal as _signal import re as _re_exec import ast as _ast_mod from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel, model_validator from .auth_guard import require_role, AuthRole try: from .priority import realtime_job as _realtime_job, background_job as _background_job except ImportError: # Fallback graceful se priority.py non ancora deployato from contextlib import asynccontextmanager import asyncio as _asyncio_fallback _FALLBACK_REALTIME_SEM = _asyncio_fallback.Semaphore(6) _FALLBACK_BACKGROUND_SEM = _asyncio_fallback.Semaphore(2) @asynccontextmanager async def _realtime_job(**_): async with _FALLBACK_REALTIME_SEM: yield @asynccontextmanager async def _background_job(**_): async with _FALLBACK_BACKGROUND_SEM: yield import logging _logger = logging.getLogger("api.exec") router = APIRouter() # ── Venv Python helpers (S756) ──────────────────────────────────────────────── # Usa il Python/pip del venv persistente (/data/venv) se disponibile. # Il venv viene creato dal terminale S755 al primo avvio. # Fallback al Python/pip di sistema se il venv non è ancora stato inizializzato. _VENV_PYTHON = '/data/venv/bin/python3' _VENV_PIP = '/data/venv/bin/pip' def _get_venv_python() -> str: """Ritorna il Python del venv se disponibile, altrimenti \'python3\' di sistema.""" import os as _os return _VENV_PYTHON if _os.path.exists(_VENV_PYTHON) else 'python3' def _get_venv_pip_cmd(pkgs: list) -> list: """ Ritorna il comando pip completo per installare pkgs nel venv (se disponibile) o con --user come fallback su sistema. S756: allineato con il venv S755 — pip install persiste tra restart HF Space. """ import os as _os if _os.path.exists(_VENV_PIP): return [_VENV_PIP, 'install', '--quiet', *pkgs] # Fallback: sistema — --user (non persistente ma funziona) return [sys.executable, '-m', 'pip', 'install', '--quiet', '--user', *pkgs] BLOCKED_CMDS = {'rm -rf /', 'mkfs', ':(){:|:&};:', 'dd if=/dev/zero'} # ── GAP-EXEC-FIX: resource limits per subprocess child ─────────────────────── # Chiamato come preexec_fn DOPO fork/PRIMA exec nel processo figlio. # Ogni setrlimit è in try/except isolato — non blocca su piattaforme che non # lo supportano (macOS dev, HF Space gVisor, ecc.). def _child_resource_limits() -> None: """GAP-EXEC-FIX: applica RLIMIT a exec_code e execute_shell.""" try: os.setsid() # P40-A: crea nuovo process group — killpg uccide l'intero albero except Exception as _exc: _logger.debug("[exec] setsid silenced %s", type(_exc).__name__) # noqa: BLE001 try: # Virtual address space: 1.5 GB — gestisce node/npm mmap senza crashare FastAPI _resource.setrlimit(_resource.RLIMIT_AS, (1_500_000_000, 1_500_000_000)) except Exception as _exc: _logger.debug("[exec] silenced %s", type(_exc).__name__) # noqa: BLE001 try: # CPU time: 120 s — defense-in-depth se asyncio kill fallisce _resource.setrlimit(_resource.RLIMIT_CPU, (120, 120)) except Exception as _exc: _logger.debug("[exec] silenced %s", type(_exc).__name__) # noqa: BLE001 try: # File descriptors: 512 — previene fd leak da processi runaway _resource.setrlimit(_resource.RLIMIT_NOFILE, (512, 512)) except Exception as _exc: _logger.debug("[exec] silenced %s", type(_exc).__name__) # noqa: BLE001 try: # Processi figli: 64 — fork bomb prevention (:(){:|:&};: già in BLOCKED_CMDS) _resource.setrlimit(_resource.RLIMIT_NPROC, (64, 64)) except Exception as _exc: _logger.debug("[exec] silenced %s", type(_exc).__name__) # noqa: BLE001 def _install_resource_limits() -> None: """INSTALL-RLIMIT: limiti rilassati per pip/npm install. npm e pip richiedono più virtual address space (mmap aggressivo) e più file descriptor rispetto all'esecuzione codice normale. Non imponiamo RLIMIT_CPU perché npm install su cache fredda può durare > 120s legittimamente. Il container Railway gestisce il wall-clock via timeout asyncio superiore. """ try: os.setsid() # P40-A: crea nuovo process group except Exception as _exc: _logger.debug("[exec] setsid silenced %s", type(_exc).__name__) # noqa: BLE001 try: # Virtual address space: 2.5 GB — npm usa mmap su file grandi (node_modules) _resource.setrlimit(_resource.RLIMIT_AS, (2_500_000_000, 2_500_000_000)) except Exception as _exc: _logger.debug("[exec] silenced %s", type(_exc).__name__) # noqa: BLE001 try: # File descriptors: 1024 — pip apre molti fd in parallelo durante install _resource.setrlimit(_resource.RLIMIT_NOFILE, (1024, 1024)) except Exception as _exc: _logger.debug("[exec] silenced %s", type(_exc).__name__) # noqa: BLE001 try: # Processi figli: 128 — npm spawna worker durante install _resource.setrlimit(_resource.RLIMIT_NPROC, (128, 128)) except Exception as _exc: _logger.debug("[exec] silenced %s", type(_exc).__name__) # noqa: BLE001 # NOTA: nessun RLIMIT_CPU — install legittime possono durare > 120s su cache fredda def _detect_oom_kill(returncode: int | None) -> str | None: """Rileva SIGKILL dal kernel (OOM killer) e restituisce messaggio chiaro. returncode == -9 significa che il processo è stato ucciso da SIGKILL. Su Railway free (512MB RAM) accade quando npm/pip supera il memory budget. Ritorna None se returncode non indica OOM kill. """ if returncode == -9: return ( '⚠️ Processo ucciso dal kernel (OOM — out of memory).\n' 'Railway free tier ha 512MB RAM. Soluzioni:\n' '• Installa i pacchetti in batch separati (max 3 alla volta)\n' '• Usa versioni più leggere dei pacchetti (es. torch-cpu invece di torch)\n' '• Verifica che /data/venv esista — il venv persistente riduce il footprint' ) return None def _killpg(proc: "asyncio.subprocess.Process") -> None: """P40-A: Uccide l'intero process group (processo + tutti i figli/nipoti). proc.kill() invia SIGKILL solo al PID diretto; i figli (npm workers, Python fork, shell pipeline) sopravvivono come zombie fino a RLIMIT_CPU (120s) o OOM killer Railway. os.killpg() invia SIGKILL all'intero process group creato da os.setsid() nel preexec_fn, garantendo cleanup completo. """ try: pgid = os.getpgid(proc.pid) os.killpg(pgid, _signal.SIGKILL) except ProcessLookupError: pass # processo già terminato — normale race condition tra timeout e exit except Exception: try: proc.kill() # fallback: almeno uccidi il processo principale except Exception: pass class ExecRequest(BaseModel): code: str lang: str = "python" @model_validator(mode='before') @classmethod def normalize_lang(cls, data: object) -> object: if isinstance(data, dict) and 'language' in data and 'lang' not in data: return {**data, 'lang': data['language']} return data class FixRequest(BaseModel): code: str error: str lang: str = "python" @model_validator(mode='before') @classmethod def normalize_lang(cls, data: object) -> object: if isinstance(data, dict) and 'language' in data and 'lang' not in data: return {**data, 'lang': data['language']} return data class ShellCmd(BaseModel): command: str timeout: int = 15 class PipInstall(BaseModel): packages: list[str] # ── AST sandbox ─────────────────────────────────────────────────────────────── _EXEC_BLOCKED_RE = _re_exec.compile( r'import\s+os\b' r'|import\s+subprocess\b' r'|import\s+socket\b' r'|__import__\s*\(' r'|\beval\s*\(' r'|\bexec\s*\(' r'|\bopen\s*\(' r'|shutil' r'|/etc/' r'|/proc/' r'|breakpoint\s*\(' r'|importlib' ) _AST_BLOCKED_BUILTINS = frozenset({ 'eval', 'exec', 'open', 'compile', 'breakpoint', '__import__', 'input', 'memoryview', 'getattr', 'setattr', 'delattr', 'vars', 'globals', 'locals', }) _AST_BLOCKED_ATTRS = frozenset({ '__class__', '__bases__', '__subclasses__', '__globals__', '__builtins__', '__code__', '__import__', '__reduce__', '__reduce_ex__', '__init_subclass__', 'mro', '__mro__', 'f_locals', 'f_globals', 'gi_frame', 'ag_frame', 'cr_frame', '__dict__', '__module__', '__loader__', '__spec__', '__file__', }) _AST_BLOCKED_MODULES = frozenset({ 'os', 'subprocess', 'socket', 'shutil', 'importlib', 'sys', 'pty', 'ctypes', 'mmap', 'signal', 'threading', 'multiprocessing', 'gc', 'inspect', 'dis', 'traceback', 'linecache', 'pickle', 'pickletools', 'marshal', 'code', 'codeop', 'pdb', 'runpy', 'builtins', '_thread', 'pathlib', 'io', 'glob', 'fnmatch', 'tempfile', 'struct', }) class _ASTSandboxChecker(_ast_mod.NodeVisitor): def __init__(self): self.violations: list[str] = [] def _add(self, reason: str) -> None: self.violations.append(reason) def visit_Import(self, node: _ast_mod.Import) -> None: for alias in node.names: top = alias.name.split('.')[0] if top in _AST_BLOCKED_MODULES: self._add(f'import {alias.name}') self.generic_visit(node) def visit_ImportFrom(self, node: _ast_mod.ImportFrom) -> None: top = (node.module or '').split('.')[0] if top in _AST_BLOCKED_MODULES: self._add(f'from {node.module} import ...') self.generic_visit(node) def visit_Call(self, node: _ast_mod.Call) -> None: if isinstance(node.func, _ast_mod.Name): if node.func.id in _AST_BLOCKED_BUILTINS: self._add(f'{node.func.id}() non consentito nel sandbox') self.generic_visit(node) def visit_Attribute(self, node: _ast_mod.Attribute) -> None: if node.attr in _AST_BLOCKED_ATTRS: self._add(f'attributo .{node.attr} non consentito nel sandbox') self.generic_visit(node) def _ast_sandbox_check(code: str) -> tuple[bool, str]: try: tree = _ast_mod.parse(code, mode='exec') except SyntaxError as e: return False, f'SyntaxError: {e}' checker = _ASTSandboxChecker() checker.visit(tree) if checker.violations: return False, checker.violations[0] return True, '' # ── Routes ──────────────────────────────────────────────────────────────────── @router.post('/api/exec') async def exec_code(req: ExecRequest, request: Request): _internal_token = os.getenv('INTERNAL_TOKEN', '') if _internal_token and request.headers.get('X-Internal-Token') != _internal_token: raise HTTPException(401, 'Unauthorized') code = req.code.strip() lang = req.lang.lower() if not code: return {'stdout': '', 'stderr': 'No code', 'exit_code': 1, 'durationMs': 0} if lang == 'python': _normalized = _re_exec.sub(r'\s+', ' ', code) _blocked_match = _EXEC_BLOCKED_RE.search(_normalized) if _blocked_match: return {'stdout': '', 'stderr': f'Blocked: pattern "{_blocked_match.group()}"', 'exit_code': 1, 'durationMs': 0} _ast_safe, _ast_reason = _ast_sandbox_check(code) if not _ast_safe: return {'stdout': '', 'stderr': f'Blocked (AST): {_ast_reason}', 'exit_code': 1, 'durationMs': 0} t0 = int(time.time() * 1000) try: async with _realtime_job(timeout_s=120.0): with tempfile.TemporaryDirectory() as tmpdir: try: if lang == 'python': cmd = [_get_venv_python(), '-c', code] elif lang in ('javascript', 'js'): cmd = ['node', '-e', code] elif lang in ('typescript', 'ts'): fname = os.path.join(tmpdir, 'snippet.ts') with open(fname, 'w') as f: f.write(code) cmd = ['npx', '--yes', 'ts-node', '--transpile-only', fname] else: return {'stdout': '', 'stderr': f'Unsupported lang: {lang}', 'exit_code': 1, 'durationMs': 0} proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=tmpdir, preexec_fn=_child_resource_limits, # GAP-EXEC-FIX: RLIMIT_AS/CPU/NOFILE/NPROC env={ 'HOME': tmpdir, 'TMPDIR': tmpdir, 'NODE_ENV': 'production', 'PATH': os.environ.get('PATH', '/usr/local/bin:/usr/bin:/bin'), # NPM-CACHE: usa /data/npm-cache persistente — riduce re-download 'npm_config_cache': '/data/npm-cache', # NODE-MEM: limita heap V8 a 384MB per stare in Railway 512MB 'NODE_OPTIONS': '--max-old-space-size=384', }, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=15) return { 'stdout': stdout.decode('utf-8', errors='replace')[:8000], 'stderr': stderr.decode('utf-8', errors='replace')[:4000], 'exit_code': proc.returncode, 'durationMs': int(time.time() * 1000) - t0, } except asyncio.TimeoutError: # S758-ProgExec: capture partial stdout before kill — progressive execution _partial_out, _partial_err = b'', b'' try: _killpg(proc) # P40-A: kill intero process group _partial_out, _partial_err = await asyncio.wait_for(proc.communicate(), timeout=2) except Exception as _exc: _logger.debug("[exec] silenced %s", type(_exc).__name__) # noqa: BLE001 return { 'stdout': _partial_out.decode('utf-8', errors='replace')[:8000], 'stderr': f'⚠️ Timeout 15s — output parziale\n' + _partial_err.decode('utf-8', errors='replace')[:2000], 'exit_code': -1, 'durationMs': 15000, 'partial': True, } except Exception as e: return {'stdout': '', 'stderr': str(e), 'exit_code': -1, 'durationMs': int(time.time() * 1000) - t0} except asyncio.TimeoutError: return {'stdout': '', 'stderr': '⚠️ Nessun slot exec disponibile (server occupato). Riprova tra qualche secondo.', 'exit_code': -1, 'durationMs': int(time.time() * 1000) - t0} @router.post('/api/execute-shell') async def execute_shell(cmd: ShellCmd, request: Request): _internal_token = os.getenv('INTERNAL_TOKEN', '') if _internal_token and request.headers.get('X-Internal-Token') != _internal_token: raise HTTPException(401, 'Unauthorized') raw = cmd.command.strip() for bad in BLOCKED_CMDS: if bad in raw: raise HTTPException(400, 'Command blocked for safety') timeout = min(max(cmd.timeout, 1), 60) try: async with _realtime_job(timeout_s=90.0): with tempfile.TemporaryDirectory() as tmpdir: try: proc = await asyncio.create_subprocess_shell( raw, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=tmpdir, preexec_fn=_child_resource_limits, # GAP-EXEC-FIX: RLIMIT_AS/CPU/NOFILE/NPROC env={**os.environ, 'HOME': tmpdir, 'TMPDIR': tmpdir}, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) return { 'stdout': stdout.decode('utf-8', errors='replace')[:8000], 'stderr': stderr.decode('utf-8', errors='replace')[:4000], 'exit_code': proc.returncode, } except asyncio.TimeoutError: # S758-ProgExec: capture partial stdout before kill — progressive execution _partial_out, _partial_err = b'', b'' try: _killpg(proc) # P40-A: kill intero process group _partial_out, _partial_err = await asyncio.wait_for(proc.communicate(), timeout=2) except Exception as _exc: _logger.debug("[exec] silenced %s", type(_exc).__name__) # noqa: BLE001 return { 'stdout': _partial_out.decode('utf-8', errors='replace')[:8000], 'stderr': f'⚠️ Timeout {timeout}s — output parziale\n' + _partial_err.decode('utf-8', errors='replace')[:2000], 'exit_code': -1, 'partial': True, } except Exception as e: return {'stdout': '', 'stderr': str(e), 'exit_code': -1} except asyncio.TimeoutError: return {'stdout': '', 'stderr': '⚠️ Nessun slot shell disponibile. Riprova tra qualche secondo.', 'exit_code': -1} @router.post('/api/pip-install') async def pip_install( body: PipInstall, role: AuthRole = Depends(require_role(AuthRole.MACHINE)), # GAP-AUTH-FIX: era aperto ): import re pkgs = [p.strip() for p in body.packages if p.strip()] if not pkgs: raise HTTPException(400, 'No packages specified') safe = re.compile(r'^[a-zA-Z0-9_\-\[\]>=<\.]+$') for p in pkgs: if not safe.match(p): raise HTTPException(400, f'Invalid package name: {p}') try: async with _background_job(timeout_s=30.0): pass # slot acquisito — pip gira fuori dal semaphore (subprocess indipendente) except asyncio.TimeoutError: raise HTTPException(429, 'Server occupato con altri job pesanti. Riprova tra 30s.') # CHUNKED-PIP: installa in batch da 3 pacchetti — previene OOM su Railway free. # Ogni batch ha timeout 45s indipendente: un batch lento non blocca i successivi. # pip cache su /data/pip-cache (persistente tra restart HF Space / Railway). _BATCH = 3 _BATCH_TIMEOUT = 45 _PIP_CACHE = '/data/pip-cache' results_ok: list[str] = [] results_fail: list[str] = [] all_stdout: list[str] = [] all_stderr: list[str] = [] for i in range(0, len(pkgs), _BATCH): batch = pkgs[i:i + _BATCH] _pip_cmd = _get_venv_pip_cmd(batch) # Aggiungi cache dir persistente per ridurre re-download tra restart _pip_cmd = _pip_cmd + ['--cache-dir', _PIP_CACHE] try: proc = await asyncio.create_subprocess_exec( *_pip_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, preexec_fn=_install_resource_limits, # INSTALL-RLIMIT: AS 2.5GB, no CPU limit ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=_BATCH_TIMEOUT) out = stdout.decode('utf-8', errors='replace')[:2000] err = stderr.decode('utf-8', errors='replace')[:1000] all_stdout.append(out) all_stderr.append(err) # OOM kill detection _oom = _detect_oom_kill(proc.returncode) if _oom: all_stderr.append(_oom) results_fail.extend(batch) elif proc.returncode == 0: results_ok.extend(batch) else: results_fail.extend(batch) except asyncio.TimeoutError: try: _killpg(proc) # P40-A: kill process group pip rimasto vivo except Exception: pass all_stderr.append(f'⚠️ Timeout {_BATCH_TIMEOUT}s su batch: {batch}') results_fail.extend(batch) except Exception as e: all_stderr.append(f'Errore batch {batch}: {e}') results_fail.extend(batch) return { 'installed': results_ok, 'failed': results_fail, 'stdout': '\n'.join(all_stdout)[:4000], 'stderr': '\n'.join(all_stderr)[:2000], 'exit_code': 0 if not results_fail else 1, 'chunked': True, 'total_batches': (len(pkgs) + _BATCH - 1) // _BATCH, } @router.post('/api/agent/fix') async def llm_fix_code( req: FixRequest, role: AuthRole = Depends(require_role(AuthRole.MACHINE)), # GAP-AUTH-FIX: era aperto ): """ LLM fix chain: Groq → OpenRouter free models. S-fix A7: rimosso anthropic/claude-3-haiku (a pagamento). """ prompt = ( f"Fix this {req.lang} code. Return ONLY the corrected code, no explanation.\n\n" f"## Code\n```{req.lang}\n{req.code[:3000]}\n```\n\n" f"## Error\n```\n{req.error[:1000]}\n```\n\n" f"## Fixed code (ONLY code, no markdown fences):\n" ) def _strip_fences(text: str) -> str: if text.startswith('```'): lines = text.split('\n') return '\n'.join(lines[1:-1]) if lines[-1].strip() == '```' else '\n'.join(lines[1:]) return text async def _call_openai_compat(base_url: str, api_key: str, model: str) -> str | None: import httpx as _httpx_fix payload = { 'model': model, 'max_tokens': 2000, 'messages': [{'role': 'user', 'content': prompt}], } async with _httpx_fix.AsyncClient(timeout=25) as _hc_fix: _r_fix = await _hc_fix.post( f'{base_url}/chat/completions', json=payload, headers={ 'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json', 'HTTP-Referer': 'https://github.com/Baida98/AI', 'X-Title': 'Agente AI Debug Loop', }, ) resp = _r_fix.json() # S750-GAP-F: guard — provider può ritornare {"error":...} senza "choices" _choices = resp.get('choices') or [] if not _choices: return None return (_choices[0].get('message', {}).get('content') or '').strip() or None _FIX_CHAIN = [] groq_key = os.getenv('GROQ_API_KEY', '') if groq_key: _FIX_CHAIN.append(('https://api.groq.com/openai/v1', groq_key, 'llama-3.3-70b-versatile')) _FIX_CHAIN.append(('https://api.groq.com/openai/v1', groq_key, 'llama-3.1-8b-instant')) or_key = os.getenv('OPENROUTER_API_KEY', '') if or_key: for m in [ 'meta-llama/llama-3.1-8b-instruct:free', 'mistralai/mistral-7b-instruct:free', 'qwen/qwen-2.5-coder-7b-instruct:free', ]: _FIX_CHAIN.append(('https://openrouter.ai/api/v1', or_key, m)) if not _FIX_CHAIN: return {'fixed_code': None, 'error': 'Nessun provider disponibile (GROQ_API_KEY o OPENROUTER_API_KEY richiesti)'} last_err = '' for (base_url, api_key, model) in _FIX_CHAIN: try: raw = await _call_openai_compat(base_url, api_key, model) if raw: return {'fixed_code': _strip_fences(raw), 'provider': model} except Exception as e: last_err = str(e) continue return {'fixed_code': None, 'error': f'Tutti i provider falliti. Ultimo: {last_err[:300]}'} # S588: 200→300 # ── GAP-2: /api/exec/tool — dispatcher generico per tool calls dal frontend ──── # Il frontend (toolExecutor.ts) chiama questo endpoint per tool come scaffold_project # che non hanno un endpoint dedicato. Il dispatcher risolve il nome tool nel TOOL_REGISTRY # e invoca la funzione associata (_fn) con gli args forniti. class ToolDispatchRequest(BaseModel): tool: str args: dict = {} @router.post('/api/exec/tool') async def exec_tool_dispatch(req: ToolDispatchRequest, request: Request): """GAP-2: dispatcher generico — risolve tool nel TOOL_REGISTRY e chiama _fn.""" _internal_token = os.getenv('INTERNAL_TOKEN', '') if _internal_token and request.headers.get('X-Internal-Token') != _internal_token: raise HTTPException(401, 'Unauthorized') try: from tools.registry import TOOL_REGISTRY # import locale — evita circular import except ImportError as _ie: return {'ok': False, 'error': f'TOOL_REGISTRY non disponibile: {_ie}'} tool_def = TOOL_REGISTRY.get(req.tool) if not tool_def: available = ', '.join(list(TOOL_REGISTRY.keys())[:20]) return {'ok': False, 'error': f"Tool '{req.tool}' non trovato. Disponibili: {available}"} _fn = tool_def.get('_fn') if not _fn: return {'ok': False, 'error': f"Tool '{req.tool}' non ha handler (_fn) — non eseguibile via dispatcher"} try: import asyncio as _asyncio if _asyncio.iscoroutinefunction(_fn): result = await _fn(**req.args) else: result = _fn(**req.args) return {'ok': True, 'tool': req.tool, 'result': result} except TypeError as _te: # Parametri sbagliati — mostra la firma corretta import inspect as _inspect _sig = str(_inspect.signature(_fn)) return {'ok': False, 'error': f"Parametri non validi per '{req.tool}'{_sig}: {str(_te)[:200]}"} except Exception as _e: _logger.exception('[exec/tool] errore in %s', req.tool) return {'ok': False, 'error': str(_e)[:500]}