From 0888c31aff4cc2d17c9e725d9922b6efcea9c703 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 30 Aug 2026 07:52:45 +0900 Subject: [PATCH 1/9] fix(pingora): exempt documentation raster evidence Signed-off-by: Seongho Bae --- docs/policies/PINGORA_EDGE_POLICY.md | 13 +++++++------ scripts/ci/pingora_edge_policy.py | 4 +++- tests/test_pingora_edge_policy.py | 16 ++++++++++++++++ 3 files changed, 26 insertions(+), 7 deletions(-) diff --git a/docs/policies/PINGORA_EDGE_POLICY.md b/docs/policies/PINGORA_EDGE_POLICY.md index 4d4c0752e1..d2dc4138fc 100644 --- a/docs/policies/PINGORA_EDGE_POLICY.md +++ b/docs/policies/PINGORA_EDGE_POLICY.md @@ -6,12 +6,13 @@ ContextualWisdomLab production and test edge runtimes use **Cloudflare Pingora** Active Nginx containers, packages, commands, configuration files, Kubernetes Nginx ingress annotations/classes, and host-service units are prohibited. -This is a runtime boundary, not a vocabulary ban. Documentation, license notices, -dedicated source fixtures under `tests/fixtures/`, the scanner source itself, and -migration histories may name Nginx. Executable integration and end-to-end test -helpers remain runtime candidates. Pull requests that modify a runtime candidate -are evaluated against the final exact head file, so deleting a legacy artifact is -allowed while preserving it or introducing a new one fails closed. +This is a runtime boundary, not a vocabulary ban. Documentation, non-executable +image evidence beneath a documentation directory, license notices, dedicated +source fixtures under `tests/fixtures/`, the scanner source itself, and migration +histories may name Nginx. Executable integration and end-to-end test helpers remain +runtime candidates. Pull requests that modify a runtime candidate are evaluated +against the final exact head file, so deleting a legacy artifact is allowed while +preserving it or introducing a new one fails closed. ## Why this is not a search-and-replace diff --git a/scripts/ci/pingora_edge_policy.py b/scripts/ci/pingora_edge_policy.py index 706fe69fc1..88d69e8137 100644 --- a/scripts/ci/pingora_edge_policy.py +++ b/scripts/ci/pingora_edge_policy.py @@ -29,6 +29,7 @@ GITHUB_API_ORIGIN = "https://api.github.com" DOCUMENT_SUFFIXES = frozenset({".md", ".mdx", ".rst", ".adoc", ".txt"}) +DOCUMENT_ASSET_SUFFIXES = frozenset({".gif", ".jpeg", ".jpg", ".png", ".webp"}) SOURCE_TEST_SUFFIXES = frozenset({".py", ".pyi", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".rs"}) LICENSE_NAMES = frozenset({"license", "license.md", "copying", "copyrights", "notice"}) DOCUMENTATION_DIRECTORIES = frozenset({"doc", "docs", "documentation"}) @@ -145,7 +146,8 @@ def _is_documentation_or_source_fixture(path: str) -> bool: or (len(pure.parts) == 1 and stem in DOCUMENTATION_ROOT_NAMES) ) if lower_name in LICENSE_NAMES or ( - is_known_documentation_path and pure.suffix.lower() in DOCUMENT_SUFFIXES + is_known_documentation_path + and pure.suffix.lower() in DOCUMENT_SUFFIXES | DOCUMENT_ASSET_SUFFIXES ): return True if pure.as_posix() == "scripts/ci/pingora_edge_policy.py": diff --git a/tests/test_pingora_edge_policy.py b/tests/test_pingora_edge_policy.py index 584e540749..48a098eed7 100644 --- a/tests/test_pingora_edge_policy.py +++ b/tests/test_pingora_edge_policy.py @@ -78,6 +78,22 @@ def test_nested_documentation_path_allows_prose_samples() -> None: assert policy.scan_content("packages/component/docs/migration.md", fixture_text()) == () +def test_documentation_image_assets_are_not_runtime_candidates() -> None: + """Binary screenshots under documentation cannot execute an edge runtime.""" + + changed = policy.ChangedFile( + "docs/screenshots/acceptance.png", + "added", + "", + patch_available=False, + ) + + assert not policy._needs_content_scan(changed) + assert policy._needs_content_scan( + policy.ChangedFile("public/acceptance.png", "added", "", patch_available=False) + ) + + @pytest.mark.parametrize("directory", ["testing", "contests", "assert", "my_tests"]) def test_scan_content_does_not_treat_test_name_substrings_as_fixtures( directory: str, From e29a96c41c76f1bc5c7996e6f731cf74d1ea433b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 30 Aug 2026 12:41:13 +0900 Subject: [PATCH 2/9] fix(pingora): validate documentation raster bytes Signed-off-by: Codex --- .../0019-cloudflare-pingora-edge-standard.md | 6 ++- docs/policies/PINGORA_EDGE_POLICY.md | 9 ++-- scripts/ci/pingora_edge_policy.py | 35 ++++++++++++- tests/test_pingora_edge_policy.py | 52 ++++++++++++++++++- 4 files changed, 92 insertions(+), 10 deletions(-) diff --git a/docs/adr/0019-cloudflare-pingora-edge-standard.md b/docs/adr/0019-cloudflare-pingora-edge-standard.md index 805e538b86..6b23006456 100644 --- a/docs/adr/0019-cloudflare-pingora-edge-standard.md +++ b/docs/adr/0019-cloudflare-pingora-edge-standard.md @@ -27,8 +27,10 @@ so a governed shared implementation is required. contracts. Environment deployment remains in `linux-cluster-ops`. 4. The organization required workflow rejects active Nginx runtime artifacts in changed final files without executing pull-request code. -5. Only dedicated source fixtures and the policy scanner may contain denied Nginx - samples; executable integration and end-to-end test helpers remain candidates +5. Documentation prose, license text, dedicated source fixtures, the policy + scanner, and byte-validated raster evidence beneath documentation directories + may contain denied Nginx samples. A documentation suffix alone is not an + exemption; executable integration and end-to-end test helpers remain candidates for enforcement. 6. Initial migration does not use Pingora's experimental cache integration. 7. PHP workloads move to an HTTP application server or reviewed FastCGI adapter diff --git a/docs/policies/PINGORA_EDGE_POLICY.md b/docs/policies/PINGORA_EDGE_POLICY.md index d2dc4138fc..ae4a226fae 100644 --- a/docs/policies/PINGORA_EDGE_POLICY.md +++ b/docs/policies/PINGORA_EDGE_POLICY.md @@ -6,8 +6,8 @@ ContextualWisdomLab production and test edge runtimes use **Cloudflare Pingora** Active Nginx containers, packages, commands, configuration files, Kubernetes Nginx ingress annotations/classes, and host-service units are prohibited. -This is a runtime boundary, not a vocabulary ban. Documentation, non-executable -image evidence beneath a documentation directory, license notices, dedicated +This is a runtime boundary, not a vocabulary ban. Documentation, byte-validated +raster evidence beneath a documentation directory, license notices, dedicated source fixtures under `tests/fixtures/`, the scanner source itself, and migration histories may name Nginx. Executable integration and end-to-end test helpers remain runtime candidates. Pull requests that modify a runtime candidate are evaluated @@ -54,8 +54,9 @@ The organization-required `required-workflow-bootstrap` job runs trusted base-branch scanner code at the immutable required-workflow SHA. It reads bounded changed-file metadata and final UTF-8 content through GitHub's REST API. It does not check out or execute pull-request content and receives only read permissions. -Malformed, truncated, binary, symlink, oversized, or unavailable evidence fails -closed. +Malformed, truncated, unexpected binary, symlink, oversized, or unavailable +evidence fails closed. Supported documentation raster evidence is fetched and +validated by file signature before it is excluded from the text scanner. ## Exception process diff --git a/scripts/ci/pingora_edge_policy.py b/scripts/ci/pingora_edge_policy.py index 88d69e8137..b4573d2be5 100644 --- a/scripts/ci/pingora_edge_policy.py +++ b/scripts/ci/pingora_edge_policy.py @@ -146,8 +146,7 @@ def _is_documentation_or_source_fixture(path: str) -> bool: or (len(pure.parts) == 1 and stem in DOCUMENTATION_ROOT_NAMES) ) if lower_name in LICENSE_NAMES or ( - is_known_documentation_path - and pure.suffix.lower() in DOCUMENT_SUFFIXES | DOCUMENT_ASSET_SUFFIXES + is_known_documentation_path and pure.suffix.lower() in DOCUMENT_SUFFIXES ): return True if pure.as_posix() == "scripts/ci/pingora_edge_policy.py": @@ -159,6 +158,32 @@ def _is_documentation_or_source_fixture(path: str) -> bool: return False +def _is_documentation_image(path: str) -> bool: + """Return whether *path* names a supported raster asset below documentation.""" + + pure = PurePosixPath(path) + return ( + any(part.lower() in DOCUMENTATION_DIRECTORIES for part in pure.parts) + and pure.suffix.lower() in DOCUMENT_ASSET_SUFFIXES + ) + + +def _is_supported_documentation_image(path: str, raw: bytes) -> bool: + """Validate a documentation image by its bounded bytes, not its suffix alone.""" + + suffix = PurePosixPath(path).suffix.lower() + signatures = { + ".gif": (b"GIF87a", b"GIF89a"), + ".jpeg": (b"\xff\xd8\xff",), + ".jpg": (b"\xff\xd8\xff",), + ".png": (b"\x89PNG\r\n\x1a\n",), + ".webp": (b"RIFF",), + } + if not any(raw.startswith(signature) for signature in signatures.get(suffix, ())): + return False + return suffix != ".webp" or len(raw) >= 12 and raw[8:12] == b"WEBP" + + def _runtime_path_rule(path: str) -> str | None: """Return a path-level violation rule for active Nginx runtime artifacts.""" @@ -296,6 +321,10 @@ def _load_file_content(api_url: str, repository: str, path: str, head_sha: str, raise PolicyError(f"GitHub content evidence for {path} is invalid base64") from exc if len(raw) != declared_size: raise PolicyError(f"GitHub content evidence for {path} has a size mismatch") + if _is_documentation_image(path): + if not _is_supported_documentation_image(path, raw): + raise PolicyError(f"Documentation image evidence {path} has invalid raster bytes") + return "" try: return raw.decode("utf-8") except UnicodeDecodeError as exc: @@ -307,6 +336,8 @@ def _needs_content_scan(changed: ChangedFile) -> bool: if changed.status == "removed" or _is_documentation_or_source_fixture(changed.path): return False + if _is_documentation_image(changed.path): + return True if not changed.patch_available: return True if _runtime_path_rule(changed.path) is not None: diff --git a/tests/test_pingora_edge_policy.py b/tests/test_pingora_edge_policy.py index 48a098eed7..9679fe4aa6 100644 --- a/tests/test_pingora_edge_policy.py +++ b/tests/test_pingora_edge_policy.py @@ -37,6 +37,17 @@ def encoded_file(content: str, *, size: int | None = None, kind: str = "file", e } +def encoded_bytes(content: bytes) -> dict[str, object]: + """Build one binary GitHub Contents API response.""" + + return { + "type": "file", + "encoding": "base64", + "size": len(content), + "content": base64.b64encode(content).decode(), + } + + def test_scan_content_rejects_runtime_paths_and_every_denied_runtime_form() -> None: """Runtime filenames and all supported active Nginx forms fail closed.""" @@ -79,7 +90,7 @@ def test_nested_documentation_path_allows_prose_samples() -> None: def test_documentation_image_assets_are_not_runtime_candidates() -> None: - """Binary screenshots under documentation cannot execute an edge runtime.""" + """Documentation screenshots require bounded byte validation.""" changed = policy.ChangedFile( "docs/screenshots/acceptance.png", @@ -88,12 +99,49 @@ def test_documentation_image_assets_are_not_runtime_candidates() -> None: patch_available=False, ) - assert not policy._needs_content_scan(changed) + assert policy._needs_content_scan(changed) assert policy._needs_content_scan( policy.ChangedFile("public/acceptance.png", "added", "", patch_available=False) ) +def test_documentation_image_requires_matching_raster_signature() -> None: + """An executable renamed with an image suffix cannot bypass enforcement.""" + + valid_png = b"\x89PNG\r\n\x1a\n" + b"bounded synthetic image" + + def valid_opener(url: str, _token: str) -> object: + if "/pulls/7/files" in url: + return [{"filename": "docs/screenshots/evidence.png", "status": "added"}] + return encoded_bytes(valid_png) + + assert policy.evaluate_pull_request( + api_url="https://api.github.test", + repository="ContextualWisdomLab/example", + pull_request=7, + head_sha="a" * 40, + event_action="opened", + token="token", + opener=valid_opener, + ) == () + + def renamed_script_opener(url: str, _token: str) -> object: + if "/pulls/8/files" in url: + return [{"filename": "docs/screenshots/evidence.png", "status": "added"}] + return encoded_bytes(b"#!/bin/sh\nnginx -g 'daemon off;'\n") + + with pytest.raises(policy.PolicyError, match="invalid raster bytes"): + policy.evaluate_pull_request( + api_url="https://api.github.test", + repository="ContextualWisdomLab/example", + pull_request=8, + head_sha="b" * 40, + event_action="opened", + token="token", + opener=renamed_script_opener, + ) + + @pytest.mark.parametrize("directory", ["testing", "contests", "assert", "my_tests"]) def test_scan_content_does_not_treat_test_name_substrings_as_fixtures( directory: str, From f67d22ac45a0f9da4fa7d9450d6c482924fa071c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 30 Aug 2026 12:53:13 +0900 Subject: [PATCH 3/9] fix(pingora): validate documentation raster evidence --- CHANGELOG.md | 3 + .../0019-cloudflare-pingora-edge-standard.md | 10 +- docs/doctoring/pingora-edge-standard.md | 6 +- docs/policies/PINGORA_EDGE_POLICY.md | 25 ++-- scripts/ci/pingora_edge_policy.py | 86 +++++++++++-- tests/test_pingora_edge_policy.py | 118 +++++++++++++++++- 6 files changed, 220 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3eab104fc2..32db0a70d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,9 @@ this file. The format follows Keep a Changelog, and versioned releases follow Semantic Versioning where the repository publishes a release. ## [Unreleased] +- Validate the final bytes of documentation raster evidence against its image + format before allowing it to name prohibited Nginx runtime forms; an image + filename alone no longer bypasses the Pingora policy scanner. - Skip trusted base Python lock materialization for exact-head reviews with no Python source or dependency-manifest changes, while preserving the fail-closed wheel-only path when Python coverage is relevant. diff --git a/docs/adr/0019-cloudflare-pingora-edge-standard.md b/docs/adr/0019-cloudflare-pingora-edge-standard.md index 805e538b86..735f0df835 100644 --- a/docs/adr/0019-cloudflare-pingora-edge-standard.md +++ b/docs/adr/0019-cloudflare-pingora-edge-standard.md @@ -27,9 +27,10 @@ so a governed shared implementation is required. contracts. Environment deployment remains in `linux-cluster-ops`. 4. The organization required workflow rejects active Nginx runtime artifacts in changed final files without executing pull-request code. -5. Only dedicated source fixtures and the policy scanner may contain denied Nginx +5. Only dedicated source fixtures, recognized raster image evidence beneath + documentation directories, and the policy scanner may contain denied Nginx samples; executable integration and end-to-end test helpers remain candidates - for enforcement. + for enforcement. Image extensions alone do not establish the exception. 6. Initial migration does not use Pingora's experimental cache integration. 7. PHP workloads move to an HTTP application server or reviewed FastCGI adapter behind Pingora before the public listener changes. @@ -66,5 +67,6 @@ so a governed shared implementation is required. The policy scanner has 100% production statement and branch coverage, bounded GitHub API evidence, path/control escaping, pagination limits, exact-head content -inspection, and fail-closed malformed-evidence tests. Product migrations require -site/proxy behavior tests and deployment-specific smoke tests before cutover. +inspection, raster-format validation, and fail-closed malformed-evidence tests. +Product migrations require site/proxy behavior tests and deployment-specific +smoke tests before cutover. diff --git a/docs/doctoring/pingora-edge-standard.md b/docs/doctoring/pingora-edge-standard.md index bb624a9c56..ad9dc2af6d 100644 --- a/docs/doctoring/pingora-edge-standard.md +++ b/docs/doctoring/pingora-edge-standard.md @@ -20,8 +20,10 @@ dependencies. The initial CWL implementation avoids experimental cache APIs. - The shared artifact is Apache-2.0 compatible with CWL permissive-license policy. - Required-workflow code is bound to its immutable central SHA and never executes pull-request content. -- Runtime evidence is bounded to one-megabyte UTF-8 regular files and a maximum of - 3,000 changed files; missing or malformed evidence fails closed. +- Runtime evidence is bounded to one-megabyte regular-file bytes, with UTF-8 + decoding for runtime candidates and supported raster-format validation for + documentation images; the maximum is 3,000 changed files and missing or + malformed evidence fails closed. - Exact-head product tests cover host/path routing, SPA fallback, security headers, WebSocket/streaming, body limits, health, metrics, TLS, and graceful shutdown as applicable. diff --git a/docs/policies/PINGORA_EDGE_POLICY.md b/docs/policies/PINGORA_EDGE_POLICY.md index d2dc4138fc..b599929b9e 100644 --- a/docs/policies/PINGORA_EDGE_POLICY.md +++ b/docs/policies/PINGORA_EDGE_POLICY.md @@ -6,13 +6,15 @@ ContextualWisdomLab production and test edge runtimes use **Cloudflare Pingora** Active Nginx containers, packages, commands, configuration files, Kubernetes Nginx ingress annotations/classes, and host-service units are prohibited. -This is a runtime boundary, not a vocabulary ban. Documentation, non-executable -image evidence beneath a documentation directory, license notices, dedicated -source fixtures under `tests/fixtures/`, the scanner source itself, and migration -histories may name Nginx. Executable integration and end-to-end test helpers remain -runtime candidates. Pull requests that modify a runtime candidate are evaluated -against the final exact head file, so deleting a legacy artifact is allowed while -preserving it or introducing a new one fails closed. +This is a runtime boundary, not a vocabulary ban. Documentation, recognized +non-executable image evidence beneath a documentation directory, license notices, +dedicated source fixtures under `tests/fixtures/`, the scanner source itself, and +migration histories may name Nginx. An image suffix alone is not an exception: +final bounded bytes must match the supported raster format. Executable integration +and end-to-end test helpers remain runtime candidates. Pull requests that modify a +runtime candidate are evaluated against the final exact head file, so deleting a +legacy artifact is allowed while preserving it or introducing a new one fails +closed. ## Why this is not a search-and-replace @@ -52,10 +54,11 @@ route/site contracts. Product repositories do not fork proxy internals. The organization-required `required-workflow-bootstrap` job runs trusted base-branch scanner code at the immutable required-workflow SHA. It reads bounded -changed-file metadata and final UTF-8 content through GitHub's REST API. It does -not check out or execute pull-request content and receives only read permissions. -Malformed, truncated, binary, symlink, oversized, or unavailable evidence fails -closed. +changed-file metadata and final file bytes through GitHub's REST API; runtime +candidates must decode as UTF-8, while documentation raster evidence must match +its supported format. It does not check out or execute pull-request content and +receives only read permissions. Malformed, truncated, unrecognized binary, +symlink, oversized, or unavailable evidence fails closed. ## Exception process diff --git a/scripts/ci/pingora_edge_policy.py b/scripts/ci/pingora_edge_policy.py index 88d69e8137..1655c4eb97 100644 --- a/scripts/ci/pingora_edge_policy.py +++ b/scripts/ci/pingora_edge_policy.py @@ -2,9 +2,9 @@ """Enforce the CWL Pingora-only edge runtime policy on pull-request changes. The checker never executes pull-request content. It reads changed-file metadata and -bounded UTF-8 file content through the GitHub REST API, then rejects active Nginx -runtime artifacts while allowing documentation, license text, and source-level -negative test fixtures. +bounded final file content through the GitHub REST API, then rejects active Nginx +runtime artifacts while allowing documentation, license text, recognized image +evidence, and source-level negative test fixtures. """ from __future__ import annotations @@ -15,6 +15,7 @@ import os import re import sys +import zlib from dataclasses import dataclass from pathlib import PurePosixPath from typing import Callable, Mapping, Sequence @@ -147,7 +148,7 @@ def _is_documentation_or_source_fixture(path: str) -> bool: ) if lower_name in LICENSE_NAMES or ( is_known_documentation_path - and pure.suffix.lower() in DOCUMENT_SUFFIXES | DOCUMENT_ASSET_SUFFIXES + and pure.suffix.lower() in DOCUMENT_SUFFIXES ): return True if pure.as_posix() == "scripts/ci/pingora_edge_policy.py": @@ -159,6 +160,60 @@ def _is_documentation_or_source_fixture(path: str) -> bool: return False +def _is_documentation_image_path(path: str) -> bool: + """Return whether *path* claims to be raster evidence below documentation.""" + + pure = PurePosixPath(path) + return bool( + pure.parts + and any(part.lower() in DOCUMENTATION_DIRECTORIES for part in pure.parts) + and pure.suffix.lower() in DOCUMENT_ASSET_SUFFIXES + ) + + +def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: + """Accept only bounded bytes with the expected raster format envelope.""" + + suffix = PurePosixPath(path).suffix.lower() + if suffix in {".jpeg", ".jpg"}: + return len(raw) >= 4 and raw.startswith(b"\xff\xd8\xff") and raw.endswith(b"\xff\xd9") + if suffix == ".gif": + return len(raw) >= 14 and raw[:6] in {b"GIF87a", b"GIF89a"} and raw[-1:] == b"\x3b" + if suffix == ".webp": + return ( + len(raw) >= 20 + and raw[:4] == b"RIFF" + and raw[8:12] == b"WEBP" + and int.from_bytes(raw[4:8], "little") == len(raw) - 8 + ) + if suffix != ".png" or len(raw) < 33 or not raw.startswith(b"\x89PNG\r\n\x1a\n"): + return False + offset = 8 + saw_idat = False + while offset + 12 <= len(raw): + length = int.from_bytes(raw[offset : offset + 4], "big") + end = offset + 12 + length + if end > len(raw): + return False + chunk_type = raw[offset + 4 : offset + 8] + chunk_data = raw[offset + 8 : offset + 8 + length] + declared_crc = int.from_bytes(raw[offset + 8 + length : end], "big") + if zlib.crc32(chunk_type + chunk_data) & 0xFFFFFFFF != declared_crc: + return False + if offset == 8 and (chunk_type != b"IHDR" or length != 13): + return False + if chunk_type == b"IHDR" and ( + int.from_bytes(chunk_data[:4], "big") == 0 + or int.from_bytes(chunk_data[4:8], "big") == 0 + ): + return False + saw_idat |= chunk_type == b"IDAT" + if chunk_type == b"IEND": + return length == 0 and saw_idat and end == len(raw) + offset = end + return False + + def _runtime_path_rule(path: str) -> str | None: """Return a path-level violation rule for active Nginx runtime artifacts.""" @@ -276,8 +331,8 @@ def _load_changed_files(api_url: str, repository: str, pull_request: int, token: raise PolicyError("GitHub changed-file pagination exceeded 3,000 files") -def _load_file_content(api_url: str, repository: str, path: str, head_sha: str, token: str, opener: OpenJson) -> str: - """Load one final head file as bounded UTF-8 text from the Contents API.""" +def _load_file_bytes(api_url: str, repository: str, path: str, head_sha: str, token: str, opener: OpenJson) -> bytes: + """Load one final head file as bounded bytes from the Contents API.""" encoded_path = quote(path, safe="/") url = f"{api_url}/repos/{repository}/contents/{encoded_path}?ref={head_sha}" @@ -296,8 +351,14 @@ def _load_file_content(api_url: str, repository: str, path: str, head_sha: str, raise PolicyError(f"GitHub content evidence for {path} is invalid base64") from exc if len(raw) != declared_size: raise PolicyError(f"GitHub content evidence for {path} has a size mismatch") + return raw + + +def _load_file_content(api_url: str, repository: str, path: str, head_sha: str, token: str, opener: OpenJson) -> str: + """Load one final head file as bounded UTF-8 text from the Contents API.""" + try: - return raw.decode("utf-8") + return _load_file_bytes(api_url, repository, path, head_sha, token, opener).decode("utf-8") except UnicodeDecodeError as exc: raise PolicyError(f"Runtime policy candidate {path} is not valid UTF-8") from exc @@ -305,7 +366,11 @@ def _load_file_content(api_url: str, repository: str, path: str, head_sha: str, def _needs_content_scan(changed: ChangedFile) -> bool: """Return whether a changed final file can carry an active edge runtime.""" - if changed.status == "removed" or _is_documentation_or_source_fixture(changed.path): + if changed.status == "removed": + return False + if _is_documentation_image_path(changed.path): + return True + if _is_documentation_or_source_fixture(changed.path): return False if not changed.patch_available: return True @@ -346,6 +411,11 @@ def evaluate_pull_request( for changed in changed_files: if not _needs_content_scan(changed): continue + if _is_documentation_image_path(changed.path): + raw = _load_file_bytes(api_url.rstrip("/"), repository, changed.path, head_sha, token, opener) + if not _is_recognized_documentation_image(changed.path, raw): + raise PolicyError(f"Documentation image evidence for {changed.path} is not a recognized image") + continue content = _load_file_content(api_url.rstrip("/"), repository, changed.path, head_sha, token, opener) violations.extend(scan_content(changed.path, content)) return tuple(violations) diff --git a/tests/test_pingora_edge_policy.py b/tests/test_pingora_edge_policy.py index 48a098eed7..a662663cd9 100644 --- a/tests/test_pingora_edge_policy.py +++ b/tests/test_pingora_edge_policy.py @@ -37,6 +37,22 @@ def encoded_file(content: str, *, size: int | None = None, kind: str = "file", e } +def encoded_binary_file(raw: bytes) -> dict[str, object]: + """Build one GitHub Contents API response for binary evidence.""" + + return { + "type": "file", + "encoding": "base64", + "size": len(raw), + "content": base64.b64encode(raw).decode(), + } + + +PNG_BYTES = base64.b64decode( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=" +) + + def test_scan_content_rejects_runtime_paths_and_every_denied_runtime_form() -> None: """Runtime filenames and all supported active Nginx forms fail closed.""" @@ -78,8 +94,8 @@ def test_nested_documentation_path_allows_prose_samples() -> None: assert policy.scan_content("packages/component/docs/migration.md", fixture_text()) == () -def test_documentation_image_assets_are_not_runtime_candidates() -> None: - """Binary screenshots under documentation cannot execute an edge runtime.""" +def test_documentation_image_assets_require_final_binary_evidence() -> None: + """Image suffixes alone cannot exempt documentation files from inspection.""" changed = policy.ChangedFile( "docs/screenshots/acceptance.png", @@ -88,12 +104,94 @@ def test_documentation_image_assets_are_not_runtime_candidates() -> None: patch_available=False, ) - assert not policy._needs_content_scan(changed) + assert policy._needs_content_scan(changed) assert policy._needs_content_scan( policy.ChangedFile("public/acceptance.png", "added", "", patch_available=False) ) +def test_documentation_image_suffix_cannot_hide_runtime_content() -> None: + """A text or script renamed to an image fails closed at final-content inspection.""" + + def opener(url: str, _token: str) -> object: + if "/pulls/17/files" in url: + return [{"filename": "docs/screenshots/acceptance.png", "status": "added"}] + return encoded_file("#!/bin/sh\nsystemctl restart nginx\n") + + with pytest.raises(policy.PolicyError, match="recognized image"): + policy.evaluate_pull_request( + api_url="https://api.github.test", + repository="ContextualWisdomLab/example", + pull_request=17, + head_sha="c" * 40, + event_action="opened", + token="token", + opener=opener, + ) + + +@pytest.mark.parametrize( + ("path", "raw", "recognized"), + [ + ("docs/acceptance.png", PNG_BYTES, True), + ("docs/acceptance.gif", b"GIF89a" + b"\x01\x00\x01\x00\x00\x00\x00" + b"\x3b", True), + ("docs/acceptance.jpg", b"\xff\xd8\xff\x00\xff\xd9", True), + ("docs/acceptance.jpeg", b"\xff\xd8\xff\x00\xff\xd9", True), + ("docs/acceptance.webp", b"RIFF" + (12).to_bytes(4, "little") + b"WEBPVP8 " + b"\x00\x00\x00\x00", True), + ("docs/acceptance.png", PNG_BYTES[:-12], False), + ("docs/acceptance.png", b"#!/bin/sh\nnginx\n", False), + ("docs/acceptance.svg", b"", False), + ], +) +def test_documentation_image_recognition_is_format_and_path_bound( + path: str, raw: bytes, recognized: bool +) -> None: + """Raster evidence requires the matching bounded format envelope.""" + + assert policy._is_recognized_documentation_image(path, raw) is recognized + + +def test_documentation_png_validation_rejects_truncated_corrupt_and_invalid_chunks() -> None: + """PNG evidence fails closed for truncated, corrupt, reordered, or empty images.""" + + truncated = bytearray(PNG_BYTES) + truncated[33:37] = (999).to_bytes(4, "big") + assert not policy._is_recognized_documentation_image("docs/acceptance.png", bytes(truncated)) + + corrupt_crc = bytearray(PNG_BYTES) + corrupt_crc[29] ^= 1 + assert not policy._is_recognized_documentation_image("docs/acceptance.png", bytes(corrupt_crc)) + + invalid_first_chunk = bytearray(PNG_BYTES) + invalid_first_chunk[12:16] = b"IDAT" + invalid_first_chunk[29:33] = policy.zlib.crc32(invalid_first_chunk[12:29]).to_bytes(4, "big") + assert not policy._is_recognized_documentation_image("docs/acceptance.png", bytes(invalid_first_chunk)) + + zero_dimension = bytearray(PNG_BYTES) + zero_dimension[16:20] = b"\x00\x00\x00\x00" + zero_dimension[29:33] = policy.zlib.crc32(zero_dimension[12:29]).to_bytes(4, "big") + assert not policy._is_recognized_documentation_image("docs/acceptance.png", bytes(zero_dimension)) + + +def test_evaluate_pull_request_accepts_recognized_documentation_image() -> None: + """A complete raster asset is accepted after final head bytes are verified.""" + + def opener(url: str, _token: str) -> object: + if "/pulls/18/files" in url: + return [{"filename": "docs/screenshots/acceptance.png", "status": "added"}] + return encoded_binary_file(PNG_BYTES) + + assert policy.evaluate_pull_request( + api_url="https://api.github.test", + repository="ContextualWisdomLab/example", + pull_request=18, + head_sha="d" * 40, + event_action="opened", + token="token", + opener=opener, + ) == () + + @pytest.mark.parametrize("directory", ["testing", "contests", "assert", "my_tests"]) def test_scan_content_does_not_treat_test_name_substrings_as_fixtures( directory: str, @@ -329,6 +427,20 @@ def opener(url: str, _token: str) -> object: assert calls[-1].endswith("page=31") +def test_changed_file_pagination_loop_bound_fails_closed() -> None: + """A full-looking final page cannot make bounded pagination loop forever.""" + + class ClaimedFullPage(list[dict[str, object]]): + """Model an adapter that reports a full page while yielding no entries.""" + + def __len__(self) -> int: + return 100 + + page = ClaimedFullPage() + with pytest.raises(policy.PolicyError, match="3,000"): + policy._load_changed_files("api", "a/b", 1, "x", lambda _url, _token: page) + + @pytest.mark.parametrize( ("payload", "message"), [ From 5e08f53774a6011504175a1ae28543e26acd3475 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 30 Aug 2026 15:04:01 +0900 Subject: [PATCH 4/9] fix(pingora): validate PNG evidence structurally --- CHANGELOG.md | 4 +- .../0019-cloudflare-pingora-edge-standard.md | 2 +- docs/doctoring/pingora-edge-standard.md | 4 +- docs/policies/PINGORA_EDGE_POLICY.md | 5 +- scripts/ci/pingora_edge_policy.py | 70 ++++++++++++++----- tests/test_pingora_edge_policy.py | 15 ++-- 6 files changed, 70 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 32db0a70d1..acc9718c78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,8 +5,8 @@ this file. The format follows Keep a Changelog, and versioned releases follow Semantic Versioning where the repository publishes a release. ## [Unreleased] -- Validate the final bytes of documentation raster evidence against its image - format before allowing it to name prohibited Nginx runtime forms; an image +- Validate the final bytes of documentation PNG evidence through chunk, CRC, + zlib-stream, dimension, and scanline checks before allowing it to name prohibited Nginx runtime forms; an image filename alone no longer bypasses the Pingora policy scanner. - Skip trusted base Python lock materialization for exact-head reviews with no Python source or dependency-manifest changes, while preserving the diff --git a/docs/adr/0019-cloudflare-pingora-edge-standard.md b/docs/adr/0019-cloudflare-pingora-edge-standard.md index 0ae16f3ac8..78eeb78728 100644 --- a/docs/adr/0019-cloudflare-pingora-edge-standard.md +++ b/docs/adr/0019-cloudflare-pingora-edge-standard.md @@ -28,7 +28,7 @@ so a governed shared implementation is required. 4. The organization required workflow rejects active Nginx runtime artifacts in changed final files without executing pull-request code. 5. Documentation prose, license text, dedicated source fixtures, the policy - scanner, and recognized raster image evidence beneath documentation directories + scanner, and structurally validated PNG evidence beneath documentation directories may contain denied Nginx samples. Image extensions alone do not establish the exception; executable integration and end-to-end test helpers remain candidates for enforcement. diff --git a/docs/doctoring/pingora-edge-standard.md b/docs/doctoring/pingora-edge-standard.md index ad9dc2af6d..d341662f9c 100644 --- a/docs/doctoring/pingora-edge-standard.md +++ b/docs/doctoring/pingora-edge-standard.md @@ -21,8 +21,8 @@ dependencies. The initial CWL implementation avoids experimental cache APIs. - Required-workflow code is bound to its immutable central SHA and never executes pull-request content. - Runtime evidence is bounded to one-megabyte regular-file bytes, with UTF-8 - decoding for runtime candidates and supported raster-format validation for - documentation images; the maximum is 3,000 changed files and missing or + decoding for runtime candidates and complete PNG chunk/CRC/zlib/scanline + validation for documentation images; the maximum is 3,000 changed files and missing or malformed evidence fails closed. - Exact-head product tests cover host/path routing, SPA fallback, security headers, WebSocket/streaming, body limits, health, metrics, TLS, and graceful shutdown as diff --git a/docs/policies/PINGORA_EDGE_POLICY.md b/docs/policies/PINGORA_EDGE_POLICY.md index b599929b9e..0b976be046 100644 --- a/docs/policies/PINGORA_EDGE_POLICY.md +++ b/docs/policies/PINGORA_EDGE_POLICY.md @@ -10,7 +10,7 @@ This is a runtime boundary, not a vocabulary ban. Documentation, recognized non-executable image evidence beneath a documentation directory, license notices, dedicated source fixtures under `tests/fixtures/`, the scanner source itself, and migration histories may name Nginx. An image suffix alone is not an exception: -final bounded bytes must match the supported raster format. Executable integration +final bounded bytes must be a structurally valid, completely decodable PNG. Executable integration and end-to-end test helpers remain runtime candidates. Pull requests that modify a runtime candidate are evaluated against the final exact head file, so deleting a legacy artifact is allowed while preserving it or introducing a new one fails @@ -55,7 +55,8 @@ route/site contracts. Product repositories do not fork proxy internals. The organization-required `required-workflow-bootstrap` job runs trusted base-branch scanner code at the immutable required-workflow SHA. It reads bounded changed-file metadata and final file bytes through GitHub's REST API; runtime -candidates must decode as UTF-8, while documentation raster evidence must match +candidates must decode as UTF-8, while documentation PNG evidence must pass +chunk-order, CRC, zlib-stream, dimension, and scanline validation and match its supported format. It does not check out or execute pull-request content and receives only read permissions. Malformed, truncated, unrecognized binary, symlink, oversized, or unavailable evidence fails closed. diff --git a/scripts/ci/pingora_edge_policy.py b/scripts/ci/pingora_edge_policy.py index 82b02dd132..f88048ec61 100644 --- a/scripts/ci/pingora_edge_policy.py +++ b/scripts/ci/pingora_edge_policy.py @@ -30,7 +30,7 @@ GITHUB_API_ORIGIN = "https://api.github.com" DOCUMENT_SUFFIXES = frozenset({".md", ".mdx", ".rst", ".adoc", ".txt"}) -DOCUMENT_ASSET_SUFFIXES = frozenset({".gif", ".jpeg", ".jpg", ".png", ".webp"}) +DOCUMENT_ASSET_SUFFIXES = frozenset({".png"}) SOURCE_TEST_SUFFIXES = frozenset({".py", ".pyi", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".rs"}) LICENSE_NAMES = frozenset({"license", "license.md", "copying", "copyrights", "notice"}) DOCUMENTATION_DIRECTORIES = frozenset({"doc", "docs", "documentation"}) @@ -175,21 +175,13 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: """Accept only bounded bytes with the expected raster format envelope.""" suffix = PurePosixPath(path).suffix.lower() - if suffix in {".jpeg", ".jpg"}: - return len(raw) >= 4 and raw.startswith(b"\xff\xd8\xff") and raw.endswith(b"\xff\xd9") - if suffix == ".gif": - return len(raw) >= 14 and raw[:6] in {b"GIF87a", b"GIF89a"} and raw[-1:] == b"\x3b" - if suffix == ".webp": - return ( - len(raw) >= 20 - and raw[:4] == b"RIFF" - and raw[8:12] == b"WEBP" - and int.from_bytes(raw[4:8], "little") == len(raw) - 8 - ) if suffix != ".png" or len(raw) < 33 or not raw.startswith(b"\x89PNG\r\n\x1a\n"): return False offset = 8 + width = height = bit_depth = color_type = None + idat_parts: list[bytes] = [] saw_idat = False + finished_idat = False while offset + 12 <= len(raw): length = int.from_bytes(raw[offset : offset + 4], "big") end = offset + 12 + length @@ -202,14 +194,54 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: return False if offset == 8 and (chunk_type != b"IHDR" or length != 13): return False - if chunk_type == b"IHDR" and ( - int.from_bytes(chunk_data[:4], "big") == 0 - or int.from_bytes(chunk_data[4:8], "big") == 0 - ): - return False - saw_idat |= chunk_type == b"IDAT" + if chunk_type == b"IHDR": + if offset != 8 or width is not None: + return False + width = int.from_bytes(chunk_data[:4], "big") + height = int.from_bytes(chunk_data[4:8], "big") + bit_depth, color_type, compression, filtering, interlace = chunk_data[8:13] + if ( + width == 0 + or height == 0 + or compression != 0 + or filtering != 0 + or interlace != 0 + or (color_type, bit_depth) + not in { + (0, 1), (0, 2), (0, 4), (0, 8), (0, 16), + (2, 8), (2, 16), + (3, 1), (3, 2), (3, 4), (3, 8), + (4, 8), (4, 16), + (6, 8), (6, 16), + } + ): + return False + elif chunk_type == b"IDAT": + if finished_idat: + return False + saw_idat = True + idat_parts.append(chunk_data) + elif saw_idat: + finished_idat = True if chunk_type == b"IEND": - return length == 0 and saw_idat and end == len(raw) + if length != 0 or not saw_idat or end != len(raw) or width is None or height is None: + return False + channels = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}[color_type] + row_bytes = (width * channels * bit_depth + 7) // 8 + expected_size = height * (row_bytes + 1) + try: + decoder = zlib.decompressobj() + pixels = decoder.decompress(b"".join(idat_parts), expected_size + 1) + pixels += decoder.flush() + except zlib.error: + return False + return ( + decoder.eof + and not decoder.unused_data + and not decoder.unconsumed_tail + and len(pixels) == expected_size + and all(pixels[row * (row_bytes + 1)] <= 4 for row in range(height)) + ) offset = end return False diff --git a/tests/test_pingora_edge_policy.py b/tests/test_pingora_edge_policy.py index 22797cc87e..6b09d3c3b9 100644 --- a/tests/test_pingora_edge_policy.py +++ b/tests/test_pingora_edge_policy.py @@ -133,10 +133,10 @@ def opener(url: str, _token: str) -> object: ("path", "raw", "recognized"), [ ("docs/acceptance.png", PNG_BYTES, True), - ("docs/acceptance.gif", b"GIF89a" + b"\x01\x00\x01\x00\x00\x00\x00" + b"\x3b", True), - ("docs/acceptance.jpg", b"\xff\xd8\xff\x00\xff\xd9", True), - ("docs/acceptance.jpeg", b"\xff\xd8\xff\x00\xff\xd9", True), - ("docs/acceptance.webp", b"RIFF" + (12).to_bytes(4, "little") + b"WEBPVP8 " + b"\x00\x00\x00\x00", True), + ("docs/acceptance.gif", b"GIF89a" + b"\x01\x00\x01\x00\x00\x00\x00" + b"\x3b", False), + ("docs/acceptance.jpg", b"\xff\xd8\xff\x00\xff\xd9", False), + ("docs/acceptance.jpeg", b"\xff\xd8\xff\x00\xff\xd9", False), + ("docs/acceptance.webp", b"RIFF" + (12).to_bytes(4, "little") + b"WEBPVP8 " + b"\x00\x00\x00\x00", False), ("docs/acceptance.png", PNG_BYTES[:-12], False), ("docs/acceptance.png", b"#!/bin/sh\nnginx\n", False), ("docs/acceptance.svg", b"", False), @@ -171,6 +171,13 @@ def test_documentation_png_validation_rejects_truncated_corrupt_and_invalid_chun zero_dimension[29:33] = policy.zlib.crc32(zero_dimension[12:29]).to_bytes(4, "big") assert not policy._is_recognized_documentation_image("docs/acceptance.png", bytes(zero_dimension)) + empty_idat = bytearray(PNG_BYTES) + empty_idat[41:45] = b"x\x9c\x03\x00" + empty_idat[37:41] = (4).to_bytes(4, "big") + del empty_idat[45:48] + empty_idat[45:49] = policy.zlib.crc32(empty_idat[41:45]).to_bytes(4, "big") + assert not policy._is_recognized_documentation_image("docs/acceptance.png", bytes(empty_idat)) + def test_evaluate_pull_request_accepts_recognized_documentation_image() -> None: """A complete raster asset is accepted after final head bytes are verified.""" From 25f4d2db6eccbea669a2c0990247a72918699cbf Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 30 Aug 2026 15:26:20 +0900 Subject: [PATCH 5/9] fix(pingora): bound complete PNG validation --- scripts/ci/pingora_edge_policy.py | 43 ++++++++++++----- tests/test_pingora_edge_policy.py | 79 +++++++++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 13 deletions(-) diff --git a/scripts/ci/pingora_edge_policy.py b/scripts/ci/pingora_edge_policy.py index f88048ec61..126f0c0dbb 100644 --- a/scripts/ci/pingora_edge_policy.py +++ b/scripts/ci/pingora_edge_policy.py @@ -25,6 +25,7 @@ MAX_FILE_BYTES = 1_048_576 MAX_RESPONSE_BYTES = 16_777_216 +MAX_IMAGE_DECODED_BYTES = MAX_RESPONSE_BYTES REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") SHA_RE = re.compile(r"^[0-9a-f]{40}$") GITHUB_API_ORIGIN = "https://api.github.com" @@ -178,7 +179,7 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: if suffix != ".png" or len(raw) < 33 or not raw.startswith(b"\x89PNG\r\n\x1a\n"): return False offset = 8 - width = height = bit_depth = color_type = None + width = height = bit_depth = color_type = interlace = None idat_parts: list[bytes] = [] saw_idat = False finished_idat = False @@ -189,6 +190,10 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: return False chunk_type = raw[offset + 4 : offset + 8] chunk_data = raw[offset + 8 : offset + 8 + length] + if len(chunk_type) != 4 or any(not (65 <= byte <= 90 or 97 <= byte <= 122) for byte in chunk_type): + return False + if chunk_type[0] & 0x20 == 0 and chunk_type not in {b"IHDR", b"IDAT", b"IEND"}: + return False declared_crc = int.from_bytes(raw[offset + 8 + length : end], "big") if zlib.crc32(chunk_type + chunk_data) & 0xFFFFFFFF != declared_crc: return False @@ -203,17 +208,12 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: if ( width == 0 or height == 0 + or width > MAX_IMAGE_DECODED_BYTES + or height > MAX_IMAGE_DECODED_BYTES or compression != 0 or filtering != 0 - or interlace != 0 - or (color_type, bit_depth) - not in { - (0, 1), (0, 2), (0, 4), (0, 8), (0, 16), - (2, 8), (2, 16), - (3, 1), (3, 2), (3, 4), (3, 8), - (4, 8), (4, 16), - (6, 8), (6, 16), - } + or interlace not in {0, 1} + or (color_type, bit_depth) not in {(4, 8), (6, 8)} ): return False elif chunk_type == b"IDAT": @@ -227,8 +227,25 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: if length != 0 or not saw_idat or end != len(raw) or width is None or height is None: return False channels = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}[color_type] - row_bytes = (width * channels * bit_depth + 7) // 8 - expected_size = height * (row_bytes + 1) + passes = [(0, 0, 1, 1)] if interlace == 0 else [ + (0, 0, 8, 8), (4, 0, 8, 8), (0, 4, 4, 8), + (2, 0, 4, 4), (0, 2, 2, 4), (1, 0, 2, 2), (0, 1, 1, 2), + ] + filter_offsets: list[int] = [] + expected_size = 0 + for x_start, y_start, x_step, y_step in passes: + pass_width = 0 if width <= x_start else (width - x_start + x_step - 1) // x_step + pass_height = 0 if height <= y_start else (height - y_start + y_step - 1) // y_step + if not pass_width or not pass_height: + continue + row_bytes = (pass_width * channels * bit_depth + 7) // 8 + filter_offsets.extend( + expected_size + row * (row_bytes + 1) + for row in range(pass_height) + ) + expected_size += pass_height * (row_bytes + 1) + if expected_size > MAX_IMAGE_DECODED_BYTES: + return False try: decoder = zlib.decompressobj() pixels = decoder.decompress(b"".join(idat_parts), expected_size + 1) @@ -240,7 +257,7 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: and not decoder.unused_data and not decoder.unconsumed_tail and len(pixels) == expected_size - and all(pixels[row * (row_bytes + 1)] <= 4 for row in range(height)) + and all(pixels[offset] <= 4 for offset in filter_offsets) ) offset = end return False diff --git a/tests/test_pingora_edge_policy.py b/tests/test_pingora_edge_policy.py index 6b09d3c3b9..7520bad903 100644 --- a/tests/test_pingora_edge_policy.py +++ b/tests/test_pingora_edge_policy.py @@ -52,6 +52,46 @@ def encoded_binary_file(raw: bytes) -> dict[str, object]: "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=" ) + +def replace_png_ihdr(raw: bytes, **fields: int) -> bytes: + """Return *raw* with selected IHDR fields and a matching CRC.""" + + image = bytearray(raw) + positions = {"width": (16, 20), "height": (20, 24), "bit_depth": (24, 25), "color_type": (25, 26), "interlace": (28, 29)} + for name, value in fields.items(): + start, end = positions[name] + image[start:end] = value.to_bytes(end - start, "big") + image[29:33] = policy.zlib.crc32(image[12:29]).to_bytes(4, "big") + return bytes(image) + + +def insert_png_chunk(raw: bytes, chunk_type: bytes, data: bytes = b"") -> bytes: + """Insert one CRC-valid chunk immediately before the first IDAT.""" + + idat_offset = raw.index(b"IDAT") - 4 + chunk = len(data).to_bytes(4, "big") + chunk_type + data + chunk += policy.zlib.crc32(chunk_type + data).to_bytes(4, "big") + return raw[:idat_offset] + chunk + raw[idat_offset:] + + +def png_chunk(chunk_type: bytes, data: bytes = b"") -> bytes: + """Build one CRC-valid PNG chunk.""" + + chunk = len(data).to_bytes(4, "big") + chunk_type + data + return chunk + policy.zlib.crc32(chunk_type + data).to_bytes(4, "big") + + +def replace_png_chunk(raw: bytes, chunk_type: bytes, data: bytes) -> bytes: + """Replace the first named PNG chunk with CRC-valid *data*.""" + + type_offset = raw.index(chunk_type, 8) + chunk_offset = type_offset - 4 + old_length = int.from_bytes(raw[chunk_offset:type_offset], "big") + old_end = type_offset + 8 + old_length + chunk = len(data).to_bytes(4, "big") + chunk_type + data + chunk += policy.zlib.crc32(chunk_type + data).to_bytes(4, "big") + return raw[:chunk_offset] + chunk + raw[old_end:] + def test_scan_content_rejects_runtime_paths_and_every_denied_runtime_form() -> None: """Runtime filenames and all supported active Nginx forms fail closed.""" @@ -178,6 +218,45 @@ def test_documentation_png_validation_rejects_truncated_corrupt_and_invalid_chun empty_idat[45:49] = policy.zlib.crc32(empty_idat[41:45]).to_bytes(4, "big") assert not policy._is_recognized_documentation_image("docs/acceptance.png", bytes(empty_idat)) + assert policy._is_recognized_documentation_image( + "docs/acceptance.png", replace_png_ihdr(PNG_BYTES, interlace=1) + ) + assert policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(PNG_BYTES, b"tEXt") + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", replace_png_ihdr(PNG_BYTES, color_type=3) + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(PNG_BYTES, b"ABCD") + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", + replace_png_ihdr(PNG_BYTES, width=policy.MAX_IMAGE_DECODED_BYTES + 1), + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(PNG_BYTES, b"IHDR", PNG_BYTES[16:29]) + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(PNG_BYTES, b"AB1D") + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", replace_png_chunk(PNG_BYTES, b"IDAT", b"not-zlib") + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", replace_png_ihdr(PNG_BYTES, width=policy.MAX_IMAGE_DECODED_BYTES) + ) + idat_offset = PNG_BYTES.index(b"IDAT") - 4 + iend_offset = PNG_BYTES.index(b"IEND") - 4 + duplicate_idat = PNG_BYTES[idat_offset:iend_offset] + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", + PNG_BYTES[:iend_offset] + png_chunk(b"tEXt") + duplicate_idat + PNG_BYTES[iend_offset:], + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", PNG_BYTES + b"trailing" + ) + def test_evaluate_pull_request_accepts_recognized_documentation_image() -> None: """A complete raster asset is accepted after final head bytes are verified.""" From c1d1495f8c043f28d3685b6e6997c5734649d25f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 30 Aug 2026 23:44:42 +0900 Subject: [PATCH 6/9] codex: address PR review feedback (#1420) --- scripts/ci/pingora_edge_policy.py | 68 ++++++++++++++++++++------ tests/test_pingora_edge_policy.py | 79 +++++++++++++++++++++++++++++++ 2 files changed, 133 insertions(+), 14 deletions(-) diff --git a/scripts/ci/pingora_edge_policy.py b/scripts/ci/pingora_edge_policy.py index 126f0c0dbb..34a34e30e7 100644 --- a/scripts/ci/pingora_edge_policy.py +++ b/scripts/ci/pingora_edge_policy.py @@ -181,6 +181,7 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: offset = 8 width = height = bit_depth = color_type = interlace = None idat_parts: list[bytes] = [] + saw_plte = False saw_idat = False finished_idat = False while offset + 12 <= len(raw): @@ -190,9 +191,18 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: return False chunk_type = raw[offset + 4 : offset + 8] chunk_data = raw[offset + 8 : offset + 8 + length] - if len(chunk_type) != 4 or any(not (65 <= byte <= 90 or 97 <= byte <= 122) for byte in chunk_type): + if ( + len(chunk_type) != 4 + or any(not (65 <= byte <= 90 or 97 <= byte <= 122) for byte in chunk_type) + or chunk_type[2] & 0x20 + ): return False - if chunk_type[0] & 0x20 == 0 and chunk_type not in {b"IHDR", b"IDAT", b"IEND"}: + if chunk_type[0] & 0x20 == 0 and chunk_type not in { + b"IHDR", + b"PLTE", + b"IDAT", + b"IEND", + }: return False declared_crc = int.from_bytes(raw[offset + 8 + length : end], "big") if zlib.crc32(chunk_type + chunk_data) & 0xFFFFFFFF != declared_crc: @@ -213,25 +223,52 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: or compression != 0 or filtering != 0 or interlace not in {0, 1} - or (color_type, bit_depth) not in {(4, 8), (6, 8)} + or bit_depth + not in { + 0: {1, 2, 4, 8, 16}, + 2: {8, 16}, + 3: {1, 2, 4, 8}, + 4: {8, 16}, + 6: {8, 16}, + }.get(color_type, set()) ): return False + elif chunk_type == b"PLTE": + if ( + width is None + or saw_plte + or saw_idat + or color_type in {0, 4} + or length == 0 + or length % 3 + or length > 768 + or (color_type == 3 and length // 3 > 2**bit_depth) + ): + return False + saw_plte = True elif chunk_type == b"IDAT": - if finished_idat: + if width is None or finished_idat or (color_type == 3 and not saw_plte): return False saw_idat = True idat_parts.append(chunk_data) elif saw_idat: finished_idat = True if chunk_type == b"IEND": - if length != 0 or not saw_idat or end != len(raw) or width is None or height is None: + if ( + length != 0 + or not saw_idat + or end != len(raw) + or width is None + or height is None + or (color_type == 3 and not saw_plte) + ): return False channels = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}[color_type] passes = [(0, 0, 1, 1)] if interlace == 0 else [ (0, 0, 8, 8), (4, 0, 8, 8), (0, 4, 4, 8), (2, 0, 4, 4), (0, 2, 2, 4), (1, 0, 2, 2), (0, 1, 1, 2), ] - filter_offsets: list[int] = [] + pass_layouts: list[tuple[int, int, int]] = [] expected_size = 0 for x_start, y_start, x_step, y_step in passes: pass_width = 0 if width <= x_start else (width - x_start + x_step - 1) // x_step @@ -239,13 +276,12 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: if not pass_width or not pass_height: continue row_bytes = (pass_width * channels * bit_depth + 7) // 8 - filter_offsets.extend( - expected_size + row * (row_bytes + 1) - for row in range(pass_height) - ) - expected_size += pass_height * (row_bytes + 1) - if expected_size > MAX_IMAGE_DECODED_BYTES: - return False + stride = row_bytes + 1 + pass_size = pass_height * stride + if pass_size > MAX_IMAGE_DECODED_BYTES - expected_size: + return False + pass_layouts.append((expected_size, stride, pass_height)) + expected_size += pass_size try: decoder = zlib.decompressobj() pixels = decoder.decompress(b"".join(idat_parts), expected_size + 1) @@ -257,7 +293,11 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: and not decoder.unused_data and not decoder.unconsumed_tail and len(pixels) == expected_size - and all(pixels[offset] <= 4 for offset in filter_offsets) + and all( + pixels[start + row * stride] <= 4 + for start, stride, rows in pass_layouts + for row in range(rows) + ) ) offset = end return False diff --git a/tests/test_pingora_edge_policy.py b/tests/test_pingora_edge_policy.py index 7520bad903..1fa7efc89e 100644 --- a/tests/test_pingora_edge_policy.py +++ b/tests/test_pingora_edge_policy.py @@ -92,6 +92,31 @@ def replace_png_chunk(raw: bytes, chunk_type: bytes, data: bytes) -> bytes: chunk += policy.zlib.crc32(chunk_type + data).to_bytes(4, "big") return raw[:chunk_offset] + chunk + raw[old_end:] + +def build_png( + *, + color_type: int, + bit_depth: int, + width: int = 2, + height: int = 2, + palette: bytes | None = None, +) -> bytes: + """Build a small structurally complete, non-interlaced PNG fixture.""" + + channels = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}[color_type] + row_bytes = (width * channels * bit_depth + 7) // 8 + pixels = b"".join(b"\x00" + bytes(row_bytes) for _ in range(height)) + ihdr = ( + width.to_bytes(4, "big") + + height.to_bytes(4, "big") + + bytes((bit_depth, color_type, 0, 0, 0)) + ) + chunks = [png_chunk(b"IHDR", ihdr)] + if palette is not None: + chunks.append(png_chunk(b"PLTE", palette)) + chunks.extend((png_chunk(b"IDAT", policy.zlib.compress(pixels)), png_chunk(b"IEND"))) + return b"\x89PNG\r\n\x1a\n" + b"".join(chunks) + def test_scan_content_rejects_runtime_paths_and_every_denied_runtime_form() -> None: """Runtime filenames and all supported active Nginx forms fail closed.""" @@ -258,6 +283,60 @@ def test_documentation_png_validation_rejects_truncated_corrupt_and_invalid_chun ) +@pytest.mark.parametrize( + ("color_type", "bit_depth"), + [ + (0, 1), + (0, 2), + (0, 4), + (0, 8), + (0, 16), + (2, 8), + (2, 16), + (3, 1), + (3, 2), + (3, 4), + (3, 8), + (4, 8), + (4, 16), + (6, 8), + (6, 16), + ], +) +def test_documentation_png_accepts_every_legal_color_and_depth_pair( + color_type: int, bit_depth: int +) -> None: + """The bounded validator accepts each PNG-defined color/depth combination.""" + + palette = b"\x00\x00\x00" if color_type == 3 else None + raw = build_png(color_type=color_type, bit_depth=bit_depth, palette=palette) + assert policy._is_recognized_documentation_image("docs/acceptance.png", raw) + + +def test_documentation_png_enforces_palette_and_chunk_name_contracts() -> None: + """Palette placement, cardinality, and the reserved chunk-name bit fail closed.""" + + indexed_without_palette = build_png(color_type=3, bit_depth=1) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", indexed_without_palette + ) + + rgba_with_palette = build_png( + color_type=6, + bit_depth=8, + palette=b"\x00\x00\x00\xff\xff\xff", + ) + assert policy._is_recognized_documentation_image( + "docs/acceptance.png", rgba_with_palette + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(rgba_with_palette, b"PLTE", b"\x00\x00\x00") + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(PNG_BYTES, b"abca") + ) + + def test_evaluate_pull_request_accepts_recognized_documentation_image() -> None: """A complete raster asset is accepted after final head bytes are verified.""" From 78f890ec28919192aa7e99ed8ac061687d795c98 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 31 Aug 2026 08:50:32 +0900 Subject: [PATCH 7/9] codex: address PR review feedback (#1420) Signed-off-by: Seongho Bae --- scripts/ci/pingora_edge_policy.py | 79 ++++++++++++++++++++++++++++--- tests/test_pingora_edge_policy.py | 52 ++++++++++++++++++++ 2 files changed, 124 insertions(+), 7 deletions(-) diff --git a/scripts/ci/pingora_edge_policy.py b/scripts/ci/pingora_edge_policy.py index 190b656cb9..17d229776f 100644 --- a/scripts/ci/pingora_edge_policy.py +++ b/scripts/ci/pingora_edge_policy.py @@ -235,7 +235,58 @@ def _is_documentation_image_path(path: str) -> bool: """Return whether *path* claims to be raster evidence below documentation.""" pure = PurePosixPath(path) - return _is_known_documentation_path(pure) and pure.suffix.lower() in DOCUMENT_ASSET_SUFFIXES + return ( + len(pure.parts) > 1 + and any(part.lower() in DOCUMENTATION_DIRECTORIES for part in pure.parts[:-1]) + and pure.suffix.lower() in DOCUMENT_ASSET_SUFFIXES + ) + + +def _paeth_predictor(left: int, above: int, upper_left: int) -> int: + """Return the PNG Paeth predictor for three reconstructed bytes.""" + + estimate = left + above - upper_left + distances = (abs(estimate - left), abs(estimate - above), abs(estimate - upper_left)) + return (left, above, upper_left)[distances.index(min(distances))] + + +def _png_rows_are_valid( + pixels: bytes, + pass_layouts: list[tuple[int, int, int, int]], + *, + bytes_per_pixel: int, + bit_depth: int, + palette_entries: int | None, +) -> bool: + """Validate filters and indexed samples for every decoded PNG pass.""" + + for start, stride, rows, pass_width in pass_layouts: + previous = bytes(stride - 1) + for row in range(rows): + row_start = start + row * stride + filter_type = pixels[row_start] + if filter_type > 4: + return False + encoded = pixels[row_start + 1 : row_start + stride] + reconstructed = bytearray(len(encoded)) + for index, value in enumerate(encoded): + left = reconstructed[index - bytes_per_pixel] if index >= bytes_per_pixel else 0 + above = previous[index] + upper_left = previous[index - bytes_per_pixel] if index >= bytes_per_pixel else 0 + predictor = (0, left, above, (left + above) // 2, _paeth_predictor(left, above, upper_left))[filter_type] + reconstructed[index] = (value + predictor) & 0xFF + previous = bytes(reconstructed) + if palette_entries is None: + continue + samples_seen = 0 + for byte in reconstructed: + for shift in range(8 - bit_depth, -1, -bit_depth): + if samples_seen == pass_width: + break + if ((byte >> shift) & ((1 << bit_depth) - 1)) >= palette_entries: + return False + samples_seen += 1 + return True def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: @@ -248,6 +299,8 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: width = height = bit_depth = color_type = interlace = None idat_parts: list[bytes] = [] saw_plte = False + palette_entries = None + saw_trns = False saw_idat = False finished_idat = False while offset + 12 <= len(raw): @@ -312,6 +365,16 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: ): return False saw_plte = True + palette_entries = length // 3 + elif chunk_type == b"tRNS": + valid_length = { + 0: length == 2, + 2: length == 6, + 3: saw_plte and 0 < length <= (palette_entries or 0), + }.get(color_type, False) + if width is None or saw_trns or saw_idat or not valid_length: + return False + saw_trns = True elif chunk_type == b"IDAT": if width is None or finished_idat or (color_type == 3 and not saw_plte): return False @@ -343,7 +406,7 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: (0, 1, 1, 2), ] ) - pass_layouts: list[tuple[int, int, int]] = [] + pass_layouts: list[tuple[int, int, int, int]] = [] expected_size = 0 for x_start, y_start, x_step, y_step in passes: pass_width = 0 if width <= x_start else (width - x_start + x_step - 1) // x_step @@ -355,7 +418,7 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: pass_size = pass_height * stride if pass_size > MAX_IMAGE_DECODED_BYTES - expected_size: return False - pass_layouts.append((expected_size, stride, pass_height)) + pass_layouts.append((expected_size, stride, pass_height, pass_width)) expected_size += pass_size try: decoder = zlib.decompressobj() @@ -368,10 +431,12 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: and not decoder.unused_data and not decoder.unconsumed_tail and len(pixels) == expected_size - and all( - pixels[start + row * stride] <= 4 - for start, stride, rows in pass_layouts - for row in range(rows) + and _png_rows_are_valid( + pixels, + pass_layouts, + bytes_per_pixel=max(1, (channels * bit_depth + 7) // 8), + bit_depth=bit_depth, + palette_entries=palette_entries if color_type == 3 else None, ) ) offset = end diff --git a/tests/test_pingora_edge_policy.py b/tests/test_pingora_edge_policy.py index 86960e3612..f777d411d1 100644 --- a/tests/test_pingora_edge_policy.py +++ b/tests/test_pingora_edge_policy.py @@ -212,6 +212,58 @@ def test_documentation_png_rejects_disguised_text_and_invalid_structure() -> Non assert not policy._is_recognized_documentation_image( "docs/acceptance.png", insert_png_chunk(rgba, b"abca") ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(rgba, b"tRNS", b"\x00") + ) + grayscale = build_png(color_type=0, bit_depth=8) + assert policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(grayscale, b"tRNS", b"\x00\x00") + ) + + +@pytest.mark.parametrize("filter_type", [0, 1, 2, 3, 4]) +def test_documentation_png_reconstructs_each_legal_filter(filter_type: int) -> None: + """Every standard PNG scanline filter is decoded before acceptance.""" + + raw = build_png(color_type=6, bit_depth=8, width=1, height=1) + raw = replace_png_chunk(raw, b"IDAT", policy.zlib.compress(bytes((filter_type, 0, 0, 0, 0)))) + assert policy._is_recognized_documentation_image("docs/acceptance.png", raw) + + +def test_documentation_png_rejects_unknown_filter() -> None: + """A structurally sized stream still fails on an unknown filter byte.""" + + raw = build_png(color_type=6, bit_depth=8, width=1, height=1) + raw = replace_png_chunk(raw, b"IDAT", policy.zlib.compress(b"\x05\x00\x00\x00\x00")) + assert not policy._is_recognized_documentation_image("docs/acceptance.png", raw) + + +@pytest.mark.parametrize("filter_type", [0, 1]) +@pytest.mark.parametrize("interlace", [0, 1]) +def test_documentation_png_rejects_missing_palette_entries( + filter_type: int, interlace: int +) -> None: + """Indexed pixels cannot reference an entry absent from PLTE.""" + + raw = build_png( + color_type=3, + bit_depth=8, + width=1, + height=1, + palette=b"\x00\x00\x00", + ) + raw = replace_png_chunk(raw, b"IDAT", policy.zlib.compress(bytes((filter_type, 1)))) + if interlace: + raw = replace_png_ihdr(raw, interlace=1) + assert not policy._is_recognized_documentation_image("docs/acceptance.png", raw) + + +def test_documentation_image_path_requires_a_documentation_directory() -> None: + """Root-level image names do not cross the documented directory boundary.""" + + assert policy._is_documentation_image_path("docs/screenshots/acceptance.png") + assert not policy._is_documentation_image_path("README.png") + assert not policy._is_documentation_image_path("CHANGELOG.png") def test_documentation_png_rejects_each_malformed_envelope_boundary() -> None: From 212ade1d1750e66851bc8ef57defa2c02a899e12 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 31 Aug 2026 08:56:14 +0900 Subject: [PATCH 8/9] codex: address PR review feedback (#1420) Signed-off-by: Seongho Bae --- scripts/ci/pingora_edge_policy.py | 10 ++++++++-- tests/test_pingora_edge_policy.py | 16 ++++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/scripts/ci/pingora_edge_policy.py b/scripts/ci/pingora_edge_policy.py index 17d229776f..1eabbdd0f5 100644 --- a/scripts/ci/pingora_edge_policy.py +++ b/scripts/ci/pingora_edge_policy.py @@ -356,6 +356,7 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: if ( width is None or saw_plte + or saw_trns or saw_idat or color_type in {0, 4} or length == 0 @@ -367,9 +368,14 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: saw_plte = True palette_entries = length // 3 elif chunk_type == b"tRNS": + sample_limit = (1 << bit_depth) - 1 if bit_depth is not None else -1 valid_length = { - 0: length == 2, - 2: length == 6, + 0: length == 2 and int.from_bytes(chunk_data, "big") <= sample_limit, + 2: length == 6 + and all( + int.from_bytes(chunk_data[index : index + 2], "big") <= sample_limit + for index in (0, 2, 4) + ), 3: saw_plte and 0 < length <= (palette_entries or 0), }.get(color_type, False) if width is None or saw_trns or saw_idat or not valid_length: diff --git a/tests/test_pingora_edge_policy.py b/tests/test_pingora_edge_policy.py index f777d411d1..21be78926a 100644 --- a/tests/test_pingora_edge_policy.py +++ b/tests/test_pingora_edge_policy.py @@ -219,6 +219,22 @@ def test_documentation_png_rejects_disguised_text_and_invalid_structure() -> Non assert policy._is_recognized_documentation_image( "docs/acceptance.png", insert_png_chunk(grayscale, b"tRNS", b"\x00\x00") ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(grayscale, b"tRNS", b"\x01\x00") + ) + truecolor = build_png(color_type=2, bit_depth=8) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", + insert_png_chunk(truecolor, b"tRNS", b"\x00\x00\x00\x00\x01\x00"), + ) + truecolor_with_palette_after_transparency = insert_png_chunk( + insert_png_chunk(truecolor, b"tRNS", b"\x00" * 6), + b"PLTE", + b"\x00\x00\x00", + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", truecolor_with_palette_after_transparency + ) @pytest.mark.parametrize("filter_type", [0, 1, 2, 3, 4]) From d84d65cc1a5b32cda3f5a723a61bab82be0c3d1e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 31 Aug 2026 14:50:32 +0900 Subject: [PATCH 9/9] fix(pingora): validate PNG color chunk contracts Signed-off-by: Seongho Bae --- scripts/ci/pingora_edge_policy.py | 50 ++++++++++++++++++++++++++++ tests/test_pingora_edge_policy.py | 55 +++++++++++++++++++++++++++++++ 2 files changed, 105 insertions(+) diff --git a/scripts/ci/pingora_edge_policy.py b/scripts/ci/pingora_edge_policy.py index 1eabbdd0f5..b3387dfb8e 100644 --- a/scripts/ci/pingora_edge_policy.py +++ b/scripts/ci/pingora_edge_policy.py @@ -303,6 +303,7 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: saw_trns = False saw_idat = False finished_idat = False + singleton_ancillary_chunks: set[bytes] = set() while offset + 12 <= len(raw): length = int.from_bytes(raw[offset : offset + 4], "big") end = offset + 12 + length @@ -381,6 +382,55 @@ def _is_recognized_documentation_image(path: str, raw: bytes) -> bool: if width is None or saw_trns or saw_idat or not valid_length: return False saw_trns = True + elif chunk_type in {b"cHRM", b"gAMA", b"iCCP", b"sBIT", b"sRGB"}: + if ( + width is None + or saw_plte + or saw_idat + or chunk_type in singleton_ancillary_chunks + or (chunk_type == b"iCCP" and b"sRGB" in singleton_ancillary_chunks) + or (chunk_type == b"sRGB" and b"iCCP" in singleton_ancillary_chunks) + ): + return False + valid_ancillary = False + if chunk_type == b"cHRM": + valid_ancillary = length == 32 + elif chunk_type == b"gAMA": + valid_ancillary = length == 4 and int.from_bytes(chunk_data, "big") != 0 + elif chunk_type == b"sRGB": + valid_ancillary = length == 1 and chunk_data[0] <= 3 + elif chunk_type == b"iCCP": + separator = chunk_data.find(b"\x00") + valid_ancillary = ( + 1 <= separator <= 79 + and separator + 2 < length + and chunk_data[separator + 1] == 0 + ) + if valid_ancillary: + try: + profile_decoder = zlib.decompressobj() + profile = profile_decoder.decompress(chunk_data[separator + 2 :], MAX_IMAGE_DECODED_BYTES + 1) + profile += profile_decoder.flush() + except zlib.error: + valid_ancillary = False + else: + valid_ancillary = ( + profile_decoder.eof + and not profile_decoder.unused_data + and not profile_decoder.unconsumed_tail + and 0 < len(profile) <= MAX_IMAGE_DECODED_BYTES + ) + elif chunk_type == b"sBIT": + expected_length = {0: 1, 2: 3, 3: 3, 4: 2, 6: 4}.get(color_type) + sample_limit = 8 if color_type == 3 else bit_depth + valid_ancillary = ( + length == expected_length + and sample_limit is not None + and all(0 < value <= sample_limit for value in chunk_data) + ) + if not valid_ancillary: + return False + singleton_ancillary_chunks.add(chunk_type) elif chunk_type == b"IDAT": if width is None or finished_idat or (color_type == 3 and not saw_plte): return False diff --git a/tests/test_pingora_edge_policy.py b/tests/test_pingora_edge_policy.py index 21be78926a..dae1fe30b5 100644 --- a/tests/test_pingora_edge_policy.py +++ b/tests/test_pingora_edge_policy.py @@ -331,6 +331,61 @@ def test_documentation_png_rejects_each_malformed_envelope_boundary() -> None: ) +@pytest.mark.parametrize( + ("chunk_type", "chunk_data"), + [ + (b"cHRM", b"\x00" * 32), + (b"gAMA", (45455).to_bytes(4, "big")), + (b"iCCP", b"profile\x00\x00" + policy.zlib.compress(b"icc-profile")), + (b"sBIT", b"\x08" * 4), + (b"sRGB", b"\x00"), + ], +) +def test_documentation_png_enforces_standard_ancillary_order_and_cardinality( + chunk_type: bytes, chunk_data: bytes +) -> None: + """Color-space and significant-bit chunks occur once before image data.""" + + raw = build_png(color_type=6, bit_depth=8, width=1, height=1) + before_idat = insert_png_chunk(raw, chunk_type, chunk_data) + assert policy._is_recognized_documentation_image( + "docs/acceptance.png", before_idat + ) + + iend_offset = raw.index(b"IEND") - 4 + after_idat = raw[:iend_offset] + png_chunk(chunk_type, chunk_data) + raw[iend_offset:] + duplicate = insert_png_chunk(before_idat, chunk_type, chunk_data) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", after_idat + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", duplicate + ) + + +def test_documentation_png_rejects_conflicting_or_malformed_color_chunks() -> None: + """Color-space chunks fail closed on conflicts and malformed payloads.""" + + raw = build_png(color_type=6, bit_depth=8, width=1, height=1) + srgb = insert_png_chunk(raw, b"sRGB", b"\x00") + conflicting = insert_png_chunk( + srgb, b"iCCP", b"profile\x00\x00" + policy.zlib.compress(b"icc-profile") + ) + for chunk_type, chunk_data in ( + (b"cHRM", b"\x00" * 31), + (b"gAMA", b"\x00" * 4), + (b"iCCP", b"profile\x00\x00not-zlib"), + (b"sBIT", b"\x09" * 4), + (b"sRGB", b"\x04"), + ): + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", insert_png_chunk(raw, chunk_type, chunk_data) + ) + assert not policy._is_recognized_documentation_image( + "docs/acceptance.png", conflicting + ) + + def test_documentation_png_always_requires_final_byte_validation() -> None: """Recognized documentation PNG paths remain final-content scan candidates."""