Bound PDF stream decoding and content parsing

README.md

@@ -389,7 +389,8 @@ Never report `no-indicators` as "authentic" or "not tampered". It means the chec
   during analysis is reported.
 - **Offline.** No uploads and no telemetry. `pdfsig` runs with `-no-ocsp` unless you pass `--online-revocation`.
 - **Nothing is executed or fetched.** Macros, JavaScript, DDE fields, embedded files and external links are only inspected.
-- **Hostile-file hardening.** Size limits, bounded ZIP/XML parsing, XML entity declarations refused, and
+- **Hostile-file hardening.** Size limits, bounded ZIP/XML parsing, bounded PDF stream decoding (XMP 2 MB,
+  page content 64 MB per page and 256 MB per document, other streams 64 MB; reported as `*.stream-too-large` / `metadata.xmp-too-large`), XML entity declarations refused, and
   external tools given a private temporary copy with no shell.
 - **Reports are sensitive.** They quote metadata and changed text, so give them the same classification
   and handling as the documents themselves.

src/pdfforensics/analyzers/content.py

@@ -13,10 +13,13 @@
 
 from pdfforensics.document import Document
 from pdfforensics.model import AnalyzerResult, Confidence, Finding, Severity
-from pdfforensics.pdfutil import get, name, s
+from pdfforensics.pdfutil import decoded_size, get, name, s
 
 TEXT_SHOW_OPS = {"Tj", "TJ", "'", '"'}
 MAX_XOBJECT_DEPTH = 6
+MAX_PAGE_CONTENT_BYTES = 64 * 1024 * 1024        # decoded content parsed per page, forms included
+MAX_DOCUMENT_CONTENT_BYTES = 256 * 1024 * 1024   # and per document
+MAX_OPERANDS = 32                                # operands kept before an operator
 
 # annotation flags (PDF 32000-1 §12.5.3)
 F_INVISIBLE, F_HIDDEN, F_PRINT, F_NOVIEW = 1, 2, 4, 32
@@ -32,20 +35,89 @@ def _shown_chars(operands: list[Any]) -> int:
     return n
 
 
-def _scan_stream(target: Any, resources: Any, stats: Counter, depth: int, seen: set) -> None:
+class _Budget:
+    """Decoded bytes the content scan may still parse, and decoded sizes already measured."""
+
+    def __init__(self) -> None:
+        self.document = MAX_DOCUMENT_CONTENT_BYTES
+        self.sizes: dict[Any, int | None] = {}
+
+    def size(self, st: pikepdf.Stream) -> int | None:
+        key = st.objgen if st.is_indirect else None
+        if key is not None and key in self.sizes:
+            return self.sizes[key]
+        size = decoded_size(st, MAX_PAGE_CONTENT_BYTES)
+        if key is not None:
+            self.sizes[key] = size
+        return size
+
+    def charge(self, streams: list[Any], stats: Counter) -> bool:
+        total = 0
+        for st in streams:
+            if isinstance(st, pikepdf.Stream):
+                size = self.size(st)
+                if size is None:
+                    return False
+                total += size
+        if stats["decoded_bytes"] + total > MAX_PAGE_CONTENT_BYTES or total > self.document:
+            return False
+        stats["decoded_bytes"] += total
+        self.document -= total
+        return True
+
+
+class _Instructions(pikepdf.StreamParser):
+    """Hands (operands, operator) to a callback as the stream is parsed, keeping no instruction list.
+
+    pikepdf.parse_content_stream builds Python objects for every token, which costs about a hundred
+    times the decoded size in memory.
+    """
+
+    def __init__(self, on_instruction: Any) -> None:
+        super().__init__()
+        self._operands: list[Any] = []
+        self._on = on_instruction
+        self.errors = 0
+
+    def handle_object(self, obj: Any, offset: int, length: int) -> None:
+        if isinstance(obj, pikepdf.Operator):
+            operands, self._operands = self._operands, []
+            try:
+                # latin-1: an operator token in a damaged stream need not be valid UTF-8
+                self._on(operands, obj.unparse().decode("latin-1"))
+            except Exception:
+                self.errors += 1
+        elif len(self._operands) < MAX_OPERANDS:
+            self._operands.append(obj)
+
+    def handle_eof(self) -> None:
+        pass
+
+
+def _scan_stream(target: Any, resources: Any, stats: Counter, depth: int, seen: set, budget: _Budget) -> None:
     """Accumulate text/image statistics for a page or form XObject."""
     if depth > MAX_XOBJECT_DEPTH:
         stats["depth_limited"] += 1
         return
