Source code for PyPDF2._reader

# Copyright (c) 2006, Mathieu Fenniak
# Copyright (c) 2007, Ashish Kulkarni <>
# All rights reserved.
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
# * Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
# * The name of the author may not be used to endorse or promote products
# derived from this software without specific prior written permission.

import re
import struct
import warnings
from hashlib import md5
from io import BytesIO
from typing import (

from ._page import PageObject, _VirtualList
from ._security import RC4_encrypt, _alg33_1, _alg34, _alg35
from ._utils import (
from .constants import CatalogAttributes as CA
from .constants import CatalogDictionary as CD
from .constants import Core as CO
from .constants import DocumentInformationAttributes as DI
from .constants import EncryptionDictAttributes as ED
from .constants import PageAttributes as PG
from .constants import PagesAttributes as PA
from .constants import StreamAttributes as SA
from .constants import TrailerKeys as TK
from .errors import PdfReadError, PdfReadWarning, PdfStreamError
from .generic import (
from .types import OutlinesType, PagemodeType
from .xmp import XmpInformation

def convert_to_int(d: bytes, size: int) -> Union[int, Tuple[Any, ...]]:
    if size > 8:
        raise PdfReadError("invalid size in convert_to_int")
    d = b_("\x00\x00\x00\x00\x00\x00\x00\x00") + b_(d)
    d = d[-8:]
    return struct.unpack(">q", d)[0]

def convertToInt(
    d: bytes, size: int
) -> Union[int, Tuple[Any, ...]]:  # pragma: no cover
    deprecate_with_replacement("convertToInt", "convert_to_int")
    return convert_to_int(d, size)

[docs]class DocumentInformation(DictionaryObject): """ A class representing the basic document metadata provided in a PDF File. This class is accessible through :py:class:`PdfReader.metadata<PyPDF2.PdfReader.metadata>`. All text properties of the document metadata have *two* properties, eg. author and author_raw. The non-raw property will always return a ``TextStringObject``, making it ideal for a case where the metadata is being displayed. The raw property can sometimes return a ``ByteStringObject``, if PyPDF2 was unable to decode the string's text encoding; this requires additional safety in the caller and therefore is not as commonly accessed. """ def __init__(self) -> None: DictionaryObject.__init__(self) def _get_text(self, key: str) -> Optional[str]: retval = self.get(key, None) if isinstance(retval, TextStringObject): return retval return None
[docs] def getText(self, key: str) -> Optional[str]: # pragma: no cover """ The text value of the specified key or None. .. deprecated:: 1.28.0 Use the attributes (e.g. :py:attr:`title` / :py:attr:`author`). """ deprecate_no_replacement("getText") return self._get_text(key)
@property def title(self) -> Optional[str]: """Read-only property accessing the document's **title**. Returns a unicode string (``TextStringObject``) or ``None`` if the title is not specified.""" return ( self._get_text(DI.TITLE) or self.get(DI.TITLE).get_object() # type: ignore if self.get(DI.TITLE) else None ) @property def title_raw(self) -> Optional[str]: """The "raw" version of title; can return a ``ByteStringObject``.""" return self.get(DI.TITLE) @property def author(self) -> Optional[str]: """Read-only property accessing the document's **author**. Returns a unicode string (``TextStringObject``) or ``None`` if the author is not specified.""" return self._get_text(DI.AUTHOR) @property def author_raw(self) -> Optional[str]: """The "raw" version of author; can return a ``ByteStringObject``.""" return self.get(DI.AUTHOR) @property def subject(self) -> Optional[str]: """Read-only property accessing the document's **subject**. Returns a unicode string (``TextStringObject``) or ``None`` if the subject is not specified.""" return self._get_text(DI.SUBJECT) @property def subject_raw(self) -> Optional[str]: """The "raw" version of subject; can return a ``ByteStringObject``.""" return self.get(DI.SUBJECT) @property def creator(self) -> Optional[str]: """Read-only property accessing the document's **creator**. If the document was converted to PDF from another format, this is the name of the application (e.g. OpenOffice) that created the original document from which it was converted. Returns a unicode string (``TextStringObject``) or ``None`` if the creator is not specified.""" return self._get_text(DI.CREATOR) @property def creator_raw(self) -> Optional[str]: """The "raw" version of creator; can return a ``ByteStringObject``.""" return self.get(DI.CREATOR) @property def producer(self) -> Optional[str]: """Read-only property accessing the document's **producer**. If the document was converted to PDF from another format, this is the name of the application (for example, OSX Quartz) that converted it to PDF. Returns a unicode string (``TextStringObject``) or ``None`` if the producer is not specified.""" return self._get_text(DI.PRODUCER) @property def producer_raw(self) -> Optional[str]: """The "raw" version of producer; can return a ``ByteStringObject``.""" return self.get(DI.PRODUCER)
[docs]class PdfReader: """ Initialize a PdfReader object. This operation can take some time, as the PDF stream's cross-reference tables are read into memory. :param stream: A File object or an object that supports the standard read and seek methods similar to a File object. Could also be a string representing a path to a PDF file. :param bool strict: Determines whether user should be warned of all problems and also causes some correctable problems to be fatal. Defaults to ``False``. :param None/str/bytes password: Decrypt PDF file at initialization. If the password is None, the file will not be decrypted. Defaults to ``None`` """ def __init__( self, stream: StrByteType, strict: bool = False, password: Union[None, str, bytes] = None, ) -> None: self.strict = strict self.flattened_pages: Optional[List[PageObject]] = None self.resolved_objects: Dict[Tuple[Any, Any], Optional[PdfObject]] = {} self.xref_index = 0 self._page_id2num: Optional[ Dict[Any, Any] ] = None # map page indirect_ref number to Page Number if hasattr(stream, "mode") and "b" not in stream.mode: # type: ignore warnings.warn( "PdfReader stream/file object is not in binary mode. " "It may not be read correctly.", PdfReadWarning, ) if isinstance(stream, str): with open(stream, "rb") as fh: stream = BytesIO(b_( = stream self._override_encryption = False if password is not None and self.decrypt(password) == 0: raise PdfReadError("Wrong password") @property def metadata(self) -> Optional[DocumentInformation]: """ Retrieve the PDF file's document information dictionary, if it exists. Note that some PDF files use metadata streams instead of docinfo dictionaries, and these metadata streams will not be accessed by this function. :return: the document information of this PDF file :rtype: :class:`DocumentInformation<pdf.DocumentInformation>` or ``None`` if none exists. """ if TK.INFO not in self.trailer: return None obj = self.trailer[TK.INFO] retval = DocumentInformation() retval.update(obj) # type: ignore return retval
[docs] def getDocumentInfo(self) -> Optional[DocumentInformation]: # pragma: no cover """ .. deprecated:: 1.28.0 Use the attribute :py:attr:`metadata` instead. """ deprecate_with_replacement("getDocumentInfo", "metadata") return self.metadata
@property def documentInfo(self) -> Optional[DocumentInformation]: # pragma: no cover """ .. deprecated:: 1.28.0 Use the attribute :py:attr:`metadata` instead. """ deprecate_with_replacement("documentInfo", "metadata") return self.metadata @property def xmp_metadata(self) -> Optional[XmpInformation]: """ XMP (Extensible Metadata Platform) data :return: a :class:`XmpInformation<xmp.XmpInformation>` instance that can be used to access XMP metadata from the document. :rtype: :class:`XmpInformation<xmp.XmpInformation>` or ``None`` if no metadata was found on the document root. """ try: self._override_encryption = True return self.trailer[TK.ROOT].xmp_metadata # type: ignore finally: self._override_encryption = False
[docs] def getXmpMetadata(self) -> Optional[XmpInformation]: # pragma: no cover """ .. deprecated:: 1.28.0 Use the attribute :py:attr:`xmp_metadata` instead. """ deprecate_with_replacement("getXmpMetadata", "xmp_metadata") return self.xmp_metadata
@property def xmpMetadata(self) -> Optional[XmpInformation]: # pragma: no cover """ .. deprecated:: 1.28.0 Use the attribute :py:attr:`xmp_metadata` instead. """ deprecate_with_replacement("xmpMetadata", "xmp_metadata") return self.xmp_metadata def _get_num_pages(self) -> int: """ Calculates the number of pages in this PDF file. :return: number of pages :rtype: int :raises PdfReadError: if file is encrypted and restrictions prevent this action. """ # Flattened pages will not work on an Encrypted PDF; # the PDF file's page count is used in this case. Otherwise, # the original method (flattened page count) is used. if self.is_encrypted: try: self._override_encryption = True self.decrypt("") return self.trailer[TK.ROOT]["/Pages"]["/Count"] # type: ignore except Exception: raise PdfReadError("File has not been decrypted") finally: self._override_encryption = False else: if self.flattened_pages is None: self._flatten() return len(self.flattened_pages) # type: ignore
[docs] def getNumPages(self) -> int: # pragma: no cover """ .. deprecated:: 1.28.0 Use :code:`len(reader.pages)` instead. """ deprecate_with_replacement("reader.getNumPages", "len(reader.pages)") return self._get_num_pages()
@property def numPages(self) -> int: # pragma: no cover """ .. deprecated:: 1.28.0 Use :code:`len(reader.pages)` instead. """ deprecate_with_replacement("reader.numPages", "len(reader.pages)") return self._get_num_pages()
[docs] def getPage(self, pageNumber: int) -> PageObject: # pragma: no cover """ .. deprecated:: 1.28.0 Use :code:`reader.pages[pageNumber]` instead. """ deprecate_with_replacement( "reader.getPage(pageNumber)", "reader.pages[pageNumber]" ) return self._get_page(pageNumber)
def _get_page(self, page_number: int) -> PageObject: """ Retrieves a page by number from this PDF file. :param int page_number: The page number to retrieve (pages begin at zero) :return: a :class:`PageObject<PyPDF2._page.PageObject>` instance. :rtype: :class:`PageObject<PyPDF2._page.PageObject>` """ # ensure that we're not trying to access an encrypted PDF # assert not self.trailer.has_key(TK.ENCRYPT) if self.flattened_pages is None: self._flatten() assert self.flattened_pages is not None, "hint for mypy" return self.flattened_pages[page_number] @property def namedDestinations(self) -> Dict[str, Any]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :py:attr:`named_destinations` instead. """ deprecate_with_replacement("namedDestinations", "named_destinations") return self.named_destinations @property def named_destinations(self) -> Dict[str, Any]: """ A read-only dictionary which maps names to :class:`Destinations<PyPDF2.generic.Destination>` """ return self._get_named_destinations() # A select group of relevant field attributes. For the complete list, # see section 8.6.2 of the PDF 1.7 reference.
[docs] def get_fields( self, tree: Optional[TreeObject] = None, retval: Optional[Dict[Any, Any]] = None, fileobj: Optional[Any] = None, ) -> Optional[Dict[str, Any]]: """ Extracts field data if this PDF contains interactive form fields. The *tree* and *retval* parameters are for recursive use. :param fileobj: A file object (usually a text file) to write a report to on all interactive form fields found. :return: A dictionary where each key is a field name, and each value is a :class:`Field<PyPDF2.generic.Field>` object. By default, the mapping name is used for keys. :rtype: dict, or ``None`` if form data could not be located. """ field_attributes = { "/FT": "Field Type", PA.PARENT: "Parent", "/T": "Field Name", "/TU": "Alternate Field Name", "/TM": "Mapping Name", "/Ff": "Field Flags", "/V": "Value", "/DV": "Default Value", } if retval is None: retval = {} catalog = cast(DictionaryObject, self.trailer[TK.ROOT]) # get the AcroForm tree if "/AcroForm" in catalog: tree = cast(Optional[TreeObject], catalog["/AcroForm"]) else: return None if tree is None: return retval self._check_kids(tree, retval, fileobj) for attr in field_attributes: if attr in tree: # Tree is a field self._build_field(tree, retval, fileobj, field_attributes) break if "/Fields" in tree: fields = cast(ArrayObject, tree["/Fields"]) for f in fields: field = f.get_object() self._build_field(field, retval, fileobj, field_attributes) return retval
[docs] def getFields( self, tree: Optional[TreeObject] = None, retval: Optional[Dict[Any, Any]] = None, fileobj: Optional[Any] = None, ) -> Optional[Dict[str, Any]]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :meth:`get_fields` instead. """ deprecate_with_replacement("getFields", "get_fields") return self.get_fields(tree, retval, fileobj)
def _build_field( self, field: Union[TreeObject, DictionaryObject], retval: Dict[Any, Any], fileobj: Any, field_attributes: Any, ) -> None: self._check_kids(field, retval, fileobj) try: key = field["/TM"] except KeyError: try: key = field["/T"] except KeyError: # Ignore no-name field for now return if fileobj: self._write_field(fileobj, field, field_attributes) fileobj.write("\n") retval[key] = Field(field) def _check_kids( self, tree: Union[TreeObject, DictionaryObject], retval: Any, fileobj: Any ) -> None: if PA.KIDS in tree: # recurse down the tree for kid in tree[PA.KIDS]: # type: ignore self.get_fields(kid.get_object(), retval, fileobj) def _write_field(self, fileobj: Any, field: Any, field_attributes: Any) -> None: order = ["/TM", "/T", "/FT", PA.PARENT, "/TU", "/Ff", "/V", "/DV"] for attr in order: attr_name = field_attributes[attr] try: if attr == "/FT": # Make the field type value more clear types = { "/Btn": "Button", "/Tx": "Text", "/Ch": "Choice", "/Sig": "Signature", } if field[attr] in types: fileobj.write(attr_name + ": " + types[field[attr]] + "\n") elif attr == PA.PARENT: # Let's just write the name of the parent try: name = field[PA.PARENT]["/TM"] except KeyError: name = field[PA.PARENT]["/T"] fileobj.write(attr_name + ": " + name + "\n") else: fileobj.write(attr_name + ": " + str(field[attr]) + "\n") except KeyError: # Field attribute is N/A or unknown, so don't write anything pass
[docs] def get_form_text_fields(self) -> Dict[str, Any]: """Retrieves form fields from the document with textual data (inputs, dropdowns)""" # Retrieve document form fields formfields = self.get_fields() if formfields is None: return {} return { formfields[field]["/T"]: formfields[field].get("/V") for field in formfields if formfields[field].get("/FT") == "/Tx" }
[docs] def getFormTextFields(self) -> Dict[str, Any]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :meth:`get_form_text_fields` instead. """ deprecate_with_replacement("getFormTextFields", "get_form_text_fields") return self.get_form_text_fields()
def _get_named_destinations( self, tree: Union[TreeObject, None] = None, retval: Optional[Any] = None, ) -> Dict[str, Any]: """ Retrieves the named destinations present in the document. :return: a dictionary which maps names to :class:`Destinations<PyPDF2.generic.Destination>`. :rtype: dict """ if retval is None: retval = {} catalog = cast(DictionaryObject, self.trailer[TK.ROOT]) # get the name tree if CA.DESTS in catalog: tree = cast(TreeObject, catalog[CA.DESTS]) elif CA.NAMES in catalog: names = cast(DictionaryObject, catalog[CA.NAMES]) if CA.DESTS in names: tree = cast(TreeObject, names[CA.DESTS]) if tree is None: return retval if PA.KIDS in tree: # recurse down the tree for kid in cast(ArrayObject, tree[PA.KIDS]): self._get_named_destinations(kid.get_object(), retval) # TABLE 3.33 Entries in a name tree node dictionary (PDF 1.7 specs) if CA.NAMES in tree: names = cast(DictionaryObject, tree[CA.NAMES]) for i in range(0, len(names), 2): key = cast(str, names[i].get_object()) value = names[i + 1].get_object() if isinstance(value, DictionaryObject) and "/D" in value: value = value["/D"] dest = self._build_destination(key, value) # type: ignore if dest is not None: retval[key] = dest return retval
[docs] def getNamedDestinations( self, tree: Union[TreeObject, None] = None, retval: Optional[Any] = None, ) -> Dict[str, Any]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :py:attr:`named_destinations` instead. """ deprecate_with_replacement("getNamedDestinations", "named_destinations") return self._get_named_destinations(tree, retval)
@property def outlines(self) -> OutlinesType: """ Read-only property for outlines present in the document. :return: a nested list of :class:`Destinations<PyPDF2.generic.Destination>`. """ return self._get_outlines() def _get_outlines( self, node: Optional[DictionaryObject] = None, outlines: Optional[Any] = None ) -> OutlinesType: if outlines is None: outlines = [] catalog = cast(DictionaryObject, self.trailer[TK.ROOT]) # get the outline dictionary and named destinations if CO.OUTLINES in catalog: try: lines = cast(DictionaryObject, catalog[CO.OUTLINES]) except PdfReadError: # this occurs if the /Outlines object reference is incorrect # for an example of such a file, see # so continue to load the file without the Bookmarks return outlines # TABLE 8.3 Entries in the outline dictionary if "/First" in lines: node = cast(DictionaryObject, lines["/First"]) self._namedDests = self._get_named_destinations() if node is None: return outlines # see if there are any more outlines while True: outline = self._build_outline(node) if outline: outlines.append(outline) # check for sub-outlines if "/First" in node: sub_outlines: List[Any] = [] self._get_outlines(cast(DictionaryObject, node["/First"]), sub_outlines) if sub_outlines: outlines.append(sub_outlines) if "/Next" not in node: break node = cast(DictionaryObject, node["/Next"]) return outlines
[docs] def getOutlines( self, node: Optional[DictionaryObject] = None, outlines: Optional[Any] = None ) -> OutlinesType: # pragma: no cover """ .. deprecated:: 1.28.0 Use :py:attr:`outlines` instead. """ deprecate_with_replacement("getOutlines", "outlines") return self._get_outlines(node, outlines)
def _get_page_number_by_indirect( self, indirect_ref: Union[None, int, NullObject, IndirectObject] ) -> int: """Generate _page_id2num""" if self._page_id2num is None: id2num = {} for i, x in enumerate(self.pages): id2num[x.indirect_ref.idnum] = i # type: ignore self._page_id2num = id2num if indirect_ref is None or isinstance(indirect_ref, NullObject): return -1 if isinstance(indirect_ref, int): idnum = indirect_ref else: idnum = indirect_ref.idnum assert self._page_id2num is not None, "hint for mypy" ret = self._page_id2num.get(idnum, -1) return ret
[docs] def get_page_number(self, page: PageObject) -> int: """ Retrieve page number of a given PageObject :param PageObject page: The page to get page number. Should be an instance of :class:`PageObject<PyPDF2._page.PageObject>` :return: the page number or -1 if page not found :rtype: int """ return self._get_page_number_by_indirect(page.indirect_ref)
[docs] def getPageNumber(self, page: PageObject) -> int: # pragma: no cover """ .. deprecated:: 1.28.0 Use :meth:`get_page_number` instead. """ deprecate_with_replacement("getPageNumber", "get_page_number") return self.get_page_number(page)
[docs] def get_destination_page_number(self, destination: Destination) -> int: """ Retrieve page number of a given Destination object. :param Destination destination: The destination to get page number. :return: the page number or -1 if page not found :rtype: int """ return self._get_page_number_by_indirect(
[docs] def getDestinationPageNumber( self, destination: Destination ) -> int: # pragma: no cover """ .. deprecated:: 1.28.0 Use :meth:`get_destination_page_number` instead. """ deprecate_with_replacement( "getDestinationPageNumber", "get_destination_page_number" ) return self.get_destination_page_number(destination)
def _build_destination( self, title: str, array: List[Union[NumberObject, IndirectObject, NullObject, DictionaryObject]], ) -> Destination: page, typ = array[0:2] array = array[2:] try: return Destination(title, page, typ, *array) # type: ignore except PdfReadError: warnings.warn(f"Unknown destination: {title} {array}", PdfReadWarning) if self.strict: raise else: # create a link to first Page tmp = self.pages[0].indirect_ref indirect_ref = NullObject() if tmp is None else tmp return Destination( title, indirect_ref, TextStringObject("/Fit") # type: ignore ) def _build_outline(self, node: DictionaryObject) -> Optional[Destination]: dest, title, outline = None, None, None if "/A" in node and "/Title" in node: # Action, section 8.5 (only type GoTo supported) title = node["/Title"] action = cast(DictionaryObject, node["/A"]) action_type = cast(NameObject, action["/S"]) if action_type == "/GoTo": dest = action["/D"] elif "/Dest" in node and "/Title" in node: # Destination, section 8.2.1 title = node["/Title"] dest = node["/Dest"] # if destination found, then create outline if dest: if isinstance(dest, ArrayObject): outline = self._build_destination(title, dest) # type: ignore elif isinstance(dest, str) and dest in self._namedDests: outline = self._namedDests[dest] outline[NameObject("/Title")] = title # type: ignore else: raise PdfReadError("Unexpected destination %r" % dest) return outline @property def pages(self) -> _VirtualList: """Read-only property that emulates a list of :py:class:`Page<PyPDF2._page.Page>` objects.""" return _VirtualList(self._get_num_pages, self._get_page) @property def page_layout(self) -> Optional[str]: """ Get the page layout. :return: Page layout currently being used. :rtype: ``str``, ``None`` if not specified .. list-table:: Valid ``layout`` values :widths: 50 200 * - /NoLayout - Layout explicitly not specified * - /SinglePage - Show one page at a time * - /OneColumn - Show one column at a time * - /TwoColumnLeft - Show pages in two columns, odd-numbered pages on the left * - /TwoColumnRight - Show pages in two columns, odd-numbered pages on the right * - /TwoPageLeft - Show two pages at a time, odd-numbered pages on the left * - /TwoPageRight - Show two pages at a time, odd-numbered pages on the right """ trailer = cast(DictionaryObject, self.trailer[TK.ROOT]) if CD.PAGE_LAYOUT in trailer: return cast(NameObject, trailer[CD.PAGE_LAYOUT]) return None
[docs] def getPageLayout(self) -> Optional[str]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :py:attr:`page_layout` instead. """ deprecate_with_replacement("getPageLayout", "page_layout") return self.page_layout
@property def pageLayout(self) -> Optional[str]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :py:attr:`page_layout` instead. """ deprecate_with_replacement("pageLayout", "page_layout") return self.page_layout @property def page_mode(self) -> Optional[PagemodeType]: """ Get the page mode. :return: Page mode currently being used. :rtype: ``str``, ``None`` if not specified .. list-table:: Valid ``mode`` values :widths: 50 200 * - /UseNone - Do not show outlines or thumbnails panels * - /UseOutlines - Show outlines (aka bookmarks) panel * - /UseThumbs - Show page thumbnails panel * - /FullScreen - Fullscreen view * - /UseOC - Show Optional Content Group (OCG) panel * - /UseAttachments - Show attachments panel """ try: return self.trailer[TK.ROOT]["/PageMode"] # type: ignore except KeyError: return None
[docs] def getPageMode(self) -> Optional[PagemodeType]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :py:attr:`page_mode` instead. """ deprecate_with_replacement("getPageMode", "page_mode") return self.page_mode
@property def pageMode(self) -> Optional[PagemodeType]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :py:attr:`page_mode` instead. """ deprecate_with_replacement("pageMode", "page_mode") return self.page_mode def _flatten( self, pages: Union[None, DictionaryObject, PageObject] = None, inherit: Optional[Dict[str, Any]] = None, indirect_ref: Optional[IndirectObject] = None, ) -> None: inheritable_page_attributes = ( NameObject(PG.RESOURCES), NameObject(PG.MEDIABOX), NameObject(PG.CROPBOX), NameObject(PG.ROTATE), ) if inherit is None: inherit = {} if pages is None: # Fix issue 327: set flattened_pages attribute only for # decrypted file catalog = self.trailer[TK.ROOT].get_object() pages = catalog["/Pages"].get_object() # type: ignore self.flattened_pages = [] t = "/Pages" if PA.TYPE in pages: t = pages[PA.TYPE] # type: ignore if t == "/Pages": for attr in inheritable_page_attributes: if attr in pages: inherit[attr] = pages[attr] for page in pages[PA.KIDS]: # type: ignore addt = {} if isinstance(page, IndirectObject): addt["indirect_ref"] = page self._flatten(page.get_object(), inherit, **addt) elif t == "/Page": for attr_in, value in list(inherit.items()): # if the page has it's own value, it does not inherit the # parent's value: if attr_in not in pages: pages[attr_in] = value page_obj = PageObject(self, indirect_ref) page_obj.update(pages) # TODO: Could flattened_pages be None at this point? self.flattened_pages.append(page_obj) # type: ignore def _get_object_from_stream( self, indirect_reference: IndirectObject ) -> Union[int, PdfObject, str]: # indirect reference to object in object stream # read the entire object stream into memory stmnum, idx = self.xref_objStm[indirect_reference.idnum] obj_stm: EncodedStreamObject = IndirectObject(stmnum, 0, self).get_object() # type: ignore # This is an xref to a stream, so its type better be a stream assert obj_stm["/Type"] == "/ObjStm" # /N is the number of indirect objects in the stream assert idx < obj_stm["/N"] stream_data = BytesIO(b_(obj_stm.get_data())) # type: ignore for i in range(obj_stm["/N"]): # type: ignore read_non_whitespace(stream_data), 1) objnum = NumberObject.read_from_stream(stream_data) read_non_whitespace(stream_data), 1) offset = NumberObject.read_from_stream(stream_data) read_non_whitespace(stream_data), 1) if objnum != indirect_reference.idnum: # We're only interested in one object continue if self.strict and idx != i: raise PdfReadError("Object is in wrong index.")["/First"] + offset), 0) # type: ignore # to cope with some case where the 'pointer' is on a white space read_non_whitespace(stream_data), 1) try: obj = read_object(stream_data, self) except PdfStreamError as exc: # Stream object cannot be read. Normally, a critical error, but # Adobe Reader doesn't complain, so continue (in strict mode?) warnings.warn( "Invalid stream (index %d) within object %d %d: %s" % (i, indirect_reference.idnum, indirect_reference.generation, exc), PdfReadWarning, ) if self.strict: raise PdfReadError("Can't read object stream: %s" % exc) # Replace with null. Hopefully it's nothing important. obj = NullObject() return obj if self.strict: raise PdfReadError("This is a fatal error in strict mode.") return NullObject()
[docs] def get_object(self, indirect_reference: IndirectObject) -> Optional[PdfObject]: retval = self.cache_get_indirect_object( indirect_reference.generation, indirect_reference.idnum ) if retval is not None: return retval if ( indirect_reference.generation == 0 and indirect_reference.idnum in self.xref_objStm ): retval = self._get_object_from_stream(indirect_reference) # type: ignore elif ( indirect_reference.generation in self.xref and indirect_reference.idnum in self.xref[indirect_reference.generation] ): start = self.xref[indirect_reference.generation][indirect_reference.idnum], 0) idnum, generation = self.read_object_header( if idnum != indirect_reference.idnum and self.xref_index: # Xref table probably had bad indexes due to not being zero-indexed if self.strict: raise PdfReadError( "Expected object ID (%d %d) does not match actual (%d %d); xref table not zero-indexed." % ( indirect_reference.idnum, indirect_reference.generation, idnum, generation, ) ) else: pass # xref table is corrected in non-strict mode elif idnum != indirect_reference.idnum and self.strict: # some other problem raise PdfReadError( "Expected object ID (%d %d) does not match actual (%d %d)." % ( indirect_reference.idnum, indirect_reference.generation, idnum, generation, ) ) if self.strict: assert generation == indirect_reference.generation retval = read_object(, self) # type: ignore # override encryption is used for the /Encrypt dictionary if not self._override_encryption and self.is_encrypted: # if we don't have the encryption key: if not hasattr(self, "_decryption_key"): raise PdfReadError("file has not been decrypted") # otherwise, decrypt here... pack1 = struct.pack("<i", indirect_reference.idnum)[:3] pack2 = struct.pack("<i", indirect_reference.generation)[:2] key = self._decryption_key + pack1 + pack2 assert len(key) == (len(self._decryption_key) + 5) md5_hash = md5(key).digest() key = md5_hash[: min(16, len(self._decryption_key) + 5)] retval = self._decrypt_object(retval, key) # type: ignore else: warnings.warn( "Object %d %d not defined." % (indirect_reference.idnum, indirect_reference.generation), PdfReadWarning, ) if self.strict: raise PdfReadError("Could not find object.") self.cache_indirect_object( indirect_reference.generation, indirect_reference.idnum, retval ) return retval
[docs] def getObject( self, indirectReference: IndirectObject ) -> Optional[PdfObject]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :meth:`get_object` instead. """ deprecate_with_replacement("getObject", "get_object") return self.get_object(indirectReference)
def _decrypt_object( self, obj: Union[ ArrayObject, BooleanObject, ByteStringObject, DictionaryObject, FloatObject, IndirectObject, NameObject, NullObject, NumberObject, StreamObject, TextStringObject, ], key: Union[str, bytes], ) -> PdfObject: if isinstance(obj, (ByteStringObject, TextStringObject)): obj = createStringObject(RC4_encrypt(key, obj.original_bytes)) elif isinstance(obj, StreamObject): obj._data = RC4_encrypt(key, obj._data) elif isinstance(obj, DictionaryObject): for dictkey, value in list(obj.items()): obj[dictkey] = self._decrypt_object(value, key) elif isinstance(obj, ArrayObject): for i in range(len(obj)): obj[i] = self._decrypt_object(obj[i], key) return obj
[docs] def read_object_header(self, stream: StreamType) -> Tuple[int, int]: # Should never be necessary to read out whitespace, since the # cross-reference table should put us in the right spot to read the # object header. In reality... some files have stupid cross reference # tables that are off by whitespace bytes. extra = False skip_over_comment(stream) extra |= skip_over_whitespace(stream), 1) idnum = read_until_whitespace(stream) extra |= skip_over_whitespace(stream), 1) generation = read_until_whitespace(stream) extra |= skip_over_whitespace(stream), 1) # although it's not used, it might still be necessary to read _obj = # noqa: F841 read_non_whitespace(stream), 1) if extra and self.strict: warnings.warn( "Superfluous whitespace found in object header %s %s" % (idnum, generation), # type: ignore PdfReadWarning, ) return int(idnum), int(generation)
[docs] def readObjectHeader( self, stream: StreamType ) -> Tuple[int, int]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :meth:`read_object_header` instead. """ deprecate_with_replacement("readObjectHeader", "read_object_header") return self.read_object_header(stream)
[docs] def cache_get_indirect_object( self, generation: int, idnum: int ) -> Optional[PdfObject]: return self.resolved_objects.get((generation, idnum))
[docs] def cacheGetIndirectObject( self, generation: int, idnum: int ) -> Optional[PdfObject]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :meth:`cache_get_indirect_object` instead. """ deprecate_with_replacement( "cacheGetIndirectObject", "cache_get_indirect_object" ) return self.cache_get_indirect_object(generation, idnum)
[docs] def cache_indirect_object( self, generation: int, idnum: int, obj: Optional[PdfObject] ) -> Optional[PdfObject]: if (generation, idnum) in self.resolved_objects: msg = f"Overwriting cache for {generation} {idnum}" if self.strict: raise PdfReadError(msg) else: warnings.warn(msg) self.resolved_objects[(generation, idnum)] = obj return obj
[docs] def cacheIndirectObject( self, generation: int, idnum: int, obj: Optional[PdfObject] ) -> Optional[PdfObject]: # pragma: no cover """ .. deprecated:: 1.28.0 Use :meth:`cache_indirect_object` instead. """ deprecate_with_replacement("cacheIndirectObject", "cache_indirect_object") return self.cache_indirect_object(generation, idnum, obj)
[docs] def read(self, stream: StreamType) -> None: # start at the end:, 2) if not stream.tell(): raise PdfReadError("Cannot read an empty file") if self.strict:, 0) header_byte = if header_byte != b"%PDF-": raise PdfReadError( "PDF starts with '{}', but '%PDF-' expected".format( header_byte.decode("utf8") ) ), 2) last_mb = stream.tell() - 1024 * 1024 + 1 # offset of last MB of stream line = b_("") while line[:5] != b_("%%EOF"): if stream.tell() < last_mb: raise PdfReadError("EOF marker not found") line = self.read_next_end_line(stream) startxref = self._find_startxref_pos(stream) # check and eventually correct the startxref only in not strict xref_issue_nr = self._get_xref_issues(stream, startxref) if xref_issue_nr != 0: if self.strict and xref_issue_nr: raise PdfReadError("Broken xref table") else: warnings.warn( f"incorrect startxref pointer({xref_issue_nr})", PdfReadWarning ) # read all cross reference tables and their trailers self.xref: Dict[Any, Any] = {} self.xref_objStm: Dict[Any, Any] = {} self.trailer = DictionaryObject() while True: # load the xref table, 0) x = if x == b_("x"): self._read_standard_xref_table(stream) read_non_whitespace(stream), 1) new_trailer = cast(Dict[str, Any], read_object(stream, self)) for key, value in new_trailer.items(): if key not in self.trailer: self.trailer[key] = value if "/Prev" in new_trailer: startxref = new_trailer["/Prev"] else: break elif xref_issue_nr: try: self._rebuild_xref_table(stream) break except Exception: xref_issue_nr = 0 elif x.isdigit(): xrefstream = self._read_pdf15_xref_stream(stream) trailer_keys = TK.ROOT, TK.ENCRYPT, TK.INFO, TK.ID for key in trailer_keys: if key in xrefstream and key not in self.trailer: self.trailer[NameObject(key)] = xrefstream.raw_get(key) if "/Prev" in xrefstream: startxref = cast(int, xrefstream["/Prev"]) else: break else: # some PDFs have /Prev=0 in the trailer, instead of no /Prev if startxref == 0: if self.strict: raise PdfReadError( "/Prev=0 in the trailer (try opening with strict=False)" ) else: warnings.warn( "/Prev=0 in the trailer - assuming there" " is no previous xref table" ) break # bad xref character at startxref. Let's see if we can find # the xref table nearby, as we've observed this error with an # off-by-one before., 1) tmp = xref_loc = tmp.find(b_("xref")) if xref_loc != -1: startxref -= 10 - xref_loc continue # No explicit xref table, try finding a cross-reference stream., 0) found = False for look in range(5): if # This is not a standard PDF, consider adding a warning startxref += look found = True break if found: continue # no xref table found at specified location raise PdfReadError("Could not find xref table at specified location") # if not zero-indexed, verify that the table is correct; change it if necessary if self.xref_index and not self.strict: loc = stream.tell() for gen in self.xref: if gen == 65535: continue for id in self.xref[gen]:[gen][id], 0) try: pid, pgen = self.read_object_header(stream) except ValueError: break if pid == id - self.xref_index: self._zero_xref(gen) break # if not, then either it's just plain wrong, or the # non-zero-index is actually correct, 0) # return to where it was
def _find_startxref_pos(self, stream: StreamType) -> int: """Find startxref entry - the location of the xref table""" line = self.read_next_end_line(stream) try: startxref = int(line) except ValueError: # 'startxref' may be on the same line as the location if not line.startswith(b_("startxref")): raise PdfReadError("startxref not found") startxref = int(line[9:].strip()) warnings.warn("startxref on same line as offset", PdfReadWarning) else: line = self.read_next_end_line(stream) if line[:9] != b_("startxref"): raise PdfReadError("startxref not found") return startxref def _read_standard_xref_table(self, stream: StreamType) -> None: # standard cross-reference table ref = if ref[:3] != b_("ref"): raise PdfReadError("xref table read error") read_non_whitespace(stream), 1) firsttime = True # check if the first time looking at the xref table while True: num = cast(int, read_object(stream, self)) if firsttime and num != 0: self.xref_index = num if self.strict: warnings.warn( "Xref table not zero-indexed. ID numbers for objects will be corrected.", PdfReadWarning, ) # if table not zero indexed, could be due to error from when PDF was created # which will lead to mismatched indices later on, only warned and corrected if self.strict==True firsttime = False read_non_whitespace(stream), 1) size = cast(int, read_object(stream, self)) read_non_whitespace(stream), 1) cnt = 0 while cnt < size: line = # It's very clear in section 3.4.3 of the PDF spec # that all cross-reference table lines are a fixed # 20 bytes (as of PDF 1.7). However, some files have # 21-byte entries (or more) due to the use of \r\n # (CRLF) EOL's. Detect that case, and adjust the line # until it does not begin with a \r (CR) or \n (LF). while line[0] in b_("\x0D\x0A"): + 1, 1) line = # On the other hand, some malformed PDF files # use a single character EOL without a preceeding # space. Detect that case, and seek the stream # back one character. (0-9 means we've bled into # the next xref entry, t means we've bled into the # text "trailer"): if line[-1] in b_("0123456789t"):, 1) offset_b, generation_b = line[:16].split(b_(" ")) offset, generation = int(offset_b), int(generation_b) if generation not in self.xref: self.xref[generation] = {} if num in self.xref[generation]: # It really seems like we should allow the last # xref table in the file to override previous # ones. Since we read the file backwards, assume # any existing key is already set correctly. pass else: self.xref[generation][num] = offset cnt += 1 num += 1 read_non_whitespace(stream), 1) trailertag = if trailertag != b_("trailer"): # more xrefs!, 1) else: break def _read_pdf15_xref_stream( self, stream: StreamType ) -> Union[ContentStream, EncodedStreamObject, DecodedStreamObject]: # PDF 1.5+ Cross-Reference Stream, 1) idnum, generation = self.read_object_header(stream) xrefstream = cast(ContentStream, read_object(stream, self)) assert xrefstream["/Type"] == "/XRef" self.cache_indirect_object(generation, idnum, xrefstream) stream_data = BytesIO(b_(xrefstream.get_data())) # Index pairs specify the subsections in the dictionary. If # none create one subsection that spans everything. idx_pairs = xrefstream.get("/Index", [0, xrefstream.get("/Size")]) entry_sizes = cast(Dict[Any, Any], xrefstream.get("/W")) assert len(entry_sizes) >= 3 if self.strict and len(entry_sizes) > 3: raise PdfReadError("Too many entry sizes: %s" % entry_sizes) def get_entry(i: int) -> Union[int, Tuple[int, ...]]: # Reads the correct number of bytes for each entry. See the # discussion of the W parameter in PDF spec table 17. if entry_sizes[i] > 0: d =[i]) return convert_to_int(d, entry_sizes[i]) # PDF Spec Table 17: A value of zero for an element in the # W array indicates...the default value shall be used if i == 0: return 1 # First value defaults to 1 else: return 0 def used_before(num: int, generation: Union[int, Tuple[int, ...]]) -> bool: # We move backwards through the xrefs, don't replace any. return num in self.xref.get(generation, []) or num in self.xref_objStm # Iterate through each subsection self._read_xref_subsections(idx_pairs, get_entry, used_before) return xrefstream @staticmethod def _get_xref_issues(stream: StreamType, startxref: int) -> int: """Return an int which indicates an issue. 0 means there is no issue.""" - 1, 0) # -1 to check character before line = if line not in b_("\r\n \t"): return 1 line = if line != b_("xref"): # not an xref so check if it is an XREF object line = b_("") while line in b_("0123456789 \t"): line = if line == b_(""): return 2 line += # 1 char already read, +2 to check "obj" if line.lower() != b_("obj"): return 3 # while in b_(" \t\r\n"): # pass # line = # check that it is xref obj # if b_("/xref") not in line.lower(): # return 4 return 0 def _rebuild_xref_table(self, stream: StreamType) -> None: self.xref = {}, 0) f_ = for m in re.finditer(b_(r"[\r\n \t][ \t]*(\d+)[ \t]+(\d+)[ \t]+obj"), f_): idnum = int( generation = int( if generation not in self.xref: self.xref[generation] = {} self.xref[generation][idnum] = m.start(1) trailer_pos = f_.rfind(b"trailer") - len(f_) + 7, 2) # code below duplicated read_non_whitespace(stream), 1) # there might be something that is not a dict (see #856) new_trailer = cast(Dict[Any, Any], read_object(stream, self)) for key, value in list(new_trailer.items()): if key not in self.trailer: self.trailer[key] = value def _read_xref_subsections( self, idx_pairs: List[int], get_entry: Callable[[int], Union[int, Tuple[int, ...]]], used_before: Callable[[int, Union[int, Tuple[int, ...]]], bool], ) -> None: last_end = 0 for start, size in self._pairs(idx_pairs): # The subsections must increase assert start >= last_end last_end = start + size for num in range(start, start + size): # The first entry is the type xref_type = get_entry(0) # The rest of the elements depend on the xref_type if xref_type == 0: # linked list of free objects next_free_object = get_entry(1) # noqa: F841 next_generation = get_entry(2) # noqa: F841 elif xref_type == 1: # objects that are in use but are not compressed byte_offset = get_entry(1) generation = get_entry(2) if generation not in self.xref: self.xref[generation] = {} if not used_before(num, generation): self.xref[generation][num] = byte_offset elif xref_type == 2: # compressed objects objstr_num = get_entry(1) obstr_idx = get_entry(2) generation = 0 # PDF spec table 18, generation is 0 if not used_before(num, generation): self.xref_objStm[num] = (objstr_num, obstr_idx) elif self.strict: raise PdfReadError("Unknown xref type: %s" % xref_type) def _zero_xref(self, generation: int) -> None: self.xref[generation] = { k - self.xref_index: v for (k, v) in list(self.xref[generation].items()) } def _pairs(self, array: List[int]) -> Iterable[Tuple[int, int]]: i = 0 while True: yield array[i], array[i + 1] i += 2 if (i + 1) >= len(array): break
[docs] def read_next_end_line(self, stream: StreamType, limit_offset: int = 0) -> bytes: line_parts = [] while True: # Prevent infinite loops in malformed PDFs if stream.tell() == 0 or stream.tell() == limit_offset: raise PdfReadError("Could not read malformed PDF file") x = if stream.tell() < 2: raise PdfReadError("EOL marker not found"), 1) if x == b_("\n") or x == b_("\r"): # \n = LF; \r = CR crlf = False while x == b_("\n") or x == b_("\r"): x = if x == b_("\n") or x == b_("\r"): # account for CR+LF, 1) crlf = True if stream.tell() < 2: raise PdfReadError("EOL marker not found"), 1) 2 if crlf else 1, 1 ) # if using CR+LF, go back 2 bytes, else 1 break else: line_parts.append(x) line_parts.reverse() return b"".join(line_parts)
[docs] def readNextEndLine( self, stream: StreamType, limit_offset: int = 0 ) -> bytes: # pragma: no cover """ .. deprecated:: 1.28.0 Use :meth:`read_next_end_line` instead. """ deprecate_with_replacement("readNextEndLine", "read_next_end_line") return self.read_next_end_line(stream, limit_offset)
[docs] def decrypt(self, password: Union[str, bytes]) -> int: """ When using an encrypted / secured PDF file with the PDF Standard encryption handler, this function will allow the file to be decrypted. It checks the given password against the document's user password and owner password, and then stores the resulting decryption key if either password is correct. It does not matter which password was matched. Both passwords provide the correct decryption key that will allow the document to be used with this library. :param str password: The password to match. :return: ``0`` if the password failed, ``1`` if the password matched the user password, and ``2`` if the password matched the owner password. :rtype: int :raises NotImplementedError: if document uses an unsupported encryption method. """ self._override_encryption = True try: return self._decrypt(password) finally: self._override_encryption = False
[docs] def decode_permissions(self, permissions_code: int) -> Dict[str, bool]: # Takes the permissions as an integer, returns the allowed access permissions = {} permissions["print"] = permissions_code & (1 << 3 - 1) != 0 # bit 3 permissions["modify"] = permissions_code & (1 << 4 - 1) != 0 # bit 4 permissions["copy"] = permissions_code & (1 << 5 - 1) != 0 # bit 5 permissions["annotations"] = permissions_code & (1 << 6 - 1) != 0 # bit 6 permissions["forms"] = permissions_code & (1 << 9 - 1) != 0 # bit 9 permissions["accessability"] = permissions_code & (1 << 10 - 1) != 0 # bit 10 permissions["assemble"] = permissions_code & (1 << 11 - 1) != 0 # bit 11 permissions["print_high_quality"] = ( permissions_code & (1 << 12 - 1) != 0 ) # bit 12 return permissions
def _decrypt(self, password: Union[str, bytes]) -> int: # Decrypts data as per Section 3.5 (page 117) of PDF spec v1.7 # "The security handler defines the use of encryption and decryption in # the document, using the rules specified by the CF, StmF, and StrF entries" encrypt = cast(DictionaryObject, self.trailer[TK.ENCRYPT].get_object()) # /Encrypt Keys: # Filter (name) : "name of the preferred security handler " # V (number) : Algorithm Code # Length (integer): Length of encryption key, in bits # CF (dictionary) : Crypt filter # StmF (name) : Name of the crypt filter that is used by default when decrypting streams # StrF (name) : The name of the crypt filter that is used when decrypting all strings in the document # R (number) : Standard security handler revision number # U (string) : A 32-byte string, based on the user password # P (integer) : Permissions allowed with user access if encrypt["/Filter"] != "/Standard": raise NotImplementedError( "only Standard PDF encryption handler is available" ) encrypt_v = cast(int, encrypt["/V"]) if encrypt_v not in (1, 2): raise NotImplementedError( "only algorithm code 1 and 2 are supported. This PDF uses code %s" % encrypt_v ) user_password, key = self._authenticate_user_password(password) if user_password: self._decryption_key = key return 1 else: rev = cast(int, encrypt["/R"].get_object()) if rev == 2: keylen = 5 else: keylen = cast(int, encrypt[SA.LENGTH].get_object()) // 8 key = _alg33_1(password, rev, keylen) real_O = cast(bytes, encrypt["/O"].get_object()) if rev == 2: userpass = RC4_encrypt(key, real_O) else: val = real_O for i in range(19, -1, -1): new_key = b_("") for l in range(len(key)): new_key += b_(chr(ord_(key[l]) ^ i)) val = RC4_encrypt(new_key, val) userpass = val owner_password, key = self._authenticate_user_password(userpass) if owner_password: self._decryption_key = key return 2 return 0 def _authenticate_user_password( self, password: Union[str, bytes] ) -> Tuple[bool, bytes]: encrypt = cast( Optional[DictionaryObject], self.trailer[TK.ENCRYPT].get_object() ) if encrypt is None: raise Exception( "_authenticateUserPassword was called on unencrypted document" ) rev = cast(int, encrypt[ED.R].get_object()) owner_entry = cast(ByteStringObject, encrypt[ED.O].get_object()) p_entry = cast(int, encrypt[ED.P].get_object()) if TK.ID in self.trailer: id_entry = cast(ArrayObject, self.trailer[TK.ID].get_object()) else: # Some documents may not have a /ID, use two empty # byte strings instead. Solves # id_entry = ArrayObject([ByteStringObject(b""), ByteStringObject(b"")]) id1_entry = id_entry[0].get_object() real_U = encrypt[ED.U].get_object().original_bytes # type: ignore if rev == 2: U, key = _alg34(password, owner_entry, p_entry, id1_entry) elif rev >= 3: U, key = _alg35( password, rev, encrypt[SA.LENGTH].get_object() // 8, # type: ignore owner_entry, p_entry, id1_entry, encrypt.get(ED.ENCRYPT_METADATA, BooleanObject(False)).get_object(), # type: ignore ) U, real_U = U[:16], real_U[:16] return U == real_U, key @property def is_encrypted(self) -> bool: """ Read-only boolean property showing whether this PDF file is encrypted. Note that this property, if true, will remain true even after the :meth:`decrypt()<PyPDF2.PdfReader.decrypt>` method is called. """ return TK.ENCRYPT in self.trailer
[docs] def getIsEncrypted(self) -> bool: # pragma: no cover """ .. deprecated:: 1.28.0 Use :py:attr:`is_encrypted` instead. """ deprecate_with_replacement("getIsEncrypted", "is_encrypted") return self.is_encrypted
@property def isEncrypted(self) -> bool: # pragma: no cover """ .. deprecated:: 1.28.0 Use :py:attr:`is_encrypted` instead. """ deprecate_with_replacement("isEncrypted", "is_encrypted") return self.is_encrypted
class PdfFileReader(PdfReader): # pragma: no cover def __init__(self, *args: Any, **kwargs: Any) -> None: deprecate_with_replacement("PdfFileReader", "PdfReader") if "strict" not in kwargs and len(args) < 2: kwargs["strict"] = True # maintain the default super().__init__(*args, **kwargs)