lambda_function.py

"""
CDR Lambda — Content Disarmament and Reconstruction
Triggered by EventBridge on S3 ObjectCreated events.

Supported formats:
  Office  — .docx .xlsx .pptx and all ZIP/OOXML variants (removes macros, OLE, external links)
  PDF     — strips JS, OpenAction, Launch, form submit actions, embedded files
  Images  — re-encodes via Pillow to purge EXIF / ICC exploits

Processing flow (see ``handler``):
  download → size/ZIP-structure validation → fail-closed routing by extension →
  format-specific CDR (``cdr_office`` / ``cdr_pdf`` / ``cdr_image``) → upload to
  SANITISED_BUCKET → publish result + delete source. Rejected/errored/unsupported files
  are quarantined instead. Unknown extensions FAIL CLOSED — they are never labelled
  sanitised.

Design rules that must not be weakened:
  * Side effects (SNS publish, source delete, metrics) are fault-isolated — they can never
    turn a successful CDR into an EventBridge retry.
  * ZIP entries are read through ``_read_zip_entry_safe`` (chunked counter, never trusts
    the central-directory ``file_size``) to defend against decompression bombs.
  * CDR drops/neutralises content in place; it never re-serialises through an Office
    library, so anything not explicitly touched is preserved.

Environment variables:
  SANITISED_BUCKET     destination bucket for clean files (required)
  QUARANTINE_BUCKET    destination for rejected/errored files (optional)
  RESULT_TOPIC_ARN     SNS topic for CDR result metadata (optional)
  CDR_MAX_FILE_BYTES   pre-download size limit in bytes (default 104857600 = 100 MB)
  CDR_MAX_ENTRY_BYTES  per-ZIP-entry decompression limit (default 209715200 = 200 MB)
  CDR_MAX_IMAGE_PIXELS decompression-bomb pixel cap for cdr_image (default 40000000 = 40 MP)
"""

import decimal
import html
import io
import json
import logging
import os
import posixpath
import re
import struct
import warnings
import xml.etree.ElementTree as ET
import zipfile
from datetime import datetime, timezone
from collections.abc import Iterable
from typing import Optional

import boto3
import openpyxl
import pikepdf
import pyxlsb
from botocore.config import Config
from PIL import Image, ImageSequence, UnidentifiedImageError

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Explicit client config rather than botocore's defaults. On a 300 s Lambda under high
# utilisation the defaults are actively harmful: a 60 s connect/read timeout with legacy
# retries lets a single hung S3 socket consume most of the invocation budget, and every
# second spent blocked holds one of the 20 reserved-concurrency slots. Bounded timeouts
# turn a stalled call into a fast, retryable failure. "standard" retry mode adds jitter
# and honours throttling responses, which matters when many concurrent invocations hit
# the same prefix. tcp_keepalive keeps warm-container sockets from being silently dropped.
_BOTO_CONFIG = Config(
    retries={"max_attempts": 3, "mode": "standard"},
    connect_timeout=5,
    read_timeout=30,
    tcp_keepalive=True,
    max_pool_connections=10,
)

s3  = boto3.client("s3", config=_BOTO_CONFIG)
sns = boto3.client("sns", config=_BOTO_CONFIG)
cw  = boto3.client("cloudwatch", config=_BOTO_CONFIG)

SANITISED_BUCKET  = os.environ["SANITISED_BUCKET"]
QUARANTINE_BUCKET = os.environ.get("QUARANTINE_BUCKET", "")
RESULT_TOPIC_ARN  = os.environ.get("RESULT_TOPIC_ARN", "")

_MAX_FILE_BYTES  = int(os.environ.get("CDR_MAX_FILE_BYTES",  str(100 * 1024 * 1024)))
_MAX_ENTRY_BYTES = int(os.environ.get("CDR_MAX_ENTRY_BYTES", str(200 * 1024 * 1024)))
# Aggregate decompression budget for one package; _MAX_ENTRY_BYTES bounds only a single
# entry, so many just-under-cap entries otherwise expand without limit (pitfall #46).
# Sized under the 1024 MB MemorySize rather than above it. Corpus max: 757 KB.
_MAX_TOTAL_ENTRY_BYTES = int(os.environ.get("CDR_MAX_TOTAL_BYTES", str(512 * 1024 * 1024)))
# Corpus max: 12 entries. Real OOXML packages run to a few thousand at the extreme.
_MAX_ZIP_ENTRIES = int(os.environ.get("CDR_MAX_ZIP_ENTRIES", "20000"))
# Image.MAX_IMAGE_PIXELS bounds a single frame; cdr_image materialises every frame at
# once, so total pixels and frame count need their own caps (pitfall #46).
_MAX_TOTAL_IMAGE_PIXELS = int(os.environ.get("CDR_MAX_TOTAL_IMAGE_PIXELS", str(80_000_000)))
_MAX_IMAGE_FRAMES = int(os.environ.get("CDR_MAX_IMAGE_FRAMES", "2000"))
# Explicit decompression-bomb cap, sized to this Lambda's 1024 MB default memory budget
# rather than relying on Pillow's own undocumented default (89.5M soft-warn / 179M
# hard-error) — the soft-warn band between those two lets a large-but-under-the-hard-cap
# image decode+re-encode silently, which at several bytes/pixel/buffer can approach the
# Lambda's memory ceiling well before Pillow's own error would ever fire.
Image.MAX_IMAGE_PIXELS = int(os.environ.get("CDR_MAX_IMAGE_PIXELS", str(40_000_000)))

# Pillow decoder allowlist (pitfall #47). IMAGE_EXTS gates which *extensions* route into
# cdr_image, but Image.open() picks its decoder by sniffing CONTENT, so without this the
# extension constrains only the save format and every plugin in Pillow's registry (43 in
# the pinned build — FLI, PSD, DDS, SGI, XPM, JPEG2000, WMF, EPS…) is reachable by naming
# a file .png. That is a native-code attack surface, not a disarm bypass: output is still
# re-encoded, but the attacker chooses which C decoder parses their bytes, which is the
# reachability precondition for a libtiff/FLI/DDS memory-corruption CVE. Deliberately the
# 7-format set, NOT the single declared extension — extension/content mismatch among these
# seven is common in legitimate files, and rejecting real business documents is a
# production incident. Keep in sync with IMAGE_EXTS.
_PILLOW_FORMATS: list[str] = ["JPEG", "PNG", "GIF", "BMP", "TIFF", "WEBP"]

# ── OOXML (Office) dangerous relationship types ────────────────────────────────
STRIP_REL_TYPES: set[str] = {
    # vbaProject ships under two interchangeable rel-type URIs in the wild: the Microsoft
    # office/2006 form AND the openxmlformats officeDocument/2006 form (LibreOffice and
    # python-docx-authored macro docs emit the latter). Both must be stripped, or the rel
    # dangles at the removed vbaProject.bin part and python-docx/Word reject the doc.
    "http://schemas.microsoft.com/office/2006/relationships/vbaProject",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/vbaProject",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/oleObject",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/externalLink",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/externalLinkPath",
    "http://schemas.microsoft.com/office/2006/relationships/attachedToolbars",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/queryTable",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/connections",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/control",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/attachedTemplate",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/subDocument",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/frame",
    # altChunk imports an arbitrary HTML/MHTML/RTF chunk that the field/macro scrub never
    # inspects (it only scans the host document.xml) — active/remote-content smuggling.
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/aFChunk",
    # Embedded package parts (the relationship form of an embedded OLE/package object).
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/package",
    # Office Web Add-in (task pane / web extension) relationships.
    "http://schemas.microsoft.com/office/2011/relationships/webextensiontaskpanes",
    "http://schemas.microsoft.com/office/2011/relationships/webextension",
    # customXml parts are stripped by STRIP_ZIP_ENTRIES ("customXml/"). Their relationships
    # MUST be dropped here too, or the surviving rel dangles at the deleted part and breaks
    # strict OPC consumers (python-docx, Word) — corrupting otherwise-legitimate documents.
    # customXml is a data-island carrier (custom doc properties / content-control databinding),
    # safe to remove wholesale. Both the 2006 type and the 2010 customXmlProps variant.
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/customXml",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/customXmlProps",
    # ── Dangling-rel siblings of parts dropped by STRIP_ZIP_ENTRIES ───────────────
    # Each of these points at a part the byte-strip removes; if the rel survives, the
    # reconstructed package dangles and strict OPC consumers (python-docx, Word) reject
    # it — the same failure mode as vbaProject/customXml above. (Matching is by LOCAL
    # NAME — see STRIP_REL_LOCALNAMES — so namespace variants are covered automatically;
    # these full URIs are kept for documentation and the membership tests.)
    "http://schemas.microsoft.com/office/2006/relationships/xlMacrosheet",        # xl/macrosheets/
    "http://schemas.microsoft.com/office/2006/relationships/xlIntlMacrosheet",
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/tags",   # ppt/tags/
    "http://schemas.microsoft.com/office/2006/relationships/activeXControlBinary",  # <app>/activeX*.bin
    # Excel long-path (>218 char) external-link / OLE-link rels — [MS-OI29500] §12.4.
    # Note the deliberately mixed year segments (2019/04 vs 2009/04) per the spec.
    "http://schemas.microsoft.com/office/2019/04/relationships/externalLinkLongPath",
    "http://schemas.microsoft.com/office/2019/04/relationships/oleObjectLinkLongPath",
    "http://schemas.microsoft.com/office/2019/04/relationships/xlExternalLinkLongPath/xlStartup",
    "http://schemas.microsoft.com/office/2019/04/relationships/xlExternalLinkLongPath/xlAlternateStartup",
    "http://schemas.microsoft.com/office/2009/04/relationships/xlExternalLinkLongPath/xlPathMissing",
    "http://schemas.microsoft.com/office/2009/04/relationships/xlExternalLinkLongPath/xlLibrary",
}


def _rel_local(rel_type: str) -> str:
    """Reduce an OPC relationship-type URI to its lowercased local name (the final path
    segment). Relationship Types are namespaced URIs, but the SAME logical relationship
    ships under multiple interchangeable namespaces in the wild — transitional
    (``schemas.openxmlformats.org/officeDocument/2006/…``), Microsoft
    (``schemas.microsoft.com/office/{2006,2011,2019/04}/…``), and ISO/IEC-29500 Strict
    (``purl.oclc.org/ooxml/officeDocument/…``, emitted after a Word "Strict Open XML"
    save). Matching on the namespace-independent local name catches all of them with one
    entry instead of enumerating every URI per type — and closes the dual-namespace bug
    class (vbaProject under both ms-office and openxmlformats was the original instance).

    Lowercased so Word's ``aFChunk`` and the ISO-standard ``afChunk`` both normalise to
    one value (altChunk is an active-content import vector — both spellings must strip).
    """
    return rel_type.rsplit("/", 1)[-1].lower()


# Local names matched by _strip_rels (namespace-independent — see _rel_local). Derived
# from the documented full-URI set above, so adding a URI there is enough; the extra
# literals below are local names with no single canonical URI in STRIP_REL_TYPES.
STRIP_REL_LOCALNAMES: set[str] = {_rel_local(t) for t in STRIP_REL_TYPES}

# External hyperlink relationship type. NOT stripped (deleting the rel would dangle the
# document's r:id reference) — instead its Target is rewritten to inert in _strip_rels
# (UNC paths leak NTLM creds; arbitrary URLs enable phishing/SSRF). Matched by local name
# so the transitional/strict namespace variants are all covered.
HYPERLINK_REL_TYPE = (
    "http://schemas.openxmlformats.org/officeDocument/2006/relationships/hyperlink"
)
HYPERLINK_REL_LOCALNAME = _rel_local(HYPERLINK_REL_TYPE)

STRIP_ZIP_ENTRIES: set[str] = {
    "word/vbaProject.bin",
    "xl/vbaProject.bin",
    "ppt/vbaProject.bin",
    "word/activeX",
    "xl/activeX",
    "ppt/activeX",
    "customXml/",
    "word/attachedToolbars/",
    "word/externalLinks/",
    "xl/externalLinks/",
    "xl/macrosheets/",
    "xl/queryTables/",
    "xl/connections.xml",
    "ppt/tags/",
    # Embedded OLE/package objects (oleObject*.bin, renamed .exe/.lnk/.hta, or a nested
    # macro doc). The oleObject relationship is already stripped, but the payload bytes
    # remain extractable unless the part itself is dropped.
    "word/embeddings/",
    "xl/embeddings/",
    "ppt/embeddings/",
    # Office Web Add-ins (task panes / content add-ins live under webextensions/). A
    # webextension auto-loads remote code from an attacker-controlled SourceLocation
    # without any VBA — remote-content execution gated only by tenant/admin policy and
    # consent prompts. Drop the parts (taskpanes.xml + the webextension*.xml definitions).
    "word/webextensions/",
    "xl/webextensions/",
    "ppt/webextensions/",
    # PostScript/EPS image parts (typically under <app>/media/). EPS is a Turing-complete
    # interpreter language and a historic RCE surface — drop the payload bytes; the
    # matching [Content_Types].xml declaration is stripped in _sanitise_content_types.
    ".eps",
    ".ps",
    # altChunk payload parts. The aFChunk RELATIONSHIP is dropped (STRIP_REL_TYPES) and
    # the <w:altChunk> element is neutralised in _strip_xml_macros, but the imported
    # chunk itself — an HTML/MHTML part such as word/afchunk.htm carrying <script>,
    # remote references, or an MHTML-smuggled payload — stayed in the package and rode
    # into SANITISED_BUCKET intact. It is never inspected by the XML scrub (which only
    # ever sees the host part), so anything extracting the package still reaches it.
    # Strip the PART as well as the reference, per the strip-part-and-rel rule.
    ".htm",
    ".html",
    ".xhtml",
    ".mht",
    ".mhtml",
}

OFFICE_EXTS: set[str] = {
    "docx", "docm", "dotx", "dotm",
    "xlsx", "xlsm", "xltx", "xltm", "xlam", "xlsb",
    "pptx", "pptm", "potx", "potm", "ppsx", "ppsm", "ppam",
}

LEGACY_EXTS: set[str] = {"doc", "xls", "ppt"}

IMAGE_EXTS: set[str] = {"jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp"}

# Formats DELIBERATELY rejected — never given a CDR handler. These carry active content
# as loose, parser-divergent structure (embedded/linked OLE, remote-template refs, control
# words a forgiving consumer acts on) rather than as cleanly-excisable content. A
# reconstruction pass can only defend the grammar IT parses; the attacker targets the
# grammar the *consumer* parses. RTF history bears this out: CVE-2017-0199 (remote OLE
# template), CVE-2017-11882 / CVE-2018-0802 (Equation Editor), CVE-2023-21716 (heap
# corruption in the font table). They are quarantined fail-closed by a dedicated handler
# block that runs BEFORE ZIP validation and the unknown-extension gate — listed explicitly
# so the rejection is a documented, order-independent decision, not an accidental gap a
# future contributor "fixes" by adding a handler. See pitfall #38.
FAIL_CLOSED_EXTS: set[str] = {"rtf"}

