diff --git a/.github/actions/validate/action.yml b/.github/actions/validate/action.yml index 08682b5..b86406c 100644 --- a/.github/actions/validate/action.yml +++ b/.github/actions/validate/action.yml @@ -30,11 +30,6 @@ runs: shell: bash run: python3 checks/media_metadata_gate.py - # The carried tree is all well formed, so this feeds the same parsers seeded malformed variants. - - name: Fuzz media parsers step - shell: bash - run: python3 checks/media_parser_fuzz_gate.py - # A redacted image replaced by its original passes every other check, so this compares hashes. - name: Check media redactions step shell: bash diff --git a/.github/workflows/deploy-site.yml b/.github/workflows/deploy-site.yml index 0fc2c6c..2a88259 100644 --- a/.github/workflows/deploy-site.yml +++ b/.github/workflows/deploy-site.yml @@ -40,7 +40,7 @@ jobs: validate: name: Validate sources job needs: [ assert-ref ] - uses: ptr727/ProjectTemplate/.github/workflows/validate-task.yml@d7041d8ce15356dfd9223e54d74c64d464f28335 # 2.0.1192 + uses: ptr727/ProjectTemplate/.github/workflows/validate-task.yml@1faf16c708bbe5fe6c4e053de0e327dd07b05b8c # 2.0.1209 permissions: contents: read with: @@ -58,7 +58,7 @@ jobs: deploy: name: Deploy site job needs: [ validate ] - uses: ptr727/ProjectTemplate/.github/workflows/deploy-site-task.yml@d7041d8ce15356dfd9223e54d74c64d464f28335 # 2.0.1192 + uses: ptr727/ProjectTemplate/.github/workflows/deploy-site-task.yml@1faf16c708bbe5fe6c4e053de0e327dd07b05b8c # 2.0.1209 with: environment: ${{ inputs.environment }} permissions: diff --git a/.github/workflows/merge-bot-pull-request.yml b/.github/workflows/merge-bot-pull-request.yml index ee60f4e..5319e41 100644 --- a/.github/workflows/merge-bot-pull-request.yml +++ b/.github/workflows/merge-bot-pull-request.yml @@ -18,7 +18,7 @@ jobs: merge-bot: name: Merge bot pull request job - uses: ptr727/ProjectTemplate/.github/workflows/merge-bot-task.yml@d7041d8ce15356dfd9223e54d74c64d464f28335 # 2.0.1192 + uses: ptr727/ProjectTemplate/.github/workflows/merge-bot-task.yml@1faf16c708bbe5fe6c4e053de0e327dd07b05b8c # 2.0.1209 secrets: CODEGEN_APP_CLIENT_ID: ${{ secrets.CODEGEN_APP_CLIENT_ID }} CODEGEN_APP_PRIVATE_KEY: ${{ secrets.CODEGEN_APP_PRIVATE_KEY }} diff --git a/.github/workflows/publish-release.yml b/.github/workflows/publish-release.yml index 6ed06e1..c4089fe 100644 --- a/.github/workflows/publish-release.yml +++ b/.github/workflows/publish-release.yml @@ -16,7 +16,7 @@ jobs: # Also replaces this repo's own dispatch-ref assertion: the hub task errors identically on a ref other than main/develop. plan: name: Plan release job - uses: ptr727/ProjectTemplate/.github/workflows/publish-plan-task.yml@d7041d8ce15356dfd9223e54d74c64d464f28335 # 2.0.1192 + uses: ptr727/ProjectTemplate/.github/workflows/publish-plan-task.yml@1faf16c708bbe5fe6c4e053de0e327dd07b05b8c # 2.0.1209 with: event_name: ${{ github.event_name }} actor: ${{ github.actor }} @@ -30,7 +30,7 @@ jobs: name: Validate sources job needs: [plan] if: ${{ needs.plan.outputs.publish == 'true' }} - uses: ptr727/ProjectTemplate/.github/workflows/validate-task.yml@d7041d8ce15356dfd9223e54d74c64d464f28335 # 2.0.1192 + uses: ptr727/ProjectTemplate/.github/workflows/validate-task.yml@1faf16c708bbe5fe6c4e053de0e327dd07b05b8c # 2.0.1209 permissions: contents: read with: @@ -54,7 +54,7 @@ jobs: # An if: on a job replaces the implicit success() over its needs. # Without the validate arm a failed gate would publish anyway. if: ${{ needs.plan.outputs.publish == 'true' && needs.validate.result == 'success' }} - uses: ptr727/ProjectTemplate/.github/workflows/build-release-task.yml@d7041d8ce15356dfd9223e54d74c64d464f28335 # 2.0.1192 + uses: ptr727/ProjectTemplate/.github/workflows/build-release-task.yml@1faf16c708bbe5fe6c4e053de0e327dd07b05b8c # 2.0.1209 permissions: contents: write actions: write diff --git a/.github/workflows/test-pull-request.yml b/.github/workflows/test-pull-request.yml index b1ee501..89f7f0f 100644 --- a/.github/workflows/test-pull-request.yml +++ b/.github/workflows/test-pull-request.yml @@ -19,7 +19,7 @@ jobs: validate: name: Validate sources job - uses: ptr727/ProjectTemplate/.github/workflows/validate-task.yml@d7041d8ce15356dfd9223e54d74c64d464f28335 # 2.0.1192 + uses: ptr727/ProjectTemplate/.github/workflows/validate-task.yml@1faf16c708bbe5fe6c4e053de0e327dd07b05b8c # 2.0.1209 permissions: contents: read with: diff --git a/README.md b/README.md index 85d7a20..c7d49d3 100644 --- a/README.md +++ b/README.md @@ -142,7 +142,7 @@ A post is public the moment it deploys, and a photo or a pasted configuration ca - **Redactions.** A house number, a face, or a serial number is covered with a flat opaque fill, never a blur. A fill on an image already published is declared in [`checks/media-redactions.json`][media-redactions] and applied by a script. [`checks/media_redactions_gate.py`][media-redactions-gate] fails when a redacted file no longer matches its recorded hash, which is how a quietly restored original is caught. - **Text.** [`checks/text_pii_gate.py`][text-pii-gate] reads every page and post under `content/` outside the imported archive years, front matter and code blocks included. It reports email addresses, hardware addresses, public IP addresses, coordinates, street addresses, and phone numbers. It also reports a URL that carries coordinates or names one specific sensor or station. A finding names the file, the line, and the class, never the value, so the CI log does not republish it. A value a post prints on purpose is declared in an allow file with a reason, and an entry that stops matching fails the gate. -A fuzzer feeds the metadata parsers malformed images in the same action, because the carried tree is all well formed and would never exercise a refusal. What no gate can judge, whether a picture locates the home or identifies a person, is the checklist in [CONTENT.md][content]. +What no gate can judge, whether a picture locates the home or identifies a person, is the checklist in [CONTENT.md][content]. ### What Happens While the Site Runs diff --git a/checks/README.md b/checks/README.md index 8b5b3f0..d10e20c 100644 --- a/checks/README.md +++ b/checks/README.md @@ -13,7 +13,6 @@ This README covers the URL contract alone. The other checks in this directory ea | [`media_metadata_gate.py`][media-metadata-gate] | carried media is not in a form the gate can vouch for | | [`media_redactions_gate.py`][media-redactions-gate] | a file declared in `media-redactions.json` is not its redacted result | | [`text_pii_gate.py`][text-pii-gate] | a Markdown file under `content/`, outside the imported archive years, carries personal data in its text | -| [`media_parser_fuzz_gate.py`][media-parser-fuzz-gate] | malformed input breaks the media gate or its normalizer | ## Why this exists @@ -200,4 +199,3 @@ Three traps in that adjudication, each of which produced a wrong answer first an [url-parity-gate]: ./url_parity_gate.py [deploy-readme-redirects]: ../deploy/README.md#how-the-redirects-are-expressed [deploy-readme-security-headers]: ../deploy/README.md#security-headers -[media-parser-fuzz-gate]: ./media_parser_fuzz_gate.py diff --git a/checks/media_parser_fuzz_gate.py b/checks/media_parser_fuzz_gate.py deleted file mode 100755 index 7bceba9..0000000 --- a/checks/media_parser_fuzz_gate.py +++ /dev/null @@ -1,1603 +0,0 @@ -#!/usr/bin/env python3 -"""Feed the media gate and its normalizer malformed input, and fail on what breaks them. - -The gate and the normalizer are only checked by the carried tree, and every file in it is -well formed. Review kept finding malformed-input cases one at a time, each one a case -somebody thought of. This generates them instead: truncated, extended, bit-flipped and -spliced variants of every container the gate recognizes, plus planted metadata that the -gate must notice wherever a decoder would read it. - -The properties, each checked on every variant: - -1. The scanner never raises. A malformed file is a finding, never a traceback. -2. The normalizer never raises, and never returns a file its own scanner rejects. -3. The normalizer returns nothing, or a file whose picture payload is unchanged. -4. Planted metadata is always reported. - A plant that declares a refusal is refused, and one that declares a rewrite is rewritten. -5. The archive walk never raises on a damaged archive. -6. The normalizer refuses a picture its Exif turns, or returns one turned the same way. - Where decoders disagree on the turn, it refuses. -7. A clean plant, holding only values the gate admits, is reported by nothing and left unchanged. -8. The icon scanner never raises, and answers the same when a site file or a zip member reaches it. - -Seeds are constructed fixtures plus a sample of the carried media and site media, read in -memory and never written anywhere. The run is deterministic for a given seed and file list, and -bounded by both a case count and a time budget, so it can run in CI. -""" - -import argparse -import contextlib -import functools -import io -import itertools -import pathlib -import random -import re -import struct -import sys -import tempfile -import time -import warnings -import zipfile -import zlib -from collections.abc import Callable, Iterable, Iterator -from typing import Any - -import media_metadata_gate as gate - -REPO = pathlib.Path(__file__).resolve().parent.parent -sys.path.insert(0, str(REPO / "scripts")) -import normalize_media as normalizer - -# A comment segment, chunk, block or atom holding nothing, so no allowlist can admit it. -PLANT_TEXT = b"planted" - -# What a plant may declare the normalizer does with it, or the exact bytes of its rewrite. -# A plant that declares nothing passes whether the normalizer refuses or rewrites it. -REFUSED = "refused" -REWRITTEN = "rewritten" - - -def jpeg_segment(marker: int, body: bytes) -> bytes: - return bytes((0xFF, marker)) + struct.pack(">H", len(body) + 2) + body - - -def exif_segment(tail: bytes = b"") -> bytes: - """An Exif APP1 carrying only Orientation, with optional bytes after its one IFD.""" - entry = struct.pack(" bytes: - """An Exif APP1 whose Exif IFD sits past bytes nothing references, as a thumbnail would.""" - gap = b"\xff\xd8" + PLANT_TEXT + b"\xff\xd9" - ifd0 = struct.pack(" bytes: - """An Exif APP1 with one IFD holding the given tag, type, count and value entries. - - A pad goes between the IFD and the values it points to, as a writer aligning them leaves one. - """ - table, values = b"", b"" - start = 8 + 2 + 12 * len(entries) + 4 + len(pad) - for tag, kind, count, value in entries: - if len(value) > 4: - field = struct.pack(" bytes: - """A fixed-shape segment's fields with the bytes at one offset replaced.""" - return fields[:at] + value + fields[at + len(value) :] - - -# A restart marker, which a fill run inside a scan precedes. -RESTART = re.compile(rb"\xff[\xd0-\xd7]") - -# The values sBIT and bKGD hold for each PNG color type, stated here rather than taken from the gate. -PNG_SIGNIFICANT = {0: 1, 2: 3, 3: 3, 4: 2, 6: 4} -PNG_BACKGROUND = {0: 2, 2: 6, 3: 1, 4: 2, 6: 6} -SRGB_GAMMA = struct.pack(">I", 45455) -SQUARE = struct.pack(">IIB", 3779, 3779, 1) - - -def png_field_plants(data: bytes, parts: list) -> list[tuple[bytes, str, str | bytes]]: - """Ancillary PNG chunks each admitted alone, planted out of their values or more than once. - - A plant goes into a seed with no ancillary chunks, so that repetition cannot report it. - """ - if len(data) < 33: - return [] - bare = data[:8] + b"".join( - data[s:e] for c, s, e in parts if c in (b"IHDR", b"PLTE", b"IDAT", b"IEND") - ) - depth, color = data[24], data[25] - significant = PNG_SIGNIFICANT.get(color, 1) - background = PNG_BACKGROUND.get(color, 2) - palette = sum(e - s - 12 for c, s, e in parts if c == b"PLTE") // 3 - alpha = {0: b"pla", 2: PLANT_TEXT, 3: bytes(palette + 1)}.get(color, b"pl") - # A palette no browser draws by goes, and one in a grayscale image is refused. - suggestion = REWRITTEN if color in (2, 6) else REFUSED - planted = [ - (b"gAMA", b"plnt", "gAMA not the sRGB gamma", REFUSED), - (b"gAMA", SRGB_GAMMA + PLANT_TEXT, "gAMA with bytes past its fields", REFUSED), - (b"cHRM", (PLANT_TEXT * 5)[:32], "cHRM not the sRGB primaries", REFUSED), - (b"sRGB", b"p", "sRGB intent not a known intent", REFUSED), - (b"pHYs", SQUARE[:4] + b"plnt\x01", "pHYs density not square", REFUSED), - (b"pHYs", SQUARE[:8] + b"p", "pHYs unit not a known unit", REFUSED), - (b"pHYs", SQUARE + PLANT_TEXT, "pHYs with bytes past its fields", REFUSED), - (b"sBIT", b"p" * significant, "sBIT not the full depth", REWRITTEN), - (b"bKGD", b"p" * background, "bKGD not zero", REWRITTEN), - (b"hIST", b"pl", "hIST chunk", REWRITTEN), - (b"tRNS", alpha, "tRNS not in its shape", REFUSED), - ] - if color == 0 and depth < 16: - planted.append((b"tRNS", b"\xff\xff", "tRNS gray past its depth", REFUSED)) - out: list[tuple[bytes, str, str | bytes]] = [ - (bare[:33] + png_chunk(chunk, body) + bare[33:], what, expect) - for chunk, body, what, expect in planted - ] - full = bytes((8 if color == 3 else depth,)) * significant - for chunk, body, expect in ( - (b"gAMA", SRGB_GAMMA, REFUSED), - (b"sRGB", b"\x00", REFUSED), - (b"pHYs", SQUARE, REFUSED), - (b"sBIT", full, REWRITTEN), - (b"bKGD", bytes(background), REWRITTEN), - ): - twice = png_chunk(chunk, body) * 2 - variant = bare[:33] + twice + bare[33:] - out.append((variant, f"repeated {chunk.decode()}", expect)) - # A decoder passes over a chunk out of its place, so its bytes are free there. - late = png_chunk(b"gAMA", SRGB_GAMMA) - out.append((bare[:-12] + late + bare[-12:], "gAMA after IDAT", REFUSED)) - if palette: - early = png_chunk(b"tRNS", bytes(1)) - out.append((bare[:33] + early + bare[33:], "tRNS before PLTE", REFUSED)) - # A palette entry no pixel can address, and a palette no browser draws by, each hold free bytes. - if color == 3 and depth <= 8: - entries = (1 << depth) + 1 - long = png_chunk(b"PLTE", (PLANT_TEXT * entries)[: 3 * entries]) - head = bare[:8] + b"".join( - long if c == b"PLTE" else data[s:e] - for c, s, e in parts - if c in (b"IHDR", b"PLTE", b"IDAT", b"IEND") - ) - out.append((head, "PLTE longer than the bit depth addresses", REFUSED)) - elif color != 3: - suggested = png_chunk(b"PLTE", PLANT_TEXT * 3) - variant = bare[:33] + suggested + bare[33:] - out.append((variant, "PLTE in a color type that draws none", suggestion)) - # A second palette fails the picture in libpng, so dropping both would draw one it did not. - out.append((bare[:33] + suggested * 2 + bare[33:], "two PLTE", REFUSED)) - if color == 2: - # A truecolor color key may precede a suggested palette, so only the palette goes. - key = png_chunk(b"tRNS", bytes(6)) - variant = bare[:33] + key + suggested + bare[33:] - out.append( - (variant, "tRNS before a suggested PLTE", bare[:33] + key + bare[33:]) - ) - # A palette after the picture data is one a decoder does not read. - palettes = [data[s:e] for c, s, e in parts if c == b"PLTE"][:1] or [ - png_chunk(b"PLTE", PLANT_TEXT * 3) - ] - moved = bare[:8] + b"".join( - data[s:e] for c, s, e in parts if c in (b"IHDR", b"IDAT") - ) - after = REFUSED if color == 3 else suggestion - out.append((moved + palettes[0] + bare[-12:], "PLTE after IDAT", after)) - # An APNG frame is refused, so a still decoder never reads a default image the gate alone passed. - control = png_chunk(b"acTL", struct.pack(">II", 1, 0)) - frame = png_chunk(b"fdAT", struct.pack(">I", 0) + PLANT_TEXT) - out.append((bare[:33] + control + bare[33:], "acTL chunk", REFUSED)) - out.append((bare[:-12] + frame + bare[-12:], "fdAT with no acTL", REFUSED)) - return out - - -def png_clean_plants(data: bytes, parts: list) -> list[tuple[bytes, str]]: - """Ancillary PNG chunks each holding a value writers use, which the gate passes and the normalizer keeps.""" - if len(data) < 33: - return [] - bare = data[:8] + b"".join( - data[s:e] for c, s, e in parts if c in (b"IHDR", b"PLTE", b"IDAT", b"IEND") - ) - depth, color = data[24], data[25] - palette = sum(e - s - 12 for c, s, e in parts if c == b"PLTE") // 3 - full = bytes((8 if color == 3 else depth,)) * PNG_SIGNIFICANT.get(color, 1) - kept = [ - *((b"sRGB", bytes((intent,))) for intent in range(4)), - *((b"cHRM", primaries) for primaries in sorted(gate.PNG_PRIMARIES)), - (b"gAMA", SRGB_GAMMA), - (b"pHYs", SQUARE), - (b"pHYs", struct.pack(">IIB", 72, 72, 0)), - (b"sBIT", full), - (b"bKGD", bytes(PNG_BACKGROUND.get(color, 2))), - ] - if color in (0, 2): - kept.append((b"tRNS", bytes(2 if color == 0 else 6))) - elif color == 3 and palette: - kept.append((b"tRNS", bytes(palette))) - at = 33 + sum(e - s for c, s, e in parts if c == b"PLTE") if color == 3 else 33 - out = [] - for chunk, body in kept: - # Transparency and background follow a palette, and every other chunk goes first. - place = at if chunk in (b"tRNS", b"bKGD") else 33 - variant = bare[:place] + png_chunk(chunk, body) + bare[place:] - out.append((variant, f"{chunk.decode()} {body.hex()}")) - # Rows of filter type zero and sample zero, laid out in the Adam7 passes an interlaced header needs. - interlaced = data[16:28] + b"\x01" - size = gate.png_picture_size(interlaced) - piece = gate.PNG_INFLATE_PIECE - rows = deflated(bytes(min(piece, size - at)) for at in range(0, size, piece)) - idat = [(s, e) for c, s, e in parts if c == b"IDAT"] - if idat: - head = data[:8] + png_chunk(b"IHDR", interlaced) + data[33 : idat[0][0]] - variant = head + png_chunk(b"IDAT", rows) + data[idat[-1][1] :] - out.append((variant, "interlaced IHDR over its Adam7 passes")) - return out - - -def png_structure_plants(data: bytes, parts: list) -> list[tuple[bytes, str, str]]: - """PNGs missing or bending a critical chunk a decoder needs, so no browser draws them and every byte is free.""" - if len(data) < 33: - return [] - signature, ihdr, iend = data[:8], data[8:33], data[-12:] - header = ihdr[8:21] - # The seed's own picture data, so each plant bends only the rule it names. - idat = b"".join(data[s:e] for c, s, e in parts if c == b"IDAT") - plte = [data[s:e] for c, s, e in parts if c == b"PLTE"] - body = b"".join(data[s:e] for c, s, e in parts if c in (b"PLTE", b"IDAT")) - ended = png_chunk(b"IEND", PLANT_TEXT) - padded = png_chunk(b"IHDR", header + PLANT_TEXT) - out = [ - (signature + idat + iend, "no IHDR", REFUSED), - (signature + idat + ihdr + iend, "IHDR not first", REFUSED), - (signature + ihdr + b"".join(plte) + iend, "no IDAT", REFUSED), - (signature + ihdr + body + ended, "IEND with a body", REWRITTEN), - (signature + padded + body + iend, "IHDR with bytes past its fields", REFUSED), - ] - for at, value, what in ( - (0, bytes(4), "IHDR width zero"), - (4, b"\x80\x00\x00\x00", "IHDR height past the range"), - ( - 0, - struct.pack(">I", gate.PNG_SIDE_LIMIT + 1), - "IHDR width past libpng's limit", - ), - ( - 4, - struct.pack(">I", gate.PNG_SIDE_LIMIT + 1), - "IHDR height past libpng's limit", - ), - (8, b"\x03", "IHDR depth not one its color type pairs with"), - (9, b"\x05", "IHDR color type not a defined one"), - (10, b"\x01", "IHDR compression not zero"), - (11, b"\x01", "IHDR filter not zero"), - (12, b"\x02", "IHDR interlace not a defined one"), - ): - bent = header[:at] + value + header[at + len(value) :] - out.append((signature + png_chunk(b"IHDR", bent) + body + iend, what, REFUSED)) - deep = header[:8] + b"\x10\x03" + header[10:] - wide = png_chunk(b"PLTE", (PLANT_TEXT * 3)[:9] * 3) - out.append( - ( - signature + png_chunk(b"IHDR", deep) + wide + idat + iend, - "palette with a 16-bit depth", - REFUSED, - ) - ) - if header[9] == 3: - variant = signature + ihdr + idat + iend - out.append((variant, "palette image with no PLTE", REFUSED)) - for at, name, expect in ((8, "IHDR", REFUSED), (len(data) - 12, "IEND", REWRITTEN)): - bent = data[: at + 8 + (13 if name == "IHDR" else 0)] + b"plnt" - out.append((bent + data[len(bent) :], f"{name} CRC not its own", expect)) - gamma = png_chunk(b"gAMA", SRGB_GAMMA)[:-4] + b"plnt" - out.append((data[:33] + gamma + data[33:], "gAMA CRC not its own", REWRITTEN)) - return out - - -def inflated(stream: bytes) -> Iterator[bytes]: - """A seed's picture data inflated in bounded pieces, so a plant built from it never holds the whole picture.""" - inflate = zlib.decompressobj() - while piece := inflate.decompress(stream, gate.PNG_INFLATE_PIECE): - yield piece - stream = inflate.unconsumed_tail - - -def all_but_last(pieces: Iterator[bytes]) -> Iterator[bytes]: - """The pieces with their final byte left off.""" - held = b"" - for piece in pieces: - yield held - held = piece - yield held[:-1] - - -def deflated(pieces: Iterable[bytes]) -> bytes: - deflate = zlib.compressobj() - return b"".join(deflate.compress(piece) for piece in pieces) + deflate.flush() - - -def png_stream_plants(data: bytes, parts: list) -> list[tuple[bytes, str, str]]: - """PNGs whose picture data is not one whole zlib stream of the length IHDR needs, so a decoder fails the picture or passes over bytes.""" - idat = [(s, e) for c, s, e in parts if c == b"IDAT"] - if len(data) < 33 or not idat: - return [] - head, tail = data[: idat[0][0]], data[idat[-1][1] :] - stream = b"".join(data[s + 8 : e - 4] for s, e in idat) - past = deflated(itertools.chain(inflated(stream), (PLANT_TEXT,))) - short = deflated(all_but_last(inflated(stream))) - pieces = inflated(stream) - first = next(pieces) - bent = deflated(itertools.chain((b"\x05" + first[1:],), pieces)) - out = [ - (png_chunk(b"IDAT", b""), "IDAT empty"), - (png_chunk(b"IDAT", PLANT_TEXT), "IDAT not a zlib stream"), - (png_chunk(b"IDAT", stream + PLANT_TEXT), "bytes after the IDAT stream"), - ( - png_chunk(b"IDAT", stream) + png_chunk(b"IDAT", PLANT_TEXT), - "IDAT after the stream's end", - ), - (png_chunk(b"IDAT", stream[:-4]), "IDAT stream cut off"), - (png_chunk(b"IDAT", past), "IDAT past its rows"), - (png_chunk(b"IDAT", short), "IDAT short of its rows"), - (png_chunk(b"IDAT", bent), "IDAT row filter 5"), - ] - variants = [(head + body + tail, what, REFUSED) for body, what in out] - header = data[16:29] - side = struct.pack(">I", gate.PNG_SIDE_LIMIT // 10) - vast = data[:8] + png_chunk(b"IHDR", side * 2 + header[8:]) + data[33:] - variants.append((vast, "IHDR picture past the size limit", REFUSED)) - interlaced = header[:12] + b"\x01" - if gate.png_picture_size(interlaced) != gate.png_picture_size(header): - whole = data[:8] + png_chunk(b"IHDR", interlaced) + data[33:] - variants.append((whole, "interlaced IHDR over rows laid out whole", REFUSED)) - return variants - - -def gif_field_plants(data: bytes, parts: list) -> list[tuple[bytes, str, str | bytes]]: - """GIF blocks each admitted alone, planted with a reserved or unused field not zero. - - A field the normalizer writes as zero comes back as the seed, which a clean seed holds zero there. - """ - if len(data) < 13: - return [] - out: list[tuple[bytes, str, str | bytes]] = [ - (data[:12] + b"p" + data[13:], "aspect ratio byte not zero", data), - (data[:11] + b"p" + data[12:], "background index not zero", data), - ( - data[:10] + bytes((data[10] | 0x70,)) + data[11:], - "color resolution not zero", - data, - ), - ( - data[:10] + bytes((data[10] | 0x08,)) + data[11:], - "screen sort flag not zero", - data, - ), - ] - if data[10] & 0x80: - table = 3 * (2 << (data[10] & 7)) - flags = b"\x07" - variant = data[:10] + flags + data[11:13] + data[13 + table :] - out.append((variant, "table size with no color table", REWRITTEN)) - # A local table on every image leaves the global one read by nothing. - local = bytearray(data) - for name, start, _ in reversed(parts): - if name == "image" and not data[start + 9] & 0x80: - local[start + 9] = data[start + 9] & 0x40 | 0x80 - local[start + 10 : start + 10] = bytes(6) - out.append((bytes(local), "global color table no image reads", REWRITTEN)) - out.append((b"GIF87a" + data[6:], "version not 89a", data)) - for name, start, end in parts: - if name == "extension 0xF9" and end - start == 8: - packed, index = data[start + 3], data[start + 6] - # Decoders differ on a disposal method the format does not define, so it is not rewritten. - for value, held, what, expect in ( - (packed | 0xE0, index, "graphic control reserved bits not zero", data), - ( - packed | 0x1C, - index, - "graphic control disposal not a known method", - REFUSED, - ), - ( - packed & 0xFE, - 0x70, - "graphic control unused transparent index", - REWRITTEN, - ), - ): - fields = bytes((value,)) + data[start + 4 : start + 6] + bytes((held,)) - variant = data[: start + 3] + fields + data[start + 7 :] - out.append((variant, what, expect)) - elif name == "image": - held = data[start + 9] - for value, what in ( - (held | 0x18, "image descriptor reserved bits not zero"), - (held | 0x20, "image descriptor sort flag not zero"), - ): - variant = data[: start + 9] + bytes((value,)) + data[start + 10 :] - out.append((variant, what, data)) - if not held & 0x80: - variant = data[: start + 9] + bytes((held | 0x07,)) + data[start + 10 :] - out.append((variant, "image table size with no local table", data)) - return out - - -ICC = b"ICC_PROFILE\x00\x01\x01" -SOF = jpeg_segment(0xC0, b"\x08\x00\x08\x00\x08\x01\x01\x11\x00") -SOS = jpeg_segment(0xDA, b"\x01\x01\x00\x00\x3f\x00") - -# APP segments each already allowed alone, planted so that their shape or repetition carries bytes. -JPEG_APP_BENT = ( - ( - [jpeg_segment(0xE0, JFIF + PLANT_TEXT)], - "JFIF with bytes past its fields", - REWRITTEN, - ), - ( - [jpeg_segment(0xE0, JFIF[:12] + b"\x03\x01" + PLANT_TEXT + b"\x00\x00")], - "JFIF thumbnail", - REWRITTEN, - ), - ( - [jpeg_segment(0xEE, ADOBE + PLANT_TEXT)], - "Adobe with bytes past its fields", - REWRITTEN, - ), - ([jpeg_segment(0xE0, JFIF)] * 2, "repeated JFIF", REWRITTEN), - ([jpeg_segment(0xEE, ADOBE)] * 2, "repeated Adobe", REFUSED), - ([exif_segment()] * 8, "repeated Exif", REFUSED), - ( - [jpeg_segment(0xE2, ICC + bytes(8)), jpeg_segment(0xE2, ICC + PLANT_TEXT)], - "repeated ICC chunk", - REFUSED, - ), - ( - [jpeg_segment(0xE2, ICC[:12] + b"\x01\x03" + PLANT_TEXT)], - "incomplete ICC profile", - REFUSED, - ), - ( - [jpeg_segment(0xE2, ICC + PLANT_TEXT)], - "ICC profile not a known profile", - REFUSED, - ), - ( - [jpeg_segment(0xE0, bend(JFIF, 5, b"pl"))], - "JFIF version not a known version", - REWRITTEN, - ), - ( - [jpeg_segment(0xE0, bend(JFIF, 7, b"p"))], - "JFIF units not a known unit", - REWRITTEN, - ), - ( - [jpeg_segment(0xE0, bend(JFIF, 8, b"plan"))], - "JFIF density not square", - REWRITTEN, - ), - ([jpeg_segment(0xEE, bend(ADOBE, 5, b"pl"))], "Adobe version not 100", REWRITTEN), - ([jpeg_segment(0xEE, bend(ADOBE, 7, b"plan"))], "Adobe flags not zero", REWRITTEN), - ( - [jpeg_segment(0xEE, bend(ADOBE, 11, b"p"))], - "Adobe transform not a known one", - REFUSED, - ), -) - - -def with_riff_size(data: bytes) -> bytes: - """Restate a WebP's RIFF size to cover the whole file, as a decoder then reads it all.""" - return data[:4] + struct.pack(" bytes: - """A structurally complete JPEG, with stuffed bytes and a restart marker in its scans.""" - sof = 0xC2 if progressive else 0xC0 - out = b"\xff\xd8" + jpeg_segment( - 0xE0, b"JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00" - ) - out += exif_segment() - out += jpeg_segment(0xDB, b"\x00" + bytes(range(64))) - out += jpeg_segment(sof, b"\x08\x00\x08\x00\x08\x01\x01\x11\x00") - out += jpeg_segment(0xC4, b"\x00" + bytes(16) + b"") - out += jpeg_segment(0xDA, b"\x01\x01\x00\x00\x3f\x00") - out += b"\x12\x34\xff\x00\x56\xff\xd0\x78\x9a" - if progressive: - out += jpeg_segment(0xC4, b"\x10" + bytes(16)) - out += jpeg_segment(0xDA, b"\x01\x01\x00\x01\x3f\x00") - out += b"\xab\xff\x00\xcd\xff\xd1\xef" - return out + b"\xff\xd9" - - -def dated_jpeg_fixture() -> bytes: - """A JPEG whose Exif holds every out-of-line allowed tag in its one shape.""" - rational = struct.pack(" bytes: - crc = zlib.crc32(name + body) & 0xFFFFFFFF - return struct.pack(">I", len(body)) + name + body + struct.pack(">I", crc) - - -def png_fixture() -> bytes: - ihdr = struct.pack(">IIBBBBB", 1, 1, 8, 0, 0, 0, 0) - idat = zlib.compress(b"\x00\x7f") - return ( - b"\x89PNG\r\n\x1a\n" - + png_chunk(b"IHDR", ihdr) - + png_chunk(b"gAMA", struct.pack(">I", 45455)) - + png_chunk(b"IDAT", idat) - + png_chunk(b"IEND", b"") - ) - - -def truecolor_png_fixture() -> bytes: - ihdr = struct.pack(">IIBBBBB", 1, 1, 8, 2, 0, 0, 0) - idat = zlib.compress(bytes(4)) - return ( - b"\x89PNG\r\n\x1a\n" - + png_chunk(b"IHDR", ihdr) - + png_chunk(b"IDAT", idat) - + png_chunk(b"IEND", b"") - ) - - -def small_png_fixture() -> bytes: - """A 5x3 grayscale PNG at two bits a sample, so rows hold a partial byte and interlacing spans several passes.""" - ihdr = struct.pack(">IIBBBBB", 5, 3, 2, 0, 0, 0, 0) - idat = zlib.compress(b"\x00\x1b\xc0" * 3) - return ( - b"\x89PNG\r\n\x1a\n" - + png_chunk(b"IHDR", ihdr) - + png_chunk(b"IDAT", idat[:5]) - + png_chunk(b"IDAT", idat[5:]) - + png_chunk(b"IEND", b"") - ) - - -def palette_png_fixture() -> bytes: - ihdr = struct.pack(">IIBBBBB", 1, 1, 1, 3, 0, 0, 0) - idat = zlib.compress(b"\x00\x00") - return ( - b"\x89PNG\r\n\x1a\n" - + png_chunk(b"IHDR", ihdr) - + png_chunk(b"PLTE", bytes(6)) - + png_chunk(b"IDAT", idat) - + png_chunk(b"IEND", b"") - ) - - -def gif_fixture() -> bytes: - head = b"GIF89a" + struct.pack(" bytes: - return ( - name + struct.pack(" bytes: - body = riff_chunk(b"VP8X", bytes(10)) + riff_chunk(b"VP8L", b"\x2f\x00\x00\x00\x00") - return b"RIFF" + struct.pack(" bytes: - """An animated WebP of one frame, which holds its own chunks inside an ANMF.""" - frame = bytes(12) + b"\x64\x00\x00\x00" - frame += riff_chunk(b"VP8L", b"\x2f\x00\x00\x00\x00") - body = riff_chunk(b"VP8X", b"\x02" + bytes(9)) + riff_chunk(b"ANIM", bytes(6)) - body += riff_chunk(b"ANMF", frame) - return b"RIFF" + struct.pack(" bytes: - return struct.pack(">I", len(body) + 8) + name + body - - -def iso_fixture() -> bytes: - moov = atom( - b"moov", atom(b"mvhd", bytes(100)) + atom(b"trak", atom(b"tkhd", bytes(84))) - ) - return atom(b"ftyp", b"isom\x00\x00\x02\x00isom") + moov + atom(b"mdat", bytes(32)) - - -def ico_fixture() -> bytes: - """An icon of two PNG entries laid end to end, each directory entry its image's own.""" - images = ((png_fixture(), 8), (truecolor_png_fixture(), 24)) - offset = 6 + gate.ICO_ENTRY * len(images) - head = gate.ICO_HEADER + struct.pack(" bytes: - buffer = io.BytesIO() - with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive: - archive.writestr("picture.png", png_fixture()) - archive.writestr("picture.jpg", jpeg_fixture(False)) - archive.writestr("notes.txt", b"not media") - for method in (zipfile.ZIP_LZMA, zipfile.ZIP_BZIP2): - picture = jpeg_fixture(True) * 8 - archive.writestr(f"picture-{method}.jpg", picture, compress_type=method) - return buffer.getvalue() - - -def container(data: bytes) -> str: - if data[:2] == b"\xff\xd8": - return "jpeg" - if data[:8] == b"\x89PNG\r\n\x1a\n": - return "png" - if data[:6] in (b"GIF87a", b"GIF89a"): - return "gif" - if data[:4] == b"RIFF" and data[8:12] == b"WEBP": - return "webp" - if data[4:8] in (b"ftyp", b"moov", b"wide", b"mdat", b"free", b"skip"): - return "iso" - if data[:4] == gate.ICO_HEADER: - return "ico" - return "other" - - -# Each mutator returns the variant and a short description of what it did. -Mutator = Callable[[random.Random, bytes], tuple[bytes, str]] - - -def truncate(rng: random.Random, data: bytes) -> tuple[bytes, str]: - cut = rng.randrange(0, len(data)) - return data[:cut], f"truncate at {cut}" - - -def extend(rng: random.Random, data: bytes) -> tuple[bytes, str]: - tail = bytes(rng.randrange(256) for _ in range(rng.choice((1, 3, 7, 8, 64)))) - return data + tail, f"append {len(tail)} bytes" - - -def bit_flip(rng: random.Random, data: bytes) -> tuple[bytes, str]: - out = bytearray(data) - flips = rng.randrange(1, 9) - for _ in range(flips): - position = rng.randrange(len(out)) - out[position] ^= 1 << rng.randrange(8) - return bytes(out), f"{flips} bit flips" - - -def smash(rng: random.Random, data: bytes) -> tuple[bytes, str]: - """Overwrite a 2- or 4-byte field with a boundary value, where a length would sit.""" - width = rng.choice((2, 4)) - position = rng.randrange(max(1, len(data) - width)) - value = rng.choice((0, 1, 2, 7, 8, 0xFF, 0x7FFF, 0xFFFF, 0x7FFFFFFF, 0xFFFFFFFF)) - value &= (1 << (8 * width)) - 1 - fmt = rng.choice((">", "<")) + ("H" if width == 2 else "I") - field = struct.pack(fmt, value) - return data[:position] + field + data[ - position + width : - ], f"smash {width} at {position}" - - -def splice(rng: random.Random, data: bytes) -> tuple[bytes, str]: - position = rng.randrange(len(data) + 1) - piece = bytes(rng.randrange(256) for _ in range(rng.randrange(1, 17))) - return data[:position] + piece + data[ - position: - ], f"splice {len(piece)} at {position}" - - -def duplicate(rng: random.Random, data: bytes) -> tuple[bytes, str]: - start = rng.randrange(len(data)) - end = min(len(data), start + rng.randrange(1, 257)) - return data[:end] + data[start:end] + data[end:], f"duplicate {start}:{end}" - - -MUTATORS: tuple[Mutator, ...] = (truncate, extend, bit_flip, smash, splice, duplicate) - - -def at_boundaries( - data: bytes, parts: list, piece: bytes, what: str -) -> list[tuple[bytes, str]]: - """Insert a piece before every parsed element, and once after the last.""" - offsets = sorted({start for _, start, _ in parts} | {len(data)}) - return [(data[:o] + piece + data[o:], f"{what} at {o}") for o in offsets] - - -def jpeg_clean_plants(data: bytes, parts: list) -> list[tuple[bytes, str]]: - """JFIF and Adobe segments each holding values writers use, planted into a seed with no APP segments.""" - bare = data[:2] + b"".join(data[s:e] for m, s, e in parts if not 0xE0 <= m <= 0xEF) - kept = [ - *(jpeg_segment(0xE0, bend(JFIF, 5, bytes((1, minor)))) for minor in range(3)), - *( - jpeg_segment(0xE0, bend(JFIF, 7, struct.pack(">BHH", units, dots, dots))) - for units, dots in ((1, 72), (1, 96), (1, 300), (2, 28), (2, 118)) - ), - *(jpeg_segment(0xEE, ADOBE[:11] + bytes((move,))) for move in range(3)), - ] - return [(bare[:2] + app + bare[2:], f"APP {app[4:].hex()}") for app in kept] - - -def gif_clean_plants(data: bytes, parts: list) -> list[tuple[bytes, str]]: - """Looping and graphic control blocks each holding values writers use, before the first image of a seed with no extensions.""" - images = [s for k, s, _ in parts if k == "image"] - if not images: - return [] - bare = b"".join( - data[s:e] for k, s, e in parts if not str(k).startswith("extension") - ) - at = images[0] - sum( - e - s for k, s, e in parts if str(k).startswith("extension") and s < images[0] - ) - kept = [ - *( - b"\x21\xff\x0bNETSCAPE2.0\x03\x01" + struct.pack(" list[tuple[bytes, str]]: - """VP8X, ANIM and ANMF fields each holding values writers use, set in a seed that already carries the chunk.""" - out = [] - for name, start, _ in parts: - body = start + 8 - if name == b"VP8X": - flags = data[body] | 0x10 - variant = data[:body] + bytes((flags,)) + data[body + 1 :] - out.append((variant, "VP8X alpha flag")) - elif name == b"ANIM": - for loops in (1, 3, 65535): - variant = data[: body + 4] + struct.pack(" list[tuple[bytes, str]]: - """Variants holding only values the gate admits, which it has to pass and the normalizer keep.""" - if kind == "png": - return png_clean_plants(data, gate.png_parts(data)[0]) - if kind == "jpeg": - return jpeg_clean_plants(data, gate.jpeg_parts(data)[0]) - if kind == "gif": - return gif_clean_plants(data, gate.gif_parts(data)[0]) - if kind == "webp": - return webp_clean_plants(data, gate.webp_parts(data)[0]) - return [] - - -def plants(kind: str, data: bytes) -> list[tuple]: - """Variants holding metadata a decoder reads, which the scanner has to report. - - A seed's own elements are where a decoder looks for the next one, so a piece planted - before any of them, between two scans included, is one the decoder reads. - """ - out: list[tuple] = [] - expect: str | bytes - if kind == "jpeg": - parts, _ = gate.jpeg_parts(data) - comment = jpeg_segment(0xFE, PLANT_TEXT) - out += [ - (variant, what, data) - for variant, what in at_boundaries(data, parts, comment, "comment") - ] - # A shape plant goes into a seed with no APP segments, so that repetition cannot report it. - bare = data[:2] + b"".join( - data[s:e] for m, s, e in parts if not 0xE0 <= m <= 0xEF - ) - # The planted Exif holds no Orientation, so dropping it leaves nothing to re-emit. - gap = bare[:2] + exif_gap_segment() + bare[2:] - out.append((gap, "bytes between Exif IFDs", bare)) - for entries, what in EXIF_BENT: - bent = exif_ifd_segment(entries) - out.append((bare[:2] + bent + bare[2:], f"Exif {what}", REWRITTEN)) - for segments, what, expect in JPEG_APP_BENT: - out.append((bare[:2] + b"".join(segments) + bare[2:], what, expect)) - padded = exif_ifd_segment([ORIENTATION, (0x0132, 2, 20, DATE)], b"p") - out.append((bare[:2] + padded + bare[2:], "Exif pad byte not zero", REWRITTEN)) - tailed = exif_segment(b"p") - out.append( - (bare[:2] + tailed + bare[2:], "Exif trailing pad byte not zero", REWRITTEN) - ) - eoi = [s for m, s, _ in parts if m == 0xD9] - if eoi: - second = b"\xff\xd8" + SOF + SOS + PLANT_TEXT - variant = data[: eoi[-1]] + second + data[eoi[-1] :] - out.append((variant, "second stream", REFUSED)) - variant = data[: eoi[-1]] + SOF + data[eoi[-1] :] - out.append((variant, "second frame header", REFUSED)) - for marker, start, end in parts: - if marker != 0xE1 or data[start + 4 : start + 10] != b"Exif\x00\x00": - continue - length = struct.unpack_from(">H", data, start + 2)[0] - tail = b"\xff\xd8" + PLANT_TEXT + b"\xff\xd9" - grown = struct.pack(">H", length + len(tail)) - variant = data[: start + 2] + grown + data[start + 4 : end] + tail - out.append((variant + data[end:], "bytes past the Exif IFDs", REWRITTEN)) - # A run of fill bytes has a length nothing reads, before a segment or after a scan alike. - for at in (parts[1][1], *eoi) if len(parts) > 1 else (): - variant = data[:at] + b"\xff" * 5 + data[at:] - out.append((variant, "fill bytes before a marker", data)) - scans = [(s, e) for m, s, e in parts if m == 0xDA] - restart = next( - (hit.start() for s, e in scans for hit in RESTART.finditer(data, s, e)), - None, - ) - if restart is not None: - variant = data[:restart] + b"\xff\xff" + data[restart:] - out.append((variant, "fill bytes inside a scan", REFUSED)) - elif kind == "png": - parts, _ = gate.png_parts(data) - text = png_chunk(b"tEXt", b"Comment\x00" + PLANT_TEXT) - texts = at_boundaries(data, parts[1:], text, "tEXt") - # One between two IDAT chunks splits the picture data, which no decoder draws past. - split = { - f"tEXt at {start}" - for (c, _, _), (d, start, _) in itertools.pairwise(parts) - if c == d == b"IDAT" - } - out += [ - (variant, what, REFUSED if what in split else REWRITTEN) - for variant, what in texts - ] - profile = png_chunk(b"iCCP", b"icc\x00\x00" + zlib.compress(PLANT_TEXT)) - variant = data[:33] + profile + data[33:] - out.append((variant, "iCCP not a known profile", REFUSED)) - out += png_field_plants(data, parts) - out += png_structure_plants(data, parts) - out += png_stream_plants(data, parts) - elif kind == "gif": - parts, _ = gate.gif_parts(data) - comment = b"\x21\xfe" + bytes((len(PLANT_TEXT),)) + PLANT_TEXT + b"\x00" - out += [ - (variant, what, data) - for variant, what in at_boundaries(data, parts[1:], comment, "comment") - ] - block = bytes((len(PLANT_TEXT),)) + PLANT_TEXT - for name, _, end in parts: - if str(name).startswith("extension"): - variant = data[: end - 1] + block + data[end - 1 :] - # A graphic control block sets transparency and timing, so it is refused rather than dropped. - expect = REFUSED if name == "extension 0xF9" else REWRITTEN - out.append((variant, f"sub-block inside {name}", expect)) - out += gif_field_plants(data, parts) - elif kind == "webp": - parts, _ = gate.webp_parts(data) - chunk = riff_chunk(b"EXIF", PLANT_TEXT) - out += [ - (with_riff_size(variant), what, data) - for variant, what in at_boundaries(data, parts, chunk, "EXIF") - ] - out.append((data + PLANT_TEXT[:7], "bytes appended", data)) - frame = riff_chunk(b"VP8L", PLANT_TEXT) - out.append((data + frame, "frame past the RIFF size", data)) - lossless = riff_chunk(b"VP8L", b"\x2f" + PLANT_TEXT) - for name, start, end in parts: - payload = start + 8 + struct.unpack_from(" bytes: - """A TIFF header and one IFD holding Orientation `first` as the given type, or none where it is 0. - - A nonzero `then` adds a second Orientation SHORT holding that value, - in the Exif sub-IFD where `sub` is set and in the same IFD otherwise. - """ - fmt = "<" if order == b"II" else ">" - if kind == 3: - value = struct.pack(fmt + "HH", first, 0) - else: - value = struct.pack(fmt + "I", first) - entries = [struct.pack(fmt + "HHI", 0x0112, kind, 1) + value] if first else [] - second = struct.pack(fmt + "HHIHH", 0x0112, 3, 1, then, 0) if then else b"" - tail = b"" - if sub: - at = 8 + 2 + 12 * (len(entries) + 1) + 4 - entries.append(struct.pack(fmt + "HHII", 0x8769, 4, 1, at)) - tail = struct.pack(fmt + "H", len(second) // 12) + second - tail += struct.pack(fmt + "I", 0) - elif second: - entries.append(second) - head = order + struct.pack(fmt + "HIH", 42, 8, len(entries)) - return head + b"".join(entries) + struct.pack(fmt + "I", 0) + tail - - -# Each pair disagrees, and each order changed what a decoder taking the last entry reads. -DISPUTED = ((6, 8), (1, 6)) -# Where the second value sits in the Exif sub-IFD, each order changed what a flattening decoder reads. -PLACES = ((False, "then"), (True, "then in the sub-IFD")) - - -def turned(kind: str, data: bytes) -> list[tuple[bytes, str, int | None]]: - """Variants whose Exif turns the picture a quarter turn, as Orientation 6 does. - - Each carries the Orientation the normalized file must hold, or None for a turn decoders - disagree on, which only a refusal keeps. - """ - out: list[tuple[bytes, str, int | None]] = [] - if kind == "jpeg": - parts, _ = gate.jpeg_parts(data) - bare = data[:2] + b"".join( - data[s:e] for m, s, e in parts if not 0xE0 <= m <= 0xEF - ) - for order, size, what in ( - (b"MM", 4, "Orientation as a big-endian LONG"), - (b"II", 4, "Orientation as a little-endian LONG"), - (b"MM", 3, "Orientation as a big-endian SHORT"), - ): - segment = jpeg_segment( - 0xE1, b"Exif\x00\x00" + orientation_tiff(order, size) - ) - out.append((bare[:2] + segment + bare[2:], what, None if size == 4 else 6)) - for (first, then), (sub, place) in itertools.product( - (*DISPUTED, (6, 6)), PLACES - ): - tiff = orientation_tiff(b"II", 3, first, then, sub) - segment = jpeg_segment(0xE1, b"Exif\x00\x00" + tiff) - what = f"Orientation {first} {place} {then}" - out.append( - (bare[:2] + segment + bare[2:], what, None if first != then else 6) - ) - # An IFD0 saying nothing reads as upright in a browser, so a turn held only elsewhere is disputed. - tiff = orientation_tiff(b"II", 3, 0, 6, sub=True) - segment = jpeg_segment(0xE1, b"Exif\x00\x00" + tiff) - out.append( - (bare[:2] + segment + bare[2:], "Orientation 6 in the sub-IFD alone", None) - ) - # A zeroed pointer points at the header, whose bytes read as an IFD of thousands of entries. - ifd0 = struct.pack(" int | None: - """The Orientation the first Exif in a file holds, 1 where it holds none, or None where disputed.""" - held = [] - if kind == "jpeg": - parts, _ = gate.jpeg_parts(data) - held = [data[s + 4 : e] for m, s, e in parts if m == 0xE1] - held = [h for h in held if h.startswith(b"Exif\x00\x00")] - elif kind == "png": - parts, _ = gate.png_parts(data) - held = [data[s + 8 : e - 4] for c, s, e in parts if c == b"eXIf"] - elif kind == "webp": - parts, _ = gate.webp_parts(data) - held = [data[s + 8 : e] for c, s, e in parts if c == b"EXIF"] - return normalizer.exif_orientation(held[0]) if held else 1 - - -class Report: - def __init__(self) -> None: - self.cases = 0 - self.failures: dict[tuple[str, str, str], tuple[int, str]] = {} - - def fail(self, prop: str, kind: str, detail: str, where: str) -> None: - key = (prop, kind, re.sub(r"0x[0-9A-F]+", "0x..", detail)) - count, first = self.failures.get(key, (0, where)) - self.failures[key] = (count + 1, first) - - -def attempt(call: Callable[[], object]) -> tuple[Any, str | None]: - """Run one parser call, returning its result or the name of what it raised.""" - try: - return call(), None - except Exception as exc: # noqa: BLE001 - return None, type(exc).__name__ - - -def check_variant(report: Report, kind: str, data: bytes, where: str) -> None: - report.cases += 1 - _, raised = attempt(lambda: gate.scan(data)) - if raised: - report.fail("1 scanner raised", kind, raised, where) - new, raised = attempt(lambda: normalizer.normalize_bytes(data)) - if raised: - report.fail("2 normalizer raised", kind, raised, where) - if not isinstance(new, bytes) or not new: - return - again, raised = attempt(lambda: gate.scan(new)) - if raised: - report.fail("1 scanner raised on normalized output", kind, raised, where) - elif again: - detail = ", ".join(sorted(again)) - report.fail("2 normalizer output rejected", kind, detail, where) - payloads, raised = attempt( - lambda: (normalizer.pixel_payload(data), normalizer.pixel_payload(new)) - ) - if raised: - report.fail("3 payload extraction raised", kind, raised, where) - elif payloads[0] is not None and payloads[0] != payloads[1]: - report.fail("3 normalizer changed the payload", kind, "", where) - - -def check_plant( - report: Report, - kind: str, - data: bytes, - where: str, - expect: str | bytes | None = None, -) -> None: - report.cases += 1 - found, raised = attempt(lambda: gate.scan(data)) - what = re.sub(r" at \d+$", "", where.split(": ", 1)[-1]) - if raised: - report.fail("1 scanner raised", kind, raised, where) - elif not found: - report.fail("4 planted metadata passed", kind, what, where) - new, raised = attempt(lambda: normalizer.normalize_bytes(data)) - if raised: - report.fail("2 normalizer raised", kind, raised, where) - return - rewritten = isinstance(new, bytes) and bool(new) - if expect == REFUSED and rewritten: - report.fail("4 normalizer rewrote a plant it should refuse", kind, what, where) - elif expect not in (None, REFUSED) and not rewritten: - report.fail("4 normalizer refused a plant it should rewrite", kind, what, where) - elif isinstance(expect, bytes) and rewritten and new != expect: - report.fail( - "4 normalizer rewrote a plant other than declared", kind, what, where - ) - if isinstance(new, bytes) and PLANT_TEXT in new: - report.fail("4 normalizer kept planted metadata", kind, what, where) - elif rewritten: - again, raised = attempt(lambda: gate.scan(new)) - if raised: - report.fail("1 scanner raised on normalized output", kind, raised, where) - elif again: - detail = ", ".join(sorted(again)) - report.fail("2 normalizer output rejected", kind, detail, where) - payloads, raised = attempt( - lambda: (normalizer.pixel_payload(data), normalizer.pixel_payload(new)) - ) - if raised: - report.fail("3 payload extraction raised", kind, raised, where) - elif payloads[0] is not None and payloads[0] != payloads[1]: - report.fail("3 normalizer changed the payload", kind, what, where) - - -def check_clean(report: Report, kind: str, data: bytes, where: str) -> None: - report.cases += 1 - what = where.split(": ", 1)[-1] - found, raised = attempt(lambda: gate.scan(data)) - if raised: - report.fail("1 scanner raised", kind, raised, where) - elif found: - report.fail("7 clean plant reported", kind, ", ".join(sorted(found)), where) - new, raised = attempt(lambda: normalizer.normalize_bytes(data)) - if raised: - report.fail("2 normalizer raised", kind, raised, where) - elif not isinstance(new, bytes) or not new: - report.fail("7 normalizer refused a clean plant", kind, what, where) - elif new != data: - report.fail("7 normalizer changed a clean plant", kind, what, where) - - -def check_turn( - report: Report, kind: str, data: bytes, where: str, shown: int | None -) -> None: - report.cases += 1 - disputed = shown is None - what = where.split(": ", 1)[-1] - found, raised = attempt(lambda: gate.scan(data)) - if raised: - report.fail("1 scanner raised", kind, raised, where) - elif disputed and not found: - report.fail("4 gate admitted a disputed orientation", kind, what, where) - new, raised = attempt(lambda: normalizer.normalize_bytes(data)) - if raised: - report.fail("2 normalizer raised", kind, raised, where) - return - if isinstance(new, bytes) and new: - again, raised = attempt(lambda: gate.scan(new)) - if raised: - report.fail("1 scanner raised on normalized output", kind, raised, where) - elif again: - detail = ", ".join(sorted(again)) - report.fail("2 normalizer output rejected", kind, detail, where) - if disputed and isinstance(new, bytes) and new: - report.fail("6 normalizer settled a disputed orientation", kind, what, where) - elif kind == "jpeg" and not disputed and not new: - # Only PNG and WebP refuse a turn, since a JPEG can carry one without its other Exif. - report.fail("6 normalizer refused an undisputed orientation", kind, what, where) - elif kind in ("png", "webp") and shown != 1 and new: - # A PNG or WebP keeps no Exif, so only a refusal keeps its turn. - report.fail("6 normalizer settled a PNG or WebP turn", kind, what, where) - elif isinstance(new, bytes) and new: - kept, raised = attempt(lambda: orientation(kind, new)) - if raised or kept != shown: - report.fail("6 normalizer lost the orientation", kind, what, where) - - -def check_icon(report: Report, data: bytes, where: str) -> None: - report.cases += 1 - found, raised = attempt(lambda: gate.scan_ico(data)) - if raised: - report.fail("1 scanner raised", "ico", raised, where) - return - for route, call in ( - ("site file", gate.scan_site), - ("zip member", gate.scan_member), - ): - again, raised = attempt(functools.partial(call, data, ".ico")) - if raised: - report.fail("8 icon scan raised", "ico", f"{route} {raised}", where) - elif again != found: - report.fail("8 icon scan answered differently", "ico", route, where) - - -def check_archive( - report: Report, data: bytes, where: str, scratch: pathlib.Path -) -> None: - report.cases += 1 - path = scratch / "fuzz.zip" - path.write_bytes(data) - _, raised = attempt(lambda: gate.walk_archive(path, "fuzz.zip")) - if raised: - report.fail("5 archive walk raised", "zip", raised, where) - with contextlib.redirect_stdout(io.StringIO()): - _, raised = attempt(lambda: normalizer.normalize_archive(path, False)) - if raised: - report.fail("5 archive normalize raised", "zip", raised, where) - - -def carried_seeds( - rng: random.Random, per_kind: int, max_size: int -) -> list[tuple[str, bytes]]: - """A deterministic sample of the carried media, capped in size so the run stays fast.""" - by_kind: dict[str, list[pathlib.Path]] = {} - for tree in gate.TREES + gate.SITE_TREES: - root = REPO / tree - if gate.symlinked(root) or not root.is_dir(): - continue - for path in sorted(root.rglob("*"), key=pathlib.Path.as_posix): - if path.is_symlink() or not path.is_file() or path.suffix.lower() == ".zip": - continue - if path.stat().st_size > max_size: - continue - with path.open("rb") as handle: - kind = container(handle.read(12)) - if tree in gate.SITE_TREES and ( - kind == "other" or path.suffix.lower() in gate.SITE_TEXT - ): - continue - by_kind.setdefault(kind, []).append(path) - seeds = [] - for kind in sorted(by_kind): - paths = by_kind[kind] - for path in rng.sample(paths, min(per_kind, len(paths))): - seeds.append((str(path.relative_to(REPO)), path.read_bytes())) - return seeds - - -def main() -> int: - parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) - parser.add_argument("--seed", type=int, default=173, help="random seed") - parser.add_argument("--cases", type=int, default=400, help="variants per seed file") - parser.add_argument( - "--per-kind", type=int, default=6, help="carried files per container" - ) - parser.add_argument( - "--max-size", type=int, default=512 * 1024, help="largest carried seed" - ) - parser.add_argument( - "--budget", type=float, default=90.0, help="seconds before stopping" - ) - parser.add_argument( - "--min-cases", - type=int, - default=4000, - help="fewest cases a run the time budget cut short may pass with", - ) - parser.add_argument( - "--no-carried", action="store_true", help="constructed fixtures only" - ) - args = parser.parse_args() - - warnings.simplefilter("ignore") - rng = random.Random(args.seed) - seeds = [ - ("fixture:jpeg", jpeg_fixture(False)), - ("fixture:jpeg-progressive", jpeg_fixture(True)), - ("fixture:jpeg-dated", dated_jpeg_fixture()), - ("fixture:png", png_fixture()), - ("fixture:png-palette", palette_png_fixture()), - ("fixture:png-truecolor", truecolor_png_fixture()), - ("fixture:png-small", small_png_fixture()), - ("fixture:gif", gif_fixture()), - ("fixture:webp", webp_fixture()), - ("fixture:webp-animated", animated_webp_fixture()), - ("fixture:iso", iso_fixture()), - ("fixture:ico", ico_fixture()), - ] - if not args.no_carried: - seeds += carried_seeds(rng, args.per_kind, args.max_size) - - report = Report() - deadline = time.monotonic() + args.budget - archive = zip_fixture() - with tempfile.TemporaryDirectory() as scratch: - for _ in range(args.cases): - if time.monotonic() > deadline: - break - variant, what = rng.choice(MUTATORS)(rng, archive) - check_archive( - report, variant, f"fixture:zip: {what}", pathlib.Path(scratch) - ) - for label, data in seeds: - kind = container(data) - if kind == "ico": - if label.startswith("fixture:"): - found, raised = attempt(functools.partial(gate.scan_ico, data)) - if not raised and found: - report.fail( - "0 fixture not clean", kind, ", ".join(sorted(found)), label - ) - check_icon(report, data, f"{label}: unmodified") - for _ in range(args.cases): - if time.monotonic() > deadline: - break - variant, what = rng.choice(MUTATORS)(rng, data) - check_icon(report, variant, f"{label}: {what}") - continue - check_variant(report, kind, data, f"{label}: unmodified") - if label.startswith("fixture:"): - found, raised = attempt(functools.partial(gate.scan, data)) - if not raised and found: - detail = ", ".join(sorted(found)) - report.fail("0 fixture not clean", kind, detail, label) - # Planting and turning split the seed with the gate's parsers and this file's own code. - planted, raised = attempt(functools.partial(plants, kind, data)) - if raised: - report.fail("0 variants not built", kind, f"plants {raised}", label) - turns, raised = attempt(functools.partial(turned, kind, data)) - if raised: - report.fail("0 variants not built", kind, f"turned {raised}", label) - for variant, what, *expect in planted or []: - check_plant(report, kind, variant, f"{label}: {what}", *expect) - kept, raised = attempt(functools.partial(clean_plants, kind, data)) - if raised: - report.fail("0 variants not built", kind, f"clean {raised}", label) - for variant, what in kept or []: - check_clean(report, kind, variant, f"{label}: {what}") - for variant, what, shown in turns or []: - check_turn(report, kind, variant, f"{label}: {what}", shown) - for _ in range(args.cases): - if time.monotonic() > deadline: - break - mutate = rng.choice(MUTATORS) - variant, what = mutate(rng, data) - check_variant(report, kind, variant, f"{label}: {what}") - stopped = time.monotonic() > deadline - - for (prop, kind, detail), (count, first) in sorted(report.failures.items()): - suffix = f" ({detail})" if detail else "" - print(f"{prop}: {kind}{suffix}, {count} case(s), first {first}") - budget = ", stopped at the time budget" if stopped else "" - print(f"\nfuzz : {report.cases} case(s) over {len(seeds)} seed(s){budget}") - short = stopped and report.cases < args.min_cases - if short: - print(f"the time budget stopped the run below {args.min_cases} case(s)") - if report.failures: - print(f"{len(report.failures)} distinct failure(s)") - return 1 if report.failures or short else 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/checks/tests/media_fixtures.py b/checks/tests/media_fixtures.py new file mode 100644 index 0000000..ab678cf --- /dev/null +++ b/checks/tests/media_fixtures.py @@ -0,0 +1,123 @@ +"""Constructed media the media gate tests build their cases from.""" + +import struct +import zlib + + +def jpeg_segment(marker: int, body: bytes) -> bytes: + return bytes((0xFF, marker)) + struct.pack(">H", len(body) + 2) + body + + +def exif_segment() -> bytes: + """An Exif APP1 carrying only Orientation.""" + entry = struct.pack(" bytes: + """An Exif APP1 with one IFD holding the given tag, type, count and value entries. + + A pad goes between the IFD and the values it points to, as a writer aligning them leaves one. + """ + table, values = b"", b"" + start = 8 + 2 + 12 * len(entries) + 4 + len(pad) + for tag, kind, count, value in entries: + if len(value) > 4: + field = struct.pack(" bytes: + """A fixed-shape segment's fields with the bytes at one offset replaced.""" + return fields[:at] + value + fields[at + len(value) :] + + +SQUARE = struct.pack(">IIB", 3779, 3779, 1) +ICC = b"ICC_PROFILE\x00\x01\x01" + + +def jpeg_fixture() -> bytes: + """A structurally complete JPEG, with stuffed bytes and a restart marker in its scan.""" + out = b"\xff\xd8" + jpeg_segment( + 0xE0, b"JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00" + ) + out += exif_segment() + out += jpeg_segment(0xDB, b"\x00" + bytes(range(64))) + out += jpeg_segment(0xC0, b"\x08\x00\x08\x00\x08\x01\x01\x11\x00") + out += jpeg_segment(0xC4, b"\x00" + bytes(16) + b"") + out += jpeg_segment(0xDA, b"\x01\x01\x00\x00\x3f\x00") + out += b"\x12\x34\xff\x00\x56\xff\xd0\x78\x9a" + return out + b"\xff\xd9" + + +def png_chunk(name: bytes, body: bytes) -> bytes: + crc = zlib.crc32(name + body) & 0xFFFFFFFF + return struct.pack(">I", len(body)) + name + body + struct.pack(">I", crc) + + +def png_fixture() -> bytes: + ihdr = struct.pack(">IIBBBBB", 1, 1, 8, 0, 0, 0, 0) + idat = zlib.compress(b"\x00\x7f") + return ( + b"\x89PNG\r\n\x1a\n" + + png_chunk(b"IHDR", ihdr) + + png_chunk(b"gAMA", struct.pack(">I", 45455)) + + png_chunk(b"IDAT", idat) + + png_chunk(b"IEND", b"") + ) + + +def palette_png_fixture() -> bytes: + ihdr = struct.pack(">IIBBBBB", 1, 1, 1, 3, 0, 0, 0) + idat = zlib.compress(b"\x00\x00") + return ( + b"\x89PNG\r\n\x1a\n" + + png_chunk(b"IHDR", ihdr) + + png_chunk(b"PLTE", bytes(6)) + + png_chunk(b"IDAT", idat) + + png_chunk(b"IEND", b"") + ) + + +def gif_fixture() -> bytes: + head = b"GIF89a" + struct.pack(" bytes: + return ( + name + struct.pack(" bytes: + body = riff_chunk(b"VP8X", bytes(10)) + riff_chunk(b"VP8L", b"\x2f\x00\x00\x00\x00") + return b"RIFF" + struct.pack(" bytes: + """An animated WebP of one frame, which holds its own chunks inside an ANMF.""" + frame = bytes(12) + b"\x64\x00\x00\x00" + frame += riff_chunk(b"VP8L", b"\x2f\x00\x00\x00\x00") + body = riff_chunk(b"VP8X", b"\x02" + bytes(9)) + riff_chunk(b"ANIM", bytes(6)) + body += riff_chunk(b"ANMF", frame) + return b"RIFF" + struct.pack(" bytes: - data = fuzz.jpeg_fixture(False) - bare = data.replace(fuzz.jpeg_segment(0xE0, fuzz.JFIF), b"") - bare = bare.replace(fuzz.exif_segment(), b"") + data = fixtures.jpeg_fixture() + bare = data.replace(fixtures.jpeg_segment(0xE0, fixtures.JFIF), b"") + bare = bare.replace(fixtures.exif_segment(), b"") return bare[:2] + b"".join(segments) + bare[2:] @@ -44,13 +42,13 @@ def icc_chunks(profile: bytes, order: tuple[int, ...]) -> list[bytes]: half = len(profile) // 2 pieces = (profile[:half], profile[half:]) return [ - fuzz.jpeg_segment(0xE2, fuzz.ICC[:12] + bytes((n, 2)) + pieces[n - 1]) + fixtures.jpeg_segment(0xE2, fixtures.ICC[:12] + bytes((n, 2)) + pieces[n - 1]) for n in order ] def png_with(chunk: bytes) -> bytes: - data = fuzz.png_fixture() + data = fixtures.png_fixture() return data[:33] + chunk + data[33:] @@ -59,20 +57,20 @@ def span(data: bytes, name: bytes) -> tuple[int, int]: def truecolor_png(chunks: bytes) -> bytes: - ihdr = fuzz.png_chunk(b"IHDR", struct.pack(">IIBBBBB", 1, 1, 8, 2, 0, 0, 0)) - idat = fuzz.png_chunk(b"IDAT", zlib.compress(bytes(4))) - end = fuzz.png_chunk(b"IEND", b"") + ihdr = fixtures.png_chunk(b"IHDR", struct.pack(">IIBBBBB", 1, 1, 8, 2, 0, 0, 0)) + idat = fixtures.png_chunk(b"IDAT", zlib.compress(bytes(4))) + end = fixtures.png_chunk(b"IEND", b"") return b"\x89PNG\r\n\x1a\n" + ihdr + chunks + idat + end def iccp(name: bytes, stream: bytes) -> bytes: - return fuzz.png_chunk(b"iCCP", name + b"\x00\x00" + stream) + return fixtures.png_chunk(b"iCCP", name + b"\x00\x00" + stream) def webp_with_profile(flags: int, profile: bytes) -> bytes: - body = fuzz.riff_chunk(b"VP8X", bytes((flags,)) + bytes(9)) - body += fuzz.riff_chunk(b"ICCP", profile) - body += fuzz.riff_chunk(b"VP8L", b"\x2f\x00\x00\x00\x00") + body = fixtures.riff_chunk(b"VP8X", bytes((flags,)) + bytes(9)) + body += fixtures.riff_chunk(b"ICCP", profile) + body += fixtures.riff_chunk(b"VP8L", b"\x2f\x00\x00\x00\x00") return b"RIFF" + struct.pack(" bytes: def text_chunk() -> bytes: - return fuzz.png_chunk(b"tEXt", b"Author\x00someone") + return fixtures.png_chunk(b"tEXt", b"Author\x00someone") def iso_with_title() -> bytes: @@ -102,7 +100,7 @@ def box(kind: bytes, body: bytes) -> bytes: class Profiles(unittest.TestCase): def test_known_jpeg_profile_passes_as_one_chunk(self) -> None: - segment = fuzz.jpeg_segment(0xE2, fuzz.ICC + PROFILE) + segment = fixtures.jpeg_segment(0xE2, fixtures.ICC + PROFILE) with known(): self.assertEqual(gate.scan(jpeg_with(segment)), set()) @@ -116,11 +114,11 @@ def test_known_jpeg_profile_in_other_chunks_is_recut(self) -> None: new = normalizer.normalize_bytes(data) self.assertIsNotNone(new) self.assertEqual(gate.scan(new), set()) - self.assertIn(fuzz.ICC + PROFILE, new) + self.assertIn(fixtures.ICC + PROFILE, new) def test_jpeg_profile_after_the_scan_is_refused(self) -> None: first, second = icc_chunks(PROFILE, (1, 2)) - whole = fuzz.jpeg_segment(0xE2, fuzz.ICC + PROFILE) + whole = fixtures.jpeg_segment(0xE2, fixtures.ICC + PROFILE) with known(): for before, after in (([first], second), ([], whole)): data = jpeg_with(*before) @@ -134,7 +132,7 @@ def test_unknown_jpeg_profile_is_refused(self) -> None: self.assertIsNone(normalizer.normalize_bytes(data)) def test_known_png_profile_in_its_canonical_body_passes(self) -> None: - data = png_with(fuzz.png_chunk(b"iCCP", gate.png_icc_body(PROFILE))) + data = png_with(fixtures.png_chunk(b"iCCP", gate.png_icc_body(PROFILE))) with known(): self.assertEqual(gate.scan(data), set()) @@ -181,7 +179,9 @@ def test_unknown_webp_profile_is_refused(self) -> None: class Fields(unittest.TestCase): def test_jfif_out_of_its_values_is_restated(self) -> None: - data = jpeg_with(fuzz.jpeg_segment(0xE0, fuzz.bend(fuzz.JFIF, 8, b"\x00\x48"))) + data = jpeg_with( + fixtures.jpeg_segment(0xE0, fixtures.bend(fixtures.JFIF, 8, b"\x00\x48")) + ) self.assertTrue(gate.scan(data)) new = normalizer.normalize_bytes(data) self.assertIsNotNone(new) @@ -194,21 +194,29 @@ def test_real_jfif_values_pass(self) -> None: (b"\x01\x00", 2, b"\x00\x00"), ): fields = ( - fuzz.JFIF[:5] + version + bytes((units,)) + density * 2 + b"\x00\x00" + fixtures.JFIF[:5] + + version + + bytes((units,)) + + density * 2 + + b"\x00\x00" ) self.assertEqual( - gate.scan(jpeg_with(fuzz.jpeg_segment(0xE0, fields))), set() + gate.scan(jpeg_with(fixtures.jpeg_segment(0xE0, fields))), set() ) def test_adobe_flags_are_cleared_and_transform_kept(self) -> None: - data = jpeg_with(fuzz.jpeg_segment(0xEE, fuzz.bend(fuzz.ADOBE, 7, b"\x40\x00"))) + data = jpeg_with( + fixtures.jpeg_segment(0xEE, fixtures.bend(fixtures.ADOBE, 7, b"\x40\x00")) + ) new = normalizer.normalize_bytes(data) self.assertIsNotNone(new) self.assertEqual(gate.scan(new), set()) - self.assertIn(fuzz.ADOBE, new) + self.assertIn(fixtures.ADOBE, new) def test_adobe_transform_out_of_its_values_is_refused(self) -> None: - data = jpeg_with(fuzz.jpeg_segment(0xEE, fuzz.bend(fuzz.ADOBE, 11, b"\x09"))) + data = jpeg_with( + fixtures.jpeg_segment(0xEE, fixtures.bend(fixtures.ADOBE, 11, b"\x09")) + ) self.assertTrue(gate.scan(data)) self.assertIsNone(normalizer.normalize_bytes(data)) @@ -216,18 +224,20 @@ def test_exif_values_in_range_pass(self) -> None: for value in range(1, 9): entry = (0x0112, 3, 1, struct.pack(" None: - entries = [fuzz.ORIENTATION, (0x0132, 2, 20, fuzz.DATE)] - segment = fuzz.exif_ifd_segment(entries, b"\x00") + entries = [fixtures.ORIENTATION, (0x0132, 2, 20, fixtures.DATE)] + segment = fixtures.exif_ifd_segment(entries, b"\x00") self.assertEqual(gate.scan(jpeg_with(segment)), set()) def test_zero_webp_pad_byte_passes(self) -> None: - self.assertEqual(gate.scan(fuzz.webp_fixture()), set()) + self.assertEqual(gate.scan(fixtures.webp_fixture()), set()) class FreeValues(unittest.TestCase): @@ -246,40 +256,42 @@ def test_png_values_writers_use_pass(self) -> None: (b"bKGD", bytes(2)), (b"tRNS", struct.pack(">H", 255)), ): - self.assertEqual(gate.scan(png_with(fuzz.png_chunk(chunk, body))), set()) + self.assertEqual( + gate.scan(png_with(fixtures.png_chunk(chunk, body))), set() + ) def test_png_undrawn_chunk_out_of_its_value_is_dropped(self) -> None: for chunk, body in ((b"sBIT", b"\x05"), (b"bKGD", b"\x00\x07")): self.assert_restated( - png_with(fuzz.png_chunk(chunk, body)), fuzz.png_fixture() + png_with(fixtures.png_chunk(chunk, body)), fixtures.png_fixture() ) def test_png_misplaced_chunk_is_dropped_or_refused(self) -> None: - data = fuzz.png_fixture() - late = fuzz.png_chunk(b"sBIT", b"\x08") + data = fixtures.png_fixture() + late = fixtures.png_chunk(b"sBIT", b"\x08") self.assert_restated(data[:-12] + late + data[-12:], data) - late = fuzz.png_chunk(b"pHYs", fuzz.SQUARE) + late = fixtures.png_chunk(b"pHYs", fixtures.SQUARE) data = data[:-12] + late + data[-12:] self.assertIn("PNG pHYs chunk out of place", gate.scan(data)) self.assertIsNone(normalizer.normalize_bytes(data)) def test_png_repeated_gamma_is_refused(self) -> None: - data = png_with(fuzz.png_chunk(b"gAMA", struct.pack(">I", 45455))) + data = png_with(fixtures.png_chunk(b"gAMA", struct.pack(">I", 45455))) self.assertIn("PNG repeated gAMA chunk", gate.scan(data)) self.assertIsNone(normalizer.normalize_bytes(data)) def test_png_repeated_palette_is_refused(self) -> None: - palette = fuzz.png_chunk(b"PLTE", bytes(6)) + palette = fixtures.png_chunk(b"PLTE", bytes(6)) data = truecolor_png(palette * 2) self.assertIn("PNG repeated PLTE chunk", gate.scan(data)) self.assertIsNone(normalizer.normalize_bytes(data)) def test_png_animation_is_refused(self) -> None: - data = fuzz.png_fixture() - control = fuzz.png_chunk(b"acTL", struct.pack(">II", 1, 0)) - frame = fuzz.png_chunk(b"fdAT", bytes(4)) + data = fixtures.png_fixture() + control = fixtures.png_chunk(b"acTL", struct.pack(">II", 1, 0)) + frame = fixtures.png_chunk(b"fdAT", bytes(4)) header = struct.pack(">IIIIIHHBB", 0, 1, 1, 0, 0, 1, 1, 0, 0) - region = fuzz.png_chunk(b"fcTL", header) + region = fixtures.png_chunk(b"fcTL", header) for chunk, bent in ( (b"acTL", png_with(control)), (b"fcTL", png_with(region)), @@ -289,18 +301,18 @@ def test_png_animation_is_refused(self) -> None: self.assertIsNone(normalizer.normalize_bytes(bent)) def test_png_suggested_palette_goes_after_its_transparency(self) -> None: - color_key = fuzz.png_chunk(b"tRNS", bytes(6)) - data = truecolor_png(color_key + fuzz.png_chunk(b"PLTE", bytes(6))) + color_key = fixtures.png_chunk(b"tRNS", bytes(6)) + data = truecolor_png(color_key + fixtures.png_chunk(b"PLTE", bytes(6))) self.assertEqual( gate.scan(data), {"PNG PLTE fields not values its format defines"} ) self.assert_restated(data, truecolor_png(color_key)) def test_png_refused_for_its_structure_is_not_inflated(self) -> None: - data = fuzz.png_fixture() + data = fixtures.png_fixture() signature, ihdr, end = data[:8], data[8:33], data[-12:] idat = data[slice(*span(data, b"IDAT"))] - palette = fuzz.palette_png_fixture() + palette = fixtures.palette_png_fixture() plte = palette[slice(*span(palette, b"PLTE"))] twice = palette.replace(plte, plte * 2) with mock.patch.object(gate.zlib, "decompressobj") as inflate: @@ -312,8 +324,8 @@ def test_png_refused_for_its_structure_is_not_inflated(self) -> None: inflate.assert_not_called() # A stream fault beside a droppable chunk is still named, since it is why the normalizer refuses. start, stop = span(data, b"IDAT") - cut = fuzz.png_chunk(b"IDAT", data[start + 8 : stop - 8]) - text = fuzz.png_chunk(b"tEXt", b"Comment\x00planted") + cut = fixtures.png_chunk(b"IDAT", data[start + 8 : stop - 8]) + text = fixtures.png_chunk(b"tEXt", b"Comment\x00planted") bent = data[:start] + cut + text + data[stop:] self.assertEqual( gate.scan(bent), @@ -322,10 +334,10 @@ def test_png_refused_for_its_structure_is_not_inflated(self) -> None: self.assertIsNone(normalizer.normalize_bytes(bent)) def test_png_missing_critical_chunk_is_refused(self) -> None: - data = fuzz.png_fixture() + data = fixtures.png_fixture() signature, ihdr, end = data[:8], data[8:33], data[-12:] idat = data[slice(*span(data, b"IDAT"))] - palette = fuzz.palette_png_fixture() + palette = fixtures.palette_png_fixture() first, last = span(palette, b"PLTE") for problem, bent in ( ("PNG without IHDR", signature + idat + end), @@ -348,38 +360,39 @@ def test_png_header_out_of_its_values_is_refused(self) -> None: bytes(4) + good[4:], good + b"\x00", ): - data = fuzz.png_fixture() - bent = data[:8] + fuzz.png_chunk(b"IHDR", header) + data[33:] + data = fixtures.png_fixture() + bent = data[:8] + fixtures.png_chunk(b"IHDR", header) + data[33:] self.assertIn( "PNG IHDR fields not values its format defines", gate.scan(bent) ) self.assertIsNone(normalizer.normalize_bytes(bent)) - data = fuzz.png_fixture() + data = fixtures.png_fixture() self.assertEqual( - gate.scan(data[:8] + fuzz.png_chunk(b"IHDR", interlaced) + data[33:]), set() + gate.scan(data[:8] + fixtures.png_chunk(b"IHDR", interlaced) + data[33:]), + set(), ) def test_png_side_past_libpng_limit_is_refused(self) -> None: limit = gate.PNG_SIDE_LIMIT for width, height in ((limit + 1, 1), (1, limit + 1)): header = struct.pack(">IIBBBBB", width, height, 8, 0, 0, 0, 0) - data = fuzz.png_fixture() - bent = data[:8] + fuzz.png_chunk(b"IHDR", header) + data[33:] + data = fixtures.png_fixture() + bent = data[:8] + fixtures.png_chunk(b"IHDR", header) + data[33:] self.assertEqual(gate.scan(bent), {"PNG IHDR side past libpng's limit"}) self.assertIsNone(normalizer.normalize_bytes(bent)) header = struct.pack(">IIBBBBB", limit, 1, 8, 0, 0, 0, 0) - rows = fuzz.png_chunk(b"IDAT", zlib.compress(bytes(limit + 1))) - data = fuzz.png_fixture() + rows = fixtures.png_chunk(b"IDAT", zlib.compress(bytes(limit + 1))) + data = fixtures.png_fixture() start, end = span(data, b"IDAT") - wide = data[:8] + fuzz.png_chunk(b"IHDR", header) + data[33:start] + rows + wide = data[:8] + fixtures.png_chunk(b"IHDR", header) + data[33:start] + rows self.assertEqual(gate.scan(wide + data[end:]), set()) def test_png_picture_past_the_size_limit_is_refused_before_inflating(self) -> None: side = gate.PNG_SIDE_LIMIT // 10 header = struct.pack(">IIBBBBB", side, side, 8, 0, 0, 0, 0) self.assertGreater(gate.png_picture_size(header), gate.SIZE_LIMIT) - data = fuzz.png_fixture() - bent = data[:8] + fuzz.png_chunk(b"IHDR", header) + data[33:] + data = fixtures.png_fixture() + bent = data[:8] + fixtures.png_chunk(b"IHDR", header) + data[33:] with mock.patch.object(gate.zlib, "decompressobj") as inflate: self.assertEqual( gate.scan(bent), {"PNG IHDR picture larger than the gate inflates"} @@ -401,7 +414,7 @@ def test_png_interlaced_size_sums_its_adam7_passes(self) -> None: self.assertEqual(gate.png_picture_size(small[:12] + b"\x00"), 3 * 2) def test_png_picture_data_not_one_whole_stream_is_refused(self) -> None: - data = fuzz.png_fixture() + data = fixtures.png_fixture() start, end = span(data, b"IDAT") stream = data[start + 8 : end - 4] rows = zlib.decompress(stream) @@ -418,27 +431,27 @@ def test_png_picture_data_not_one_whole_stream_is_refused(self) -> None: [zlib.compress(b"\x05" + rows[1:])], ), ): - idat = b"".join(fuzz.png_chunk(b"IDAT", body) for body in bodies) + idat = b"".join(fixtures.png_chunk(b"IDAT", body) for body in bodies) bent = data[:start] + idat + data[end:] self.assertEqual(gate.scan(bent), {problem}) self.assertIsNone(normalizer.normalize_bytes(bent)) # A stream split across chunks at any byte is still one stream. split = [stream[:1], stream[1:3], stream[3:]] - whole = b"".join(fuzz.png_chunk(b"IDAT", body) for body in split) + whole = b"".join(fixtures.png_chunk(b"IDAT", body) for body in split) self.assertEqual(gate.scan(data[:start] + whole + data[end:]), set()) - text = fuzz.png_chunk(b"tEXt", b"Comment\x00planted") - parted = fuzz.png_chunk(b"IDAT", split[0]) + text - parted += fuzz.png_chunk(b"IDAT", b"".join(split[1:])) + text = fixtures.png_chunk(b"tEXt", b"Comment\x00planted") + parted = fixtures.png_chunk(b"IDAT", split[0]) + text + parted += fixtures.png_chunk(b"IDAT", b"".join(split[1:])) bent = data[:start] + parted + data[end:] self.assertIn("PNG IDAT chunks not consecutive", gate.scan(bent)) self.assertIsNone(normalizer.normalize_bytes(bent)) def test_png_end_with_a_body_is_emptied(self) -> None: - data = fuzz.png_fixture() - self.assert_restated(data[:-12] + fuzz.png_chunk(b"IEND", b"\x00"), data) + data = fixtures.png_fixture() + self.assert_restated(data[:-12] + fixtures.png_chunk(b"IEND", b"\x00"), data) def test_png_chunk_whose_crc_is_not_its_own_is_dropped_or_refused(self) -> None: - data = fuzz.png_fixture() + data = fixtures.png_fixture() self.assert_restated(data[:-4] + bytes(4), data) start, end = span(data, b"gAMA") bent = data[: end - 4] + bytes(4) + data[end:] @@ -449,19 +462,19 @@ def test_png_chunk_whose_crc_is_not_its_own_is_dropped_or_refused(self) -> None: self.assertIsNone(normalizer.normalize_bytes(bent)) def test_gif_unused_control_fields_are_zeroed(self) -> None: - data = fuzz.gif_fixture() + data = fixtures.gif_fixture() at = data.index(b"\x21\xf9") + 3 bent = data[:at] + b"\xe0" + data[at + 1 : at + 3] + b"\x05" + data[at + 4 :] self.assert_restated(bent, data) def test_gif_aspect_ratio_is_zeroed(self) -> None: - data = fuzz.gif_fixture() + data = fixtures.gif_fixture() bent = data[:12] + b"\x31" + data[13:] self.assertIn("GIF aspect ratio not zero", gate.scan(bent)) self.assert_restated(bent, data) def test_gif_unused_descriptor_fields_are_zeroed(self) -> None: - data = fuzz.gif_fixture() + data = fixtures.gif_fixture() at = data.index(b"\x2c") + 9 for bent in ( b"GIF87a" + data[6:], @@ -471,13 +484,13 @@ def test_gif_unused_descriptor_fields_are_zeroed(self) -> None: self.assert_restated(bent, data) def test_gif_unread_global_table_is_dropped(self) -> None: - data = fuzz.gif_fixture() + data = fixtures.gif_fixture() at = data.index(b"\x2c") + 9 local = data[:at] + b"\x80" + bytes(6) + data[at + 1 :] self.assert_restated(local, data[:10] + bytes(3) + local[19:]) def test_webp_background_is_zeroed_and_loop_count_kept(self) -> None: - data = fuzz.animated_webp_fixture() + data = fixtures.animated_webp_fixture() at = data.index(b"ANIM") + 8 looped = data[:at] + bytes(4) + b"\x03\x00" + data[at + 6 :] self.assertEqual(gate.scan(looped), set()) @@ -485,12 +498,12 @@ def test_webp_background_is_zeroed_and_loop_count_kept(self) -> None: self.assert_restated(bent, looped) def test_jpeg_fill_before_a_marker_is_dropped(self) -> None: - data = fuzz.jpeg_fixture(False) + data = fixtures.jpeg_fixture() for at in (2, len(data) - 2): self.assert_restated(data[:at] + b"\xff" * 3 + data[at:], data) def test_jpeg_fill_inside_a_scan_is_refused(self) -> None: - data = fuzz.jpeg_fixture(False).replace(b"\xff\xd0", b"\xff\xff\xd0") + data = fixtures.jpeg_fixture().replace(b"\xff\xd0", b"\xff\xff\xd0") self.assertIn("JPEG fill bytes inside a scan", gate.scan(data)) self.assertIsNone(normalizer.normalize_bytes(data)) @@ -649,31 +662,6 @@ def test_tree_under_a_symlinked_directory_is_named_and_not_read(self) -> None: self.link("static", target) self.assertEqual(gate.findings(), [("static/media", {"symlink, not read"})]) - def seeds(self) -> list[str]: - """The carried seeds the fuzzer samples from the scratch repository.""" - with mock.patch.object(fuzz, "REPO", self.root): - seeds = fuzz.carried_seeds(random.Random(0), 9, 1 << 20) - return [name for name, _ in seeds] - - def test_fuzzer_samples_a_real_tree(self) -> None: - held = self.root / "static" / "media" / "held.png" - held.parent.mkdir(parents=True) - held.write_bytes(png_with(text_chunk())) - self.assertEqual( - self.seeds(), [str(pathlib.Path("static", "media", "held.png"))] - ) - - def test_fuzzer_does_not_sample_a_symlinked_root(self) -> None: - self.link("sites", self.outside()) - self.assertEqual(self.seeds(), []) - - def test_fuzzer_does_not_sample_a_root_under_a_symlink(self) -> None: - target = self.outside() - (target / "media").mkdir() - (target / "media" / "held.png").write_bytes(png_with(text_chunk())) - self.link("static", target) - self.assertEqual(self.seeds(), []) - def test_normalizer_does_not_follow_a_symlinked_root(self) -> None: target = self.outside() before = (target / "held.png").read_bytes() @@ -826,17 +814,19 @@ def test_icon_inside_an_archive_is_read(self) -> None: self.assertIn(self.key("icons.zip") + "!favicon.ico", found) def test_large_icon_entry_states_its_side_as_zero(self) -> None: - ihdr = fuzz.png_chunk(b"IHDR", struct.pack(">IIBBBBB", 512, 1, 8, 0, 0, 0, 0)) - idat = fuzz.png_chunk(b"IDAT", zlib.compress(bytes(513))) - image = b"\x89PNG\r\n\x1a\n" + ihdr + idat + fuzz.png_chunk(b"IEND", b"") + ihdr = fixtures.png_chunk( + b"IHDR", struct.pack(">IIBBBBB", 512, 1, 8, 0, 0, 0, 0) + ) + idat = fixtures.png_chunk(b"IDAT", zlib.compress(bytes(513))) + image = b"\x89PNG\r\n\x1a\n" + ihdr + idat + fixtures.png_chunk(b"IEND", b"") data = ico_of(image) wide = data[:6] + b"\x00" + data[7:] self.assertEqual(self.found("favicon.ico", wide), {}) def test_clean_site_png_and_icon_pass(self) -> None: - self.found("apple-touch-icon.png", fuzz.png_fixture()) + self.found("apple-touch-icon.png", fixtures.png_fixture()) self.assertEqual( - self.found("favicon.ico", ico_of(*[fuzz.png_fixture()] * 2)), {} + self.found("favicon.ico", ico_of(*[fixtures.png_fixture()] * 2)), {} ) def test_icon_entry_holding_text_is_reported(self) -> None: @@ -845,13 +835,13 @@ def test_icon_entry_holding_text_is_reported(self) -> None: self.assertTrue(all(tag.startswith("ICO entry 0: ") for tag in tags), tags) def test_icon_bytes_outside_its_entries_are_reported(self) -> None: - found = self.found("favicon.ico", ico_of(fuzz.png_fixture(), tail=b"note")) + found = self.found("favicon.ico", ico_of(fixtures.png_fixture(), tail=b"note")) self.assertEqual( found[self.key("favicon.ico")], {"ICO bytes past the last entry"} ) def test_icon_entry_past_the_end_is_reported(self) -> None: - data = ico_of(fuzz.png_fixture()) + data = ico_of(fixtures.png_fixture()) found = self.found("favicon.ico", data[:-1]) self.assertEqual( found[self.key("favicon.ico")], @@ -859,7 +849,7 @@ def test_icon_entry_past_the_end_is_reported(self) -> None: ) def test_icon_bytes_before_an_entry_are_reported(self) -> None: - image = fuzz.png_fixture() + image = fixtures.png_fixture() entry = struct.pack(" None: self.assertEqual(found[self.key("favicon.ico")], {"ICO entry 0 not a PNG"}) def test_icon_directory_disagreeing_with_its_image_is_reported(self) -> None: - found = self.found("favicon.ico", ico_of(fuzz.png_fixture(), bits=32)) + found = self.found("favicon.ico", ico_of(fixtures.png_fixture(), bits=32)) self.assertEqual( found[self.key("favicon.ico")], {"ICO entry 0 directory not its image's own"},