69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
import re
|
|
from collections.abc import Sequence
|
|
from typing import NamedTuple
|
|
|
|
from mitmproxy import ctx
|
|
from mitmproxy import exceptions
|
|
from mitmproxy import flowfilter
|
|
from mitmproxy import http
|
|
from mitmproxy.utils.spec import parse_spec
|
|
|
|
|
|
class MapRemoteSpec(NamedTuple):
|
|
matches: flowfilter.TFilter
|
|
subject: str
|
|
replacement: str
|
|
|
|
|
|
def parse_map_remote_spec(option: str) -> MapRemoteSpec:
|
|
spec = MapRemoteSpec(*parse_spec(option))
|
|
|
|
try:
|
|
re.compile(spec.subject)
|
|
except re.error as e:
|
|
raise ValueError(f"Invalid regular expression {spec.subject!r} ({e})")
|
|
|
|
return spec
|
|
|
|
|
|
class MapRemote:
|
|
def __init__(self) -> None:
|
|
self.replacements: list[MapRemoteSpec] = []
|
|
|
|
def load(self, loader):
|
|
loader.add_option(
|
|
"map_remote",
|
|
Sequence[str],
|
|
[],
|
|
"""
|
|
Map remote resources to another remote URL using a pattern of the form
|
|
"[/flow-filter]/url-regex/replacement", where the separator can
|
|
be any character.
|
|
""",
|
|
)
|
|
|
|
def configure(self, updated):
|
|
if "map_remote" in updated:
|
|
self.replacements = []
|
|
for option in ctx.options.map_remote:
|
|
try:
|
|
spec = parse_map_remote_spec(option)
|
|
except ValueError as e:
|
|
raise exceptions.OptionsError(
|
|
f"Cannot parse map_remote option {option}: {e}"
|
|
) from e
|
|
|
|
self.replacements.append(spec)
|
|
|
|
def request(self, flow: http.HTTPFlow) -> None:
|
|
if flow.response or flow.error or not flow.live:
|
|
return
|
|
for spec in self.replacements:
|
|
if spec.matches(flow):
|
|
url = flow.request.pretty_url
|
|
new_url = re.sub(spec.subject, spec.replacement, url)
|
|
# this is a bit messy: setting .url also updates the host header,
|
|
# so we really only do that if the replacement affected the URL.
|
|
if url != new_url:
|
|
flow.request.url = new_url # type: ignore
|