EXT_REMAP: dict[str, str] = {
    "docm": "docx", "dotm": "dotx",
    "xlsm": "xlsx", "xltm": "xltx", "xlam": "xlsx", "xlsb": "xlsx",
    "pptm": "pptx", "potm": "potx", "ppsm": "ppsx", "ppam": "pptx",
}

# Includes both container-level types and real OPC part-level (*.main+xml) types.
MACRO_CONTENT_TYPE_REMAP: dict[str, str] = {
    "application/vnd.ms-word.document.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    "application/vnd.ms-word.document.macroEnabled.main+xml":
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document.main+xml",
    "application/vnd.ms-word.template.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.wordprocessingml.template",
    "application/vnd.ms-word.template.macroEnabledTemplate.main+xml":
        "application/vnd.openxmlformats-officedocument.wordprocessingml.template.main+xml",
    "application/vnd.ms-excel.sheet.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "application/vnd.ms-excel.sheet.macroEnabled.main+xml":
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet.main+xml",
    "application/vnd.ms-excel.sheet.binary.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "application/vnd.ms-excel.template.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.spreadsheetml.template",
    "application/vnd.ms-excel.template.macroEnabled.main+xml":
        "application/vnd.openxmlformats-officedocument.spreadsheetml.template.main+xml",
    "application/vnd.ms-excel.addin.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "application/vnd.ms-powerpoint.presentation.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.presentationml.presentation",
    "application/vnd.ms-powerpoint.presentation.macroEnabled.main+xml":
        "application/vnd.openxmlformats-officedocument.presentationml.presentation.main+xml",
    "application/vnd.ms-powerpoint.template.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.presentationml.template",
    "application/vnd.ms-powerpoint.template.macroEnabled.main+xml":
        "application/vnd.openxmlformats-officedocument.presentationml.template.main+xml",
    "application/vnd.ms-powerpoint.slideshow.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.presentationml.slideshow",
    "application/vnd.ms-powerpoint.slideshow.macroEnabled.main+xml":
        "application/vnd.openxmlformats-officedocument.presentationml.slideshow.main+xml",
    "application/vnd.ms-powerpoint.addin.macroEnabled.12":
        "application/vnd.openxmlformats-officedocument.presentationml.presentation",
}

# PostScript/EPS content types (lowercased). EPS is a Turing-complete interpreter
# language and a historic RCE surface; a valid OOXML package can legitimately declare
# such a part, so we strip both the [Content_Types].xml declaration and the part bytes.
POSTSCRIPT_CONTENT_TYPES: set[str] = {
    "application/postscript",
    "application/eps",
    "application/x-eps",
    "image/eps",
    "image/x-eps",
}


def _is_postscript_ct(ct: str) -> bool:
    """True if ``ct`` is a PostScript/EPS content type. Normalises case, surrounding
    whitespace, and any RFC-2045 parameter suffix (``;charset=…``) before comparison so
    a declaration like ``application/postscript; charset=utf-8`` cannot evade the set."""
    return ct.split(";", 1)[0].strip().lower() in POSTSCRIPT_CONTENT_TYPES

_ZIP_MAGIC           = b"\x50\x4b\x03\x04"
_SAFE_COMPRESS_METHODS = {0, 8}  # stored, deflate

# ── Pure CDR decision core ──────────────────────────────────────────────────────

# Disposition values returned by cdr_dispatch in result["status"]:
#   "sanitised"          — content disarmed; result["data"] is the clean bytes
#   "rejected"           — ZIP structural anomaly; hard reject (source must be deleted)
#   "unsupported-format" — legacy OLE, fail-closed carrier (RTF), or unknown extension
# cdr_dispatch performs NO I/O (no S3/SNS/CloudWatch). It is the single decision path
# shared by the Lambda handler and any local consumer (e.g. the FastAPI service in
# app.py), so the two can never drift apart on a security decision. Side-effect signals
# the caller may need to act on are returned, not emitted:
#   result["metric"]     — "zip-anomaly" | "passthrough" | None  (caller emits to CloudWatch)
#   result["delete_source"] — bool: whether the source object should be deleted on this path

def cdr_dispatch(data: bytes, ext: str, *, max_file_bytes: Optional[int] = None) -> dict:
    """Pure CDR gate: decide and (if applicable) disarm ``data`` for extension ``ext``.

    Mirrors the routing in ``handler`` exactly — size guard, legacy-OLE reject,
    fail-closed carrier reject, ZIP structural validation, unknown-extension fail-close,
    then format-specific CDR — but performs no I/O. Returns a dict:

        {
          "status": "sanitised" | "rejected" | "unsupported-format",
          "data":   bytes | None,        # clean output (only when sanitised)
          "original_ext":  str,
          "sanitised_ext": str,
          "cdr_mode":   str | None,
          "report":     dict | None,
          "reason":     str | None,      # why rejected/unsupported
          "metric":     str | None,      # "zip-anomaly" | "passthrough"
          "delete_source": bool,
        }

    Raises whatever the format CDR functions raise (e.g. on a corrupt PDF) — the caller
    decides quarantine-and-reraise vs. surface-the-error, matching the Lambda error path.
    """
    ext           = ext.lower()
    sanitised_ext = EXT_REMAP.get(ext, ext)
    limit         = _MAX_FILE_BYTES if max_file_bytes is None else max_file_bytes

    def _result(status, **kw):
        base = {
            "status": status, "data": None, "original_ext": ext,
            "sanitised_ext": sanitised_ext, "cdr_mode": None, "report": None,
            "reason": None, "metric": None, "delete_source": False,
        }
        base.update(kw)
        return base

    # ── Pre-CDR size guard ────────────────────────────────────────────────────
    if len(data) > limit:
        return _result("rejected", reason="file too large", delete_source=False)

    # ── Legacy binary (OLE) formats — unsupported, fail closed ────────────────
    if ext in LEGACY_EXTS:
        return _result("unsupported-format",
                       reason="OLE binary format not supported",
                       metric="passthrough", delete_source=True)

    # ── Deliberately-rejected carriers (RTF) — FAIL CLOSED (highest priority) ──
    if ext in FAIL_CLOSED_EXTS:
        return _result("unsupported-format",
                       reason=f"format rejected by design: {ext}",
                       metric="passthrough", delete_source=True)

    # ── ZIP structural validation (Office only) ───────────────────────────────
    if ext in OFFICE_EXTS:
        valid, zip_anomalies = _validate_zip_structure(data)
        if not valid:
            return _result("rejected", reason=zip_anomalies[0],
                           metric="zip-anomaly", delete_source=True)

    # ── Unknown extension — FAIL CLOSED ───────────────────────────────────────
    if ext not in OFFICE_EXTS and ext != "pdf" and ext not in IMAGE_EXTS:
        return _result("unsupported-format",
                       reason=f"unsupported extension: {ext}",
                       metric="passthrough", delete_source=True)

    # A CdrReject is a deterministic verdict — hard-reject rather than let it escape as
    # an error and burn the EventBridge retry budget on input that fails identically.
    try:
        if ext in OFFICE_EXTS:
            clean_bytes, report = cdr_office(data, ext)
        elif ext == "pdf":
            clean_bytes, report = cdr_pdf(data)
        else:  # one of IMAGE_EXTS, per the fail-closed guard above
            clean_bytes, report = cdr_image(data, ext)
    except CdrReject as exc:
        return _result("rejected", reason=str(exc),
                       metric="zip-anomaly", delete_source=True)

    # An empty payload must never be uploaded labelled "sanitised".
    if not clean_bytes:
        raise ValueError(
            f"CDR produced no output for ext={ext} — refusing to label empty content sanitised"
        )

    return _result("sanitised", data=clean_bytes, report=report,
                   cdr_mode=report.get("cdr_mode", "full"), delete_source=True)


# ── Entry point ────────────────────────────────────────────────────────────────

def handler(event: dict, context) -> dict:
    """Lambda entry point — receives an EventBridge S3 ObjectCreated event."""
    detail = event.get("detail", {})
    try:
        bucket = detail["bucket"]["name"]
        key    = detail["object"]["key"]
    except KeyError as exc:
        logger.error("Malformed EventBridge event — missing field %s: %s", exc, json.dumps(event)[:512])
        raise ValueError(f"Malformed event: missing {exc}") from exc

    size = detail.get("object", {}).get("size", 0)
    logger.info("CDR start: bucket=%s key=%s size=%d", bucket, key, size)

    # Validate key for path traversal (S3 keys are literal but defend against confused callers)
    if ".." in key.split("/"):
        logger.warning("Key contains path traversal segments, rejecting: %s", key)
        _publish_result_safe(bucket, key, "rejected", {"reason": "invalid key"})
        return {"status": "rejected", "reason": "invalid key"}

    # ── Pre-download size guard ───────────────────────────────────────────────
    if size > _MAX_FILE_BYTES:
        logger.warning("File too large: key=%s size=%d max=%d", key, size, _MAX_FILE_BYTES)
        if QUARANTINE_BUCKET:
            try:
                s3.copy_object(
                    CopySource={"Bucket": bucket, "Key": key},
                    Bucket=QUARANTINE_BUCKET,
                    Key=f"oversized/{key}",
                    TaggingDirective="REPLACE",
                    Tagging="cdr-status=rejected&cdr-reason=file-too-large",
                )
            except Exception as exc:
                logger.warning("Could not copy oversized file to quarantine: key=%s error=%s", key, exc)
        _publish_result_safe(bucket, key, "rejected", {"reason": "file too large", "size": size})
        return {"status": "rejected", "reason": "file too large"}

    # ── Download ──────────────────────────────────────────────────────────────
    try:
        file_bytes, content_type = _download(bucket, key)
    except Exception as exc:
        _classify_download_error(bucket, key, exc)
        raise

    # Take the extension from the BASENAME. Splitting the whole key on "." reads a dot in
    # a directory name as an extension ("reports.v2/summary" -> ext "v2/summary"), which
    # fails closed but logs and alarms under a nonsense extension.
    basename      = posixpath.basename(key)
    ext           = basename.rsplit(".", 1)[-1].lower() if "." in basename else ""

    # ── Pure CDR decision core ────────────────────────────────────────────────
    # cdr_dispatch makes every routing/disarm decision with NO I/O; the handler maps
    # its result onto S3/SNS/CloudWatch. The CDR functions can still raise (corrupt PDF
    # etc.) — that is the error path below.
    try:
        decision = cdr_dispatch(file_bytes, ext)
    except Exception as exc:
        logger.exception("CDR processing failed: key=%s error=%s", key, exc)
        if QUARANTINE_BUCKET:
            try:
                _upload(QUARANTINE_BUCKET, f"error/{key}", file_bytes, content_type,
                        {"cdr-status": "error", "cdr-error": str(exc)[:256]})
            except Exception as q_exc:
                logger.warning("Quarantine upload failed: key=%s error=%s", key, q_exc)
        _publish_result_safe(bucket, key, "error", {"error": str(exc), "original_ext": ext})
        raise

    sanitised_ext = decision["sanitised_ext"]

    # Emit the side-effect metric the decision asked for (CloudWatch — caller's job).
    if decision["metric"] == "zip-anomaly":
        _emit_zip_anomaly_metric()
    elif decision["metric"] == "passthrough":
        _emit_passthrough_metric(ext)

    # ── Reject / unsupported routing ──────────────────────────────────────────
    if decision["status"] != "sanitised":
        reason       = decision["reason"]
        status       = decision["status"]
        prefix       = "rejected" if status == "rejected" else "unsupported"
        tag_status   = "rejected" if status == "rejected" else "unsupported-format"
        logger.warning("CDR %s: key=%s ext=%s reason=%s", status, key, ext, reason)
        quarantine_failed = False
        if QUARANTINE_BUCKET:
            try:
                _upload(QUARANTINE_BUCKET, f"{prefix}/{key}", file_bytes, content_type,
                        {"cdr-status": tag_status, "cdr-reason": str(reason)[:256],
                         "cdr-original-ext": ext, "cdr-timestamp": _now()})
            except Exception as q_exc:
                logger.warning("Quarantine upload failed: key=%s error=%s", key, q_exc)
                quarantine_failed = True
        _publish_result_safe(bucket, key, status, {"reason": reason, "original_ext": ext})
        # If quarantine was supposed to hold the only remaining copy of this file and the
        # upload failed, deleting the source would destroy it entirely — never delete in
        # that case (pitfall: never destroy the only copy of unprocessable input).
        if decision["delete_source"] and not quarantine_failed:
            _delete_source_safe(bucket, key)
        return {"status": status, "reason": reason}

    # ── Sanitised output ──────────────────────────────────────────────────────
    clean_bytes           = decision["data"]
    report                = decision["report"]
    dest_key              = _sanitised_key(key, sanitised_ext)
    cdr_mode              = decision["cdr_mode"]
    sanitised_content_type = _content_type_for_ext(sanitised_ext, content_type)
    removal_count         = len(report.get("removed", []))

    _upload(
        SANITISED_BUCKET, dest_key, clean_bytes, sanitised_content_type,
        {
            "cdr-status":       "sanitised",
            "cdr-source":       f"s3://{bucket}/{key}",
            "cdr-timestamp":    _now(),
            "cdr-removals":     str(removal_count),
            "cdr-original-ext": ext,
            # A structural anomaly is a hard reject, so anything reaching this upload had
            # none by construction. (Previously computed from a local that was always [].)
            "cdr-zip-anomaly":  "false",
            "cdr-mode":         cdr_mode,
        },
    )

    logger.info("CDR complete: key=%s dest=%s ext=%s removals=%d mode=%s",
                key, dest_key, ext, removal_count, cdr_mode)

    result_payload = {
        "original_ext":  ext,
        "sanitised_ext": sanitised_ext,
        "cdr_mode":      cdr_mode,
        "zip_anomalies": [],
        "report":        report,
    }
    _publish_result_safe(bucket, key, "sanitised", result_payload)
    _delete_source_safe(bucket, key)

    return {
        "status":      "sanitised",
        "destination": f"s3://{SANITISED_BUCKET}/{dest_key}",
        "report":      result_payload,
    }


# ── Office CDR ─────────────────────────────────────────────────────────────────

def _is_xml_ct(ct: str) -> bool:
    """True when a content type's media subtype is XML. ``+xml`` covers the OOXML family
    (…wordprocessingml.document.main+xml, …drawing+xml, …); ``text/xml`` and
    ``application/xml`` cover the plain forms.

    Any ``;`` parameter is discarded before the suffix test. OPC declares a bare media
    type, so ``…main+xml; charset=utf-8`` is malformed and both LibreOffice and
    python-docx refuse the package outright — the parameter is stripped for robustness,
    not to close a live bypass (see pitfall #58). Without it the trailing parameter moves
    ``+xml`` off the end of the string and the part is never classified as XML."""
    ct = ct.split(";", 1)[0].strip().lower()
    return ct.endswith("+xml") or ct in ("text/xml", "application/xml")


