chore: merge main into PR #2801
This commit is contained in:
morluto
@@ -0,0 +1,10 @@
|
||||
# Exercised runtime graph: CPython 3.14.7, Linux x86_64 guest.
|
||||
aioquic==1.2.0
|
||||
attrs==26.1.0
|
||||
certifi==2026.7.22
|
||||
cffi==2.1.1
|
||||
cryptography==50.0.2
|
||||
pycparser==3.11
|
||||
pylsqpack==0.3.24
|
||||
pyOpenSSL==26.4.0
|
||||
service-identity==26.1.0
|
||||
@@ -0,0 +1,416 @@
|
||||
"""Real loopback HTTP/3 laboratory using aioquic and caller-supplied mitmdump.
|
||||
|
||||
Run inside a disposable Linux guest. All native records are written by
|
||||
mitmproxy's real save addon, not constructed fixtures. Endpoint stream IDs are
|
||||
observations scoped to that endpoint; no MITM stream mapping is claimed.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import signal
|
||||
import ssl
|
||||
import subprocess
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import aioquic
|
||||
from aioquic.asyncio import QuicConnectionProtocol, connect, serve
|
||||
from aioquic.h3.connection import H3Connection
|
||||
from aioquic.h3.events import DataReceived, HeadersReceived
|
||||
from aioquic.quic.configuration import QuicConfiguration
|
||||
from aioquic.quic.events import ConnectionTerminated, HandshakeCompleted, StreamReset
|
||||
from aioquic.quic.packet import QuicProtocolVersion
|
||||
|
||||
|
||||
def binary(value):
|
||||
return {"base64": base64.b64encode(value).decode("ascii"), "bytes": len(value),
|
||||
"sha256": hashlib.sha256(value).hexdigest()}
|
||||
|
||||
|
||||
def pairs(headers):
|
||||
return [[binary(name), binary(value)] for name, value in headers]
|
||||
|
||||
|
||||
class Journal:
|
||||
def __init__(self, root):
|
||||
self.root = root
|
||||
self.events = []
|
||||
self.requests = {}
|
||||
self.responses = {}
|
||||
self.tasks = set()
|
||||
self.children = []
|
||||
self.writers = []
|
||||
|
||||
def emit(self, event, **data):
|
||||
value = {"event": event, "monotonic": time.monotonic(), **data}
|
||||
self.events.append(value)
|
||||
with (self.root / "endpoints.jsonl").open("a") as handle:
|
||||
handle.write(json.dumps(value) + "\n")
|
||||
|
||||
def task(self, coroutine):
|
||||
task = asyncio.create_task(coroutine)
|
||||
self.tasks.add(task)
|
||||
return task
|
||||
|
||||
|
||||
|
||||
class Endpoint(QuicConnectionProtocol):
|
||||
def __init__(self, *args, journal, endpoint, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.journal, self.endpoint = journal, endpoint
|
||||
self.http = H3Connection(self._quic)
|
||||
self.messages = {}
|
||||
self.waiters = {}
|
||||
self.concurrent = []
|
||||
self.identity = f"{endpoint}-{len(journal.events)}"
|
||||
|
||||
def quic_event_received(self, event):
|
||||
if isinstance(event, HandshakeCompleted):
|
||||
# This is the negotiated runtime value in pinned aioquic 1.2.0,
|
||||
# not configuration.supported_versions or mitmproxy's QUICv1 label.
|
||||
self.journal.emit("handshake", endpoint=self.endpoint, connection=self.identity,
|
||||
alpn=event.alpn_protocol, quic_version=int(self._quic._version))
|
||||
if isinstance(event, (StreamReset, ConnectionTerminated)):
|
||||
self.journal.emit(type(event).__name__, endpoint=self.endpoint,
|
||||
connection=self.identity, error_code=event.error_code,
|
||||
stream_id=getattr(event, "stream_id", None),
|
||||
reason=getattr(event, "reason_phrase", None))
|
||||
targets = self.waiters.values() if isinstance(event, ConnectionTerminated) else [self.waiters.get(event.stream_id)]
|
||||
for waiter in targets:
|
||||
if waiter is not None and not waiter.done():
|
||||
waiter.set_exception(ConnectionError(type(event).__name__))
|
||||
for item in self.http.handle_event(event):
|
||||
if isinstance(item, (HeadersReceived, DataReceived)):
|
||||
message = self.messages.setdefault(item.stream_id, {"headers": [], "body": bytearray()})
|
||||
if isinstance(item, HeadersReceived):
|
||||
message["headers"].extend(item.headers)
|
||||
else:
|
||||
message["body"].extend(item.data)
|
||||
if item.stream_ended:
|
||||
self.completed(item.stream_id, message)
|
||||
|
||||
def completed(self, stream_id, message):
|
||||
body = bytes(message["body"])
|
||||
if self.endpoint == "client":
|
||||
self.journal.emit("response-received", endpoint="client", connection=self.identity,
|
||||
stream_id=stream_id, headers=pairs(message["headers"]), body=binary(body))
|
||||
self.waiters[stream_id].set_result(message)
|
||||
return
|
||||
headers = dict(message["headers"])
|
||||
case = headers.get(b"x-lab-case", b"curl").decode("ascii")
|
||||
self.journal.requests[case] = {"headers": pairs(message["headers"]), "body": binary(body)}
|
||||
self.journal.emit("request-received", endpoint="origin", connection=self.identity,
|
||||
stream_id=stream_id, case=case, **self.journal.requests[case])
|
||||
if case.startswith("parallel-"):
|
||||
self.concurrent.append((case, stream_id, body))
|
||||
if len(self.concurrent) == 3:
|
||||
self.journal.task(self.answer_parallel())
|
||||
else:
|
||||
self.answer(case, stream_id, b"\x00\xfeanswer:" + case.encode() + b":" + body[::-1])
|
||||
|
||||
async def answer_parallel(self):
|
||||
for case, stream_id, body in reversed(self.concurrent):
|
||||
self.answer(case, stream_id, b"\x00\xfeanswer:" + case.encode() + b":" + body[::-1])
|
||||
await asyncio.sleep(0.15)
|
||||
|
||||
def answer(self, case, stream_id, body):
|
||||
headers = [(b":status", b"200"), (b"content-type", b"application/octet-stream"),
|
||||
(b"content-length", str(len(body)).encode()),
|
||||
(b"x-lab-case", case.encode()), (b"x-duplicate", b"one"), (b"x-duplicate", b"two")]
|
||||
self.journal.responses[case] = {"headers": pairs(headers), "body": binary(body), "finished": True}
|
||||
self.journal.emit("response-sent", endpoint="origin", connection=self.identity,
|
||||
stream_id=stream_id, case=case, **self.journal.responses[case])
|
||||
self.http.send_headers(stream_id, headers)
|
||||
self.http.send_data(stream_id, body, end_stream=True)
|
||||
self.transmit()
|
||||
|
||||
def request(self, case, body):
|
||||
stream_id = self._quic.get_next_available_stream_id()
|
||||
headers = [(b":method", b"POST"), (b":scheme", b"https"),
|
||||
(b":authority", b"lab.test:18443"), (b":path", b"/same?token=ordinary&x=%2F"),
|
||||
(b"x-lab-case", case.encode()), (b"content-type", b"application/octet-stream"),
|
||||
(b"content-length", str(len(body)).encode()),
|
||||
(b"x-duplicate", b"first"), (b"x-duplicate", b"second")]
|
||||
self.waiters[stream_id] = asyncio.get_running_loop().create_future()
|
||||
self.journal.emit("request-sent", endpoint="client", connection=self.identity,
|
||||
stream_id=stream_id, case=case, headers=pairs(headers), body=binary(body), finished=True)
|
||||
self.http.send_headers(stream_id, headers)
|
||||
self.http.send_data(stream_id, body, end_stream=True)
|
||||
self.transmit()
|
||||
return stream_id, self.waiters[stream_id]
|
||||
|
||||
|
||||
|
||||
class Proxy:
|
||||
def __init__(self, args, root, name, http3=True):
|
||||
self.args, self.root, self.name, self.http3 = args, root, name, http3
|
||||
self.process = None
|
||||
self.log = None
|
||||
|
||||
async def start(self):
|
||||
self.log = (self.root / f"{self.name}.proxy.log").open("wb")
|
||||
command = [self.args.mitmdump, "--mode", "reverse:https://lab.test:19443@127.0.0.1:18443",
|
||||
"--set", f"confdir={self.root / 'proxy-config'}", "--set", "connection_strategy=lazy",
|
||||
"--set", "keep_host_header=true", "--set", "store_streamed_bodies=false",
|
||||
"--set", f"rea_lab_log={self.root / (self.name + '.producer.jsonl')}",
|
||||
"-w", str(self.root / (self.name + '.mitm')),
|
||||
"-s", str(Path(__file__).with_name("observe-mitmproxy-http3.py"))]
|
||||
command.extend(["--set", f"ssl_verify_upstream_trusted_ca={self.root / 'origin-ca.pem'}"])
|
||||
if not self.http3:
|
||||
command.extend(["--set", "http3=false"])
|
||||
self.process = subprocess.Popen(command, stdout=self.log, stderr=subprocess.STDOUT,
|
||||
start_new_session=True)
|
||||
lease = {"runId": os.environ["REA_PROCESS_RUN_ID"], "leaderPid": self.process.pid,
|
||||
"processGroupId": self.process.pid}
|
||||
with (self.root / "process-groups.jsonl").open("a") as handle:
|
||||
handle.write(json.dumps(lease) + "\n")
|
||||
deadline = time.monotonic() + 15
|
||||
journal = self.root / f"{self.name}.producer.jsonl"
|
||||
while time.monotonic() < deadline:
|
||||
if self.process.poll() is not None:
|
||||
raise RuntimeError(f"Proxy exited: {self.log.name}")
|
||||
if journal.exists() and '"event": "ready"' in journal.read_text():
|
||||
return self
|
||||
await asyncio.sleep(0.05)
|
||||
raise TimeoutError(f"Proxy readiness: {self.log.name}")
|
||||
|
||||
async def stop(self):
|
||||
if self.process is not None and self.process.poll() is None:
|
||||
# The standalone mitmdump distribution has a bootloader parent.
|
||||
# Signal its owned session together, including the interpreter child.
|
||||
os.killpg(self.process.pid, signal.SIGINT)
|
||||
try:
|
||||
await asyncio.wait_for(asyncio.to_thread(self.process.wait), 10)
|
||||
except TimeoutError:
|
||||
os.killpg(self.process.pid, signal.SIGKILL)
|
||||
await asyncio.to_thread(self.process.wait)
|
||||
raise
|
||||
if self.process is not None:
|
||||
for _ in range(40):
|
||||
if group_is_empty(self.process.pid):
|
||||
return
|
||||
await asyncio.sleep(0.05)
|
||||
raise RuntimeError(f"Owned proxy group still live: {self.process.pid}")
|
||||
|
||||
async def close_log(self):
|
||||
if self.log is not None:
|
||||
self.log.close()
|
||||
|
||||
|
||||
def certificates(root):
|
||||
for name in ["origin"]:
|
||||
subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
|
||||
"-keyout", str(root / f"{name}-key.pem"), "-out", str(root / f"{name}-ca.pem"),
|
||||
"-days", "2", "-subj", "/CN=lab.test", "-addext", "subjectAltName=DNS:lab.test",
|
||||
"-addext", "basicConstraints=critical,CA:TRUE"], check=True,
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
|
||||
|
||||
def client_config(root):
|
||||
config = QuicConfiguration(is_client=True, alpn_protocols=["h3"], supported_versions=[QuicProtocolVersion.VERSION_1],
|
||||
server_name="lab.test", idle_timeout=3)
|
||||
config.load_verify_locations(cafile=str(root / "proxy-config/mitmproxy-ca-cert.pem"))
|
||||
return config
|
||||
|
||||
|
||||
def connection(root, journal):
|
||||
return connect("127.0.0.1", 18443, configuration=client_config(root),
|
||||
create_protocol=lambda *a, **kw: Endpoint(*a, journal=journal, endpoint="client", **kw))
|
||||
|
||||
|
||||
async def curl_request(root, journal, name, version, path="/curl", maximum=8):
|
||||
child = await asyncio.create_subprocess_exec("curl", version, "--silent", "--show-error",
|
||||
"--max-time", str(maximum), "--cacert", str(root / "proxy-config/mitmproxy-ca-cert.pem"),
|
||||
"--resolve", "lab.test:18443:127.0.0.1", "--noproxy", "*", "-H", f"x-lab-case: {name}",
|
||||
"--output", str(root / (name + ".body")), "--dump-header", str(root / (name + ".headers")),
|
||||
"--write-out", "%{json}", "https://lab.test:18443" + path,
|
||||
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
|
||||
journal.children.append(child)
|
||||
out, err = await child.communicate()
|
||||
(root / (name + ".json")).write_bytes(out)
|
||||
(root / (name + ".stderr")).write_bytes(err)
|
||||
return child.returncode, json.loads(out)
|
||||
|
||||
|
||||
async def tcp_origin(reader, writer, journal):
|
||||
journal.writers.append(writer)
|
||||
try:
|
||||
head = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), 5)
|
||||
case = "tcp"
|
||||
for line in head.split(b"\r\n")[1:]:
|
||||
if line.lower().startswith(b"x-lab-case:"):
|
||||
case = line.split(b":", 1)[1].strip().decode("ascii")
|
||||
body = b"\x00\xfftcp-origin"
|
||||
journal.emit("tcp-request", case=case, alpn=writer.get_extra_info("ssl_object").selected_alpn_protocol(),
|
||||
headers=binary(head), body=binary(body))
|
||||
writer.write(b"HTTP/1.1 200 OK\r\ncontent-type: application/octet-stream\r\ncontent-length: " +
|
||||
str(len(body)).encode() + b"\r\nconnection: close\r\n\r\n" + body)
|
||||
await writer.drain()
|
||||
finally:
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
|
||||
def group_is_empty(group):
|
||||
for entry in Path("/proc").iterdir():
|
||||
if not entry.name.isdigit():
|
||||
continue
|
||||
try:
|
||||
fields = (entry / "stat").read_text().rsplit(")", 1)[1].split()
|
||||
except FileNotFoundError:
|
||||
continue
|
||||
if fields[0] != "Z" and int(fields[2]) == group:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
async def attempt_all(actions):
|
||||
failures = []
|
||||
for name, action in actions:
|
||||
try:
|
||||
await asyncio.wait_for(action(), 12)
|
||||
except BaseException as error:
|
||||
failures.append((name, error))
|
||||
return failures
|
||||
|
||||
|
||||
async def close_resource(resource):
|
||||
resource.close()
|
||||
|
||||
|
||||
async def finish_child(child):
|
||||
if child.returncode is None:
|
||||
child.terminate()
|
||||
try:
|
||||
await asyncio.wait_for(child.wait(), 3)
|
||||
except TimeoutError:
|
||||
child.kill()
|
||||
await child.wait()
|
||||
|
||||
|
||||
async def finish_tasks(journal):
|
||||
for task in journal.tasks:
|
||||
if not task.done():
|
||||
task.cancel()
|
||||
results = await asyncio.gather(*journal.tasks, return_exceptions=True)
|
||||
failures = [error for error in results if isinstance(error, BaseException)
|
||||
and not isinstance(error, asyncio.CancelledError)]
|
||||
if failures:
|
||||
raise BaseExceptionGroup("Endpoint tasks failed", failures)
|
||||
|
||||
|
||||
async def run(args):
|
||||
if aioquic.__version__ != "1.2.0":
|
||||
raise RuntimeError("This laboratory requires aioquic 1.2.0")
|
||||
root = Path(args.output).resolve()
|
||||
root.mkdir(mode=0o700)
|
||||
os.umask(0o077)
|
||||
journal = Journal(root)
|
||||
loop = asyncio.get_running_loop()
|
||||
asynchronous_errors = []
|
||||
def unexpected_exception(active_loop, context):
|
||||
asynchronous_errors.append(context.get("message", "Unexpected asynchronous error"))
|
||||
active_loop.default_exception_handler(context)
|
||||
loop.set_exception_handler(unexpected_exception)
|
||||
certificates(root)
|
||||
server = tcp_server = None
|
||||
proxies = []
|
||||
scenarios = []
|
||||
primary_error = None
|
||||
cleanup_errors = []
|
||||
try:
|
||||
config = QuicConfiguration(is_client=False, alpn_protocols=["h3"],
|
||||
supported_versions=[QuicProtocolVersion.VERSION_1], idle_timeout=3)
|
||||
config.load_cert_chain(str(root / "origin-ca.pem"), str(root / "origin-key.pem"))
|
||||
server = await serve("127.0.0.1", 19443, configuration=config,
|
||||
create_protocol=lambda *a, **kw: Endpoint(*a, journal=journal, endpoint="origin", **kw))
|
||||
tls_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||||
tls_context.load_cert_chain(str(root / "origin-ca.pem"), str(root / "origin-key.pem"))
|
||||
tls_context.set_alpn_protocols(["http/1.1"])
|
||||
tcp_server = await asyncio.start_server(lambda r, w: journal.task(tcp_origin(r, w, journal)),
|
||||
"127.0.0.1", 19443, ssl=tls_context)
|
||||
proxy = Proxy(args, root, "normal")
|
||||
proxies.append(proxy)
|
||||
await proxy.start()
|
||||
async with connection(root, journal) as client:
|
||||
requests = [client.request(f"parallel-{i}", bytes([0, 255, i]) + bytes(range(256))) for i in range(3)]
|
||||
await asyncio.wait_for(asyncio.gather(*(future for _, future in requests)), 10)
|
||||
# cURL supplies an independently implemented HTTP/3 smoke case.
|
||||
code, result = await curl_request(root, journal, "curl", "--http3-only")
|
||||
if code != 0:
|
||||
raise RuntimeError("HTTP/3 cURL failed; inspect curl.stderr")
|
||||
if result["http_version"] != "3":
|
||||
raise AssertionError("cURL silently downgraded")
|
||||
await proxy.stop()
|
||||
scenarios.append({"name": "normal", "capture": "normal.mitm"})
|
||||
|
||||
proxy = Proxy(args, root, "fallback", http3=False)
|
||||
proxies.append(proxy)
|
||||
await proxy.start()
|
||||
code, result = await curl_request(root, journal, "fallback", "--http3", path="/fallback")
|
||||
if code != 0 or result["http_version"] != "2":
|
||||
raise AssertionError(f"Fallback control: {result}")
|
||||
await proxy.stop()
|
||||
scenarios.append({"name": "fallback", "capture": "fallback.mitm"})
|
||||
|
||||
except BaseException as error:
|
||||
primary_error = error
|
||||
finally:
|
||||
actions = []
|
||||
for proxy in proxies:
|
||||
actions.extend([(f"proxy-{proxy.name}", proxy.stop),
|
||||
(f"proxy-log-{proxy.name}", proxy.close_log)])
|
||||
if server is not None:
|
||||
actions.append(("quic-server", lambda: close_resource(server)))
|
||||
if tcp_server is not None:
|
||||
actions.extend([("tcp-listener-close", lambda: close_resource(tcp_server)),
|
||||
("tcp-listener-wait", tcp_server.wait_closed)])
|
||||
for index, writer in enumerate(journal.writers):
|
||||
actions.extend([(f"tcp-writer-close-{index}", lambda w=writer: close_resource(w)),
|
||||
(f"tcp-writer-wait-{index}", writer.wait_closed)])
|
||||
actions.append(("endpoint-tasks", lambda: finish_tasks(journal)))
|
||||
for index, child in enumerate(journal.children):
|
||||
actions.append((f"curl-child-{index}", lambda c=child: finish_child(c)))
|
||||
cleanup_errors = await attempt_all(actions)
|
||||
# This runs after every close attempt, including failed setup.
|
||||
groups = []
|
||||
async def audit_groups():
|
||||
for proxy in proxies:
|
||||
if proxy.process is not None:
|
||||
groups.append({"group": proxy.process.pid, "empty": group_is_empty(proxy.process.pid)})
|
||||
if any(not item["empty"] for item in groups):
|
||||
raise RuntimeError("Proxy groups remain live")
|
||||
cleanup_errors.extend(await attempt_all([("process-groups", audit_groups)]))
|
||||
try:
|
||||
(root / "producer-cleanup.json").write_text(json.dumps({
|
||||
"primary_error": None if primary_error is None else str(primary_error),
|
||||
"cleanup_errors": [{"action": name, "type": type(error).__name__, "message": str(error)}
|
||||
for name, error in cleanup_errors], "groups": groups,
|
||||
"asynchronous_errors": asynchronous_errors}, indent=2))
|
||||
except BaseException as error:
|
||||
cleanup_errors.append(("cleanup-report", error))
|
||||
if primary_error is not None:
|
||||
for name, error in cleanup_errors:
|
||||
primary_error.add_note(f"Cleanup {name}: {error}")
|
||||
raise primary_error
|
||||
if cleanup_errors:
|
||||
raise BaseExceptionGroup("HTTP/3 producer cleanup failed", [error for _, error in cleanup_errors])
|
||||
if asynchronous_errors:
|
||||
raise RuntimeError(f"Unexpected asynchronous errors: {asynchronous_errors}")
|
||||
(root / "manifest.json").write_text(json.dumps({"profile": "http3-minimal-v1", "aioquic": aioquic.__version__,
|
||||
"scenarios": scenarios, "requests": journal.requests, "responses": journal.responses,
|
||||
"events": journal.events, "limitations": [
|
||||
"Endpoint stream IDs are scoped observations, not mitmproxy serialized IDs or mappings.",
|
||||
"The fixture reads negotiated _version from pinned aioquic 1.2.0.",
|
||||
"Only QUIC v1, Linux reverse proxy and the selected endpoints are exercised."]}, indent=2))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--mitmdump", required=True)
|
||||
parser.add_argument("--output", required=True)
|
||||
asyncio.run(run(parser.parse_args()))
|
||||
@@ -0,0 +1,56 @@
|
||||
"""Lifecycle regressions; these tests do not establish HTTP/3 provider support."""
|
||||
import importlib.util
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
spec = importlib.util.spec_from_file_location("http3_lab", Path(__file__).with_name("http3-lab.py"))
|
||||
lab = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(lab)
|
||||
|
||||
|
||||
class CleanupTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_all_actions_run_after_multiple_close_failures(self):
|
||||
attempted = []
|
||||
async def fail():
|
||||
attempted.append("failed-close")
|
||||
raise RuntimeError("close failed")
|
||||
async def audit():
|
||||
attempted.append("audit")
|
||||
failures = await lab.attempt_all([("one", fail), ("two", fail), ("audit", audit)])
|
||||
self.assertEqual(attempted, ["failed-close", "failed-close", "audit"])
|
||||
self.assertEqual([name for name, _ in failures], ["one", "two"])
|
||||
|
||||
async def test_partial_setup_keeps_primary_error_and_audits_after_close_failure(self):
|
||||
original_serve = lab.serve
|
||||
closed = []
|
||||
class ClosingServer:
|
||||
def __init__(self, server):
|
||||
self.server = server
|
||||
def close(self):
|
||||
self.server.close()
|
||||
closed.append(True)
|
||||
raise RuntimeError("injected close failure")
|
||||
async def serve(*args, **kwargs):
|
||||
return ClosingServer(await original_serve(*args, **kwargs))
|
||||
async def fail_tcp(*args, **kwargs):
|
||||
raise RuntimeError("injected primary setup failure")
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory) / "capture"
|
||||
with patch.object(lab, "serve", serve), patch.object(lab.asyncio, "start_server", fail_tcp):
|
||||
with self.assertRaisesRegex(RuntimeError, "injected primary setup failure") as caught:
|
||||
await lab.run(SimpleNamespace(output=str(root), mitmdump="unused"))
|
||||
self.assertEqual(closed, [True])
|
||||
self.assertIn("injected close failure", " ".join(caught.exception.__notes__))
|
||||
report = json.loads((root / "producer-cleanup.json").read_text())
|
||||
self.assertEqual(report["primary_error"], "injected primary setup failure")
|
||||
self.assertEqual(report["cleanup_errors"][0]["action"], "quic-server")
|
||||
self.assertEqual(report["groups"], [])
|
||||
self.assertFalse((root / "manifest.json").exists())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,64 @@
|
||||
"""Test-only observation of real flows; never synthesizes flow or stream IDs."""
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from mitmproxy import ctx
|
||||
|
||||
|
||||
def bytes_view(value):
|
||||
if value is None:
|
||||
return None
|
||||
return {"base64": base64.b64encode(value).decode("ascii"),
|
||||
"bytes": len(value), "sha256": hashlib.sha256(value).hexdigest()}
|
||||
|
||||
|
||||
class Observer:
|
||||
def load(self, loader):
|
||||
loader.add_option("rea_lab_log", str, "", "HTTP/3 verifier observations")
|
||||
|
||||
def emit(self, event, **data):
|
||||
with Path(ctx.options.rea_lab_log).open("a", encoding="utf-8") as handle:
|
||||
handle.write(json.dumps({"event": event, **data}) + "\n")
|
||||
|
||||
def running(self):
|
||||
self.emit("ready")
|
||||
|
||||
def connection(self, conn):
|
||||
return {"id": conn.id, "transport_protocol": conn.transport_protocol,
|
||||
"alpn": bytes_view(conn.alpn), "tls_version": conn.tls_version,
|
||||
"timestamp_tls_setup": conn.timestamp_tls_setup,
|
||||
"error": conn.error}
|
||||
|
||||
def tls_established_client(self, data):
|
||||
self.emit("tls-established-client", connection=self.connection(data.conn))
|
||||
|
||||
def tls_established_server(self, data):
|
||||
self.emit("tls-established-server", connection=self.connection(data.conn))
|
||||
|
||||
def flow(self, event, flow):
|
||||
def message(message):
|
||||
if message is None:
|
||||
return None
|
||||
return {"http_version": message.http_version,
|
||||
"content": bytes_view(message.raw_content),
|
||||
"timestamp_end": message.timestamp_end}
|
||||
self.emit(event, id=flow.id, case=flow.request.headers.get("x-lab-case", "curl"),
|
||||
request=message(flow.request), response=message(flow.response),
|
||||
error=None if flow.error is None else flow.error.msg,
|
||||
client_conn=self.connection(flow.client_conn),
|
||||
server_conn=self.connection(flow.server_conn))
|
||||
|
||||
def request(self, flow):
|
||||
self.flow("request", flow)
|
||||
|
||||
def response(self, flow):
|
||||
self.flow("response", flow)
|
||||
|
||||
def error(self, flow):
|
||||
self.flow("error", flow)
|
||||
|
||||
|
||||
addons = [Observer()]
|
||||
@@ -0,0 +1,100 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
|
||||
import {
|
||||
inspectWebPageInputSchema,
|
||||
listBrowserTargetsInputSchema,
|
||||
} from "../../dist/domain/browserObservation.js";
|
||||
import { compareWebCapturesInputSchema } from "../../dist/domain/webCaptureDiffSchemas.js";
|
||||
import { artifactCliEvidence } from "./artifact-e2e.mjs";
|
||||
|
||||
/** Verify browser-socket discovery and inspection against HTTP discovery. */
|
||||
export async function verifyBrowserSocketTransport({
|
||||
provider,
|
||||
profile,
|
||||
endpoint,
|
||||
target,
|
||||
origin,
|
||||
}) {
|
||||
const [port, path] = (
|
||||
await readFile(join(profile, "DevToolsActivePort"), "utf8")
|
||||
).split("\n");
|
||||
const socket = `ws://127.0.0.1:${port}${path}`;
|
||||
const version = await (await fetch(`${endpoint}/json/version`)).json();
|
||||
assert.equal(
|
||||
new URL(version.webSocketDebuggerUrl).pathname,
|
||||
path,
|
||||
"DevToolsActivePort and /json/version named different browser sockets",
|
||||
);
|
||||
const list = async (cdpEndpoint) => {
|
||||
const listed = await provider.listTargets(
|
||||
listBrowserTargetsInputSchema.parse({
|
||||
cdp_endpoint: cdpEndpoint,
|
||||
allowed_origins: [origin],
|
||||
}),
|
||||
);
|
||||
if (!listed.ok) throw listed.error;
|
||||
return listed.value;
|
||||
};
|
||||
const throughHttp = await list(endpoint);
|
||||
const throughSocket = await list(socket);
|
||||
// Target.getTargets reports attachment state that /json/list may omit.
|
||||
const withoutAttachment = (targets) =>
|
||||
targets.map(({ attached: _attached, ...rest }) => rest);
|
||||
assert.ok(throughSocket.targets.length > 0, "Browser socket listed no page");
|
||||
assert.deepEqual(
|
||||
withoutAttachment(throughSocket.targets),
|
||||
withoutAttachment(throughHttp.targets),
|
||||
);
|
||||
assert.ok(
|
||||
throughSocket.targets.every(
|
||||
({ attached }) => typeof attached === "boolean",
|
||||
),
|
||||
"Browser socket omitted target attachment state",
|
||||
);
|
||||
assert.equal(throughSocket.browser.product, throughHttp.browser.product);
|
||||
|
||||
const evidence = await artifactCliEvidence("inspect-web-page", socket, [
|
||||
target,
|
||||
"--allowed-origins",
|
||||
origin,
|
||||
"--observation-ms",
|
||||
"100",
|
||||
]);
|
||||
assert.equal(evidence.normalized_result.target.target_id, target);
|
||||
assert.ok(
|
||||
evidence.normalized_result.dom.nodes.length > 0,
|
||||
"Browser-socket inspection returned no DOM nodes",
|
||||
);
|
||||
assert.equal(
|
||||
evidence.parameters.cdp_endpoint,
|
||||
`ws://127.0.0.1:${port}/devtools/browser/[REDACTED]`,
|
||||
);
|
||||
assert.ok(
|
||||
!JSON.stringify(evidence).includes(path),
|
||||
"Browser-socket Evidence retained the socket identifier",
|
||||
);
|
||||
const throughHttpInspection = await provider.inspectPage(
|
||||
inspectWebPageInputSchema.parse({
|
||||
cdp_endpoint: endpoint,
|
||||
allowed_origins: [origin],
|
||||
target_id: target,
|
||||
observation_ms: 100,
|
||||
}),
|
||||
);
|
||||
if (!throughHttpInspection.ok) throw throughHttpInspection.error;
|
||||
const compared = await provider.compareCaptures(
|
||||
compareWebCapturesInputSchema.parse({
|
||||
before: { inspection: throughHttpInspection.value },
|
||||
after: { inspection: evidence.normalized_result },
|
||||
}),
|
||||
);
|
||||
if (!compared.ok) throw compared.error;
|
||||
assert.equal(compared.value.dimensions.dom_structure.status, "unchanged");
|
||||
assert.equal(compared.value.dimensions.scripts.status, "unchanged");
|
||||
return {
|
||||
targets: throughSocket.targets.length,
|
||||
revision: throughSocket.browser.revision,
|
||||
};
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
export function assertCleanLinuxHopperRuntime(
|
||||
launcherPaths?: readonly string[],
|
||||
procRoot?: string,
|
||||
): Promise<void>;
|
||||
|
||||
@@ -16,6 +16,7 @@ export const assertCleanLinuxHopperRuntime = async (
|
||||
? []
|
||||
: [process.env.HOPPER_LAUNCHER_PATH]),
|
||||
],
|
||||
procRoot = "/proc",
|
||||
) => {
|
||||
const { stdout } = await execFileAsync("ps", ["-ax", "-o", "command="]);
|
||||
if (/\/opt\/hopper\/bin\/Hopper|\bXvfb\b|hopper-demo-x11\.py/u.test(stdout))
|
||||
@@ -32,17 +33,17 @@ export const assertCleanLinuxHopperRuntime = async (
|
||||
// /proc exposes the executed file even when argv[0] names a removed symlink.
|
||||
// Compare file identity, not a display path or a process name.
|
||||
let inspectionError;
|
||||
for (const entry of await readdir("/proc", { withFileTypes: true })) {
|
||||
for (const entry of await readdir(procRoot, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory() || !/^\d+$/u.test(entry.name)) continue;
|
||||
let executable;
|
||||
try {
|
||||
executable = await stat(`/proc/${entry.name}/exe`, { bigint: true });
|
||||
executable = await stat(`${procRoot}/${entry.name}/exe`, { bigint: true });
|
||||
} catch (error) {
|
||||
if (error?.code === "ENOENT" || error?.code === "ESRCH") continue;
|
||||
// Other users' executables may be unreadable. The setup lane launches
|
||||
// providers as the caller; unreadable caller-owned processes fail closed.
|
||||
if (error?.code === "EACCES") {
|
||||
const owner = await stat(`/proc/${entry.name}`).catch((cause) => {
|
||||
const owner = await stat(`${procRoot}/${entry.name}`).catch((cause) => {
|
||||
if (cause?.code === "ENOENT") return undefined;
|
||||
throw cause;
|
||||
});
|
||||
|
||||
@@ -0,0 +1,243 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { createHash } from "node:crypto";
|
||||
import { isUtf8 } from "node:buffer";
|
||||
const digest = (bytes) => createHash("sha256").update(bytes).digest("hex");
|
||||
// The native decoder displays strict UTF-8 and retains invalid bytes as null text.
|
||||
const textOf = (expected) => {
|
||||
const value = Buffer.from(expected.base64, "base64");
|
||||
return isUtf8(value) ? value.toString("utf8") : null;
|
||||
};
|
||||
const field = (record, pointer) =>
|
||||
record.binary_fields.find((item) => item.pointer === pointer);
|
||||
const bytes = (record, pointer) => {
|
||||
const value = field(record, pointer);
|
||||
assert.equal(
|
||||
value?.state,
|
||||
"retained",
|
||||
`Missing retained bytes at ${pointer}`,
|
||||
);
|
||||
const decoded = Buffer.from(value.content_base64, "base64");
|
||||
assert.equal(value.bytes, decoded.length);
|
||||
assert.equal(value.sha256, digest(decoded));
|
||||
return decoded;
|
||||
};
|
||||
const binaryEqual = (record, pointer, expected) => {
|
||||
const decoded = bytes(record, pointer);
|
||||
assert.deepEqual(decoded, Buffer.from(expected.base64, "base64"), pointer);
|
||||
assert.equal(field(record, pointer).sha256, expected.sha256);
|
||||
assert.equal(field(record, pointer).bytes, expected.bytes);
|
||||
};
|
||||
const caseOf = (record) =>
|
||||
record.reported.request.headers.find(
|
||||
([name]) => name === "x-lab-case",
|
||||
)?.[1] ?? "curl";
|
||||
const byCase = (value, name) => {
|
||||
const selected = value.records.filter((record) => caseOf(record) === name);
|
||||
assert.equal(
|
||||
selected.length,
|
||||
1,
|
||||
`Expected exactly one producer record for ${name}`,
|
||||
);
|
||||
return selected[0];
|
||||
};
|
||||
const connectionH3 = (record, side) => {
|
||||
const conn = record.reported[side];
|
||||
assert.equal(conn.transport_protocol, "udp");
|
||||
assert.equal(bytes(record, `/${side}/alpn`).toString(), "h3");
|
||||
assert.equal(conn.alpn, "h3", `${side}.alpn readable value`);
|
||||
assert.equal(typeof conn.timestamp_tls_setup, "number");
|
||||
};
|
||||
|
||||
function assertHeaders(record, side, headers) {
|
||||
const regular = headers.filter(
|
||||
([name]) => !Buffer.from(name.base64, "base64").toString().startsWith(":"),
|
||||
);
|
||||
assert.equal(record.reported[side].headers.length, regular.length);
|
||||
for (const [index, pair] of regular.entries()) {
|
||||
binaryEqual(record, `/${side}/headers/${index}/0`, pair[0]);
|
||||
binaryEqual(record, `/${side}/headers/${index}/1`, pair[1]);
|
||||
assert.deepEqual(
|
||||
record.reported[side].headers[index],
|
||||
pair.map(textOf),
|
||||
`${side}.headers[${index}] readable values`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
export function assertEndpointHeaders(actual, expected) {
|
||||
// The proxy rebuilds pseudo fields from its HTTP state. Compare their values
|
||||
// by name while retaining the exact order and duplicates of regular headers.
|
||||
const partition = (headers) => {
|
||||
const pseudo = new Map();
|
||||
const regular = [];
|
||||
for (const pair of headers) {
|
||||
const name = Buffer.from(pair[0].base64, "base64").toString();
|
||||
if (name.startsWith(":")) {
|
||||
assert.equal(pseudo.has(name), false);
|
||||
pseudo.set(name, pair);
|
||||
} else regular.push(pair);
|
||||
}
|
||||
return { pseudo, regular };
|
||||
};
|
||||
assert.deepEqual(partition(actual), partition(expected));
|
||||
}
|
||||
|
||||
export function requireHttp3(record) {
|
||||
for (const side of ["request", "response"])
|
||||
assert.equal(record.reported[side].http_version, "HTTP/3");
|
||||
connectionH3(record, "client_conn");
|
||||
connectionH3(record, "server_conn");
|
||||
}
|
||||
|
||||
export function verifyNormalRecords(normal, manifest, finalProducer) {
|
||||
for (const record of normal.records) {
|
||||
const name = caseOf(record);
|
||||
const producer = finalProducer.find(
|
||||
(event) => event.id === record.reported.id,
|
||||
);
|
||||
assert.ok(producer, `No actual producer observation for ${name}`);
|
||||
assert.equal(record.reported.error, null);
|
||||
assert.equal(producer.error, null);
|
||||
assert.equal(record.reported.response.status_code, 200);
|
||||
requireHttp3(record);
|
||||
for (const side of ["request", "response"]) {
|
||||
assert.equal(
|
||||
record.reported[side].http_version,
|
||||
producer[side].http_version,
|
||||
);
|
||||
binaryEqual(record, `/${side}/content`, producer[side].content);
|
||||
assertHeaders(
|
||||
record,
|
||||
side,
|
||||
manifest[side === "request" ? "requests" : "responses"][name].headers,
|
||||
);
|
||||
}
|
||||
// Native UUIDs must be retained as UUIDs; the producer exports no stream mapping.
|
||||
for (const side of ["client_conn", "server_conn"])
|
||||
for (const key of ["id", "tls_version", "timestamp_tls_setup", "error"])
|
||||
assert.equal(
|
||||
record.reported[side][key],
|
||||
producer[side][key],
|
||||
`${side}.${key} differs from producer observation ${producer.id}`,
|
||||
);
|
||||
binaryEqual(record, "/request/content", manifest.requests[name].body);
|
||||
binaryEqual(record, "/response/content", manifest.responses[name].body);
|
||||
const pseudo = {
|
||||
":method": "method",
|
||||
":scheme": "scheme",
|
||||
":authority": "authority",
|
||||
":path": "path",
|
||||
};
|
||||
for (const [headerName, headerValue] of manifest.requests[name].headers) {
|
||||
const decodedName = Buffer.from(headerName.base64, "base64").toString();
|
||||
if (pseudo[decodedName]) {
|
||||
binaryEqual(record, `/request/${pseudo[decodedName]}`, headerValue);
|
||||
assert.equal(
|
||||
record.reported.request[pseudo[decodedName]],
|
||||
textOf(headerValue),
|
||||
`request.${pseudo[decodedName]} readable value`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export { bytes, caseOf, byCase };
|
||||
|
||||
export function verifyConcurrency(normal, manifest) {
|
||||
const parallelNames = ["parallel-0", "parallel-1", "parallel-2"];
|
||||
const sent = manifest.events.filter(
|
||||
(event) =>
|
||||
event.event === "request-sent" && parallelNames.includes(event.case),
|
||||
);
|
||||
const received = manifest.events.filter(
|
||||
(event) =>
|
||||
event.event === "request-received" && parallelNames.includes(event.case),
|
||||
);
|
||||
const responses = manifest.events.filter(
|
||||
(event) =>
|
||||
event.event === "response-sent" && parallelNames.includes(event.case),
|
||||
);
|
||||
assert.equal(new Set(sent.map((event) => event.connection)).size, 1);
|
||||
assert.equal(new Set(received.map((event) => event.connection)).size, 1);
|
||||
assert.equal(new Set(sent.map((event) => event.stream_id)).size, 3);
|
||||
assert.equal(new Set(received.map((event) => event.stream_id)).size, 3);
|
||||
assert.ok(
|
||||
Math.max(...received.map((event) => event.monotonic)) <
|
||||
responses[0].monotonic,
|
||||
);
|
||||
assert.deepEqual(
|
||||
responses.map((event) => event.case),
|
||||
received.map((event) => event.case).reverse(),
|
||||
);
|
||||
for (const side of ["client_conn", "server_conn"])
|
||||
assert.equal(
|
||||
new Set(
|
||||
parallelNames.map((name) => byCase(normal, name).reported[side].id),
|
||||
).size,
|
||||
1,
|
||||
);
|
||||
const normalConnections = new Set(
|
||||
[...sent, ...received].map((event) => event.connection),
|
||||
);
|
||||
for (const connection of normalConnections) {
|
||||
const handshakes = manifest.events.filter(
|
||||
(event) => event.event === "handshake" && event.connection === connection,
|
||||
);
|
||||
assert.equal(handshakes.length, 1);
|
||||
assert.equal(handshakes[0].alpn, "h3");
|
||||
assert.equal(handshakes[0].quic_version, 1);
|
||||
}
|
||||
assert.equal(
|
||||
new Set(normal.records.map((record) => record.reported.id)).size,
|
||||
4,
|
||||
);
|
||||
assert.equal(sent.length, 3);
|
||||
const receivedByClient = manifest.events.filter(
|
||||
(event) =>
|
||||
event.event === "response-received" &&
|
||||
sent.some(
|
||||
(request) =>
|
||||
request.connection === event.connection &&
|
||||
request.stream_id === event.stream_id,
|
||||
),
|
||||
);
|
||||
assert.equal(receivedByClient.length, 3);
|
||||
for (const request of sent) {
|
||||
const response = receivedByClient.find(
|
||||
(event) =>
|
||||
event.connection === request.connection &&
|
||||
event.stream_id === request.stream_id,
|
||||
);
|
||||
assert.ok(
|
||||
response,
|
||||
"Each sent request needs an independently observed client response",
|
||||
);
|
||||
assert.deepEqual(response.body, manifest.responses[request.case].body);
|
||||
assert.deepEqual(request.body, manifest.requests[request.case].body);
|
||||
assertEndpointHeaders(
|
||||
request.headers,
|
||||
manifest.requests[request.case].headers,
|
||||
);
|
||||
assertEndpointHeaders(
|
||||
response.headers,
|
||||
manifest.responses[request.case].headers,
|
||||
);
|
||||
}
|
||||
return {
|
||||
sent: sent.map((event) => event.case),
|
||||
completed_at_origin: received.map((event) => event.case),
|
||||
dispatched_at_origin: responses.map((event) => event.case),
|
||||
completed_at_client: receivedByClient.map(
|
||||
(event) =>
|
||||
sent.find(
|
||||
(request) =>
|
||||
request.connection === event.connection &&
|
||||
request.stream_id === event.stream_id,
|
||||
).case,
|
||||
),
|
||||
saved_in_capture: normal.records
|
||||
.filter((record) => parallelNames.includes(caseOf(record)))
|
||||
.map(caseOf),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { createHash } from "node:crypto";
|
||||
import { test } from "node:test";
|
||||
import { verifyNormalRecords } from "./http3-capture-assertions.mjs";
|
||||
|
||||
function connectionFactsFixture() {
|
||||
const body = Buffer.from([0, 255, 128, 1]);
|
||||
const view = (bytes) => ({
|
||||
base64: bytes.toString("base64"),
|
||||
bytes: bytes.length,
|
||||
sha256: createHash("sha256").update(bytes).digest("hex"),
|
||||
});
|
||||
const producer = {
|
||||
id: "flow-id",
|
||||
error: null,
|
||||
request: { http_version: "HTTP/3", content: view(body) },
|
||||
response: { http_version: "HTTP/3", content: view(body) },
|
||||
client_conn: {
|
||||
id: "client-id",
|
||||
transport_protocol: "udp",
|
||||
tls_version: "QUICv1",
|
||||
timestamp_tls_setup: 1_700_000_000.125678,
|
||||
error: null,
|
||||
},
|
||||
server_conn: {
|
||||
id: "server-id",
|
||||
transport_protocol: "udp",
|
||||
tls_version: "QUICv1",
|
||||
timestamp_tls_setup: 1_700_000_000.987654,
|
||||
error: null,
|
||||
},
|
||||
};
|
||||
const reported = structuredClone(producer);
|
||||
const requestFields = {
|
||||
method: "POST",
|
||||
scheme: "https",
|
||||
authority: "lab.test:18443",
|
||||
path: "/echo?case=readable",
|
||||
};
|
||||
Object.assign(reported.request, requestFields);
|
||||
const headers = [
|
||||
[Buffer.from("x-repeat"), Buffer.from("first-α")],
|
||||
[Buffer.from("x-repeat"), Buffer.from("second-β")],
|
||||
[Buffer.from("x-binary"), Buffer.from([255])],
|
||||
];
|
||||
for (const side of ["request", "response"])
|
||||
reported[side].headers = [
|
||||
["x-repeat", "first-α"],
|
||||
["x-repeat", "second-β"],
|
||||
["x-binary", null],
|
||||
];
|
||||
reported.client_conn.alpn = "h3";
|
||||
reported.server_conn.alpn = "h3";
|
||||
reported.response.status_code = 200;
|
||||
const binary_fields = [
|
||||
["/request/content", body],
|
||||
["/response/content", body],
|
||||
["/client_conn/alpn", Buffer.from("h3")],
|
||||
["/server_conn/alpn", Buffer.from("h3")],
|
||||
...Object.entries(requestFields).map(([name, value]) => [
|
||||
`/request/${name}`,
|
||||
Buffer.from(value),
|
||||
]),
|
||||
...["request", "response"].flatMap((side) =>
|
||||
headers.flatMap((pair, index) =>
|
||||
pair.map((value, part) => [`/${side}/headers/${index}/${part}`, value]),
|
||||
),
|
||||
),
|
||||
].map(([pointer, bytes]) => {
|
||||
const { base64, ...facts } = view(bytes);
|
||||
return {
|
||||
pointer,
|
||||
representation: "producer-bytes",
|
||||
state: "retained",
|
||||
content_base64: base64,
|
||||
...facts,
|
||||
};
|
||||
});
|
||||
return {
|
||||
normal: { records: [{ reported, binary_fields }] },
|
||||
manifest: {
|
||||
requests: {
|
||||
curl: {
|
||||
headers: [
|
||||
...Object.entries(requestFields).map(([name, value]) => [
|
||||
view(Buffer.from(`:${name}`)),
|
||||
view(Buffer.from(value)),
|
||||
]),
|
||||
...headers.map((pair) => pair.map(view)),
|
||||
],
|
||||
body: view(body),
|
||||
},
|
||||
},
|
||||
responses: {
|
||||
curl: {
|
||||
headers: headers.map((pair) => pair.map(view)),
|
||||
body: view(body),
|
||||
},
|
||||
},
|
||||
},
|
||||
producer: [producer],
|
||||
};
|
||||
}
|
||||
|
||||
test("positive assertions retain exact producer connection facts on both legs", () => {
|
||||
const { normal, manifest, producer } = connectionFactsFixture();
|
||||
verifyNormalRecords(normal, manifest, producer);
|
||||
});
|
||||
|
||||
for (const [field, value] of [
|
||||
["method", "GET"],
|
||||
["scheme", "http"],
|
||||
["authority", "other.test:18443"],
|
||||
["path", "/different"],
|
||||
])
|
||||
test(`positive assertions reject changed readable request ${field}`, () => {
|
||||
const { normal, manifest, producer } = connectionFactsFixture();
|
||||
normal.records[0].reported.request[field] = value;
|
||||
assert.throws(() => verifyNormalRecords(normal, manifest, producer), {
|
||||
code: "ERR_ASSERTION",
|
||||
});
|
||||
});
|
||||
|
||||
for (const side of ["client_conn", "server_conn"])
|
||||
test(`positive assertions reject null readable ${side} ALPN`, () => {
|
||||
const { normal, manifest, producer } = connectionFactsFixture();
|
||||
normal.records[0].reported[side].alpn = null;
|
||||
assert.throws(() => verifyNormalRecords(normal, manifest, producer), {
|
||||
code: "ERR_ASSERTION",
|
||||
});
|
||||
});
|
||||
|
||||
for (const side of ["request", "response"])
|
||||
for (const [index, part, value] of [
|
||||
[0, 0, "x-changed"],
|
||||
[0, 1, "changed-first"],
|
||||
[1, 1, "changed-second"],
|
||||
[2, 1, "�"],
|
||||
])
|
||||
test(`positive assertions reject changed readable ${side} header ${index}/${part}`, () => {
|
||||
const { normal, manifest, producer } = connectionFactsFixture();
|
||||
normal.records[0].reported[side].headers[index][part] = value;
|
||||
assert.throws(() => verifyNormalRecords(normal, manifest, producer), {
|
||||
code: "ERR_ASSERTION",
|
||||
});
|
||||
});
|
||||
|
||||
for (const side of ["client_conn", "server_conn"])
|
||||
for (const [field, value] of [
|
||||
["tls_version", null],
|
||||
["tls_version", "changed-producer-label"],
|
||||
["error", "changed-connection-error"],
|
||||
["error", undefined],
|
||||
["timestamp_tls_setup", 0],
|
||||
["timestamp_tls_setup", 1_700_000_000],
|
||||
])
|
||||
test(`positive assertions reject ${side}.${field} changed to ${String(value)}`, () => {
|
||||
const { normal, manifest, producer } = connectionFactsFixture();
|
||||
normal.records[0].reported[side][field] = value;
|
||||
assert.throws(() => verifyNormalRecords(normal, manifest, producer), {
|
||||
code: "ERR_ASSERTION",
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,143 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { mkdir, readFile, rename, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import { setTimeout } from "node:timers/promises";
|
||||
import { cleanupOwnedProcessGroup } from "../../dist/process/ProcessOwnership.js";
|
||||
import { observeOwnedProcessGroups } from "../../dist/process/ProcessOwnershipObservation.js";
|
||||
|
||||
export const describeError = (error) => ({
|
||||
name: error?.name ?? "Error",
|
||||
message: error?.message ?? String(error),
|
||||
...(error?.stack ? { stack: error.stack } : {}),
|
||||
});
|
||||
|
||||
export async function attemptAll(actions) {
|
||||
const failures = [];
|
||||
for (const [name, action] of actions) {
|
||||
try {
|
||||
await action();
|
||||
} catch (error) {
|
||||
failures.push({ action: name, ...describeError(error) });
|
||||
}
|
||||
}
|
||||
return failures;
|
||||
}
|
||||
|
||||
async function latest(root, report) {
|
||||
const temporary = join(root, `.verification-${report.run_id}.tmp`);
|
||||
await writeFile(temporary, JSON.stringify(report, null, 2), { flag: "wx" });
|
||||
await rename(temporary, join(root, "verification-latest.json"));
|
||||
}
|
||||
|
||||
async function withLatestLock(root, action) {
|
||||
// Keep this pathname stable: removing it could split concurrent lock owners.
|
||||
// SQLite releases its OS-backed lock when the connection/process closes.
|
||||
const database = new DatabaseSync(join(root, ".verification-lock.sqlite"));
|
||||
try {
|
||||
for (;;) {
|
||||
try {
|
||||
database.exec("BEGIN EXCLUSIVE");
|
||||
break;
|
||||
} catch (error) {
|
||||
if (
|
||||
error.code !== "ERR_SQLITE_ERROR" ||
|
||||
(error.errcode !== 5 && error.errcode !== 6)
|
||||
)
|
||||
throw error;
|
||||
// Yield while another process (or async operation here) owns the lock.
|
||||
// A synchronous busy timeout would block that operation's fs callbacks.
|
||||
await setTimeout(10);
|
||||
}
|
||||
}
|
||||
return await action();
|
||||
} finally {
|
||||
database.close();
|
||||
}
|
||||
}
|
||||
|
||||
export async function prepareAttempt(root, inspectExisting, run) {
|
||||
if (!inspectExisting) await mkdir(root, { mode: 0o700 });
|
||||
const directory = join(root, "verifications", run.run_id);
|
||||
await mkdir(directory, { recursive: true, mode: 0o700 });
|
||||
await withLatestLock(root, async () => {
|
||||
// Preserve older reports as history, without leaving an ambiguous success
|
||||
// at the capture root. Registration is the shared marker's ordering point.
|
||||
try {
|
||||
await rename(
|
||||
join(root, "verification.json"),
|
||||
join(directory, "legacy-verification.json"),
|
||||
);
|
||||
} catch (error) {
|
||||
if (error.code !== "ENOENT") throw error;
|
||||
}
|
||||
await latest(root, {
|
||||
run_id: run.run_id,
|
||||
status: "running",
|
||||
started_at: new Date().toISOString(),
|
||||
directory,
|
||||
});
|
||||
});
|
||||
return { directory, captureRoot: join(root, "captures") };
|
||||
}
|
||||
|
||||
export async function publishAttempt(root, directory, report) {
|
||||
await writeFile(
|
||||
join(directory, "result.json"),
|
||||
JSON.stringify(report, null, 2),
|
||||
{ flag: "wx" },
|
||||
);
|
||||
if (report.status === "passed")
|
||||
await writeFile(
|
||||
join(directory, "verification.json"),
|
||||
JSON.stringify(report, null, 2),
|
||||
{ flag: "wx" },
|
||||
);
|
||||
await withLatestLock(root, async () => {
|
||||
const current = JSON.parse(
|
||||
await readFile(join(root, "verification-latest.json"), "utf8"),
|
||||
);
|
||||
if (current.run_id !== report.run_id) return;
|
||||
// Ownership check and replacement share one lock with attempt registration.
|
||||
// An older attempt still keeps its reports, but cannot replace this marker.
|
||||
await latest(root, {
|
||||
run_id: report.run_id,
|
||||
status: report.status,
|
||||
directory,
|
||||
finished_at: report.finished_at,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export async function finishProducerGroups(captureRoot, runId) {
|
||||
let leases;
|
||||
try {
|
||||
leases = (await readFile(join(captureRoot, "process-groups.jsonl"), "utf8"))
|
||||
.trim()
|
||||
.split("\n")
|
||||
.filter(Boolean)
|
||||
.map(JSON.parse);
|
||||
} catch (error) {
|
||||
if (error.code === "ENOENT") return { groups: [], observation: [] };
|
||||
throw error;
|
||||
}
|
||||
for (const lease of leases) assert.equal(lease.runId, runId);
|
||||
const failures = await attemptAll(
|
||||
leases.map((lease) => [
|
||||
`producer-group-${lease.processGroupId}`,
|
||||
async () => {
|
||||
const result = await cleanupOwnedProcessGroup(lease);
|
||||
assert.equal(result.cleaned, true, JSON.stringify(result));
|
||||
},
|
||||
]),
|
||||
);
|
||||
// Inspect even when one close failed; the groups can outlive their bootloader
|
||||
// parent and disappear from a plain descendant-only observation.
|
||||
const observation = [
|
||||
...(await observeOwnedProcessGroups(
|
||||
runId,
|
||||
leases.map((lease) => lease.processGroupId),
|
||||
)),
|
||||
];
|
||||
return { groups: leases, failures, observation };
|
||||
}
|
||||
@@ -0,0 +1,433 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { test } from "node:test";
|
||||
import { spawn } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { once } from "node:events";
|
||||
import { setTimeout } from "node:timers/promises";
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import {
|
||||
attemptAll,
|
||||
prepareAttempt,
|
||||
publishAttempt,
|
||||
finishProducerGroups,
|
||||
} from "./http3-verifier-lifecycle.mjs";
|
||||
import { cleanupOwnedProcessGroup } from "../../dist/process/ProcessOwnership.js";
|
||||
import { runOwnedCommand } from "../../dist/process/OwnedCommand.js";
|
||||
import { verifyConcurrency } from "./http3-capture-assertions.mjs";
|
||||
|
||||
async function waitForFile(path) {
|
||||
const signal = AbortSignal.timeout(5_000);
|
||||
for (;;) {
|
||||
try {
|
||||
await readFile(path);
|
||||
return;
|
||||
} catch (error) {
|
||||
if (error.code !== "ENOENT") throw error;
|
||||
}
|
||||
await setTimeout(10, undefined, { signal });
|
||||
}
|
||||
}
|
||||
|
||||
function startAttemptWorker(root, report) {
|
||||
const worker = `
|
||||
import { readFile, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { setTimeout } from "node:timers/promises";
|
||||
import { prepareAttempt, publishAttempt } from ${JSON.stringify(
|
||||
new URL("./http3-verifier-lifecycle.mjs", import.meta.url).href,
|
||||
)};
|
||||
const [root, serialized] = process.argv.slice(1);
|
||||
const report = JSON.parse(serialized);
|
||||
const attempt = await prepareAttempt(root, true, report);
|
||||
await writeFile(join(root, report.run_id + ".ready"), "");
|
||||
for (;;) {
|
||||
try { await readFile(join(root, report.run_id + ".finish")); break; }
|
||||
catch (error) { if (error.code !== "ENOENT") throw error; }
|
||||
await setTimeout(10);
|
||||
}
|
||||
await publishAttempt(root, attempt.directory, report);
|
||||
`;
|
||||
return runOwnedCommand(
|
||||
{
|
||||
command: process.execPath,
|
||||
arguments: [
|
||||
"--input-type=module",
|
||||
"--eval",
|
||||
worker,
|
||||
root,
|
||||
JSON.stringify(report),
|
||||
],
|
||||
cwd: root,
|
||||
hostEnvironment: process.env,
|
||||
runId: randomUUID(),
|
||||
},
|
||||
{ timeoutMs: 10_000, diagnosticBytes: 4096 },
|
||||
).then(
|
||||
(result) => ({ result }),
|
||||
(error) => ({ error }),
|
||||
);
|
||||
}
|
||||
|
||||
test("cleanup continues through failed closes and reaches its audit", async () => {
|
||||
const attempted = [];
|
||||
const failures = await attemptAll(
|
||||
["client", "transport", "audit"].map((name) => [
|
||||
name,
|
||||
async () => {
|
||||
attempted.push(name);
|
||||
if (name !== "audit") throw new Error(`${name} failed`);
|
||||
},
|
||||
]),
|
||||
);
|
||||
assert.deepEqual(attempted, ["client", "transport", "audit"]);
|
||||
assert.deepEqual(
|
||||
failures.map((failure) => failure.action),
|
||||
["client", "transport"],
|
||||
);
|
||||
});
|
||||
|
||||
test("a failed reinspection cannot leave the earlier success as the current result", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "rea-http3-results-"));
|
||||
try {
|
||||
await writeFile(
|
||||
join(root, "verification.json"),
|
||||
'{"status":"passed","run_id":"legacy"}',
|
||||
);
|
||||
const first = await prepareAttempt(root, true, { run_id: "first" });
|
||||
await publishAttempt(root, first.directory, {
|
||||
run_id: "first",
|
||||
status: "passed",
|
||||
});
|
||||
const next = await prepareAttempt(root, true, { run_id: "next" });
|
||||
assert.equal(
|
||||
JSON.parse(await readFile(join(root, "verification-latest.json"))).status,
|
||||
"running",
|
||||
);
|
||||
await publishAttempt(root, next.directory, {
|
||||
run_id: "next",
|
||||
status: "failed",
|
||||
error: { message: "new failure" },
|
||||
});
|
||||
assert.equal(
|
||||
JSON.parse(await readFile(join(root, "verification-latest.json"))).run_id,
|
||||
"next",
|
||||
);
|
||||
assert.equal(
|
||||
JSON.parse(await readFile(join(root, "verification-latest.json"))).status,
|
||||
"failed",
|
||||
);
|
||||
await assert.rejects(readFile(join(next.directory, "verification.json")), {
|
||||
code: "ENOENT",
|
||||
});
|
||||
await assert.rejects(readFile(join(root, "verification.json")), {
|
||||
code: "ENOENT",
|
||||
});
|
||||
assert.equal(
|
||||
JSON.parse(await readFile(join(first.directory, "verification.json")))
|
||||
.status,
|
||||
"passed",
|
||||
);
|
||||
assert.equal(
|
||||
JSON.parse(
|
||||
await readFile(join(first.directory, "legacy-verification.json")),
|
||||
).run_id,
|
||||
"legacy",
|
||||
);
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("separate overlapping attempts finishing in reverse order retain the newer failure", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "rea-http3-overlap-"));
|
||||
const commands = [];
|
||||
const failed = {
|
||||
run_id: "newer",
|
||||
status: "failed",
|
||||
error: { message: "newer inspection failed" },
|
||||
cleanup_failures: [{ action: "transport-close", message: "close failed" }],
|
||||
};
|
||||
let primaryError;
|
||||
const start = (report) => {
|
||||
const command = startAttemptWorker(root, report);
|
||||
commands.push(command);
|
||||
return command;
|
||||
};
|
||||
try {
|
||||
const older = start({ run_id: "older", status: "passed" });
|
||||
await waitForFile(join(root, "older.ready"));
|
||||
const newer = start(failed);
|
||||
await waitForFile(join(root, "newer.ready"));
|
||||
await writeFile(join(root, "newer.finish"), "");
|
||||
const newerResult = await newer;
|
||||
if (newerResult.error) throw newerResult.error;
|
||||
assert.equal(newerResult.result.exitCode, 0);
|
||||
assert.equal(newerResult.result.signal, null);
|
||||
const marker = await readFile(join(root, "verification-latest.json"));
|
||||
await writeFile(join(root, "older.finish"), "");
|
||||
const olderResult = await older;
|
||||
if (olderResult.error) throw olderResult.error;
|
||||
assert.equal(olderResult.result.exitCode, 0);
|
||||
assert.equal(olderResult.result.signal, null);
|
||||
assert.deepEqual(
|
||||
await readFile(join(root, "verification-latest.json")),
|
||||
marker,
|
||||
);
|
||||
assert.deepEqual(
|
||||
JSON.parse(await readFile(join(root, "verifications/newer/result.json"))),
|
||||
failed,
|
||||
);
|
||||
await assert.rejects(
|
||||
readFile(join(root, "verifications/newer/verification.json")),
|
||||
{
|
||||
code: "ENOENT",
|
||||
},
|
||||
);
|
||||
assert.equal(
|
||||
JSON.parse(
|
||||
await readFile(join(root, "verifications/older/verification.json")),
|
||||
).status,
|
||||
"passed",
|
||||
);
|
||||
} catch (error) {
|
||||
primaryError = error;
|
||||
} finally {
|
||||
// Release both barriers even after an assertion/setup failure. Each command
|
||||
// retains native supervision through termination before the root is removed.
|
||||
const failures = await attemptAll(
|
||||
["older", "newer"].map((name) => [
|
||||
name,
|
||||
() => writeFile(join(root, name + ".finish"), ""),
|
||||
]),
|
||||
);
|
||||
const outcomes = await Promise.all(commands);
|
||||
const retained = outcomes.filter(
|
||||
({ error }) => error?.cleanupFailure || error?.cleanupResult,
|
||||
);
|
||||
if (retained.length)
|
||||
primaryError = new AggregateError(
|
||||
[primaryError, ...retained.map(({ error }) => error)].filter(Boolean),
|
||||
`Worker cleanup failed; retained test evidence at ${root}`,
|
||||
{ cause: primaryError },
|
||||
);
|
||||
else
|
||||
failures.push(
|
||||
...(await attemptAll([
|
||||
["root", () => rm(root, { recursive: true, force: true })],
|
||||
])),
|
||||
);
|
||||
if (failures.length)
|
||||
primaryError = new AggregateError(
|
||||
[primaryError, ...failures].filter(Boolean),
|
||||
"Test cleanup failed",
|
||||
{ cause: primaryError },
|
||||
);
|
||||
}
|
||||
if (primaryError) throw primaryError;
|
||||
});
|
||||
|
||||
test("publication waits for the lock before reading the current owner", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "rea-http3-lock-"));
|
||||
let database;
|
||||
let publication;
|
||||
let primaryError;
|
||||
try {
|
||||
const older = await prepareAttempt(root, true, { run_id: "older" });
|
||||
database = new DatabaseSync(join(root, ".verification-lock.sqlite"));
|
||||
database.exec("BEGIN EXCLUSIVE");
|
||||
publication = publishAttempt(root, older.directory, {
|
||||
run_id: "older",
|
||||
status: "passed",
|
||||
});
|
||||
await waitForFile(join(older.directory, "verification.json"));
|
||||
const marker = join(root, "verification-latest.json");
|
||||
assert.equal(JSON.parse(await readFile(marker)).status, "running");
|
||||
// This connection represents another marker writer. Its replacement must
|
||||
// be observed only after the blocked publisher acquires the same lock.
|
||||
const newer = { run_id: "newer", status: "failed" };
|
||||
await writeFile(marker, JSON.stringify(newer));
|
||||
database.close();
|
||||
database = undefined;
|
||||
await publication;
|
||||
assert.deepEqual(JSON.parse(await readFile(marker)), newer);
|
||||
} catch (error) {
|
||||
primaryError = error;
|
||||
} finally {
|
||||
const failures = await attemptAll([
|
||||
["lock", () => database?.close()],
|
||||
[
|
||||
"publication",
|
||||
async () => {
|
||||
if (publication) await publication;
|
||||
},
|
||||
],
|
||||
["root", () => rm(root, { recursive: true, force: true })],
|
||||
]);
|
||||
if (failures.length)
|
||||
primaryError = new AggregateError(
|
||||
[primaryError, ...failures].filter(Boolean),
|
||||
"Test cleanup failed",
|
||||
{ cause: primaryError },
|
||||
);
|
||||
}
|
||||
if (primaryError) throw primaryError;
|
||||
});
|
||||
|
||||
test("a failed marker read releases the lock and preserves the attempt report", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "rea-http3-lock-failure-"));
|
||||
try {
|
||||
const older = await prepareAttempt(root, true, { run_id: "older" });
|
||||
await writeFile(join(root, "verification-latest.json"), "{");
|
||||
const report = {
|
||||
run_id: "older",
|
||||
status: "failed",
|
||||
error: { message: "inspection failed" },
|
||||
};
|
||||
await assert.rejects(
|
||||
publishAttempt(root, older.directory, report),
|
||||
SyntaxError,
|
||||
);
|
||||
assert.deepEqual(
|
||||
JSON.parse(await readFile(join(older.directory, "result.json"))),
|
||||
report,
|
||||
);
|
||||
const next = await prepareAttempt(root, true, { run_id: "next" });
|
||||
await publishAttempt(root, next.directory, {
|
||||
run_id: "next",
|
||||
status: "passed",
|
||||
});
|
||||
assert.equal(
|
||||
JSON.parse(await readFile(join(root, "verification-latest.json"))).run_id,
|
||||
"next",
|
||||
);
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("dispatch reverses origin completion; client and capture may arrive in other orders", () => {
|
||||
const names = ["parallel-0", "parallel-1", "parallel-2"];
|
||||
const header = (text) => ({ base64: Buffer.from(text).toString("base64") });
|
||||
const messages = Object.fromEntries(
|
||||
names.map((name) => [
|
||||
name,
|
||||
{
|
||||
headers: [[header("x-lab-case"), header(name)]],
|
||||
body: { base64: header(name).base64 },
|
||||
},
|
||||
]),
|
||||
);
|
||||
const event = (kind, index, connection, monotonic) => ({
|
||||
event: kind,
|
||||
case: names[index],
|
||||
stream_id: index * 4,
|
||||
connection,
|
||||
monotonic,
|
||||
...messages[names[index]],
|
||||
});
|
||||
const manifest = {
|
||||
requests: messages,
|
||||
responses: messages,
|
||||
events: [
|
||||
...[0, 1, 2].map((i) => event("request-sent", i, "client", i)),
|
||||
...[1, 2, 0].map((i) => event("request-received", i, "origin", 10)),
|
||||
...[0, 2, 1].map((i) => event("response-sent", i, "origin", 20)),
|
||||
...[2, 0, 1].map((i) => event("response-received", i, "client", 30)),
|
||||
...["client", "origin"].map((connection) => ({
|
||||
event: "handshake",
|
||||
connection,
|
||||
alpn: "h3",
|
||||
quic_version: 1,
|
||||
})),
|
||||
],
|
||||
};
|
||||
const record = (name) => ({
|
||||
reported: {
|
||||
id: name,
|
||||
request: { headers: [["x-lab-case", name]] },
|
||||
client_conn: { id: "client-uuid" },
|
||||
server_conn: { id: "origin-uuid" },
|
||||
},
|
||||
});
|
||||
const normal = {
|
||||
records: [...[1, 0, 2].map((i) => record(names[i])), record("curl")],
|
||||
};
|
||||
const orders = verifyConcurrency(normal, manifest);
|
||||
assert.deepEqual(orders.dispatched_at_origin, [
|
||||
"parallel-0",
|
||||
"parallel-2",
|
||||
"parallel-1",
|
||||
]);
|
||||
manifest.events = manifest.events.filter(
|
||||
(item) => item.event !== "response-sent",
|
||||
);
|
||||
manifest.events.push(
|
||||
...[2, 1, 0].map((i) => event("response-sent", i, "origin", 20)),
|
||||
);
|
||||
assert.throws(() => verifyConcurrency(normal, manifest), {
|
||||
code: "ERR_ASSERTION",
|
||||
});
|
||||
});
|
||||
|
||||
test("cleanup verifies and retires a separately owned producer group", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "rea-http3-group-"));
|
||||
const runId = randomUUID();
|
||||
const child = spawn("/usr/bin/sleep", ["60"], {
|
||||
detached: true,
|
||||
env: { ...process.env, REA_PROCESS_RUN_ID: runId },
|
||||
stdio: "ignore",
|
||||
});
|
||||
let primaryError;
|
||||
let cleaned = false;
|
||||
try {
|
||||
await once(child, "spawn");
|
||||
await writeFile(
|
||||
join(root, "process-groups.jsonl"),
|
||||
JSON.stringify({
|
||||
runId,
|
||||
leaderPid: child.pid,
|
||||
processGroupId: child.pid,
|
||||
}) + "\n",
|
||||
);
|
||||
const result = await finishProducerGroups(root, runId);
|
||||
assert.deepEqual(result.failures, []);
|
||||
assert.equal(result.observation[0][1].state, "empty");
|
||||
} catch (error) {
|
||||
primaryError = error;
|
||||
} finally {
|
||||
const failures = await attemptAll([
|
||||
[
|
||||
"process",
|
||||
async () => {
|
||||
const result = await cleanupOwnedProcessGroup({
|
||||
runId,
|
||||
leaderPid: child.pid,
|
||||
processGroupId: child.pid,
|
||||
});
|
||||
assert.equal(
|
||||
result.cleaned,
|
||||
true,
|
||||
`Retained ownership evidence at ${root}: ${JSON.stringify(result)}`,
|
||||
);
|
||||
cleaned = true;
|
||||
},
|
||||
],
|
||||
[
|
||||
"root",
|
||||
async () => {
|
||||
if (cleaned) await rm(root, { recursive: true, force: true });
|
||||
},
|
||||
],
|
||||
]);
|
||||
if (failures.length)
|
||||
primaryError = new AggregateError(
|
||||
[primaryError, ...failures].filter(Boolean),
|
||||
"Test cleanup failed",
|
||||
{ cause: primaryError },
|
||||
);
|
||||
}
|
||||
if (primaryError) throw primaryError;
|
||||
});
|
||||
@@ -0,0 +1,408 @@
|
||||
#!/usr/bin/env node
|
||||
import assert from "node:assert/strict";
|
||||
import { createHash } from "node:crypto";
|
||||
import { readFile, writeFile } from "node:fs/promises";
|
||||
import { isAbsolute, join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { Client } from "@modelcontextprotocol/client";
|
||||
import { StdioClientTransport } from "@modelcontextprotocol/client/stdio";
|
||||
import { Ajv2020 } from "ajv/dist/2020.js";
|
||||
import { runOwnedCommand } from "../dist/process/OwnedCommand.js";
|
||||
import { toolContract } from "../dist/contracts/toolContracts.js";
|
||||
import { createVerifierRun, completeVerifierRun } from "./lib/verifier-run.mjs";
|
||||
import {
|
||||
bytes,
|
||||
byCase,
|
||||
verifyNormalRecords,
|
||||
requireHttp3,
|
||||
verifyConcurrency,
|
||||
} from "./lib/http3-capture-assertions.mjs";
|
||||
import {
|
||||
attemptAll,
|
||||
describeError,
|
||||
prepareAttempt,
|
||||
publishAttempt,
|
||||
finishProducerGroups,
|
||||
} from "./lib/http3-verifier-lifecycle.mjs";
|
||||
import { mcpTextValue } from "./lib/mcp-verifier-results.mjs";
|
||||
|
||||
const root = process.argv[2];
|
||||
const inspectExisting = process.argv[3] === "--inspect-existing";
|
||||
const mitmdump = process.env.REA_MITMDUMP_COMMAND;
|
||||
const python = process.env.REA_HTTP3_PYTHON_COMMAND;
|
||||
assert.equal(
|
||||
process.platform,
|
||||
"linux",
|
||||
"HTTP/3 native verification requires Linux",
|
||||
);
|
||||
assert.ok(
|
||||
root && isAbsolute(root),
|
||||
"Select an absolute, new evidence directory",
|
||||
);
|
||||
assert.ok(
|
||||
mitmdump && isAbsolute(mitmdump),
|
||||
"Select caller-supplied mitmdump 12.2.3",
|
||||
);
|
||||
if (!inspectExisting)
|
||||
assert.ok(
|
||||
python && isAbsolute(python),
|
||||
"Select a Python environment with aioquic 1.2.0",
|
||||
);
|
||||
const run = createVerifierRun();
|
||||
const cwd = fileURLToPath(new URL("..", import.meta.url));
|
||||
const entrypoint = fileURLToPath(new URL("./rea.mjs", import.meta.url));
|
||||
const environment = Object.fromEntries(
|
||||
Object.entries(process.env).filter(([, value]) => typeof value === "string"),
|
||||
);
|
||||
const client = new Client({
|
||||
name: "real-http3-capture-verifier",
|
||||
version: "1",
|
||||
});
|
||||
const transport = new StdioClientTransport({
|
||||
command: process.execPath,
|
||||
args: [entrypoint, "mcp"],
|
||||
env: environment,
|
||||
stderr: "pipe",
|
||||
});
|
||||
const contract = toolContract("inspect_web_network_capture");
|
||||
const maxMcpStderrBytes = 64 * 1024;
|
||||
let mcpStderr = Buffer.alloc(0);
|
||||
transport.stderr?.on("data", (chunk) => {
|
||||
mcpStderr = Buffer.concat([mcpStderr, chunk]);
|
||||
if (mcpStderr.length > maxMcpStderrBytes)
|
||||
mcpStderr = mcpStderr.subarray(-maxMcpStderrBytes);
|
||||
});
|
||||
const checks = [];
|
||||
const retainedEvidence = [];
|
||||
const originals = new Map();
|
||||
let directory;
|
||||
let captureRoot = join(root, "captures");
|
||||
let primaryError;
|
||||
let verifier;
|
||||
let producerAudit;
|
||||
let cleanupFailures = [];
|
||||
let validateOutput;
|
||||
|
||||
const digest = (bytes) => createHash("sha256").update(bytes).digest("hex");
|
||||
const readJson = async (path) => JSON.parse(await readFile(path, "utf8"));
|
||||
const readLines = async (path) =>
|
||||
(await readFile(path, "utf8"))
|
||||
.trim()
|
||||
.split("\n")
|
||||
.filter(Boolean)
|
||||
.map(JSON.parse);
|
||||
|
||||
async function cli(input, name) {
|
||||
const command = await runOwnedCommand(
|
||||
{
|
||||
command: process.execPath,
|
||||
arguments: [
|
||||
entrypoint,
|
||||
"inspect-web-network-capture",
|
||||
input.capture_path,
|
||||
input.format,
|
||||
"--json",
|
||||
],
|
||||
cwd,
|
||||
hostEnvironment: environment,
|
||||
runId: run.run_id,
|
||||
},
|
||||
{
|
||||
timeoutMs: 40_000,
|
||||
diagnosticBytes: 8 * 1024 * 1024,
|
||||
},
|
||||
);
|
||||
assert.notEqual(command.diagnosticTruncated, true);
|
||||
await writeFile(
|
||||
join(directory, `${name}.cli.stdout.json`),
|
||||
command.stdout.text,
|
||||
);
|
||||
await writeFile(
|
||||
join(directory, `${name}.cli.stderr.log`),
|
||||
command.stderr.text,
|
||||
);
|
||||
const value = JSON.parse(command.stdout.text);
|
||||
return contract.outputSchema.parse(value);
|
||||
}
|
||||
|
||||
async function mcp(input, name) {
|
||||
const reply = await client.callTool({
|
||||
name: contract.name,
|
||||
arguments: input,
|
||||
});
|
||||
await writeFile(
|
||||
join(directory, `${name}.mcp.reply.json`),
|
||||
JSON.stringify(reply, null, 2),
|
||||
);
|
||||
const value = JSON.parse(mcpTextValue(reply));
|
||||
assert.notEqual(reply.isError, true, mcpTextValue(reply));
|
||||
assert.ok(
|
||||
validateOutput(reply.structuredContent),
|
||||
JSON.stringify(validateOutput.errors),
|
||||
);
|
||||
const evidence = contract.outputSchema.parse(value);
|
||||
assert.deepEqual(reply.structuredContent, value);
|
||||
retainedEvidence.push(evidence);
|
||||
return evidence;
|
||||
}
|
||||
|
||||
async function inspect(name) {
|
||||
const input = {
|
||||
capture_path: join(captureRoot, `${name}.mitm`),
|
||||
format: "mitmproxy",
|
||||
};
|
||||
const cliResult = await cli(input, name);
|
||||
const mcpResult = await mcp(input, name);
|
||||
assert.deepEqual(cliResult.normalized_result, mcpResult.normalized_result);
|
||||
for (const evidence of [cliResult, mcpResult]) {
|
||||
assert.equal(evidence.authority, "historical-reference");
|
||||
assert.equal(evidence.confidence, "observed");
|
||||
assert.equal(evidence.provider.version, "12.2.3");
|
||||
assert.equal(evidence.normalized_result.runtime_attribution, "unknown");
|
||||
assert.deepEqual(evidence.provider, evidence.normalized_result.decoder);
|
||||
assert.equal(
|
||||
evidence.subject.digest.sha256,
|
||||
digest(await readFile(input.capture_path)),
|
||||
);
|
||||
}
|
||||
checks.push({
|
||||
case: name,
|
||||
outcome: "passed",
|
||||
records: mcpResult.normalized_result.records.length,
|
||||
});
|
||||
return mcpResult.normalized_result;
|
||||
}
|
||||
|
||||
try {
|
||||
({ directory, captureRoot } = await prepareAttempt(
|
||||
root,
|
||||
inspectExisting,
|
||||
run,
|
||||
));
|
||||
if (!inspectExisting) {
|
||||
const produced = await runOwnedCommand(
|
||||
{
|
||||
command: python,
|
||||
arguments: [
|
||||
fileURLToPath(new URL("./fixtures/http3-lab.py", import.meta.url)),
|
||||
"--mitmdump",
|
||||
mitmdump,
|
||||
"--output",
|
||||
captureRoot,
|
||||
],
|
||||
cwd,
|
||||
hostEnvironment: environment,
|
||||
runId: run.run_id,
|
||||
},
|
||||
{ timeoutMs: 120_000, diagnosticBytes: 1024 * 1024 },
|
||||
);
|
||||
await writeFile(
|
||||
join(directory, "producer.stdout.log"),
|
||||
produced.stdout.text,
|
||||
);
|
||||
await writeFile(
|
||||
join(directory, "producer.stderr.log"),
|
||||
produced.stderr.text,
|
||||
);
|
||||
}
|
||||
const manifest = await readJson(join(captureRoot, "manifest.json"));
|
||||
assert.equal(manifest.aioquic, "1.2.0");
|
||||
assert.equal(manifest.profile, "http3-minimal-v1");
|
||||
await client.connect(transport);
|
||||
const advertised = (await client.listTools()).tools.find(
|
||||
(tool) => tool.name === contract.name,
|
||||
);
|
||||
assert.ok(advertised?.outputSchema, "Missing named advertised schema");
|
||||
const ajv = new Ajv2020({ strict: false, validateFormats: false });
|
||||
assert.ok(ajv.validateSchema(advertised.inputSchema));
|
||||
assert.ok(ajv.validateSchema(advertised.outputSchema));
|
||||
validateOutput = ajv.compile(advertised.outputSchema);
|
||||
const inputValidator = ajv.compile(advertised.inputSchema);
|
||||
assert.ok(
|
||||
inputValidator({
|
||||
capture_path: join(captureRoot, "normal.mitm"),
|
||||
format: "mitmproxy",
|
||||
}),
|
||||
);
|
||||
assert.equal(
|
||||
inputValidator({
|
||||
capture_path: join(captureRoot, "normal.mitm"),
|
||||
format: "mitmproxy",
|
||||
stream_id: 0,
|
||||
}),
|
||||
false,
|
||||
);
|
||||
await writeFile(
|
||||
join(directory, "advertised-tool.json"),
|
||||
JSON.stringify(advertised, null, 2),
|
||||
);
|
||||
|
||||
for (const name of ["normal", "fallback"])
|
||||
originals.set(name, await readFile(join(captureRoot, `${name}.mitm`)));
|
||||
const normal = await inspect("normal");
|
||||
assert.equal(normal.total_records, 4);
|
||||
const finalProducer = (
|
||||
await readLines(join(captureRoot, "normal.producer.jsonl"))
|
||||
).filter((event) => ["response", "error"].includes(event.event));
|
||||
verifyNormalRecords(normal, manifest, finalProducer);
|
||||
const orders = verifyConcurrency(normal, manifest);
|
||||
assert.equal(
|
||||
(await readJson(join(captureRoot, "curl.json"))).http_version,
|
||||
"3",
|
||||
);
|
||||
assert.deepEqual(
|
||||
await readFile(join(captureRoot, "curl.body")),
|
||||
bytes(byCase(normal, "curl"), "/response/content"),
|
||||
);
|
||||
checks.push({
|
||||
case: "concurrency-bytes",
|
||||
outcome: "passed",
|
||||
parallel_streams: 3,
|
||||
orders,
|
||||
});
|
||||
const fallback = await inspect("fallback");
|
||||
assert.equal(fallback.total_records, 1);
|
||||
const record = byCase(fallback, "fallback");
|
||||
assert.throws(
|
||||
() => requireHttp3(record),
|
||||
{ code: "ERR_ASSERTION" },
|
||||
"The same positive HTTP/3 check must reject the fallback record",
|
||||
);
|
||||
assert.equal(record.reported.client_conn.transport_protocol, "tcp");
|
||||
assert.equal(record.reported.server_conn.transport_protocol, "tcp");
|
||||
assert.equal(record.reported.request.http_version, "HTTP/2.0");
|
||||
assert.equal(record.reported.response.http_version, "HTTP/1.1");
|
||||
assert.equal(record.reported.server_conn.alpn, "http/1.1");
|
||||
assert.equal(
|
||||
(await readJson(join(captureRoot, "fallback.json"))).http_version,
|
||||
"2",
|
||||
);
|
||||
checks.push({
|
||||
case: "fallback-detected",
|
||||
outcome: "passed",
|
||||
client_http: "2",
|
||||
origin_http: "1.1",
|
||||
});
|
||||
|
||||
const bundleReply = await client.callTool({
|
||||
name: "get_evidence_bundle",
|
||||
arguments: {},
|
||||
});
|
||||
assert.notEqual(bundleReply.isError, true, mcpTextValue(bundleReply));
|
||||
const bundle = JSON.parse(mcpTextValue(bundleReply)).result;
|
||||
const records = bundle.records;
|
||||
assert.ok(Array.isArray(records), "Evidence bundle omitted its records");
|
||||
for (const evidence of retainedEvidence) {
|
||||
const retained = records.find(
|
||||
(record) => record.evidence_id === evidence.evidence_id,
|
||||
);
|
||||
assert.deepEqual(
|
||||
retained,
|
||||
evidence,
|
||||
`MCP must retain the complete Evidence for ${evidence.evidence_id}`,
|
||||
);
|
||||
}
|
||||
await writeFile(
|
||||
join(directory, "session-bundle.reply.json"),
|
||||
JSON.stringify(bundleReply, null, 2),
|
||||
);
|
||||
checks.push({
|
||||
case: "mcp-session-retention",
|
||||
outcome: "passed",
|
||||
records: retainedEvidence.length,
|
||||
});
|
||||
} catch (error) {
|
||||
primaryError = error;
|
||||
} finally {
|
||||
cleanupFailures = await attemptAll([
|
||||
["client-close", () => client.close()],
|
||||
["transport-close", () => transport.close()],
|
||||
[
|
||||
"producer-groups",
|
||||
async () => {
|
||||
if (!inspectExisting) {
|
||||
producerAudit = await finishProducerGroups(captureRoot, run.run_id);
|
||||
assert.deepEqual(producerAudit.failures ?? [], []);
|
||||
for (const [, observation] of producerAudit.observation)
|
||||
assert.equal(
|
||||
observation.state,
|
||||
"empty",
|
||||
JSON.stringify(observation),
|
||||
);
|
||||
}
|
||||
},
|
||||
],
|
||||
...[...originals].map(([name, original]) => [
|
||||
`input-immutability-${name}`,
|
||||
async () => {
|
||||
assert.deepEqual(
|
||||
await readFile(join(captureRoot, `${name}.mitm`)),
|
||||
original,
|
||||
);
|
||||
},
|
||||
]),
|
||||
[
|
||||
"mcp-stderr-diagnostics",
|
||||
async () => {
|
||||
if (directory)
|
||||
await writeFile(join(directory, "mcp.stderr.log"), mcpStderr);
|
||||
},
|
||||
],
|
||||
[
|
||||
"failed-command-diagnostics",
|
||||
async () => {
|
||||
if (directory && primaryError?.snapshot) {
|
||||
await writeFile(
|
||||
join(directory, "failed-command.stdout.log"),
|
||||
primaryError.snapshot.stdout.text,
|
||||
);
|
||||
await writeFile(
|
||||
join(directory, "failed-command.stderr.log"),
|
||||
primaryError.snapshot.stderr.text,
|
||||
);
|
||||
}
|
||||
},
|
||||
],
|
||||
[
|
||||
"final-lineage",
|
||||
async () => {
|
||||
verifier = await completeVerifierRun(run);
|
||||
assert.equal(verifier.process_lineage.status, "verified");
|
||||
assert.deepEqual(verifier.process_lineage.descendants, []);
|
||||
},
|
||||
],
|
||||
]);
|
||||
}
|
||||
const report = {
|
||||
run_id: run.run_id,
|
||||
status: primaryError || cleanupFailures.length ? "failed" : "passed",
|
||||
finished_at: new Date().toISOString(),
|
||||
...(primaryError
|
||||
? {
|
||||
error: describeError(primaryError),
|
||||
mcp_stderr_log: "mcp.stderr.log",
|
||||
}
|
||||
: {}),
|
||||
cleanup_failures: cleanupFailures,
|
||||
checks,
|
||||
verifier,
|
||||
producer_audit: producerAudit,
|
||||
scope:
|
||||
"Real Linux reverse HTTP/3, mitmproxy 12.2.3, aioquic 1.2.0 and guest cURL",
|
||||
limitations: [
|
||||
"No local/WireGuard, browser, QUIC v2, migration, 0-RTT, mTLS or replay claim.",
|
||||
"No HAR, interruption, certificate-failure or body-completeness coverage in this profile.",
|
||||
],
|
||||
};
|
||||
if (directory) {
|
||||
try {
|
||||
await publishAttempt(root, directory, report);
|
||||
} catch (error) {
|
||||
console.error("Could not publish this attempt's result:", error);
|
||||
primaryError ??= error;
|
||||
}
|
||||
}
|
||||
if (primaryError) throw primaryError;
|
||||
if (cleanupFailures.length)
|
||||
throw new AggregateError(cleanupFailures, "HTTP/3 verifier cleanup failed");
|
||||
console.log(JSON.stringify(report, null, 2));
|
||||
@@ -45,6 +45,7 @@ import { verifyBrowserNetworkEvidence } from "./lib/browser-network-e2e.mjs";
|
||||
import { verifyBrowserScriptExport } from "./lib/browser-script-export-e2e.mjs";
|
||||
import { verifyBrowserModules } from "./lib/browser-module-e2e.mjs";
|
||||
import { verifyBrowserDomDestinations } from "./lib/browser-dom-destinations-e2e.mjs";
|
||||
import { verifyBrowserSocketTransport } from "./lib/browser-socket-e2e.mjs";
|
||||
import { verifyBrowserCaptureMetadataBudget } from "./lib/browser-capture-metadata-budget-e2e.mjs";
|
||||
import { artifactCliEvidence, artifactMcpResult } from "./lib/artifact-e2e.mjs";
|
||||
import { verifyScenarioEnvironment } from "./lib/browser-scenario-environment-e2e.mjs";
|
||||
@@ -234,6 +235,13 @@ try {
|
||||
|
||||
pageProxy = await startPageCdpProxy(endpoint);
|
||||
await verifyPageScopedTransport(provider, pageProxy, site.origin);
|
||||
const browserSocketTransport = await verifyBrowserSocketTransport({
|
||||
provider,
|
||||
profile,
|
||||
endpoint,
|
||||
target,
|
||||
origin: site.origin,
|
||||
});
|
||||
const attachedScenarioInput = browserScenario(
|
||||
{
|
||||
mode: "connect",
|
||||
@@ -377,6 +385,7 @@ try {
|
||||
bundle_cli_and_stdio_mcp: true,
|
||||
sessionEvents: session.value.timeline.length,
|
||||
pageScopedTransport: true,
|
||||
browserSocketTransport,
|
||||
screenshotBytes: screenshot.value.artifact.bytes,
|
||||
largeScreenshot,
|
||||
popupEvents,
|
||||
|
||||
Reference in New Issue
Block a user