File size: 4,619 Bytes
14acdb1 | 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 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | """
Admin-only sandboxed execution for skill scripts.
This is NEVER reachable from the live public chat agent or any CORS-tagged public endpoint —
only from the admin-authenticated POST /api/chat-agent/skills/{id}/scripts/{index}/run route in
chat_agent_routes.py (guarded by require_tenant_admin, same as every other mutating skill route).
A visitor's chat message can never trigger this; only an authenticated admin explicitly clicking
"Run" in the Skills UI can.
What this isolation protects against: a script accidentally (or deliberately) reading this
container's real secrets (OPENAI_API_KEY, GOOGLE_CLIENT_SECRET, SESSION_SECRET, etc — the
subprocess env is a fixed minimal allowlist, never os.environ), and a hung/runaway script (a
wall-clock timeout kills it). What it does NOT protect against: a deliberately malicious script
run by the admin themselves — there is no OS-level sandbox (no network restriction, no filesystem
restriction beyond the script's own temp working directory, no cgroup/resource limits). That's an
acceptable boundary here because the admin already has equivalent-or-greater capability through
the rest of the admin panel. See SKILLS-SECURITY.md for the full writeup.
"""
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
# Empty string covers scripts saved before a language was set.
_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.
Only Python is actually executed — we can't guarantee any other interpreter exists in this
container, so any other `language` value returns status="unsupported" without attempting
execution (never silently no-ops as success).
"""
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="chat_agent_script_")
started = time.monotonic()
try:
script_path = Path(tmpdir) / "script.py"
script_path.write_text(content)
# Deliberately NOT os.environ — a fixed minimal allowlist so the script can't read
# OPENAI_API_KEY/GOOGLE_CLIENT_SECRET/SESSION_SECRET/etc, even by accident
# (e.g. `print(os.environ)`).
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)
|