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>
65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
"""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()]
|