+    if isinstance(target, pikepdf.Page):
+        contents = get(target.obj, "/Contents")
+        streams = list(contents) if isinstance(contents, pikepdf.Array) else [contents]
+    else:
+        streams = [target]
     try:
-        instructions = pikepdf.parse_content_stream(target)
+        affordable = budget.charge(streams, stats)
     except Exception:
         stats["unparsable_streams"] += 1
         return
+    if not affordable:
+        stats["streams_over_limit"] += 1
+        return
+
     tr_stack = [0]
     xobjects = get(resources, "/XObject") if resources is not None else None
-    for operands, operator in instructions:
-        op = str(operator)
+    forms: list[pikepdf.Stream] = []
+
+    def step(operands: list[Any], op: str) -> None:
         if op == "q":
             tr_stack.append(tr_stack[-1])
         elif op == "Q":
@@ -57,7 +129,7 @@ def _scan_stream(target: Any, resources: Any, stats: Counter, depth: int, seen:
             except Exception:
                 pass
         elif op in TEXT_SHOW_OPS:
-            n = _shown_chars(list(operands))
+            n = _shown_chars(operands)
             stats["text_chars"] += n
             if tr_stack[-1] == 3:
                 stats["invisible_chars"] += n
@@ -66,7 +138,7 @@ def _scan_stream(target: Any, resources: Any, stats: Counter, depth: int, seen:
         elif op == "Do" and operands and xobjects is not None:
             xo = get(xobjects, str(operands[0]))
             if not isinstance(xo, pikepdf.Stream):
-                continue
+                return
             sub = name(get(xo, "/Subtype"))
             if sub == "/Image":
                 stats["images"] += 1
@@ -76,11 +148,24 @@ def _scan_stream(target: Any, resources: Any, stats: Counter, depth: int, seen:
                     pass
             elif sub == "/Form":
                 key = xo.objgen if xo.is_indirect else id(xo)
-                if key in seen:
-                    continue
-                seen.add(key)
-                stats["form_xobjects"] += 1
-                _scan_stream(xo, get(xo, "/Resources") or resources, stats, depth + 1, seen)
+                if key not in seen:
+                    seen.add(key)
+                    forms.append(xo)
+
+    try:
+        parser = _Instructions(step)
+        if isinstance(target, pikepdf.Page):
+            target.parse_contents(parser)
+        else:
+            pikepdf.Object._parse_stream(target, parser)  # what parse_content_stream uses for streams
+    except Exception:
+        stats["unparsable_streams"] += 1
+        return
+    if parser.errors:
+        stats["instruction_errors"] += parser.errors
+    for xo in forms:
+        stats["form_xobjects"] += 1
+        _scan_stream(xo, get(xo, "/Resources") or resources, stats, depth + 1, seen, budget)
 
 
 def analyze(doc: Document) -> AnalyzerResult:
@@ -89,19 +174,20 @@ def analyze(doc: Document) -> AnalyzerResult:
     pages_info: list[dict[str, Any]] = []
     invisible_pages, ocr_like_pages, hidden_annots, print_only_annots = [], [], [], []
     annot_types: Counter = Counter()
+    budget = _Budget()
 
     for idx, page in enumerate(pdf.pages):
         if idx >= doc.options.max_pages:
             res.facts["pages_truncated"] = True
             break
         stats: Counter = Counter()
         try:
-            _scan_stream(page, get(page.obj, "/Resources"), stats, 0, set())
+            _scan_stream(page, get(page.obj, "/Resources"), stats, 0, set(), budget)
         except Exception as exc:
             stats["errors"] += 1
             res.facts.setdefault("page_errors", []).append(f"page {idx + 1}: {exc}")
         pno = idx + 1
-        info = {"page": pno, **{k: v for k, v in stats.items()}}
+        info = {"page": pno, **{k: v for k, v in stats.items() if k != "decoded_bytes"}}
         pages_info.append(info)
         if stats["invisible_chars"]:
             entry = {"page": pno, "invisible_chars": stats["invisible_chars"], "text_chars": stats["text_chars"],
@@ -130,6 +216,19 @@ def analyze(doc: Document) -> AnalyzerResult:
                     continue
 
     res.facts.update({"pages": pages_info[:200], "annotation_types": dict(annot_types)})
+    over = [p["page"] for p in pages_info if p.get("streams_over_limit")]
+    if over:
+        res.findings.append(Finding(
+            id="content.stream-too-large",
+            title=f"Content on {len(over)} page(s) exceeds the decoding budget; not checked for hidden text",
+            severity=Severity.LOW, confidence=Confidence.HIGH, category="content",
+            explanation=(f"Page content is only parsed up to {MAX_PAGE_CONTENT_BYTES // (1024 * 1024)} MB of decoded "
+                         f"data per page and {MAX_DOCUMENT_CONTENT_BYTES // (1024 * 1024)} MB per document. Content "
+                         "beyond that, or content that could not be decoded safely, was not parsed, so invisible "
+                         "text on these pages was not checked."),
+            evidence={"pages": over[:50], "page_limit_bytes": MAX_PAGE_CONTENT_BYTES,
+                      "document_limit_bytes": MAX_DOCUMENT_CONTENT_BYTES},
+            benign_explanations=["Very complex vector drawing, such as a detailed map or plan"]))
 
     # optional content hidden by default / different on print
     ocp = get(pdf.Root, "/OCProperties")

src/pdfforensics/analyzers/metadata.py

@@ -11,9 +11,10 @@
 
 from pdfforensics.document import Document
 from pdfforensics.model import AnalyzerResult, Confidence, Finding, Severity
-from pdfforensics.pdfutil import date_diff, get, iso, parse_pdf_date, parse_xmp_date, s
+from pdfforensics.pdfutil import StreamTooLarge, date_diff, get, iso, parse_pdf_date, parse_xmp_date, read_stream, s
 
 MAX_XMP_BYTES = 2 * 1024 * 1024
+XMP_TOO_LARGE = "XMP stream larger than 2 MB; skipped"
 EXACT_TOLERANCE = timedelta(minutes=2)
 NAIVE_TOLERANCE = timedelta(hours=14)  # max UTC offset spread when a timezone is missing
 
@@ -83,11 +84,13 @@ def read_xmp(pdf: pikepdf.Pdf) -> tuple[dict[str, Any] | None, str | None]:
     if not isinstance(md, pikepdf.Stream):
         return None, None
     try:
-        raw = md.read_bytes()
+        raw, truncated = read_stream(md, MAX_XMP_BYTES)
+    except StreamTooLarge:
+        return None, XMP_TOO_LARGE
     except Exception as exc:
         return None, f"XMP stream unreadable: {exc}"
-    if len(raw) > MAX_XMP_BYTES:
-        return None, "XMP stream larger than 2 MB; skipped"
+    if truncated:
+        return None, XMP_TOO_LARGE
     text = raw.decode("utf-8", errors="replace")
     # XMP packets may carry a BOM / xpacket PI; ET handles PIs, strip leading junk.
     start = text.find("<")
@@ -146,6 +149,15 @@ def analyze(doc: Document) -> AnalyzerResult:
                       "trailer_id_parts_equal": len(ids) == 2 and ids[0] == ids[1]})
     if xmp_err:
         res.facts["xmp_error"] = xmp_err
+    if xmp_err == XMP_TOO_LARGE:
+        res.findings.append(Finding(
+            id="metadata.xmp-too-large", title="XMP metadata stream too large to decode; not analysed",
+            severity=Severity.LOW, confidence=Confidence.HIGH, category="metadata",
+            explanation=("The XMP stream decodes to more than 2 MB, so it was not decoded and XMP dates and history "
+                         "were not checked. Real XMP packets are a few kilobytes; a very large one can be padding "
+                         "or a decompression bomb."),
+            evidence={"limit_bytes": MAX_XMP_BYTES},
+            benign_explanations=["Producer embedding large thumbnails or history in XMP"]))
 
     now = datetime.now(timezone.utc)
     c_info = parse_pdf_date(info.get("CreationDate", ""))

src/pdfforensics/analyzers/structure.py

@@ -8,7 +8,7 @@
 
 from pdfforensics.document import Document
 from pdfforensics.model import AnalyzerResult, Confidence, Finding, Severity
-from pdfforensics.pdfutil import _walk, get, name
+from pdfforensics.pdfutil import MAX_DECODED_STREAM_BYTES, StreamTooLarge, _walk, decoded_size, get, name, read_stream
 
 SKIP_TYPES = {"/ObjStm", "/XRef"}
 _OBJ_AT = re.compile(rb"\s*(\d+)\s+(\d+)\s+obj")
@@ -17,15 +17,21 @@
 _TEXT_OPS = re.compile(rb"\bBT\b.*?(Tj|TJ|')", re.S)
 
 
-def _is_content_bearing(obj: pikepdf.Object, kind: str) -> bool:
+def _is_content_bearing(obj: pikepdf.Object, kind: str, oversize: list[pikepdf.Object] | None = None) -> bool:
     if kind in ("/Page", "/Annot") or "/Rect" in obj:
         return True
     if "/Length1" in obj or (kind in ("/Metadata", "/EmbeddedFile", "/XObject")
                              and name(get(obj, "/Subtype")) != "/Form"):
         return False  # font programs, XMP, attachments, images
     if isinstance(obj, pikepdf.Stream) and name(get(obj, "/Subtype")) in ("", "/Form"):
         try:
-            data = obj.read_bytes()[:262144]
+            data, truncated = read_stream(obj, 262144)
+            if truncated and oversize is not None and decoded_size(obj, MAX_DECODED_STREAM_BYTES) is None:
+                oversize.append(obj)
+        except StreamTooLarge:
+            if oversize is not None:
+                oversize.append(obj)
+            return False
         except Exception:
             return False
         return bool(_TEXT_OPS.search(data))
@@ -106,6 +112,7 @@ def analyze(doc: Document) -> AnalyzerResult:
         reachable |= _linearization_objects(doc)
         unreachable: list[str] = []
         content_bearing: list[str] = []
+        oversize: list[pikepdf.Object] = []
         for obj in pdf.objects:
             if obj is None or not isinstance(obj, pikepdf.Object):
                 continue
@@ -121,7 +128,7 @@ def analyze(doc: Document) -> AnalyzerResult:
                     or ("stream" if isinstance(obj, pikepdf.Stream) else "dictionary"))
             label = f"{obj.objgen[0]} {obj.objgen[1]} R ({kind})"
             unreachable.append(label)
-            if _is_content_bearing(obj, kind):
+            if _is_content_bearing(obj, kind, oversize):
                 content_bearing.append(label)
         res.facts["unreachable_objects"] = len(unreachable)
         if unreachable:
@@ -136,6 +143,17 @@ def analyze(doc: Document) -> AnalyzerResult:
                              "removed content. Inspect them with 'qpdf --show-object=N --filtered-stream-data'."),
                 evidence={"content_bearing": content_bearing[:30], "objects": unreachable[:30]},
                 benign_explanations=["Many generators leave unused fonts/resources behind"]))
+        if oversize:
+            res.findings.append(Finding(
+                id="structure.stream-too-large",
+                title=f"{len(oversize)} unreferenced stream(s) decode to more than "
+                      f"{MAX_DECODED_STREAM_BYTES // (1024 * 1024)} MB or could not be decoded safely",
+                severity=Severity.LOW, confidence=Confidence.HIGH, category="structure",
+                explanation=("These streams were only sampled, not decoded in full. An unreferenced stream that "
+                             "expands this much is a decompression-bomb pattern."),
+                evidence={"objects": [f"{o.objgen[0]} {o.objgen[1]} R" for o in oversize[:30]],
+                          "limit_bytes": MAX_DECODED_STREAM_BYTES},
+                benign_explanations=["Leftover large image or font data from an earlier save"]))
     except Exception as exc:
         res.facts["unreachable_error"] = str(exc)