thePlug/app/services/mitmproxy/validator.py

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,
        )