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

716 statements  

1from __future__ import annotations 

2 

3__lazy_modules__ = {"calendar", "mmap", "zlib"} 

4 

5import calendar 

6import codecs 

7import collections 

8import mmap 

9import os 

10import re 

11import sys 

12import time 

13import zlib 

14from typing import Any, NamedTuple 

15 

16from . import ImageFile 

17 

18TYPE_CHECKING = False 

19if TYPE_CHECKING: 

20 from typing import IO, Self 

21 

22 _DictBase = collections.UserDict[str | bytes, Any] 

23else: 

24 _DictBase = collections.UserDict 

25 

26 

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") 

31 

32 

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} 

76 

77 

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) 

83 

84 

85class PdfFormatError(RuntimeError): 

86 """An error that probably indicates a syntactic or semantic error in the 

87 PDF file structure""" 

88 

89 pass 

90 

91 

92def check_format_condition(condition: bool, error_message: str) -> None: 

93 if not condition: 

94 raise PdfFormatError(error_message) 

95 

96 

97class IndirectReferenceTuple(NamedTuple): 

98 object_id: int 

99 generation: int 

100 

101 

102class IndirectReference(IndirectReferenceTuple): 

103 def __str__(self) -> str: 

104 return f"{self.object_id} {self.generation} R" 

105 

106 def __bytes__(self) -> bytes: 

107 return self.__str__().encode("us-ascii") 

108 

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 

114 

115 def __ne__(self, other: object) -> bool: 

116 return not (self == other) 

117 

118 def __hash__(self) -> int: 

119 return hash((self.object_id, self.generation)) 

120 

121 

122class IndirectObjectDef(IndirectReference): 

123 def __str__(self) -> str: 

124 return f"{self.object_id} {self.generation} obj" 

125 

126 

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 

138 

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] 

148 

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] 

154 

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) 

168 

169 def __contains__(self, key: int) -> bool: 

170 return key in self.existing_entries or key in self.new_entries 

171 

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 ) 

178 

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()) 

183 

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 

222 

223 

224class PdfName: 

225 name: bytes 

226 

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") 

234 

235 def name_as_str(self) -> str: 

236 return self.name.decode("us-ascii") 

237 

238 def __eq__(self, other: object) -> bool: 

239 return ( 

240 isinstance(other, PdfName) and other.name == self.name 

241 ) or other == self.name 

242 

243 def __hash__(self) -> int: 

244 return hash(self.name) 

245 

246 def __repr__(self) -> str: 

247 return f"{self.__class__.__name__}({repr(self.name)})" 

248 

249 @classmethod 

250 def from_pdf_stream(cls, data: bytes) -> PdfName: 

251 return cls(PdfParser.interpret_name(data)) 

252 

253 allowed_chars = set(range(33, 127)) - {ord(c) for c in "#%/()<>[]{}"} 

254 

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) 

263 

264 

265class PdfArray(list[Any]): 

266 def __bytes__(self) -> bytes: 

267 return b"[ " + b" ".join(pdf_repr(x) for x in self) + b" ]" 

268 

269 

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 

276 

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:] 

287 

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]) 

294 

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 

303 

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) 

316 

317 

318class PdfBinary: 

319 def __init__(self, data: list[int] | bytes) -> None: 

320 self.data = data 

321 

322 def __bytes__(self) -> bytes: 

323 return b"<%s>" % b"".join(b"%02X" % b for b in self.data) 

324 

325 

326class PdfStream: 

327 def __init__(self, dictionary: PdfDict, buf: bytes) -> None: 

328 self.dictionary = dictionary 

329 self.buf = buf 

330 

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) 

346 

347 

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) 

375 

376 

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 """ 

382 

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() 

435 

436 def __enter__(self) -> Self: 

437 return self 

438 

439 def __exit__(self, *args: object) -> None: 

440 self.close() 

441 

442 def start_writing(self) -> None: 

443 self.close_buf() 

444 self.seek_end() 

445 

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 

451 

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 

457 

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 

464 

465 def seek_end(self) -> None: 

466 assert self.f is not None 

467 self.f.seek(0, os.SEEK_END) 

468 

469 def write_header(self) -> None: 

470 assert self.f is not None 

471 self.f.write(b"%PDF-1.4\n") 

472 

473 def write_comment(self, s: str) -> None: 

474 assert self.f is not None 

475 self.f.write(f"% {s}\n".encode()) 

476 

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 

491 

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 = [] 

522 

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 ) 

548 

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) 

558 

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 

582 

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] 

588 

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"" 

600 

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[:] 

636 

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 

643 

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 ) 

683 

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"]) 

706 

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"] 

736 

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) 

746 

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 

777 

778 re_hashes_in_name = re.compile(rb"([^#]*)(#([0-9a-fA-F]{2}))?") 

779 

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) 

792 

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 ) 

841 

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) 

981 

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 } 

1003 

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) 

1031 

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)") 

1042 

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 

1075 

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 

1092 

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