Kaoru0822andClaude Opus 5.5
110151e523
File URLs cannot carry userinfo (an "@" in a file host fails the parse), and a special authority whose host or port WHATWG rejects fails the whole URL. Neither has credentials, so the bridge now keeps their source text and byte sidecar. Checked against Node's URL over 30,000 spellings: no under-redaction, over-redaction 20,540 -> 5,865 (remaining: empty userinfo, IPv4 range, IDNA, non-special and baseless references). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
215 lines
9.9 KiB
Python
215 lines
9.9 KiB
Python
"""Source-owned tests for the mitmproxy adapter's record projection."""
|
|
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
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, scheme=None):
|
|
state = {"response": {"headers": [[b"Location", value]]}}
|
|
if scheme is not None:
|
|
state["request"] = {"scheme": scheme}
|
|
record = capture.project_record(state, [])
|
|
sidecar = next(
|
|
field for field in record["binary_fields"] if field["pointer"] == "/response/headers/0/1"
|
|
)
|
|
return record["reported"]["response"]["headers"][0][1], sidecar, record["redactions"]
|
|
|
|
|
|
class TransportUserinfoTests(unittest.TestCase):
|
|
def test_ignorable_controls_cannot_hide_authority_credentials(self):
|
|
for control in [b"\t", b"\n", b"\r"]:
|
|
for value in [b"https://" + control + b"/user:pass@host/path", b"ht" + control + b"tps://user:" + control + b"pass@host/path"]:
|
|
with self.subTest(value=value):
|
|
text, sidecar, redactions = location(value, b"https")
|
|
self.assertNotIn("user", text)
|
|
self.assertNotIn("pass", text)
|
|
self.assertEqual(sidecar["state"], "redacted")
|
|
self.assertEqual(len(redactions), 1)
|
|
|
|
def test_mixed_network_separators_remove_credentials(self):
|
|
for prefix in [b"\\\\", b"/\\", b"\\/"]:
|
|
with self.subTest(prefix=prefix):
|
|
text, sidecar, redactions = location(prefix + b"user:pass@host/path", b"https")
|
|
self.assertEqual(text, prefix.decode("ascii") + "host/path")
|
|
self.assertEqual(sidecar["state"], "redacted")
|
|
self.assertEqual(len(redactions), 1)
|
|
|
|
def test_non_special_base_keeps_backslashes_inside_network_authorities(self):
|
|
text, sidecar, redactions = location(b"//user\\name:pass@host/x", b"foo")
|
|
self.assertEqual(text, "//host/x")
|
|
self.assertEqual(sidecar["state"], "redacted")
|
|
self.assertEqual(len(redactions), 1)
|
|
text, sidecar, redactions = location(b"/\\invoice@customer", b"foo")
|
|
self.assertEqual(text, "/\\invoice@customer")
|
|
self.assertEqual(sidecar["state"], "retained")
|
|
self.assertEqual(redactions, [])
|
|
|
|
def test_native_request_paths_preserve_origin_form_and_redact_absolute_form(self):
|
|
for value in [b"/\\invoice@customer", b"//invoice@customer", b"\\\\invoice@customer"]:
|
|
with self.subTest(value=value):
|
|
record = capture.project_record({"request": {"path": value}}, [])
|
|
self.assertEqual(record["reported"]["request"]["path"], value.decode("ascii"))
|
|
self.assertEqual(record["binary_fields"][0]["state"], "retained")
|
|
self.assertEqual(record["redactions"], [])
|
|
record = capture.project_record({"request": {"path": b"https://user:pass@host/x"}}, [])
|
|
self.assertEqual(record["reported"]["request"]["path"], "https://host/x")
|
|
self.assertEqual(record["binary_fields"][0]["state"], "redacted")
|
|
|
|
def test_same_scheme_locations_preserve_relative_path_text_and_bytes(self):
|
|
for value in [b"https:invoice@customer", b"https:/invoice@customer", b"https:\\invoice@customer"]:
|
|
with self.subTest(value=value):
|
|
text, sidecar, redactions = location(value, b"https")
|
|
self.assertEqual(text, value.decode("ascii"))
|
|
self.assertEqual(sidecar["state"], "retained")
|
|
self.assertEqual(redactions, [])
|
|
|
|
def test_location_authorities_still_remove_credentials_with_a_base_scheme(self):
|
|
for value, scheme, expected in [
|
|
(b"https://user:pass@host/path", b"https", "https://host/path"),
|
|
(b"https:\\\\user:pass@host/path", b"https", "https:\\\\host/path"),
|
|
(b"https:user:pass@host/path", b"http", "https:host/path"),
|
|
]:
|
|
with self.subTest(value=value, scheme=scheme):
|
|
text, sidecar, redactions = location(value, scheme)
|
|
self.assertEqual(text, expected)
|
|
self.assertEqual(sidecar["state"], "redacted")
|
|
self.assertEqual(len(redactions), 1)
|
|
|
|
def test_backup_location_uses_its_own_request_scheme(self):
|
|
record = capture.project_record({
|
|
"request": {"scheme": b"http"},
|
|
"backup": {
|
|
"request": {"scheme": b"https"},
|
|
"response": {"headers": [[b"Location", b"https:invoice@customer"]]},
|
|
},
|
|
}, [])
|
|
self.assertEqual(record["reported"]["backup"]["response"]["headers"][0][1], "https:invoice@customer")
|
|
self.assertEqual(record["redactions"], [])
|
|
|
|
def test_backslash_ends_a_special_scheme_authority(self):
|
|
text, sidecar, redactions = location(b"https://example.test\\part@tail")
|
|
self.assertEqual(text, "https://example.test\\part@tail")
|
|
self.assertEqual(sidecar["state"], "retained")
|
|
self.assertIsNotNone(sidecar["content_base64"])
|
|
self.assertEqual(redactions, [])
|
|
|
|
def test_userinfo_before_a_backslash_is_still_removed(self):
|
|
text, sidecar, redactions = location(b"https://user:pw@example.test\\path")
|
|
self.assertEqual(text, "https://example.test\\path")
|
|
self.assertEqual(sidecar["state"], "redacted")
|
|
self.assertEqual(
|
|
redactions,
|
|
[{"pointer": "/response/headers/0/1", "reason": "transport-credential"}],
|
|
)
|
|
|
|
def test_ordinary_userinfo_is_removed(self):
|
|
text, _sidecar, _redactions = location(b"https://user:pw@example.test/a@b")
|
|
self.assertEqual(text, "https://example.test/a@b")
|
|
|
|
def test_empty_userinfo_marker_is_removed(self):
|
|
self.assertEqual(
|
|
capture.without_userinfo("https://@host"), "https://host"
|
|
)
|
|
|
|
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_base_keeps_paths_after_the_fixed_network_boundary(self):
|
|
for value in [b"///u@h", b"////u@h"]:
|
|
with self.subTest(value=value):
|
|
text, sidecar, redactions = location(value, b"file")
|
|
self.assertEqual(text, value.decode("ascii"))
|
|
self.assertEqual(sidecar["state"], "retained")
|
|
self.assertIsNotNone(sidecar["content_base64"])
|
|
self.assertEqual(redactions, [])
|
|
|
|
def test_file_urls_keep_path_text(self):
|
|
self.assertEqual(capture.without_userinfo("file:///a@b"), "file:///a@b")
|
|
|
|
def test_file_urls_reject_userinfo_and_keep_their_source(self):
|
|
# WHATWG rejects an "@" in a file host, so nothing parses as credentials.
|
|
self.assertEqual(
|
|
capture.without_userinfo("file://user:pass@host/x"), "file://user:pass@host/x"
|
|
)
|
|
for value in [b"\\\\user:pass@host/x", b"/\\user:pass@host/x", b"//user:pass@host/x"]:
|
|
with self.subTest(value=value):
|
|
text, sidecar, redactions = location(value, b"file")
|
|
self.assertEqual(text, value.decode("ascii"))
|
|
self.assertEqual(sidecar["state"], "retained")
|
|
self.assertEqual(redactions, [])
|
|
|
|
def test_rejected_special_hosts_keep_their_source(self):
|
|
# A rejected host or port fails the whole URL, so it has no userinfo.
|
|
for value in [
|
|
"https:user@%",
|
|
"https://user@",
|
|
"https://user:pass@h^st/x",
|
|
"https://user@ho st/x",
|
|
"https://user@host:99999/x",
|
|
"https://user@host:12a",
|
|
"https://user@[::1",
|
|
]:
|
|
with self.subTest(value=value):
|
|
self.assertEqual(capture.without_userinfo(value), value)
|
|
text, sidecar, redactions = location(b"https://user:pass@h^st/x", b"https")
|
|
self.assertEqual(text, "https://user:pass@h^st/x")
|
|
self.assertEqual(sidecar["state"], "retained")
|
|
self.assertEqual(redactions, [])
|
|
|
|
def test_accepted_special_hosts_still_remove_userinfo(self):
|
|
for value, expected in [
|
|
("https:user@%41", "https:%41"),
|
|
("https://user@[::1]:8443/x", "https://[::1]:8443/x"),
|
|
("https://user@host:/x", "https://host:/x"),
|
|
("wss://user:pass@host:65535", "wss://host:65535"),
|
|
]:
|
|
with self.subTest(value=value):
|
|
self.assertEqual(capture.without_userinfo(value), expected)
|
|
|
|
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")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|