def _declared_parts(ct_xml: bytes, entry_names: Iterable[str],
                    matches_ct) -> set[str]:
    """Return the normalised part names (lower, leading '/' stripped) that
    ``[Content_Types].xml`` declares with a content type satisfying ``matches_ct``.

    OPC offers **two** declaration mechanisms and both are authoritative:

    * ``Override`` binds a content type to one exact PartName, and
    * ``Default`` binds it to *every* part whose extension matches.

    Resolving only ``Override`` (pitfall #54) left the ``Default`` half open: a package
    declaring ``<Default Extension="dat" ContentType="…wordprocessingml.document.main+xml"/>``
    with the officeDocument relationship pointing at ``word/doc.dat`` needs no Override at
    all, and python-docx opened the sanitised output and still saw a live DDEAUTO payload
    (pitfall #55). ``Default`` is resolved against the real ZIP entry names because it binds
    by extension rather than by name.

    Returns an empty set on unparseable XML — the suffix rules remain the backstop."""
    parts: set[str] = set()
    _reject_xml_doctype(ct_xml, "[Content_Types].xml")
    try:
        root = ET.fromstring(ct_xml)
    except ET.ParseError:
        return parts
    ns = "http://schemas.openxmlformats.org/package/2006/content-types"
    default_exts: set[str] = set()
    for child in root:
        if not matches_ct(child.get("ContentType", "")):
            continue
        if child.tag == f"{{{ns}}}Override":
            pn = child.get("PartName", "").replace("\\", "/").lstrip("/").lower()
            if pn:
                parts.add(pn)
        elif child.tag == f"{{{ns}}}Default":
            ext = child.get("Extension", "").strip().lstrip(".").lower()
            if ext:
                default_exts.add(ext)
    if default_exts:
        for name in entry_names:
            norm = name.replace("\\", "/").lstrip("/").lower()
            _, _, ext = norm.rpartition(".")
            if ext and ext in default_exts:
                # Both spellings: callers match this set against the plain lowered name in
                # one place and against the posixpath-canonical form in another, and a
                # part named "./word/doc.dat" must be caught either way.
                parts.add(norm)
                parts.add(posixpath.normpath(norm).lstrip("/"))
    return parts


def _postscript_override_parts(ct_xml: bytes,
                               entry_names: Iterable[str] = ()) -> set[str]:
    """Part names declared with a PostScript/EPS content type in ``[Content_Types].xml``.

    The declaration is the authoritative signal for the part-byte strip: an EPS payload can
    be declared PostScript while stored as ``word/media/image1.png``, which the ``.eps``/
    ``.ps`` suffix rule in STRIP_ZIP_ENTRIES would miss. Covers both declaration forms —
    see ``_declared_parts``."""
    return _declared_parts(ct_xml, entry_names, _is_postscript_ct)


def _xml_override_parts(ct_xml: bytes, entry_names: Iterable[str] = ()) -> set[str]:
    """Part names ``[Content_Types].xml`` declares as XML, whatever their filename suffix.

    A real OOXML consumer locates the main document part by following the package
    relationship and reads it as XML because the *content type* says so — the ``.xml``
    suffix is a convention, not the binding. Covers both declaration forms — see
    ``_declared_parts``."""
    return _declared_parts(ct_xml, entry_names, _is_xml_ct)


# Relationship types (local name) whose Target is a sheet part. `xlBinaryIndex` is the
# type xlsb actually uses for its BIFF12 sheet binaries; the rest are the OOXML sheet
# types, included so a package that binds a .bin under a conventional sheet rel is caught
# too. Matched on local name because the same type ships under both the
# officeDocument/2006 and microsoft.com/office namespaces (mirrors `_strip_rels`).
# Cell-value prefixes Excel treats as the start of a formula. '=' is the only one openpyxl
# turns into a live <f> element in xlsx; the other three matter on CSV export (pitfall #50).
_FORMULA_INJECTION_PREFIXES: tuple[str, ...] = ("=", "+", "-", "@")

_WORKSHEET_REL_TYPES: frozenset[str] = frozenset({
    "xlbinaryindex", "worksheet", "chartsheet", "dialogsheet", "macrosheet",
})


def _xlsb_worksheet_parts(src: zipfile.ZipFile, budget: "_DecompressionBudget") -> set[str]:
    """Return the canonical part names an xlsb's workbook relationships bind to a sheet.

    OPC part names are arbitrary — a real parser (pyxlsb included) follows the ``Target``
    in ``xl/_rels/workbook.bin.rels``, not the conventional ``xl/worksheets/sheetN.bin``
    path. Matching on the conventional name let a relocated sheet binary skip conversion
    (pitfall #49), so resolve the rels instead.

    Deliberately matched on the relationship type's LOCAL NAME (mirrors the rel-type
    matching in ``_strip_rels``): the same worksheet type ships under both the
    ``officeDocument/2006`` and ``microsoft.com/office`` namespaces, and a full-URI compare
    misses one of them. Targets are resolved relative to the rels file's owning directory
    and canonicalised the same way the ``STRIP_ZIP_ENTRIES`` match is, so ``../`` and
    ``//`` cannot dodge the comparison.

    A ``CdrReject`` from ``_read_zip_entry_safe`` (decompression budget trip) is allowed to
    propagate: swallowing it would yield an empty set and fall through to the pass-through
    ZIP path, which is exactly the bypass this function exists to close.
    """
    parts: set[str] = set()
    for item in src.infolist():
        name = item.filename.replace("\\", "/").lower()
        # Any *.rels under a _rels/ directory — not just xl/_rels/workbook.bin.rels, since
        # the workbook part itself is named by the package root rels.
        if not (name.endswith(".rels") and "_rels/" in name):
            continue
        raw = _read_zip_entry_safe(src, item, budget)
        try:
            _reject_xml_doctype(raw, item.filename)
            root = ET.fromstring(raw)
        except (ET.ParseError, CdrReject):
            continue
        base = posixpath.dirname(posixpath.dirname(name))  # strip "_rels/<file>.rels"
        for child in root:
            rel_type = child.get("Type", "").rsplit("/", 1)[-1].lower()
            if rel_type not in _WORKSHEET_REL_TYPES:
                continue
            target = child.get("Target", "").replace("\\", "/")
            if not target or "://" in target:
                continue  # external target — never a local part
            resolved = target.lstrip("/") if target.startswith("/") else posixpath.join(base, target)
            canonical = posixpath.normpath(resolved).lstrip("/").lower()
            if canonical.endswith(".bin"):
                parts.add(canonical)
    return parts


def cdr_office(data: bytes, ext: str) -> tuple[bytes, dict]:
    """Disarm an OOXML (ZIP) Office file by rebuilding the archive entry-by-entry.

    Drops dangerous parts (VBA, ActiveX, embeddings, external links, …), scrubs the
    surviving `.rels`, `[Content_Types].xml`, and XML parts, and dispatches xlsb worksheet
    binaries to ``cdr_xlsb``. Never re-serialises through an Office library, so anything
    not explicitly touched is preserved byte-for-byte.

    Returns ``(clean_bytes, report)`` where report is ``{"format", "removed", "cdr_mode"}``.
    """
    removed: list[str] = []
    out_buf = io.BytesIO()
    budget = _DecompressionBudget()

    with zipfile.ZipFile(io.BytesIO(data), "r") as src:
        # Pre-pass: resolve which parts [Content_Types].xml declares as PostScript, and
        # which it declares as XML, via either OPC declaration form (Override on an exact
        # PartName, or Default on an extension). infolist() order is not guaranteed, so
        # this must run before the main loop to catch such parts wherever they appear.
        # The entry names are passed in because Default binds by extension, not by name.
        ps_parts: set[str] = set()
        xml_parts: set[str] = set()
        entry_names = [i.filename for i in src.infolist()]
        for item in src.infolist():
            if item.filename.replace("\\", "/").lower() == "[content_types].xml":
                try:
                    ct_raw = _read_zip_entry_safe(src, item, budget)
                    ps_parts = _postscript_override_parts(ct_raw, entry_names)
                    xml_parts = _xml_override_parts(ct_raw, entry_names)
                except ValueError:
                    # Bomb-guard trip; suffix rules remain the backstop.
                    ps_parts = set()
                    xml_parts = set()
                break

        # Pitfall #49: resolve xlsb worksheet parts from the workbook relationships, not
        # from a hardcoded "xl/worksheets/sheet*.bin" name. OPC part names are arbitrary —
        # the rel Target is what a real parser follows — so an xlsb whose sheet binary is
        # relocated to e.g. "xl/binparts/data1.bin" skipped conversion entirely and its
        # BIFF12 records (FORMULA / DDE / OLEOBJECT / DEFINEDNAME) passed through
        # byte-for-byte into the sanitised bucket. Resolved *before* the main loop because
        # infolist() order is not guaranteed and the rels part may follow the sheet.
        # A budget trip inside raises CdrReject, which propagates — deliberately NOT caught
        # and converted into "no sheets found", since that would fall through to the ZIP
        # path and pass BIFF12 bytes through unconverted. Fail closed.
        xlsb_sheet_parts: set[str] = (
            _xlsb_worksheet_parts(src, budget) if ext == "xlsb" else set()
        )

        with zipfile.ZipFile(out_buf, "w", compression=zipfile.ZIP_DEFLATED) as dst:
            for item in src.infolist():
                # Normalise backslashes: spec mandates forward slashes
                name_lower = item.filename.replace("\\", "/").lower()

                # Canonical form for the STRIP_ZIP_ENTRIES match only: collapses "./",
                # "//", leading "/", and ".." segments the way OPC part-name / Word's path
                # resolution does, so "./word/vbaProject.bin" or "word//vbaProject.bin"
                # can't dodge the literal-prefix/suffix check while still resolving to the
                # same part at open time.
                name_canonical = posixpath.normpath(name_lower).lstrip("/")

                # 1. Drop entries matching dangerous file/prefix patterns
                skip = False
                for pattern in STRIP_ZIP_ENTRIES:
                    if pattern.endswith("/"):
                        if name_canonical.startswith(pattern.lower()):
                            removed.append(item.filename)
                            skip = True
                            break
                    else:
                        if name_canonical.endswith(pattern.lower()):
                            removed.append(item.filename)
                            skip = True
                            break
                if skip:
                    continue

                # 1b. Drop parts declared PostScript/EPS by an Override (content-type
                #     -driven, regardless of file extension — closes the Override bypass).
                if name_lower in ps_parts:
                    removed.append(item.filename)
                    continue

                # 2. Drop ActiveX control bins/xmls
                if re.search(r"activex/activex\d*\.(bin|xml)$", name_lower):
                    removed.append(item.filename)
                    continue

                # 3. Decompression bomb guard — read in chunks to catch falsified
                #    file_size, and charge the read against the package-wide budget so a
                #    package of individually-legal entries cannot decompress to tens of GB.
                raw = _read_zip_entry_safe(src, item, budget)

                # 4. Sanitise [Content_Types].xml
                if name_lower == "[content_types].xml":
                    raw, ct_removed = _sanitise_content_types(raw)
                    removed.extend(ct_removed)

                # 5. Scrub relationship files
                elif name_lower.endswith(".rels"):
                    raw, rels_removed = _strip_rels(raw)
                    removed.extend(rels_removed)

                # 6. Scrub macro attributes from XML content. `.vml` is included: a VML
                #    drawing part is XML that carries shape definitions with the same
                #    action attributes (onClick/onAction) and field-code carriers the
                #    scrub targets, and it is the classic home of the OLE-object shape.
                #    Matching only ".xml" left every vmlDrawing*.vml part unscrubbed.
                #    The Override-declared set is consulted alongside the suffix because
                #    OPC binds a content type to an exact PartName irrespective of the
                #    filename: a document part stored as "word/document.bin" and declared
                #    wordprocessingml via an Override is still read as XML by every real
                #    consumer, so a suffix-only dispatch left it entirely unscrubbed
                #    (pitfall #54). Checked before the xlsb branch cannot apply, since an
                #    xlsb sheet binary is never declared with an XML content type.
                elif (name_lower.endswith(".xml") or name_lower.endswith(".vml")
                      or name_canonical in xml_parts):
                    raw, xml_removed = _strip_xml_macros(raw, item.filename)
                    removed.extend(xml_removed)

                # 7. xlsb worksheet binary — convert the whole file via cdr_xlsb()
                #    rather than attempting ZIP-level surgery on BIFF12 records. Only
                #    worksheet binaries (xl/worksheets/sheet*.bin) trigger conversion;
                #    other .bin parts (e.g. xl/workbook.bin metadata) must not divert a
                #    VBA-only/metadata-only xlsb away from the normal ZIP CDR path.
                elif ext == "xlsb" and name_canonical in xlsb_sheet_parts:
                    return cdr_xlsb(data)

                dst.writestr(item, raw)

    return out_buf.getvalue(), {"format": ext, "removed": removed, "cdr_mode": "full"}


def cdr_xlsb(data: bytes) -> tuple[bytes, dict]:
    """Convert an xlsb file to clean xlsx by extracting cell values via pyxlsb
    and re-serialising with openpyxl. Strips all BIFF12 formula records, DDE
    references, external links, VBA, and metadata — only plain cell values survive.

    Security notes:
    - String values starting with '=' are forced to plain text to prevent openpyxl
      from serialising them as live formulas in the output xlsx.
    - Each xlsb ZIP entry is read through _read_zip_entry_safe before pyxlsb processes
      the file, enforcing the same decompression-bomb limit as the standard ZIP CDR path.
    """
    removed = ["BIFF12 binary sheet(s) converted to xlsx (all formulas, DDE, and active content stripped)"]

    # Enforce per-entry decompression limit on the xlsb ZIP before handing to pyxlsb.
    # pyxlsb reads entries internally, bypassing _read_zip_entry_safe — pre-reading
    # all entries here ensures a decompression bomb cannot OOM the Lambda.
    _budget = _DecompressionBudget()
    with zipfile.ZipFile(io.BytesIO(data), "r") as _zf:
        for _item in _zf.infolist():
            # Raises CdrReject on the per-entry cap or the aggregate package budget.
            _read_zip_entry_safe(_zf, _item, _budget)

    wb_out = openpyxl.Workbook(write_only=False)
    wb_out.remove(wb_out.active)  # remove default empty sheet

    # Pitfall #49: malformed/hostile BIFF12 makes pyxlsb and openpyxl raise plain
    # KeyError/ValueError/IndexError — an unresolvable sheet rel target, a sheet name with
    # a character openpyxl forbids, an out-of-range column index. Those are deterministic
    # verdicts about the bytes, so translate them into CdrReject; left as generic errors
    # they re-raise and burn the full EventBridge retry budget on input that fails
    # identically every attempt, ending in the DLQ instead of quarantine.
    try:
        _convert_xlsb_sheets(data, wb_out)
    except CdrReject:
        raise
    except (KeyError, ValueError, IndexError, StopIteration, struct.error) as exc:
        raise CdrReject(f"unparseable xlsb workbook: {type(exc).__name__}") from exc

    out_buf = io.BytesIO()
    wb_out.save(out_buf)
    return out_buf.getvalue(), {
        "format": "xlsb",
        "converted_to": "xlsx",
        "removed": removed,
        "cdr_mode": "full",
    }


