js_sandbox.py

"""backend/js_sandbox.py — P20-JS-SANDBOX: Secure JS/TS execution.

Security layers (in order):
  1. Blocklist pre-filter (_JS_BLOCKED_RE) — blocks obvious CJS dangerous patterns
  2. Deno sandbox (primary, if available): OS-level deny-all for fs/net/env/subprocess
  3. Node vm fallback: vm.runInContext in isolated Context (no require/process/fs)
     + --disallow-code-generation-from-strings

Deno detection: $DENO_PATH env var, then /data/deno/bin/deno, /usr/local/bin/deno,
/home/user/.deno/bin/deno, /root/.deno/bin/deno, then PATH lookup.

Why Deno beats blocklist:
  - Blocklist bypassed via dynamic require, obfuscation, prototype chain attacks
  - Deno: OS-level deny-all — every permission must be explicitly granted with --allow-*
  - Node vm: JS-level isolation — sandbox Context has no require/process/fs globals
"""
import os, asyncio, re, logging
from typing import Optional

_logger = logging.getLogger("api.js_sandbox")

# ── Deno detection ─────────────────────────────────────────────────────────
_DENO_CANDIDATES = [
    os.environ.get("DENO_PATH", ""),
    "/data/deno/bin/deno",
    "/usr/local/bin/deno",
    "/home/user/.deno/bin/deno",
    "/root/.deno/bin/deno",
    "deno",  # PATH lookup
]


def _find_deno() -> Optional[str]:
    """Return path to Deno binary if available, else None."""
    import shutil
    for candidate in _DENO_CANDIDATES:
        if not candidate:
            continue
        if os.path.isabs(candidate):
            if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
                return candidate
        else:
            found = shutil.which(candidate)
            if found:
                return found
    return None


_DENO_BIN: Optional[str] = _find_deno()
_logger.info("[js_sandbox] Deno: %s", _DENO_BIN or "NOT FOUND — using Node vm fallback")

# ── JS/TS blocklist (defense-in-depth — belt, Deno/vm are suspenders) ─────
# Kept even when Deno is available: O(1) pre-filter before any subprocess spawn.
_JS_BLOCKED_RE = re.compile(
    r'\brequire\s*\(\s*[\'"]child_process[\'"]\s*\)'
    r'|\bprocess\.env\b'
    r'|\bfs\s*\.\s*(read|write|unlink|rm)'
    r'|\bexecSync\b|\bspawnSync\b'
    r'|\brequire\s*\(\s*[\'"](?:net|http|https|dgram|cluster|worker_threads)[\'"]'
    r'\s*\)',
    re.IGNORECASE,
)

# ── Node vm runner script (written to tmpdir per-call) ─────────────────────
# Reads user code from argv[2]. vm.createContext: no require/process/fs globals.
_NODE_VM_RUNNER = r"""'use strict';
const vm = require('vm');
const fs = require('fs');
const code = fs.readFileSync(process.argv[2], 'utf8');
const sandbox = {
  console: {
    log:   (...a) => process.stdout.write(a.map(String).join(' ') + '\n'),
    error: (...a) => process.stderr.write(a.map(String).join(' ') + '\n'),
    warn:  (...a) => process.stderr.write('[warn] ' + a.map(String).join(' ') + '\n'),
    info:  (...a) => process.stdout.write('[info] ' + a.map(String).join(' ') + '\n'),
  },
  Math, JSON, Date, Array, Object, String, Number, Boolean, RegExp,
  Error, TypeError, RangeError, SyntaxError, ReferenceError,
  Map, Set, WeakMap, WeakSet, Symbol, BigInt, Promise,
  parseInt, parseFloat, isNaN, isFinite, encodeURIComponent, decodeURIComponent,
};
const ctx = vm.createContext(sandbox);
try {
  vm.runInContext(code, ctx, { timeout: 10000, filename: 'sandbox.js' });
} catch (e) {
  process.stderr.write(String(e) + '\n');
  process.exit(1);
}
"""


