diff --git a/bridge/mitmproxy/capture.py b/bridge/mitmproxy/capture.py index 4123060120..c5ca704f69 100644 --- a/bridge/mitmproxy/capture.py +++ b/bridge/mitmproxy/capture.py @@ -6,11 +6,13 @@ The containing mitmdump process and its configuration directory are REA-owned. import base64 import hashlib +import ipaddress import json import math import re import resource from pathlib import Path +from urllib.parse import unquote_to_bytes from mitmproxy import ctx, version from mitmproxy.io import tnetstring @@ -46,6 +48,8 @@ SCHEME_PREFIX = r"^[\x00-\x20]*([a-z][a-z0-9+.-]*):" NETWORK_PREFIX = r"^[\x00-\x20]*[/\\]{2}" URL_IGNORABLE_CONTROLS = "\t\n\r" URL_CONTROL_TRANSLATION = {ord(char): None for char in URL_IGNORABLE_CONTROLS} +C0_CONTROL_OR_SPACE = "".join(chr(code) for code in range(0x21)) +FORBIDDEN_DOMAIN_CODE_POINTS = set(C0_CONTROL_OR_SPACE) | set("#%/:<>?@[\\]^|\x7f") def authority_start(text, base_scheme=None): @@ -54,11 +58,14 @@ def authority_start(text, base_scheme=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. + userinfo. File URLs never carry userinfo: an "@" in a file host makes + WHATWG reject the whole URL, so there are no credentials to remove. """ 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": + if name == "file" or (name is None and base_scheme == "file"): + return None + if name is not None and name in SPECIAL_SCHEMES: # A same-scheme Location without two slash-equivalent separators is # relative to the recorded request, so its @ belongs to the path. following = text[scheme.end():scheme.end() + 2] @@ -69,7 +76,7 @@ def authority_start(text, base_scheme=None): if scheme is not None: if text[scheme.end():scheme.end() + 2] != "//": return None - return scheme.end() + 2, name in SPECIAL_SCHEMES + return scheme.end() + 2, False # A non-special base treats backslashes as authority data and only # recognizes two forward slashes as a network reference. if base_scheme is not None and base_scheme not in SPECIAL_SCHEMES: @@ -79,13 +86,87 @@ def authority_start(text, base_scheme=None): if prefix is None: return None start = prefix.end() - if base_scheme == "file": - return start, True while start < len(text) and text[start] in "/\\": start += 1 return start, True +IPV4_DIGITS = {8: "01234567", 10: "0123456789", 16: "0123456789abcdefABCDEF"} + + +def ipv4_number(part): + """WHATWG IPv4 number parser: decimal, 0x hex or 0-prefixed octal, else None.""" + if part == "": + return None + radix, digits = 10, part + if part[:2] in ("0x", "0X"): + radix, digits = 16, part[2:] + elif len(part) > 1 and part[0] == "0": + radix, digits = 8, part[1:] + if digits == "": + return 0 + if any(char not in IPV4_DIGITS[radix] for char in digits): + return None + # Bound decimal text before int(): Python limits decimal conversions to + # 4300 digits, and anything past 10 significant digits exceeds IPv4. + significant = digits.lstrip("0") + if radix == 10 and len(significant) > 10: + return 2**32 + return int(digits, radix) + + +def domain_parses(domain): + """Apply WHATWG's ends-in-a-number check and IPv4 parser to a domain.""" + parts = domain.split(".") + if parts[-1] == "" and len(parts) > 1: + parts.pop() + last = parts[-1] + if not ((last != "" and last.isascii() and last.isdigit()) or ipv4_number(last) is not None): + return True + numbers = [ipv4_number(part) for part in parts] + if len(parts) > 4 or any(number is None for number in numbers): + return False + return all(number <= 255 for number in numbers[:-1]) and numbers[-1] < 256 ** (5 - len(numbers)) + + +def special_host_parses(host_port): + """Whether WHATWG could accept a special authority's host and port. + + A rejected host fails the whole URL, so it has no credentials to remove. + This mirrors the forbidden-code-point, empty-host, IPv4, IPv6 and port + rules for ASCII hosts. UTS #46 mapping is not modeled: a non-ASCII host + may map to an accepted one (fullwidth digits, ideographic full stops, + ignored code points), so it is treated as accepted and still redacts. + """ + if host_port.startswith("["): + host_end = host_port.find("]") + 1 + literal = host_port[1 : host_end - 1] + if host_end == 0 or "%" in literal: + return False + try: + ipaddress.IPv6Address(literal) + except ValueError: + return False + else: + colon = host_port.find(":") + host_end = len(host_port) if colon < 0 else colon + host = unquote_to_bytes(host_port[:host_end]).decode("utf-8", errors="replace") + # Mapping never removes ASCII, so these rejections hold either way. + if host == "" or any(char in FORBIDDEN_DOMAIN_CODE_POINTS for char in host): + return False + if host.isascii() and not domain_parses(host.lower()): + return False + port = host_port[host_end:] + if port == "": + return True + digits = port[1:] + if port[0] != ":" or not (digits == "" or (digits.isascii() and digits.isdigit())): + return False + # Bound the text first: int() refuses decimal strings over 4300 digits. + significant = digits.lstrip("0") + return len(significant) <= 5 and int(significant or "0") <= 65535 + + def without_userinfo(value, base_scheme=None): """Remove an authority's userinfo using the WHATWG authority boundary. @@ -96,7 +177,8 @@ def without_userinfo(value, base_scheme=None): text = value.decode("latin-1") if binary else value # WHATWG removes these controls before interpreting scheme/separators. # Keep that parse view separate from the exact source spelling and bytes. - parsed_text = text.translate(URL_CONTROL_TRANSLATION) + # It also drops trailing C0 controls and spaces; offsets are unchanged. + parsed_text = text.translate(URL_CONTROL_TRANSLATION).rstrip(C0_CONTROL_OR_SPACE) located = authority_start(parsed_text, base_scheme) if located is None: return value @@ -106,7 +188,7 @@ def without_userinfo(value, base_scheme=None): while end < len(parsed_text) and parsed_text[end] not in terminators: end += 1 at = parsed_text.rfind("@", start, end) - if at < start: + if at < start or (special and not special_host_parses(parsed_text[at + 1 : end])): return value parsed_offset = 0 raw_start = 0 diff --git a/bridge/mitmproxy/test_capture.py b/bridge/mitmproxy/test_capture.py index f0a1106b1b..3863312cb9 100644 --- a/bridge/mitmproxy/test_capture.py +++ b/bridge/mitmproxy/test_capture.py @@ -165,6 +165,97 @@ class TransportUserinfoTests(unittest.TestCase): 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"), + ("https://user@0x7f.1/", "https://0x7f.1/"), + ("https://user@1.2.3.4./", "https://1.2.3.4./"), + ("https://user@4294967295/", "https://4294967295/"), + ("https://user@1.2.3.08a/", "https://1.2.3.08a/"), + ("https://user@[::ffff:1.2.3.4]/", "https://[::ffff:1.2.3.4]/"), + ]: + with self.subTest(value=value): + self.assertEqual(capture.without_userinfo(value), expected) + + def test_unicode_hosts_that_may_map_to_accepted_hosts_remove_userinfo(self): + # UTS #46 maps these to 1.2.3.4; mapping is not modeled, so redact. + for host in ["%EF%BC%91.2.3.4", "1。2.3.4", "%C2%AD1.2.3.4", "１.2.3.4"]: + with self.subTest(host=host): + self.assertEqual( + capture.without_userinfo(f"https://user:pw@{host}/"), f"https://{host}/" + ) + text, sidecar, redactions = location(b"https://user:pw@%EF%BC%91.2.3.4/", b"https") + self.assertEqual(text, "https://%EF%BC%91.2.3.4/") + self.assertEqual(sidecar["state"], "redacted") + self.assertEqual(len(redactions), 1) + # ASCII rejections still hold for non-ASCII hosts. + self.assertEqual(capture.without_userinfo("https://user@１^/"), "https://user@１^/") + + def test_trailing_controls_and_spaces_do_not_hide_userinfo(self): + # WHATWG strips trailing C0 controls and spaces before parsing. + for suffix in [b" ", b"\x01", b"\x00 \x1f"]: + with self.subTest(suffix=suffix): + value = b"https://user:pw@example.test" + suffix + text, sidecar, redactions = location(value, b"https") + self.assertEqual(text, "https://example.test" + suffix.decode("latin-1")) + self.assertEqual(sidecar["state"], "redacted") + self.assertEqual(len(redactions), 1) + + def test_rejected_ip_literals_keep_their_source(self): + for value in [ + "https://user:pass@256.256.256.256/path", + "https://user@1.2.3.4.5/", + "https://user@1.2.3.09/", + "https://user@4294967296/", + "https://user@a.0x/", + "https://user@1..2/", + "https://user@1." + "9" * 5000 + "/", + "https://user@[1::2::3]/", + "https://user@[::1%25eth0]/", + ]: + with self.subTest(value=value[:40]): + self.assertEqual(capture.without_userinfo(value), value) + + def test_overlong_ports_are_compared_without_numeric_overflow(self): + # Python 3.11+ refuses int() on decimal strings over 4300 digits. + accepted = "https://user@host:" + "0" * 5000 + "80/x" + self.assertEqual(capture.without_userinfo(accepted), "https://host:" + "0" * 5000 + "80/x") + rejected = "https://user@host:" + "9" * 5000 + "/x" + self.assertEqual(capture.without_userinfo(rejected), rejected) + 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")