175 lines
8.1 KiB
Python
175 lines
8.1 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_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()
|