diff --git a/README.md b/README.md index f13c5f2..78ecb18 100644 --- a/README.md +++ b/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. diff --git a/src/pdfforensics/analyzers/content.py b/src/pdfforensics/analyzers/content.py index 8dd120b..8721a07 100644 --- a/src/pdfforensics/analyzers/content.py +++ b/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,6 +174,7 @@ 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: @@ -96,12 +182,12 @@ def analyze(doc: Document) -> AnalyzerResult: 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") diff --git a/src/pdfforensics/analyzers/metadata.py b/src/pdfforensics/analyzers/metadata.py index 177c657..279c299 100644 --- a/src/pdfforensics/analyzers/metadata.py +++ b/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", "")) diff --git a/src/pdfforensics/analyzers/structure.py b/src/pdfforensics/analyzers/structure.py index 39dda1a..ff53dd1 100644 --- a/src/pdfforensics/analyzers/structure.py +++ b/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,7 +17,7 @@ _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") @@ -25,7 +25,13 @@ def _is_content_bearing(obj: pikepdf.Object, kind: str) -> bool: 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) diff --git a/src/pdfforensics/pdfutil.py b/src/pdfforensics/pdfutil.py index 99eacf9..e8c3f59 100644 --- a/src/pdfforensics/pdfutil.py +++ b/src/pdfforensics/pdfutil.py @@ -4,6 +4,7 @@ import hashlib import re +import zlib from datetime import datetime, timedelta, timezone from typing import Any, Iterator @@ -82,6 +83,157 @@ def object_digest(obj: pikepdf.Object) -> str: return h.hexdigest() +# --------------------------------------------------------------------------- bounded stream decoding + +_FLATE = {"/FlateDecode", "/Fl"} +# Worst-case expansion per filter, used for chains that are not decoded incrementally here. +_MAX_EXPANSION = {"/ASCIIHexDecode": 1, "/AHx": 1, "/ASCII85Decode": 1, "/A85": 1, + "/RunLengthDecode": 128, "/RL": 128, "/LZWDecode": 4096, "/LZW": 4096, + "/FlateDecode": 1032, "/Fl": 1032} +_CHUNK = 1 << 16 +MAX_FILTERS = 8 # longer filter chains are not decoded +MAX_DECODED_STREAM_BYTES = 64 * 1024 * 1024 # largest stream an analyser decodes in full +# Other filter chains are handed to qpdf, which decodes them at once. They are only decoded when +# their worst case fits this ceiling; the real size is then checked against the caller's limit. +QPDF_DECODE_CEILING = 256 * 1024 * 1024 + + +class StreamTooLarge(ValueError): + """The stream decodes, or could decode, to more than can be handled safely.""" + + +def _filters(stream: pikepdf.Stream) -> tuple[list[str], bool]: + f = stream.get("/Filter") + names = [str(x) for x in f] if isinstance(f, pikepdf.Array) else ([str(f)] if f is not None else []) + parms = stream.get("/DecodeParms") + plist = list(parms) if isinstance(parms, pikepdf.Array) else [parms] + predictor = False + for p in plist: + try: + predictor = predictor or (isinstance(p, pikepdf.Dictionary) and int(p.get("/Predictor", 1)) > 1) + except Exception: + predictor = True + return names, predictor + + +def _salvage(d: Any, data: bytes, room: int, keep: bool, out: list[bytes]) -> int: + """After a zlib error, re-feed the failed window byte by byte to keep the output before the bad byte.""" + produced = 0 + for i in range(len(data)): + if produced > room or d.eof: + break + buf = data[i:i + 1] + try: + while buf or not d.eof: + chunk = d.decompress(buf, min(_CHUNK, room + 1 - produced)) + buf = d.unconsumed_tail + if not chunk: + break + produced += len(chunk) + if keep: + out.append(chunk) + if produced > room: + break + except zlib.error: + break + return produced + + +def _inflate(data: bytes, limit: int, keep: bool = True) -> tuple[bytes, int]: + """Inflate ``data``, producing at most ``limit + 1`` bytes. Returns (output if keep, bytes produced). + + Like qpdf, corrupt input keeps everything decoded before the error; an unreadable zlib header + raises ValueError, as qpdf does. + """ + d = zlib.decompressobj() + out: list[bytes] = [] + produced = 0 + buf = data + while produced <= limit and not d.eof: + snapshot = d.copy() + try: + chunk = d.decompress(buf, min(_CHUNK, limit + 1 - produced)) + except zlib.error as exc: + salvaged = _salvage(snapshot, buf, limit - produced, keep, out) + if produced == 0 and salvaged == 0: + raise ValueError(f"invalid Flate data: {exc}") from exc + produced += salvaged + break + buf = d.unconsumed_tail + if not chunk and not buf: + break + produced += len(chunk) + if keep: + out.append(chunk) + return b"".join(out), produced + + +def _flate_only(names: list[str], predictor: bool) -> bool: + return bool(names) and not predictor and all(n in _FLATE for n in names) + + +def _worst_case_size(stream: pikepdf.Stream, names: list[str]) -> int: + raw = stream.read_raw_bytes() + size = len(raw) + for i, n in enumerate(names): + if i == 0 and n in _FLATE: + size = _inflate(raw, QPDF_DECODE_CEILING, keep=False)[1] + else: + size *= _MAX_EXPANSION.get(n, 1) + return size + + +def _qpdf_decode(stream: pikepdf.Stream, names: list[str]) -> bytes: + if _worst_case_size(stream, names) > QPDF_DECODE_CEILING: + raise StreamTooLarge("filter chain could expand beyond the safe decoding ceiling") + return stream.read_bytes() + + +def read_stream(stream: pikepdf.Stream, limit: int) -> tuple[bytes, bool]: + """Decoded stream data, cut at ``limit`` bytes. Returns (data, truncated). + + Unfiltered and Flate-only streams are decoded incrementally and never beyond ``limit``. Other + filter chains are decoded by qpdf when their worst case fits QPDF_DECODE_CEILING; otherwise, or + for chains longer than MAX_FILTERS, StreamTooLarge is raised. + """ + names, predictor = _filters(stream) + if len(names) > MAX_FILTERS: + raise StreamTooLarge(f"{len(names)} filters in one stream") + if not names: + raw = stream.read_raw_bytes() + return raw[:limit], len(raw) > limit + if _flate_only(names, predictor): + data, truncated = stream.read_raw_bytes(), False + for _ in names: + data, produced = _inflate(data, limit) + truncated = truncated or produced > limit + return data[:limit], truncated + data = _qpdf_decode(stream, names) + return data[:limit], len(data) > limit + + +def decoded_size(stream: pikepdf.Stream, cap: int) -> int | None: + """The stream's decoded size if it is at most ``cap`` bytes, else None. Keeps no decoded data.""" + names, predictor = _filters(stream) + if len(names) > MAX_FILTERS: + return None + try: + if not names: + size = len(stream.read_raw_bytes()) + elif _flate_only(names, predictor): + data, size = stream.read_raw_bytes(), 0 + for i, _ in enumerate(names): + last = i == len(names) - 1 + data, size = _inflate(data, cap, keep=not last) + if size > cap: + return None + else: + size = len(_qpdf_decode(stream, names)) + except StreamTooLarge: + return None + return size if size <= cap else None + + # --------------------------------------------------------------------------- dates _PDF_DATE_RE = re.compile( diff --git a/src/pdfforensics/summary.py b/src/pdfforensics/summary.py index bbd192c..135e561 100644 --- a/src/pdfforensics/summary.py +++ b/src/pdfforensics/summary.py @@ -233,6 +233,8 @@ def _limitations(r: dict[str, Any]) -> list[str]: ids = {f["id"] for f in r["findings"]} if "signature.not-validated" in ids: out.append("Some signatures could not be verified cryptographically.") + if ids & {"metadata.xmp-too-large", "structure.stream-too-large", "content.stream-too-large"}: + out.append("Some streams exceeded the decoding limits and were not analysed in full.") if "analysis.objects-truncated" in ids or r.get("facts", {}).get("content", {}).get("pages_truncated"): out.append("Very large document: some objects or pages were outside the configured scan limits.") if r.get("facts", {}).get("revisions", {}).get("revisions_skipped"): diff --git a/tests/pdfgen.py b/tests/pdfgen.py index d0d0bf8..0354f4e 100644 --- a/tests/pdfgen.py +++ b/tests/pdfgen.py @@ -130,6 +130,58 @@ def append_unlinked_update(data: bytes, new_objects: dict[int, bytes]) -> bytes: return bytes(out) +def flate_bomb_obj(mib: int, extra: bytes = b"") -> bytes: + """A FlateDecode stream object of spaces that expands to ``mib`` MiB (about 1 KiB per MiB on disk).""" + c = zlib.compressobj(9) + comp = b"".join(c.compress(b" " * (1 << 20)) for _ in range(mib)) + c.flush() + return (b"<< /Length %d /Filter /FlateDecode " % len(comp)) + extra + b" >>\nstream\n" + comp + b"\nendstream" + + +def lzw_encode(data: bytes) -> bytes: + """PDF LZWDecode encoding (EarlyChange 1), for fixtures of older files.""" + table = {bytes([i]): i for i in range(256)} + nxt, width, acc, nbits, out = 258, 9, 0, 0, bytearray() + + def emit(code: int) -> None: + nonlocal acc, nbits + acc, nbits = (acc << width) | code, nbits + width + while nbits >= 8: + nbits -= 8 + out.append((acc >> nbits) & 0xFF) + + emit(256) + w = b"" + for c in data: + wc = w + bytes([c]) + if wc in table: + w = wc + continue + emit(table[w]) + table[wc] = nxt + nxt += 1 + if nxt + 1 > (1 << width) and width < 12: + width += 1 + if nxt >= 4094: + emit(256) + table, nxt, width = {bytes([i]): i for i in range(256)}, 258, 9 + w = bytes([c]) + if w: + emit(table[w]) + emit(257) + if nbits: + out.append((acc << (8 - nbits)) & 0xFF) + return bytes(out) + + +def with_metadata_stream(data: bytes, body: bytes) -> bytes: + """Point the catalog's /Metadata at a new stream object with the given body.""" + with pikepdf.open(io.BytesIO(data)) as pdf: + root_num = pdf.Root.objgen[0] + root_body = pdf.Root.unparse(resolved=True) + size = int(re.findall(rb"/Size\s+(\d+)", data)[-1]) + return append_update(data, {size: body, root_num: root_body[:-2] + b" /Metadata %d 0 R >>" % size}) + + def stream_obj(content: bytes) -> bytes: return b"<< /Length %d >>\nstream\n" % len(content) + content + b"\nendstream" diff --git a/tests/test_limits.py b/tests/test_limits.py new file mode 100644 index 0000000..1c9d1b4 --- /dev/null +++ b/tests/test_limits.py @@ -0,0 +1,161 @@ +"""Stream decoding is bounded: small hostile files must not decode to gigabytes.""" + +import io +import re +import zlib + +import pikepdf +import pytest + +import pdfgen +from conftest import finding, ids +from pdfforensics import pdfutil + + +_OWNERS: list[pikepdf.Pdf] = [] # a stream is only valid while its Pdf is alive + + +def _stream(data: bytes, filters: list[str]) -> pikepdf.Stream: + pdf = pikepdf.new() + _OWNERS.append(pdf) + st = pikepdf.Stream(pdf, b"") + names = [pikepdf.Name(f) for f in filters] + st.write(data, filter=(names[0] if len(names) == 1 else pikepdf.Array(names)) if names else None) + return st + + +@pytest.fixture +def no_full_decode(monkeypatch): + def refuse(self, *a, **k): + raise AssertionError("read_bytes() decodes the whole stream") + monkeypatch.setattr(pikepdf.Object, "read_bytes", refuse) + + +def test_flate_stream_is_read_up_to_the_limit_only(no_full_decode): + st = _stream(zlib.compress(b" " * (64 << 20), 9), ["/FlateDecode"]) + data, truncated = pdfutil.read_stream(st, 1 << 20) + assert truncated and len(data) == 1 << 20 + assert pdfutil.decoded_size(st, 1 << 20) is None + + +def test_nested_flate_is_bounded(no_full_decode): + inner = zlib.compress(b"x" * (32 << 20), 9) + st = _stream(zlib.compress(inner, 9), ["/FlateDecode", "/FlateDecode"]) + data, truncated = pdfutil.read_stream(st, 4096) + assert truncated and data == b"x" * 4096 + + +def test_small_streams_decode_exactly(): + text = b"BT /F1 12 Tf (Total: 100 SEK) Tj ET" + assert pdfutil.read_stream(_stream(zlib.compress(text), ["/FlateDecode"]), 1 << 20) == (text, False) + assert pdfutil.read_stream(_stream(text, []), 1 << 20) == (text, False) + hexed = zlib.compress(text).hex().encode() + b">" + assert pdfutil.read_stream(_stream(hexed, ["/ASCIIHexDecode", "/FlateDecode"]), 1 << 20) == (text, False) + + +def test_non_flate_chain_whose_worst_case_is_too_large_is_refused(no_full_decode): + rle = (bytes([129]) + b" ") * (1 << 21) + b"\x80" # each pair may expand to 128 bytes: 512 MiB + st = _stream(rle, ["/RunLengthDecode"]) + assert pdfutil.decoded_size(st, 1 << 20) is None + with pytest.raises(pdfutil.StreamTooLarge): + pdfutil.read_stream(st, 1 << 20) + + +def test_legacy_lzw_stream_decodes_like_qpdf(): + text = b" ".join(b"BT /F1 12 Tf 72 %d Td (Line %d) Tj ET" % (700 - i, i) for i in range(4000)) + st = _stream(pdfgen.lzw_encode(text), ["/LZWDecode"]) + assert len(st.read_raw_bytes()) > 20000 + assert pdfutil.read_stream(st, 1 << 20) == (text, False) + assert pdfutil.decoded_size(st, 64 << 20) == len(text) + + +@pytest.mark.parametrize("damage", ["checksum", "middle"]) +def test_corrupt_flate_keeps_what_qpdf_keeps(damage): + text = bytes(range(256)) * 3000 + comp = zlib.compress(text, 6) + if damage == "checksum": + comp = comp[:-4] + b"\0\0\0\0" + else: + mid = len(comp) // 2 + comp = comp[:mid] + b"\xff" * 40 + comp[mid + 40:] + st = _stream(comp, ["/FlateDecode"]) + assert pdfutil.read_stream(st, 1 << 30)[0] == st.read_bytes() + + +def test_unreadable_flate_header_raises_like_qpdf(): + st = _stream(b"\x00\x00" + zlib.compress(b"x" * 100)[2:], ["/FlateDecode"]) + with pytest.raises(Exception): + st.read_bytes() + with pytest.raises(ValueError): + pdfutil.read_stream(st, 1 << 20) + + +def test_xmp_bomb_is_reported_not_decoded(analyze): + data = pdfgen.with_metadata_stream(pdfgen.build(), pdfgen.flate_bomb_obj(64, b"/Type /Metadata /Subtype /XML")) + r = analyze(data) + assert finding(r, "metadata.xmp-too-large")["category"] == "metadata" + assert "Metadata dates and producer information are consistent." not in r["summary"]["ruled_out"] + + +def test_unreferenced_stream_bomb_is_reported(analyze): + r = analyze(pdfgen.with_orphan(pdfgen.build(), pdfgen.flate_bomb_obj(96))) + assert "structure.stream-too-large" in ids(r) + + +def test_page_content_bomb_is_reported_not_parsed(analyze): + base = pdfgen.build() + r = analyze(pdfgen.append_update(base, {pdfgen.content_objnum(base): pdfgen.flate_bomb_obj(96)})) + assert "content.stream-too-large" in ids(r) + assert "No hidden text, print-only annotations or hidden layers." not in r["summary"]["ruled_out"] + assert "analysis.error" not in ids(r) + + +def test_unreferenced_hex_and_flate_content_is_still_recognised(analyze): + body = b"BT /F1 12 Tf 72 700 Td (Total: 100 SEK) Tj ET\n" * 200 + hexed = zlib.compress(body).hex().encode() + b">" + obj = b"<< /Length %d /Filter [/ASCIIHexDecode /FlateDecode] >>\nstream\n" % len(hexed) + hexed + b"\nendstream" + r = analyze(pdfgen.with_orphan(pdfgen.build(), obj)) + assert finding(r, "structure.unreachable-objects")["severity"] == "low" + assert "structure.stream-too-large" not in ids(r) + + +def test_legacy_lzw_page_content_is_scanned(analyze): + base = pdfgen.build() + text = b" ".join(b"BT /F1 12 Tf 72 %d Td (Line %d) Tj ET" % (700 - i % 600, i) for i in range(4000)) + comp = pdfgen.lzw_encode(text) + obj = b"<< /Length %d /Filter /LZWDecode >>\nstream\n" % len(comp) + comp + b"\nendstream" + r = analyze(pdfgen.append_update(base, {pdfgen.content_objnum(base): obj})) + assert "content.stream-too-large" not in ids(r) + assert r["facts"]["content"]["pages"][0]["text_chars"] > 20000 + + +def test_repeated_content_streams_share_one_page_budget(analyze, monkeypatch): + from pdfforensics.analyzers import content + monkeypatch.setattr(content, "MAX_PAGE_CONTENT_BYTES", 1 << 20) + base = pdfgen.build() + size = int(re.findall(rb"/Size\s+(\d+)", base)[-1]) + with pikepdf.open(io.BytesIO(base)) as pdf: + page_num = pdf.pages[0].obj.objgen[0] + page_body = pdf.pages[0].obj.unparse(resolved=True) + chunk = pdfgen.stream_obj(b"0 g\n" * (600 << 8)) # 600 KiB each, under the limit alone + page = re.sub(rb"/Contents \d+ 0 R", b"/Contents [%d 0 R %d 0 R]" % (size, size), page_body) + r = analyze(pdfgen.append_update(base, {size: chunk, page_num: page})) + assert "content.stream-too-large" in ids(r) + + +def test_content_is_parsed_without_building_an_instruction_list(analyze, monkeypatch): + def refuse(*a, **k): + raise AssertionError("parse_content_stream builds every instruction in memory") + monkeypatch.setattr(pikepdf, "parse_content_stream", refuse) + r = analyze(pdfgen.build()) + assert r["facts"]["content"]["pages"][0]["text_chars"] > 0 + assert "content.stream-too-large" not in ids(r) + + +def test_damaged_operator_does_not_stop_the_page_scan(analyze): + base = pdfgen.build() + body = b"q \xe5\xff Q BT /F1 12 Tf 72 700 Td (Total: 100 SEK) Tj ET" + r = analyze(pdfgen.append_update(base, {pdfgen.content_objnum(base): pdfgen.stream_obj(body)})) + page = r["facts"]["content"]["pages"][0] + assert page["text_chars"] == len("Total: 100 SEK") + assert not r["facts"]["content"].get("page_errors")