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
1# Copyright (c) 2006, Mathieu Fenniak
2# All rights reserved.
3#
4# Redistribution and use in source and binary forms, with or without
5# modification, are permitted provided that the following conditions are
6# met:
7#
8# * Redistributions of source code must retain the above copyright notice,
9# this list of conditions and the following disclaimer.
10# * Redistributions in binary form must reproduce the above copyright notice,
11# this list of conditions and the following disclaimer in the documentation
12# and/or other materials provided with the distribution.
13# * The name of the author may not be used to endorse or promote products
14# derived from this software without specific prior written permission.
15#
16# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
17# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
20# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
21# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
22# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
23# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
24# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
25# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
26# POSSIBILITY OF SUCH DAMAGE.
28__author__ = "Mathieu Fenniak"
29__author_email__ = "biziqe@mathieu.fenniak.net"
31import logging
32import os
33import re
34import sys
35from collections.abc import Iterable, Sequence
36from io import BytesIO
37from math import ceil
38from typing import (
39 Any,
40 Callable,
41 NamedTuple,
42 Optional,
43 Union,
44 cast,
45)
47from .. import get_configuration
48from .._protocols import PdfReaderProtocol, PdfWriterProtocol, XmpInformationProtocol
49from .._utils import (
50 WHITESPACES,
51 BinaryStreamType,
52 StreamType,
53 deprecation_no_replacement,
54 logger_warning,
55 read_non_whitespace,
56 read_until_regex,
57 read_until_whitespace,
58 skip_over_comment,
59 skip_over_comments,
60)
61from ..constants import (
62 CheckboxRadioButtonAttributes,
63 FieldDictionaryAttributes,
64 OutlineFontFlag,
65 StreamAttributes,
66)
67from ..constants import FilterTypes as FT
68from ..constants import TypArguments as TA
69from ..constants import TypFitArguments as TF
70from ..errors import STREAM_TRUNCATED_PREMATURELY, LimitReachedError, PdfReadError, PdfStreamError
71from ._base import (
72 BooleanObject,
73 ByteStringObject,
74 FloatObject,
75 IndirectObject,
76 NameObject,
77 NullObject,
78 NumberObject,
79 PdfObject,
80 TextStringObject,
81 is_null_or_none,
82)
83from ._fit import Fit
84from ._image_inline import (
85 extract_inline__ascii85_decode,
86 extract_inline__ascii_hex_decode,
87 extract_inline__dct_decode,
88 extract_inline__run_length_decode,
89 extract_inline_default,
90)
91from ._utils import StreamParsingBudget, read_hex_string_from_stream, read_string_from_stream
93if sys.version_info >= (3, 11):
94 from typing import Self
95else:
96 from typing_extensions import Self
98logger = logging.getLogger(__name__)
100IndirectPattern = re.compile(rb"[+-]?(\d+)\s+(\d+)\s+R[^a-zA-Z]")
102# Direct objects of these types are neither streams nor indirect objects.
103# Large arrays consist mostly of them, for example in the /ParentTree of the
104# structure tree, so the clone loops check the exact type first and skip the
105# slower isinstance() checks.
106_SCALAR_TYPES = frozenset(
107 (BooleanObject, ByteStringObject, FloatObject, NameObject, NullObject, NumberObject, TextStringObject)
108)
111class ArrayObject(list[Any], PdfObject):
112 def replicate(
113 self,
114 pdf_dest: PdfWriterProtocol,
115 ) -> "ArrayObject":
116 arr = cast(
117 "ArrayObject",
118 self._reference_clone(ArrayObject(), pdf_dest, False),
119 )
120 for data in self:
121 if hasattr(data, "replicate"):
122 arr.append(data.replicate(pdf_dest))
123 else:
124 arr.append(data)
125 return arr
127 def clone(
128 self,
129 pdf_dest: PdfWriterProtocol,
130 force_duplicate: bool = False,
131 ignore_fields: Optional[Sequence[Union[str, int]]] = (),
132 ) -> "ArrayObject":
133 """Clone object into pdf_dest."""
134 try:
135 if self.indirect_reference.pdf == pdf_dest and not force_duplicate: # type: ignore[union-attr]
136 return self
137 except Exception:
138 pass
139 arr = cast(
140 "ArrayObject",
141 self._reference_clone(ArrayObject(), pdf_dest, force_duplicate=True),
142 )
143 for data in self:
144 if type(data) in _SCALAR_TYPES:
145 # Same as the generic case below, but without the isinstance() checks.
146 arr.append(data.clone(pdf_dest, force_duplicate, ignore_fields))
147 elif isinstance(data, StreamObject):
148 dup = data._reference_clone(
149 data.clone(pdf_dest, force_duplicate, ignore_fields),
150 pdf_dest,
151 force_duplicate,
152 )
153 arr.append(dup.indirect_reference)
154 elif isinstance(data, IndirectObject) and isinstance(resolved := data.get_object(), StreamObject):
155 dup = data._reference_clone(
156 resolved.clone(pdf_dest, force_duplicate=True, ignore_fields=ignore_fields),
157 pdf_dest,
158 force_duplicate,
159 )
160 arr.append(dup.indirect_reference)
161 elif hasattr(data, "clone"):
162 arr.append(data.clone(pdf_dest, force_duplicate, ignore_fields))
163 else:
164 arr.append(data)
165 return arr
167 def hash_bin(self) -> int:
168 """
169 Used to detect modified object.
171 Returns:
172 Hash considering type and value.
174 """
175 return hash((self.__class__, tuple(x.hash_bin() for x in self)))
177 def items(self) -> Iterable[Any]:
178 """Emulate DictionaryObject.items for a list (index, object)."""
179 return enumerate(self)
181 def _to_lst(self, lst: Any) -> list[Any]:
182 # Convert to list, internal
183 result: list[Any]
184 if isinstance(lst, (list, tuple, set)):
185 result = list(lst)
186 elif isinstance(lst, PdfObject):
187 result = [lst]
188 elif isinstance(lst, str):
189 if lst[0] == "/":
190 result = [NameObject(lst)]
191 else:
192 result = [TextStringObject(lst)]
193 elif isinstance(lst, bytes):
194 result = [ByteStringObject(lst)]
195 else: # for numbers,...
196 result = [lst]
197 return result
199 def __add__(self, lst: Any) -> "ArrayObject":
200 """
201 Allow extension by adding list or add one element only
203 Args:
204 lst: any list, tuples are extended the list.
205 other types(numbers,...) will be appended.
206 if str is passed it will be converted into TextStringObject
207 or NameObject (if starting with "/")
208 if bytes is passed it will be converted into ByteStringObject
210 Returns:
211 ArrayObject with all elements
213 """
214 temp = ArrayObject(self)
215 temp.extend(self._to_lst(lst))
216 return temp
218 def __iadd__(self, lst: Any) -> Self:
219 """
220 Allow extension by adding list or add one element only
222 Args:
223 lst: any list, tuples are extended the list.
224 other types(numbers,...) will be appended.
225 if str is passed it will be converted into TextStringObject
226 or NameObject (if starting with "/")
227 if bytes is passed it will be converted into ByteStringObject
229 """
230 self.extend(self._to_lst(lst))
231 return self
233 def __isub__(self, lst: Any) -> Self:
234 """Allow to remove items"""
235 for x in self._to_lst(lst):
236 try:
237 index = self.index(x)
238 del self[index]
239 except ValueError:
240 pass
241 return self
243 def write_to_stream(
244 self, stream: StreamType, encryption_key: Union[str, bytes, None] = None
245 ) -> None:
246 if encryption_key is not None: # deprecated
247 deprecation_no_replacement(
248 "the encryption_key parameter of write_to_stream", "5.0.0"
249 )
250 stream.write(b"[")
251 for data in self:
252 stream.write(b" ")
253 data.write_to_stream(stream)
254 stream.write(b" ]")
256 @staticmethod
257 def _check_limits(*, remaining_elements: int, parsing_budget: Optional[StreamParsingBudget]) -> int:
258 remaining_elements -= 1
259 if remaining_elements < 0:
260 raise LimitReachedError("Reached maximum array element count.")
261 if parsing_budget is not None:
262 parsing_budget.remaining_budget -= 1
263 if parsing_budget.remaining_budget < 0:
264 raise LimitReachedError("Reached maximum number of tokens.")
265 return remaining_elements
267 @staticmethod
268 def read_from_stream(
269 stream: StreamType,
270 pdf: Optional[PdfReaderProtocol],
271 forced_encoding: Union[str, list[str], dict[int, str], None] = None,
272 *,
273 parsing_budget: Optional[StreamParsingBudget] = None,
274 ) -> "ArrayObject":
275 arr = ArrayObject()
276 tmp = stream.read(1)
277 if tmp != b"[":
278 raise PdfReadError("Could not read array")
279 remaining_elements = get_configuration().array_maximum_element_count
280 while True:
281 # skip leading whitespace
282 tok = stream.read(1)
283 while tok.isspace() or tok in WHITESPACES:
284 tok = stream.read(1)
285 if tok == b"":
286 break
287 if tok == b"%":
288 stream.seek(-1, 1)
289 skip_over_comment(stream)
290 continue
291 # check for array ending
292 if tok == b"]":
293 break
295 remaining_elements = ArrayObject._check_limits(
296 remaining_elements=remaining_elements, parsing_budget=parsing_budget
297 )
298 if tok == b"n":
299 # Shortcut for null, as some arrays consist mostly of nulls
300 rest = stream.read(3)
301 if rest == b"ull":
302 arr.append(NullObject())
303 continue
304 stream.seek(-len(rest), 1)
305 stream.seek(-1, 1)
306 # read and append object
307 arr.append(read_object(stream, pdf, forced_encoding, parsing_budget=parsing_budget))
308 return arr
311class DictionaryObject(dict[Any, Any], PdfObject):
312 def replicate(
313 self,
314 pdf_dest: PdfWriterProtocol,
315 ) -> "DictionaryObject":
316 d__ = cast(
317 "DictionaryObject",
318 self._reference_clone(self.__class__(), pdf_dest, False),
319 )
320 for k, v in self.items():
321 d__[k.replicate(pdf_dest)] = (
322 v.replicate(pdf_dest) if hasattr(v, "replicate") else v
323 )
324 return d__
326 def clone(
327 self,
328 pdf_dest: PdfWriterProtocol,
329 force_duplicate: bool = False,
330 ignore_fields: Optional[Sequence[Union[str, int]]] = (),
331 ) -> "DictionaryObject":
332 """Clone object into pdf_dest."""
333 try:
334 if self.indirect_reference.pdf == pdf_dest and not force_duplicate: # type: ignore[union-attr]
335 return self
336 except Exception:
337 pass
339 visited: set[tuple[int, int]] = set() # (idnum, generation)
340 d__ = cast(
341 "DictionaryObject",
342 self._reference_clone(self.__class__(), pdf_dest, force_duplicate),
343 )
344 if ignore_fields is None:
345 ignore_fields = []
346 if len(d__.keys()) == 0:
347 d__._clone(self, pdf_dest, force_duplicate, ignore_fields, visited)
348 return d__
350 def _clone(
351 self,
352 src: "DictionaryObject",
353 pdf_dest: PdfWriterProtocol,
354 force_duplicate: bool,
355 ignore_fields: Optional[Sequence[Union[str, int]]],
356 visited: set[tuple[int, int]], # (idnum, generation)
357 ) -> None:
358 """
359 Update the object from src.
361 Args:
362 src: "DictionaryObject":
363 pdf_dest:
364 force_duplicate:
365 ignore_fields:
367 """
368 # First we remove the ignore_fields
369 # that are for a limited number of levels
370 assert ignore_fields is not None
371 ignore_fields = list(ignore_fields)
372 x = 0
373 while x < len(ignore_fields):
374 if isinstance(ignore_fields[x], int):
375 if cast(int, ignore_fields[x]) <= 0:
376 del ignore_fields[x]
377 del ignore_fields[x]
378 continue
379 ignore_fields[x] -= 1 # type:ignore
380 x += 1
381 # Check if this is a chain list, we need to loop to prevent recur
382 if any(
383 field not in ignore_fields
384 and field in src
385 and isinstance(src.raw_get(field), IndirectObject)
386 and isinstance(src[field], DictionaryObject)
387 and (
388 src.get("/Type", None) is None
389 or cast(DictionaryObject, src[field]).get("/Type", None) is None
390 or src.get("/Type", None)
391 == cast(DictionaryObject, src[field]).get("/Type", None)
392 )
393 for field in ["/Next", "/Prev", "/N", "/V"]
394 ):
395 ignore_fields = list(ignore_fields)
396 for lst in (("/Next", "/Prev"), ("/N", "/V")):
397 for k in lst:
398 objs = []
399 if (
400 k in src
401 and k not in self
402 and isinstance(src.raw_get(k), IndirectObject)
403 and isinstance(src[k], DictionaryObject)
404 # If need to go further the idea is to check
405 # that the types are the same
406 and (
407 src.get("/Type", None) is None
408 or cast(DictionaryObject, src[k]).get("/Type", None) is None
409 or src.get("/Type", None)
410 == cast(DictionaryObject, src[k]).get("/Type", None)
411 )
412 ):
413 cur_obj: Optional[DictionaryObject] = cast(
414 "DictionaryObject", src[k]
415 )
416 prev_obj: Optional[DictionaryObject] = self
417 while cur_obj is not None:
418 clon = cast(
419 "DictionaryObject",
420 cur_obj._reference_clone(
421 cur_obj.__class__(), pdf_dest, force_duplicate
422 ),
423 )
424 # Check to see if we've previously processed our item
425 if clon.indirect_reference is not None:
426 idnum = clon.indirect_reference.idnum
427 generation = clon.indirect_reference.generation
428 if (idnum, generation) in visited:
429 cur_obj = None
430 break
431 visited.add((idnum, generation))
432 objs.append((cur_obj, clon))
433 assert prev_obj is not None
434 prev_obj[NameObject(k)] = clon.indirect_reference
435 prev_obj = clon
436 try:
437 if cur_obj == src:
438 cur_obj = None
439 else:
440 cur_obj = cast("DictionaryObject", cur_obj[k])
441 except Exception:
442 cur_obj = None
443 for s, c in objs:
444 c._clone(
445 s, pdf_dest, force_duplicate, ignore_fields, visited
446 )
448 for k, v in src.items():
449 if k not in ignore_fields:
450 # The type check does not change the result. It only skips the
451 # slower isinstance() check for the most common values.
452 if type(v) not in _SCALAR_TYPES and isinstance(v, StreamObject):
453 if not hasattr(v, "indirect_reference"):
454 v.indirect_reference = None
455 vv = v.clone(pdf_dest, force_duplicate, ignore_fields)
456 assert vv.indirect_reference is not None
457 self[k.clone(pdf_dest)] = vv.indirect_reference
458 elif k not in self:
459 self[NameObject(k)] = (
460 v.clone(pdf_dest, force_duplicate, ignore_fields)
461 if hasattr(v, "clone")
462 else v
463 )
465 def hash_bin(self) -> int:
466 """
467 Used to detect modified object.
469 Returns:
470 Hash considering type and value.
472 """
473 return hash(
474 (self.__class__, tuple(((k, v.hash_bin()) for k, v in self.items())))
475 )
477 def raw_get(self, key: Any) -> Any:
478 return dict.__getitem__(self, key)
480 def get_inherited(self, key: str, default: Any = None) -> Any:
481 """
482 Returns the value of a key or from the parent if not found.
483 If not found returns default.
485 Args:
486 key: string identifying the field to return
488 default: default value to return
490 Returns:
491 Current key or inherited one, otherwise default value.
493 """
494 current = self
495 visited: set[int] = set()
497 while True:
498 # Detect cyclic parent references
499 obj_id = id(current)
500 if obj_id in visited:
501 raise LimitReachedError(f"Detected cycle in /Parent hierarchy when retrieving value for key {key!r}.")
502 visited.add(obj_id)
504 if key in current:
505 return current[key]
507 if "/Parent" not in current:
508 return default
510 # Walk upward
511 current = cast(
512 "DictionaryObject",
513 current["/Parent"].get_object(),
514 )
516 def __setitem__(self, key: Any, value: Any) -> Any:
517 if not isinstance(key, PdfObject):
518 raise ValueError("Key must be a PdfObject")
519 if not isinstance(value, PdfObject):
520 raise ValueError("Value must be a PdfObject")
521 return dict.__setitem__(self, key, value)
523 def setdefault(self, key: Any, value: Optional[Any] = None) -> Any:
524 if not isinstance(key, PdfObject):
525 raise ValueError("Key must be a PdfObject")
526 if not isinstance(value, PdfObject):
527 raise ValueError("Value must be a PdfObject")
528 return dict.setdefault(self, key, value)
530 def __getitem__(self, key: Any) -> PdfObject:
531 return cast(PdfObject, dict.__getitem__(self, key).get_object())
533 @property
534 def xmp_metadata(self) -> Optional[XmpInformationProtocol]:
535 """
536 Retrieve XMP (Extensible Metadata Platform) data relevant to this
537 object, if available.
539 See Table 347 — Additional entries in a metadata stream dictionary.
541 Returns:
542 Returns a :class:`~pypdf.xmp.XmpInformation` instance
543 that can be used to access XMP metadata from the document. Can also
544 return None if no metadata was found on the document root.
546 """
547 from ..xmp import XmpInformation # noqa: PLC0415
549 metadata = self.get("/Metadata", None)
550 if is_null_or_none(metadata):
551 return None
552 assert metadata is not None, "mypy"
553 metadata = metadata.get_object()
554 return XmpInformation(metadata)
556 def write_to_stream(
557 self, stream: StreamType, encryption_key: Union[str, bytes, None] = None
558 ) -> None:
559 if encryption_key is not None: # deprecated
560 deprecation_no_replacement(
561 "the encryption_key parameter of write_to_stream", "5.0.0"
562 )
563 stream.write(b"<<\n")
564 for key, value in self.items():
565 if len(key) > 2 and key[1] == "%" and key[-1] == "%":
566 continue
567 key.write_to_stream(stream, encryption_key)
568 stream.write(b" ")
569 value.write_to_stream(stream)
570 stream.write(b"\n")
571 stream.write(b">>")
573 @classmethod
574 def _get_next_object_position(
575 cls, position_before: int, position_end: int, generations: list[int], pdf: PdfReaderProtocol
576 ) -> int:
577 out = position_end
578 for generation in generations:
579 for x in pdf.xref[generation].values():
580 if position_before < x <= position_end:
581 out = min(out, x)
582 return out
584 @classmethod
585 def _read_unsized_from_stream(
586 cls, *, stream: BinaryStreamType, pdf: PdfReaderProtocol, length: int,
587 ) -> bytes:
588 current_position = stream.tell()
590 # Determine stream size.
591 try:
592 stream.seek(0, os.SEEK_END)
593 stream_length = stream.tell()
594 finally:
595 stream.seek(current_position)
597 object_position = cls._get_next_object_position(
598 position_before=current_position, position_end=stream_length, generations=list(pdf.xref), pdf=pdf
599 )
601 bytes_to_read = object_position - current_position
602 if bytes_to_read >= length:
603 raise LimitReachedError(f"Requested length of {bytes_to_read} exceeds maximum allowed length.")
605 # Read until the next object position.
606 read_value = stream.read(bytes_to_read)
607 endstream_position = read_value.find(b"endstream")
608 if endstream_position < 0:
609 raise PdfReadError(
610 f"Unable to find 'endstream' marker for obj starting at {current_position}."
611 )
612 # 9 = len(b"endstream")
613 stream.seek(current_position + endstream_position + 9)
614 return read_value[: endstream_position - 1]
616 @staticmethod
617 def read_from_stream(
618 stream: StreamType,
619 pdf: Optional[PdfReaderProtocol],
620 forced_encoding: Union[str, list[str], dict[int, str], None] = None,
621 *,
622 parsing_budget: Optional[StreamParsingBudget] = None,
623 ) -> "DictionaryObject":
624 tmp = stream.read(2)
625 if tmp != b"<<":
626 raise PdfReadError(
627 f"Dictionary read error at byte {hex(stream.tell())}: "
628 "stream must begin with '<<'"
629 )
630 data: dict[Any, Any] = {}
631 remaining_count = get_configuration().dictionary_maximum_element_count
632 while True:
633 tok = read_non_whitespace(stream)
634 if tok == b"\x00":
635 continue
636 if tok == b"%":
637 stream.seek(-1, 1)
638 skip_over_comment(stream)
639 continue
640 if not tok:
641 raise PdfStreamError(STREAM_TRUNCATED_PREMATURELY)
643 if tok == b">":
644 stream.read(1)
645 break
646 stream.seek(-1, 1)
648 remaining_count -= 1
649 if remaining_count < 0:
650 raise LimitReachedError("Reached maximum number of dictionary entries.")
651 if parsing_budget is not None:
652 parsing_budget.remaining_budget -= 2
653 if parsing_budget.remaining_budget < 0:
654 raise LimitReachedError("Reached maximum number of tokens.")
656 try:
657 try:
658 key = read_object(stream, pdf, parsing_budget=parsing_budget)
659 if isinstance(key, NullObject):
660 break
661 if not isinstance(key, NameObject):
662 raise PdfReadError(
663 f"Expecting a NameObject for key but found {key!r}"
664 )
665 except PdfReadError as exc:
666 if pdf is not None and pdf.strict:
667 raise
668 logger_warning("%(exception)r", source=__name__, exception=exc)
669 continue
670 tok = read_non_whitespace(stream)
671 stream.seek(-1, 1)
672 value = read_object(stream, pdf, forced_encoding, parsing_budget=parsing_budget)
673 except LimitReachedError:
674 # Always propagate security issues.
675 raise
676 except RecursionError as exc:
677 raise PdfReadError(exc.__repr__()) from exc
678 except Exception as exc:
679 if pdf is not None and pdf.strict:
680 raise PdfReadError(exc.__repr__()) from exc
681 logger_warning("%(exception)r", source=__name__, exception=exc)
682 retval = DictionaryObject()
683 retval.update(data)
684 return retval # return partial data
686 if key not in data:
687 data[key] = value
688 else:
689 # multiple definitions of key not permitted
690 msg = (
691 "Multiple definitions in dictionary at byte "
692 "%(position)s for key %(key)s"
693 )
694 values = {"position": hex(stream.tell()), "key": key}
695 if pdf is not None and pdf.strict:
696 raise PdfReadError(msg % values)
697 logger_warning(msg, source=__name__, **values)
699 DictionaryObject._read_stream_data(stream=stream, pdf=pdf, data=data)
700 if "__streamdata__" in data:
701 return StreamObject.initialize_from_dictionary(data)
702 retval = DictionaryObject()
703 retval.update(data)
704 return retval
706 @staticmethod
707 def _read_stream_data(
708 *,
709 stream: StreamType,
710 pdf: Optional[PdfReaderProtocol],
711 data: dict[Any, Any]
712 ) -> None:
713 pos = stream.tell()
714 s = read_non_whitespace(stream)
715 if s == b"s" and stream.read(5) == b"tream":
716 eol = stream.read(1)
717 # Occasional PDF file output has spaces after 'stream' keyword but before EOL.
718 # patch provided by Danial Sandler
719 while eol == b" ":
720 eol = stream.read(1)
721 if eol not in (b"\n", b"\r"):
722 raise PdfStreamError("Stream data must be followed by a newline")
723 if eol == b"\r" and stream.read(1) != b"\n":
724 stream.seek(-1, 1)
725 # this is a stream object, not a dictionary
726 if StreamAttributes.LENGTH not in data:
727 if pdf is not None and pdf.strict:
728 raise PdfStreamError("Stream length not defined")
729 logger_warning(
730 "Stream length not defined @pos=%(position)d",
731 source=__name__,
732 position=stream.tell(),
733 )
734 data[NameObject(StreamAttributes.LENGTH)] = NumberObject(-1)
735 length = data[StreamAttributes.LENGTH]
736 if isinstance(length, IndirectObject):
737 t = stream.tell()
738 assert pdf is not None, "mypy"
739 length = pdf.get_object(length)
740 stream.seek(t, 0)
741 if length is None: # if the PDF is damaged
742 length = -1
743 pstart = stream.tell()
745 from .._configuration import get_configuration # noqa: PLC0415
746 configuration = get_configuration()
747 if length >= 0:
748 if length > configuration.maximum_declared_stream_length:
749 raise LimitReachedError(f"Declared stream length of {length} exceeds maximum allowed length.")
751 data["__streamdata__"] = stream.read(length)
752 else:
753 data["__streamdata__"] = read_until_regex(
754 stream=stream, regex=re.compile(b"endstream"), length=configuration.maximum_declared_stream_length,
755 )
756 e = read_non_whitespace(stream)
757 ndstream = stream.read(8)
758 if (e + ndstream) != b"endstream":
759 # the odd PDF file has a length that is too long, so
760 # we need to read backwards to find the "endstream" ending.
761 # ReportLab (unknown version) generates files with this bug,
762 # and Python users into PDF files tend to be our audience.
763 # we need to do this to correct the streamdata and chop off
764 # an extra character.
765 pos = stream.tell()
766 stream.seek(-10, 1)
767 end = stream.read(9)
768 if end == b"endstream":
769 # we found it by looking back one character further.
770 data["__streamdata__"] = data["__streamdata__"][:-1]
771 elif pdf is not None and not pdf.strict:
772 stream.seek(pstart, 0)
773 data["__streamdata__"] = DictionaryObject._read_unsized_from_stream(
774 stream=stream, pdf=pdf, length=configuration.maximum_declared_stream_length
775 )
776 pos = stream.tell()
777 else:
778 stream.seek(pos, 0)
779 raise PdfReadError(
780 "Unable to find 'endstream' marker after stream at byte "
781 f"{hex(stream.tell())} (nd='{ndstream!r}', end='{end!r}')."
782 )
783 else:
784 stream.seek(pos, 0)
787class TreeObject(DictionaryObject):
788 def __init__(self, dct: Optional[DictionaryObject] = None) -> None:
789 DictionaryObject.__init__(self)
790 if dct:
791 self.update(dct)
793 def has_children(self) -> bool:
794 return "/First" in self
796 def __iter__(self) -> Any:
797 return self.children()
799 def children(self) -> Iterable[Any]:
800 if not self.has_children():
801 return
803 child_ref = self[NameObject("/First")]
804 last = self[NameObject("/Last")]
805 child = child_ref.get_object()
806 visited: set[int] = set()
807 while True:
808 child_id = id(child)
809 if child_id in visited:
810 logger_warning("Detected cycle in outline structure for %(child)s", source=__name__, child=child)
811 return
812 visited.add(child_id)
814 yield child
816 if child == last:
817 return
818 child_ref = child.get(NameObject("/Next")) # type: ignore[union-attr]
819 if is_null_or_none(child_ref):
820 return
821 child = child_ref.get_object()
823 def add_child(self, child: Any, pdf: PdfWriterProtocol) -> None:
824 self.insert_child(child, None, pdf)
826 def inc_parent_counter_default(
827 self, parent: Union[IndirectObject, "TreeObject", None], n: int
828 ) -> None:
829 if is_null_or_none(parent):
830 return
831 assert parent is not None, "mypy"
832 parent = cast("TreeObject", parent.get_object())
833 if "/Count" in parent:
834 parent[NameObject("/Count")] = NumberObject(
835 max(0, cast(int, parent[NameObject("/Count")]) + n)
836 )
837 self.inc_parent_counter_default(parent.get("/Parent", None), n)
839 def inc_parent_counter_outline(
840 self, parent: Union[IndirectObject, "TreeObject", None], n: int
841 ) -> None:
842 if is_null_or_none(parent):
843 return
844 assert parent is not None, "mypy"
845 parent = cast("TreeObject", parent.get_object())
846 # BooleanObject requires comparison with == not is
847 opn = parent.get("/%is_open%", True) == True # noqa: E712
848 c = cast(int, parent.get("/Count", 0))
849 if c < 0:
850 c = abs(c)
851 parent[NameObject("/Count")] = NumberObject((c + n) * (1 if opn else -1))
852 if not opn:
853 return
854 self.inc_parent_counter_outline(parent.get("/Parent", None), n)
856 def insert_child(
857 self,
858 child: Any,
859 before: Any,
860 pdf: PdfWriterProtocol,
861 inc_parent_counter: Optional[Callable[..., Any]] = None,
862 ) -> IndirectObject:
863 if inc_parent_counter is None:
864 inc_parent_counter = self.inc_parent_counter_default
865 child_obj = child.get_object()
866 assert child.indirect_reference is not None, "mypy"
867 child_reference: IndirectObject = child.indirect_reference
869 prev: Optional[DictionaryObject]
870 if "/First" not in self: # no child yet
871 self[NameObject("/First")] = child_reference
872 self[NameObject("/Count")] = NumberObject(0)
873 self[NameObject("/Last")] = child_reference
874 child_obj[NameObject("/Parent")] = self.indirect_reference
875 inc_parent_counter(self, child_obj.get("/Count", 1))
876 if "/Next" in child_obj:
877 del child_obj["/Next"]
878 if "/Prev" in child_obj:
879 del child_obj["/Prev"]
880 return child_reference
881 prev = cast("DictionaryObject", self["/Last"])
883 visited: set[int] = set()
884 while prev.indirect_reference != before:
885 prev_id = id(prev)
886 if prev_id in visited:
887 raise LimitReachedError("Detected cycle in tree structure.")
888 visited.add(prev_id)
889 if "/Next" in prev:
890 prev = cast("TreeObject", prev["/Next"])
891 continue
893 # append at the end
894 prev[NameObject("/Next")] = cast("TreeObject", child_reference)
895 child_obj[NameObject("/Prev")] = prev.indirect_reference
896 child_obj[NameObject("/Parent")] = self.indirect_reference
897 if "/Next" in child_obj:
898 del child_obj["/Next"]
899 self[NameObject("/Last")] = child_reference
900 inc_parent_counter(self, child_obj.get("/Count", 1))
901 return child_reference
902 try: # insert as first or in the middle
903 assert isinstance(prev["/Prev"], DictionaryObject)
904 prev["/Prev"][NameObject("/Next")] = child_reference
905 child_obj[NameObject("/Prev")] = prev["/Prev"]
906 except Exception: # it means we are inserting in first position
907 child_obj.pop("/Next", None)
908 child_obj[NameObject("/Next")] = prev
909 prev[NameObject("/Prev")] = child_reference
910 child_obj[NameObject("/Parent")] = self.indirect_reference
911 inc_parent_counter(self, child_obj.get("/Count", 1))
912 return child_reference
914 def _remove_node_from_tree(
915 self, *, previous: Any, previous_reference: Any, current: Any, last: Any
916 ) -> None:
917 """Adjust the pointers of the linked list and tree node count."""
918 next_reference = current.get("/Next", None)
919 if previous is None:
920 if next_reference:
921 # Removing first tree node
922 next_object = next_reference.get_object()
923 del next_object["/Prev"]
924 self[NameObject("/First")] = next_reference
925 self[NameObject("/Count")] = NumberObject(
926 self[NameObject("/Count")] - 1 # type: ignore[operator]
927 )
929 else:
930 # Removing only tree node
931 self[NameObject("/Count")] = NumberObject(0)
932 del self["/First"]
933 if NameObject("/Last") in self:
934 del self[NameObject("/Last")]
935 else:
936 if next_reference:
937 # Removing middle tree node
938 next_object = next_reference.get_object()
939 next_object[NameObject("/Prev")] = previous_reference
940 previous[NameObject("/Next")] = next_reference
941 else:
942 # Removing last tree node
943 assert current == last
944 del previous["/Next"]
945 self[NameObject("/Last")] = previous_reference
946 self[NameObject("/Count")] = NumberObject(self["/Count"] - 1) # type: ignore[operator]
948 def remove_child(self, child: Any) -> None:
949 child_object = child.get_object()
951 if "/Parent" not in child_object:
952 raise ValueError("Removed child does not appear to be a tree item")
953 if child_object["/Parent"] != self:
954 raise ValueError("Removed child is not a member of this tree")
956 found = False
957 previous_reference = None
958 previous = None
959 current_reference: Optional[Any] = self["/First"]
960 current: Optional[dict[str, Any]] = current_reference.get_object() # type: ignore[union-attr]
961 last_reference = self["/Last"]
962 last = last_reference.get_object()
963 visited: set[int] = set()
964 while current is not None:
965 if current == child_object:
966 TreeObject._remove_node_from_tree(
967 self, previous=previous, previous_reference=previous_reference, current=current, last=last
968 )
969 found = True
970 break
971 current_id = id(current)
972 if current_id in visited:
973 raise LimitReachedError("Detected cyclic /Next hierarchy.")
974 visited.add(current_id)
976 # Go to the next node.
977 previous_reference = current_reference
978 previous = current
979 if "/Next" not in current:
980 break
981 current_reference = current["/Next"]
982 current = current_reference.get_object()
984 if not found:
985 raise ValueError("Removal couldn't find item in tree")
987 _reset_node_tree_relationship(child_object)
989 def remove_from_tree(self) -> None:
990 """Remove the object from the tree it is in."""
991 if "/Parent" not in self:
992 raise ValueError("Removed child does not appear to be a tree item")
993 cast("TreeObject", self["/Parent"]).remove_child(self)
995 def empty_tree(self) -> None:
996 for child in self:
997 child_obj = child.get_object()
998 _reset_node_tree_relationship(child_obj)
1000 if NameObject("/Count") in self:
1001 del self[NameObject("/Count")]
1002 if NameObject("/First") in self:
1003 del self[NameObject("/First")]
1004 if NameObject("/Last") in self:
1005 del self[NameObject("/Last")]
1008def _reset_node_tree_relationship(child_obj: Any) -> None:
1009 """
1010 Call this after a node has been removed from a tree.
1012 This resets the nodes attributes in respect to that tree.
1014 Args:
1015 child_obj:
1017 """
1018 del child_obj[NameObject("/Parent")]
1019 if NameObject("/Next") in child_obj:
1020 del child_obj[NameObject("/Next")]
1021 if NameObject("/Prev") in child_obj:
1022 del child_obj[NameObject("/Prev")]
1025class _FontFileCharacterMap(NamedTuple):
1026 """The character map pypdf._cmap derived from an embedded font program. See #4156."""
1028 digest: bytes
1029 """SHA-256 digest of the decoded font data the map was derived from."""
1030 map_dict: dict[Any, Any]
1031 int_entry: list[int]
1034class StreamObject(DictionaryObject):
1035 _font_file_character_map: Optional[_FontFileCharacterMap] = None
1036 """
1037 The character map pypdf._cmap derived from this stream as an embedded font program. Parsing a CFF font
1038 program with fontTools is expensive, and the same program is reached from every font resource and page
1039 referencing it. See #4156.
1040 """
1042 def __init__(self) -> None:
1043 self._data: bytes = b""
1044 self.decoded_self: Optional[DecodedStreamObject] = None
1046 def replicate(
1047 self,
1048 pdf_dest: PdfWriterProtocol,
1049 ) -> "StreamObject":
1050 d__ = cast(
1051 "StreamObject",
1052 self._reference_clone(self.__class__(), pdf_dest, False),
1053 )
1054 d__._data = self._data
1055 try:
1056 decoded_self = self.decoded_self
1057 if decoded_self is None:
1058 self.decoded_self = None
1059 else:
1060 self.decoded_self = cast(
1061 "DecodedStreamObject", decoded_self.replicate(pdf_dest)
1062 )
1063 except Exception:
1064 pass
1065 for k, v in self.items():
1066 d__[k.replicate(pdf_dest)] = (
1067 v.replicate(pdf_dest) if hasattr(v, "replicate") else v
1068 )
1069 return d__
1071 def _clone(
1072 self,
1073 src: DictionaryObject,
1074 pdf_dest: PdfWriterProtocol,
1075 force_duplicate: bool,
1076 ignore_fields: Optional[Sequence[Union[str, int]]],
1077 visited: set[tuple[int, int]],
1078 ) -> None:
1079 """
1080 Update the object from src.
1082 Args:
1083 src:
1084 pdf_dest:
1085 force_duplicate:
1086 ignore_fields:
1088 """
1089 self._data = cast("StreamObject", src)._data
1090 try:
1091 decoded_self = cast("StreamObject", src).decoded_self
1092 if decoded_self is None:
1093 self.decoded_self = None
1094 else:
1095 self.decoded_self = cast(
1096 "DecodedStreamObject",
1097 decoded_self.clone(pdf_dest, force_duplicate, ignore_fields),
1098 )
1099 except Exception:
1100 pass
1101 super()._clone(src, pdf_dest, force_duplicate, ignore_fields, visited)
1103 def hash_bin(self) -> int:
1104 """
1105 Used to detect modified object.
1107 Returns:
1108 Hash considering type and value.
1110 """
1111 # Use _data to prevent errors on non-decoded streams.
1112 return hash((super().hash_bin(), self._data))
1114 def get_data(self) -> bytes:
1115 return self._data
1117 def set_data(self, data: bytes) -> None:
1118 self._data = data
1120 def hash_value_data(self) -> bytes:
1121 data = super().hash_value_data()
1122 data += self.get_data()
1123 return data
1125 def write_to_stream(
1126 self, stream: StreamType, encryption_key: Union[str, bytes, None] = None
1127 ) -> None:
1128 if encryption_key is not None: # deprecated
1129 deprecation_no_replacement(
1130 "the encryption_key parameter of write_to_stream", "5.0.0"
1131 )
1132 self[NameObject(StreamAttributes.LENGTH)] = NumberObject(len(self._data))
1133 DictionaryObject.write_to_stream(self, stream)
1134 del self[StreamAttributes.LENGTH]
1135 stream.write(b"\nstream\n")
1136 stream.write(self._data)
1137 stream.write(b"\nendstream")
1139 @staticmethod
1140 def initialize_from_dictionary(
1141 data: dict[str, Any]
1142 ) -> Union["EncodedStreamObject", "DecodedStreamObject"]:
1143 retval: Union[EncodedStreamObject, DecodedStreamObject]
1144 if StreamAttributes.FILTER in data:
1145 retval = EncodedStreamObject()
1146 else:
1147 retval = DecodedStreamObject()
1148 retval._data = data["__streamdata__"]
1149 del data["__streamdata__"]
1150 if StreamAttributes.LENGTH in data:
1151 del data[StreamAttributes.LENGTH]
1152 retval.update(data)
1153 return retval
1155 def flate_encode(self, level: int = -1) -> "EncodedStreamObject":
1156 from ..filters import FlateDecode # noqa: PLC0415
1158 if StreamAttributes.FILTER in self:
1159 f = self[StreamAttributes.FILTER]
1160 if isinstance(f, ArrayObject):
1161 f = ArrayObject([NameObject(FT.FLATE_DECODE), *f])
1162 try:
1163 params = ArrayObject(
1164 [NullObject(), *self.get(StreamAttributes.DECODE_PARMS, ArrayObject())]
1165 )
1166 except TypeError:
1167 # case of error where the * operator is not working (not an array
1168 params = ArrayObject(
1169 [NullObject(), self.get(StreamAttributes.DECODE_PARMS, ArrayObject())]
1170 )
1171 else:
1172 f = ArrayObject([NameObject(FT.FLATE_DECODE), f])
1173 params = ArrayObject(
1174 [NullObject(), self.get(StreamAttributes.DECODE_PARMS, NullObject())]
1175 )
1176 else:
1177 f = NameObject(FT.FLATE_DECODE)
1178 params = None
1179 retval = EncodedStreamObject()
1180 retval.update(self)
1181 retval[NameObject(StreamAttributes.FILTER)] = f
1182 if params is not None:
1183 retval[NameObject(StreamAttributes.DECODE_PARMS)] = params
1184 retval._data = FlateDecode.encode(self._data, level)
1185 return retval
1187 def decode_as_image(self, pillow_parameters: Union[dict[str, Any], None] = None) -> Any:
1188 """
1189 Try to decode the stream object as an image
1191 Args:
1192 pillow_parameters: parameters provided to Pillow Image.save() method,
1193 cf. <https://pillow.readthedocs.io/en/stable/reference/Image.html#PIL.Image.Image.save>
1195 Returns:
1196 a PIL image if proper decoding has been found
1197 Raises:
1198 Exception: Errors during decoding will be reported.
1199 It is recommended to catch exceptions to prevent
1200 stops in your program.
1202 """
1203 from ._image_xobject import _xobj_to_image # noqa: PLC0415
1205 if self.get("/Subtype", "") != "/Image":
1206 try:
1207 logger_warning( # pragma: no cover
1208 "%(indirect_reference)s does not seem to be an Image",
1209 source=__name__,
1210 indirect_reference=self.indirect_reference,
1211 )
1212 except AttributeError:
1213 logger_warning( # pragma: no cover
1214 "%(obj)r object does not seem to be an Image",
1215 source=__name__,
1216 obj=self,
1217 )
1218 extension, _, img = _xobj_to_image(self, pillow_parameters)
1219 if extension is None:
1220 return None # pragma: no cover
1221 return img
1224class DecodedStreamObject(StreamObject):
1225 pass
1228class EncodedStreamObject(StreamObject):
1229 def __init__(self) -> None:
1230 self.decoded_self: Optional[DecodedStreamObject] = None
1232 # This overrides the parent method
1233 def get_data(self) -> bytes:
1234 from ..filters import decode_stream_data # noqa: PLC0415
1236 if self.decoded_self is not None:
1237 # Cached version of decoded object
1238 return self.decoded_self.get_data()
1240 # Create decoded object
1241 decoded = DecodedStreamObject()
1242 decoded.set_data(decode_stream_data(self))
1243 for key, value in self.items():
1244 if key not in (StreamAttributes.LENGTH, StreamAttributes.FILTER, StreamAttributes.DECODE_PARMS):
1245 decoded[key] = value
1246 self.decoded_self = decoded
1247 return decoded.get_data()
1249 # This overrides the parent method:
1250 def set_data(self, data: bytes) -> None:
1251 from ..filters import FlateDecode # noqa: PLC0415
1253 if self.get(StreamAttributes.FILTER, "") in (FT.FLATE_DECODE, [FT.FLATE_DECODE]):
1254 if not isinstance(data, bytes):
1255 raise TypeError("Data must be bytes")
1256 if self.decoded_self is None:
1257 self.get_data() # to create self.decoded_self
1258 assert self.decoded_self is not None, "mypy"
1259 self.decoded_self.set_data(data)
1260 super().set_data(FlateDecode.encode(data))
1261 else:
1262 raise PdfReadError(
1263 "Streams encoded with a filter different from FlateDecode are not supported"
1264 )
1267CONTENT_STREAM_ARRAY_MAX_LENGTH = 10_000
1270class ContentStream(DecodedStreamObject):
1271 """
1272 In order to be fast, this data structure can contain either:
1274 * raw data in ._data
1275 * parsed stream operations in ._operations.
1277 At any time, ContentStream object can either have both of those fields defined,
1278 or one field defined and the other set to None.
1280 These fields are "rebuilt" lazily, when accessed:
1282 * when .get_data() is called, if ._data is None, it is rebuilt from ._operations.
1283 * when .operations is called, if ._operations is None, it is rebuilt from ._data.
1285 Conversely, these fields can be invalidated:
1287 * when .set_data() is called, ._operations is set to None.
1288 * when .operations is set, ._data is set to None.
1289 """
1290 _OPERATOR_LENGTH_LIMIT = 128
1292 def __init__(
1293 self,
1294 stream: Any,
1295 pdf: Any,
1296 forced_encoding: Union[str, list[str], dict[int, str], None] = None,
1297 ) -> None:
1298 self.pdf = pdf
1299 self._operations: list[tuple[Any, bytes]] = []
1301 # stream may be a StreamObject or an ArrayObject containing
1302 # StreamObjects to be concatenated together.
1303 if stream is None:
1304 super().set_data(b"")
1305 else:
1306 stream = stream.get_object()
1307 if isinstance(stream, ArrayObject):
1308 from pypdf._configuration import get_configuration # noqa: PLC0415
1309 configuration = get_configuration()
1311 if (stream_length := len(stream)) > CONTENT_STREAM_ARRAY_MAX_LENGTH:
1312 raise LimitReachedError(
1313 f"Array-based stream has {stream_length} > {CONTENT_STREAM_ARRAY_MAX_LENGTH} elements."
1314 )
1315 data = bytearray()
1316 length = 0
1317 for s in stream:
1318 s_resolved = s.get_object()
1319 if isinstance(s_resolved, NullObject):
1320 continue
1321 if not isinstance(s_resolved, StreamObject):
1322 # No need to emit an exception here for now - the PDF structure
1323 # seems to already be broken beforehand in these cases.
1324 logger_warning(
1325 "Expected StreamObject, got %(type_name)s instead. Data might be wrong.",
1326 source=__name__,
1327 type_name=type(s_resolved).__name__,
1328 )
1329 else:
1330 new_data = s_resolved.get_data()
1331 length += len(new_data)
1332 if length > configuration.array_based_stream_maximum_output_length:
1333 raise LimitReachedError(
1334 f"Array-based stream has at least {length} > "
1335 f"{configuration.array_based_stream_maximum_output_length} output bytes."
1336 )
1337 data += new_data
1338 if len(data) == 0 or data[-1:] != b"\n":
1339 # There should be no direct need to check for a change of one byte.
1340 length += 1
1341 data += b"\n"
1342 super().set_data(bytes(data))
1343 else:
1344 stream_data = stream.get_data()
1345 assert stream_data is not None
1346 super().set_data(stream_data)
1347 self.forced_encoding = forced_encoding
1349 def replicate(
1350 self,
1351 pdf_dest: PdfWriterProtocol,
1352 ) -> "ContentStream":
1353 d__ = cast(
1354 "ContentStream",
1355 self._reference_clone(self.__class__(None, None), pdf_dest, False),
1356 )
1357 d__._data = self._data
1358 try:
1359 decoded_self = self.decoded_self
1360 if decoded_self is None:
1361 self.decoded_self = None
1362 else:
1363 self.decoded_self = cast(
1364 "DecodedStreamObject", decoded_self.replicate(pdf_dest)
1365 )
1366 except Exception:
1367 pass
1368 for k, v in self.items():
1369 d__[k.replicate(pdf_dest)] = (
1370 v.replicate(pdf_dest) if hasattr(v, "replicate") else v
1371 )
1372 return d__
1373 d__.set_data(self._data)
1374 d__.pdf = pdf_dest
1375 d__._operations = list(self._operations)
1376 d__.forced_encoding = self.forced_encoding
1377 return d__
1379 def clone(
1380 self,
1381 pdf_dest: Any,
1382 force_duplicate: bool = False,
1383 ignore_fields: Optional[Sequence[Union[str, int]]] = (),
1384 ) -> "ContentStream":
1385 """
1386 Clone object into pdf_dest.
1388 Args:
1389 pdf_dest:
1390 force_duplicate:
1391 ignore_fields:
1393 Returns:
1394 The cloned ContentStream
1396 """
1397 try:
1398 if self.indirect_reference.pdf == pdf_dest and not force_duplicate: # type: ignore[union-attr]
1399 return self
1400 except Exception:
1401 pass
1403 visited: set[tuple[int, int]] = set()
1404 d__ = cast(
1405 "ContentStream",
1406 self._reference_clone(
1407 self.__class__(None, None), pdf_dest, force_duplicate
1408 ),
1409 )
1410 if ignore_fields is None:
1411 ignore_fields = []
1412 d__._clone(self, pdf_dest, force_duplicate, ignore_fields, visited)
1413 return d__
1415 def _clone(
1416 self,
1417 src: DictionaryObject,
1418 pdf_dest: PdfWriterProtocol,
1419 force_duplicate: bool,
1420 ignore_fields: Optional[Sequence[Union[str, int]]],
1421 visited: set[tuple[int, int]],
1422 ) -> None:
1423 """
1424 Update the object from src.
1426 Args:
1427 src:
1428 pdf_dest:
1429 force_duplicate:
1430 ignore_fields:
1432 """
1433 src_cs = cast("ContentStream", src)
1434 super().set_data(src_cs._data)
1435 self.pdf = pdf_dest
1436 self._operations = list(src_cs._operations)
1437 self.forced_encoding = src_cs.forced_encoding
1438 # no need to call DictionaryObjection or anything
1439 # like super(DictionaryObject,self)._clone(src, pdf_dest, force_duplicate, ignore_fields, visited)
1441 @staticmethod
1442 def _check_limits(
1443 *,
1444 remaining_elements: Optional[int] = None,
1445 parsing_budget: StreamParsingBudget,
1446 token_decrement: int
1447 ) -> Optional[int]:
1448 if remaining_elements is not None:
1449 remaining_elements -= 1
1450 if remaining_elements < 0:
1451 raise LimitReachedError("Reached maximum number of tokens.")
1452 parsing_budget.remaining_budget -= token_decrement
1453 if parsing_budget.remaining_budget < 0:
1454 raise LimitReachedError("Reached maximum number of tokens per stream.")
1455 return remaining_elements
1457 def _parse_content_stream(self, stream: StreamType) -> None:
1458 # 7.8.2 Content Streams
1459 stream.seek(0, 0)
1460 operands: list[Union[int, str, PdfObject]] = []
1461 parsing_budget = StreamParsingBudget(remaining_budget=get_configuration().stream_parsing_work_maximum_count)
1462 while True:
1463 peek = read_non_whitespace(stream)
1464 if peek in (b"", 0):
1465 break
1466 stream.seek(-1, 1)
1467 if peek.isalpha() or peek in (b"'", b'"'):
1468 operator = read_until_regex(
1469 stream=stream, regex=NameObject.delimiter_pattern, length=self._OPERATOR_LENGTH_LIMIT
1470 )
1471 if operator == b"BI":
1472 # begin inline image - a completely different parsing
1473 # mechanism is required, of course... thanks buddy...
1474 assert operands == []
1475 self._check_limits(parsing_budget=parsing_budget, token_decrement=2)
1476 ii = self._read_inline_image(stream, parsing_budget=parsing_budget)
1477 self._operations.append((ii, b"INLINE IMAGE"))
1478 else:
1479 self._check_limits(parsing_budget=parsing_budget, token_decrement=1)
1480 self._operations.append((operands, operator))
1481 operands = []
1482 elif peek == b"%":
1483 # If we encounter a comment in the content stream, we have to
1484 # handle it here. Typically, read_object will handle
1485 # encountering a comment -- but read_object assumes that
1486 # following the comment must be the object we're trying to
1487 # read. In this case, it could be an operator instead.
1488 while peek not in (b"\r", b"\n", b""):
1489 peek = stream.read(1)
1490 else:
1491 self._check_limits(parsing_budget=parsing_budget, token_decrement=1)
1492 operands.append(read_object(stream, None, self.forced_encoding, parsing_budget=parsing_budget))
1494 def _read_inline_image(self, stream: StreamType, parsing_budget: StreamParsingBudget) -> dict[str, Any]:
1495 # begin reading just after the "BI" - begin image
1496 # first read the dictionary of settings.
1497 settings = DictionaryObject()
1498 remaining_count: Optional[int] = get_configuration().dictionary_maximum_element_count
1499 while True:
1500 tok = read_non_whitespace(stream)
1501 if not tok:
1502 raise PdfReadError("Unexpected end of stream.")
1503 stream.seek(-1, 1)
1504 if tok == b"I":
1505 # "ID" - begin of image data
1506 break
1507 remaining_count = self._check_limits(
1508 remaining_elements=remaining_count, parsing_budget=parsing_budget, token_decrement=2
1509 )
1511 key = read_object(stream, self.pdf, parsing_budget=parsing_budget)
1512 _ = read_non_whitespace(stream)
1513 stream.seek(-1, 1)
1514 value = read_object(stream, self.pdf, parsing_budget=parsing_budget)
1515 settings[key] = value
1516 # left at beginning of ID
1517 tmp = stream.read(3)
1518 assert tmp[:2] == b"ID"
1519 filtr = settings.get("/F", settings.get("/Filter", "not set"))
1520 savpos = stream.tell()
1521 if isinstance(filtr, list):
1522 filtr = filtr[0] if filtr else "not set" # used forencoding
1523 if not isinstance(filtr, str):
1524 # A valid filter is a name or an array of names; anything else
1525 # cannot match the abbreviations below, so treat it as unfiltered.
1526 filtr = "not set"
1527 if "AHx" in filtr or "ASCIIHexDecode" in filtr:
1528 data = extract_inline__ascii_hex_decode(stream)
1529 elif "A85" in filtr or "ASCII85Decode" in filtr:
1530 data = extract_inline__ascii85_decode(stream)
1531 elif "RL" in filtr or "RunLengthDecode" in filtr:
1532 data = extract_inline__run_length_decode(stream)
1533 elif "DCT" in filtr or "DCTDecode" in filtr:
1534 data = extract_inline__dct_decode(stream)
1535 elif filtr == "not set":
1536 cs = settings.get("/CS", "")
1537 if isinstance(cs, list):
1538 cs = cs[0] if cs else ""
1539 if not isinstance(cs, str):
1540 cs = ""
1541 if "RGB" in cs:
1542 lcs: float = 3
1543 elif "CMYK" in cs:
1544 lcs = 4
1545 else:
1546 bits = settings.get(
1547 "/BPC",
1548 8 if cs in {"/I", "/G", "/Indexed", "/DeviceGray"} else -1,
1549 )
1550 if isinstance(bits, (int, float)) and bits > 0:
1551 lcs = bits / 8.0
1552 else:
1553 data = extract_inline_default(stream)
1554 lcs = -1
1555 width = settings.get("/W")
1556 height = settings.get("/H")
1557 if lcs > 0 and isinstance(width, int) and isinstance(height, int):
1558 data = stream.read(ceil(width * lcs) * height)
1559 elif lcs > 0:
1560 # Without usable dimensions the raw sample length is unknown;
1561 # fall back to scanning for the `EI` marker.
1562 data = extract_inline_default(stream)
1563 # Move to the `EI` if possible.
1564 ei = read_non_whitespace(stream)
1565 stream.seek(-1, 1)
1566 else:
1567 data = extract_inline_default(stream)
1569 ei = stream.read(3)
1570 # An `EI` at the very end of the stream yields only two bytes; rewinding
1571 # unconditionally would step back into the marker (#3468).
1572 if len(ei) == 3:
1573 stream.seek(-1, 1)
1574 ei_trailing = ei[2:3]
1575 if ei[:2] != b"EI" or (ei_trailing != b"" and ei_trailing not in WHITESPACES):
1576 # Deal with wrong/missing `EI` tags. Example: Wrong dimensions specified above.
1577 stream.seek(savpos, 0)
1578 data = extract_inline_default(stream)
1579 ei = stream.read(3)
1580 if len(ei) == 3:
1581 stream.seek(-1, 1)
1582 ei_trailing = ei[2:3]
1583 if ei[:2] != b"EI" or (
1584 ei_trailing != b"" and ei_trailing not in WHITESPACES
1585 ): # pragma: no cover
1586 # Check the same condition again. This should never fail as
1587 # edge cases are covered by `extract_inline_default` above,
1588 # but check this to make sure that we are behind the `EI` afterwards.
1589 raise PdfStreamError(
1590 f"Could not extract inline image, even using fallback. Expected 'EI', got {ei!r}"
1591 )
1592 return {"settings": settings, "data": data}
1594 # This overrides the parent method
1595 def get_data(self) -> bytes:
1596 if not self._data:
1597 new_data = BytesIO()
1598 for operands, operator in self._operations:
1599 if operator == b"INLINE IMAGE":
1600 new_data.write(b"BI")
1601 dict_text = BytesIO()
1602 operands["settings"].write_to_stream(dict_text)
1603 new_data.write(dict_text.getvalue()[2:-2])
1604 new_data.write(b"ID ")
1605 new_data.write(operands["data"])
1606 new_data.write(b"EI")
1607 else:
1608 for op in operands:
1609 op.write_to_stream(new_data)
1610 new_data.write(b" ")
1611 new_data.write(operator)
1612 new_data.write(b"\n")
1613 self._data = new_data.getvalue()
1614 return self._data
1616 # This overrides the parent method
1617 def set_data(self, data: bytes) -> None:
1618 super().set_data(data)
1619 self._operations = []
1621 @property
1622 def operations(self) -> list[tuple[Any, bytes]]:
1623 if not self._operations and self._data:
1624 self._parse_content_stream(BytesIO(self._data))
1625 self._data = b""
1626 return self._operations
1628 @operations.setter
1629 def operations(self, operations: list[tuple[Any, bytes]]) -> None:
1630 self._operations = operations
1631 self._data = b""
1633 def isolate_graphics_state(self) -> None:
1634 if self._operations:
1635 self._operations.insert(0, ([], b"q"))
1636 self._operations.append(([], b"Q"))
1637 elif self._data:
1638 self._data = b"q\n" + self._data + b"\nQ\n"
1640 # This overrides the parent method
1641 def write_to_stream(
1642 self, stream: StreamType, encryption_key: Union[str, bytes, None] = None
1643 ) -> None:
1644 if not self._data and self._operations:
1645 self.get_data() # this ensures ._data is rebuilt
1646 super().write_to_stream(stream, encryption_key)
1649# Bound comment scanning without consuming Python recursion depth.
1650_MAX_LEADING_COMMENTS = 1000
1653def read_object( # noqa: PLR0911
1654 stream: StreamType,
1655 pdf: Optional[PdfReaderProtocol],
1656 forced_encoding: Union[str, list[str], dict[int, str], None] = None,
1657 *,
1658 parsing_budget: Optional[StreamParsingBudget] = None,
1659) -> PdfObject:
1660 """
1661 Read the next object from the stream.
1663 Args:
1664 stream: The stream to read from.
1665 pdf: The corresponding PDF file.
1666 forced_encoding: The encoding to enforce, if any.
1667 parsing_budget: The allowed parsing budget, optional. If set, only applies to dictionaries,
1668 arrays and content streams.
1670 Returns:
1671 The parsed object.
1672 """
1673 tok = stream.read(1)
1674 stream.seek(-1, 1) # reset to start
1675 if tok == b"%":
1676 # Skip consecutive comments in a loop instead of recursing per comment.
1677 skip_over_comments(stream, limit=_MAX_LEADING_COMMENTS)
1678 tok = stream.read(1)
1679 stream.seek(-1, 1) # reset to start
1680 if tok == b"/":
1681 return NameObject.read_from_stream(stream, pdf)
1682 if tok == b"<":
1683 # hexadecimal string OR dictionary
1684 peek = stream.read(2)
1685 stream.seek(-2, 1) # reset to start
1686 if peek == b"<<":
1687 return DictionaryObject.read_from_stream(stream, pdf, forced_encoding, parsing_budget=parsing_budget)
1688 return read_hex_string_from_stream(stream, forced_encoding)
1689 if tok == b"[":
1690 return ArrayObject.read_from_stream(stream, pdf, forced_encoding, parsing_budget=parsing_budget)
1691 if tok in (b"t", b"f"):
1692 return BooleanObject.read_from_stream(stream)
1693 if tok == b"(":
1694 return read_string_from_stream(stream, forced_encoding)
1695 if tok == b"e" and stream.read(6) == b"endobj":
1696 return NullObject()
1697 if tok == b"n":
1698 return NullObject.read_from_stream(stream)
1699 if tok in b"0123456789+-.":
1700 # number object OR indirect reference
1701 peek = stream.read(20)
1702 match = IndirectPattern.match(peek)
1703 if match is not None:
1704 assert pdf is not None, "mypy"
1705 # Continue after the "R", the match includes the character following it
1706 stream.seek(match.end() - 1 - len(peek), 1)
1707 # The object number includes an optional sign
1708 return IndirectObject(int(peek[: match.end(1)]), int(match[2]), pdf)
1709 number_end = NumberObject.NumberPattern.search(peek)
1710 if number_end is not None:
1711 # The whole number is in the peek, there is no need to read it again
1712 num = peek[: number_end.start()]
1713 stream.seek(len(num) - len(peek), 1)
1714 if b"." in num:
1715 return FloatObject(num)
1716 return NumberObject(num)
1717 stream.seek(-len(peek), 1) # reset to start
1718 return NumberObject.read_from_stream(stream)
1719 pos = stream.tell()
1720 stream.seek(-20, 1)
1721 stream_extract = stream.read(80)
1722 stream.seek(pos)
1723 read_until_whitespace(stream)
1724 raise PdfReadError(
1725 f"Invalid Elementary Object starting with {tok!r} @{pos}: {stream_extract!r}"
1726 )
1729class Field(TreeObject):
1730 """
1731 A class representing a field dictionary.
1733 This class is accessed through
1734 :meth:`get_fields()<pypdf.PdfReader.get_fields>`
1735 """
1737 def __init__(self, data: DictionaryObject) -> None:
1738 DictionaryObject.__init__(self)
1739 field_attributes = (
1740 FieldDictionaryAttributes.attributes()
1741 + CheckboxRadioButtonAttributes.attributes()
1742 )
1743 self.indirect_reference = data.indirect_reference
1744 for attr in field_attributes:
1745 try:
1746 self[NameObject(attr)] = data[attr]
1747 except KeyError:
1748 pass
1749 if isinstance(self.get("/V"), EncodedStreamObject):
1750 d = cast(EncodedStreamObject, self[NameObject("/V")]).get_data()
1751 if isinstance(d, bytes):
1752 d_str = d.decode()
1753 elif d is None:
1754 d_str = ""
1755 else:
1756 raise Exception("Should never happen")
1757 self[NameObject("/V")] = TextStringObject(d_str)
1759 # TABLE 8.69 Entries common to all field dictionaries
1760 @property
1761 def field_type(self) -> Optional[NameObject]:
1762 """Read-only property accessing the type of this field."""
1763 return self.get(FieldDictionaryAttributes.FT)
1765 @property
1766 def parent(self) -> Optional[DictionaryObject]:
1767 """Read-only property accessing the parent of this field."""
1768 return self.get(FieldDictionaryAttributes.Parent)
1770 @property
1771 def kids(self) -> Optional["ArrayObject"]:
1772 """Read-only property accessing the kids of this field."""
1773 return self.get(FieldDictionaryAttributes.Kids)
1775 @property
1776 def name(self) -> Optional[str]:
1777 """Read-only property accessing the name of this field."""
1778 return self.get(FieldDictionaryAttributes.T)
1780 @property
1781 def alternate_name(self) -> Optional[str]:
1782 """Read-only property accessing the alternate name of this field."""
1783 return self.get(FieldDictionaryAttributes.TU)
1785 @property
1786 def mapping_name(self) -> Optional[str]:
1787 """
1788 Read-only property accessing the mapping name of this field.
1790 This name is used by pypdf as a key in the dictionary returned by
1791 :meth:`get_fields()<pypdf.PdfReader.get_fields>`
1792 """
1793 return self.get(FieldDictionaryAttributes.TM)
1795 @property
1796 def flags(self) -> Optional[int]:
1797 """
1798 Read-only property accessing the field flags, specifying various
1799 characteristics of the field (see Table 8.70 of the PDF 1.7 reference).
1800 """
1801 return self.get(FieldDictionaryAttributes.Ff)
1803 @property
1804 def value(self) -> Optional[Any]:
1805 """
1806 Read-only property accessing the value of this field.
1808 Format varies based on field type.
1809 """
1810 return self.get(FieldDictionaryAttributes.V)
1812 @property
1813 def default_value(self) -> Optional[Any]:
1814 """Read-only property accessing the default value of this field."""
1815 return self.get(FieldDictionaryAttributes.DV)
1817 @property
1818 def additional_actions(self) -> Optional[DictionaryObject]:
1819 """
1820 Read-only property accessing the additional actions dictionary.
1822 This dictionary defines the field's behavior in response to trigger
1823 events. See Section 8.5.2 of the PDF 1.7 reference.
1824 """
1825 return self.get(FieldDictionaryAttributes.AA)
1828class Destination(TreeObject):
1829 """
1830 A class representing a destination within a PDF file.
1832 See section 12.3.2 of the PDF 2.0 reference.
1834 Args:
1835 title: Title of this destination.
1836 page: Reference to the page of this destination. Should
1837 be an instance of :class:`IndirectObject<pypdf.generic.IndirectObject>`.
1838 fit: How the destination is displayed.
1840 Raises:
1841 PdfReadError: If destination type is invalid.
1843 """
1845 node: Optional[
1846 DictionaryObject
1847 ] = None # node provide access to the original Object
1849 def remove_from_tree(self) -> None:
1850 """
1851 Remove the outline item this destination was built from.
1853 `reader.outline` and `writer.outline` yield detached copies rather than
1854 the nodes themselves, so the copy never has a `/Parent` and removing it
1855 would be a no-op. `node` is the dictionary in the document.
1856 """
1857 if self.node is None:
1858 super().remove_from_tree()
1859 elif "/Parent" not in self.node:
1860 raise ValueError("Removed child does not appear to be a tree item")
1861 else:
1862 TreeObject.remove_child(
1863 cast("TreeObject", self.node["/Parent"]), self.node
1864 )
1866 def __init__(
1867 self,
1868 title: Union[str, bytes],
1869 page: Union[NumberObject, IndirectObject, NullObject, DictionaryObject],
1870 fit: Fit,
1871 ) -> None:
1872 self._filtered_children: list[Any] = [] # used in PdfWriter
1874 typ = fit.fit_type
1875 args = fit.fit_args
1877 DictionaryObject.__init__(self)
1878 self[NameObject("/Title")] = TextStringObject(title)
1879 self[NameObject("/Page")] = page
1880 self[NameObject("/Type")] = typ
1882 # from table 8.2 of the PDF 1.7 reference.
1883 if typ == "/XYZ":
1884 if len(args) < 1: # left is missing : should never occur
1885 args.append(NumberObject(0.0))
1886 if len(args) < 2: # top is missing
1887 args.append(NumberObject(0.0))
1888 if len(args) < 3: # zoom is missing
1889 args.append(NumberObject(0.0))
1890 # surplus arguments are ignored rather than failing the unpacking
1891 (
1892 self[NameObject(TA.LEFT)],
1893 self[NameObject(TA.TOP)],
1894 self[NameObject("/Zoom")],
1895 ) = args[:3]
1896 elif len(args) == 0:
1897 pass
1898 elif typ == TF.FIT_R:
1899 if len(args) == 4: # a wrong number of arguments degrades to null
1900 (
1901 self[NameObject(TA.LEFT)],
1902 self[NameObject(TA.BOTTOM)],
1903 self[NameObject(TA.RIGHT)],
1904 self[NameObject(TA.TOP)],
1905 ) = args
1906 else:
1907 (
1908 self[NameObject(TA.LEFT)],
1909 self[NameObject(TA.BOTTOM)],
1910 self[NameObject(TA.RIGHT)],
1911 self[NameObject(TA.TOP)],
1912 ) = (NullObject(), NullObject(), NullObject(), NullObject())
1913 elif typ in [TF.FIT_H, TF.FIT_BH]:
1914 try: # Prefer to be more robust not only to null parameters
1915 (self[NameObject(TA.TOP)],) = args
1916 except Exception:
1917 (self[NameObject(TA.TOP)],) = (NullObject(),)
1918 elif typ in [TF.FIT_V, TF.FIT_BV]:
1919 try: # Prefer to be more robust not only to null parameters
1920 (self[NameObject(TA.LEFT)],) = args
1921 except Exception:
1922 (self[NameObject(TA.LEFT)],) = (NullObject(),)
1923 elif typ in [TF.FIT, TF.FIT_B]:
1924 pass
1925 else:
1926 raise PdfReadError(f"Unknown Destination Type: {typ!r}")
1928 @property
1929 def dest_array(self) -> "ArrayObject":
1930 return ArrayObject(
1931 [self.raw_get("/Page"), self["/Type"]]
1932 + [
1933 self[x]
1934 for x in ["/Left", "/Bottom", "/Right", "/Top", "/Zoom"]
1935 if x in self
1936 ]
1937 )
1939 def write_to_stream(
1940 self, stream: StreamType, encryption_key: Union[str, bytes, None] = None
1941 ) -> None:
1942 if encryption_key is not None: # deprecated
1943 deprecation_no_replacement(
1944 "the encryption_key parameter of write_to_stream", "5.0.0"
1945 )
1946 stream.write(b"<<\n")
1947 key = NameObject("/D")
1948 key.write_to_stream(stream)
1949 stream.write(b" ")
1950 value = self.dest_array
1951 value.write_to_stream(stream)
1953 key = NameObject("/S")
1954 key.write_to_stream(stream)
1955 stream.write(b" ")
1956 value_s = NameObject("/GoTo")
1957 value_s.write_to_stream(stream)
1959 stream.write(b"\n")
1960 stream.write(b">>")
1962 @property
1963 def title(self) -> Optional[str]:
1964 """Read-only property accessing the destination title."""
1965 return self.get("/Title")
1967 @property
1968 def page(self) -> Optional[IndirectObject]:
1969 """Read-only property accessing the IndirectObject of the destination page."""
1970 return self.get("/Page")
1972 @property
1973 def typ(self) -> Optional[str]:
1974 """Read-only property accessing the destination type."""
1975 return self.get("/Type")
1977 @property
1978 def zoom(self) -> Optional[int]:
1979 """Read-only property accessing the zoom factor."""
1980 return self.get("/Zoom", None)
1982 @property
1983 def left(self) -> Optional[FloatObject]:
1984 """Read-only property accessing the left horizontal coordinate."""
1985 return self.get("/Left", None)
1987 @property
1988 def right(self) -> Optional[FloatObject]:
1989 """Read-only property accessing the right horizontal coordinate."""
1990 return self.get("/Right", None)
1992 @property
1993 def top(self) -> Optional[FloatObject]:
1994 """Read-only property accessing the top vertical coordinate."""
1995 return self.get("/Top", None)
1997 @property
1998 def bottom(self) -> Optional[FloatObject]:
1999 """Read-only property accessing the bottom vertical coordinate."""
2000 return self.get("/Bottom", None)
2002 @property
2003 def color(self) -> Optional["ArrayObject"]:
2004 """Read-only property accessing the color in (R, G, B) with values 0.0-1.0."""
2005 return cast(
2006 "ArrayObject",
2007 self.get("/C", ArrayObject([FloatObject(0), FloatObject(0), FloatObject(0)])),
2008 )
2010 @property
2011 def font_format(self) -> Optional[OutlineFontFlag]:
2012 """
2013 Read-only property accessing the font type.
2015 1=italic, 2=bold, 3=both
2016 """
2017 return OutlineFontFlag(self.get("/F", 0))
2019 @property
2020 def outline_count(self) -> Optional[int]:
2021 """
2022 Read-only property accessing the outline count.
2024 positive = expanded
2025 negative = collapsed
2026 absolute value = number of visible descendants at all levels
2027 """
2028 return self.get("/Count", None)