417 lines
19 KiB
Python
417 lines
19 KiB
Python
"""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()))
|