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