def _convert_xlsb_sheets(data: bytes, wb_out: "openpyxl.Workbook") -> None:
    """Read every sheet's cell values via pyxlsb and append them to ``wb_out``.

    Split out of ``cdr_xlsb`` so the whole pyxlsb/openpyxl traversal sits behind one
    deterministic-reject boundary (pitfall #49)."""
    with pyxlsb.open_workbook(io.BytesIO(data)) as wb_in:
        for sheet_name in wb_in.sheets:
            ws_out = wb_out.create_sheet(title=sheet_name)
            with wb_in.get_sheet(sheet_name) as ws_in:
                for row in ws_in.rows(sparse=True):
                    if not row:
                        continue
                    max_col = max(cell.c for cell in row)
                    out_row: list = [None] * (max_col + 1)
                    for cell in row:
                        v = cell.v
                        # Force formula-injection prefixes to plain text (pitfall #50).
                        # Only '=' makes openpyxl serialise a live <f> element, so that is
                        # the whole xlsx threat and the rest are inert *here*. They are not
                        # inert downstream: '+', '-' and '@' are the standard CSV-injection
                        # prefixes, and Excel evaluates all four when a sanitised sheet is
                        # exported to CSV and reopened — so a '@SUM'/'+DDE' payload survives
                        # CDR as text and goes live one export later. Leading whitespace is
                        # stripped before the check because Excel tolerates it ahead of the
                        # prefix. Prefixing with an apostrophe is the standard force-text
                        # convention and is idempotent for our purposes.
                        if isinstance(v, str) and v.lstrip("\t\r\n ").startswith(
                            _FORMULA_INJECTION_PREFIXES
                        ):
                            v = "'" + v
                        out_row[cell.c] = v
                    ws_out.append(out_row)


class _DecompressionBudget:
    """Running total of decompressed bytes across every entry of ONE package.

    The per-entry cap alone bounds no aggregate: a package whose entries are each under
    _MAX_ENTRY_BYTES can still decompress to tens of GB in total (see
    _MAX_TOTAL_ENTRY_BYTES). One budget instance is threaded through every read of a
    single archive so the total is enforced, not just the maximum.
    """

    __slots__ = ("used", "limit")

    def __init__(self, limit: int = None):
        self.used = 0
        self.limit = _MAX_TOTAL_ENTRY_BYTES if limit is None else limit

    def consume(self, n: int, entry_name: str) -> None:
        self.used += n
        if self.used > self.limit:
            raise CdrReject(
                f"package exceeds total decompression budget {self.limit} bytes "
                f"(while reading '{entry_name}')"
            )


def _read_zip_entry_safe(src: zipfile.ZipFile, item: zipfile.ZipInfo,
                         budget: Optional["_DecompressionBudget"] = None) -> bytes:
    """Read a ZIP entry in chunks, raising if the actual decompressed size exceeds the limit.
    Does not trust item.file_size which can be falsified in the central directory.

    ``budget``, when supplied, additionally enforces the aggregate across all entries of
    the package — checked per chunk, so an oversized entry is abandoned mid-read rather
    than after it has already been buffered.
    """
    buf = bytearray()  # extend-in-place: a chunk list + b"".join() doubles peak memory,
    total = 0          # 400 MB for one max-size entry inside a 1024 MB container.
    with src.open(item) as f:
        while True:
            chunk = f.read(65536)
            if not chunk:
                break
            total += len(chunk)
            if total > _MAX_ENTRY_BYTES:
                raise CdrReject(
                    f"ZIP entry '{item.filename}' exceeds decompression limit {_MAX_ENTRY_BYTES}"
                )
            if budget is not None:
                budget.consume(len(chunk), item.filename)
            buf.extend(chunk)
    return bytes(buf)


def _strip_rels(data: bytes) -> tuple[bytes, list[str]]:
    """Scrub an OPC ``.rels`` part: drop relationships of dangerous types (VBA, OLE,
    external links, altChunk, …) and rewrite external hyperlink Targets to inert while
    keeping the rel (so the document's ``r:id`` references don't dangle).

    Returns ``(rels_bytes, removed)``; unparseable XML is returned unchanged.
    """
    removed: list[str] = []
    _reject_xml_doctype(data, "(.rels part)")
    try:
        root = ET.fromstring(data)
    except ET.ParseError:
        return data, removed

    ns = "http://schemas.openxmlformats.org/package/2006/relationships"
    to_remove = []
    for child in list(root):
        rel_type = child.get("Type", "")
        local = _rel_local(rel_type)
        if local in STRIP_REL_LOCALNAMES:
            to_remove.append(child)
            removed.append(f"rel:{child.get('Id')} type={local}")
            continue

        # External hyperlinks: neutralise the Target in place rather than deleting the rel
        # (deleting would dangle the document's r:id reference). UNC targets (\\host\share)
        # leak NTLM credentials; arbitrary URLs enable phishing/SSRF on click.
        if (
            local == HYPERLINK_REL_LOCALNAME
            and child.get("TargetMode") == "External"
            and child.get("Target")
        ):
            child.set("Target", "https://_CDR_REMOVED_/")
            removed.append(f"rel:{child.get('Id')} external hyperlink target neutralised")

    for child in to_remove:
        root.remove(child)

    ET.register_namespace("", ns)
    return ET.tostring(root, encoding="unicode", xml_declaration=False).encode(), removed


def _sanitise_content_types(data: bytes) -> tuple[bytes, list[str]]:
    """Rewrite ``[Content_Types].xml``: remap macro-enabled OPC part content types to
    their clean equivalents (both the ``*.main+xml`` Override and ``.12`` Default forms)
    and drop VBA/ActiveX Default entries.

    Returns ``(content_types_bytes, removed)``; unparseable XML is returned unchanged.
    """
    removed: list[str] = []
    _reject_xml_doctype(data, "[Content_Types].xml")
    try:
        root = ET.fromstring(data)
    except ET.ParseError:
        return data, removed

    ns = "http://schemas.openxmlformats.org/package/2006/content-types"
    ET.register_namespace("", ns)

    to_remove = []
    for child in list(root):
        ct = child.get("ContentType", "")

        if ct in MACRO_CONTENT_TYPE_REMAP:
            child.set("ContentType", MACRO_CONTENT_TYPE_REMAP[ct])
            removed.append(f"content-type replaced: {ct.split('.')[-1]}")

        elif child.tag == f"{{{ns}}}Default" and (
            "vba" in ct.lower() or "activex" in ct.lower()
        ):
            to_remove.append(child)
            removed.append(f"content-type removed: {ct}")

        elif child.tag == f"{{{ns}}}Override" and (
            "activex" in ct.lower() or "vba" in ct.lower()
        ):
            # An Override pinning a content type to /word/vbaProject.bin dangles once the
            # part is stripped (some authors emit one alongside the Default Extension="bin"
            # entry). Drop it for the same reason as the Default vba branch above.
            to_remove.append(child)
            removed.append(f"content-type removed: {ct}")

        # PostScript/EPS parts declare a Turing-complete interpreter language (the
        # GhostScript -dSAFER bypass family is real history). Drop the declaration in
        # both Default and Override forms. Part bytes are dropped by STRIP_ZIP_ENTRIES
        # (suffix) AND by the content-type-driven pre-pass in cdr_office (which catches a
        # PostScript Override on an arbitrarily-named part, e.g. /word/media/image1.png).
        elif _is_postscript_ct(ct):
            to_remove.append(child)
            removed.append(f"content-type removed: {ct}")

    for child in to_remove:
        root.remove(child)

    return ET.tostring(root, encoding="unicode", xml_declaration=False).encode(), removed


# Field-code keywords that are dangerous regardless of arguments. Used to decide whether
# a run of entity-encoded characters decodes to a threat that must be neutralised.
_ENCODED_FIELD_KEYWORDS = (
    "AUTOOPEN", "AUTOEXIT", "AUTOCLOSE", "AUTONEW", "MACROBUTTON", "DDEAUTO", "DDE",
    "EXEC", "INCLUDETEXT", "INCLUDEPICTURE", "INCLUDE", "WEBSERVICE", "HYPERLINK",
    "RTD", "CALL", "REGISTER", "LINK",
)
# Word-boundary matcher for the above: plain `kw in decoded_text` is a substring check and
# false-positives on benign words that merely contain a keyword (e.g. "PADDED" contains
# "DDE", "UNLINKING" contains "LINK", "RECALL" contains "CALL") — corrupting legitimate
# text that was never a field code.
_ENCODED_FIELD_KEYWORDS_RE = re.compile(
    r'\b(?:' + '|'.join(_ENCODED_FIELD_KEYWORDS) + r')\b'
)

# A run of ASCII letters and/or numeric character references, e.g. &#68;&#68;&#69; or
# &#77;ACROBUTTON (partial encoding). Such a run only ever spans plain text — never markup
# — so neutralising it cannot corrupt XML structure or disturb legitimate single escapes
# like &amp; (which decodes to '&', not a letter, so it is not absorbed into a run).
#
# A SINGLE quantified alternation `(?:A|B)+` is used deliberately: the earlier form with
# two unbounded `(?:...)*` groups around a required token had quadratic backtracking
# (a long letter run with no '&#...;' made .sub() retry from every index → ReDoS). The
# `_replace` callback applies the "must contain an actual numeric reference" condition
# instead of the regex, so correctness is preserved without the backtracking shape.
_NUMERIC_ENTITY_RUN = re.compile(r'(?:[A-Za-z]|&#[xX]?[0-9a-fA-F]+;)+')


def _neutralise_encoded_field_codes(text: str) -> tuple[str, int]:
    """Strip entity-encoded field-code keywords without decoding the whole document.

    Returns (text, count). Only runs that actually contain a numeric character reference
    AND decode to a dangerous keyword are replaced; everything else — including benign
    single escapes such as ``&amp;`` and plain words — is left byte-for-byte unchanged.
    """
    if "&#" not in text:  # cheap short-circuit: no numeric references → nothing to do
        return text, 0
    count = 0

    def _replace(match: "re.Match[str]") -> str:
        nonlocal count
        run = match.group(0)
        if "&#" not in run:  # a pure-letter run is never an encoded keyword
            return run
        if _ENCODED_FIELD_KEYWORDS_RE.search(html.unescape(run).upper()):
            count += 1
            return "_CDR_REMOVED_"
        return run

    return _NUMERIC_ENTITY_RUN.sub(_replace, text), count


# Auto-execute field names that carry no argument (AUTOOPEN fires on document open, etc.).
_FIELD_AUTOEXEC_RE = re.compile(r'\b(AUTOOPEN|AUTOEXIT|AUTOCLOSE|AUTONEW)\b', re.IGNORECASE)

# Keyword + argument field codes. Neutralise the argument, preserving the keyword so the
# field structure stays well-formed. INCLUDE*/LINK fetch external resources (SSRF/exfil);
# WEBSERVICE auto-fetches a URL on open; HYPERLINK UNC paths trigger NTLM credential theft.
# `[^\s<>]+` (not `\S+`) bounds the argument so a match can never cross an XML tag boundary
# even if a field carrier somehow contained one.
_FIELD_KEYWORD_ARG_RE = re.compile(
    r'\b(MACROBUTTON|DDE|DDEAUTO|AUTO|EXEC|INCLUDE|INCLUDETEXT|INCLUDEPICTURE|LINK'
    r'|WEBSERVICE|HYPERLINK)\s+[^\s<>]+',
    re.IGNORECASE,
)

# Field-code CARRIERS: the only places Word evaluates a field code.
#   <w:instrText …>…</w:instrText>  — the run that holds the field instruction text
#   <w:fldSimple … w:instr="…" …>   — the simple-field instruction attribute
# Namespace prefix is matched loosely ([A-Za-z0-9]*:) so non-"w" prefixes are still covered.
#
# _INSTRTEXT_OPEN_RE/_INSTRTEXT_CLOSE_RE are matched by a manual linear scanner
# (_iter_instrtext_elements), NOT combined into one open-body-close regex with a `.*?`
# body group: a `.*?` body scans to end-of-string on every opening tag that has no
# matching close, which is O(n) work per occurrence — O(n^2) total on a document.xml
# crafted with many unmatched "<w:instrText ...>" opens (Lambda-timeout DoS well within
# the 200MB per-entry decompression cap). The manual scanner finds each close with a
# single forward str.find/search per open, giving linear total work.
_INSTRTEXT_OPEN_RE = re.compile(r'<[A-Za-z0-9]*:?instrText\b[^>]*>', re.IGNORECASE)
_INSTRTEXT_CLOSE_RE = re.compile(r'</[A-Za-z0-9]*:?instrText\s*>', re.IGNORECASE)
_FLDSIMPLE_INSTR_RE = re.compile(
    r'(\b[A-Za-z0-9]*:?instr=")([^"]*)(")',
    re.IGNORECASE,
)


def _scrub_field_string(s: str) -> tuple[str, int]:
    """Neutralise auto-exec + keyword-argument field codes inside a single field-code
    string (the content of an instrText element or a fldSimple instr attribute). Returns
    (scrubbed, count). Operates only on field-code text, so it can never corrupt markup."""
    s, n1 = _FIELD_AUTOEXEC_RE.subn(r'\1_CDR_REMOVED_', s)
    s, n2 = _FIELD_KEYWORD_ARG_RE.subn(r'\1 _CDR_REMOVED_', s)
    return s, n1 + n2


def _scrub_instrtext_elements(text: str) -> tuple[str, int]:
    """Linear-time equivalent of a single
    ``(<w:instrText…>)(.*?)(</w:instrText>)`` substitution: for each opening tag found,
    look for the next closing tag starting only from the end of that opening tag (never
    re-scanning text already consumed by a prior element), so total work is O(n) instead
    of the O(n^2) a lazy DOTALL body group produces on many unmatched opening tags."""
    total = 0
    out: list[str] = []
    pos = 0
    for open_m in _INSTRTEXT_OPEN_RE.finditer(text):
        if open_m.start() < pos:
            continue  # inside a previously-consumed element
        out.append(text[pos:open_m.start()])
        close_m = _INSTRTEXT_CLOSE_RE.search(text, open_m.end())
        if close_m is None:
            # Unterminated element: nothing more can be a instrText body; stop scrubbing.
            out.append(text[open_m.start():])
            pos = len(text)
            break
        body, n = _scrub_field_string(text[open_m.end():close_m.start()])
        total += n
        out.append(open_m.group(0))
        out.append(body)
        out.append(close_m.group(0))
        pos = close_m.end()
    out.append(text[pos:])
    return "".join(out), total


