reverse-engineering-anything/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, [], 64)
            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")] }}, [], 64)
        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)]}}, [], 64)
            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
        body = b"A" * 12345
        state = {"response": {"content": body}}
        control = adapter.project_record(state, [], 64)
        field = control["binary_fields"][0]
        assert field["state"] == "retained" and field["bytes"] == len(body), control
        for literal in ["12345", "2345"]:
            # Only the generated byte count matches this exclusion.
            assert literal not in control["reported"]["response"]["content"]
            assert literal not in field["content_base64"] and literal not in field["sha256"]
            projected = adapter.project_record(state, [literal], 64)
            assert projected["reported"]["response"]["content"] is None, projected
            assert projected["binary_fields"] == [{"pointer": "/response/content", "representation": "producer-bytes", "state": "redacted", "content_base64": None, "bytes": None, "sha256": None}], projected
            assert {"pointer": "/response/content", "reason": "explicit-sensitive-value"} in projected["redactions"], projected
            assert literal not in json.dumps(projected), 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.metadata["numeric_spellings"] = {"small": 1e-7, "large": 1e20, "ordinary": 42}
        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()]