157 lines
4.7 KiB
Python
157 lines
4.7 KiB
Python
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()
|
|
]
|