diff --git a/bridge/mitmproxy/capture.py b/bridge/mitmproxy/capture.py index 00487d68b..5104130ab 100644 --- a/bridge/mitmproxy/capture.py +++ b/bridge/mitmproxy/capture.py @@ -42,31 +42,54 @@ class BoundedReader: SPECIAL_SCHEMES = {"ftp", "file", "http", "https", "ws", "wss"} +SCHEME_PREFIX = r"^[\x00-\x20]*([a-z][a-z0-9+.-]*):" URL_PREFIX = r"^([\x00-\x20]*(?:([a-z][a-z0-9+.-]*):)?//)" +def authority_start(text): + """Return where an authority begins and whether it is special, or None. + + Follows WHATWG URL parsing as src/domain/recordedTransportUrl.ts does: + special schemes (and scheme-relative references) skip every leading "/" + or "\\" before the authority, so `https://\\user:pass@host` still has + userinfo. File URLs carry no userinfo and keep their "//" boundary. + """ + scheme = re.match(SCHEME_PREFIX, text, flags=re.I) + name = scheme.group(1).lower() if scheme is not None else None + if name is not None and name in SPECIAL_SCHEMES and name != "file": + start = scheme.end() + else: + prefix = re.match(URL_PREFIX, text, flags=re.I) + if prefix is None: + return None + if prefix.group(2) is not None: + return prefix.end(), prefix.group(2).lower() in SPECIAL_SCHEMES + start = prefix.end() + while start < len(text) and text[start] in "/\\": + start += 1 + return start, True + + def without_userinfo(value): """Remove an authority's userinfo using the WHATWG authority boundary. - Special schemes (and scheme-relative references) also end the authority - at a backslash, so later path text containing "@" is not userinfo. This - matches src/domain/recordedTransportUrl.ts. + Special schemes and scheme-relative references also end the authority + at a backslash, so later path text containing "@" is not userinfo. """ binary = isinstance(value, bytes) text = value.decode("latin-1") if binary else value - prefix = re.match(URL_PREFIX, text, flags=re.I) - if prefix is None: + located = authority_start(text) + if located is None: return value - scheme = prefix.group(2) - special = scheme is None or scheme.lower() in SPECIAL_SCHEMES + start, special = located terminators = "/?#\\" if special else "/?#" - end = prefix.end() + end = start while end < len(text) and text[end] not in terminators: end += 1 - at = text.rfind("@", prefix.end(), end) + at = text.rfind("@", start, end) if at < 0: return value - safe = text[: prefix.end()] + text[at + 1 :] + safe = text[:start] + text[at + 1 :] return safe.encode("latin-1") if binary else safe diff --git a/bridge/mitmproxy/test_capture.py b/bridge/mitmproxy/test_capture.py index fe9f4f7af..8b1217e59 100644 --- a/bridge/mitmproxy/test_capture.py +++ b/bridge/mitmproxy/test_capture.py @@ -5,20 +5,31 @@ import os import sys import types import unittest +from unittest import mock -fake_mitmproxy = sys.modules.setdefault("mitmproxy", types.ModuleType("mitmproxy")) -fake_mitmproxy.ctx = types.SimpleNamespace() -fake_mitmproxy.version = types.SimpleNamespace(VERSION="12.2.3") -fake_io = sys.modules.setdefault("mitmproxy.io", types.ModuleType("mitmproxy.io")) -fake_io.tnetstring = sys.modules.setdefault( - "mitmproxy.io.tnetstring", types.ModuleType("mitmproxy.io.tnetstring") -) -spec = importlib.util.spec_from_file_location( - "rea_mitmproxy_capture", os.path.join(os.path.dirname(__file__), "capture.py") -) -capture = importlib.util.module_from_spec(spec) -spec.loader.exec_module(capture) +def load_capture(): + """Load the adapter with fresh mitmproxy stand-ins scoped to its import.""" + fake_mitmproxy = types.ModuleType("mitmproxy") + fake_mitmproxy.ctx = types.SimpleNamespace() + fake_mitmproxy.version = types.SimpleNamespace(VERSION="12.2.3") + fake_io = types.ModuleType("mitmproxy.io") + fake_io.tnetstring = types.ModuleType("mitmproxy.io.tnetstring") + stand_ins = { + "mitmproxy": fake_mitmproxy, + "mitmproxy.io": fake_io, + "mitmproxy.io.tnetstring": fake_io.tnetstring, + } + spec = importlib.util.spec_from_file_location( + "rea_mitmproxy_capture", os.path.join(os.path.dirname(__file__), "capture.py") + ) + module = importlib.util.module_from_spec(spec) + with mock.patch.dict(sys.modules, stand_ins): + spec.loader.exec_module(module) + return module + + +capture = load_capture() def location(value): @@ -50,6 +61,23 @@ class TransportUserinfoTests(unittest.TestCase): text, _sidecar, _redactions = location(b"https://user:pw@example.test/a@b") self.assertEqual(text, "https://example.test/a@b") + def test_special_schemes_skip_leading_slashes_before_userinfo(self): + for value, expected in [ + (b"https://\\user:pass@host/path", b"https://\\host/path"), + (b"https:\\\\user:pass@host/path", b"https:\\\\host/path"), + (b"https:user:pass@host", b"https:host"), + (b"//\\user:pass@host", b"//\\host"), + ]: + with self.subTest(value=value): + self.assertEqual(capture.without_userinfo(value), expected) + text, sidecar, redactions = location(b"https://\\user:pass@host/path") + self.assertEqual(text, "https://\\host/path") + self.assertEqual(sidecar["state"], "redacted") + self.assertEqual(len(redactions), 1) + + def test_file_urls_keep_path_text(self): + self.assertEqual(capture.without_userinfo("file:///a@b"), "file:///a@b") + def test_non_special_schemes_keep_backslashes_in_the_authority(self): self.assertEqual(capture.without_userinfo("foo://a\\b@host/x"), "foo://host/x") self.assertEqual(capture.without_userinfo(b"//a\\b@host"), b"//a\\b@host")