def _scrub_field_code_carriers(text: str) -> tuple[str, int]:
    """Run the auto-exec / keyword-argument field-code scrub ONLY inside field carriers
    (<w:instrText> content and <w:fldSimple w:instr="…">), leaving all other XML untouched.
    Returns (text, total_count)."""
    total = 0

    def _fldsimple_attr(m: "re.Match[str]") -> str:
        nonlocal total
        val, n = _scrub_field_string(m.group(2))
        total += n
        return m.group(1) + val + m.group(3)

    text, n1 = _scrub_instrtext_elements(text)
    text = _FLDSIMPLE_INSTR_RE.sub(_fldsimple_attr, text)
    return text, n1 + total


class CdrReject(Exception):
    """A deterministic, unprocessable-input verdict raised from inside a CDR function.

    ``cdr_dispatch`` turns this into a ``rejected`` disposition (quarantine + delete
    source), NOT into the error path. The distinction matters under load: an error
    re-raises so EventBridge retries — correct for a possibly-transient fault, but pure
    waste for input that will fail identically on every attempt, and it multiplies an
    attacker's cost-per-upload by the retry count. Structural verdicts that can never
    succeed on retry belong here.
    """


# XML byte-order marks. A part in any of these encodings is valid XML that Word/Excel
# parse happily, but every neutralisation pass in this module is written against the
# decoded text — decoding UTF-16 as UTF-8 turns "DDEAUTO" into "D\x00D\x00E\x00…", which
# no keyword regex matches, so a UTF-16 document.xml sails through every pass untouched.
# Decode with the real encoding, scrub, and re-encode with the SAME encoding (BOM
# included) so the declaration stays truthful and the file stays byte-faithful.
_XML_BOMS: tuple[tuple[bytes, str], ...] = (
    (b"\x00\x00\xfe\xff", "utf-32-be"),
    (b"\xff\xfe\x00\x00", "utf-32-le"),
    (b"\xef\xbb\xbf",     "utf-8-sig"),
    (b"\xfe\xff",         "utf-16-be"),
    (b"\xff\xfe",         "utf-16-le"),
)


def _decode_xml_part(data: bytes):
    """Decode an XML part to text, returning ``(text, encode_fn)`` where ``encode_fn``
    re-encodes scrubbed text back to the part's original encoding.

    Never lossy. The previous ``decode("utf-8", errors="replace")`` silently rewrote every
    byte it could not decode to U+FFFD and then wrote that back out — so a legitimate
    latin-1-encoded part came out of CDR with its accented characters destroyed (real
    corruption of a clean document), while a UTF-16 part evaded the scrub entirely.
    """
    for bom, enc in _XML_BOMS:
        if data.startswith(bom):
            body = data[len(bom):] if enc != "utf-8-sig" else data
            try:
                text = body.decode(enc)
            except UnicodeDecodeError:
                break
            if enc == "utf-8-sig":
                return text, lambda s: s.encode("utf-8-sig")
            return text, lambda s, e=enc, b=bom: b + s.encode(e)

    try:
        return data.decode("utf-8"), lambda s: s.encode("utf-8")
    except UnicodeDecodeError:
        # Not UTF-8 and no BOM — most likely a legacy single-byte encoding. latin-1 maps
        # every byte 1:1 to U+0000–U+00FF, so decode→scrub→encode round-trips byte-exactly
        # for anything the ASCII-only patterns don't match, and the ASCII keywords the
        # patterns DO look for are still visible. Lossless where "replace" was destructive.
        return data.decode("latin-1"), lambda s: s.encode("latin-1", errors="replace")


# ECMA-376 Part 2 forbids a DTD in any OPC package part, and Word/Excel reject one. An
# XML part carrying <!DOCTYPE ...> is therefore never legitimate, and it is the entry
# point for entity-expansion amplification (billion laughs / quadratic blowup) against
# the ElementTree parses in _strip_rels and _sanitise_content_types — verified locally:
# a nested-entity [Content_Types].xml expands ~1000x through ET.fromstring today.
# Reject the package rather than parse it.
_XML_DOCTYPE_RE = re.compile(rb"<!DOCTYPE", re.IGNORECASE)


def _reject_xml_doctype(data: bytes, filename: str) -> None:
    """Hard-reject an OPC XML part containing a DTD (spec-forbidden; entity-expansion
    amplification vector). Only the prolog can hold one, so only the head is scanned."""
    if _XML_DOCTYPE_RE.search(data[:4096]):
        raise CdrReject(f"XML part '{filename}' contains a DTD (forbidden in OPC packages)")


def _strip_xml_macros(data: bytes, filename: str) -> tuple[bytes, list[str]]:
    """Neutralise dangerous content inside an Office XML part via text-level regex passes.

    Covers: entity-encoded field codes (e.g. ``&#68;&#68;&#69;`` = DDE), auto-exec and
    keyword+argument field codes (AUTOOPEN, DDE, WEBSERVICE, …), ``onClick``/``onAction``
    action attributes, the DDE pipe form (``cmd|'/c calc'!A1``), and ``<w:altChunk>``
    imports. Matches in place (replacing with ``_CDR_REMOVED_``) so XML structure and
    benign escapes like ``&amp;`` survive — it never decodes the whole part.

    Returns ``(xml_bytes, removed)``.
    """
    removed: list[str] = []
    _reject_xml_doctype(data, filename)
    text, _encode = _decode_xml_part(data)

    # Neutralise entity-encoded field codes (e.g. &#68;&#68;&#69; = "DDE") WITHOUT
    # decoding the whole document. Word entity-decodes content before evaluating field
    # codes, so an attacker can hide a keyword inside numeric character references. A
    # numeric character reference only ever encodes a single text character — never XML
    # markup — so collapsing a run of them and checking whether it decodes to a dangerous
    # keyword lets us strip the threat while leaving legitimate escapes like &amp; intact.
    text, n_ent = _neutralise_encoded_field_codes(text)
    if n_ent:
        removed.append(f"{filename}: {n_ent} entity-encoded field code(s)")

    # Auto-exec + keyword-argument field codes are neutralised ONLY inside the parts of the
    # XML where Word actually evaluates field codes — never over the raw part. A field code
    # lives in a `<w:instrText>…</w:instrText>` run, or in the `w:instr="…"` attribute of a
    # `<w:fldSimple>`. Scanning the whole part (the prior approach) matched keywords like
    # LINK/AUTO/HYPERLINK inside legitimate element/attribute NAMES, and the `\S+` swallowed
    # XML markup, emitting a value-less `_CDR_REMOVED_` token → invalid XML that strict
    # consumers (python-docx/Word) reject (false positives in python-docx-authored styles.xml).
    # Scoping to the field carriers keeps detection on real field codes while never touching
    # markup. (Excel cell-formula field forms like DDE/WEBSERVICE(...) and the DDE pipe form
    # are still handled by the n3 pass below, which is XML-boundary-safe.)
    cleaned, n = _scrub_field_code_carriers(text)
    if n:
        removed.append(f"{filename}: {n} dangerous field code(s)")

    cleaned, n2 = re.subn(
        r'''\b(onClick|onAction|r:link|r:action)=(?:"[^"]*"|'[^']*')''',
        '',
        cleaned,
    )
    if n2:
        removed.append(f"{filename}: {n2} action attribute(s)")

    # altChunk imports an external HTML/MHTML/RTF chunk into the document; the imported
    # content bypasses every field/macro scrub (which only sees the host part). The
    # aFChunk relationship is dropped in STRIP_REL_TYPES, but also rename the element so
    # the import cannot fire even if a rel survives. Handles both <w:altChunk .../> and
    # paired <w:altChunk>…</w:altChunk> by neutralising the tag name.
    cleaned, n_alt = re.subn(
        r'(<|</)\s*[A-Za-z0-9]*:?altChunk\b',
        r'\1_CDR_REMOVED_altChunk',
        cleaned,
        flags=re.IGNORECASE,
    )
    if n_alt:
        removed.append(f"{filename}: {n_alt} altChunk import(s) neutralised")

    # DDE pipe pattern allows spaces in the argument (e.g. cmd|' /c calc'!A1). The app
    # name before the pipe may be bare, double-quoted, single-quoted, or bracketed —
    # Excel accepts "cmd"|, 'cmd'|, and [cmd]| forms, so all must be neutralised.
    #
    # Quantifiers are BOUNDED ({1,64} / {1,256}) to prevent quadratic backtracking: an
    # unbounded app-name alternative followed by unbounded [^!]+ with no '!suffix' made
    # the engine retry O(n^2) on a long `a…|b…` string → ReDoS. The argument also excludes
    # '<' and '>' so a match cannot span an XML tag boundary and corrupt structure.
    #
    # The post-'!' item is anchored to an EXCEL CELL REFERENCE shape (A1 / $B$2 / R1C1)
    # rather than any word — a real DDE link target is always a cell ref. This avoids the
    # false positive on benign prose like "Profit|Loss!Important" (which would otherwise be
    # corrupted to an unbalanced "_CDR_REMOVED_(").
    cleaned, n3 = re.subn(
        r"\b(WEBSERVICE|RTD|HYPERLINK|DDE|DDEAUTO|CALL|REGISTER)\s*\("
        r'|(?:"[^"]{1,256}"|\'[^\']{1,256}\'|\[[^\]]{1,256}\]|[^\s\[\]"\'|<>]{1,64})'
        r'\|[^!<>]{1,256}!(?:\$?[A-Za-z]{1,3}\$?[0-9]{1,7}|R[0-9]{1,7}C[0-9]{1,7})\b',
        '_CDR_REMOVED_',
        cleaned,
        flags=re.IGNORECASE,
    )
    if n3:
        removed.append(f"{filename}: {n3} dangerous formula/DDE reference(s)")

    return _encode(cleaned), removed


# ── PDF CDR ────────────────────────────────────────────────────────────────────

def cdr_pdf(data: bytes) -> tuple[bytes, dict]:
    """Disarm a PDF with pikepdf: remove catalog-level active content (``/OpenAction``,
    ``/AA``, JavaScript, ``/Names./EmbeddedFiles``), sweep AcroForm fields and root, walk
    pages (all annotation ``/A``/``/AA``, ``/FileAttachment`` file specs, multimedia
    subtypes) and the outline tree, drop article threads, external stream file refs and
    malformed Type3 ``/FontMatrix`` values, then re-serialise via ``pdf.save()`` — which
    also drops any appended/polyglot bytes.

    Returns ``(clean_bytes, {"format": "pdf", "removed": [...]})``.
    """
    removed: list[str] = []

    # Magic check before QPDF sees the bytes (pitfall #47). The Office path validates
    # magic + structure before zlib touches anything; PDF handed raw bytes straight to
    # QPDF's C++ parser behind only the size cap. The spec allows the header within the
    # first 1024 bytes, so match that rather than requiring offset 0 — the subsequent
    # pdf.save() rebuild drops any leading polyglot prefix anyway.
    if b"%PDF-" not in data[:1024]:
        raise CdrReject("no %PDF- header in the first 1024 bytes")

    with pikepdf.open(io.BytesIO(data)) as pdf:
        catalog = pdf.Root

        for dangerous_key in ("/OpenAction", "/AA", "/JavaScript", "/JS", "/Names", "/AcroForm"):
            if dangerous_key in catalog:
                if dangerous_key == "/AcroForm":
                    acroform = catalog["/AcroForm"]
                    # /AA (Additional Actions) on the AcroForm root fires JavaScript on
                    # form calculate/focus — strip it here; the catalog-level /AA loop
                    # above only covers catalog["/AA"], not the AcroForm dict's own.
                    for action_key in ("/XFA", "/CO", "/AA"):
                        if action_key in acroform:
                            del acroform[action_key]
                            removed.append(f"AcroForm{action_key}")
                    removed.extend(_strip_acroform_fields(acroform.get("/Fields", [])))
                elif dangerous_key == "/Names":
                    _DANGEROUS_NAMES = {"/JavaScript", "/EmbeddedFiles", "/AlternatePresentations"}
                    names_dict = catalog["/Names"]
                    for name_key in _DANGEROUS_NAMES:
                        if name_key in names_dict:
                            del names_dict[name_key]
                            removed.append(f"/Names.{name_key}")
                else:
                    del catalog[dangerous_key]
                    removed.append(dangerous_key)

        for page_num, page in enumerate(pdf.pages):
            removed.extend(_strip_pdf_page(page, page_num))

        removed.extend(_strip_pdf_outlines(catalog))
        removed.extend(_strip_pdf_threads(catalog))
        removed.extend(_neutralise_pdf_risky_image_filters(pdf))
        removed.extend(_strip_pdf_external_stream_refs(pdf))
        removed.extend(_strip_pdf_fontmatrix(pdf))

        if "/Metadata" in catalog:
            del catalog["/Metadata"]
            removed.append("/Metadata")
        try:
            if pdf.docinfo:
                pdf.docinfo.clear()
                removed.append("/Info")
        except Exception as exc:
            logger.warning("Could not clear PDF docinfo: %s", exc)

        out_buf = io.BytesIO()
        pdf.save(out_buf)

    return out_buf.getvalue(), {"format": "pdf", "removed": removed}


# Image compression filters with a history of decoder-level RCE: JBIG2 (CVE-2009-0658 and
# the FORCEDENTRY CVE-2021-30860 family) and JPEG2000 / JPXDecode. These are not "active
# content" the action sweep catches — the exploit is in the *decoder* parsing a crafted
# stream. pdfid flags /JBIG2Decode for exactly this reason. We can't safely transcode the
# image without invoking the very decoder we distrust, so we neutralise the stream: replace
# its bytes with a 1×1 inert image and drop the dangerous filter. The page keeps a (blank)
# image in that slot; the malicious payload never reaches a viewer's JBIG2/JPX decoder.
_RISKY_IMAGE_FILTERS = {"/JBIG2Decode", "/JPXDecode"}


def _filter_names(filt) -> list[str]:
    """Normalise a stream's ``/Filter`` (a Name, an Array of Names, or — malformed — some
    other type) to a list of name strings. pikepdf resolves indirect refs and normalises
    name tokens (so ``/JBIG2#44ecode`` → ``/JBIG2Decode``) before we see them."""
    if filt is None:
        return []
    if isinstance(filt, pikepdf.Name):
        return [str(filt)]
    try:
        return [str(x) for x in filt]
    except TypeError:
        # A non-Name, non-iterable /Filter is spec-violating. Surface it as a sentinel so
        # the caller fails closed rather than silently treating the stream as clean.
        return ["<unparsable-filter>"]


