From 0684f520dc0b49a66b386a7389dbd7b7fffb7bf6 Mon Sep 17 00:00:00 2001 From: David Tapiador Date: Fri, 25 Sep 2026 15:27:55 +0200 Subject: [PATCH] Revert generated compression replay runtime Reverts the generated runtime changes from #4055 and #4067. --- tests/generated-test/test-server | 367 ++----------------------------- 1 file changed, 16 insertions(+), 351 deletions(-) diff --git a/tests/generated-test/test-server b/tests/generated-test/test-server index ff395f697d..417fcc086c 100755 --- a/tests/generated-test/test-server +++ b/tests/generated-test/test-server @@ -10,15 +10,12 @@ from __future__ import annotations import argparse import base64 -import ctypes import json import os import re import tempfile import threading import uuid -import zlib -from ctypes.util import find_library from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -47,18 +44,6 @@ SAFE_BROWSER_HEADERS = { "content-security-policy": "default-src 'none'; sandbox", "x-content-type-options": "nosniff", } -ZSTD_CONTENTSIZE_UNKNOWN = (1 << 64) - 1 -ZSTD_CONTENTSIZE_ERROR = (1 << 64) - 2 -ZSTD_MAGIC = b"\x28\xb5\x2f\xfd" -ZSTD_SINGLE_SEGMENT_FLAG = 0x20 -ZSTD_DESCRIPTOR_LOW_BITS_MASK = 0x3F -ZSTD_MIN_RAW_FRAME_SIZE = 6 -ZSTD_FCS_TWO_BYTE_OFFSET = 0x100 -UINT8_MAX = (1 << 8) - 1 -UINT16_WITH_OFFSET_MAX = (1 << 16) - 1 + ZSTD_FCS_TWO_BYTE_OFFSET -UINT32_MAX = (1 << 32) - 1 -ZSTD_RAW_BLOCK_SIZE = 128 * 1024 -MAX_DECOMPRESSED_BODY_SIZE = 256 * 1024 * 1024 class RecordingDatabase: @@ -67,11 +52,9 @@ class RecordingDatabase: self.lock = threading.RLock() self.shards: dict[tuple[str, str], dict[str, Any]] = {} self.shard_paths: dict[tuple[str, str], Path] = {} - self.request_plans: dict[tuple[str, str, str], dict[str, Any]] = {} self.sessions: dict[str, dict[str, Any]] = {} self.fallback_consumed: set[tuple[str, str, str, int]] = set() self._load() - self._load_request_plans() def _load(self) -> None: manifest_path = self.root / "manifest.json" @@ -87,17 +70,6 @@ class RecordingDatabase: self.shards[key] = shard self.shard_paths[key] = path - def _load_request_plans(self) -> None: - root = self.root.parent / "test-runner-data" - manifest_path = root / "manifest.json" - if not manifest_path.exists(): - return - manifest = _read_json(manifest_path) - for item in manifest.get("scenarios", []): - plan = _read_json(root / item["file"]) - key = (item["version"], item["feature"], item["scenario"]) - self.request_plans[key] = plan.get("request", {}) - def start(self, version: str, feature: str, scenario: str, mode: str) -> dict[str, Any]: with self.lock: key = (version, feature) @@ -119,14 +91,13 @@ class RecordingDatabase: "key": key, "scenario": scenario, "recording": recording, - "request_plan": self.request_plans.get((version, feature, scenario)), "cursor": 0, "captures": [], "frozen_at": frozen_at, } return {"session": session_id, "frozen_at": frozen_at} - def replay(self, session_id: str | None, actual: dict[str, Any]) -> tuple[dict[str, Any], str | None]: + def replay(self, session_id: str | None, actual: dict[str, Any]) -> dict[str, Any]: with self.lock: if session_id: session = self.sessions.get(session_id) @@ -138,32 +109,20 @@ class RecordingDatabase: if cursor >= len(interactions): raise LookupError(f"Recording has no interaction #{cursor + 1}") expected = interactions[cursor] - request_plan = session["request_plan"] - if not _requests_match(expected["request"], actual, request_plan): + if not _requests_match(expected["request"], actual): raise RequestMismatchError(expected["request"], actual, cursor) session["cursor"] += 1 - compression = ( - request_plan.get("compression") - if request_plan and _request_matches_plan(expected["request"], request_plan) - else expected["request"].get("compression") - ) - return expected["response"], compression + return expected["response"] for (version, feature), shard in sorted(self.shards.items()): for recording in shard["recordings"]: for index, interaction in enumerate(recording["interactions"]): consumed_key = (version, feature, recording["scenario"], index) - request_plan = self.request_plans.get((version, feature, recording["scenario"])) if consumed_key not in self.fallback_consumed and _requests_match( - interaction["request"], actual, request_plan + interaction["request"], actual ): self.fallback_consumed.add(consumed_key) - compression = ( - request_plan.get("compression") - if request_plan and _request_matches_plan(interaction["request"], request_plan) - else interaction["request"].get("compression") - ) - return interaction["response"], compression + return interaction["response"] raise LookupError("No unconsumed interaction matches this request") def next_request(self, session_id: str) -> dict[str, Any]: @@ -339,21 +298,14 @@ class TestRequestHandler(BaseHTTPRequestHandler): def _handle_api_request(self) -> None: body = self._read_body() - actual = _normalise_request( - self.command, - self.path, - self.headers.get("content-type", ""), - self.headers.get("content-encoding", ""), - body, - ) + actual = _normalise_request(self.command, self.path, self.headers.get("content-type", ""), body) session_id = self.headers.get(SESSION_HEADER) if self.server.mode == "replay": - response, compression = self.server.database.replay(session_id, actual) + response = self.server.database.replay(session_id, actual) else: response = self._forward(body) self.server.database.capture(session_id, actual, response) - compression = None - self._send_recorded_response(response, compression=compression) + self._send_recorded_response(response) def _forward(self, body: bytes) -> dict[str, Any]: if not self.server.upstream: @@ -389,37 +341,17 @@ class TestRequestHandler(BaseHTTPRequestHandler): length = int(self.headers.get("content-length", "0")) return self.rfile.read(length) if length else b"" - def _send_recorded_response(self, response: dict[str, Any], *, compression: str | None = None) -> None: + def _send_recorded_response(self, response: dict[str, Any]) -> None: body_data = response.get("body", {}) if body_data.get("encoding") == "base64": body = base64.b64decode(body_data.get("value", "")) else: body = body_data.get("value", "").encode("utf-8") status = response["status"] - response_headers = response.get("headers", {}) - recorded_encoding = next( - (str(value) for key, value in response_headers.items() if key.lower() == "content-encoding"), - "", - ) - recorded_compression = recorded_encoding.strip().casefold() - if compression and _status_allows_message_content(status): - if recorded_compression: - decoded = _decompress_body(body, recorded_compression) - if decoded is None: - raise ValueError(f"Unsupported or invalid recorded response compression: {recorded_encoding}") - body = decoded - body = _compress_body(body, compression) self.send_response(status, response.get("reason")) - for key, value in response_headers.items(): - if ( - key.lower() - not in HOP_BY_HOP_HEADERS | {"content-encoding", "content-length"} | SAFE_BROWSER_HEADERS.keys() - ): + for key, value in response.get("headers", {}).items(): + if key.lower() not in HOP_BY_HOP_HEADERS | {"content-length"} | SAFE_BROWSER_HEADERS.keys(): self.send_header(key, value) - if compression and _status_allows_message_content(status): - self.send_header("content-encoding", compression) - elif recorded_encoding and _status_allows_message_content(status): - self.send_header("content-encoding", recorded_encoding) for key, value in SAFE_BROWSER_HEADERS.items(): self.send_header(key, value) if _status_allows_message_content(status): @@ -440,50 +372,21 @@ class TestRequestHandler(BaseHTTPRequestHandler): self.wfile.write(body) -def _normalise_request( - method: str, - raw_path: str, - content_type: str, - content_encoding: str, - body: bytes, -) -> dict[str, Any]: +def _normalise_request(method: str, raw_path: str, content_type: str, body: bytes) -> dict[str, Any]: parsed = urlsplit(raw_path) - compression = content_encoding.strip().casefold() - normalised_body = _normalise_body(body, content_type, compression) - request = { + normalised_body = _normalise_body(body, content_type) + return { "method": method.upper(), "path": _normalise_path(parsed.path), "query": sorted([list(pair) for pair in parse_qsl(parsed.query, keep_blank_values=True)]), "content_type": _normalise_content_type(content_type, normalised_body), "body": normalised_body, } - if compression: - request["compression"] = compression - return request -def _normalise_body(body: bytes, content_type: str, compression: str = "") -> dict[str, Any]: +def _normalise_body(body: bytes, content_type: str) -> dict[str, Any]: if not body: - if compression in {"gzip", "deflate", "zstd1"}: - return {"type": "invalid-compression", "value": ""} return {"type": "empty", "value": None} - if compression == "zstd1": - decompressed = _decompress_zstd(body) - if decompressed is None: - return { - "type": "invalid-compression", - "value": base64.b64encode(body).decode("ascii"), - } - body = decompressed - elif compression in {"gzip", "deflate"}: - wbits = zlib.MAX_WBITS | 16 if compression == "gzip" else zlib.MAX_WBITS - decompressed = _decompress_zlib(body, wbits=wbits) - if decompressed is None: - return { - "type": "invalid-compression", - "value": base64.b64encode(body).decode("ascii"), - } - body = decompressed media_type = _media_type(content_type) text = body.decode("utf-8", errors="surrogateescape") if media_type.endswith("json"): @@ -498,224 +401,6 @@ def _normalise_body(body: bytes, content_type: str, compression: str = "") -> di return {"type": "text", "value": text} -def _decompress_zlib( - body: bytes, - *, - wbits: int, - max_size: int = MAX_DECOMPRESSED_BODY_SIZE, -) -> bytes | None: - """Decompress exactly one complete zlib stream with bounded output.""" - decompressor = zlib.decompressobj(wbits) - output = bytearray() - pending = body - try: - while pending: - remaining = max_size - len(output) + 1 - chunk = decompressor.decompress(pending, remaining) - output.extend(chunk) - if len(output) > max_size: - return None - unconsumed = decompressor.unconsumed_tail - if not unconsumed: - break - if len(unconsumed) == len(pending) and not chunk: - return None - pending = unconsumed - except zlib.error: - return None - if not decompressor.eof or decompressor.unused_data or decompressor.unconsumed_tail: - return None - return bytes(output) - - -def _compress_body(body: bytes, compression: str) -> bytes: - if compression == "gzip": - compressor = zlib.compressobj(wbits=zlib.MAX_WBITS | 16) - return compressor.compress(body) + compressor.flush() - if compression == "deflate": - return zlib.compress(body) - if compression == "zstd1": - return _compress_zstd(body) - raise ValueError(f"Unsupported response compression: {compression}") - - -def _decompress_body(body: bytes, compression: str) -> bytes | None: - if compression == "gzip": - return _decompress_zlib(body, wbits=zlib.MAX_WBITS | 16) - if compression == "deflate": - return _decompress_zlib(body, wbits=zlib.MAX_WBITS) - if compression == "zstd1": - return _decompress_zstd(body) - return None - - -def _compress_zstd(body: bytes) -> bytes: - """Compress one Zstandard frame with the platform libzstd.""" - library = _load_zstd() - if library is None: - return _compress_zstd_raw_frame(body) - try: - library.ZSTD_isError.argtypes = [ctypes.c_size_t] - library.ZSTD_isError.restype = ctypes.c_uint - library.ZSTD_compressBound.argtypes = [ctypes.c_size_t] - library.ZSTD_compressBound.restype = ctypes.c_size_t - library.ZSTD_compress.argtypes = [ - ctypes.c_void_p, - ctypes.c_size_t, - ctypes.c_void_p, - ctypes.c_size_t, - ctypes.c_int, - ] - library.ZSTD_compress.restype = ctypes.c_size_t - - source = ctypes.create_string_buffer(body) - capacity = library.ZSTD_compressBound(len(body)) - destination = ctypes.create_string_buffer(max(1, capacity)) - compressed_size = library.ZSTD_compress(destination, capacity, source, len(body), 3) - if library.ZSTD_isError(compressed_size): - return _compress_zstd_raw_frame(body) - return destination.raw[:compressed_size] - except (AttributeError, OSError, OverflowError, TypeError): - return _compress_zstd_raw_frame(body) - - -def _compress_zstd_raw_frame(body: bytes) -> bytes: - """Create a valid Zstandard frame made only of dependency-free raw blocks.""" - size = len(body) - if size <= UINT8_MAX: - descriptor = 0x20 - content_size = size.to_bytes(1, "little") - elif size <= UINT16_WITH_OFFSET_MAX: - descriptor = 0x60 - content_size = (size - ZSTD_FCS_TWO_BYTE_OFFSET).to_bytes(2, "little") - elif size <= UINT32_MAX: - descriptor = 0xA0 - content_size = size.to_bytes(4, "little") - else: - descriptor = 0xE0 - content_size = size.to_bytes(8, "little") - - frame = bytearray(ZSTD_MAGIC) - frame.append(descriptor) - frame.extend(content_size) - offset = 0 - while offset < size: - chunk = body[offset : offset + ZSTD_RAW_BLOCK_SIZE] - offset += len(chunk) - header = (len(chunk) << 3) | int(offset == size) - frame.extend(header.to_bytes(3, "little")) - frame.extend(chunk) - if not body: - frame.extend(b"\x01\x00\x00") - return bytes(frame) - - -def _load_zstd() -> Any | None: - candidates = ( - find_library("zstd"), - "libzstd.so.1", - "libzstd.dylib", - "/opt/homebrew/lib/libzstd.dylib", - "/usr/local/lib/libzstd.dylib", - "libzstd.dll", - "zstd.dll", - ) - for candidate in dict.fromkeys(item for item in candidates if item): - try: - return ctypes.CDLL(candidate) - except OSError: - continue - return None - - -def _decompress_zstd(body: bytes) -> bytes | None: - """Decompress one complete Zstandard frame with the platform libzstd.""" - library = _load_zstd() - if library is None: - return _decompress_zstd_raw_frame(body) - - try: - library.ZSTD_isError.argtypes = [ctypes.c_size_t] - library.ZSTD_isError.restype = ctypes.c_uint - library.ZSTD_findFrameCompressedSize.argtypes = [ctypes.c_void_p, ctypes.c_size_t] - library.ZSTD_findFrameCompressedSize.restype = ctypes.c_size_t - library.ZSTD_getFrameContentSize.argtypes = [ctypes.c_void_p, ctypes.c_size_t] - library.ZSTD_getFrameContentSize.restype = ctypes.c_ulonglong - library.ZSTD_decompressBound.argtypes = [ctypes.c_void_p, ctypes.c_size_t] - library.ZSTD_decompressBound.restype = ctypes.c_ulonglong - library.ZSTD_decompress.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_void_p, ctypes.c_size_t] - library.ZSTD_decompress.restype = ctypes.c_size_t - - source = ctypes.create_string_buffer(body) - source_pointer = ctypes.cast(source, ctypes.c_void_p) - compressed_size = library.ZSTD_findFrameCompressedSize(source_pointer, len(body)) - if library.ZSTD_isError(compressed_size) or compressed_size != len(body): - return None - - capacity = library.ZSTD_getFrameContentSize(source_pointer, len(body)) - if capacity == ZSTD_CONTENTSIZE_ERROR: - return None - if capacity == ZSTD_CONTENTSIZE_UNKNOWN: - capacity = library.ZSTD_decompressBound(source_pointer, len(body)) - if capacity > MAX_DECOMPRESSED_BODY_SIZE: - return None - - destination = ctypes.create_string_buffer(max(1, capacity)) - decompressed_size = library.ZSTD_decompress(destination, capacity, source_pointer, len(body)) - if library.ZSTD_isError(decompressed_size) or decompressed_size > capacity: - return None - return destination.raw[:decompressed_size] - except (AttributeError, OSError, OverflowError, TypeError): - return None - - -def _decompress_zstd_raw_frame(body: bytes) -> bytes | None: - """Decode the raw-block Zstandard frames emitted by the dependency-free fallback.""" - if len(body) < ZSTD_MIN_RAW_FRAME_SIZE or body[:4] != ZSTD_MAGIC: - return None - descriptor = body[4] - if descriptor & ZSTD_DESCRIPTOR_LOW_BITS_MASK != ZSTD_SINGLE_SEGMENT_FLAG: - return None - size_flag = descriptor >> 6 - size_bytes = (1, 2, 4, 8)[size_flag] - cursor = 5 - if len(body) < cursor + size_bytes: - return None - expected_size = int.from_bytes(body[cursor : cursor + size_bytes], "little") - if size_flag == 1: - expected_size += ZSTD_FCS_TWO_BYTE_OFFSET - if expected_size > MAX_DECOMPRESSED_BODY_SIZE: - return None - cursor += size_bytes - output = bytearray() - last_block = False - while not last_block: - if len(body) < cursor + 3: - return None - header = int.from_bytes(body[cursor : cursor + 3], "little") - cursor += 3 - last_block = bool(header & 1) - block_type = (header >> 1) & 0x3 - block_size = header >> 3 - if block_type == 0: - if len(body) < cursor + block_size: - return None - output.extend(body[cursor : cursor + block_size]) - cursor += block_size - elif block_type == 1: - if cursor >= len(body): - return None - output.extend(body[cursor : cursor + 1] * block_size) - cursor += 1 - else: - return None - if len(output) > MAX_DECOMPRESSED_BODY_SIZE: - return None - if cursor != len(body) or len(output) != expected_size: - return None - return bytes(output) - - def _normalise_json(value: Any) -> Any: if isinstance(value, dict): return {key: _normalise_json(item) for key, item in value.items()} @@ -730,33 +415,13 @@ def _normalise_content_type(content_type: str, body: dict[str, Any]) -> str: return "" if body["type"] == "empty" else _media_type(content_type) -def _requests_match( - expected: dict[str, Any], - actual: dict[str, Any], - request_plan: dict[str, Any] | None = None, -) -> bool: +def _requests_match(expected: dict[str, Any], actual: dict[str, Any]) -> bool: comparable_fields = ("method", "path", "query", "content_type") if any(expected[field] != actual[field] for field in comparable_fields): return False - expected_compression = expected.get("compression") - if request_plan and _request_matches_plan(expected, request_plan): - expected_compression = request_plan.get("compression", expected_compression) - if expected_compression is not None and actual.get("compression", "") != expected_compression: - return False return _bodies_match(expected["body"], actual["body"]) -def _request_matches_plan(request: dict[str, Any], plan: dict[str, Any]) -> bool: - if request["method"] != plan.get("method"): - return False - path = plan.get("path") - if not path: - return False - parts = re.split(r"(\{[^/{}]+\})", path) - pattern = "".join(r"[^/]+" if part.startswith("{") else re.escape(part) for part in parts) - return re.fullmatch(pattern, request["path"]) is not None - - def _bodies_match(expected: dict[str, Any], actual: dict[str, Any]) -> bool: if expected == actual: return True