""" This module implements support for PDF-specific signing functionality. """ import asyncio import enum import logging import uuid import warnings from dataclasses import dataclass, field from datetime import datetime from typing import IO, Any, Dict, Iterable, List, Optional, Set, Tuple, Union import tzlocal from asn1crypto import algos, cms, crl, keys, ocsp from asn1crypto import pdf as asn1_pdf from asn1crypto.core import VOID from cryptography.hazmat.primitives import hashes from pyhanko_certvalidator import CertificateValidator, ValidationContext from pyhanko_certvalidator.errors import PathBuildingError, PathValidationError from pyhanko_certvalidator.path import ValidationPath from pyhanko_certvalidator.validate import ACValidationResult, async_validate_ac from pyhanko.pdf_utils import generic, misc from pyhanko.pdf_utils.crypt import pdfmac from pyhanko.pdf_utils.generic import pdf_name from pyhanko.pdf_utils.incremental_writer import IncrementalPdfFileWriter from pyhanko.pdf_utils.reader import PdfFileReader from pyhanko.pdf_utils.writer import BasePdfFileWriter from pyhanko.sign import attributes from pyhanko.sign.ades.api import CAdESSignedAttrSpec from pyhanko.sign.fields import ( FieldMDPSpec, InvisSigSettings, MDPPerm, SeedLockDocument, SigAuthType, SigFieldSpec, SigSeedSubFilter, SigSeedValFlags, SigSeedValueSpec, enumerate_sig_fields, ) from pyhanko.sign.general import ( SigningError, find_unique_cms_attribute, get_cms_hash_algo_for_mechanism, get_pyca_cryptography_hash, simple_cms_attribute, ) from pyhanko.sign.timestamps import TimeStamper from pyhanko.stamp import BaseStampStyle from . import constants from .cms_embedder import ( PdfCMSEmbedder, SigAppearanceSetup, SigIOSetup, SigMDPSetup, SigObjSetup, ) from .pdf_byterange import ( BuildProps, DocumentTimestamp, PreparedByteRangeDigest, SignatureObject, ) from .pdf_cms import PdfCMSSignedAttributes, Signer, select_suitable_signing_md __all__ = [ 'PdfSignatureMetadata', 'DSSContentSettings', 'TimestampDSSContentSettings', 'GeneralDSSContentSettings', 'SigDSSPlacementPreference', 'PdfTimeStamper', 'PdfSigner', 'PdfSigningSession', 'PdfTBSDocument', 'PdfPostSignatureDocument', 'PreSignValidationStatus', 'PostSignInstructions', ] from ...pdf_utils.crypt import SerialisedCredential from ..attributes import CMSAttributeProvider logger = logging.getLogger(__name__) @dataclass(frozen=True) class GeneralDSSContentSettings: """ .. versionadded:: 0.8.0 Settings that govern DSS creation and updating in general. """ include_vri: bool = True """ Flag to control whether to create and update entries in the VRI dictionary. The default is to always update the VRI dictionary. .. note:: The VRI dictionary is a relic of the past that is effectively deprecated in the current PAdES standards, and most modern validators don't rely on it being there. That said, there's no real harm in creating these entries, other than that it occasionally forces DSS updates where none would otherwise be necessary, and that it prevents the DSS from being updated prior to signing (as opposed to after signing). """ skip_if_unneeded: bool = True """ Do not perform a write if updating the DSS would not add any new information. .. note:: This setting is only used if the DSS update would happen in its own revision. """ class SigDSSPlacementPreference(enum.Enum): """ .. versionadded:: 0.8.0 Preference for where to perform a DSS update with validation information for a specific signature. """ TOGETHER_WITH_SIGNATURE = enum.auto() """ Update the DSS in the revision that contains the signature. Doing so can be useful to create a PAdES-B-LT signature in a single revision. Such signatures can be processed by a validator that isn't capable of incremental update analysis. .. warning:: This setting can only be used if :attr:`include_vri` is ``False``. """ SEPARATE_REVISION = enum.auto() """ Always perform the DSS update in a separate revision, after the signature, but before any timestamps are added. .. note:: This is the old default behaviour. """ TOGETHER_WITH_NEXT_TS = enum.auto() """ If the signing workflow includes a document timestamp after the signature, update the DSS in the same revision as the timestamp. In the absence of document timestamps, this is equivalent to :attr:`SEPARATE_REVISION`. .. warning:: This option controls the addition of validation info for the signature and its associated signature timestamp, not the validation info for the document timestamp itself. See :attr:`.DSSContentSettings.next_ts_settings`. In most practical situations, the distinction is only relevant in interrupted signing workflows (see :ref:`interrupted-signing`), where the lifecycle of the validation context is out of pyHanko's hands. """ @dataclass(frozen=True) class TimestampDSSContentSettings(GeneralDSSContentSettings): """ .. versionadded:: 0.8.0 Settings for a DSS update with validation information for a document timestamp. .. note:: In most workflows, adding a document timestamp doesn't trigger any DSS updates beyond VRI additions, because the same TSA is used for signature timestamps and for document timestamps. """ update_before_ts: bool = False """ Perform DSS update before creating the timestamp, instead of after. .. warning:: This setting can only be used if :attr:`include_vri` is ``False``. """ def assert_viable(self): """ Check settings for consistency, and raise :class:`.SigningError` otherwise. """ if self.include_vri and self.update_before_ts: raise SigningError( "If VRI entries are to be included, DSS updates can only be " "performed after the timestamp in question was created." ) @dataclass(frozen=True) class DSSContentSettings(GeneralDSSContentSettings): """ .. versionadded:: 0.8.0 Settings for a DSS update with validation information for a signature. """ placement: SigDSSPlacementPreference = ( SigDSSPlacementPreference.TOGETHER_WITH_NEXT_TS ) """ Preference for where to perform a DSS update with validation information for a specific signature. See :class:`.SigDSSPlacementPreference`. The default is :attr:`.SigDSSPlacementPreference.TOGETHER_WITH_NEXT_TS`. """ next_ts_settings: Optional[TimestampDSSContentSettings] = None """ Explicit settings for DSS updates pertaining to a document timestamp added as part of the same signing workflow, if applicable. If ``None``, a default will be generated based on the values of this settings object. .. note:: When consuming :class:`.DSSContentSettings` objects, you should call :meth:`get_settings_for_ts` instead of relying on the value of this field. """ def get_settings_for_ts(self) -> TimestampDSSContentSettings: """ Retrieve DSS update settings for document timestamps that are part of our signing workflow, if there are any. """ ts_settings = self.next_ts_settings if ts_settings is not None: return ts_settings update_before_ts = ( self.placement == SigDSSPlacementPreference.TOGETHER_WITH_SIGNATURE ) return TimestampDSSContentSettings( include_vri=self.include_vri, skip_if_unneeded=self.skip_if_unneeded, update_before_ts=update_before_ts, ) def assert_viable(self): """ Check settings for consistency, and raise :class:`.SigningError` otherwise. """ pre_sign = ( self.placement == SigDSSPlacementPreference.TOGETHER_WITH_SIGNATURE ) if self.include_vri and pre_sign: raise SigningError( "If VRI entries are to be included, DSS updates can only be " "performed after the signature in question was created." ) self.get_settings_for_ts().assert_viable() @dataclass(frozen=True) class PdfSignatureMetadata: """ Specification for a PDF signature. """ field_name: Optional[str] = None """ The name of the form field to contain the signature. If there is only one available signature field, the name may be inferred. """ md_algorithm: Optional[str] = None """ The name of the digest algorithm to use. It should be supported by `pyca/cryptography`. If ``None``, :func:`.select_suitable_signing_md` will be invoked to generate a suitable default, unless a seed value dictionary happens to be available. """ location: Optional[str] = None """ Location of signing. """ reason: Optional[str] = None """ Reason for signing (textual). """ contact_info: Optional[str] = None """ Information provided by the signer to enable the receiver to contact the signer to verify the signature. """ name: Optional[str] = None """ Name of the signer. This value is usually not necessary to set, since it should appear on the signer's certificate, but there are cases where it might be useful to specify it here (e.g. in situations where signing is delegated to a trusted third party). """ app_build_props: Optional[BuildProps] = None """ Properties of the application that created the signature. If specified, this data will be recorded in the **Prop_Build** dictionary of the signature. """ prop_auth_time: Optional[int] = None """ Number of seconds since signer was last authenticated. """ prop_auth_type: Optional[SigAuthType] = None """ Signature /Prop_AuthType to use. This should be one of :attr:`~.fields.SigAuthType.PIN` or :attr:`~.fields.SigAuthType.PASSWORD` or :attr:`~.fields.SigAuthType.FINGERPRINT` If not specified, this property won't be set on the signature dictionary. """ certify: bool = False """ Sign with an author (certification) signature, as opposed to an approval signature. A document can contain at most one such signature, and it must be the first one. """ # TODO Does this restriction also apply to prior document timestamps? subfilter: Optional[SigSeedSubFilter] = None """ Signature subfilter to use. This should be one of :attr:`~.fields.SigSeedSubFilter.ADOBE_PKCS7_DETACHED` or :attr:`~.fields.SigSeedSubFilter.PADES`. If not specified, the value may be inferred from the signature field's seed value dictionary. Failing that, :attr:`~.fields.SigSeedSubFilter.ADOBE_PKCS7_DETACHED` is used as the default value. """ embed_validation_info: bool = False """ Flag indicating whether validation info (OCSP responses and/or CRLs) should be embedded or not. This is necessary to be able to validate signatures long after they have been made. This flag requires :attr:`validation_context` to be set. The precise manner in which the validation info is embedded depends on the (effective) value of :attr:`subfilter`: * With :attr:`~.fields.SigSeedSubFilter.ADOBE_PKCS7_DETACHED`, the validation information will be embedded inside the CMS object containing the signature. * With :attr:`~.fields.SigSeedSubFilter.PADES`, the validation information will be embedded into the document security store (DSS). """ use_pades_lta: bool = False """ If ``True``, the signer will append an additional document timestamp after writing the signature's validation information to the document security store (DSS). This flag is only meaningful if :attr:`subfilter` is :attr:`~.fields.SigSeedSubFilter.PADES`. The PAdES B-LTA profile solves the long-term validation problem by adding a timestamp chain to the document after the regular signatures, which is updated with new timestamps at regular intervals. This provides an audit trail that ensures the long-term integrity of the validation information in the DSS, since OCSP responses and CRLs also have a finite lifetime. See also :meth:`.PdfTimeStamper.update_archival_timestamp_chain`. """ timestamp_field_name: Optional[str] = None """ Name of the timestamp field created when :attr:`use_pades_lta` is ``True``. If not specified, a unique name will be generated using :mod:`uuid`. """ validation_context: Optional[ValidationContext] = None """ The validation context to use when validating signatures. If provided, the signer's certificate and any timestamp certificates will be validated before signing. This parameter is mandatory when :attr:`embed_validation_info` is ``True``. """ docmdp_permissions: MDPPerm = MDPPerm.FILL_FORMS """ Indicates the document modification policy that will be in force after this signature is created. Only relevant for certification signatures or signatures that apply locking. .. warning:: For non-certification signatures, this is only explicitly allowed since PDF 2.0 (ISO 32000-2), so older software may not respect this setting on approval signatures. """ signer_key_usage: Set[str] = field( default_factory=lambda: constants.DEFAULT_SIGNER_KEY_USAGE ) """ Key usage extensions required for the signer's certificate. Defaults to ``non_repudiation`` only, but sometimes ``digital_signature`` or a combination of both may be more appropriate. See :class:`x509.KeyUsage` for a complete list. Only relevant if a validation context is also provided. """ cades_signed_attr_spec: Optional[CAdESSignedAttrSpec] = None """ .. versionadded:: 0.5.0 Specification for CAdES-specific attributes. """ dss_settings: DSSContentSettings = DSSContentSettings() """ .. versionadded:: 0.8.0 DSS output settings. See :class:`.DSSContentSettings`. """ tight_size_estimates: bool = False """ .. versionadded:: 0.8.0 When estimating the size of a signature container, do not add safety margins. .. note:: This should be OK if the entire CMS object is produced by pyHanko, and the signing scheme produces signatures of a fixed size. However, if the signature container includes unsigned attributes such as signature timestamps, the size of the signature is never entirely predictable. """ ac_validation_context: Optional[ValidationContext] = None """ .. versionadded:: 0.11.0 Validation context for attribute certificates """ def _ensure_esic_ext(pdf_writer: BasePdfFileWriter): """ Helper function to ensure that the output PDF is at least PDF 1.7, and that the relevant ESIC extension for PAdES is enabled if the version lower than 2.0. """ pdf_writer.ensure_output_version(version=(1, 7)) if pdf_writer.output_version < (2, 0): pdf_writer.register_extension(constants.ESIC_EXTENSION_1) def add_mac_to_external_cms( pdf_out: BasePdfFileWriter, md_algorithm: str, signed_data: cms.SignedData, document_digest: bytes, dry_run=False, ): """ Utility function to add a MAC token to an externally generated CMS payload. The payload is modified in-place. :param pdf_out: Writer to contain output. :param md_algorithm: The message digest algorithm to use for the MAC. :param document_digest: The document digest that should go into the MAC token. :param signed_data: A CMS signed data value. :param dry_run: Whether this is considered a dry run for size estimation purposes. """ mac_handler = pdf_out._init_mac_handler(md_algorithm) si = signed_data['signer_infos'][0] signature_bytes = si['signature'].native md = hashes.Hash(get_pyca_cryptography_hash(md_algorithm)) md.update(signature_bytes) sig_digest = md.finalize() if dry_run: # sig_digest always has the right length document_digest = sig_digest mac_token = mac_handler.build_pdfmac_token( document_digest=document_digest, signature_digest=sig_digest, dry_run=dry_run, ) ua = si['unsigned_attrs'] if ua is VOID: ua = cms.CMSAttributes([]) ua.append(simple_cms_attribute('pdf_mac_data', mac_token)) si['unsigned_attrs'] = ua def _ensure_iso32001_ext(pdf_writer: BasePdfFileWriter): pdf_writer.ensure_output_version(version=(2, 0)) pdf_writer.register_extension(constants.ISO32001) def _ensure_iso32002_ext(pdf_writer: BasePdfFileWriter): pdf_writer.ensure_output_version(version=(2, 0)) pdf_writer.register_extension(constants.ISO32002) def _is_iso32002_curve(pubkey: keys.PublicKeyInfo): kind, curve_id = pubkey.curve return kind == 'named' and curve_id in constants.ISO32002_CURVE_NAMES class PdfTimeStamper: """ Class to encapsulate the process of appending document timestamps to PDF files. """ def __init__( self, timestamper: TimeStamper, field_name: Optional[str] = None, invis_settings: InvisSigSettings = InvisSigSettings(), readable_field_name: str = "Timestamp", ): self.default_timestamper = timestamper self._field_name = field_name self._readable_field_name = readable_field_name self._invis_settings = invis_settings @property def field_name(self) -> str: """ Retrieve or generate the field name for the signature field to contain the document timestamp. :return: The field name, as a (Python) string. """ return self._field_name or ('Timestamp-' + str(uuid.uuid4())) # TODO I'm not entirely sure that allowing validation_paths to be cached # is wise. In principle, the TSA could issue their next timestamp with a # different certificate (e.g. due to load balancing), which would require # validation regardless. def timestamp_pdf( self, pdf_out: IncrementalPdfFileWriter, md_algorithm, validation_context=None, bytes_reserved=None, validation_paths=None, timestamper: Optional[TimeStamper] = None, *, in_place=False, output=None, dss_settings: TimestampDSSContentSettings = TimestampDSSContentSettings(), chunk_size=misc.DEFAULT_CHUNK_SIZE, tight_size_estimates: bool = False, ): """ .. versionchanged:: 0.9.0 Wrapper around :meth:`async_timestamp_pdf`. Timestamp the contents of ``pdf_out``. Note that ``pdf_out`` should not be written to after this operation. :param pdf_out: An :class:`.IncrementalPdfFileWriter`. :param md_algorithm: The hash algorithm to use when computing message digests. :param validation_context: The :class:`.pyhanko_certvalidator.ValidationContext` against which the TSA response should be validated. This validation context will also be used to update the DSS. :param bytes_reserved: Bytes to reserve for the CMS object in the PDF file. If not specified, make an estimate based on a dummy signature. .. warning:: Since the CMS object is written to the output file as a hexadecimal string, you should request **twice** the (estimated) number of bytes in the DER-encoded version of the CMS object. :param validation_paths: If the validation path(s) for the TSA's certificate are already known, you can pass them using this parameter to avoid having to run the validation logic again. :param timestamper: Override the default :class:`.TimeStamper` associated with this :class:`.PdfTimeStamper`. :param output: Write the output to the specified output stream. If ``None``, write to a new :class:`.BytesIO` object. Default is ``None``. :param in_place: Sign the original input stream in-place. This parameter overrides ``output``. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. :param dss_settings: DSS output settings. See :class:`.TimestampDSSContentSettings`. :param tight_size_estimates: When estimating the size of a document timestamp container, do not add safety margins. .. note:: External TSAs cannot be relied upon to always produce the exact same output length, which makes this option risky to use. :return: The output stream containing the signed output. """ result = asyncio.run( self.async_timestamp_pdf( pdf_out, md_algorithm, validation_context=validation_context, bytes_reserved=bytes_reserved, validation_paths=validation_paths, timestamper=timestamper, in_place=in_place, output=output, chunk_size=chunk_size, dss_settings=dss_settings, tight_size_estimates=tight_size_estimates, ) ) return result async def async_timestamp_pdf( self, pdf_out: IncrementalPdfFileWriter, md_algorithm, validation_context=None, bytes_reserved=None, validation_paths=None, timestamper: Optional[TimeStamper] = None, *, in_place=False, output=None, dss_settings: TimestampDSSContentSettings = TimestampDSSContentSettings(), chunk_size=misc.DEFAULT_CHUNK_SIZE, tight_size_estimates: bool = False, embed_roots: bool = True, ): """ .. versionadded:: 0.9.0 Timestamp the contents of ``pdf_out``. Note that ``pdf_out`` should not be written to after this operation. :param pdf_out: An :class:`.IncrementalPdfFileWriter`. :param md_algorithm: The hash algorithm to use when computing message digests. :param validation_context: The :class:`.pyhanko_certvalidator.ValidationContext` against which the TSA response should be validated. This validation context will also be used to update the DSS. :param bytes_reserved: Bytes to reserve for the CMS object in the PDF file. If not specified, make an estimate based on a dummy signature. .. warning:: Since the CMS object is written to the output file as a hexadecimal string, you should request **twice** the (estimated) number of bytes in the DER-encoded version of the CMS object. :param validation_paths: If the validation path(s) for the TSA's certificate are already known, you can pass them using this parameter to avoid having to run the validation logic again. :param timestamper: Override the default :class:`.TimeStamper` associated with this :class:`.PdfTimeStamper`. :param output: Write the output to the specified output stream. If ``None``, write to a new :class:`.BytesIO` object. Default is ``None``. :param in_place: Sign the original input stream in-place. This parameter overrides ``output``. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. :param dss_settings: DSS output settings. See :class:`.TimestampDSSContentSettings`. :param tight_size_estimates: When estimating the size of a document timestamp container, do not add safety margins. .. note:: External TSAs cannot be relied upon to always produce the exact same output length, which makes this option risky to use. :param embed_roots: Option that controls whether the root certificate of each validation path should be embedded into the DSS. The default is ``True``. .. note:: Trust roots are configured by the validator, so embedding them typically does nothing in a typical validation process. Therefore they can be safely omitted in most cases. Nonetheless, embedding the roots can be useful for documentation purposes. :return: The output stream containing the signed output. """ _ensure_esic_ext(pdf_out) need_mac = ( pdf_out.security_handler is not None and pdf_out.security_handler.pdf_mac_enabled ) from pyhanko.sign import validation timestamper = timestamper or self.default_timestamper if validation_context is not None: paths_coro = timestamper.validation_paths(validation_context) if validation_paths is None: validation_paths = [] async for path in paths_coro: validation_paths.append(path) dss_settings.assert_viable() if dss_settings.update_before_ts: # NOTE: we have to disable VRI in this scenario validation.DocumentSecurityStore.supply_dss_in_writer( pdf_out, sig_contents=None, paths=validation_paths, validation_context=validation_context, embed_roots=embed_roots, ) field_name = self.field_name if bytes_reserved is None: test_tst = await timestamper.async_dummy_response(md_algorithm) if need_mac: add_mac_to_external_cms( pdf_out, md_algorithm=md_algorithm, signed_data=test_tst['content'], document_digest=b"", dry_run=True, ) test_len = len(test_tst.dump()) * 2 if tight_size_estimates: bytes_reserved = test_len else: # see sign_pdf comments bytes_reserved = test_len + 2 * (test_len // 4) timestamp_obj = DocumentTimestamp(bytes_reserved=bytes_reserved) field_spec = SigFieldSpec( sig_field_name=field_name, invis_sig_settings=self._invis_settings, readable_field_name=self._readable_field_name, ) cms_writer = PdfCMSEmbedder(new_field_spec=field_spec).write_cms( field_name=field_name, writer=pdf_out, # for LTA, requiring existing_fields_only doesn't make sense # since we should in principle be able to add document timestamps # ad infinitum. existing_fields_only=False, ) next(cms_writer) sig_obj_ref = cms_writer.send( SigObjSetup(sig_placeholder=timestamp_obj) ) if need_mac: _register_mac_on_sig(pdf_out, sig_obj_ref) sig_io = SigIOSetup( md_algorithm=md_algorithm, in_place=in_place, output=output, chunk_size=chunk_size, ) prep_digest: PreparedByteRangeDigest prep_digest, res_output = cms_writer.send(sig_io) timestamp_cms = await timestamper.async_timestamp( prep_digest.document_digest, md_algorithm ) if need_mac: add_mac_to_external_cms( pdf_out, md_algorithm=md_algorithm, signed_data=timestamp_cms['content'], document_digest=prep_digest.document_digest, ) sig_contents = cms_writer.send(timestamp_cms) # update the DSS if necessary if validation_context is not None and not dss_settings.update_before_ts: if not dss_settings.include_vri: sig_contents = None # FIXME in this case, the ser/deser step is unnecessary # and inefficient; should probably rewrite # using supply_dss_in_writer credential_ser: Optional[SerialisedCredential] = None if pdf_out.security_handler: credential = pdf_out.security_handler.extract_credential() if credential: credential_ser = credential.serialise() strict = True if isinstance(pdf_out, IncrementalPdfFileWriter): strict = pdf_out.prev.strict validation.DocumentSecurityStore.add_dss( output_stream=res_output, sig_contents=sig_contents, paths=validation_paths, validation_context=validation_context, force_write=not dss_settings.skip_if_unneeded, embed_roots=embed_roots, file_credential=credential_ser, strict=strict, ) return misc.finalise_output(output, res_output) def update_archival_timestamp_chain( self, reader: PdfFileReader, validation_context, in_place=True, output=None, chunk_size=misc.DEFAULT_CHUNK_SIZE, default_md_algorithm=constants.DEFAULT_MD, ): """ .. versionchanged:: 0.9.0 Wrapper around :meth:`async_update_archival_timestamp_chain`. Validate the last timestamp in the timestamp chain on a PDF file, and write an updated version to an output stream. :param reader: A :class:`PdfReader` encapsulating the input file. :param validation_context: :class:`.pyhanko_certvalidator.ValidationContext` object to validate the last timestamp. :param output: Write the output to the specified output stream. If ``None``, write to a new :class:`.BytesIO` object. Default is ``None``. :param in_place: Sign the original input stream in-place. This parameter overrides ``output``. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. :param default_md_algorithm: Message digest to use if there are no preceding timestamps in the file. :return: The output stream containing the signed output. """ coro = self.async_update_archival_timestamp_chain( reader=reader, validation_context=validation_context, in_place=in_place, output=output, chunk_size=chunk_size, default_md_algorithm=default_md_algorithm, ) return asyncio.run(coro) async def async_update_archival_timestamp_chain( self, reader: PdfFileReader, validation_context, in_place=True, output=None, chunk_size=misc.DEFAULT_CHUNK_SIZE, default_md_algorithm=constants.DEFAULT_MD, embed_roots: bool = True, ): """ .. versionadded:: 0.9.0 Validate the last timestamp in the timestamp chain on a PDF file, and write an updated version to an output stream. :param reader: A :class:`PdfReader` encapsulating the input file. :param validation_context: :class:`.pyhanko_certvalidator.ValidationContext` object to validate the last timestamp. :param output: Write the output to the specified output stream. If ``None``, write to a new :class:`.BytesIO` object. Default is ``None``. :param in_place: Sign the original input stream in-place. This parameter overrides ``output``. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. :param default_md_algorithm: Message digest to use if there are no preceding timestamps in the file. :param embed_roots: Option that controls whether the root certificate of each validation path should be embedded into the DSS. The default is ``True``. .. note:: Trust roots are configured by the validator, so embedding them typically does nothing in a typical validation process. Therefore they can be safely omitted in most cases. Nonetheless, embedding the roots can be useful for documentation purposes. :return: The output stream containing the signed output. """ # TODO expose DSS fine-tuning here as well # In principle, we only have to validate that the last timestamp token # in the current chain is valid. # TODO: add an option to validate the entire timestamp chain # plus all signatures from .. import validation from ..validation.status import TimestampSignatureStatus timestamps = validation.get_timestamp_chain(reader) try: last_timestamp = next(timestamps) except StopIteration: logger.warning( "Document does not have any document timestamps yet. " "This may cause unexpected results." ) last_timestamp = None # Validate the previous timestamp if present tst_status: Optional[TimestampSignatureStatus] = None if last_timestamp is None: md_algorithm = default_md_algorithm else: expected_imprint = last_timestamp.compute_digest() tst_token = last_timestamp.signed_data # run validation logic last_tst_status = await validation.ltv.establish_timestamp_trust( tst_token, validation_context, expected_imprint ) md_algorithm = last_tst_status.md_algorithm tst_status = last_tst_status # Prepare output if in_place: res_output = reader.stream else: res_output = misc.prepare_rw_output_stream(output) reader.stream.seek(0) misc.chunked_write(bytearray(chunk_size), reader.stream, res_output) pdf_out = IncrementalPdfFileWriter(res_output) if last_timestamp is not None: assert tst_status is not None # update the DSS validation.DocumentSecurityStore.supply_dss_in_writer( pdf_out, last_timestamp.pkcs7_content, paths=(tst_status.validation_path,), validation_context=validation_context, embed_roots=embed_roots, ) # append a new timestamp await self.async_timestamp_pdf( pdf_out, md_algorithm, validation_context, in_place=True, embed_roots=embed_roots, ) return misc.finalise_output(output, res_output) def _signatures_exist(handler): try: next(enumerate_sig_fields(handler, filled_status=True)) return True except StopIteration: return False class PdfSigner: """ .. versionchanged: 0.7.0 This class is no longer a subclass of :class:`.PdfTimeStamper`. Class to handle PDF signatures in general. :param signature_meta: The specification of the signature to add. :param signer: :class:`.Signer` object to use to produce the signature object. :param timestamper: :class:`.TimeStamper` object to use to produce any time stamp tokens that might be required. :param stamp_style: Stamp style specification to determine the visible style of the signature, typically an object of type :class:`.TextStampStyle` or :class:`.QRStampStyle`. Defaults to :const:`constants.DEFAULT_SIGNING_STAMP_STYLE`. :param new_field_spec: If a new field is to be created, this parameter allows the caller to specify the field's properties in the form of a :class:`.SigFieldSpec`. This parameter is only meaningful if ``existing_fields_only`` is ``False``. """ _ignore_sv = False def __init__( self, signature_meta: PdfSignatureMetadata, signer: Signer, *, timestamper: Optional[TimeStamper] = None, stamp_style: Optional[BaseStampStyle] = None, new_field_spec: Optional[SigFieldSpec] = None, ): self.signature_meta = signature_meta if ( new_field_spec is not None and new_field_spec.sig_field_name != signature_meta.field_name ): raise SigningError( "Field names specified in SigFieldSpec and " "PdfSignatureMetadata do not agree." ) self.signer = signer stamp_style = stamp_style or constants.DEFAULT_SIGNING_STAMP_STYLE self.stamp_style: BaseStampStyle = stamp_style self.signer_hash_algo: Optional[str] try: mech = self.signer.get_signature_mechanism_for_digest(None) self.signer_hash_algo = get_cms_hash_algo_for_mechanism(mech) except ValueError: self.signer_hash_algo = None self.new_field_spec = new_field_spec self.default_timestamper = timestamper @property def default_md_for_signer(self) -> Optional[str]: """ Name of the default message digest algorithm for this signer, if there is one. This method will try the :attr:`~.PdfSignatureMetadata.md_algorithm` attribute on the signer's :attr:`signature_meta`, or try to retrieve the digest algorithm associated with the underlying :class:`~pyhanko.sign.signers.pdf_cms.Signer`. :return: The name of the message digest algorithm, or ``None``. """ return self.signature_meta.md_algorithm or self.signer_hash_algo def _enforce_certification_constraints(self, reader: PdfFileReader): # TODO we really should take into account the /DocMDP constraints # of _all_ previous signatures, i.e. also approval signatures with # locking instructions etc. if self.signature_meta.certify and _signatures_exist(reader): raise SigningError( "Certification signatures must be the first signature " "in a given document." ) from pyhanko.sign.validation import read_certification_data cd = read_certification_data(reader) # if there is no author signature, we don't have to do anything if cd is None: return if cd.permission == MDPPerm.NO_CHANGES: raise SigningError("Author signature forbids all changes") def _retrieve_seed_value_spec( self, sig_field ) -> Optional[SigSeedValueSpec]: # for testing & debugging if self._ignore_sv: return None sv_dict = sig_field.get('/SV') if sv_dict is None: return None return SigSeedValueSpec.from_pdf_object(sv_dict) def _select_md_algorithm(self, sv_spec: Optional[SigSeedValueSpec]) -> str: signature_meta = self.signature_meta # priority order for the message digest algorithm # (1) If signature_meta specifies a message digest algorithm, use it # (it has been cleared by the SV dictionary checker already) # (2) Use the first algorithm specified in the seed value dictionary, # if a suggestion is present # (3) fall back to select_suitable_signing_md() if sv_spec is not None and sv_spec.digest_methods: sv_md_algorithm = sv_spec.digest_methods[0] else: sv_md_algorithm = None if self.default_md_for_signer is not None: md_algorithm = self.default_md_for_signer elif sv_md_algorithm is not None: md_algorithm = sv_md_algorithm elif self.signer.signing_cert is not None: md_algorithm = select_suitable_signing_md( self.signer.signing_cert.public_key ) else: raise SigningError( "Could not select a default digest algorithm. Please supply " "a value in the signature settings, or configure the signer " "with an explicit signature mechanism that includes a digest " "algorithm specification." ) algorithm_policy = ( signature_meta.validation_context.algorithm_policy if signature_meta.validation_context is not None else None ) if algorithm_policy is not None: now = datetime.now() md_algo_obj = algos.DigestAlgorithm({'algorithm': md_algorithm}) if not algorithm_policy.digest_algorithm_allowed(md_algo_obj, now): raise SigningError( f"The hash algorithm {md_algorithm} is not allowed in the " f"specified validation context (usage policy of type " f"{algorithm_policy.__class__.__name__})." ) return md_algorithm def register_extensions( self, pdf_out: BasePdfFileWriter, *, md_algorithm: str ): if self.signature_meta.subfilter == SigSeedSubFilter.PADES: _ensure_esic_ext(pdf_out) if ( pdf_out.security_handler is not None and pdf_out.security_handler.pdf_mac_enabled ): pdf_out.register_extension(pdfmac.ISO32004) try: sig_mech = self.signer.get_signature_mechanism_for_digest( md_algorithm ) sig_algo = sig_mech.signature_algo except (SigningError, ValueError) as e: logger.debug( f"Failed to introspect signature mechanism: {str(e)}. " f"Will forgo algorithm-based automatic extension registration.", ) return if sig_algo == 'ed25519': _ensure_iso32002_ext(pdf_out) elif sig_algo == 'ed448': _ensure_iso32001_ext(pdf_out) _ensure_iso32002_ext(pdf_out) else: if md_algorithm.startswith('sha3') or md_algorithm == 'shake256': _ensure_iso32001_ext(pdf_out) cert = self.signer.signing_cert if ( cert and sig_algo == 'ecdsa' and _is_iso32002_curve(cert.public_key) ): _ensure_iso32002_ext(pdf_out) def init_signing_session( self, pdf_out: BasePdfFileWriter, existing_fields_only=False ) -> 'PdfSigningSession': """ Initialise a signing session with this :class:`.PdfSigner` for a specified PDF file writer. This step in the signing process handles all field-level operations prior to signing: it creates the target form field if necessary, and makes sure the seed value dictionary gets processed. See also :meth:`digest_doc_for_signing` and :meth:`sign_pdf`. :param pdf_out: The writer containing the PDF file to be signed. :param existing_fields_only: If ``True``, never create a new empty signature field to contain the signature. If ``False``, a new field may be created if no field matching :attr:`~.PdfSignatureMetadata.field_name` exists. :return: A :class:`.PdfSigningSession` object modelling the signing session in its post-setup stage. """ if isinstance(pdf_out, IncrementalPdfFileWriter): # ensure we're not signing a hybrid reference doc prev = pdf_out.prev if prev.strict and prev.xrefs.hybrid_xrefs_present: raise SigningError( "Attempting to sign document with hybrid cross-reference " "sections while hybrid xrefs are disabled" ) timestamper = self.default_timestamper signature_meta: PdfSignatureMetadata = self.signature_meta cms_writer = PdfCMSEmbedder( new_field_spec=self.new_field_spec ).write_cms( field_name=signature_meta.field_name, writer=pdf_out, existing_fields_only=existing_fields_only, ) # let the CMS writer put in a field for us, if necessary sig_field_ref = next(cms_writer) sig_field = sig_field_ref.get_object() # Fetch seed values (if present) to prepare for signing sv_spec = self._retrieve_seed_value_spec(sig_field) # Check DocMDP settings to see if we're allowed to add a signature if isinstance(pdf_out, IncrementalPdfFileWriter): self._enforce_certification_constraints(pdf_out.prev) md_algorithm = self._select_md_algorithm(sv_spec) self.register_extensions(pdf_out, md_algorithm=md_algorithm) # check if a timestamp is required without a timestamper being # available already, and furnish one if necessary if ( sv_spec is not None and sv_spec.timestamp_required and timestamper is None ): timestamper = sv_spec.build_timestamper() # subfilter: try signature_meta and SV dict, fall back # to /adbe.pkcs7.detached by default subfilter = signature_meta.subfilter if subfilter is None: if sv_spec is not None and sv_spec.subfilters: subfilter = sv_spec.subfilters[0] else: subfilter = SigSeedSubFilter.ADOBE_PKCS7_DETACHED session = PdfSigningSession( self, pdf_out, cms_writer, sig_field, md_algorithm, timestamper, subfilter, sv_spec=sv_spec, ) return session def digest_doc_for_signing( self, pdf_out: BasePdfFileWriter, existing_fields_only=False, bytes_reserved=None, *, appearance_text_params=None, in_place=False, output=None, chunk_size=misc.DEFAULT_CHUNK_SIZE, ) -> Tuple[PreparedByteRangeDigest, 'PdfTBSDocument', IO]: """ .. deprecated:: 0.9.0 Use :meth:`async_digest_doc_for_signing` instead. Set up all stages of the signing process up to and including the point where the signature placeholder is allocated, and the document's ``/ByteRange`` digest is computed. See :meth:`sign_pdf` for a less granular, more high-level approach. .. note:: This method is useful in remote signing scenarios, where you might want to free up resources while waiting for the remote signer to respond. The :class:`.PreparedByteRangeDigest` object returned allows you to keep track of the required state to fill the signature container at some later point in time. :param pdf_out: A PDF file writer (usually an :class:`.IncrementalPdfFileWriter`) containing the data to sign. :param existing_fields_only: If ``True``, never create a new empty signature field to contain the signature. If ``False``, a new field may be created if no field matching :attr:`~.PdfSignatureMetadata.field_name` exists. :param bytes_reserved: Bytes to reserve for the CMS object in the PDF file. If not specified, make an estimate based on a dummy signature. .. warning:: Since the CMS object is written to the output file as a hexadecimal string, you should request **twice** the (estimated) number of bytes in the DER-encoded version of the CMS object. :param appearance_text_params: Dictionary with text parameters that will be passed to the signature appearance constructor (if applicable). :param output: Write the output to the specified output stream. If ``None``, write to a new :class:`.BytesIO` object. Default is ``None``. :param in_place: Sign the original input stream in-place. This parameter overrides ``output``. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. :return: A tuple containing a :class:`.PreparedByteRangeDigest` object, a :class:`.PdfTBSDocument` object and an output handle to which the document in its current state has been written. """ warnings.warn( "'digest_doc_for_signing' is deprecated, use " "'async_digest_doc_for_signing' instead", DeprecationWarning, ) result = asyncio.run( self.async_digest_doc_for_signing( pdf_out, existing_fields_only=existing_fields_only, bytes_reserved=bytes_reserved, appearance_text_params=appearance_text_params, in_place=in_place, output=output, chunk_size=chunk_size, ) ) return result async def async_digest_doc_for_signing( self, pdf_out: BasePdfFileWriter, existing_fields_only=False, bytes_reserved=None, *, appearance_text_params=None, in_place=False, output=None, chunk_size=misc.DEFAULT_CHUNK_SIZE, ) -> Tuple[PreparedByteRangeDigest, 'PdfTBSDocument', IO]: """ .. versionadded:: 0.9.0 Set up all stages of the signing process up to and including the point where the signature placeholder is allocated, and the document's ``/ByteRange`` digest is computed. See :meth:`sign_pdf` for a less granular, more high-level approach. .. note:: This method is useful in remote signing scenarios, where you might want to free up resources while waiting for the remote signer to respond. The :class:`.PreparedByteRangeDigest` object returned allows you to keep track of the required state to fill the signature container at some later point in time. :param pdf_out: A PDF file writer (usually an :class:`.IncrementalPdfFileWriter`) containing the data to sign. :param existing_fields_only: If ``True``, never create a new empty signature field to contain the signature. If ``False``, a new field may be created if no field matching :attr:`~.PdfSignatureMetadata.field_name` exists. :param bytes_reserved: Bytes to reserve for the CMS object in the PDF file. If not specified, make an estimate based on a dummy signature. .. warning:: Since the CMS object is written to the output file as a hexadecimal string, you should request **twice** the (estimated) number of bytes in the DER-encoded version of the CMS object. :param appearance_text_params: Dictionary with text parameters that will be passed to the signature appearance constructor (if applicable). :param output: Write the output to the specified output stream. If ``None``, write to a new :class:`.BytesIO` object. Default is ``None``. :param in_place: Sign the original input stream in-place. This parameter overrides ``output``. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. :return: A tuple containing a :class:`.PreparedByteRangeDigest` object, a :class:`.PdfTBSDocument` object and an output handle to which the document in its current state has been written. """ signing_session = self.init_signing_session( pdf_out, existing_fields_only=existing_fields_only, ) validation_info = await signing_session.perform_presign_validation( pdf_out ) if bytes_reserved is None: estimation = signing_session.estimate_signature_container_size( validation_info=validation_info, tight=self.signature_meta.tight_size_estimates, ) bytes_reserved = await estimation tbs_document = signing_session.prepare_tbs_document( validation_info=validation_info, bytes_reserved=bytes_reserved, appearance_text_params=appearance_text_params, ) prepared_br_digest, res_output = tbs_document.digest_tbs_document( in_place=in_place, chunk_size=chunk_size, output=output ) return ( prepared_br_digest, tbs_document, misc.finalise_output(output, res_output), ) def sign_pdf( self, pdf_out: BasePdfFileWriter, existing_fields_only=False, bytes_reserved=None, *, appearance_text_params=None, in_place=False, output=None, chunk_size=misc.DEFAULT_CHUNK_SIZE, ): """ .. versionchanged:: 0.9.0 Wrapper around :meth:`async_sign_pdf`. Sign a PDF file using the provided output writer. :param pdf_out: A PDF file writer (usually an :class:`.IncrementalPdfFileWriter`) containing the data to sign. :param existing_fields_only: If ``True``, never create a new empty signature field to contain the signature. If ``False``, a new field may be created if no field matching :attr:`~.PdfSignatureMetadata.field_name` exists. :param bytes_reserved: Bytes to reserve for the CMS object in the PDF file. If not specified, make an estimate based on a dummy signature. :param appearance_text_params: Dictionary with text parameters that will be passed to the signature appearance constructor (if applicable). :param output: Write the output to the specified output stream. If ``None``, write to a new :class:`.BytesIO` object. Default is ``None``. :param in_place: Sign the original input stream in-place. This parameter overrides ``output``. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. :return: The output stream containing the signed data. """ result = asyncio.run( self.async_sign_pdf( pdf_out, existing_fields_only=existing_fields_only, bytes_reserved=bytes_reserved, appearance_text_params=appearance_text_params, in_place=in_place, output=output, chunk_size=chunk_size, ) ) return result async def async_sign_pdf( self, pdf_out: BasePdfFileWriter, existing_fields_only=False, bytes_reserved=None, *, appearance_text_params=None, in_place=False, output=None, chunk_size=misc.DEFAULT_CHUNK_SIZE, ): """ .. versionadded:: 0.9.0 Sign a PDF file using the provided output writer. :param pdf_out: A PDF file writer (usually an :class:`.IncrementalPdfFileWriter`) containing the data to sign. :param existing_fields_only: If ``True``, never create a new empty signature field to contain the signature. If ``False``, a new field may be created if no field matching :attr:`~.PdfSignatureMetadata.field_name` exists. :param bytes_reserved: Bytes to reserve for the CMS object in the PDF file. If not specified, make an estimate based on a dummy signature. :param appearance_text_params: Dictionary with text parameters that will be passed to the signature appearance constructor (if applicable). :param output: Write the output to the specified output stream. If ``None``, write to a new :class:`.BytesIO` object. Default is ``None``. :param in_place: Sign the original input stream in-place. This parameter overrides ``output``. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. :return: The output stream containing the signed data. """ signing_session = self.init_signing_session( pdf_out, existing_fields_only=existing_fields_only, ) validation_info = await signing_session.perform_presign_validation( pdf_out ) if bytes_reserved is None: estimation = signing_session.estimate_signature_container_size( validation_info, tight=self.signature_meta.tight_size_estimates ) bytes_reserved = await estimation tbs_document = signing_session.prepare_tbs_document( validation_info=validation_info, bytes_reserved=bytes_reserved, appearance_text_params=appearance_text_params, ) prepared_br_digest, res_output = tbs_document.digest_tbs_document( in_place=in_place, chunk_size=chunk_size, output=output ) post_signing_doc = await tbs_document.perform_signature( document_digest=prepared_br_digest.document_digest, pdf_cms_signed_attrs=PdfCMSSignedAttributes( signing_time=signing_session.system_time, adobe_revinfo_attr=( None if validation_info is None else validation_info.adobe_revinfo_attr ), cades_signed_attrs=self.signature_meta.cades_signed_attr_spec, ), ) await post_signing_doc.post_signature_processing( res_output, chunk_size=chunk_size ) # we put the finalisation step after the DSS manipulations, since # otherwise we'd also run into issues with non-seekable output buffers return misc.finalise_output(output, res_output) @dataclass(frozen=True) class PreSignValidationStatus: """ .. versionadded:: 0.7.0 Container for validation data collected prior to creating a signature, e.g. for later inclusion in a document's DSS, or as a signed attribute on the signature. """ signer_path: ValidationPath """ Validation path for the signer's certificate. """ validation_paths: List[ValidationPath] """ List of other relevant validation paths. """ ts_validation_paths: Optional[List[ValidationPath]] = None """ List of validation paths relevant for embedded timestamps. """ adobe_revinfo_attr: Optional[asn1_pdf.RevocationInfoArchival] = None """ Preformatted revocation info attribute to include, if requested by the settings. """ ocsps_to_embed: List[ocsp.OCSPResponse] = field(default_factory=list) """ List of OCSP responses collected so far. """ crls_to_embed: List[crl.CertificateList] = field(default_factory=list) """ List of CRLS collected so far. """ ac_validation_paths: List[ValidationPath] = field(default_factory=list) """ List of validation paths relevant for embedded attribute certificates. """ def _register_mac_on_sig( pdf_out: BasePdfFileWriter, sig_obj_ref: generic.IndirectObject ): pdf_out.set_custom_trailer_entry( pdf_name('/AuthCode'), generic.DictionaryObject( { pdf_name('/MACLocation'): pdf_name('/AttachedToSig'), pdf_name('/SigObjRef'): sig_obj_ref, } ), ) class PdfSigningSession: """ .. versionadded:: 0.7.0 Class modelling a PDF signing session in its initial state. The ``__init__`` method is internal API, get an instance using :meth:`.PdfSigner.init_signing_session`. """ def __init__( self, pdf_signer: PdfSigner, pdf_out: BasePdfFileWriter, cms_writer, sig_field, md_algorithm: str, timestamper: Optional[TimeStamper], subfilter: SigSeedSubFilter, system_time: Optional[datetime] = None, sv_spec: Optional[SigSeedValueSpec] = None, ): self.pdf_signer = pdf_signer self.pdf_out = pdf_out self.sig_field = sig_field self.cms_writer = cms_writer self.md_algorithm = md_algorithm self.timestamper = timestamper self.subfilter = subfilter self.use_pades = subfilter == SigSeedSubFilter.PADES self.system_time = system_time or datetime.now( tz=tzlocal.get_localzone() ) self.sv_spec = sv_spec async def perform_presign_validation( self, pdf_out: Optional[BasePdfFileWriter] = None ) -> Optional[PreSignValidationStatus]: """ Perform certificate validation checks for the signer's certificate, including any necessary revocation checks. This function will also attempt to validate & collect revocation information for the relevant TSA (by requesting a dummy timestamp). :param pdf_out: Current PDF writer. Technically optional; only used to look for the end of the timestamp chain in the previous revision when producing a PAdES-LTA signature in a document that is already signed (to ensure that the timestamp chain is uninterrupted). :return: A :class:`PreSignValidationStatus` object, or ``None`` if there is no validation context available. """ pdf_signer = self.pdf_signer validation_paths = [] signature_meta = pdf_signer.signature_meta validation_context = signature_meta.validation_context if signature_meta.embed_validation_info: if self.pdf_signer.signer.signing_cert is None: raise SigningError( "A signer's certificate must be provided if " "validation/revocation info is to be embedded into the " "signature." ) elif validation_context is None: raise SigningError( "A validation context must be provided if " "validation/revocation info is to be embedded into the " "signature." ) elif not validation_context.fetching_allowed: logger.warning( "Validation/revocation info will be embedded, but " "fetching is not allowed. This may give rise to unexpected " "results." ) validation_context = signature_meta.validation_context # if there's no validation context, bail early if validation_context is None: return None signer_path = await self._perform_presign_signer_validation( validation_context, signature_meta.signer_key_usage ) validation_paths.append(signer_path) # If LTA: # if the original document already included a document timestamp, # we need to collect revocation information for it, to preserve # the integrity of the timestamp chain if signature_meta.use_pades_lta and isinstance( pdf_out, IncrementalPdfFileWriter ): prev_tsa_path = await self._perform_prev_ts_validation( validation_context, pdf_out.prev ) if prev_tsa_path is not None: validation_paths.append(prev_tsa_path) timestamper = self.timestamper # Finally, fetch validation information for the TSA that we're going to # use for our own TS if timestamper is not None: async_ts_paths = timestamper.validation_paths(validation_context) ts_paths = [] async for ts_path in async_ts_paths: validation_paths.append(ts_path) ts_paths.append(ts_path) else: ts_paths = None aa_paths: List[ValidationPath] # fetch attribute certificate validation paths if signature_meta.ac_validation_context is not None: async_aa_paths = self._perform_presign_ac_validation( signature_meta.ac_validation_context ) aa_paths = [] async for aa_path in async_aa_paths: validation_paths.append(aa_path) aa_paths.append(aa_path) else: aa_paths = [] # do we need adobe-style revocation info? if signature_meta.embed_validation_info and not self.use_pades: assert validation_context is not None # checked earlier revinfo = Signer.format_revinfo( ocsp_responses=validation_context.ocsps, crls=validation_context.crls, ) else: # PAdES prescribes another mechanism for embedding revocation info revinfo = None return PreSignValidationStatus( validation_paths=validation_paths, signer_path=signer_path, ts_validation_paths=ts_paths, adobe_revinfo_attr=revinfo, ocsps_to_embed=validation_context.ocsps, crls_to_embed=validation_context.crls, ac_validation_paths=aa_paths, ) async def _perform_presign_ac_validation( self, validation_context: ValidationContext ): signer = self.pdf_signer.signer attr_certs = list(signer.attribute_certs) cades_attr_spec = self.pdf_signer.signature_meta.cades_signed_attr_spec # also make sure to pull in the validation chains for all attribute # certificates included in the signer-attributes-v2 attr, if there is # one. if ( cades_attr_spec is not None and cades_attr_spec.signer_attributes is not None ): attr_certs.extend(cades_attr_spec.signer_attributes.certified_attrs) ac_jobs = [ async_validate_ac( ac, validation_context, holder_cert=signer.signing_cert ) for ac in attr_certs ] for ac_job in asyncio.as_completed(ac_jobs): result: ACValidationResult = await ac_job yield result.aa_path async def _perform_presign_signer_validation( self, validation_context, key_usage ): signer = self.pdf_signer.signer # validate cert # (this also keeps track of any validation data automagically) validator = CertificateValidator( signer.signing_cert, intermediate_certs=signer.cert_registry, validation_context=validation_context, ) try: signer_cert_validation_path = await validator.async_validate_usage( key_usage ) except (PathBuildingError, PathValidationError) as e: raise SigningError( "The signer's certificate could not be validated" ) from e return signer_cert_validation_path async def _perform_prev_ts_validation( self, validation_context, prev_reader ): signer = self.pdf_signer.signer from pyhanko.sign.validation import get_timestamp_chain # try to grab the most recent document timestamp last_ts = None try: last_ts = next(get_timestamp_chain(prev_reader)) except StopIteration: pass last_ts_validation_path = None if last_ts is not None: ts_validator = CertificateValidator( last_ts.signer_cert, intermediate_certs=signer.cert_registry, validation_context=validation_context, ) try: validate_coro = ts_validator.async_validate_usage( set(), extended_key_usage={"time_stamping"} ) last_ts_validation_path = await validate_coro except (PathBuildingError, PathValidationError) as e: raise SigningError( "Requested a PAdES-LTA signature on an existing " "document, but the most recent timestamp " "could not be validated.", e, ) return last_ts_validation_path def _apply_locking_rules(self) -> SigMDPSetup: # TODO allow equivalent functionality to the /Lock dictionary # to be specified in PdfSignatureMetadata # this helper method handles /Lock dictionary and certification # semantics. # The fallback rules are messy and ad-hoc; behaviour is mostly # documented by tests. # read recommendations and/or requirements from the SV dictionary sv_spec = self.sv_spec sig_field = self.sig_field signature_meta = self.pdf_signer.signature_meta if sv_spec is not None: sv_lock_lut: Dict[Any, List[MDPPerm]] = { SeedLockDocument.LOCK: [MDPPerm.NO_CHANGES], SeedLockDocument.DO_NOT_LOCK: [ MDPPerm.FILL_FORMS, MDPPerm.ANNOTATE, ], } sv_lock_values: List[MDPPerm] = sv_lock_lut.get( sv_spec.lock_document, [] ) sv_lock_value_req = sv_lock_values is not None and bool( sv_spec.flags & SigSeedValFlags.LOCK_DOCUMENT ) else: sv_lock_values = [] sv_lock_value_req = False lock = lock_dict = None # init the DocMDP value with what the /LockDocument setting in the SV # dict recommends. If the constraint is mandatory, it might conflict # with the /Lock dictionary, but we'll deal with that later. docmdp_perms = sv_lock_values[0] if sv_lock_values else None try: lock_dict = sig_field['/Lock'] lock = FieldMDPSpec.from_pdf_object(lock_dict) docmdp_value = lock_dict['/P'] docmdp_perms = MDPPerm(docmdp_value) if sv_lock_value_req and docmdp_perms not in sv_lock_values: raise SigningError( "Inconsistency in form field data. " "The field lock dictionary imposes the DocMDP policy " f"'{docmdp_perms}', but the seed value " "dictionary's /LockDocument does not allow that." ) except KeyError: pass except ValueError as e: raise SigningError("Failed to read /Lock dictionary", e) meta_perms = signature_meta.docmdp_permissions meta_certify = signature_meta.certify # only pull meta_perms into the validation if we're trying to make a # cert sig, or there already is some other docmdp_perms value available. # (in other words, if there's no SV dict or /Lock, and we're not # certifying, this will be skipped) if meta_perms is not None and ( meta_certify or docmdp_perms is not None ): if sv_lock_value_req and meta_perms not in sv_lock_values: # in this case, we have to override docmdp_perms = sv_lock_values[0] else: # choose the stricter option if both are available docmdp_perms = ( meta_perms if docmdp_perms is None else (min(docmdp_perms, meta_perms)) ) if docmdp_perms != meta_perms: logger.warning( f"DocMDP policy '{meta_perms}', was requested, " f"but the signature field settings do " f"not allow that. Setting '{docmdp_perms}' instead." ) # if not certifying and docmdp_perms is not None, ensure the # appropriate permission in the Lock dictionary is set if not meta_certify and docmdp_perms is not None: if lock_dict is None: # set a field lock that doesn't do anything sig_field['/Lock'] = lock_dict = generic.DictionaryObject( { pdf_name('/Action'): pdf_name('/Include'), pdf_name('/Fields'): generic.ArrayObject(), } ) lock_dict['/P'] = generic.NumberObject(docmdp_perms.value) return SigMDPSetup( certify=meta_certify, field_lock=lock, docmdp_perms=docmdp_perms, md_algorithm=self.md_algorithm, ) def _enforce_seed_value_constraints( self, validation_path: Optional[ValidationPath] ): sv_spec = self.sv_spec pdf_signer = self.pdf_signer signature_meta = pdf_signer.signature_meta assert sv_spec is not None # Enforce mandatory seed values (except LOCK_DOCUMENT, which is handled # elsewhere) flags: SigSeedValFlags = sv_spec.flags if sv_spec.cert is not None: if pdf_signer.signer.signing_cert is None: raise SigningError( "Cannot verify seed value constraints on the signer's " "certificate since it is not available" ) sv_spec.cert.satisfied_by( pdf_signer.signer.signing_cert, validation_path ) if sv_spec.seed_signature_type is not None: sv_certify = sv_spec.seed_signature_type.certification_signature() if sv_certify != signature_meta.certify: def _type(certify): return 'a certification' if certify else 'an approval' raise SigningError( "The seed value dictionary's /MDP entry specifies that " f"this field should contain {_type(sv_certify)} " f"signature, but {_type(signature_meta.certify)} " "was requested." ) sv_mdp_perm = sv_spec.seed_signature_type.mdp_perm if sv_certify and sv_mdp_perm != signature_meta.docmdp_permissions: raise SigningError( "The seed value dictionary specified that this " "certification signature should use the MDP policy " f"'{sv_mdp_perm}', " f"but '{signature_meta.docmdp_permissions}' was " "requested." ) if not flags: return sv_spec selected_sf = signature_meta.subfilter if ( flags & SigSeedValFlags.SUBFILTER ) and sv_spec.subfilters is not None: # empty array = no supported subfilters if not sv_spec.subfilters: raise NotImplementedError( "The signature encodings mandated by the seed value " "dictionary are not supported." ) # standard mandates that we take the first available subfilter mandated_sf: SigSeedSubFilter = sv_spec.subfilters[0] if selected_sf is not None and mandated_sf != selected_sf: raise SigningError( "The seed value dictionary mandates subfilter '%s', " "but '%s' was requested." % (mandated_sf.value, selected_sf.value) ) # SV dict serves as a source of defaults as well if selected_sf is None and sv_spec.subfilters is not None: selected_sf = sv_spec.subfilters[0] if ( flags & SigSeedValFlags.APPEARANCE_FILTER ) and sv_spec.appearance is not None: raise SigningError( "pyHanko does not define any named appearances, but " "the seed value dictionary requires that the named appearance " f"'{sv_spec.appearance}' be used." ) if ( flags & SigSeedValFlags.ADD_REV_INFO ) and sv_spec.add_rev_info is not None: if sv_spec.add_rev_info != signature_meta.embed_validation_info: raise SigningError( "The seed value dict mandates that revocation info %sbe " "added; adjust PdfSignatureMetadata settings accordingly." % ("" if sv_spec.add_rev_info else "not ") ) if ( sv_spec.add_rev_info and selected_sf != SigSeedSubFilter.ADOBE_PKCS7_DETACHED ): raise SigningError( "The seed value dict mandates that Adobe-style revocation " "info be added; this requires subfilter '%s'" % (SigSeedSubFilter.ADOBE_PKCS7_DETACHED.value) ) if ( flags & SigSeedValFlags.DIGEST_METHOD ) and sv_spec.digest_methods is not None: selected_md = pdf_signer.default_md_for_signer if selected_md is not None: selected_md = selected_md.lower() if selected_md not in sv_spec.digest_methods: raise SigningError( "The selected message digest %s is not allowed by the " "seed value dictionary. Please select one of %s." % (selected_md, ", ".join(sv_spec.digest_methods)) ) if flags & SigSeedValFlags.REASONS: reasons = sv_spec.reasons # standard says that omission of the /Reasons key amounts to # a prohibition in this case must_omit = not reasons or reasons == ["."] reason_given = signature_meta.reason if must_omit and reason_given is not None: raise SigningError( "The seed value dictionary prohibits giving a reason " "for signing." ) if not must_omit and reasons and reason_given not in reasons: raise SigningError( "Reason \"%s\" is not a valid reason for signing, " "please choose one of the following: %s." % ( reason_given, ", ".join("\"%s\"" % s for s in reasons), ) ) async def estimate_signature_container_size( self, validation_info: Optional[PreSignValidationStatus], tight=False ): md_algorithm = self.md_algorithm signature_meta = self.pdf_signer.signature_meta signer = self.pdf_signer.signer if signer.signing_cert is None: raise SigningError( "Automatic signature size estimation is not available without " "a signer's certificate. Space must be allocated manually " "using bytes_reserved=..." ) # estimate bytes_reserved by creating a fake CMS object md_spec = get_pyca_cryptography_hash(md_algorithm) test_md = hashes.Hash(md_spec).finalize() signed_attrs = PdfCMSSignedAttributes( signing_time=self.system_time, adobe_revinfo_attr=( None if validation_info is None else validation_info.adobe_revinfo_attr ), cades_signed_attrs=signature_meta.cades_signed_attr_spec, ) test_signature_cms = await signer.async_sign( test_md, md_algorithm, use_pades=self.use_pades, dry_run=True, timestamper=self.timestamper, signed_attr_settings=signed_attrs, ) # Note: multiply by 2 to account for the fact that this byte dump # will be embedded into the resulting PDF as a hexadecimal # string test_len = len(test_signature_cms.dump()) * 2 if tight: bytes_reserved = test_len else: # External actors such as timestamping servers can't be relied on to # always return exactly the same response, so we build in a 50% # error margin (+ ensure that bytes_reserved is even) bytes_reserved = test_len + 2 * (test_len // 4) return bytes_reserved def prepare_tbs_document( self, validation_info: Optional[PreSignValidationStatus], bytes_reserved, appearance_text_params=None, ) -> 'PdfTBSDocument': """ Set up the signature appearance (if necessary) and signature dictionary in the PDF file, to put the document in its final pre-signing state. :param validation_info: Validation information collected prior to signing. :param bytes_reserved: Bytes to reserve for the signature container. :param appearance_text_params: Optional text parameters for the signature appearance content. :return: A :class:`.PdfTBSDocument` describing the document in its final pre-signing state. """ pdf_signer = self.pdf_signer signature_meta = self.pdf_signer.signature_meta if self.sv_spec is not None: # process the field's seed value constraints self._enforce_seed_value_constraints( None if validation_info is None else validation_info.signer_path ) signer = pdf_signer.signer embed_roots = signer.embed_roots # take care of DSS updates, if they have to happen now dss_settings = signature_meta.dss_settings if self.use_pades and validation_info is not None: # Check consistency of settings dss_settings.assert_viable() if ( dss_settings.placement == SigDSSPlacementPreference.TOGETHER_WITH_SIGNATURE ): from pyhanko.sign import validation pdf_out = self.pdf_out # source info directly from the validation_info object # for consistency # NOTE: we have to disable VRI in this scenario validation.DocumentSecurityStore.supply_dss_in_writer( pdf_out, sig_contents=None, paths=validation_info.validation_paths, ocsps=validation_info.ocsps_to_embed, crls=validation_info.crls_to_embed, embed_roots=embed_roots, ) md_algorithm = self.md_algorithm sig_mdp_setup = self._apply_locking_rules() # Prepare instructions to the CMS writer to set up the # (PDF) signature object and its appearance system_time = self.system_time name_specified = signature_meta.name sig_appearance = SigAppearanceSetup( style=pdf_signer.stamp_style, name=name_specified or signer.subject_name, timestamp=system_time, text_params=appearance_text_params, ) sig_obj = SignatureObject( bytes_reserved=bytes_reserved, subfilter=self.subfilter, timestamp=system_time, name=name_specified if name_specified else None, location=signature_meta.location, reason=signature_meta.reason, contact_info=signature_meta.contact_info, prop_auth_time=signature_meta.prop_auth_time, prop_auth_type=signature_meta.prop_auth_type, app_build_props=signature_meta.app_build_props, ) # Pass in the SignatureObject settings sig_obj_ref = self.cms_writer.send( SigObjSetup( sig_placeholder=sig_obj, mdp_setup=sig_mdp_setup, appearance_setup=sig_appearance, ) ) signer = pdf_signer.signer security_handler = self.pdf_out.security_handler if security_handler is not None and security_handler.pdf_mac_enabled: _register_mac_on_sig(self.pdf_out, sig_obj_ref) PdfMacAttrProviderSpec(self.pdf_out).install( signer, self.timestamper ) # At this point, the document is in its final pre-signing state # Last job: prepare instructions for the post-signing workflow signature_meta = pdf_signer.signature_meta validation_context = signature_meta.validation_context post_signing_instr = doc_timestamper = None if self.use_pades and signature_meta.embed_validation_info: if validation_info is None: raise SigningError( "Requested embedding of validation info, " "but none was collected" ) if signature_meta.use_pades_lta: doc_timestamper = self.timestamper # if necessary/supported, extract a file access credential # to perform post-signing operations later credential_ser: Optional[SerialisedCredential] = None if security_handler is not None: credential = security_handler.extract_credential() if credential is not None: credential_ser = credential.serialise() # Preserve strict mode setting if we're doing # an incremental signature strict = True if isinstance(self.pdf_out, IncrementalPdfFileWriter): strict = self.pdf_out.prev.strict post_signing_instr = PostSignInstructions( validation_info=validation_info, # use the same algorithm # TODO make this configurable? Some TSAs only allow one choice # of MD, and forcing our signers to use the same one to handle # might be overly restrictive (esp. for things like EdDSA where # the MD is essentially fixed) timestamp_md_algorithm=md_algorithm, timestamper=doc_timestamper, timestamp_field_name=signature_meta.timestamp_field_name, dss_settings=signature_meta.dss_settings, tight_size_estimates=signature_meta.tight_size_estimates, embed_roots=embed_roots, file_credential=credential_ser, strict=strict, ) return PdfTBSDocument( cms_writer=self.cms_writer, signer=signer, md_algorithm=md_algorithm, timestamper=self.timestamper, use_pades=self.use_pades, post_sign_instructions=post_signing_instr, validation_context=validation_context, ) @dataclass(frozen=True) class PostSignInstructions: """ .. versionadded:: 0.7.0 Container class housing instructions for incremental updates to the document after the signature has been put in place. Necessary for PAdES-LT and PAdES-LTA workflows. """ validation_info: PreSignValidationStatus """ Validation information to embed in the DSS (if not already present). """ timestamper: Optional[TimeStamper] = None """ Timestamper to use for produce document timestamps. If ``None``, no timestamp will be added. """ timestamp_md_algorithm: Optional[str] = None """ Digest algorithm to use when producing timestamps. Defaults to :const:`~pyhanko.sign.signers.constants.DEFAULT_MD`. """ timestamp_field_name: Optional[str] = None """ Name of the timestamp field to use. If not specified, a field name will be generated. """ dss_settings: DSSContentSettings = DSSContentSettings() """ .. versionadded:: 0.8.0 Settings to fine-tune DSS generation. """ tight_size_estimates: bool = False """ .. versionadded:: 0.8.0 When estimating the size of a document timestamp container, do not add safety margins. .. note:: External TSAs cannot be relied upon to always produce the exact same output length, which makes this option risky to use. """ embed_roots: bool = True """ .. versionadded:: 0.9.0 Option that controls whether the root certificate of each validation path should be embedded into the DSS. The default is ``True``. .. note:: Trust roots are configured by the validator, so embedding them typically does nothing in a typical validation process. Therefore they can be safely omitted in most cases. Nonetheless, embedding the roots can be useful for documentation purposes. .. note:: This setting is not part of :class:`.DSSContentSettings` because its value is taken from the corresponding property on the :class:`.Signer` involved, not from the initial configuration. """ file_credential: Optional[SerialisedCredential] = None """ .. versionadded:: 0.13.0 Serialised file credential, to update encrypted files. """ strict: bool = True """ .. versionadded:: 0.25.2 Controls whether to open the signed document in strict mode before applying post-signing instructions. """ class PdfMacAttrProvider(attributes.CMSAttributeProvider): attribute_type = 'pdf_mac_data' def __init__( self, *, document_digest: bytes, signature_digest: bytes, mac_handler: 'pdfmac.PdfMacTokenHandler', ): self.document_digest = document_digest self.signature_digest = signature_digest self.mac_handler = mac_handler async def build_attr_value(self, dry_run=False): return self.mac_handler.build_pdfmac_token( document_digest=self.document_digest, signature_digest=self.signature_digest, dry_run=dry_run, ) class PdfMacAttrProviderSpec(attributes.UnsignedAttributeProviderSpec): """ Internal handler to deal with signatures under ISO/TS 32004. """ def __init__( self, pdf_out: BasePdfFileWriter, document_digest: Optional[bytes] = None, ): self.pdf_out = pdf_out self.document_digest = document_digest self.wrapped: Optional[attributes.UnsignedAttributeProviderSpec] = None self._orig: Optional[attributes.UnsignedAttributeProviderSpec] = None self._target: Optional[Signer] = None def install(self, signer: Signer, timestamper: Optional[TimeStamper]): """ Install the provider on a signer. NOTE: this is a stateful operation and won't work if the signer is shared, but we have a safeguard in place for that (will fail during size estimation under normal circumstances). """ # TODO find a way to handle this set/reset process more cleanly, # and/or expose more control. Would wrapping Signer be easier # in practice? It's cleaner in theory, but Signer is a bad mix # between extensible API and structural logic, so that might have to # wait. assert self._orig is None self._target = signer self._orig = signer.unsigned_attr_prov_spec self.wrapped = signer._unsigned_attr_provider_spec(timestamper) signer.unsigned_attr_prov_spec = self def uninstall(self): """ Remove the provider from its target (internal API). """ self._target.unsigned_attr_prov_spec = self._orig def unsigned_attr_providers( self, signature: bytes, signed_attrs: cms.CMSAttributes, digest_algorithm: str, ) -> Iterable[CMSAttributeProvider]: assert self.wrapped is not None, "Unsigned attr provider not installed" other_provs = self.wrapped.unsigned_attr_providers( signature, signed_attrs, digest_algorithm ) for prov in other_provs: if isinstance(prov, PdfMacAttrProvider): raise SigningError( "Other PDF MAC attribute provider found. This may be the " "result of signer instances being shared across multiple " "documents. This workflow is not supported for encrypted " "documents. Either turn off MAC generation, or use " "separate signer instances for different documents." ) yield prov md = hashes.Hash(get_pyca_cryptography_hash(digest_algorithm)) md.update(signature) sig_digest = md.finalize() document_digest = self.document_digest if not document_digest: # since by default we're aligning the hash function with that of # the underlying signature anyhow, # we can simply "steal" the value of the messageDigest attribute document_digest = find_unique_cms_attribute( signed_attrs, 'message_digest' ).native assert document_digest is not None yield PdfMacAttrProvider( document_digest=document_digest, signature_digest=sig_digest, mac_handler=self.pdf_out._init_mac_handler(digest_algorithm), ) class PdfTBSDocument: """ .. versionadded:: 0.7.0 A PDF document in its final pre-signing state. The ``__init__`` method is internal API, get an instance using :meth:`.PdfSigningSession.prepare_tbs_document`. Alternatively, use :meth:`resume_signing` or :meth:`finish_signing` to continue a previously interrupted signing process without instantiating a new :class:`.PdfTBSDocument` object. """ def __init__( self, cms_writer, signer: Signer, md_algorithm: str, use_pades: bool, timestamper: Optional[TimeStamper] = None, post_sign_instructions: Optional[PostSignInstructions] = None, validation_context: Optional[ValidationContext] = None, ): self.cms_writer = cms_writer self.signer = signer self.md_algorithm = md_algorithm self.timestamper = timestamper self.use_pades = use_pades self.post_sign_instructions = post_sign_instructions self.validation_context = validation_context def digest_tbs_document( self, *, output: Optional[IO] = None, in_place: bool = False, chunk_size=misc.DEFAULT_CHUNK_SIZE, ) -> Tuple[PreparedByteRangeDigest, IO]: """ Write the document to an output stream and compute the digest, while keeping track of the (future) location of the signature contents in the output stream. The digest can then be passed to the next part of the signing pipeline. .. warning:: This method can only be called once. :param output: Write the output to the specified output stream. If ``None``, write to a new :class:`.BytesIO` object. Default is ``None``. :param in_place: Sign the original input stream in-place. This parameter overrides ``output``. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. :return: A tuple containing a :class:`.PreparedByteRangeDigest` and the output stream to which the output was written. """ # pass in I/O parameters, get back a hash return self.cms_writer.send( SigIOSetup( md_algorithm=self.md_algorithm, in_place=in_place, chunk_size=chunk_size, output=output, ) ) async def perform_signature( self, document_digest: bytes, pdf_cms_signed_attrs: PdfCMSSignedAttributes, ) -> 'PdfPostSignatureDocument': """ Perform the relevant cryptographic signing operations on the document digest, and write the resulting CMS object to the appropriate location in the output stream. .. warning:: This method can only be called once, and must be invoked after :meth:`digest_tbs_document`. :param document_digest: Digest of the document, as computed over the relevant ``/ByteRange``. :param pdf_cms_signed_attrs: Description of the signed attributes to include. :return: A :class:`.PdfPostSignatureDocument` object. """ signer = self.signer signature_cms = await signer.async_sign( document_digest, self.md_algorithm, use_pades=self.use_pades, timestamper=self.timestamper, signed_attr_settings=pdf_cms_signed_attrs, ) # ... and feed it to the CMS writer sig_contents = self.cms_writer.send(signature_cms) # lastly, restore the signer if isinstance(signer.unsigned_attr_prov_spec, PdfMacAttrProviderSpec): signer.unsigned_attr_prov_spec.uninstall() return PdfPostSignatureDocument( sig_contents, post_sign_instr=self.post_sign_instructions, validation_context=self.validation_context, ) @classmethod def resume_signing( cls, output: IO, prepared_digest: PreparedByteRangeDigest, signature_cms: Union[bytes, cms.ContentInfo], post_sign_instr: Optional[PostSignInstructions] = None, validation_context: Optional[ValidationContext] = None, ) -> 'PdfPostSignatureDocument': """ Resume signing after obtaining a CMS object from an external source. This is a class method; it doesn't require a :class:`.PdfTBSDocument` instance. Contrast with :meth:`perform_signature`. :param output: Output stream housing the document in its final pre-signing state. This stream must at least be writable and seekable, and also readable if post-signature processing is required. :param prepared_digest: The prepared digest returned by a prior call to :meth:`digest_tbs_document`. :param signature_cms: CMS object to embed in the signature dictionary. :param post_sign_instr: Instructions for post-signing processing (DSS updates and document timestamps). :param validation_context: Validation context to use in post-signing operations. This is mainly intended for TSA certificate validation, but it can also contain additional validation data to embed in the DSS. :return: A :class:`PdfPostSignatureDocument`. """ sig_contents = prepared_digest.fill_with_cms(output, signature_cms) return PdfPostSignatureDocument( sig_contents, post_sign_instr=post_sign_instr, validation_context=validation_context, ) @classmethod def finish_signing( cls, output: IO, prepared_digest: PreparedByteRangeDigest, signature_cms: Union[bytes, cms.ContentInfo], post_sign_instr: Optional[PostSignInstructions] = None, validation_context: Optional[ValidationContext] = None, chunk_size=misc.DEFAULT_CHUNK_SIZE, ): """ Finish signing after obtaining a CMS object from an external source, and perform any required post-signature processing. This is a class method; it doesn't require a :class:`.PdfTBSDocument` instance. Contrast with :meth:`perform_signature`. :param output: Output stream housing the document in its final pre-signing state. :param prepared_digest: The prepared digest returned by a prior call to :meth:`digest_tbs_document`. :param signature_cms: CMS object to embed in the signature dictionary. :param post_sign_instr: Instructions for post-signing processing (DSS updates and document timestamps). :param validation_context: Validation context to use in post-signing operations. This is mainly intended for TSA certificate validation, but it can also contain additional validation data to embed in the DSS. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. """ asyncio.run( cls.async_finish_signing( output, prepared_digest, signature_cms, post_sign_instr=post_sign_instr, validation_context=validation_context, chunk_size=chunk_size, ) ) @classmethod async def async_finish_signing( cls, output: IO, prepared_digest: PreparedByteRangeDigest, signature_cms: Union[bytes, cms.ContentInfo], post_sign_instr: Optional[PostSignInstructions] = None, validation_context: Optional[ValidationContext] = None, chunk_size=misc.DEFAULT_CHUNK_SIZE, ): """ Finish signing after obtaining a CMS object from an external source, and perform any required post-signature processing. This is a class method; it doesn't require a :class:`.PdfTBSDocument` instance. Contrast with :meth:`perform_signature`. :param output: Output stream housing the document in its final pre-signing state. :param prepared_digest: The prepared digest returned by a prior call to :meth:`digest_tbs_document`. :param signature_cms: CMS object to embed in the signature dictionary. :param post_sign_instr: Instructions for post-signing processing (DSS updates and document timestamps). :param validation_context: Validation context to use in post-signing operations. This is mainly intended for TSA certificate validation, but it can also contain additional validation data to embed in the DSS. :param chunk_size: Size of the internal buffer (in bytes) used to feed data to the message digest function if the input stream does not support ``memoryview``. """ # TODO at this point, the output stream no longer needs to be readable, # just seekable, unless there's a timestamp requirement. # Might want to factor that out for speed at some point. rw_output = misc.prepare_rw_output_stream(output) post_sign = cls.resume_signing( rw_output, prepared_digest=prepared_digest, signature_cms=signature_cms, post_sign_instr=post_sign_instr, validation_context=validation_context, ) await post_sign.post_signature_processing( rw_output, chunk_size=chunk_size ) class PdfPostSignatureDocument: """ .. versionadded:: 0.7.0 Represents the final phase of the PDF signing process """ def __init__( self, sig_contents: bytes, post_sign_instr: Optional[PostSignInstructions] = None, validation_context: Optional[ValidationContext] = None, ): self.sig_contents = sig_contents self.post_sign_instructions = post_sign_instr self.validation_context = validation_context async def post_signature_processing( self, output: IO, chunk_size=misc.DEFAULT_CHUNK_SIZE ): """ Handle DSS updates and LTA timestamps, if applicable. :param output: I/O buffer containing the signed document. Must support reading, writing and seeking. :param chunk_size: Chunk size to use for I/O operations that do not support the buffer protocol. """ instr = self.post_sign_instructions if instr is None: return validation_context = self.validation_context validation_info = instr.validation_info dss_settings = instr.dss_settings from pyhanko.sign import validation # If we're resuming a signing operation, the (new) validation context # might not have all relevant OCSP responses / CRLs available. # Hence why we also pass in the data from the pre-signing check. # The DSS handling code will deal with deduplication. dss_op_kwargs: Dict[str, Any] dss_op_kwargs = dict( paths=validation_info.validation_paths, validation_context=validation_context, ocsps=validation_info.ocsps_to_embed, crls=validation_info.crls_to_embed, embed_roots=instr.embed_roots, ) if dss_settings.include_vri: dss_op_kwargs['sig_contents'] = self.sig_contents else: dss_op_kwargs['sig_contents'] = None timestamper = instr.timestamper # Separate DSS revision if no TS that would otherwise be bundled with it # or explicitly requested as separate dss_placement = dss_settings.placement separate_dss_revision = False if dss_placement == SigDSSPlacementPreference.SEPARATE_REVISION: separate_dss_revision = True elif dss_placement == SigDSSPlacementPreference.TOGETHER_WITH_NEXT_TS: separate_dss_revision = timestamper is None if separate_dss_revision: if not dss_settings.skip_if_unneeded: dss_op_kwargs['force_write'] = True validation.DocumentSecurityStore.add_dss( output_stream=output, **dss_op_kwargs, file_credential=instr.file_credential, strict=instr.strict, ) if timestamper is not None: # append a document timestamp after the DSS update w = IncrementalPdfFileWriter(output, strict=instr.strict) if ( w.security_handler is not None and instr.file_credential is not None ): w.security_handler.authenticate(instr.file_credential) # we let the SH throw errors on access as necessary pdf_timestamper = PdfTimeStamper( timestamper, field_name=instr.timestamp_field_name ) if dss_placement == SigDSSPlacementPreference.TOGETHER_WITH_NEXT_TS: validation.DocumentSecurityStore.supply_dss_in_writer( w, **dss_op_kwargs ) await pdf_timestamper.async_timestamp_pdf( w, instr.timestamp_md_algorithm or constants.DEFAULT_MD, validation_context, validation_paths=validation_info.validation_paths, in_place=True, timestamper=timestamper, chunk_size=chunk_size, dss_settings=dss_settings.get_settings_for_ts(), tight_size_estimates=instr.tight_size_estimates, embed_roots=instr.embed_roots, )