def _neutralise_pdf_risky_image_filters(pdf) -> list[str]:
    """Neutralise every use of a decoder-RCE-prone image filter (JBIG2 / JPX).

    Covers BOTH places such a filter can live:
      * **Stream objects** (image XObjects, form XObjects, /SMask sub-images, appearance
        streams) — rewritten to a 1×1 inert image with the filter dropped.
      * **Inline images** (``BI … /F /JBIG2Decode … ID <bytes> EI``) inside *any* reachable
        content stream — these live in operator tokens, not stream objects, so they are
        invisible to the sweep above; a PDF carrying one is **hard-rejected** (raises),
        because inline JBIG2/JPX is spec-violating and vanishingly rare in legitimate
        files, so failing closed is correct and won't false-positive.

    FAIL CLOSED: if a stream is identified as risky (or its ``/Filter`` can't be parsed)
    but neutralisation throws, we RE-RAISE rather than swallow — a silent fail-open that
    leaves a risky stream in a file labelled "sanitised" is the worst outcome for a CDR
    control. ``cdr_pdf``'s caller then quarantines the whole PDF (handler error path)."""
    removed: list[str] = []

    # ── (a) stream objects ────────────────────────────────────────────────────
    # Walk the graph rather than pdf.objects. A PDF stream is *always* an indirect object
    # (the spec requires it — a stream's data is addressed via the xref), so enumerating
    # the object table happens to reach every one of them today, and probing confirmed it.
    # But that is a property of pikepdf's parser, not an invariant this sweep should rest
    # on: a lenient parser accepting a malformed direct stream, or a future pikepdf that
    # surfaces one, silently reopens the #52 hole. _walk_pdf_nodes seeds from pdf.objects
    # as well, so this is a strict superset — orphans are still covered.
    for obj in _walk_pdf_nodes(pdf):
        if not isinstance(obj, pikepdf.Stream):
            continue
        try:
            names = _filter_names(obj.get("/Filter"))
        except Exception as exc:
            # Could not even read /Filter on this stream — can't prove it's safe.
            raise ValueError(f"unreadable /Filter during risky-image sweep: {exc}") from exc
        risky = _RISKY_IMAGE_FILTERS.intersection(names)
        if "<unparsable-filter>" in names:
            raise ValueError("stream with unparsable /Filter — failing closed")
        if not risky:
            continue
        # Identified as risky — from here a failure must NOT be swallowed.
        obj.write(b"\x00")
        obj.Width = 1
        obj.Height = 1
        obj.BitsPerComponent = 1
        obj.ColorSpace = pikepdf.Name("/DeviceGray")
        for k in ("/Filter", "/DecodeParms", "/SMask", "/Decode"):
            if k in obj:
                del obj[k]
        removed.append(f"risky image filter {','.join(sorted(risky))} (neutralised)")

    # ── (b) inline images in content streams ──────────────────────────────────
    removed.extend(_reject_inline_risky_images(pdf))
    return removed


def _reject_inline_risky_images(pdf) -> list[str]:
    """Scan every reachable stream for an inline image declaring a risky filter. Such an
    image bypasses the object-level sweep entirely (it lives in operator tokens, not a
    stream object) and survives ``pdf.save()``. Hard-reject the PDF if one is found — fail
    closed. Walks the whole object graph rather than enumerating known containers: see the
    note at the loop below for the six container shapes that enumeration missed."""
    def _check(obj, where: str) -> None:
        try:
            ops = pikepdf.parse_content_stream(obj)
        except Exception:
            return  # not a content stream / unparseable — object sweep covers real streams
        for operands, operator in ops:
            # Now that every reachable stream is scanned, `ops` can come from a stream that
            # merely *looked* parseable — an embedded file, a font program — whose operator
            # tokens are arbitrary binary. str() on those raises UnicodeDecodeError, which
            # must not abort the sweep. Only the risky-filter raise below may propagate.
            try:
                if str(operator) != "INLINE IMAGE":
                    continue
                filter_lists = [
                    [str(f) for f in getattr(operand, "filters", [])]
                    for operand in operands
                ]
            except Exception:
                continue
            for filters in filter_lists:
                if _RISKY_IMAGE_FILTERS.intersection(filters):
                    raise ValueError(
                        f"inline image with decoder-RCE filter "
                        f"{sorted(_RISKY_IMAGE_FILTERS.intersection(filters))} in {where} "
                        f"— rejected (cannot be safely neutralised in place)"
                    )

    # Pages first, so the common case reports a page-shaped location.
    for page_num, page in enumerate(pdf.pages):
        _check(page, f"page[{page_num}] content stream")

    # Then EVERY reachable stream. Enumerating specific containers (form XObjects one
    # level deep, /AP appearances) missed inline images in nested form XObjects, Type3
    # /CharProcs glyph procedures, tiling /Pattern streams and ExtGState /SMask /G groups
    # — all content streams, none reachable through the old page→/XObject→/AP traversal.
    # Gating on /Subtype == /Form was the same attacker-controlled-discriminator mistake as
    # pitfall #51a: dropping /Subtype walked the payload straight through. parse_content_
    # stream simply fails on a non-content stream, so scanning everything costs nothing.
    for obj in _walk_pdf_nodes(pdf):
        if isinstance(obj, pikepdf.Stream):
            _check(obj, f"stream {obj.objgen}")
    return []


# pdf.objects enumerates only *indirect* objects. A dictionary written inline — e.g. a
# Type3 font sitting directly in /Resources /Font /F1 rather than behind an object
# reference — is never visited by it, so any sweep built on pdf.objects alone is bypassed
# by the attacker simply not making their payload indirect (pitfall #52). Walk the
# reachable graph instead: the object *tree*, not the object *table*.
_PDF_WALK_MAX_NODES = 500_000


def _walk_pdf_nodes(pdf):
    """Yield every dictionary and stream reachable from the document root.

    Covers direct (inline) objects, which ``pdf.objects`` omits, as well as every indirect
    object still referenced from the catalog. Indirect objects are additionally seeded from
    ``pdf.objects`` so an orphan that is not reachable from the root — but is still present
    in the file, and so still parsed by a lenient viewer — is not skipped either.

    Cycles are broken with a seen-set keyed on ``objgen`` for indirect objects and on the
    node's *path from the root* for direct ones. ``id()`` must not be used as that key:
    pikepdf hands back a fresh Python wrapper on every access, so a temporary wrapper can
    be collected and its address reused by the next one, which would make an unvisited node
    collide with a ``seen`` entry and be silently skipped. ``_PDF_WALK_MAX_NODES`` bounds a
    hostile file that nests direct dictionaries deeply enough to make the walk itself the
    denial of service; exceeding it raises ``CdrReject`` rather than truncating, because a
    partial walk silently narrows every sweep built on this generator (pitfall #59)."""
    seen: set = set()
    stack: list = []

    def key_of(node, path):
        objgen = getattr(node, "objgen", None)
        # (0, 0) is pikepdf's sentinel for a direct object: not a real object number, and
        # shared by every direct object in the file — so key those by path instead.
        if objgen and objgen != (0, 0):
            return ("i", objgen)
        return ("d", path)

    def push(node, path):
        if isinstance(node, (pikepdf.Dictionary, pikepdf.Array, pikepdf.Stream)):
            stack.append((node, path))

    try:
        push(pdf.Root, "/Root")
    except Exception as exc:
        logger.debug("Could not seed /Root for PDF walk: %s", exc)
    # Materialise the page tree before seeding: pikepdf builds page objects lazily, and an
    # inline /Resources dictionary under a page that has never been touched is not yet
    # reachable through the handles the walk sees.
    try:
        for page_index, page in enumerate(pdf.pages):
            push(page.obj, f"/Page[{page_index}]")
    except Exception as exc:
        logger.debug("Could not seed pages for PDF walk: %s", exc)
    for obj_index, obj in enumerate(pdf.objects):
        push(obj, f"/Objects[{obj_index}]")

    visited = 0
    while stack:
        node, path = stack.pop()
        node_key = key_of(node, path)
        if node_key in seen:
            continue
        seen.add(node_key)
        visited += 1
        if visited > _PDF_WALK_MAX_NODES:
            # FAIL CLOSED. This cap is a DoS bound, but truncating the walk silently
            # completes every sweep built on it with a partial view: whichever nodes had
            # not been popped yet are never examined, cdr_pdf reports nothing unusual, and
            # the file shipped to SANITISED_BUCKET carries whatever was in the unvisited
            # tail. Verified: a 10.6 MiB file (well under _MAX_FILE_BYTES) with 700k filler
            # dictionaries ordered so the payload pops last left a live CVE-2024-4367
            # /FontMatrix at /Page[0]/Resources/AAA_early in the output with an EMPTY
            # removed-list. Node count is attacker-controlled, so a `return` here hands the
            # attacker the sweep's coverage. See pitfall #59.
            logger.warning("PDF object walk hit the %d-node cap; rejecting",
                           _PDF_WALK_MAX_NODES)
            raise CdrReject(
                f"PDF object graph exceeds the {_PDF_WALK_MAX_NODES}-node walk cap; "
                "cannot prove the whole graph was swept")
        if isinstance(node, pikepdf.Array):
            try:
                for element_index, element in enumerate(node):
                    push(element, f"{path}[{element_index}]")
            except Exception as exc:
                logger.debug("Could not iterate array during PDF walk: %s", exc)
            continue
        yield node
        try:
            for name, value in node.items():
                push(value, f"{path}{name}")
        except Exception as exc:
            logger.debug("Could not iterate dict during PDF walk: %s", exc)


# A stream's /F key is an *external file specification*: the viewer fetches the stream's
# data from that path instead of using the embedded bytes. Pointed at a UNC share
# (\\host\share) it leaks NTLM credentials on open, with no action object anywhere in the
# file — so the unconditional annotation /A//AA sweep never sees it. Same trick as the
# Office-path UNC handling (see the WEBSERVICE/HYPERLINK notes above), different container.
# /Alternates and /OPI are the same fetch-on-open shape for image XObjects.
_EXTERNAL_STREAM_REF_KEYS = ("/F", "/Alternates", "/OPI")

# /F on a stream is *always* an external fetch, so it goes unconditionally. /Alternates and
# /OPI are only dangerous when something underneath them names an external file: an
# /Alternates array whose images are all embedded streams is legitimate fidelity data, and
# deleting it degrades the file for no security gain. These two are therefore scoped to
# subtrees that actually carry an /F.
_CONDITIONAL_EXTERNAL_REF_KEYS = ("/Alternates", "/OPI")
_EXTERNAL_REF_SCAN_MAX_NODES = 10_000


def _subtree_has_external_ref(node, depth: int = 0, budget: list | None = None) -> bool:
    """True if any dictionary under ``node`` carries an external file spec.

    Fails *closed*: anything unreadable, cyclic beyond the budget, or deeper than the cap
    counts as external, so an obfuscated subtree is removed rather than kept."""
    if budget is None:
        budget = [_EXTERNAL_REF_SCAN_MAX_NODES]
    budget[0] -= 1
    if budget[0] <= 0 or depth > 64:
        return True
    try:
        if isinstance(node, pikepdf.Array):
            return any(_subtree_has_external_ref(e, depth + 1, budget) for e in node)
        if isinstance(node, (pikepdf.Dictionary, pikepdf.Stream)):
            # /F here means the same thing it means on a stream: fetch from elsewhere.
            # (On a file-spec dictionary it is the path itself; either way it is external.)
            if "/F" in node or "/UF" in node or "/URL" in node:
                return True
            return any(_subtree_has_external_ref(v, depth + 1, budget)
                       for v in node.values())
    except Exception:
        return True
    return False


def _strip_pdf_external_stream_refs(pdf) -> list[str]:
    """Delete external-file references from every stream object.

    A stream carrying ``/F`` (plus its ``/FDecodeParms``/``/FFilter`` companions) fetches
    its data from that path on open. The dangerous case is a UNC or remote URL; we do not
    try to distinguish "safe" local paths, because a CDR output should be self-contained by
    construction — an embedded stream that needs an external fetch is already outside what
    we are willing to pass through. The stream's own bytes are left intact, so a legitimate
    file simply renders from the data it already carries."""
    streams = [o for o in _walk_pdf_nodes(pdf) if isinstance(o, pikepdf.Stream)]

    # Decide the conditional keys up front, against the *untouched* document. Stripping /F
    # from a nested alternate image would otherwise make its parent's /Alternates subtree
    # look clean by the time we reach it, so whether the parent is kept would depend on
    # traversal order. Snapshot the verdicts first, then mutate.
    doomed: set = set()
    for obj in streams:
        for key in _CONDITIONAL_EXTERNAL_REF_KEYS:
            try:
                if key in obj and _subtree_has_external_ref(obj[key]):
                    doomed.add((obj.objgen, key))
            except Exception as exc:
                raise ValueError(
                    f"Could not inspect {key} on a PDF stream"
                ) from exc

    removed: list[str] = []
    for obj in streams:
        # Per-key try, not per-stream: a single malformed key must not abort the sweep of
        # the remaining keys on the same stream and leave a live /F behind.
        for key in _EXTERNAL_STREAM_REF_KEYS:
            try:
                if key not in obj:
                    continue
                if (key in _CONDITIONAL_EXTERNAL_REF_KEYS
                        and (obj.objgen, key) not in doomed):
                    continue
                del obj[key]
                # /FDecodeParms and /FFilter only have meaning alongside /F; leaving
                # them behind would be harmless but incoherent.
                if key == "/F":
                    for companion in ("/FDecodeParms", "/FFilter"):
                        if companion in obj:
                            del obj[companion]
                removed.append(f"stream external file ref {key}")
            except Exception as exc:
                # Fail closed. Reaching here means the key was present and the delete
                # failed, so a live external-fetch reference is still on the stream;
                # swallowing that would pass a credential-leaking file off as sanitised.
                raise ValueError(
                    f"Could not remove external file reference {key} from PDF stream"
                ) from exc
    return removed


_DEFAULT_FONT_MATRIX = [0.001, 0, 0, 0.001, 0, 0]


def _is_pdf_number(element) -> bool:
    """True only for a genuine PDF numeric object.

    ``bool`` is deliberately excluded: it is a subclass of ``int`` in Python, so a naive
    ``isinstance(x, int)`` accepts a PDF boolean — which serialises as the *token* ``true``
    and is exactly the kind of non-numeric element that reaches a viewer's string
    interpolation. pikepdf hands back ``Decimal`` for reals and ``int`` for integers."""
    if isinstance(element, bool):
        return False
    return isinstance(element, (int, float, decimal.Decimal))


