File size: 3,411 Bytes
005df22 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """
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)
|