feat(proxy): implement proxy code validation and error handling

app/api/proxy/router.py

@@ -10,6 +10,7 @@ from db.proxy import (
 )
 from fastapi import APIRouter, Depends, HTTPException, status
 from models.proxy import ProxyCreateInput, ProxyUpdateInput
+from utils.proxy.validator import ProxyCodeValidationError, validate_proxy_code
 
 router = APIRouter(
     prefix="/proxy",
@@ -18,6 +19,16 @@ router = APIRouter(
 )
 
 
+async def _validate_code_or_raise(code: str) -> None:
+    try:
+        await validate_proxy_code(code)
+    except ProxyCodeValidationError as error:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=error.to_http_detail(),
+        ) from error
+
+
 @router.get("")
 async def list_proxies_handler():
     return await list_proxy_documents()
@@ -43,6 +54,8 @@ async def create_proxy_handler(payload: ProxyCreateInput):
             detail=f"Proxy already exists on port {payload.proxy_port}",
         )
 
+    await _validate_code_or_raise(payload.code)
+
     return await create_proxy(
         proxy_port=payload.proxy_port,
         name=payload.name,
@@ -53,6 +66,9 @@ async def create_proxy_handler(payload: ProxyCreateInput):
 
 @router.put("/{proxy_port}")
 async def edit_proxy(proxy_port: int, payload: ProxyUpdateInput):
+    if payload.code is not None:
+        await _validate_code_or_raise(payload.code)
+
     updated_proxy = await update_proxy(proxy_port, payload)
     if updated_proxy is None:
         raise HTTPException(

app/utils/proxy/validator.py

@@ -0,0 +1,164 @@
+import asyncio
+import json
+import os
+import shutil
+import tempfile
+from typing import Any
+
+TARGET_RUFF_CODES = ("F821", "F822", "F823", "E999")
+
+
+class ProxyCodeValidationError(Exception):
+    def __init__(
+        self,
+        message: str,
+        *,
+        errors: list[dict[str, Any]] | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.message = message
+        self.errors = errors or []
+
+    def to_http_detail(self) -> dict[str, Any]:
+        return {
+            "message": self.message,
+            "errors": self.errors,
+        }
+
+
+def _validate_python_syntax(code: str) -> None:
+    try:
+        compile(code, "<proxy_code>", "exec")
+    except SyntaxError as error:
+        line = error.lineno if error.lineno is not None else 0
+        column = error.offset if error.offset is not None else 0
+        detail = {
+            "code": "COMPILE",
+            "message": error.msg,
+            "location": {"row": line, "column": column},
+        }
+        raise ProxyCodeValidationError(
+            "Proxy code has invalid Python syntax",
+            errors=[detail],
+        ) from error
+
+
+def _get_ruff_executable() -> list[str]:
+    ruff_path = shutil.which("ruff")
+    if ruff_path:
+        return [ruff_path]
+
+    uv_path = shutil.which("uv")
+    if uv_path:
+        return [uv_path, "run", "ruff"]
+
+    raise ProxyCodeValidationError("Unable to run ruff: executable not found")
+
+
+def _parse_ruff_output(output: str) -> list[dict[str, Any]]:
+    if not output.strip():
+        return []
+
+    try:
+        parsed = json.loads(output)
+    except json.JSONDecodeError as error:
+        raise ProxyCodeValidationError("Unable to parse ruff JSON output") from error
+
+    if not isinstance(parsed, list):
+        raise ProxyCodeValidationError("Unexpected ruff output format")
+
+    errors: list[dict[str, Any]] = []
+    for item in parsed:
+        if not isinstance(item, dict):
+            continue
+        code = item.get("code")
+        if code not in TARGET_RUFF_CODES:
+            continue
+
+        message = item.get("message")
+        raw_location = item.get("location")
+        location = raw_location if isinstance(raw_location, dict) else {}
+        row = location.get("row", 0)
+        column = location.get("column", 0)
+
+        errors.append(
+            {
+                "code": code,
+                "message": str(message) if message is not None else "",
+                "location": {"row": row, "column": column},
+            }
+        )
+
+    return errors
+
+
+async def _run_ruff_check(code: str) -> list[dict[str, Any]]:
+    command = _get_ruff_executable()
+
+    file_path = ""
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            suffix=".py",
+            encoding="utf-8",
+            delete=False,
+        ) as handle:
+            handle.write(code)
+            file_path = handle.name
+
+        process = await asyncio.create_subprocess_exec(
+            *command,
+            "check",
+            "--output-format=json",
+            "--select",
+            ",".join(TARGET_RUFF_CODES),
+            file_path,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+    finally:
+        if file_path:
+            try:
+                os.unlink(file_path)
+            except FileNotFoundError:
+                pass
+
+    stdout_text = stdout.decode("utf-8", errors="replace")
+    stderr_text = stderr.decode("utf-8", errors="replace")
+
+    if process.returncode not in (0, 1):
+        raise ProxyCodeValidationError(
+            "Ruff check failed to execute",
+            errors=[
+                {
+                    "code": "RUFF",
+                    "message": stderr_text.strip() or "Unknown ruff failure",
+                    "location": {"row": 0, "column": 0},
+                }
+            ],
+        )
+
+    if stderr_text.strip() and process.returncode == 0:
+        raise ProxyCodeValidationError(
+            "Ruff emitted unexpected stderr output",
+            errors=[
+                {
+                    "code": "RUFF",
+                    "message": stderr_text.strip(),
+                    "location": {"row": 0, "column": 0},
+                }
+            ],
+        )
+
+    return _parse_ruff_output(stdout_text)
+
+
+async def validate_proxy_code(code: str) -> None:
+    _validate_python_syntax(code)
+    errors = await _run_ruff_check(code)
+    if errors:
+        raise ProxyCodeValidationError(
+            "Proxy code contains ruff violations",
+            errors=errors,
+        )