Compare commits
21
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
110151e523 | ||
|
|
b93e585d68 | ||
|
|
929aa67336 | ||
|
|
4e8475e965 | ||
|
|
557148ed3f | ||
|
|
f839046776 | ||
|
|
281c4b8ec5 | ||
|
|
a0da18d7fa | ||
|
|
2c6e074cac | ||
|
|
29c871e51b | ||
|
|
c0018ca403 | ||
|
|
ad16d250e3 | ||
|
|
af2f55cda9 | ||
|
|
4d5d1f5daa | ||
|
|
0aff997b31 | ||
|
|
e8361b7715 | ||
|
|
72e931f3a3 | ||
|
|
0266e6552b | ||
|
|
7d972a7e85 | ||
|
|
e1fb40460d | ||
|
|
553cd6e358 |
+133
-12
@@ -11,6 +11,7 @@ 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
|
||||
@@ -41,6 +42,113 @@ class BoundedReader:
|
||||
return self.handle.read(count)
|
||||
|
||||
|
||||
SPECIAL_SCHEMES = {"ftp", "file", "http", "https", "ws", "wss"}
|
||||
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}
|
||||
FORBIDDEN_DOMAIN_CODE_POINTS = {chr(code) for code in range(0x21)} | set("#%/:<>?@[\\]^|\x7f")
|
||||
|
||||
|
||||
def authority_start(text, base_scheme=None):
|
||||
"""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 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 == "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]
|
||||
if name == base_scheme and (len(following) < 2 or any(char not in "/\\" for char in following)):
|
||||
return None
|
||||
start = scheme.end()
|
||||
else:
|
||||
if scheme is not None:
|
||||
if text[scheme.end():scheme.end() + 2] != "//":
|
||||
return None
|
||||
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:
|
||||
prefix = re.match(r"^[\x00-\x20]*//", text)
|
||||
return None if prefix is None else (prefix.end(), False)
|
||||
prefix = re.match(NETWORK_PREFIX, text)
|
||||
if prefix is None:
|
||||
return None
|
||||
start = prefix.end()
|
||||
while start < len(text) and text[start] in "/\\":
|
||||
start += 1
|
||||
return start, True
|
||||
|
||||
|
||||
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 and port rules; IDNA,
|
||||
IPv4 range and IPv6 syntax failures are not modeled and still redact.
|
||||
"""
|
||||
if host_port.startswith("["):
|
||||
host_end = host_port.find("]") + 1
|
||||
if host_end == 0:
|
||||
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")
|
||||
if host == "" or any(char in FORBIDDEN_DOMAIN_CODE_POINTS for char in host):
|
||||
return False
|
||||
port = host_port[host_end:]
|
||||
if port == "":
|
||||
return True
|
||||
digits = port[1:]
|
||||
return port[0] == ":" and (digits == "" or (digits.isascii() and digits.isdigit() and int(digits) <= 65535))
|
||||
|
||||
|
||||
def without_userinfo(value, base_scheme=None):
|
||||
"""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.
|
||||
"""
|
||||
binary = isinstance(value, bytes)
|
||||
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)
|
||||
located = authority_start(parsed_text, base_scheme)
|
||||
if located is None:
|
||||
return value
|
||||
start, special = located
|
||||
terminators = "/?#\\" if special else "/?#"
|
||||
end = start
|
||||
while end < len(parsed_text) and parsed_text[end] not in terminators:
|
||||
end += 1
|
||||
at = parsed_text.rfind("@", start, end)
|
||||
if at < start or (special and not special_host_parses(parsed_text[at + 1 : end])):
|
||||
return value
|
||||
parsed_offset = 0
|
||||
raw_start = 0
|
||||
for raw_offset, char in enumerate(text):
|
||||
if char in URL_IGNORABLE_CONTROLS:
|
||||
continue
|
||||
if parsed_offset == start:
|
||||
raw_start = raw_offset
|
||||
if parsed_offset == at:
|
||||
safe = text[:raw_start] + text[raw_offset + 1 :]
|
||||
return safe.encode("latin-1") if binary else safe
|
||||
parsed_offset += 1
|
||||
raise CaptureFailure("decoder", "Normalized URL authority could not be bound to its source.")
|
||||
|
||||
|
||||
def project_record(state, secrets):
|
||||
binaries, numbers, redactions = [], [], []
|
||||
byte_secrets = []
|
||||
@@ -58,16 +166,23 @@ def project_record(state, secrets):
|
||||
return None
|
||||
return text
|
||||
|
||||
def transport_value(value, path, transport_url):
|
||||
def transport_value(value, path, transport_url, base_scheme):
|
||||
if not (transport_url or re.fullmatch(r"/(?:backup/)*request/(?:path|authority)", path)):
|
||||
return value, False
|
||||
binary = isinstance(value, bytes)
|
||||
pattern = rb"^([\x00-\x20]*(?:[a-z][a-z0-9+.-]*:)?//)[^/?#]*@" if binary else r"^([\x00-\x20]*(?:[a-z][a-z0-9+.-]*:)?//)[^/?#]*@"
|
||||
replacement = rb"\1" if binary else r"\1"
|
||||
separator = b"@" if binary else "@"
|
||||
safe_url = re.sub(pattern, replacement, value, flags=re.I)
|
||||
if path.endswith("/authority") and separator in safe_url:
|
||||
safe_url = safe_url.rsplit(separator, 1)[1]
|
||||
if transport_url:
|
||||
safe_url = without_userinfo(value, base_scheme)
|
||||
elif path.endswith("/authority"):
|
||||
safe_url = value.rsplit(separator, 1)[-1]
|
||||
else:
|
||||
# Native request.path is origin-form unless it explicitly carries
|
||||
# a scheme (absolute-form). Leading separators are path evidence,
|
||||
# not a header URL reference's network authority.
|
||||
text = value.decode("latin-1") if binary else value
|
||||
safe_url = without_userinfo(value) if re.match(
|
||||
SCHEME_PREFIX, text.translate(URL_CONTROL_TRANSLATION), flags=re.I
|
||||
) else value
|
||||
changed = safe_url != value
|
||||
if changed:
|
||||
redactions.append({"pointer": path, "reason": "transport-credential"})
|
||||
@@ -80,7 +195,7 @@ def project_record(state, secrets):
|
||||
del numbers[before[1]:]
|
||||
del redactions[before[2]:]
|
||||
|
||||
def visit(value, path, depth=0, credential=False, transport_url=False):
|
||||
def visit(value, path, depth=0, credential=False, transport_url=False, base_scheme=None):
|
||||
if depth > 64:
|
||||
raise CaptureFailure("input-limit", "Native capture exceeds the 64-level evidence nesting budget.", path)
|
||||
if credential:
|
||||
@@ -90,7 +205,7 @@ def project_record(state, secrets):
|
||||
binaries.append({"pointer": path, "representation": "producer-bytes", "state": "redacted", "content_base64": None, "bytes": None, "sha256": None})
|
||||
return None
|
||||
if isinstance(value, bytes):
|
||||
safe_bytes, credential_url = transport_value(value, path, transport_url)
|
||||
safe_bytes, credential_url = transport_value(value, path, transport_url, base_scheme)
|
||||
content_base64 = base64.b64encode(value).decode("ascii")
|
||||
digest = hashlib.sha256(value).hexdigest()
|
||||
hidden = any(secret in value or secret in safe_bytes for secret in byte_secrets) or any(secret in content_base64 or secret in digest or secret in str(len(value)) for secret in secrets)
|
||||
@@ -107,7 +222,7 @@ def project_record(state, secrets):
|
||||
return None
|
||||
return safe_text
|
||||
if isinstance(value, str):
|
||||
safe_text, _ = transport_value(value, path, transport_url)
|
||||
safe_text, _ = transport_value(value, path, transport_url, base_scheme)
|
||||
return redact(safe_text, path, value)
|
||||
if isinstance(value, bool) or value is None:
|
||||
return value
|
||||
@@ -125,10 +240,16 @@ def project_record(state, secrets):
|
||||
for index, field in enumerate(value):
|
||||
if not isinstance(field, (list, tuple)) or len(field) != 2 or not isinstance(field[0], bytes):
|
||||
raise CaptureFailure("format", "Native header/trailer is not an ordered byte pair.", pointer(path, index))
|
||||
result.append([visit(field[0], pointer(pointer(path, index), 0), depth + 2), visit(field[1], pointer(pointer(path, index), 1), depth + 2, field[0].lower() in CREDENTIALS, field[0].lower() in {b"location", b"referer", b"origin"})])
|
||||
result.append([visit(field[0], pointer(pointer(path, index), 0), depth + 2), visit(field[1], pointer(pointer(path, index), 1), depth + 2, field[0].lower() in CREDENTIALS, field[0].lower() in {b"location", b"referer", b"origin"}, base_scheme if field[0].lower() == b"location" else None)])
|
||||
return result
|
||||
return [visit(item, pointer(path, index), depth + 1) for index, item in enumerate(value)]
|
||||
return [visit(item, pointer(path, index), depth + 1, base_scheme=base_scheme) for index, item in enumerate(value)]
|
||||
if isinstance(value, dict):
|
||||
if path == "" or re.fullmatch(r"/backup(?:/backup)*", path):
|
||||
request = value.get("request")
|
||||
scheme = request.get("scheme") if isinstance(request, dict) else None
|
||||
if isinstance(scheme, bytes):
|
||||
scheme = scheme.decode("latin-1")
|
||||
base_scheme = scheme.lower() if isinstance(scheme, str) and re.fullmatch(r"[a-z][a-z0-9+.-]*", scheme, flags=re.I) else None
|
||||
result = {}
|
||||
for key, item in value.items():
|
||||
if not isinstance(key, str):
|
||||
@@ -139,7 +260,7 @@ def project_record(state, secrets):
|
||||
validate_omitted(item, pointer(path, key), depth + 1)
|
||||
redactions.append({"pointer": path, "reason": "explicit-sensitive-value", "scope": "property-name"})
|
||||
continue
|
||||
result[key] = visit(item, pointer(path, key), depth + 1)
|
||||
result[key] = visit(item, pointer(path, key), depth + 1, base_scheme=base_scheme)
|
||||
return result
|
||||
raise CaptureFailure("decoder", "Native parser returned an unexpected value type: " + type(value).__name__ + ".", path)
|
||||
|
||||
|
||||
@@ -0,0 +1,214 @@
|
||||
"""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()
|
||||
+1
-1
@@ -138,7 +138,7 @@
|
||||
"test": "npm run test:with-build",
|
||||
"test:with-build": "npm run test:prepare && vitest run",
|
||||
"test:run": "vitest run",
|
||||
"test:python": "python3 -m unittest bridge/native/test_rea_lldb_tracer.py -v && python3 -m unittest discover -s experiments/analysis-efficiency -p 'test_*.py' -v",
|
||||
"test:python": "python3 -m unittest bridge/native/test_rea_lldb_tracer.py bridge/mitmproxy/test_capture.py -v && python3 -m unittest discover -s experiments/analysis-efficiency -p 'test_*.py' -v",
|
||||
"test:local": "node scripts/run-development-tests.mjs local",
|
||||
"test:fast": "node scripts/ci/test-group.mjs source",
|
||||
"test:built": "node scripts/ci/test-group.mjs built",
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
export function assertCleanLinuxHopperRuntime(
|
||||
launcherPaths?: readonly string[],
|
||||
procRoot?: string,
|
||||
listProcesses?: () => Promise<{ stdout: string }>,
|
||||
): Promise<void>;
|
||||
|
||||
@@ -17,8 +17,9 @@ export const assertCleanLinuxHopperRuntime = async (
|
||||
: [process.env.HOPPER_LAUNCHER_PATH]),
|
||||
],
|
||||
procRoot = "/proc",
|
||||
listProcesses = () => execFileAsync("ps", ["-ax", "-o", "command="]),
|
||||
) => {
|
||||
const { stdout } = await execFileAsync("ps", ["-ax", "-o", "command="]);
|
||||
const { stdout } = await listProcesses();
|
||||
if (/\/opt\/hopper\/bin\/Hopper|\bXvfb\b|hopper-demo-x11\.py/u.test(stdout))
|
||||
throw new Error("Clean Linux runner retained a Hopper demo process");
|
||||
const launchers = [];
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
import { expect, it } from "vitest";
|
||||
|
||||
import { parseRecordedTransportUrl } from "./recordedTransportUrl.js";
|
||||
|
||||
it("recognizes an empty userinfo marker at the authority start", () => {
|
||||
expect(parseRecordedTransportUrl("https://@host")).toMatchObject({
|
||||
origin: "https://host",
|
||||
path: "/",
|
||||
credentials: { start: 8, end: 9 },
|
||||
});
|
||||
});
|
||||
@@ -34,6 +34,10 @@ it
|
||||
const procRoot = join(directory, "proc");
|
||||
await mkdir(procRoot);
|
||||
const alternateMode = mode === "alternate launcher";
|
||||
let orphanPid: number | undefined;
|
||||
const isolatedProcessList = async () => ({
|
||||
stdout: alternateMode || orphanPid === undefined ? "" : "Xvfb",
|
||||
});
|
||||
if (alternateMode) {
|
||||
await copyFile("/usr/bin/sleep", launcher);
|
||||
await chmod(launcher, 0o700);
|
||||
@@ -47,6 +51,7 @@ it
|
||||
await assertCleanLinuxHopperRuntime(
|
||||
alternateMode ? [launcher] : [],
|
||||
procRoot,
|
||||
isolatedProcessList,
|
||||
);
|
||||
const command = alternateMode ? alternate : process.execPath;
|
||||
const arguments_ = alternateMode ? ["30"] : ["-e", childCode];
|
||||
@@ -69,7 +74,6 @@ it
|
||||
const parentPid = parent.child.pid;
|
||||
if (parentPid === undefined)
|
||||
throw new Error("Fixture parent did not spawn");
|
||||
let orphanPid: number | undefined;
|
||||
try {
|
||||
const { stdout } = await parent;
|
||||
const { pid } = JSON.parse(stdout) as { pid: number };
|
||||
@@ -115,6 +119,7 @@ it
|
||||
assertCleanLinuxHopperRuntime(
|
||||
alternateMode ? [launcher] : [],
|
||||
procRoot,
|
||||
isolatedProcessList,
|
||||
),
|
||||
).rejects.toThrow("Clean Linux runner retained a Hopper demo process");
|
||||
// Observation must not terminate the fixture, even though it resembles Xvfb.
|
||||
|
||||
Reference in New Issue
Block a user