Seth
Claude Sonnet 5
Chat Agent Skills: real YAML, templates/, virtual file browser, sandboxed script execution
14acdb1 | """ | |
| 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) | |