def _strip_pdf_fontmatrix(pdf) -> list[str]:
    """Normalise every malformed ``/FontMatrix`` to the spec default.

    CVE-2024-4367: PDF.js interpolated ``/FontMatrix`` entries straight into generated
    JavaScript, so a non-numeric element (a string) broke out and ran arbitrary JS in the
    viewer. The array is *font geometry* — it has no business holding anything but six
    numbers, and a malformed one cannot be "cleaned" element-wise without guessing intent,
    so it gets the spec default ``[0.001 0 0 0.001 0 0]``. Glyphs may render at the wrong
    scale, which is the correct trade for a file that was already lying about its geometry.

    Deliberately NOT gated on ``/Subtype == /Type3``. The vulnerable sink parses the array
    wherever it finds it, and ``/Subtype`` is attacker-controlled — gating on it lets a
    payload through by simply relabelling the font (or omitting ``/Subtype`` entirely).
    Any dictionary carrying a ``/FontMatrix`` is checked; a well-formed one is untouched,
    so the broader scope costs nothing on legitimate files."""
    removed: list[str] = []
    for obj in _walk_pdf_nodes(pdf):
        if not isinstance(obj, (pikepdf.Dictionary, pikepdf.Stream)):
            continue
        try:
            if "/FontMatrix" not in obj:
                continue
            matrix = obj["/FontMatrix"]
        except Exception as exc:
            # Probing the key failed, so we cannot show the matrix is safe. Treat an
            # unreadable /FontMatrix as hostile rather than assuming it is absent.
            raise ValueError("Could not inspect /FontMatrix on a PDF object") from exc
        numeric = True
        try:
            if len(matrix) != 6:
                numeric = False
            else:
                for element in matrix:
                    if not _is_pdf_number(element):
                        numeric = False
                        break
        except TypeError:
            numeric = False  # not even an array — spec-violating
        if numeric:
            continue
        try:
            obj["/FontMatrix"] = pikepdf.Array(_DEFAULT_FONT_MATRIX)
        except Exception as exc:
            # Fail closed: a risky /FontMatrix was positively identified and the reset
            # did not take, so the CVE-2024-4367 payload is still in the output.
            raise ValueError("Could not neutralise a malformed PDF /FontMatrix") from exc
        removed.append("/FontMatrix (non-numeric, reset to default)")
    return removed


def _strip_pdf_threads(catalog) -> list[str]:
    """Drop the document's article-thread structure (``/Threads``).

    Thread beads carry ``/I`` info dictionaries and file specs that survive every other
    sweep (they hang off the catalog, not a page, an annotation, or an action). Threads are
    a navigation nicety with no rendering role, so removing the array wholesale is safe."""
    if "/Threads" not in catalog:
        return []
    del catalog["/Threads"]
    return ["/Threads"]


def _strip_pdf_page(page, page_num: int) -> list[str]:
    """Strip active content from one PDF page: the page ``/AA``, plus every annotation's
    ``/A``/``/AA`` actions, ``/FileAttachment`` file specs, and multimedia subtypes.
    Returns the list of removals (for the report)."""
    removed: list[str] = []

    if "/AA" in page:
        del page["/AA"]
        removed.append(f"page[{page_num}]/AA")

    # Multimedia annotation subtypes carry executable/remote payloads in /RichMediaContent,
    # /Movie, /Sound dicts that the action sweep never inspects. Neutralise the subtype so
    # the viewer treats the annotation as inert (the visual rectangle remains).
    MULTIMEDIA_SUBTYPES = {"/RichMedia", "/Screen", "/3D", "/Movie", "/Sound"}
    if "/Annots" in page:
        for annot in page["/Annots"]:
            try:
                subtype = annot.get("/Subtype")
                # /FileAttachment annotations embed a file directly on the page via a
                # /FS file specification — a smuggling path around the catalog-level
                # /Names./EmbeddedFiles strip. Scrub the file spec so nothing extractable
                # remains; the visible annotation is left as an inert marker.
                if subtype == "/FileAttachment":
                    for fs_key in ("/FS", "/EF"):
                        if fs_key in annot:
                            del annot[fs_key]
                    removed.append(f"page[{page_num}] FileAttachment file spec")
                if subtype in MULTIMEDIA_SUBTYPES:
                    for mm_key in ("/RichMediaContent", "/RichMediaSettings", "/Movie", "/Sound"):
                        if mm_key in annot:
                            del annot[mm_key]
                    annot["/Subtype"] = pikepdf.Name("/CDRRemoved")
                    removed.append(f"page[{page_num}] {subtype} multimedia annotation")
                # Delete every action unconditionally. The action denylist approach left
                # gaps (/GoToE, /Rendition, /SetOCGState, …); a CDR tool should drop ALL
                # annotation actions — the visual annotation stays inert, consistent with
                # how _strip_acroform_fields already treats form fields.
                if "/A" in annot:
                    del annot["/A"]
                    removed.append(f"page[{page_num}] annot/A")
                if "/AA" in annot:
                    del annot["/AA"]
                    removed.append(f"page[{page_num}] annot/AA")
            except Exception as exc:
                logger.debug("Skipped malformed annotation on page %d: %s", page_num, exc)

    return removed


_MAX_WALK_NODES = 100_000  # cap on any recursive-tree-turned-iterative PDF walk


def _strip_pdf_outlines(catalog) -> list[str]:
    """Walk the document outline (bookmark) tree, deleting /A and /AA actions. An outline
    item's action (/JavaScript, /Launch, …) fires on bookmark click and is never reached
    by the page/annotation/AcroForm sweeps.

    Iterative (not recursive): a legitimately deep (non-cyclic) outline chain of a few
    thousand items used to silently stop being swept once a fixed recursion-depth cutoff
    was hit — well short of Python's default 1000-frame recursion limit — leaving deeper
    nodes' /A /AA actions un-stripped while the function returned normally (pitfall #43).
    An explicit stack has no such limit; the identity (`.objgen`) seen-set alone is
    sufficient to block cyclic /Next or /First chains, and _MAX_WALK_NODES bounds total
    work against a pathologically large (but acyclic) tree."""
    removed: list[str] = []
    if "/Outlines" not in catalog:
        return removed

    seen: set = set()
    stack = [catalog["/Outlines"].get("/First")]
    visited_count = 0

    while stack:
        node = stack.pop()
        if visited_count >= _MAX_WALK_NODES:
            # FAIL CLOSED (pitfall #59/#60): `continue`-ing here drains the stack without
            # examining it, so outline items beyond the cap keep their /A /AA actions and
            # cdr_pdf still returns a "sanitised" file. Node count is attacker-controlled
            # — a 100_002-node outline tree fits well under _MAX_FILE_BYTES.
            logger.warning("PDF outline walk hit the %d-node cap; rejecting",
                           _MAX_WALK_NODES)
            raise CdrReject(
                f"PDF outline tree exceeds the {_MAX_WALK_NODES}-node walk cap; "
                "cannot prove all outline actions were stripped")
        if node is None:
            continue
        try:
            ident = node.objgen
        except Exception:
            ident = None
        if ident is not None:
            if ident in seen:
                continue
            seen.add(ident)
        visited_count += 1
        try:
            for action_key in ("/A", "/AA"):
                if action_key in node:
                    del node[action_key]
                    removed.append(f"outline{action_key}")
            nxt = node.get("/Next")
            if nxt is not None:
                stack.append(nxt)
            child = node.get("/First")
            if child is not None:
                stack.append(child)
        except Exception as exc:
            logger.debug("Skipped malformed outline node: %s", exc)
    return removed


def _strip_acroform_fields(fields) -> list[str]:
    """Strip dangerous action keys from AcroForm field/widget dicts, walking /Kids.

    Iterative (not recursive) for the same reason as _strip_pdf_outlines: a fixed
    recursion-depth cutoff silently stops sweeping partway through a legitimately deep
    (non-cyclic) /Kids chain, leaving deeper fields' /JS /A /AA un-stripped while
    returning normally (pitfall #43). The identity (`.objgen`) seen-set alone is
    sufficient to block a cyclic /Kids chain (a field listing itself, or two fields
    listing each other); _MAX_WALK_NODES bounds total work against a pathologically
    large (but acyclic) field tree."""
    removed: list[str] = []
    seen: set = set()
    stack = list(fields) if fields else []
    visited_count = 0
    while stack:
        field = stack.pop()
        if visited_count >= _MAX_WALK_NODES:
            # FAIL CLOSED (pitfall #59/#60): `break`ing here abandons the rest of the
            # field tree, leaving deeper /JS /A /AA live in a file that is then reported
            # sanitised. See the identical bound in _walk_pdf_nodes.
            logger.warning("PDF AcroForm walk hit the %d-node cap; rejecting",
                           _MAX_WALK_NODES)
            raise CdrReject(
                f"PDF AcroForm field tree exceeds the {_MAX_WALK_NODES}-node walk cap; "
                "cannot prove all field actions were stripped")
        try:
            try:
                ident = field.objgen
            except Exception:
                ident = None
            if ident is not None:
                if ident in seen:
                    continue
                seen.add(ident)
            visited_count += 1
            for key in ("/A", "/AA", "/JS", "/JavaScript"):
                if key in field:
                    del field[key]
                    removed.append(f"AcroForm field {key}")
            if "/Kids" in field:
                stack.extend(field["/Kids"])
        except Exception as exc:
            logger.debug("Skipped malformed AcroForm field: %s", exc)
    return removed


# ── Image CDR ──────────────────────────────────────────────────────────────────

def cdr_image(data: bytes, ext: str) -> tuple[bytes, dict]:
    """Re-encode an image through Pillow to produce a pixel-only copy, discarding all
    metadata channels (EXIF/ICC/XMP, GIF comment blocks). Multi-frame TIFFs are
    re-encoded frame-by-frame so metadata is stripped from every frame; the output keeps
    the original format/Content-Type.

    Returns ``(clean_bytes, {"format", "removed", "original_bytes", "sanitised_bytes"})``.
    """
    removed: list[str] = []
    original_size = len(data)

    fmt_map = {"jpg": "JPEG", "jpeg": "JPEG", "png": "PNG",
               "bmp": "BMP", "tiff": "TIFF", "webp": "WEBP", "gif": "GIF"}
    save_fmt = fmt_map.get(ext, "PNG")

    # Between MAX_IMAGE_PIXELS and 2x that, Pillow only emits DecompressionBombWarning
    # (default warning filters do not raise) and proceeds to decode anyway. Escalate to
    # an error, scoped to this call only, so every image over our explicit cap fails
    # closed instead of silently decoding (and consuming memory) in the soft-warn band.
    with warnings.catch_warnings():
        warnings.simplefilter("error", Image.DecompressionBombWarning)
        # A format outside _PILLOW_FORMATS raises UnidentifiedImageError. That is a
        # deterministic verdict about the bytes, not a transient fault, so translate it
        # into CdrReject — otherwise it escapes as a generic error and EventBridge retries
        # input that will fail identically every time.
        try:
            img_ctx = Image.open(io.BytesIO(data), formats=_PILLOW_FORMATS)
        except UnidentifiedImageError as exc:
            raise CdrReject(
                f"image content does not match any allowed format {_PILLOW_FORMATS} "
                f"(declared extension: {ext})"
            ) from exc
        with img_ctx as img:
            if img.info.get("exif") or img.info.get("IFD"):
                removed.append("EXIF")
            if "icc_profile" in img.info:
                removed.append("ICC profile")
            if "xmp" in img.info:
                removed.append("XMP")
            if img.info.get("comment"):
                removed.append("comment")

            n_frames = getattr(img, "n_frames", 1)
            out_buf = io.BytesIO()

            if save_fmt in ("TIFF", "GIF", "WEBP") and n_frames > 1:
                # Re-encode every frame through Pillow to produce a pixel-only multi-frame
                # image. Iterating via ImageSequence ensures each frame is decoded
                # independently and no IFD tags or per-frame metadata survive the round-trip.
                # Saving only `img` (the first frame) here would silently collapse an
                # animated GIF/WEBP/TIFF to a single static frame.
                # Image.MAX_IMAGE_PIXELS bounds ONE frame; nothing bounded the total.
                # Every frame is materialised at once here, so a 2-frame 40 MP animation
                # (a few hundred KB compressed) decodes to ~240 MB of buffers and a
                # handful of frames reaches the 1024 MB container ceiling. Bound the
                # cumulative pixel count and the frame count, and collect in ONE pass —
                # the previous code ran ImageSequence.Iterator twice, decoding every
                # frame a second time purely to read its duration.
                frame_mode = "RGB" if save_fmt == "TIFF" else img.mode
                frames = []
                durations = []
                total_pixels = 0
                for frame in ImageSequence.Iterator(img):
                    if len(frames) >= _MAX_IMAGE_FRAMES:
                        raise CdrReject(
                            f"animated image exceeds frame cap {_MAX_IMAGE_FRAMES}"
                        )
                    total_pixels += frame.width * frame.height
                    if total_pixels > _MAX_TOTAL_IMAGE_PIXELS:
                        raise CdrReject(
                            f"animated image exceeds total pixel budget "
                            f"{_MAX_TOTAL_IMAGE_PIXELS} across {len(frames) + 1} frames"
                        )
                    durations.append(frame.info.get("duration", 0))
                    frames.append(frame.convert(frame_mode).copy())
                # Pitfall #48: the PNG and TIFF encoders pull `icc_profile` out of the
                # image's info dict (which survives .convert()/.copy()) and re-embed it
                # unless explicitly overridden — so an attacker-supplied ICC blob rode
                # through a "sanitised" image while `removed` claimed it was stripped.
                # JPEG/WEBP/GIF happen to drop it, but pass None on every path rather than
                # depending on per-encoder behaviour.
                save_all_kwargs: dict = {
                    "save_all": True, "append_images": frames[1:], "icc_profile": None,
                }
                if save_fmt in ("GIF", "WEBP"):
                    save_all_kwargs["duration"] = durations
                    save_all_kwargs["loop"] = img.info.get("loop", 0)
                if save_fmt == "GIF":
                    save_all_kwargs["comment"] = b""
                frames[0].save(out_buf, format=save_fmt, **save_all_kwargs)
                removed.append(f"multi-frame {save_fmt} metadata ({n_frames} frames re-encoded)")
            else:
                target_mode = "RGBA" if (ext == "png" and img.mode in ("RGBA", "LA", "PA")) else "RGB"
                clean_img = img.convert(target_mode)

                # icc_profile=None on every path — see pitfall #48 above.
                save_kwargs: dict = {"icc_profile": None}
                if save_fmt == "JPEG":
                    save_kwargs |= {"quality": 95, "optimize": True}
                elif save_fmt == "PNG":
                    save_kwargs |= {"optimize": True}
                elif save_fmt == "GIF":
                    # Explicitly suppress comment extension blocks — Pillow carries them
                    # through re-encode unless overridden.
                    save_kwargs = {"comment": b""}

                clean_img.save(out_buf, format=save_fmt, **save_kwargs)

            new_size = out_buf.tell()

    return out_buf.getvalue(), {
        "format":          ext,
        "removed":         removed,
        "original_bytes":  original_size,
        "sanitised_bytes": new_size,
    }


