spb/uqo-chat

1"""Client for the sandbox-runner service (POST /run). Never executes code in-process."""23from __future__ import annotations45import re6from dataclasses import dataclass, field7from typing import Any89import httpx1011from app.core.config import Settings1213# Defence in depth — the real isolation lives in sandbox-runner.14HOSTILE_PATTERNS = [15    r"\bsubprocess\b", r"\bsocket\b", r"os\.system", r"os\.popen", r"\bctypes\b",16    r"__import__\s*\(\s*['\"]os['\"]", r"\bshutil\.rmtree\b", r"\bpty\b", r"\bmultiprocessing\b",17    r"\burllib\b", r"\brequests\b", r"\bhttpx\b", r"\bsignal\.", r"os\.fork", r"os\.exec",18    r"\bimportlib\b", r"open\(\s*['\"]/(etc|Users|home|var|proc)", r"\beval\s*\(", r"\bexec\s*\(",19]202122@dataclass23class RunResult:24    stdout: str = ""25    stderr: str = ""26    exit_code: int = 027    duration_ms: int = 028    files_out: list[dict[str, Any]] = field(default_factory=list)29    truncated: bool = False30    error: str | None = None313233def prefilter(code: str) -> str | None:34    for pat in HOSTILE_PATTERNS:35        if re.search(pat, code):36            return f"Code refusé : motif non autorisé « {pat.strip(chr(92) + 'b')} »."37    return None383940class SandboxClient:41    def __init__(self, settings: Settings) -> None:42        self.s = settings4344    async def run(self, code: str, files_in: list[dict[str, str]] | None = None,45                  timeout_s: int | None = None) -> RunResult:46        refusal = prefilter(code)47        if refusal:48            return RunResult(stderr=refusal, exit_code=126, error=refusal)49        timeout = timeout_s or self.s.SANDBOX_TIMEOUT_S50        headers = {}51        if self.s.SANDBOX_TOKEN.get_secret_value():52            headers["Authorization"] = f"Bearer {self.s.SANDBOX_TOKEN.get_secret_value()}"53        try:54            async with httpx.AsyncClient(timeout=timeout + 15) as c:55                r = await c.post(f"{self.s.SANDBOX_URL}/run", headers=headers,56                                 json={"code": code, "files_in": files_in or [],57                                       "timeout_s": timeout})58        except httpx.HTTPError as exc:59            return RunResult(error=f"Sandbox injoignable : {exc}", exit_code=127)60        if r.status_code >= 400:61            return RunResult(error=f"Sandbox HTTP {r.status_code}: {r.text[:300]}", exit_code=127)62        d = r.json()63        return RunResult(64            stdout=d.get("stdout", ""), stderr=d.get("stderr", ""),65            exit_code=int(d.get("exit_code", 0)), duration_ms=int(d.get("duration_ms", 0)),66            files_out=d.get("files_out", []), truncated=bool(d.get("truncated", False)),67        )68