fix(capture): skip special-scheme leading slashes before userinfo
Ending the authority at a backslash exposed a credential: for `https://\user:pass@host/path`, WHATWG parsing skips every leading "/" or "\" after a special scheme (and in scheme-relative references), so `user:pass` is userinfo, but the new boundary stopped at the first backslash and found no "@". The same applied to `https:\\user:pass@host` and `https:user:pass@host`. Start the authority after that whole slash run for non-file special schemes and scheme-relative references. File URLs keep their "//" boundary because they carry no userinfo. Load the adapter in tests with fresh mitmproxy stand-ins scoped to its import instead of leaving them in sys.modules. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
553cd6e358
commit
e1fb40460d
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
Loading…
Reference in New Issue