ebaaa79ca0
* test(http3): add real MITM capture verification lab * test(http3): register the optional verification command * fix(http3): preserve the latest verification attempt owner * test(cli): admit the dyld slice fixture on x64 hosts * test(http3): verify exact producer TLS connection facts * test(http3): verify readable capture fields against observed bytes * fix(comparison): recheck historical calls-only changes (#2223) * fix(comparison): preserve incomplete function call evidence Keep missing, unknown, partial and unresolved call relationships unknown in function comparisons and reconstruction claims. Aggregate IDA caller and callee coverage without overwriting uncertainty. Validate with 144 focused tests, typecheck/lint, docs checks and test discovery. Real providers were not verified; the full headless IDA session suite requires unavailable Windows native controls. * fix(comparison): recheck historical calls-only changes --------- Co-authored-by: morluto <76467478+morluto@users.noreply.github.com> * fix(android): make DEX descriptor pattern portable * fix(ci): isolate Linux Hopper process fixture inventory * fix(http3): retain MCP startup diagnostics --------- Co-authored-by: morluto <76467478+morluto@users.noreply.github.com> Co-authored-by: Lucien <164857476+dakjdakd@users.noreply.github.com>
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)
< |