Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/PIL/PdfParser.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3__lazy_modules__ = {"calendar", "mmap", "zlib"}
5import calendar
6import codecs
7import collections
8import mmap
9import os
10import re
11import sys
12import time
13import zlib
14from typing import Any, NamedTuple
16from . import ImageFile
18TYPE_CHECKING = False
19if TYPE_CHECKING:
20 from typing import IO, Self
22 _DictBase = collections.UserDict[str | bytes, Any]
23else:
24 _DictBase = collections.UserDict
27# see 7.9.2.2 Text String Type on page 86 and D.3 PDFDocEncoding Character Set
28# on page 656
29def encode_text(s: str) -> bytes:
30 return codecs.BOM_UTF16_BE + s.encode("utf_16_be")
33PDFDocEncoding = {
34 0x16: "\u0017",
35 0x18: "\u02d8",
36 0x19: "\u02c7",
37 0x1A: "\u02c6",
38 0x1B: "\u02d9",
39 0x1C: "\u02dd",
40 0x1D: "\u02db",
41 0x1E: "\u02da",
42 0x1F: "\u02dc",
43 0x80: "\u2022",
44 0x81: "\u2020",
45 0x82: "\u2021",
46 0x83: "\u2026",
47 0x84: "\u2014",
48 0x85: "\u2013",
49 0x86: "\u0192",
50 0x87: "\u2044",
51 0x88: "\u2039",
52 0x89: "\u203a",
53 0x8A: "\u2212",
54 0x8B: "\u2030",
55 0x8C: "\u201e",
56 0x8D: "\u201c",
57 0x8E: "\u201d",
58 0x8F: "\u2018",
59 0x90: "\u2019",
60 0x91: "\u201a",
61 0x92: "\u2122",
62 0x93: "\ufb01",
63 0x94: "\ufb02",
64 0x95: "\u0141",
65 0x96: "\u0152",
66 0x97: "\u0160",
67 0x98: "\u0178",
68 0x99: "\u017d",
69 0x9A: "\u0131",
70 0x9B: "\u0142",
71 0x9C: "\u0153",
72 0x9D: "\u0161",
73 0x9E: "\u017e",
74 0xA0: "\u20ac",
75}
78def decode_text(b: bytes) -> str:
79 if b[: len(codecs.BOM_UTF16_BE)] == codecs.BOM_UTF16_BE:
80 return b[len(codecs.BOM_UTF16_BE) :].decode("utf_16_be")
81 else:
82 return "".join(PDFDocEncoding.get(byte, chr(byte)) for byte in b)
85class PdfFormatError(RuntimeError):
86 """An error that probably indicates a syntactic or semantic error in the
87 PDF file structure"""
89 pass
92def check_format_condition(condition: bool, error_message: str) -> None:
93 if not condition:
94 raise PdfFormatError(error_message)
97class IndirectReferenceTuple(NamedTuple):
98 object_id: int
99 generation: int
102class IndirectReference(IndirectReferenceTuple):
103 def __str__(self) -> str:
104 return f"{self.object_id} {self.generation} R"
106 def __bytes__(self) -> bytes:
107 return self.__str__().encode("us-ascii")
109 def __eq__(self, other: object) -> bool:
110 if self.__class__ is not other.__class__:
111 return False
112 assert isinstance(other, IndirectReference)
113 return other.object_id == self.object_id and other.generation == self.generation
115 def __ne__(self, other: object) -> bool:
116 return not (self == other)
118 def __hash__(self) -> int:
119 return hash((self.object_id, self.generation))
122class IndirectObjectDef(IndirectReference):
123 def __str__(self) -> str:
124 return f"{self.object_id} {self.generation} obj"
127class XrefTable:
128 def __init__(self) -> None:
129 self.existing_entries: dict[int, tuple[int, int]] = (
130 {}
131 ) # object ID => (offset, generation)
132 self.new_entries: dict[int, tuple[int, int]] = (
133 {}
134 ) # object ID => (offset, generation)
135 self.deleted_entries = {0: 65536} # object ID => generation
136 self.reading_finished = False
137 self._max_object_id = 0
139 def __setitem__(self, key: int, value: tuple[int, int]) -> None:
140 if self.reading_finished:
141 self.new_entries[key] = value
142 else:
143 self.existing_entries[key] = value
144 if key > self._max_object_id:
145 self._max_object_id = key
146 if key in self.deleted_entries:
147 del self.deleted_entries[key]
149 def __getitem__(self, key: int) -> tuple[int, int]:
150 try:
151 return self.new_entries[key]
152 except KeyError:
153 return self.existing_entries[key]
155 def __delitem__(self, key: int) -> None:
156 if key in self.new_entries:
157 generation = self.new_entries[key][1] + 1
158 del self.new_entries[key]
159 self.deleted_entries[key] = generation
160 elif key in self.existing_entries:
161 generation = self.existing_entries[key][1] + 1
162 self.deleted_entries[key] = generation
163 elif key in self.deleted_entries:
164 generation = self.deleted_entries[key]
165 else:
166 msg = f"object ID {key} cannot be deleted because it doesn't exist"
167 raise IndexError(msg)
169 def __contains__(self, key: int) -> bool:
170 return key in self.existing_entries or key in self.new_entries
172 def __len__(self) -> int:
173 return len(
174 set(self.existing_entries.keys())
175 | set(self.new_entries.keys())
176 | set(self.deleted_entries.keys())
177 )
179 def keys(self) -> set[int]:
180 return (
181 set(self.existing_entries.keys()) - set(self.deleted_entries.keys())
182 ) | set(self.new_entries.keys())
184 def write(self, f: IO[bytes]) -> int:
185 keys = sorted(set(self.new_entries.keys()) | set(self.deleted_entries.keys()))
186 deleted_keys = sorted(self.deleted_entries)
187 startxref = f.tell()
188 f.write(b"xref\n")
189 while keys:
190 # find a contiguous sequence of object IDs
191 prev: int | None = None
192 for index, key in enumerate(keys):
193 if prev is None or prev + 1 == key:
194 prev = key
195 else:
196 contiguous_keys = keys[:index]
197 keys = keys[index:]
198 break
199 else:
200 contiguous_keys = keys
201 keys = []
202 f.write(b"%d %d\n" % (contiguous_keys[0], len(contiguous_keys)))
203 for object_id in contiguous_keys:
204 if object_id in self.new_entries:
205 f.write(b"%010d %05d n \n" % self.new_entries[object_id])
206 else:
207 this_deleted_object_id = deleted_keys.pop(0)
208 check_format_condition(
209 object_id == this_deleted_object_id,
210 f"expected the next deleted object ID to be {object_id}, "
211 f"instead found {this_deleted_object_id}",
212 )
213 try:
214 next_in_linked_list = deleted_keys[0]
215 except IndexError:
216 next_in_linked_list = 0
217 f.write(
218 b"%010d %05d f \n"
219 % (next_in_linked_list, self.deleted_entries[object_id])
220 )
221 return startxref
224class PdfName:
225 name: bytes
227 def __init__(self, name: PdfName | bytes | str) -> None:
228 if isinstance(name, PdfName):
229 self.name = name.name
230 elif isinstance(name, bytes):
231 self.name = name
232 else:
233 self.name = name.encode("us-ascii")
235 def name_as_str(self) -> str:
236 return self.name.decode("us-ascii")
238 def __eq__(self, other: object) -> bool:
239 return (
240 isinstance(other, PdfName) and other.name == self.name
241 ) or other == self.name
243 def __hash__(self) -> int:
244 return hash(self.name)
246 def __repr__(self) -> str:
247 return f"{self.__class__.__name__}({repr(self.name)})"
249 @classmethod
250 def from_pdf_stream(cls, data: bytes) -> PdfName:
251 return cls(PdfParser.interpret_name(data))
253 allowed_chars = set(range(33, 127)) - {ord(c) for c in "#%/()<>[]{}"}
255 def __bytes__(self) -> bytes:
256 result = bytearray(b"/")
257 for b in self.name:
258 if b in self.allowed_chars:
259 result.append(b)
260 else:
261 result.extend(b"#%02X" % b)
262 return bytes(result)
265class PdfArray(list[Any]):
266 def __bytes__(self) -> bytes:
267 return b"[ " + b" ".join(pdf_repr(x) for x in self) + b" ]"
270class PdfDict(_DictBase):
271 def __setattr__(self, key: str, value: Any) -> None:
272 if key == "data":
273 collections.UserDict.__setattr__(self, key, value)
274 else:
275 self[key.encode("us-ascii")] = value
277 def __getattr__(self, key: str) -> str | time.struct_time:
278 try:
279 value = self[key.encode("us-ascii")]
280 except KeyError as e:
281 raise AttributeError(key) from e
282 if isinstance(value, bytes):
283 value = decode_text(value)
284 if key.endswith("Date"):
285 if value.startswith("D:"):
286 value = value[2:]
288 relationship = "Z"
289 if len(value) > 17:
290 relationship = value[14]
291 offset = int(value[15:17]) * 60
292 if len(value) > 20:
293 offset += int(value[18:20])
295 format = "%Y%m%d%H%M%S"[: len(value) - 2]
296 value = time.strptime(value[: len(format) + 2], format)
297 if relationship in ["+", "-"]:
298 offset *= 60
299 if relationship == "+":
300 offset *= -1
301 value = time.gmtime(calendar.timegm(value) + offset)
302 return value
304 def __bytes__(self) -> bytes:
305 out = bytearray(b"<<")
306 for key, value in self.items():
307 if value is None:
308 continue
309 value = pdf_repr(value)
310 out.extend(b"\n")
311 out.extend(bytes(PdfName(key)))
312 out.extend(b" ")
313 out.extend(value)
314 out.extend(b"\n>>")
315 return bytes(out)
318class PdfBinary:
319 def __init__(self, data: list[int] | bytes) -> None:
320 self.data = data
322 def __bytes__(self) -> bytes:
323 return b"<%s>" % b"".join(b"%02X" % b for b in self.data)
326class PdfStream:
327 def __init__(self, dictionary: PdfDict, buf: bytes) -> None:
328 self.dictionary = dictionary
329 self.buf = buf
331 def decode(self, max_length: int = ImageFile.SAFEBLOCK) -> bytes:
332 try:
333 filter = self.dictionary[b"Filter"]
334 except KeyError:
335 return self.buf
336 if filter == b"FlateDecode":
337 dobj = zlib.decompressobj()
338 plaintext = dobj.decompress(self.buf, max_length)
339 if dobj.unconsumed_tail:
340 msg = "Decompressed data too large"
341 raise ValueError(msg)
342 return plaintext
343 else:
344 msg = f"stream filter {repr(filter)} unknown/unsupported"
345 raise NotImplementedError(msg)
348def pdf_repr(x: Any) -> bytes:
349 if x is True:
350 return b"true"
351 elif x is False:
352 return b"false"
353 elif x is None:
354 return b"null"
355 elif isinstance(x, (PdfName, PdfDict, PdfArray, PdfBinary)):
356 return bytes(x)
357 elif isinstance(x, (int, float)):
358 return str(x).encode("us-ascii")
359 elif isinstance(x, time.struct_time):
360 return b"(D:" + time.strftime("%Y%m%d%H%M%SZ", x).encode("us-ascii") + b")"
361 elif isinstance(x, dict):
362 return bytes(PdfDict(x))
363 elif isinstance(x, list):
364 return bytes(PdfArray(x))
365 elif isinstance(x, str):
366 return pdf_repr(encode_text(x))
367 elif isinstance(x, bytes):
368 # XXX escape more chars? handle binary garbage
369 x = x.replace(b"\\", b"\\\\")
370 x = x.replace(b"(", b"\\(")
371 x = x.replace(b")", b"\\)")
372 return b"(" + x + b")"
373 else:
374 return bytes(x)
377class PdfParser:
378 """Based on
379 https://www.adobe.com/content/dam/acom/en/devnet/acrobat/pdfs/PDF32000_2008.pdf
380 Supports PDF up to 1.4
381 """
383 def __init__(
384 self,
385 filename: str | None = None,
386 f: IO[bytes] | None = None,
387 buf: bytes | bytearray | None = None,
388 start_offset: int = 0,
389 mode: str = "rb",
390 ) -> None:
391 if buf and f:
392 msg = "specify buf or f or filename, but not both buf and f"
393 raise RuntimeError(msg)
394 self.filename = filename
395 self.buf: bytes | bytearray | memoryview | mmap.mmap | None = buf
396 self.f = f
397 self.start_offset = start_offset
398 self.should_close_buf = False
399 self.should_close_file = False
400 if filename is not None and f is None:
401 self.f = f = open(filename, mode)
402 self.should_close_file = True
403 if f is not None:
404 self.buf = self.get_buf_from_file(f)
405 self.should_close_buf = True
406 if not filename and hasattr(f, "name"):
407 self.filename = f.name
408 self.cached_objects: dict[IndirectReference, Any] = {}
409 self.root_ref: IndirectReference | None
410 self.info_ref: IndirectReference | None
411 self.pages_ref: IndirectReference | None
412 self.last_xref_section_offset: int | None
413 if self.buf:
414 try:
415 self.read_pdf_info()
416 except PdfFormatError:
417 self.close()
418 raise
419 else:
420 self.file_size_total = self.file_size_this = 0
421 self.root = PdfDict()
422 self.root_ref = None
423 self.info = PdfDict()
424 self.info_ref = None
425 self.page_tree_root = PdfDict()
426 self.pages: list[IndirectReference] = []
427 self.orig_pages: list[IndirectReference] = []
428 self.pages_ref = None
429 self.last_xref_section_offset = None
430 self.trailer_dict: dict[bytes, Any] = {}
431 self.xref_table = XrefTable()
432 self.xref_table.reading_finished = True
433 if f:
434 self.seek_end()
436 def __enter__(self) -> Self:
437 return self
439 def __exit__(self, *args: object) -> None:
440 self.close()
442 def start_writing(self) -> None:
443 self.close_buf()
444 self.seek_end()
446 def close_buf(self) -> None:
447 if isinstance(self.buf, memoryview):
448 self.buf.release()
449 if sys.implementation.name == "pypy":
450 import gc
452 # See https://github.com/pypy/pypy/pull/5616
453 gc.collect()
454 elif isinstance(self.buf, mmap.mmap):
455 self.buf.close()
456 self.buf = None
458 def close(self) -> None:
459 if self.should_close_buf:
460 self.close_buf()
461 if self.f is not None and self.should_close_file:
462 self.f.close()
463 self.f = None
465 def seek_end(self) -> None:
466 assert self.f is not None
467 self.f.seek(0, os.SEEK_END)
469 def write_header(self) -> None:
470 assert self.f is not None
471 self.f.write(b"%PDF-1.4\n")
473 def write_comment(self, s: str) -> None:
474 assert self.f is not None
475 self.f.write(f"% {s}\n".encode())
477 def write_catalog(self) -> IndirectReference:
478 assert self.f is not None
479 self.del_root()
480 self.root_ref = self.next_object_id(self.f.tell())
481 self.pages_ref = self.next_object_id(0)
482 self.rewrite_pages()
483 self.write_obj(self.root_ref, Type=PdfName(b"Catalog"), Pages=self.pages_ref)
484 self.write_obj(
485 self.pages_ref,
486 Type=PdfName(b"Pages"),
487 Count=len(self.pages),
488 Kids=self.pages,
489 )
490 return self.root_ref
492 def rewrite_pages(self) -> None:
493 pages_tree_nodes_to_delete = []
494 for i, page_ref in enumerate(self.orig_pages):
495 page_info = self.cached_objects[page_ref]
496 del self.xref_table[page_ref.object_id]
497 pages_tree_nodes_to_delete.append(page_info[PdfName(b"Parent")])
498 if page_ref not in self.pages:
499 # the page has been deleted
500 continue
501 # make dict keys into strings for passing to write_page
502 stringified_page_info = {}
503 for key, value in page_info.items():
504 # key should be a PdfName
505 stringified_page_info[key.name_as_str()] = value
506 stringified_page_info["Parent"] = self.pages_ref
507 new_page_ref = self.write_page(None, **stringified_page_info)
508 for j, cur_page_ref in enumerate(self.pages):
509 if cur_page_ref == page_ref:
510 # replace the page reference with the new one
511 self.pages[j] = new_page_ref
512 # delete redundant Pages tree nodes from xref table
513 for pages_tree_node_ref in pages_tree_nodes_to_delete:
514 processed_refs = []
515 while pages_tree_node_ref and pages_tree_node_ref not in processed_refs:
516 processed_refs.append(pages_tree_node_ref)
517 pages_tree_node = self.cached_objects[pages_tree_node_ref]
518 if pages_tree_node_ref.object_id in self.xref_table:
519 del self.xref_table[pages_tree_node_ref.object_id]
520 pages_tree_node_ref = pages_tree_node.get(b"Parent", None)
521 self.orig_pages = []
523 def write_xref_and_trailer(
524 self, new_root_ref: IndirectReference | None = None
525 ) -> None:
526 assert self.f is not None
527 if new_root_ref:
528 self.del_root()
529 self.root_ref = new_root_ref
530 if self.info:
531 self.info_ref = self.write_obj(None, self.info)
532 start_xref = self.xref_table.write(self.f)
533 num_entries = len(self.xref_table)
534 trailer_dict: dict[str | bytes, Any] = {
535 b"Root": self.root_ref,
536 b"Size": num_entries,
537 }
538 if self.last_xref_section_offset is not None:
539 trailer_dict[b"Prev"] = self.last_xref_section_offset
540 if self.info:
541 trailer_dict[b"Info"] = self.info_ref
542 self.last_xref_section_offset = start_xref
543 self.f.write(
544 b"trailer\n"
545 + bytes(PdfDict(trailer_dict))
546 + b"\nstartxref\n%d\n%%%%EOF" % start_xref
547 )
549 def write_page(
550 self, ref: int | IndirectReference | None, *objs: Any, **dict_obj: Any
551 ) -> IndirectReference:
552 obj_ref = self.pages[ref] if isinstance(ref, int) else ref
553 if "Type" not in dict_obj:
554 dict_obj["Type"] = PdfName(b"Page")
555 if "Parent" not in dict_obj:
556 dict_obj["Parent"] = self.pages_ref
557 return self.write_obj(obj_ref, *objs, **dict_obj)
559 def write_obj(
560 self, ref: IndirectReference | None, *objs: Any, **dict_obj: Any
561 ) -> IndirectReference:
562 assert self.f is not None
563 f = self.f
564 if ref is None:
565 ref = self.next_object_id(f.tell())
566 else:
567 self.xref_table[ref.object_id] = (f.tell(), ref.generation)
568 f.write(bytes(IndirectObjectDef(*ref)))
569 stream = dict_obj.pop("stream", None)
570 if stream is not None:
571 dict_obj["Length"] = len(stream)
572 if dict_obj:
573 f.write(pdf_repr(dict_obj))
574 for obj in objs:
575 f.write(pdf_repr(obj))
576 if stream is not None:
577 f.write(b"stream\n")
578 f.write(stream)
579 f.write(b"\nendstream\n")
580 f.write(b"endobj\n")
581 return ref
583 def del_root(self) -> None:
584 if self.root_ref is None:
585 return
586 del self.xref_table[self.root_ref.object_id]
587 del self.xref_table[self.root[b"Pages"].object_id]
589 @staticmethod
590 def get_buf_from_file(f: IO[bytes]) -> bytes | mmap.mmap:
591 if hasattr(f, "getbuffer"):
592 return f.getbuffer()
593 elif hasattr(f, "getvalue"):
594 return f.getvalue()
595 else:
596 try:
597 return mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
598 except ValueError: # cannot mmap an empty file
599 return b""
601 def read_pdf_info(self) -> None:
602 assert self.buf is not None
603 self.file_size_total = len(self.buf)
604 self.file_size_this = self.file_size_total - self.start_offset
605 self.read_trailer()
606 check_format_condition(
607 self.trailer_dict.get(b"Root") is not None, "Root is missing"
608 )
609 self.root_ref = self.trailer_dict[b"Root"]
610 assert self.root_ref is not None
611 self.info_ref = self.trailer_dict.get(b"Info", None)
612 self.root = PdfDict(self.read_indirect(self.root_ref))
613 if self.info_ref is None:
614 self.info = PdfDict()
615 else:
616 self.info = PdfDict(self.read_indirect(self.info_ref))
617 check_format_condition(b"Type" in self.root, "/Type missing in Root")
618 check_format_condition(
619 self.root[b"Type"] == b"Catalog", "/Type in Root is not /Catalog"
620 )
621 check_format_condition(
622 self.root.get(b"Pages") is not None, "/Pages missing in Root"
623 )
624 check_format_condition(
625 isinstance(self.root[b"Pages"], IndirectReference),
626 "/Pages in Root is not an indirect reference",
627 )
628 self.pages_ref = self.root[b"Pages"]
629 assert self.pages_ref is not None
630 self.page_tree_root = self.read_indirect(self.pages_ref)
631 self.pages = self.linearize_page_tree()
632 # save the original list of page references
633 # in case the user modifies, adds or deletes some pages
634 # and we need to rewrite the pages and their list
635 self.orig_pages = self.pages[:]
637 def next_object_id(self, offset: int | None = None) -> IndirectReference:
638 # TODO: support reuse of deleted objects
639 reference = IndirectReference(self.xref_table._max_object_id + 1, 0)
640 if offset is not None:
641 self.xref_table[reference.object_id] = (offset, 0)
642 return reference
644 delimiter = rb"[][()<>{}/%]"
645 delimiter_or_ws = rb"[][()<>{}/%\000\011\012\014\015\040]"
646 whitespace = rb"[\000\011\012\014\015\040]"
647 whitespace_or_hex = rb"[\000\011\012\014\015\0400-9a-fA-F]"
648 whitespace_optional = whitespace + b"*"
649 whitespace_mandatory = whitespace + b"+"
650 # No "\012" aka "\n" or "\015" aka "\r":
651 whitespace_optional_no_nl = rb"[\000\011\014\040]*"
652 newline_only = rb"[\r\n]+"
653 newline = whitespace_optional_no_nl + newline_only + whitespace_optional_no_nl
654 re_trailer_end = re.compile(
655 whitespace_mandatory
656 + rb"trailer"
657 + whitespace_optional
658 + rb"<<(.*>>)"
659 + newline
660 + rb"startxref"
661 + newline
662 + rb"([0-9]+)"
663 + newline
664 + rb"%%EOF"
665 + whitespace_optional
666 + rb"$",
667 re.DOTALL,
668 )
669 re_trailer_prev = re.compile(
670 whitespace_optional
671 + rb"trailer"
672 + whitespace_optional
673 + rb"<<(.*?>>)"
674 + newline
675 + rb"startxref"
676 + newline
677 + rb"([0-9]+)"
678 + newline
679 + rb"%%EOF"
680 + whitespace_optional,
681 re.DOTALL,
682 )
684 def read_trailer(self) -> None:
685 assert self.buf is not None
686 search_start_offset = len(self.buf) - 16384
687 if search_start_offset < self.start_offset:
688 search_start_offset = self.start_offset
689 m = self.re_trailer_end.search(self.buf, search_start_offset)
690 check_format_condition(m is not None, "trailer end not found")
691 # make sure we found the LAST trailer
692 last_match = m
693 while m:
694 last_match = m
695 m = self.re_trailer_end.search(self.buf, m.start() + 16)
696 if not m:
697 m = last_match
698 assert m is not None
699 trailer_data = m.group(1)
700 self.last_xref_section_offset = int(m.group(2))
701 self.trailer_dict = self.interpret_trailer(trailer_data)
702 self.xref_table = XrefTable()
703 self.read_xref_table(xref_section_offset=self.last_xref_section_offset)
704 if b"Prev" in self.trailer_dict:
705 self.read_prev_trailer(self.trailer_dict[b"Prev"])
707 def read_prev_trailer(
708 self, xref_section_offset: int, processed_offsets: list[int] | None = None
709 ) -> None:
710 assert self.buf is not None
711 if processed_offsets is None:
712 processed_offsets = []
713 while True:
714 check_format_condition(
715 xref_section_offset not in processed_offsets, "trailer loop found"
716 )
717 trailer_offset = self.read_xref_table(
718 xref_section_offset=xref_section_offset
719 )
720 m = self.re_trailer_prev.search(
721 self.buf[trailer_offset : trailer_offset + 16384]
722 )
723 check_format_condition(m is not None, "previous trailer not found")
724 assert m is not None
725 trailer_data = m.group(1)
726 check_format_condition(
727 int(m.group(2)) == xref_section_offset,
728 "xref section offset in previous trailer "
729 "doesn't match what was expected",
730 )
731 trailer_dict = self.interpret_trailer(trailer_data)
732 if b"Prev" not in trailer_dict:
733 break
734 processed_offsets.append(xref_section_offset)
735 xref_section_offset = trailer_dict[b"Prev"]
737 re_whitespace_optional = re.compile(whitespace_optional)
738 re_name = re.compile(
739 whitespace_optional
740 + rb"/([!-$&'*-.0-;=?-Z\\^-z|~]+)(?="
741 + delimiter_or_ws
742 + rb")"
743 )
744 re_dict_start = re.compile(whitespace_optional + rb"<<")
745 re_dict_end = re.compile(whitespace_optional + rb">>" + whitespace_optional)
747 @classmethod
748 def interpret_trailer(cls, trailer_data: bytes) -> dict[bytes, Any]:
749 trailer = {}
750 offset = 0
751 while True:
752 m = cls.re_name.match(trailer_data, offset)
753 if not m:
754 m = cls.re_dict_end.match(trailer_data, offset)
755 check_format_condition(
756 m is not None and m.end() == len(trailer_data),
757 "name not found in trailer, remaining data: "
758 + repr(trailer_data[offset:]),
759 )
760 break
761 key = cls.interpret_name(m.group(1))
762 assert isinstance(key, bytes)
763 value, value_offset = cls.get_value(trailer_data, m.end())
764 trailer[key] = value
765 if value_offset is None:
766 break
767 offset = value_offset
768 check_format_condition(
769 b"Size" in trailer and isinstance(trailer[b"Size"], int),
770 "/Size not in trailer or not an integer",
771 )
772 check_format_condition(
773 b"Root" in trailer and isinstance(trailer[b"Root"], IndirectReference),
774 "/Root not in trailer or not an indirect reference",
775 )
776 return trailer
778 re_hashes_in_name = re.compile(rb"([^#]*)(#([0-9a-fA-F]{2}))?")
780 @classmethod
781 def interpret_name(cls, raw: bytes, as_text: bool = False) -> str | bytes:
782 name = b""
783 for m in cls.re_hashes_in_name.finditer(raw):
784 if m.group(3):
785 name += m.group(1) + bytearray.fromhex(m.group(3).decode("us-ascii"))
786 else:
787 name += m.group(1)
788 if as_text:
789 return name.decode("utf-8")
790 else:
791 return bytes(name)
793 re_null = re.compile(whitespace_optional + rb"null(?=" + delimiter_or_ws + rb")")
794 re_true = re.compile(whitespace_optional + rb"true(?=" + delimiter_or_ws + rb")")
795 re_false = re.compile(whitespace_optional + rb"false(?=" + delimiter_or_ws + rb")")
796 re_int = re.compile(
797 whitespace_optional + rb"([-+]?[0-9]+)(?=" + delimiter_or_ws + rb")"
798 )
799 re_real = re.compile(
800 whitespace_optional
801 + rb"([-+]?([0-9]+\.[0-9]*|[0-9]*\.[0-9]+))(?="
802 + delimiter_or_ws
803 + rb")"
804 )
805 re_array_start = re.compile(whitespace_optional + rb"\[")
806 re_array_end = re.compile(whitespace_optional + rb"]")
807 re_string_hex = re.compile(
808 whitespace_optional + rb"<(" + whitespace_or_hex + rb"*)>"
809 )
810 re_string_lit = re.compile(whitespace_optional + rb"\(")
811 re_indirect_reference = re.compile(
812 whitespace_optional
813 + rb"([-+]?[0-9]+)"
814 + whitespace_mandatory
815 + rb"([-+]?[0-9]+)"
816 + whitespace_mandatory
817 + rb"R(?="
818 + delimiter_or_ws
819 + rb")"
820 )
821 re_indirect_def_start = re.compile(
822 whitespace_optional
823 + rb"([-+]?[0-9]+)"
824 + whitespace_mandatory
825 + rb"([-+]?[0-9]+)"
826 + whitespace_mandatory
827 + rb"obj(?="
828 + delimiter_or_ws
829 + rb")"
830 )
831 re_indirect_def_end = re.compile(
832 whitespace_optional + rb"endobj(?=" + delimiter_or_ws + rb")"
833 )
834 re_comment = re.compile(
835 rb"(" + whitespace_optional + rb"%[^\r\n]*" + newline + rb")*"
836 )
837 re_stream_start = re.compile(whitespace_optional + rb"stream\r?\n")
838 re_stream_end = re.compile(
839 whitespace_optional + rb"endstream(?=" + delimiter_or_ws + rb")"
840 )
842 @classmethod
843 def get_value(
844 cls,
845 data: bytes | bytearray | memoryview | mmap.mmap,
846 offset: int,
847 expect_indirect: IndirectReference | None = None,
848 max_nesting: int = 900,
849 ) -> tuple[Any, int | None]:
850 if max_nesting == 0:
851 return None, None
852 m = cls.re_comment.match(data, offset)
853 if m:
854 offset = m.end()
855 m = cls.re_indirect_def_start.match(data, offset)
856 if m:
857 check_format_condition(
858 int(m.group(1)) > 0,
859 "indirect object definition: object ID must be greater than 0",
860 )
861 check_format_condition(
862 int(m.group(2)) >= 0,
863 "indirect object definition: generation must be non-negative",
864 )
865 check_format_condition(
866 expect_indirect is None
867 or expect_indirect
868 == IndirectReference(int(m.group(1)), int(m.group(2))),
869 "indirect object definition different than expected",
870 )
871 object, object_offset = cls.get_value(
872 data, m.end(), max_nesting=max_nesting - 1
873 )
874 if object_offset is None:
875 return object, None
876 m = cls.re_indirect_def_end.match(data, object_offset)
877 check_format_condition(
878 m is not None, "indirect object definition end not found"
879 )
880 assert m is not None
881 return object, m.end()
882 check_format_condition(
883 not expect_indirect, "indirect object definition not found"
884 )
885 m = cls.re_indirect_reference.match(data, offset)
886 if m:
887 check_format_condition(
888 int(m.group(1)) > 0,
889 "indirect object reference: object ID must be greater than 0",
890 )
891 check_format_condition(
892 int(m.group(2)) >= 0,
893 "indirect object reference: generation must be non-negative",
894 )
895 return IndirectReference(int(m.group(1)), int(m.group(2))), m.end()
896 m = cls.re_dict_start.match(data, offset)
897 if m:
898 offset = m.end()
899 result: dict[Any, Any] = {}
900 m = cls.re_dict_end.match(data, offset)
901 current_offset: int | None = offset
902 while not m:
903 assert current_offset is not None
904 m = cls.re_name.match(data, current_offset)
905 if not m:
906 msg = "key must be a name"
907 raise PdfFormatError(msg)
908 key = PdfName(cls.interpret_name(m.group(1)))
909 value, current_offset = cls.get_value(
910 data, m.end(), max_nesting=max_nesting - 1
911 )
912 result[key] = value
913 if current_offset is None:
914 return result, None
915 m = cls.re_dict_end.match(data, current_offset)
916 current_offset = m.end()
917 m = cls.re_stream_start.match(data, current_offset)
918 if m:
919 stream_len = result.get(b"Length")
920 if stream_len is None or not isinstance(stream_len, int):
921 msg = f"bad or missing Length in stream dict ({stream_len})"
922 raise PdfFormatError(msg)
923 stream_data = bytes(data[m.end() : m.end() + stream_len])
924 m = cls.re_stream_end.match(data, m.end() + stream_len)
925 check_format_condition(m is not None, "stream end not found")
926 assert m is not None
927 current_offset = m.end()
928 return PdfStream(PdfDict(result), stream_data), current_offset
929 return PdfDict(result), current_offset
930 m = cls.re_array_start.match(data, offset)
931 if m:
932 offset = m.end()
933 results = []
934 m = cls.re_array_end.match(data, offset)
935 current_offset = offset
936 while not m:
937 assert current_offset is not None
938 value, current_offset = cls.get_value(
939 data, current_offset, max_nesting=max_nesting - 1
940 )
941 results.append(value)
942 if current_offset is None:
943 return results, None
944 m = cls.re_array_end.match(data, current_offset)
945 return results, m.end()
946 m = cls.re_null.match(data, offset)
947 if m:
948 return None, m.end()
949 m = cls.re_true.match(data, offset)
950 if m:
951 return True, m.end()
952 m = cls.re_false.match(data, offset)
953 if m:
954 return False, m.end()
955 m = cls.re_name.match(data, offset)
956 if m:
957 return PdfName(cls.interpret_name(m.group(1))), m.end()
958 m = cls.re_int.match(data, offset)
959 if m:
960 return int(m.group(1)), m.end()
961 m = cls.re_real.match(data, offset)
962 if m:
963 # XXX Decimal instead of float???
964 return float(m.group(1)), m.end()
965 m = cls.re_string_hex.match(data, offset)
966 if m:
967 # filter out whitespace
968 hex_string = bytearray(
969 b for b in m.group(1) if b in b"0123456789abcdefABCDEF"
970 )
971 if len(hex_string) % 2 == 1:
972 # append a 0 if the length is not even - yes, at the end
973 hex_string.append(ord(b"0"))
974 return bytearray.fromhex(hex_string.decode("us-ascii")), m.end()
975 m = cls.re_string_lit.match(data, offset)
976 if m:
977 return cls.get_literal_string(data, m.end())
978 # return None, offset # fallback (only for debugging)
979 msg = f"unrecognized object: {repr(data[offset : offset + 32])}"
980 raise PdfFormatError(msg)
982 re_lit_str_token = re.compile(
983 rb"(\\[nrtbf()\\])|(\\[0-9]{1,3})|(\\(\r\n|\r|\n))|(\r\n|\r|\n)|(\()|(\))"
984 )
985 escaped_chars = {
986 b"n": b"\n",
987 b"r": b"\r",
988 b"t": b"\t",
989 b"b": b"\b",
990 b"f": b"\f",
991 b"(": b"(",
992 b")": b")",
993 b"\\": b"\\",
994 ord(b"n"): b"\n",
995 ord(b"r"): b"\r",
996 ord(b"t"): b"\t",
997 ord(b"b"): b"\b",
998 ord(b"f"): b"\f",
999 ord(b"("): b"(",
1000 ord(b")"): b")",
1001 ord(b"\\"): b"\\",
1002 }
1004 @classmethod
1005 def get_literal_string(
1006 cls, data: bytes | bytearray | memoryview | mmap.mmap, offset: int
1007 ) -> tuple[bytes, int]:
1008 nesting_depth = 0
1009 result = bytearray()
1010 for m in cls.re_lit_str_token.finditer(data, offset):
1011 result.extend(data[offset : m.start()])
1012 if m.group(1):
1013 result.extend(cls.escaped_chars[m.group(1)[1]])
1014 elif m.group(2):
1015 result.append(int(m.group(2)[1:], 8))
1016 elif m.group(3):
1017 pass
1018 elif m.group(5):
1019 result.extend(b"\n")
1020 elif m.group(6):
1021 result.extend(b"(")
1022 nesting_depth += 1
1023 elif m.group(7):
1024 if nesting_depth == 0:
1025 return bytes(result), m.end()
1026 result.extend(b")")
1027 nesting_depth -= 1
1028 offset = m.end()
1029 msg = "unfinished literal string"
1030 raise PdfFormatError(msg)
1032 re_xref_section_start = re.compile(whitespace_optional + rb"xref" + newline)
1033 re_xref_subsection_start = re.compile(
1034 whitespace_optional
1035 + rb"([0-9]+)"
1036 + whitespace_mandatory
1037 + rb"([0-9]+)"
1038 + whitespace_optional
1039 + newline_only
1040 )
1041 re_xref_entry = re.compile(rb"([0-9]{10}) ([0-9]{5}) ([fn])( \r| \n|\r\n)")
1043 def read_xref_table(self, xref_section_offset: int) -> int:
1044 assert self.buf is not None
1045 subsection_found = False
1046 m = self.re_xref_section_start.match(
1047 self.buf, xref_section_offset + self.start_offset
1048 )
1049 check_format_condition(m is not None, "xref section start not found")
1050 assert m is not None
1051 offset = m.end()
1052 while True:
1053 m = self.re_xref_subsection_start.match(self.buf, offset)
1054 if not m:
1055 check_format_condition(
1056 subsection_found, "xref subsection start not found"
1057 )
1058 break
1059 subsection_found = True
1060 offset = m.end()
1061 first_object = int(m.group(1))
1062 num_objects = int(m.group(2))
1063 for i in range(first_object, first_object + num_objects):
1064 m = self.re_xref_entry.match(self.buf, offset)
1065 check_format_condition(m is not None, "xref entry not found")
1066 assert m is not None
1067 offset = m.end()
1068 is_free = m.group(3) == b"f"
1069 if not is_free:
1070 generation = int(m.group(2))
1071 new_entry = (int(m.group(1)), generation)
1072 if i not in self.xref_table:
1073 self.xref_table[i] = new_entry
1074 return offset
1076 def read_indirect(self, ref: IndirectReference, max_nesting: int = 900) -> Any:
1077 offset, generation = self.xref_table[ref[0]]
1078 check_format_condition(
1079 generation == ref[1],
1080 f"expected to find generation {ref[1]} for object ID {ref[0]} in xref "
1081 f"table, instead found generation {generation} at offset {offset}",
1082 )
1083 assert self.buf is not None
1084 value = self.get_value(
1085 self.buf,
1086 offset + self.start_offset,
1087 expect_indirect=IndirectReference(*ref),
1088 max_nesting=max_nesting,
1089 )[0]
1090 self.cached_objects[ref] = value
1091 return value
1093 def linearize_page_tree(
1094 self, node: PdfDict | None = None
1095 ) -> list[IndirectReference]:
1096 processed_ids = set()
1097 if node is not None:
1098 page_node = node
1099 else:
1100 page_node = self.page_tree_root
1101 if self.pages_ref is not None:
1102 processed_ids.add(self.pages_ref.object_id)
1103 check_format_condition(
1104 page_node[b"Type"] == b"Pages", "/Type of page tree node is not /Pages"
1105 )
1106 pages = []
1107 # Walk the tree with a stack of iterators rather than recursion,
1108 # so that a deep page tree cannot exhaust the call stack
1109 stack = [iter(page_node[b"Kids"])]
1110 while stack:
1111 kid = next(stack[-1], None)
1112 if kid is None:
1113 stack.pop()
1114 continue
1115 check_format_condition(
1116 kid.object_id not in processed_ids,
1117 f"page tree contains a cyclic or duplicate reference to {kid}",
1118 )
1119 processed_ids.add(kid.object_id)
1120 kid_object = self.read_indirect(kid)
1121 if kid_object[b"Type"] == b"Page":
1122 pages.append(kid)
1123 else:
1124 check_format_condition(
1125 kid_object[b"Type"] == b"Pages",
1126 "/Type of page tree node is not /Pages",
1127 )
1128 stack.append(iter(kid_object[b"Kids"]))
1129 return pages