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() ]