rea/scripts/fixtures/generate-mitmproxy-capture.py

import importlib.util
import json
from pathlib import Path
from mitmproxy import connection, ctx, http
from mitmproxy.io import FlowWriter
from mitmproxy.io import tnetstring
from mitmproxy.addons.savehar import SaveHar
from mitmproxy.websocket import WebSocketData, WebSocketMessage

class Generator:
    def load(self, loader):
        loader.add_option("rea_fixture_root", str, "", "Owned synthetic fixture root")
        loader.add_option("rea_adapter_path", str, "", "REA native adapter under verification")

    def running(self):
        root = Path(ctx.options.rea_fixture_root)
        spec = importlib.util.spec_from_file_location("rea_capture_under_test", ctx.options.rea_adapter_path)
        adapter = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(adapter)
        classifications = [
            (adapter.CaptureFailure("format", "Malformed retained record.", "/record"), "format"),
            (adapter.CaptureFailure("input-limit", "Nesting budget.", "/extension"), "input-limit"),
            (MemoryError("payload must not enter diagnostics"), "resource-limit"),
            (RuntimeError("payload must not enter diagnostics"), "decoder"),
            (OSError("payload must not enter diagnostics"), "decoder"),
        ]
        for error, expected in classifications:
            reply = adapter.failure_reply(error)
            assert reply["reason"] == expected, reply
            assert "payload must not enter diagnostics" not in json.dumps(reply)
        for state, expected in [({"metadata": {1: b"bad-key"}}, "format"), ({"metadata": object()}, "decoder")]:
            try:
                adapter.project_record(state, [])
            except adapter.CaptureFailure as error:
                assert error.reason == expected, (error.reason, expected)
                assert error.pointer == "/metadata", error.pointer
            else:
                raise AssertionError("Invalid native representation was accepted")
        non_utf8 = adapter.project_record({"request": {"headers": [(b"Referer", b"https://byte-user:byte-secret@example.test/path\xff")] }}, [])
        assert non_utf8["reported"]["request"]["headers"][0][1] is None, non_utf8
        assert non_utf8["binary_fields"][-1]["state"] == "redacted", non_utf8
        for value in [" \tHTTPS://ows-user:ows-secret@example.test/path\t ", b" \tHTTPS://ows-user:ows-secret@example.test/path\t "]:
            projected = adapter.project_record({"request": {"headers": [(b"Referer", value)]}}, [])
            assert projected["reported"]["request"]["headers"][0][1] == " \tHTTPS://example.test/path\t ", projected
            assert "ows-secret" not in json.dumps(projected), projected
            if isinstance(value, bytes):
                assert projected["binary_fields"][-1]["state"] == "redacted", projected
        (root / "native-error-classifications.json").write_text(json.dumps({"passed": len(classifications) + 2}))
        first = http.HTTPFlow(connection.Client(peername=("127.0.0.1", 1), sockname=("127.0.0.1", 2)), connection.Server(address=("example.test", 80)))
        first.id = "original-producer-id"
        first.request = http.Request.make("POST", "http://example.test/a?token=ordinary", b"\x00\xffbody", [(b"Authorization", b"Bearer native-transport-secret"), (b"X-Duplicate", b"one"), (b"X-Duplicate", b"two")])
        first.request.path = "http://user:password@example.test/a?token=ordinary#fragment"
        first.response = http.Response.make(200, b"\x00\xfeanswer", [(b"Set-Cookie", b"session=cookie-secret"), (b"Location", b"https://redirect-user:redirect-password@example.test/b?token=ordinary")])
        first.metadata = {"big_integer": 9007199254740993, "nonfinite": float("inf"), "isLosslessNumber": True, "value": "ordinary-extension", "response": {"headers": [["Authorization", "ordinary-extension-credential-name"]]}}
        first.websocket = WebSocketData(messages=[WebSocketMessage(2, True, b"\x00\xfdmessage", 123.5)])
        first.backup()
        first.comment = "caller-selected-fixture"
        second = first.copy()
        second.id = first.id
        second.request.raw_content = None
        second.response.raw_content = b""
        second.websocket = None
        with (root / "flows.mitm").open("wb") as handle:
            writer = FlowWriter(handle)
            writer.add(first)
            writer.add(second)
        har_first, har_second = first.copy(), second.copy()
        har_first.request.path = "/a?token=ordinary#fragment"
        har_second.request.path = "/a?token=ordinary#fragment"
        har_first.response.raw_content = b"\x00\xff\x01\xfe"
        markers = {"sensitive": "REDACTED", "bracket": "literal[", "overlap": "secret", "ordinary": "unmarked"}
        har = SaveHar().make_har([har_first, har_second])
        har["log"]["entries"][0]["_markers"] = markers
        har["log"]["entries"][0]["_private_properties"] = {"private-property/~": {"number": 123, "text": "private-property"}, "kept": 7}
        (root / "producer.har").write_text(json.dumps(har))
        har_text = json.dumps(har)
        for name, members in [
            ("equal-number", '"_duplicate": 1, "_duplicate": 1'),
            ("equal-null", '"_duplicate": null, "_duplicate": null'),
            ("equal-object", '"_duplicate": {"value": 1}, "_duplicate": {"value": 1}'),
            ("escaped-key", '"_duplicate": 1, "\\u005fduplicate": 1'),
        ]:
            duplicate = har_text.replace('"cache": {}', '"cache": {}, ' + members, 1)
            assert duplicate != har_text
            (root / ("duplicate-" + name + ".har")).write_text(duplicate)
        duplicate = har_text.replace('"version": "1.2"', '"version": "1.2", "version": "1.2"', 1)
        assert duplicate != har_text
        (root / "duplicate-core.har").write_text(duplicate)
        prototype = json.loads(har_text)
        prototype["_extension"] = {"__proto__": {"preserved": 7}, "constructor": "ordinary"}
        (root / "prototype.har").write_text(json.dumps(prototype))
        prototype_native = first.get_state()
        prototype_native["metadata"]["__proto__"] = {"preserved": 7}
        with (root / "prototype.mitm").open("wb") as handle:
            tnetstring.dump(prototype_native, handle)
        invalid_har = json.loads(json.dumps(har))
        invalid_har["log"]["entries"][0]["response"]["content"].update({"encoding": "base64", "text": "AR=="})
        (root / "invalid-private-parent.har").write_text(json.dumps(invalid_har))
        strings = first.get_state()
        for state in [strings, strings["backup"]]:
            state["request"]["path"] = "https://native-user:string-password@example.test/string-path"
            state["request"]["authority"] = "authority-user:authority-password@example.test"
            state["request"]["headers"] = [(b"Referer", "https://referer-user:referer-password@example.test/from"), (b"Origin", "//origin-user:origin-password@example.test")]
            state["response"]["headers"] = [(b"Location", "https://location-user:location-password@example.test/to")]
            state["metadata"]["_markers"] = dict(markers)
            state["metadata"]["_private_properties"] = {"private-property/~": {"number": 123, "bytes": b"private-property"}, "kept": 7}
        with (root / "string-urls.mitm").open("wb") as handle:
            tnetstring.dump(strings, handle)
        strings["metadata"]["_private_properties"]["private-property/~"] = {1: b"unsupported-key"}
        with (root / "invalid-private-key.mitm").open("wb") as handle:
            tnetstring.dump(strings, handle)
        extension = "leaf"
        for _ in range(66):
            extension = {"child": extension}
        har["_extension"] = extension
        (root / "deep.har").write_text(json.dumps(har))
        deep = first.get_state()
        deep["_extension"] = extension
        with (root / "deep.mitm").open("wb") as handle:
            tnetstring.dump(deep, handle)
        for name, value in [("invalid-credential-key", {1: b"bad-key"}), ("deep-credential", extension)]:
            malformed = first.get_state()
            malformed["request"]["headers"] = [(b"Authorization", value)]
            with (root / (name + ".mitm")).open("wb") as handle:
                tnetstring.dump(malformed, handle)
        for prefix, name in [(" \t", "ows"), ("\r\n \v\f\x00\x1f", "controls")]:
            url = prefix + "HTTPS://ows-user:ows-secret@example.test/path?token=ordinary#fragment\t "
            relative = prefix + "//ows-user:ows-secret@example.test/path?token=ordinary#fragment\t "
            whitespace_har = json.loads((root / "producer.har").read_text())
            entry = whitespace_har["log"]["entries"][0]
            for side, header in [("response", "Location"), ("request", "Referer"), ("request", "Origin")]:
                entry[side]["headers"] = [{"name": header, "value": url}] + entry[side]["headers"]
            (root / (name + ".har")).write_text(json.dumps(whitespace_har))
            whitespace_native = first.get_state()
            for state in [whitespace_native, whitespace_native["backup"]]:
                state["request"]["headers"] = [(b"Referer", url.encode("utf-8")), (b"Origin", relative)]
                state["response"]["headers"] = [(b"Location", url)]
            with (root / (name + ".mitm")).open("wb") as handle:
                tnetstring.dump(whitespace_native, handle)
        non_utf8_native = first.get_state()
        for state in [non_utf8_native, non_utf8_native["backup"]]:
            state["request"]["path"] = b"https://byte-user:byte-secret@example.test/path\xff"
            state["request"]["authority"] = b"byte-user:byte-secret@example.test\xff"
            state["request"]["headers"] = [(b"Referer", b" https://byte-user:byte-secret@example.test/from\xff"), (b"Origin", b"//byte-user:byte-secret@example.test\xff")]
            state["response"]["headers"] = [(b"Location", b"https://byte-user:byte-secret@example.test/to\xff")]
        with (root / "non-utf8-urls.mitm").open("wb") as handle:
            tnetstring.dump(non_utf8_native, handle)
        (root / "oracle.json").write_text(json.dumps({"records": 2, "request_base64": "AP9ib2R5", "response_base64": "AP5hbnN3ZXI=", "websocket_base64": "AP1tZXNzYWdl", "id": first.id}))
        ctx.master.shutdown()

addons = [Generator()]