| """ |
| Admin-only sandboxed execution for Qualifier skill scripts. |
| |
| Near-identical to chat_agent_script_runner.py (same isolation boundary: fixed minimal env |
| allowlist, wall-clock timeout, no OS-level sandbox beyond that). Kept as its own module rather than |
| imported, for the same "separate copy per agent" reasoning as the rest of the qualifier_*.py files |
| — this is reachable only from POST /api/qualifier/skills/{id}/scripts/{index}/run, guarded by |
| require_tenant_admin, and never by the background qualification run itself. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import shutil |
| import subprocess |
| import sys |
| import tempfile |
| import time |
| from pathlib import Path |
| from typing import Any, Dict |
|
|
| MAX_OUTPUT_CHARS = 20_000 |
| DEFAULT_TIMEOUT_SECONDS = 10 |
|
|
| _SUPPORTED_LANGUAGES = {"python", "py", ""} |
|
|
|
|
| def run_script_sandboxed(script: Dict[str, Any], timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS) -> Dict[str, Any]: |
| """Run one skill script in an isolated subprocess and return its result.""" |
| language = (script.get("language") or "").strip().lower() |
| content = script.get("content") or "" |
|
|
| if language not in _SUPPORTED_LANGUAGES: |
| return { |
| "status": "unsupported", |
| "detail": f'Only Python scripts can be run in this environment (got "{language}").', |
| "stdout": "", |
| "stderr": "", |
| "duration_ms": 0, |
| } |
| if not content.strip(): |
| return {"status": "error", "detail": "Script has no content.", "stdout": "", "stderr": "", "duration_ms": 0} |
|
|
| tmpdir = tempfile.mkdtemp(prefix="qualifier_script_") |
| started = time.monotonic() |
| try: |
| script_path = Path(tmpdir) / "script.py" |
| script_path.write_text(content) |
|
|
| minimal_env = { |
| "PATH": "/usr/bin:/bin:/usr/local/bin", |
| "LANG": "C.UTF-8", |
| "HOME": tmpdir, |
| "PYTHONDONTWRITEBYTECODE": "1", |
| } |
|
|
| proc = subprocess.run( |
| [sys.executable, str(script_path)], |
| cwd=tmpdir, |
| env=minimal_env, |
| timeout=timeout_seconds, |
| capture_output=True, |
| text=True, |
| ) |
| duration_ms = int((time.monotonic() - started) * 1000) |
| return { |
| "status": "ok" if proc.returncode == 0 else "error", |
| "detail": "" if proc.returncode == 0 else f"Exited with code {proc.returncode}", |
| "stdout": proc.stdout[:MAX_OUTPUT_CHARS], |
| "stderr": proc.stderr[:MAX_OUTPUT_CHARS], |
| "duration_ms": duration_ms, |
| } |
| except subprocess.TimeoutExpired as e: |
| duration_ms = int((time.monotonic() - started) * 1000) |
| stdout = e.stdout if isinstance(e.stdout, str) else (e.stdout or b"").decode("utf-8", "replace") |
| stderr = e.stderr if isinstance(e.stderr, str) else (e.stderr or b"").decode("utf-8", "replace") |
| return { |
| "status": "timeout", |
| "detail": f"Script did not finish within {timeout_seconds}s and was killed.", |
| "stdout": stdout[:MAX_OUTPUT_CHARS], |
| "stderr": stderr[:MAX_OUTPUT_CHARS], |
| "duration_ms": duration_ms, |
| } |
| except Exception as e: |
| duration_ms = int((time.monotonic() - started) * 1000) |
| return {"status": "error", "detail": str(e)[:500], "stdout": "", "stderr": "", "duration_ms": duration_ms} |
| finally: |
| shutil.rmtree(tmpdir, ignore_errors=True) |
|
|
