BIG BIG BIG BIG BIG UPDATES

Co-authored-by: Copilot <[email protected]>

backend/app/services/mitmproxy/options.py

@@ -0,0 +1,156 @@
+import asyncio
+import re
+import shutil
+from dataclasses import dataclass
+from typing import Any
+
+import yaml
+
+RESERVED_OPTIONS = {"mode", "listen_host", "listen_port", "scripts"}
+OPTIONS_DOCS_URL = "https://docs.mitmproxy.org/stable/concepts/options/"
+
+
+@dataclass(slots=True)
+class MitmOption:
+    name: str
+    value_type: str
+    default: Any
+    description: str
+    choices: list[str]
+
+
+class MitmOptionError(Exception):
+    pass
+
+
+catalog_cache: dict[str, MitmOption] | None = None
+
+
+async def load_options() -> dict[str, MitmOption]:
+    global catalog_cache
+    if catalog_cache is None:
+        catalog_cache = await read_options()
+    return catalog_cache
+
+
+async def read_options(command: str = "mitmdump") -> dict[str, MitmOption]:
+    command_path = shutil.which(command)
+    if command_path is None:
+        raise MitmOptionError("mitmdump executable not found in PATH")
+
+    process = await asyncio.create_subprocess_exec(
+        command_path,
+        "--options",
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await process.communicate()
+    if process.returncode != 0:
+        message = stderr.decode("utf-8", errors="replace").strip()
+        raise MitmOptionError(message or "Unable to load mitmdump options")
+
+    text = stdout.decode("utf-8", errors="replace")
+    defaults = yaml.safe_load(text) or {}
+    return parse_options(text, defaults)
+
+
+def parse_options(text: str, defaults: dict[str, Any]) -> dict[str, MitmOption]:
+    comments: list[str] = []
+    options: dict[str, MitmOption] = {}
+
+    for line in text.splitlines():
+        stripped = line.strip()
+        if not stripped:
+            comments = []
+            continue
+        if stripped.startswith("#"):
+            comments.append(stripped[1:].strip())
+            continue
+        if not re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*:", stripped):
+            continue
+
+        name = stripped.split(":", 1)[0]
+        description = " ".join(comments)
+        options[name] = MitmOption(
+            name=name,
+            value_type=parse_type(description),
+            default=defaults.get(name),
+            description=description,
+            choices=parse_choices(description),
+        )
+        comments = []
+
+    return options
+
+
+def parse_type(description: str) -> str:
+    match = re.search(r"Type ([^.]+)\.", description)
+    if match is None:
+        return "unknown"
+    return match.group(1).strip()
+
+
+def parse_choices(description: str) -> list[str]:
+    match = re.search(r"Valid values are ([^.]+)\.", description)
+    if match is None:
+        return []
+    return re.findall(r"'([^']+)'", match.group(1))
+
+
+async def validate_options(options: dict[str, Any]) -> None:
+    catalog = await load_options()
+    errors: list[str] = []
+
+    for name, value in options.items():
+        option = catalog.get(name)
+        if option is None:
+            errors.append(f"Unknown mitmdump option: {name}")
+            continue
+        if name in RESERVED_OPTIONS:
+            errors.append(f"Option is managed by thePlug and cannot be set: {name}")
+            continue
+        message = validate_value(option, value)
+        if message:
+            errors.append(f"{name}: {message}")
+
+    if errors:
+        raise MitmOptionError("; ".join(errors))
+
+
+def validate_value(option: MitmOption, value: Any) -> str | None:
+    value_type = option.value_type
+    if option.choices and str(value) not in option.choices:
+        return f"expected one of {', '.join(option.choices)}"
+    if value is None:
+        if value_type.startswith("optional "):
+            return None
+        return "cannot be null"
+    if value_type == "bool" and not isinstance(value, bool):
+        return "expected bool"
+    if value_type == "int" and not isinstance(value, int):
+        return "expected int"
+    if value_type == "str" and not isinstance(value, str):
+        return "expected str"
+    if value_type == "sequence of str":
+        if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
+            return "expected list[str]"
+    if value_type == "optional str" and not isinstance(value, str):
+        return "expected str or null"
+    if value_type == "optional int" and not isinstance(value, int):
+        return "expected int or null"
+    return None
+
+
+def public_options(options: dict[str, MitmOption]) -> list[dict[str, Any]]:
+    return [
+        {
+            "name": option.name,
+            "type": option.value_type,
+            "default": option.default,
+            "description": option.description,
+            "choices": option.choices,
+            "reserved": option.name in RESERVED_OPTIONS,
+            "docs_url": f"{OPTIONS_DOCS_URL}#{option.name}",
+        }
+        for option in options.values()
+    ]