# ── Helpers ────────────────────────────────────────────────────────────────────

def _download(bucket: str, key: str) -> tuple[bytes, str]:
    """Fetch an S3 object, returning ``(body, content_type)``. Raises ``ValueError`` if
    the object's ``ContentLength`` exceeds ``_MAX_FILE_BYTES`` (a download-time size guard,
    independent of the pre-download EventBridge check)."""
    resp = s3.get_object(Bucket=bucket, Key=key)
    # Defence-in-depth against a post-event object swap: the EventBridge size field is
    # checked pre-download, but the object could be replaced before we read it. Trust the
    # S3 response headers, not the event, and refuse to buffer a multi-GB body into the
    # 1024 MB container. A terminal ValueError here exhausts retries into the DLQ (alarmed)
    # rather than OOM-crashing mid-read.
    content_length = resp.get("ContentLength", 0)
    if content_length > _MAX_FILE_BYTES:
        raise ValueError(
            f"S3 object content length {content_length} exceeds max {_MAX_FILE_BYTES}"
        )
    # ContentType is absent on objects stored without one (and on some S3-compatible
    # endpoints). Indexing would raise KeyError here, which the caller re-raises into the
    # retry/DLQ path — a download that actually succeeded, failed as if the object were
    # unreadable. The value is only ever a fallback for the output Content-Type anyway.
    return resp["Body"].read(), resp.get("ContentType") or "application/octet-stream"


def _classify_download_error(bucket: str, key: str, exc: Exception) -> None:
    """Log a structured message distinguishing NoSuchKey from other download failures."""
    from botocore.exceptions import ClientError
    if isinstance(exc, ClientError) and exc.response["Error"]["Code"] == "NoSuchKey":
        logger.warning("Source object not found (possible race/double-process): bucket=%s key=%s", bucket, key)
        _publish_result_safe(bucket, key, "source-missing", {"reason": "NoSuchKey"})
    else:
        logger.error("Download failed: bucket=%s key=%s error=%s", bucket, key, exc)


# S3 permits letters, digits, spaces, and + - = . _ : / @ in a tag key/value, and the
# value must be the LITERAL string (percent-encoding is rejected as InvalidTag). But the
# Tagging request is a `k=v&k=v` query string, so '=' and '&' inside a value break parsing
# (InvalidArgument) and could let an attacker-controlled filename inject extra pairs.
# We therefore allow S3's set MINUS '=' and '&', replacing anything else with '_'.
_TAG_SAFE = re.compile(r"[^A-Za-z0-9 +\-._:/@]")


def _enc_tag(value: str, limit: int) -> str:
    """Make a string safe to use as an S3 object-tag key/value: replace any character
    outside S3's allowed set with '_' (this also neutralises '&'/'=' injection), then cap
    at ``limit`` chars (S3 allows 128 for keys, 256 for values)."""
    return _TAG_SAFE.sub("_", value)[:limit]


# SNS Subject must be ASCII and rejects non-printable control characters; publish()
# raises InvalidParameterException otherwise. An attacker-controlled key with, say, a
# newline or non-ASCII byte would break the publish call, and because publish is
# fault-isolated (never blocks the success/delete path), the exception is silently
# swallowed — the CDR result for exactly that (adversarial) file is never published.
_SNS_SUBJECT_SAFE = re.compile(r"[^\x20-\x7e]")


def _sns_subject_safe(value: str, limit: int = 100) -> str:
    """Make a string safe to use as an SNS Subject: strip non-printable-ASCII/control
    characters, then cap at ``limit`` chars."""
    return _SNS_SUBJECT_SAFE.sub("_", value)[:limit]


def _upload(bucket: str, key: str, data: bytes,
            content_type: str, tags: dict[str, str]) -> None:
    """Put an object to S3 with AES256 encryption and a sanitised tag set. Tag keys/values
    are reduced to S3's allowed character set (see ``_enc_tag``), which also prevents
    ``&``/``=`` tag-pair injection from an attacker-controlled filename."""
    tag_str = "&".join(
        f"{_enc_tag(k, 128)}={_enc_tag(v, 256)}" for k, v in tags.items()
    )
    s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=data,
        ContentType=content_type,
        Tagging=tag_str,
        ServerSideEncryption="AES256",
    )


_EXT_CONTENT_TYPE: dict[str, str] = {
    "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    "dotx": "application/vnd.openxmlformats-officedocument.wordprocessingml.template",
    "xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "xltx": "application/vnd.openxmlformats-officedocument.spreadsheetml.template",
    "xlsb": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
    "potx": "application/vnd.openxmlformats-officedocument.presentationml.template",
    "ppsx": "application/vnd.openxmlformats-officedocument.presentationml.slideshow",
    "pdf":  "application/pdf",
    "jpg":  "image/jpeg",
    "jpeg": "image/jpeg",
    "png":  "image/png",
    "gif":  "image/gif",
    "bmp":  "image/bmp",
    "tiff": "image/tiff",
    "webp": "image/webp",
}


def _content_type_for_ext(ext: str, fallback: str) -> str:
    """Map an extension to its canonical MIME type, or return ``fallback`` if unknown."""
    return _EXT_CONTENT_TYPE.get(ext, fallback)


def _sanitised_key(key: str, sanitised_ext: Optional[str] = None) -> str:
    """Build the destination key under the ``sanitised/`` prefix, optionally swapping the
    extension (e.g. ``docm`` → ``docx`` after macro removal).

    When the extension changes, the original extension is kept in the basename (e.g.
    ``foo.docm`` -> ``foo.docm.docx``) rather than dropped (``foo.docm`` -> ``foo.docx``).
    Dropping it lets two distinct source uploads — a macro-bearing ``foo.docm`` and an
    unrelated, already-clean ``foo.docx`` — collide on the same destination key, silently
    overwriting one sanitised output with the other."""
    if sanitised_ext and "." in key:
        orig_ext = key.rsplit(".", 1)[-1]
        if orig_ext.lower() != sanitised_ext.lower():
            key = f"{key}.{sanitised_ext}"
    return f"sanitised/{key}"


def _now() -> str:
    """Current UTC time as an ISO-8601 string (used in result/tag timestamps)."""
    return datetime.now(timezone.utc).isoformat()


# SNS hard limit is 256 KB. Leave headroom for the envelope (source/status/timestamp
# keys, JSON structure) and for multi-byte UTF-8 expansion of attacker-controlled names.
_SNS_REMOVED_BYTE_BUDGET = 200 * 1024


def _cap_removed_by_bytes(items: list, budget: int) -> list:
    """Trim a 'removed' list so its JSON-serialised size stays within budget bytes.

    Entry names are attacker-controlled and can be tens of KB each, so a fixed entry
    count is not enough to keep an SNS message under 256 KB — bound by serialised size.
    """
    kept: list = []
    used = 0
    for i, entry in enumerate(items):
        size = len(json.dumps(entry)) + 1  # +1 for the comma separator
        if used + size > budget:
            kept.append(f"... and {len(items) - i} more (truncated)")
            return kept
        kept.append(entry)
        used += size
    return kept


def _truncate_removed(d: dict) -> dict:
    """Return a copy of d with any 'removed' list capped at 100 entries AND bounded by
    serialised byte size, at any nesting level, so the SNS message stays under 256 KB."""
    d = dict(d)
    if isinstance(d.get("removed"), list):
        capped = d["removed"]
        if len(capped) > 100:
            capped = capped[:100] + [f"... and {len(d['removed']) - 100} more"]
        d["removed"] = _cap_removed_by_bytes(capped, _SNS_REMOVED_BYTE_BUDGET)
    if isinstance(d.get("report"), dict):
        inner = d["report"]
        if isinstance(inner.get("removed"), list):
            inner = dict(inner)
            capped = inner["removed"]
            if len(capped) > 100:
                capped = capped[:100] + [f"... and {len(inner['removed']) - 100} more"]
            inner["removed"] = _cap_removed_by_bytes(capped, _SNS_REMOVED_BYTE_BUDGET)
            d["report"] = inner
    return d


def _publish_result_safe(source_bucket: str, key: str, status: str, report: dict) -> None:
    """Publish to SNS; on failure log a warning but never block the success/delete path."""
    if not RESULT_TOPIC_ARN:
        return
    try:
        # Inside the try: truncation/serialisation must not be able to break the
        # success/delete path either (fault isolation covers the whole publish).
        report = _truncate_removed(report)
        sns.publish(
            TopicArn=RESULT_TOPIC_ARN,
            Subject=_sns_subject_safe(f"CDR/{status}: {key}"),
            Message=json.dumps({
                "source":    f"s3://{source_bucket}/{key}",
                "status":    status,
                "timestamp": _now(),
                "report":    report,
            }),
            MessageAttributes={
                "status": {"DataType": "String", "StringValue": status},
            },
        )
    except Exception as exc:
        logger.warning("SNS publish failed: bucket=%s key=%s status=%s error=%s",
                       source_bucket, key, status, exc)


def _delete_source_safe(bucket: str, key: str) -> None:
    """Delete source object; on failure log a warning but never raise."""
    try:
        s3.delete_object(Bucket=bucket, Key=key)
    except Exception as exc:
        logger.warning("delete_object failed (sanitised copy is safe): bucket=%s key=%s error=%s",
                       bucket, key, exc)


def _emit_zip_anomaly_metric() -> None:
    """Publish one ``CDR/Validation/ZipAnomalies`` datapoint on a ZIP structural
    hard-reject. Failure to emit is logged, never raised (metrics must not break CDR)."""
    try:
        cw.put_metric_data(
            Namespace="CDR/Validation",
            MetricData=[{"MetricName": "ZipAnomalies", "Value": 1, "Unit": "Count"}],
        )
    except Exception as exc:
        logger.warning("CloudWatch metric emission failed: %s", exc)


def _emit_passthrough_metric(ext: str) -> None:
    """Emit a metric when an unknown extension is seen (now quarantined, fail-closed).

    Emits TWO datapoints: a dimensionless rollup that CdrPassthroughAlarm watches (a
    CloudWatch alarm cannot aggregate across dimension values), plus a per-extension
    breakdown for diagnostics. The dimensionless point is mandatory — without it the
    alarm monitors a series that never receives data and stays permanently non-breaching.
    """
    try:
        # CloudWatch rejects an empty Dimensions[].Value client-side (ParamValidationError)
        # — a file with no extension at all would otherwise take down BOTH datapoints in
        # this request, including the mandatory dimensionless rollup above.
        dim_value = ext[:32] if ext else "(none)"
        cw.put_metric_data(
            Namespace="CDR/Validation",
            MetricData=[
                {"MetricName": "PassthroughFiles", "Value": 1, "Unit": "Count"},
                {"MetricName": "PassthroughFiles", "Value": 1, "Unit": "Count",
                 "Dimensions": [{"Name": "Extension", "Value": dim_value}]},
            ],
        )
    except Exception as exc:
        logger.warning("CloudWatch passthrough metric emission failed: %s", exc)


def _validate_zip_structure(data: bytes) -> tuple[bool, list[str]]:
    """
    Validate ZIP structural integrity before CDR.
    Returns (is_valid_zip, anomalies).
    is_valid_zip=False → hard reject (quarantine + delete source).
    """
    if len(data) < 4:
        return False, ["file too small to be a valid ZIP"]

    if data[:4] != _ZIP_MAGIC:
        return False, [f"invalid magic bytes: {data[:4].hex().upper()} (expected 504B0304)"]

    try:
        zf_ctx = zipfile.ZipFile(io.BytesIO(data), "r")
    except zipfile.BadZipFile as exc:
        return False, [f"corrupt ZIP: {exc}"]

    anomalies: list[str] = []
    with zf_ctx as zf:
        infos = zf.infolist()
        # Per-entry and total byte caps say nothing about entry COUNT: a package of
        # hundreds of thousands of empty entries costs per-entry parse, re-deflate, and
        # report-append work with no byte cap ever tripping.
        if len(infos) > _MAX_ZIP_ENTRIES:
            return False, [f"too many ZIP entries: {len(infos)} (max {_MAX_ZIP_ENTRIES})"]

        names: list[str] = []
        for info in infos:
            if info.compress_type not in _SAFE_COMPRESS_METHODS:
                return False, [
                    f"non-standard compression method {info.compress_type} "
                    f"on entry '{info.filename}'"
                ]

            try:
                raw_offset   = info.header_offset
                local_method = int.from_bytes(data[raw_offset + 8: raw_offset + 10], "little")
                if local_method != info.compress_type:
                    return False, [
                        f"compression method mismatch on '{info.filename}': "
                        f"local={local_method} central={info.compress_type}"
                    ]
            except Exception:
                pass

            # Entry names that escape the package root. OPC part names are absolute,
            # slash-rooted, and traversal-free; an entry like "../../../etc/cron.d/x" or
            # "/abs/path" is never a legitimate part. Left unchecked, such a name passes
            # validation, survives CDR (nothing else inspects entry names outside the
            # denylist match), and is written verbatim into the SANITISED_BUCKET output —
            # so the gateway hands a zip-slip archive to every downstream consumer that
            # extracts it, stamped "sanitised". Reject the package outright.
            cname = info.filename.replace("\\", "/")
            if (
                cname.startswith("/")
                or ".." in cname.split("/")
                or (len(cname) > 1 and cname[1] == ":")  # "C:/…" drive-absolute
            ):
                return False, [f"unsafe ZIP entry name: '{info.filename}'"]

            names.append(info.filename)

        # Duplicate detection must run on the CANONICAL part name, not the raw entry
        # name. Word resolves "./word/document.xml", "word//document.xml", and
        # "word/document.xml" to the same OPC part, and part-name comparison is
        # case-insensitive — so a raw-string comparison lets an attacker ship two
        # entries that collide at open time (the consumer picks one, CDR reports on the
        # other) while the duplicate-entry hard reject sees two distinct names. Verified:
        # a package with both "word/document.xml" and "./word/document.xml" passed
        # validation before this check. (Same canonicalisation as the STRIP_ZIP_ENTRIES
        # match in cdr_office — pitfall #43.)
        seen: set[str] = set()
        for name in names:
            canonical = posixpath.normpath(name.replace("\\", "/").lower()).lstrip("/")
            if canonical in seen:
                return False, [f"duplicate ZIP entry: '{name}'"]
            seen.add(canonical)

        # Every OOXML package MUST contain [Content_Types].xml at the root. A ZIP that
        # lacks it is not a real Office document — reject rather than CDR-and-label it
        # "sanitised" (an arbitrary ZIP renamed .docx would otherwise pass through).
        if not any(n.replace("\\", "/").lower() == "[content_types].xml" for n in names):
            return False, ["not an OOXML package: missing [Content_Types].xml"]

    return True, anomalies