Compare commits

...
Author SHA1 Message Date
Kaoru0822andClaude Opus 5.5 110151e523 fix(capture): keep URLs that WHATWG rejects instead of redacting userinfo
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>
2026-10-14 00:26:29 +00:00
Kaoru0822 b93e585d68 Merge remote-tracking branch 'origin/main' into fix/issue-2671 2026-10-14 00:22:25 +00:00
morluto 929aa67336 chore(main): merge pinned CI checkpoint 2895a18 2026-10-14 03:12:47 +08:00
morluto 4e8475e965 test(ci): isolate alternate launcher process fixture 2026-10-14 03:02:01 +08:00
morluto 557148ed3f fix(ci): declare process inventory callback 2026-10-14 02:25:19 +08:00
morluto f839046776 chore(merge): refresh PR #2767 with latest main 2026-10-14 02:23:57 +08:00
Morluto 281c4b8ec5 test(ci): isolate Hopper process inventory fixtures 2026-10-14 02:15:33 +08:00
Morluto a0da18d7fa test(capture): cover empty URL userinfo markers 2026-10-14 02:14:51 +08:00
Morluto 2c6e074cac fix(capture): remove empty URL userinfo markers 2026-10-14 02:14:51 +08:00
morluto 29c871e51b chore(merge): refresh PR #2767 with main 2026-10-14 02:10:19 +08:00
REA MaintenanceandKaoru0822 c0018ca403 fix(ci): isolate Linux Hopper process fixture inventory 2026-10-14 01:23:30 +08:00
morlutoandKaoru0822 ad16d250e3 fix(android): make DEX descriptor pattern portable 2026-10-14 01:23:25 +08:00
Lucienandmorluto af2f55cda9 fix(comparison): recheck historical calls-only changes (#2223)
* fix(comparison): preserve incomplete function call evidence

Keep missing, unknown, partial and unresolved call relationships unknown in function comparisons and reconstruction claims. Aggregate IDA caller and callee coverage without overwriting uncertainty.

Validate with 144 focused tests, typecheck/lint, docs checks and test discovery. Real providers were not verified; the full headless IDA session suite requires unavailable Windows native controls.

* fix(comparison): recheck historical calls-only changes

---------

Co-authored-by: morluto <76467478+morluto@users.noreply.github.com>
2026-10-14 01:23:19 +08:00
morluto 4d5d1f5daa fix(capture): preserve file-relative path evidence after separators 2026-10-13 23:02:00 +08:00
morluto 0aff997b31 fix(capture): distinguish native paths and URL base authorities 2026-10-13 21:35:04 +08:00
morluto e8361b7715 fix(capture): parse URL controls and mixed network authorities 2026-10-13 20:31:54 +08:00
morluto 72e931f3a3 chore(ci): retain contributor updates alongside verification fixes 2026-10-13 20:21:10 +08:00
morluto 0266e6552b fix(capture): preserve same-scheme relative redirect paths 2026-10-13 20:19:00 +08:00
Kaoru0822 7d972a7e85 Merge remote-tracking branch 'origin/main' into fix/issue-2671 2026-10-13 12:04:14 +00:00
Kaoru0822andClaude Opus 5.5 e1fb40460d 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>
2026-10-13 11:25:05 +00:00
Kaoru0822andClaude Opus 5.5 553cd6e358 fix(capture): end special-scheme URL authorities at a backslash
The native mitmproxy adapter removed transport userinfo with
`^(...//)[^/?#]*@`, so the authority ran to the last "@" before "/", "?"
or "#". For `https://example.test\part@tail` it treated
`example.test\part` as userinfo, reported `https://tail` as the
Location, and withdrew the original bytes as a transport credential.
WHATWG URL parsing, used by the shared TypeScript owner
(recordedTransportUrl.ts) and by HAR captures, ends a special-scheme or
scheme-relative authority at a backslash, so that "@" is ordinary path
text.

Locate the authority with the same boundary rule before removing
userinfo. Bytes are handled as Latin-1 text so unrelated bytes stay
identical. Userinfo before the backslash is still removed and redacted.
Add source unit tests for the adapter projection to test:python.

Fixes #2671

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-13 11:06:25 +00:00
7 changed files with 368 additions and 15 deletions
+133 -12
View File
@@ -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)
+214
View File
@@ -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
View File
@@ -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>;
+2 -1
View File
@@ -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 = [];
+11
View File
@@ -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.