From 33d55ec897f0ee3feb462a1f000dde026f95152e Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Mon, 20 Jul 2026 09:30:04 -0700 Subject: [PATCH] Release: raster-data-URI redaction fast-path (#6311, @inch772) (#6367) * fix(redaction): skip native raster data URIs * fix(redaction): accept mixed-case raster MIME types * fix(redaction): validate complete raster payloads * docs(changelog): stamp #6311 raster-data-URI redaction fast-path --------- Co-authored-by: Su Ahn Lee <11433303+inch772@users.noreply.github.com> Co-authored-by: nesquena-hermes --- CHANGELOG.md | 2 + api/helpers.py | 420 +++++++++++++++++++++++- tests/test_raster_data_uri_redaction.py | 362 ++++++++++++++++++++ 3 files changed, 782 insertions(+), 2 deletions(-) create mode 100644 tests/test_raster_data_uri_redaction.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e75b0b7..2e9dc877 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,8 @@ ### Fixed +- **Large embedded images no longer slow down API response redaction.** Native raster image data URIs (`data:image/png|jpeg|gif|webp|bmp;base64,…`) are skipped by the credential-redaction pass instead of being scanned byte-by-byte on every response — a multi-megabyte image is opaque binary, not text a credential regex should rewrite. The fast path only applies after the *entire* base64 payload is validated (canonical round-trip + structural walk), so a string that merely starts with an image header but carries trailing or appended text still goes through full redaction — a credential can never ride along in an image's tail. Non-image strings, and sibling secrets in the same payload, are unaffected. Thanks @inch772. (#6311) + - **Gateway-backed sessions can once again send with a MoA (Mixture-of-Agents) default model.** A legitimate gateway-default MoA send was being rejected as if it were a disallowed per-request override. The chat-start handler now distinguishes the two: a send is allowed only when the *configured* provider is MoA, the resolved model matches the configured default, and no explicit picker override was made — all evaluated against one request-owned config snapshot captured atomically, so a concurrent profile/config change can't flip the decision mid-request. Explicit `moa_config` overrides on gateway sessions still return 409. Thanks @rodboev. (#5869, #5853) - **Content-search and sidebar polling are faster: run-journal summaries are now cached.** Repeated reads of an unchanged run journal reuse a cached summary instead of re-parsing the whole file, keyed on a complete filesystem signature (device, inode, size, mtime, and ctime) so the cache is always dropped the instant the journal is appended to, replaced, truncated, or recreated — including a same-size rewrite that preserves mtime. Bounded LRU; a first-write race can never pin a stale empty result. Thanks @sjungwon03. (#6291) diff --git a/api/helpers.py b/api/helpers.py index 1a060435..d285cfe4 100644 --- a/api/helpers.py +++ b/api/helpers.py @@ -1,6 +1,8 @@ """ Hermes Web UI -- HTTP helper functions. """ +import base64 as _base64 +import binascii as _binascii import functools import json as _json import logging @@ -341,6 +343,13 @@ def _build_redact_fn(): r"""(Authorization:\s*(?:Bearer|Bot)\s+)([^\s'",\]\)]+)""", _re.IGNORECASE, ) + # A rejected image data URI can place a syntactically valid AWS key at an + # arbitrary base64 alignment, immediately after another base64 character. + # The general credential regex uses token boundaries to avoid rewriting + # prose identifiers; this defense-in-depth pass ensures the fail-closed + # image path still removes the maintainer's embedded-suffix attack even + # when hermes-agent is unavailable and the local fallback owns redaction. + _EMBEDDED_AWS_ACCESS_KEY_RE = _re.compile(r"AKIA[A-Z0-9]{16}") _ENV_RE = _re.compile( r"([A-Z0-9_]{0,50}(?:API_?KEY|TOKEN|SECRET|PASSWORD|PASSWD|CREDENTIAL|AUTH)[A-Z0-9_]{0,50})" r"\s*=\s*(['\"]?)(\S+)\2" @@ -408,6 +417,7 @@ def _build_redact_fn(): if not isinstance(text, str) or not text: return text text = _CRED_RE.sub(lambda m: _mask(m.group(1)), text) + text = _EMBEDDED_AWS_ACCESS_KEY_RE.sub(lambda m: _mask(m.group(0)), text) text = _AUTH_HDR_RE.sub(lambda m: m.group(1) + _mask(m.group(2)), text) text = _ENV_RE.sub(_env_replacement, text) text = _PRIVKEY_RE.sub("[REDACTED PRIVATE KEY]", text) @@ -576,6 +586,360 @@ def _redact_text(text: str, *, _enabled: bool | None = None) -> str: return _redact_fn_cached(text) +_RASTER_IMAGE_DATA_URI_PREFIXES = ( + ("data:image/png;base64,", "png"), + ("data:image/jpeg;base64,", "jpeg"), + ("data:image/jpg;base64,", "jpeg"), + ("data:image/gif;base64,", "gif"), + ("data:image/webp;base64,", "webp"), + ("data:image/bmp;base64,", "bmp"), +) + + +def _is_native_raster_data_uri(text: str) -> bool: + """Return whether *text* is one complete, canonical raster data URI. + + Native image content is opaque binary, not text that the credential regexes + can safely rewrite. The exemption is a credential-boundary decision, so a + matching header or magic prefix is not enough: decode the entire canonical + base64 payload and require the image format to terminate exactly at the end + of the decoded bytes. Any malformed, ambiguous, or trailing content falls + through to normal text redaction. + """ + if not isinstance(text, str): + return False + image_kind = None + payload_start = 0 + for prefix, candidate_kind in _RASTER_IMAGE_DATA_URI_PREFIXES: + # URI schemes and MIME type tokens are case-insensitive. Only normalize + # this short header slice — never the multi-megabyte base64 payload. + if text[:len(prefix)].lower() == prefix: + image_kind = candidate_kind + payload_start = len(prefix) + break + if image_kind is None: + return False + + payload = text[payload_start:] + if not payload: + return False + try: + raw = _base64.b64decode(payload, validate=True) + except (_binascii.Error, ValueError): + return False + # validate=True rejects foreign characters and misplaced padding; this + # round-trip also rejects non-canonical pad bits and missing/extra padding. + if _base64.b64encode(raw).decode("ascii") != payload: + return False + + if image_kind == "png": + return _is_complete_png(raw) + if image_kind == "jpeg": + return _is_complete_jpeg(raw) + if image_kind == "gif": + return _is_complete_gif(raw) + if image_kind == "webp": + return _is_complete_webp(raw) + if image_kind == "bmp": + return _is_complete_bmp(raw) + return False + + +def _is_complete_png(raw: bytes) -> bool: + if not raw.startswith(b"\x89PNG\r\n\x1a\n"): + return False + pos = 8 + chunk_index = 0 + saw_idat = False + while pos < len(raw): + if pos + 12 > len(raw): + return False + length = int.from_bytes(raw[pos:pos + 4], "big") + chunk_type = raw[pos + 4:pos + 8] + data_start = pos + 8 + data_end = data_start + length + chunk_end = data_end + 4 + if chunk_end > len(raw): + return False + if not all((65 <= value <= 90) or (97 <= value <= 122) for value in chunk_type): + return False + if not 65 <= chunk_type[2] <= 90: # PNG reserved bit must be zero. + return False + expected_crc = int.from_bytes(raw[data_end:chunk_end], "big") + actual_crc = _binascii.crc32(chunk_type + raw[data_start:data_end]) & 0xFFFFFFFF + if actual_crc != expected_crc: + return False + + if chunk_index == 0: + if chunk_type != b"IHDR" or length != 13: + return False + width = int.from_bytes(raw[data_start:data_start + 4], "big") + height = int.from_bytes(raw[data_start + 4:data_start + 8], "big") + bit_depth = raw[data_start + 8] + color_type = raw[data_start + 9] + valid_depths = { + 0: {1, 2, 4, 8, 16}, + 2: {8, 16}, + 3: {1, 2, 4, 8}, + 4: {8, 16}, + 6: {8, 16}, + } + if ( + not width + or not height + or bit_depth not in valid_depths.get(color_type, set()) + or raw[data_start + 10] != 0 + or raw[data_start + 11] != 0 + or raw[data_start + 12] not in {0, 1} + ): + return False + elif chunk_type == b"IHDR": + return False + + if chunk_type == b"IDAT": + saw_idat = True + if chunk_type == b"IEND": + return length == 0 and saw_idat and chunk_end == len(raw) + pos = chunk_end + chunk_index += 1 + return False + + +_JPEG_SOF_MARKERS = { + 0xC0, 0xC1, 0xC2, 0xC3, + 0xC5, 0xC6, 0xC7, + 0xC9, 0xCA, 0xCB, + 0xCD, 0xCE, 0xCF, +} + + +def _is_complete_jpeg(raw: bytes) -> bool: + if len(raw) < 4 or raw[:2] != b"\xff\xd8": + return False + pos = 2 + saw_sof = False + saw_scan = False + while pos < len(raw): + marker_start = pos + if raw[pos] != 0xFF: + return False + while pos < len(raw) and raw[pos] == 0xFF: + pos += 1 + if pos >= len(raw): + return False + marker = raw[pos] + pos += 1 + if marker == 0xD9: + return saw_sof and saw_scan and pos == len(raw) + if marker in {0x00, 0x01, 0xD8} or 0xD0 <= marker <= 0xD7: + return False + if pos + 2 > len(raw): + return False + segment_length = int.from_bytes(raw[pos:pos + 2], "big") + if segment_length < 2: + return False + segment_end = pos + segment_length + if segment_end > len(raw): + return False + if marker in _JPEG_SOF_MARKERS: + if segment_length < 8: + return False + saw_sof = True + if marker != 0xDA: + pos = segment_end + continue + + saw_scan = True + pos = segment_end + while pos < len(raw): + if raw[pos] != 0xFF: + pos += 1 + continue + marker_start = pos + while pos < len(raw) and raw[pos] == 0xFF: + pos += 1 + if pos >= len(raw): + return False + scan_marker = raw[pos] + if scan_marker == 0x00 or 0xD0 <= scan_marker <= 0xD7: + pos += 1 + continue + pos = marker_start + break + else: + return False + return False + + +def _gif_subblocks_end(raw: bytes, pos: int) -> int | None: + while pos < len(raw): + size = raw[pos] + pos += 1 + if size == 0: + return pos + if pos + size > len(raw): + return None + pos += size + return None + + +def _is_complete_gif(raw: bytes) -> bool: + if len(raw) < 14 or not raw.startswith((b"GIF87a", b"GIF89a")): + return False + width = int.from_bytes(raw[6:8], "little") + height = int.from_bytes(raw[8:10], "little") + if not width or not height: + return False + packed = raw[10] + pos = 13 + if packed & 0x80: + pos += 3 * (1 << ((packed & 0x07) + 1)) + if pos > len(raw): + return False + saw_image = False + while pos < len(raw): + introducer = raw[pos] + pos += 1 + if introducer == 0x3B: + return saw_image and pos == len(raw) + if introducer == 0x21: + if pos >= len(raw): + return False + pos += 1 # extension label + end = _gif_subblocks_end(raw, pos) + if end is None: + return False + pos = end + continue + if introducer != 0x2C or pos + 9 > len(raw): + return False + image_width = int.from_bytes(raw[pos + 4:pos + 6], "little") + image_height = int.from_bytes(raw[pos + 6:pos + 8], "little") + image_packed = raw[pos + 8] + if not image_width or not image_height: + return False + pos += 9 + if image_packed & 0x80: + pos += 3 * (1 << ((image_packed & 0x07) + 1)) + if pos >= len(raw): + return False + lzw_minimum_code_size = raw[pos] + if not 2 <= lzw_minimum_code_size <= 11: + return False + pos += 1 + end = _gif_subblocks_end(raw, pos) + if end is None: + return False + pos = end + saw_image = True + return False + + +def _is_webp_image_chunk(chunk_type: bytes, data: bytes) -> bool: + if chunk_type == b"VP8 ": + if len(data) < 10 or data[3:6] != b"\x9d\x01\x2a": + return False + width = int.from_bytes(data[6:8], "little") & 0x3FFF + height = int.from_bytes(data[8:10], "little") & 0x3FFF + return bool(width and height) + if chunk_type == b"VP8L": + # The three high bits of the fifth byte are the version number. The + # current lossless bitstream defines only version zero. + return len(data) >= 5 and data[0] == 0x2F and not data[4] & 0xE0 + return False + + +def _is_complete_webp_frame(data: bytes) -> bool: + """Validate the nested chunks in one extended-WebP animation frame.""" + if len(data) < 16 or data[15] & 0xFC: + return False + pos = 16 + saw_image = False + while pos < len(data): + if pos + 8 > len(data): + return False + chunk_type = data[pos:pos + 4] + chunk_size = int.from_bytes(data[pos + 4:pos + 8], "little") + data_start = pos + 8 + data_end = data_start + chunk_size + chunk_end = data_end + (chunk_size & 1) + if chunk_end > len(data): + return False + chunk_data = data[data_start:data_end] + if chunk_type in {b"VP8 ", b"VP8L"}: + if saw_image or not _is_webp_image_chunk(chunk_type, chunk_data): + return False + saw_image = True + elif chunk_type != b"ALPH": + return False + pos = chunk_end + return saw_image and pos == len(data) + + +def _is_complete_webp(raw: bytes) -> bool: + if ( + len(raw) < 20 + or raw[:4] != b"RIFF" + or raw[8:12] != b"WEBP" + or int.from_bytes(raw[4:8], "little") + 8 != len(raw) + ): + return False + pos = 12 + saw_image = False + while pos < len(raw): + if pos + 8 > len(raw): + return False + chunk_type = raw[pos:pos + 4] + chunk_size = int.from_bytes(raw[pos + 4:pos + 8], "little") + data_start = pos + 8 + data_end = data_start + chunk_size + chunk_end = data_end + (chunk_size & 1) + if chunk_end > len(raw): + return False + data = raw[data_start:data_end] + if chunk_type in {b"VP8 ", b"VP8L"}: + if saw_image or not _is_webp_image_chunk(chunk_type, data): + return False + saw_image = True + elif chunk_type == b"VP8X": + if len(data) != 10 or data[0] & 0x81 or any(data[1:4]): + return False + elif chunk_type == b"ANMF": + if not _is_complete_webp_frame(data): + return False + saw_image = True + pos = chunk_end + return saw_image and pos == len(raw) + + +def _is_complete_bmp(raw: bytes) -> bool: + if len(raw) < 26 or raw[:2] != b"BM": + return False + if int.from_bytes(raw[2:6], "little") != len(raw): + return False + pixel_offset = int.from_bytes(raw[10:14], "little") + dib_size = int.from_bytes(raw[14:18], "little") + if dib_size == 12: + width = int.from_bytes(raw[18:20], "little") + height = int.from_bytes(raw[20:22], "little") + planes = int.from_bytes(raw[22:24], "little") + bits_per_pixel = int.from_bytes(raw[24:26], "little") + elif dib_size >= 40 and 14 + dib_size <= len(raw): + width = int.from_bytes(raw[18:22], "little", signed=True) + height = int.from_bytes(raw[22:26], "little", signed=True) + planes = int.from_bytes(raw[26:28], "little") + bits_per_pixel = int.from_bytes(raw[28:30], "little") + else: + return False + return ( + bool(width) + and bool(height) + and planes == 1 + and bits_per_pixel in {1, 2, 4, 8, 16, 24, 32} + and 14 + dib_size <= pixel_offset < len(raw) + ) + + def _redact_value(v, *, _enabled: bool | None = None): """Recursively redact credentials from strings, dicts, and lists. @@ -585,12 +949,64 @@ def _redact_value(v, *, _enabled: bool | None = None): if isinstance(v, str): return _redact_text(v, _enabled=_enabled) if isinstance(v, dict): - return {k: _redact_value(val, _enabled=_enabled) for k, val in v.items()} + return { + key: _redact_value(value, _enabled=_enabled) + for key, value in v.items() + } if isinstance(v, list): return [_redact_value(item, _enabled=_enabled) for item in v] return v +def _redact_message_content_part(part, *, _enabled: bool): + """Redact one canonical ``messages[*].content[*]`` part. + + The raster exemption exists only at this authoritative schema position. + Image-shaped dictionaries in metadata, tools, todos, journals, or arbitrary + nested values remain on the normal fail-closed redaction path. + """ + if not _enabled or not ( + isinstance(part, dict) + and part.get("type") == "image_url" + and isinstance(part.get("image_url"), dict) + ): + return _redact_value(part, _enabled=_enabled) + result = {} + for key, value in part.items(): + if key != "image_url": + result[key] = _redact_value(value, _enabled=_enabled) + continue + result[key] = { + image_key: image_value + if image_key == "url" and _is_native_raster_data_uri(image_value) + else _redact_value(image_value, _enabled=_enabled) + for image_key, image_value in value.items() + } + return result + + +def _redact_messages(messages, *, _enabled: bool): + if not isinstance(messages, list): + return _redact_value(messages, _enabled=_enabled) + redacted = [] + for message in messages: + if not isinstance(message, dict): + redacted.append(_redact_value(message, _enabled=_enabled)) + continue + item = {} + allow_native_image = message.get("role") == "user" + for key, value in message.items(): + if allow_native_image and key == "content" and isinstance(value, list): + item[key] = [ + _redact_message_content_part(part, _enabled=_enabled) + for part in value + ] + else: + item[key] = _redact_value(value, _enabled=_enabled) + redacted.append(item) + return redacted + + def redact_session_data(session_dict: dict) -> dict: """Redact credentials from message content, tool data, and session sidecars. @@ -610,7 +1026,7 @@ def redact_session_data(session_dict: dict) -> dict: if isinstance(result.get('title'), str): result['title'] = _redact_text(result['title'], _enabled=_enabled) if 'messages' in result: - result['messages'] = _redact_value(result['messages'], _enabled=_enabled) + result['messages'] = _redact_messages(result['messages'], _enabled=_enabled) if 'tool_calls' in result: result['tool_calls'] = _redact_value(result['tool_calls'], _enabled=_enabled) if 'todo_state' in result: diff --git a/tests/test_raster_data_uri_redaction.py b/tests/test_raster_data_uri_redaction.py new file mode 100644 index 00000000..99c49f0d --- /dev/null +++ b/tests/test_raster_data_uri_redaction.py @@ -0,0 +1,362 @@ +"""Regression coverage for large native image redaction overhead. + +Native multimodal messages store raster images as base64 data URIs. Those +opaque image bytes must not be sent through the text credential redactor: a +large image can randomly contain one of the cheap prefilter's markers and then +pay for every regex pass in the agent redactor. +""" +import base64 +import json +import struct +import zlib + +import pytest + +from api import helpers +from api.session_export_html import render_session_html + + +_FAKE_AWS_KEY = "AKIATESTFAKEKEY12345" +_ONE_PIXEL_JPEG = base64.b64decode( + "/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAP//////////////////////////////////////////////////////////////////////////////////////2wBDAf//////////////////////////////////////////////////////////////////////////////////////wAARCAABAAEDASIAAhEBAxEB/8QAFQABAQAAAAAAAAAAAAAAAAAAAAf/xAAUEAEAAAAAAAAAAAAAAAAAAAAA/9oADAMBAAIQAxAAAAF//8QAFBABAAAAAAAAAAAAAAAAAAAAAP/aAAgBAQABBQJ//8QAFBEBAAAAAAAAAAAAAAAAAAAAAP/aAAgBAwEBPwF//8QAFBEBAAAAAAAAAAAAAAAAAAAAAP/aAAgBAgEBPwF//8QAFBABAAAAAAAAAAAAAAAAAAAAAP/aAAgBAQAGPwJ//8QAFBABAAAAAAAAAAAAAAAAAAAAAP/aAAgBAQABPyF//9oADAMBAAIAAwAAABAf/8QAFBEBAAAAAAAAAAAAAAAAAAAAAP/aAAgBAwEBPxB//8QAFBEBAAAAAAAAAAAAAAAAAAAAAP/aAAgBAgEBPxB//8QAFBABAAAAAAAAAAAAAAAAAAAAAP/aAAgBAQABPxB//9k=" +) +_ONE_PIXEL_GIF = base64.b64decode( + "R0lGODlhAQABAIAAAAAAAP///ywAAAAAAQABAAACAUwAOw==" +) +_ONE_PIXEL_WEBP = base64.b64decode( + "UklGRiIAAABXRUJQVlA4IBYAAAAwAQCdASoBAAEAAUAmJaQAA3AA/vuU" +) + + +def _one_pixel_bmp() -> bytes: + pixel_data = b"\x00\x00\x00\x00" # BGR black + row padding + file_size = 54 + len(pixel_data) + file_header = b"BM" + struct.pack(" bytes: + crc = zlib.crc32(kind + data) & 0xFFFFFFFF + return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", crc) + + +def _valid_png_bytes_with_sensitive_marker(*, extra_bytes: int = 0) -> bytes: + """Build a real 1x1 RGBA PNG whose base64 contains an ``AKIA`` quartet.""" + marker_bytes = base64.b64decode("AKIA") + ihdr = struct.pack(">IIBBBBB", 1, 1, 8, 6, 0, 0, 0) + # The first data byte starts at absolute offset 41. One byte of padding + # aligns marker_bytes to a base64 quantum, preserving the literal AKIA + # quartet that exercises the redaction prefilter. + marker_chunk = _png_chunk( + b"raNd", + b"\x00" + marker_bytes + (b"image-bytes" * 32) + (b"x" * extra_bytes), + ) + idat = _png_chunk(b"IDAT", zlib.compress(b"\x00\x00\x00\x00\x00")) + prefix = b"\x89PNG\r\n\x1a\n" + _png_chunk(b"IHDR", ihdr) + marker_chunk + idat + # Make the complete PNG length a multiple of three. This lets the + # maintainer-supplied post-IEND attack append decoded bytes whose standard + # base64 representation remains the literal fake credential. + for pad_len in range(3): + raw = prefix + _png_chunk(b"paDd", b"x" * pad_len) + _png_chunk(b"IEND", b"") + if len(raw) % 3 == 0: + encoded = base64.b64encode(raw).decode("ascii") + assert "AKIA" in encoded + return raw + raise AssertionError("could not align synthetic PNG") + + +def _png_data_uri_with_sensitive_marker() -> str: + raw = _valid_png_bytes_with_sensitive_marker() + encoded = base64.b64encode(raw).decode("ascii") + assert "AKIA" in encoded + return f"data:image/png;base64,{encoded}" + + +def _png_data_uri_with_post_iend_credential() -> str: + raw = _valid_png_bytes_with_sensitive_marker() + suffix = base64.b64decode(_FAKE_AWS_KEY, validate=True) + encoded = base64.b64encode(raw + suffix).decode("ascii") + assert encoded.endswith(_FAKE_AWS_KEY) + return f"data:image/png;base64,{encoded}" + + +def test_native_raster_data_uri_bypasses_text_redactor(monkeypatch): + uri = _png_data_uri_with_sensitive_marker() + calls = [] + monkeypatch.setattr( + helpers, + "_redact_fn_cached", + lambda text: calls.append(text) or "unexpected-redaction", + ) + + content_part = { + "type": "image_url", + "image_url": {"url": uri, "detail": "auto"}, + } + + redacted = helpers.redact_session_data( + {"messages": [{"role": "user", "content": [content_part]}]} + ) + assert redacted["messages"][0]["content"][0] == content_part + assert calls == [] + + +def test_multimegabyte_native_raster_still_bypasses_text_redactor(monkeypatch): + raw = _valid_png_bytes_with_sensitive_marker(extra_bytes=1_000_000) + uri = f"data:image/png;base64,{base64.b64encode(raw).decode('ascii')}" + calls = [] + monkeypatch.setattr( + helpers, + "_redact_fn_cached", + lambda text: calls.append(text) or "unexpected-redaction", + ) + + result = helpers.redact_session_data( + { + "messages": [ + { + "role": "user", + "content": [{"type": "image_url", "image_url": {"url": uri}}], + } + ] + } + ) + + assert result["messages"][0]["content"][0]["image_url"]["url"] == uri + assert calls == [] + + +def test_native_raster_data_uri_accepts_uppercase_mime(monkeypatch): + uri = _png_data_uri_with_sensitive_marker().replace( + "data:image/png", + "data:image/PNG", + 1, + ) + calls = [] + monkeypatch.setattr( + helpers, + "_redact_fn_cached", + lambda text: calls.append(text) or "unexpected-redaction", + ) + + content_part = {"type": "image_url", "image_url": {"url": uri}} + + redacted = helpers.redact_session_data( + {"messages": [{"role": "user", "content": [content_part]}]} + ) + assert redacted["messages"][0]["content"][0] == content_part + assert calls == [] + + +def test_raster_data_uri_outside_image_part_keeps_security_boundary(monkeypatch): + uri = _png_data_uri_with_sensitive_marker() + calls = [] + monkeypatch.setattr( + helpers, + "_redact_fn_cached", + lambda text: calls.append(text) or "redacted", + ) + + assert helpers._redact_value({"content": uri}, _enabled=True) == { + "content": "redacted", + } + assert calls == [uri] + + +def test_non_raster_image_part_keeps_security_boundary(monkeypatch): + uri = "data:image/svg+xml;base64," + base64.b64encode( + b" " + base64.b64decode("AKIA") + b" sensitive text" + ).decode("ascii") + assert "AKIA" in uri + calls = [] + monkeypatch.setattr( + helpers, + "_redact_fn_cached", + lambda text: calls.append(text) or "redacted", + ) + + result = helpers._redact_value( + {"type": "image_url", "image_url": {"url": uri}}, + _enabled=True, + ) + + assert result["image_url"]["url"] == "redacted" + assert calls == [uri] + + +def test_declared_raster_with_wrong_magic_keeps_security_boundary(monkeypatch): + encoded = base64.b64encode(b"not-a-png" + base64.b64decode("AKIA")).decode("ascii") + uri = f"data:image/png;base64,{encoded}" + calls = [] + monkeypatch.setattr( + helpers, + "_redact_fn_cached", + lambda text: calls.append(text) or "redacted", + ) + + result = helpers._redact_value( + {"type": "image_url", "image_url": {"url": uri}}, + _enabled=True, + ) + + assert result["image_url"]["url"] == "redacted" + assert calls == [uri] + + +def test_post_iend_credential_falls_through_to_text_redaction(monkeypatch): + uri = _png_data_uri_with_post_iend_credential() + calls = [] + + def redact(text): + calls.append(text) + return text.replace(_FAKE_AWS_KEY, "[REDACTED]") + + monkeypatch.setattr(helpers, "_redact_fn_cached", redact) + content_part = {"type": "image_url", "image_url": {"url": uri}} + + result = helpers.redact_session_data( + {"messages": [{"role": "user", "content": [content_part]}]} + ) + + assert _FAKE_AWS_KEY not in result["messages"][0]["content"][0]["image_url"]["url"] + assert calls == [uri] + assert helpers._is_native_raster_data_uri(uri) is False + + +def test_post_iend_credential_is_removed_by_real_redactor(): + uri = _png_data_uri_with_post_iend_credential() + result = helpers.redact_session_data( + { + "messages": [ + { + "role": "user", + "content": [{"type": "image_url", "image_url": {"url": uri}}], + } + ] + } + ) + + assert _FAKE_AWS_KEY not in json.dumps(result) + + +def test_image_exemption_is_scoped_to_direct_message_content_parts(monkeypatch): + valid_uri = _png_data_uri_with_sensitive_marker() + calls = [] + monkeypatch.setattr( + helpers, + "_redact_fn_cached", + lambda text: calls.append(text) or text.replace("AKIA", "[REDACTED]"), + ) + image_shaped = {"type": "image_url", "image_url": {"url": valid_uri}} + session = { + "messages": [ + { + "role": "user", + "content": [ + image_shaped, + {"type": "text", "text": f"sibling={_FAKE_AWS_KEY}"}, + ], + "metadata": image_shaped, + }, + {"role": "assistant", "content": [image_shaped]}, + ], + "tool_calls": [image_shaped], + "todo_state": {"image": image_shaped}, + "runtime_journal_snapshot": {"messages": [image_shaped]}, + } + + result = helpers.redact_session_data(session) + + # Only the canonical messages[*].content[*] image URL stays opaque. + assert result["messages"][0]["content"][0]["image_url"]["url"] == valid_uri + assert _FAKE_AWS_KEY not in result["messages"][0]["content"][1]["text"] + assert "AKIA" not in result["messages"][0]["metadata"]["image_url"]["url"] + assert "AKIA" not in result["messages"][1]["content"][0]["image_url"]["url"] + assert "AKIA" not in result["tool_calls"][0]["image_url"]["url"] + assert "AKIA" not in result["todo_state"]["image"]["image_url"]["url"] + assert "AKIA" not in result["runtime_journal_snapshot"]["messages"][0]["image_url"]["url"] + assert calls + + +def test_crafted_image_is_redacted_from_json_html_and_journal(monkeypatch): + uri = _png_data_uri_with_post_iend_credential() + monkeypatch.setattr( + helpers, + "_redact_fn_cached", + lambda text: text.replace(_FAKE_AWS_KEY, "[REDACTED]"), + ) + crafted_part = {"type": "image_url", "image_url": {"url": uri}} + safe = helpers.redact_session_data( + { + "title": "security regression", + "messages": [{"role": "user", "content": [crafted_part]}], + "runtime_journal_snapshot": { + "messages": [crafted_part], + "persisted_payload": uri, + }, + } + ) + + json_export = json.dumps(safe) + html_export = render_session_html(safe) + assert _FAKE_AWS_KEY not in json_export + assert _FAKE_AWS_KEY not in html_export + assert _FAKE_AWS_KEY not in json.dumps(safe["runtime_journal_snapshot"]) + + +def test_uncertain_base64_forms_fail_closed(): + valid_uri = _png_data_uri_with_sensitive_marker() + header, payload = valid_uri.split(",", 1) + hostile = [ + f"{header},{payload[:16]}\n{payload[16:]}", + f"{valid_uri}?token={_FAKE_AWS_KEY}", + f"{valid_uri} {_FAKE_AWS_KEY}", + f"{header},{payload[:-1]}", + f"{header},{payload}=A", + ] + + assert all(not helpers._is_native_raster_data_uri(value) for value in hostile) + + +@pytest.mark.parametrize( + "mime,raw", + [ + ("png", _valid_png_bytes_with_sensitive_marker()), + ("jpeg", _ONE_PIXEL_JPEG), + ("gif", _ONE_PIXEL_GIF), + ("webp", _ONE_PIXEL_WEBP), + ("bmp", _one_pixel_bmp()), + ], +) +def test_supported_raster_requires_exact_terminal_boundary(mime, raw): + exact = f"data:image/{mime};base64,{base64.b64encode(raw).decode('ascii')}" + trailing = ( + f"data:image/{mime};base64," + f"{base64.b64encode(raw + b'trailing-bytes').decode('ascii')}" + ) + + assert helpers._is_native_raster_data_uri(exact) is True + assert helpers._is_native_raster_data_uri(trailing) is False + + +def test_webp_animation_frame_requires_real_nested_image_chunk(): + fake_frame = b"\x00" * 16 + anmf = b"ANMF" + struct.pack("