async def run_js_sandbox(
    code: str,
    lang: str,
    tmpdir: str,
    timeout: float = 12.0,
    preexec_fn=None,
    safe_env: Optional[dict] = None,
) -> dict:
    """Execute JS/TS code in a sandbox. Returns dict with stdout, stderr, returncode.

    Security: blocklist pre-filter (belt) → Deno deny-all sandbox (primary)
              → Node vm.createContext (fallback).
    TypeScript: supported natively by Deno only; Node vm fallback refuses TS.
    """
    lang_norm = lang.lower()
    is_ts = lang_norm in ("typescript", "ts")

    # Belt: fast pre-filter — catches obvious CJS/process/fs dangerous patterns
    bad = _JS_BLOCKED_RE.search(code)
    if bad:
        return {
            "stdout": "",
            "stderr": f"js_blocked: pattern '{bad.group()}' non permesso nel sandbox",
            "returncode": 1,
        }

    if _DENO_BIN:
        return await _run_with_deno(code, is_ts, tmpdir, timeout, preexec_fn, safe_env or {})

    if is_ts:
        return {
            "stdout": "",
            "stderr": (
                "TypeScript sandbox: Deno non disponibile — TS non supportato nel fallback Node vm. "
                "Installa Deno (https://deno.land) o invia codice in JavaScript."
            ),
            "returncode": 1,
        }
    return await _run_with_node_vm(code, tmpdir, timeout, preexec_fn, safe_env or {})


async def _run_with_deno(
    code: str, is_ts: bool, tmpdir: str,
    timeout: float, preexec_fn, env: dict,
) -> dict:
    """Primary sandbox: Deno run — default deny-all (no --allow-* flags passed).

    Deno 1.x+: omitting all --allow-* flags = deny-all for fs/net/env/subprocess.
    --no-prompt: never prompt interactively for permissions (would hang).
    """
    ext = ".ts" if is_ts else ".js"
    fname = os.path.join(tmpdir, f"sandbox{ext}")
    with open(fname, "w") as fh:
        fh.write(code)

    # No --allow-* = deny-all by default (all Deno versions).
    cmd = [_DENO_BIN, "run", "--no-prompt", fname]

    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=tmpdir,
            env=env,
            preexec_fn=preexec_fn,
        )
        try:
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        except asyncio.TimeoutError:
            try:
                proc.kill()
            except Exception:
                pass
            await proc.wait()
            return {"stdout": "", "stderr": f"\u26a0\ufe0f Timeout {timeout:.0f}s (Deno sandbox)", "returncode": -1}
        return {
            "stdout": stdout.decode("utf-8", errors="replace")[:8000],
            "stderr": stderr.decode("utf-8", errors="replace")[:4000],
            "returncode": proc.returncode,
        }
    except Exception as exc:
        _logger.warning("[js_sandbox/deno] spawn error: %s", exc)
        return {"stdout": "", "stderr": str(exc)[:300], "returncode": 1}


async def _run_with_node_vm(
    code: str, tmpdir: str,
    timeout: float, preexec_fn, env: dict,
) -> dict:
    """Fallback sandbox: Node.js vm.createContext — no require/process/fs in Context.

    Not an OS-level sandbox (prototype chain escapes exist in theory), but combined
    with the blocklist pre-filter provides solid defense-in-depth for most threats.
    """
    code_file = os.path.join(tmpdir, "user_code.js")
    runner_file = os.path.join(tmpdir, "_vm_runner.js")
    with open(code_file, "w") as fh:
        fh.write(code)
    with open(runner_file, "w") as fh:
        fh.write(_NODE_VM_RUNNER)

    cmd = [
        "node",
        "--disallow-code-generation-from-strings",
        runner_file,
        code_file,
    ]
    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=tmpdir,
            env=env,
            preexec_fn=preexec_fn,
        )
        try:
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        except asyncio.TimeoutError:
            try:
                proc.kill()
            except Exception:
                pass
            await proc.wait()
            return {"stdout": "", "stderr": f"\u26a0\ufe0f Timeout {timeout:.0f}s (Node vm sandbox)", "returncode": -1}
        return {
            "stdout": stdout.decode("utf-8", errors="replace")[:8000],
            "stderr": stderr.decode("utf-8", errors="replace")[:4000],
            "returncode": proc.returncode,
        }
    except Exception as exc:
        _logger.warning("[js_sandbox/node-vm] spawn error: %s", exc)
        return {"stdout": "", "stderr": str(exc)[:300], "returncode": 1}