""" This module contains a number of functions to handle AdES signature validation. .. danger:: This API is incubating, and not all features of the spec have been fully implemented at this stage. There will be bugs, and API changes may still occur. """ import asyncio import dataclasses import itertools import logging from copy import copy from dataclasses import dataclass from datetime import datetime, timezone from typing import ( Any, Dict, FrozenSet, Generator, Generic, Iterable, Iterator, List, Optional, Set, Tuple, Type, TypeVar, Union, overload, ) import tzlocal from asn1crypto import cms, keys from asn1crypto import pdf as asn1_pdf from asn1crypto import tsp, x509 from asn1crypto.crl import CertificateList from asn1crypto.ocsp import OCSPResponse from pyhanko_certvalidator import ValidationContext from pyhanko_certvalidator.authority import CertTrustAnchor, TrustAnchor from pyhanko_certvalidator.context import ( CertValidationPolicySpec, ValidationDataHandlers, ) from pyhanko_certvalidator.errors import PathError from pyhanko_certvalidator.ltv.ades_past import past_validate from pyhanko_certvalidator.ltv.poe import ( KnownPOE, POEManager, POEType, ValidationObject, ValidationObjectType, digest_for_poe, ) from pyhanko_certvalidator.ltv.time_slide import ades_gather_prima_facie_revinfo from pyhanko_certvalidator.ltv.types import ValidationTimingInfo from pyhanko_certvalidator.path import ValidationPath from pyhanko_certvalidator.policy_decl import ( AlgorithmUsagePolicy, NonRevokedStatusAssertion, RevocationCheckingRule, ) from pyhanko_certvalidator.registry import PathBuilder, TrustManager from pyhanko_certvalidator.revinfo.archival import CRLContainer, OCSPContainer from pyhanko_certvalidator.revinfo.validate_crl import CRLOfInterest from pyhanko_certvalidator.revinfo.validate_ocsp import OCSPResponseOfInterest from pyhanko.pdf_utils.reader import HistoricalResolver, PdfFileReader from pyhanko.sign.ades.report import ( AdESFailure, AdESIndeterminate, AdESPassed, AdESStatus, AdESSubIndic, ) from pyhanko.sign.general import ( CMSExtractionError, CMSStructuralError, MultivaluedAttributeError, NonexistentAttributeError, extract_certificate_info, find_cms_attribute, find_unique_cms_attribute, ) from pyhanko.sign.validation import ( DocumentSecurityStore, EmbeddedPdfSignature, errors, generic_cms, ) from pyhanko.sign.validation.settings import KeyUsageConstraints from pyhanko.sign.validation.status import ( DocumentTimestampStatus, PdfSignatureStatus, RevocationDetails, SignatureCoverageLevel, SignatureStatus, SignerAttributeStatus, StandardCMSSignatureStatus, TimestampSignatureStatus, ) from ..diff_analysis import DiffPolicy from .dss import enumerate_ocsp_certs from .errors import NoDSSFoundError from .policy_decl import ( LocalKnowledge, PdfSignatureValidationSpec, RevinfoOnlineFetchingRule, RevocationInfoGatheringSpec, SignatureValidationSpec, bootstrap_validation_data_handlers, ) from .utils import CMSAlgorithmUsagePolicy __all__ = [ 'ades_basic_validation', 'ades_with_time_validation', 'ades_lta_validation', 'ades_timestamp_validation', 'simulate_future_ades_lta_validation', 'AdESBasicValidationResult', 'AdESWithTimeValidationResult', 'AdESLTAValidationResult', 'derive_validation_object_identifier', ] logger = logging.getLogger(__name__) StatusType = TypeVar('StatusType', bound=SignatureStatus, covariant=True) def derive_validation_object_binary_data( vo: ValidationObject, ) -> Optional[bytes]: if vo.object_type == ValidationObjectType.CERTIFICATE: return vo.value.dump() elif vo.object_type == ValidationObjectType.CRL: return vo.value.crl_data.dump() elif vo.object_type == ValidationObjectType.OCSP_RESPONSE: return vo.value.ocsp_response_data.dump() elif vo.object_type in ( ValidationObjectType.SIGNED_DATA, ValidationObjectType.TIMESTAMP, ): return vo.value['signer_infos'][0]['signature'].native else: return None def derive_validation_object_identifier(vo: ValidationObject) -> Optional[str]: # TODO for certs and signers, it could make sense to somehow encode # a human-readable "slugified" representation of the common name # to identify things at a glance. if vo.object_type == ValidationObjectType.CERTIFICATE: marker = digest_for_poe(vo.value.dump()).hex() elif vo.object_type == ValidationObjectType.CRL: marker = digest_for_poe(vo.value.crl_data.dump()).hex() elif vo.object_type == ValidationObjectType.OCSP_RESPONSE: marker = digest_for_poe(vo.value.ocsp_response_data.dump()).hex() elif vo.object_type in ( ValidationObjectType.SIGNED_DATA, ValidationObjectType.TIMESTAMP, ): marker = digest_for_poe( vo.value['signer_infos'][0]['signature'].native ).hex() else: return None return f'vo-{vo.object_type.value}-{marker}' class ValidationObjectSet: def __init__(self, *object_collections: Iterable[ValidationObject]): def _pairs(): for obj in itertools.chain(*object_collections): ident = derive_validation_object_identifier(obj) if ident: yield ident, obj self._things = {k: v for k, v in _pairs()} def __iter__(self) -> Iterator[ValidationObject]: return iter(self._things.values()) @staticmethod def empty(): return ValidationObjectSet(()) @dataclass(frozen=True) class AdESBasicValidationResult(Generic[StatusType]): """ Result of validation of basic signatures. ETSI EN 319 102-1, § 5.3 """ ades_subindic: AdESSubIndic """ AdES subindication. """ api_status: Optional[StatusType] """ A status descriptor object from pyHanko's own validation API. Will be an instance of :class:`.SignatureStatus` or a subclass thereof. """ failure_msg: Optional[str] """ A string describing the reason why validation failed, if applicable. """ validation_objects: ValidationObjectSet """ Validation objects that were potentially relevant for the validation process. """ @dataclass class _InternalBasicValidationResult: ades_subindic: AdESSubIndic signature_poe_time: Optional[datetime] signature_not_before_time: Optional[datetime] validation_path: Optional[ValidationPath] status_kwargs: dict = dataclasses.field(default_factory=dict) trust_subindic_update: Optional[AdESSubIndic] = None signature_ts_validity: Optional[TimestampSignatureStatus] = None content_ts_validity: Optional[TimestampSignatureStatus] = None signer_attr_status: Optional[SignerAttributeStatus] = None def update(self, status_cls, with_ts, with_attrs): status_kwargs = self.status_kwargs status_kwargs['validation_path'] = self.validation_path if self.trust_subindic_update: status_kwargs['trust_problem_indic'] = self.trust_subindic_update if with_ts and self.signature_ts_validity: status_kwargs['timestamp_validity'] = self.signature_ts_validity if with_ts and self.content_ts_validity: status_kwargs['content_timestamp_validity'] = ( self.content_ts_validity ) if with_attrs and self.signer_attr_status: status_kwargs['ac_attrs'] = self.signer_attr_status.ac_attrs status_kwargs['cades_signer_attrs'] = ( self.signer_attr_status.cades_signer_attrs ) status_kwargs['ac_validation_errs'] = ( self.signer_attr_status.ac_validation_errs ) return status_cls(**status_kwargs) @overload async def ades_timestamp_validation( tst_signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, expected_tst_imprint: bytes, *, status_cls: Type[StatusType], timing_info: Optional[ValidationTimingInfo] = None, validation_data_handlers: Optional[ValidationDataHandlers] = None, extra_status_kwargs: Optional[Dict[str, Any]] = None, ) -> AdESBasicValidationResult: ... @overload async def ades_timestamp_validation( tst_signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, expected_tst_imprint: bytes, *, timing_info: Optional[ValidationTimingInfo] = None, validation_data_handlers: Optional[ValidationDataHandlers] = None, extra_status_kwargs: Optional[Dict[str, Any]] = None, ) -> AdESBasicValidationResult: ... async def ades_timestamp_validation( tst_signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, expected_tst_imprint: bytes, timing_info: Optional[ValidationTimingInfo] = None, validation_data_handlers: Optional[ValidationDataHandlers] = None, extra_status_kwargs: Optional[Dict[str, Any]] = None, status_cls=TimestampSignatureStatus, ) -> AdESBasicValidationResult: """ Validate a timestamp token according to ETSI EN 319 102-1 § 5.4. :param tst_signed_data: The ``SignedData`` value of the timestamp. :param validation_spec: Validation settings to apply. :param expected_tst_imprint: The expected message imprint in the timestamp token. :param timing_info: Data object describing the timing of the validation. Defaults to :meth:`.ValidationTimingInfo.now`. :param validation_data_handlers: Data handlers to manage validation data. :param extra_status_kwargs: Extra keyword arguments to pass to the signature status object's ``__init__`` function. :param status_cls: The class of the resulting status object in pyHanko's internal validation API. :return: A :class:`.AdESBasicValidationResult`. """ timing_info = timing_info or ValidationTimingInfo.now() cert_validation_policy = ( validation_spec.ts_cert_validation_policy or validation_spec.cert_validation_policy ) if validation_data_handlers is None: validation_data_handlers = bootstrap_validation_data_handlers( spec=validation_spec, timing_info=timing_info ) validation_context = cert_validation_policy.build_validation_context( timing_info=timing_info, handlers=validation_data_handlers ) return await _ades_timestamp_validation_from_context( tst_signed_data, validation_context, expected_tst_imprint, extra_status_kwargs=extra_status_kwargs, status_cls=status_cls, ) def _ades_signature_crypto_policy_check( signer_info: cms.SignerInfo, algo_policy: AlgorithmUsagePolicy, control_time: datetime, public_key: Optional[keys.PublicKeyInfo], ): sig_algo: cms.SignedDigestAlgorithm = signer_info['signature_algorithm'] sig_allowed = algo_policy.signature_algorithm_allowed( sig_algo, control_time, public_key=public_key ) if not sig_allowed: msg = ( f"Signature algorithm {sig_algo.signature_algo} not allowed as " f"of {control_time}, which is " f"the time of the earliest PoE for the signature." ) raise errors.SignatureValidationError( msg, ades_subindication=( AdESIndeterminate.CRYPTO_CONSTRAINTS_FAILURE if sig_allowed.not_allowed_after is None else AdESIndeterminate.CRYPTO_CONSTRAINTS_FAILURE_NO_POE ), ) def _enumerate_validation_objects( validation_context: Optional[ValidationContext], ) -> Generator[ValidationObject, None, None]: if validation_context is None: return for ocsp in validation_context.ocsps: for cont in OCSPContainer.load_multi(ocsp): yield ValidationObject(ValidationObjectType.OCSP_RESPONSE, cont) for cert in enumerate_ocsp_certs(ocsp): yield ValidationObject(ValidationObjectType.CERTIFICATE, cert) for crl in validation_context.crls: yield ValidationObject(ValidationObjectType.CRL, CRLContainer(crl)) def _enumerate_certs_in_paths( status: Union[SignatureStatus, _InternalBasicValidationResult, None], ): if status is None: return path = status.validation_path if path: for cert in path.iter_certs(include_root=True): yield ValidationObject(ValidationObjectType.CERTIFICATE, cert) if isinstance(status, StandardCMSSignatureStatus): yield from _enumerate_certs_in_paths(status.timestamp_validity) yield from _enumerate_certs_in_paths(status.content_timestamp_validity) if isinstance(status, _InternalBasicValidationResult): yield from _enumerate_certs_in_paths(status.signature_ts_validity) yield from _enumerate_certs_in_paths(status.content_ts_validity) async def _ades_timestamp_validation_from_context( tst_signed_data: cms.SignedData, validation_context: ValidationContext, expected_tst_imprint: bytes, extra_status_kwargs: Optional[Dict[str, Any]] = None, status_cls=TimestampSignatureStatus, ) -> AdESBasicValidationResult: vos = ValidationObjectSet(_enumerate_validation_objects(validation_context)) status_kwargs = dict(extra_status_kwargs or {}) status_kwargs_from_validation = await generic_cms.validate_tst_signed_data( tst_signed_data, validation_context=validation_context, expected_tst_imprint=expected_tst_imprint, ) status_kwargs.update(status_kwargs_from_validation) # noinspection PyArgumentList status = status_cls(**status_kwargs) if not status.intact: return AdESBasicValidationResult( ades_subindic=AdESFailure.HASH_FAILURE, api_status=status, failure_msg=None, validation_objects=vos, ) elif not status.valid: return AdESBasicValidationResult( ades_subindic=AdESFailure.SIG_CRYPTO_FAILURE, api_status=status, failure_msg=None, validation_objects=vos, ) interm_result = await _process_basic_validation( tst_signed_data, status, validation_context, ac_validation_context=None, signature_not_before_time=None, ) interm_result.status_kwargs = status_kwargs vos = ValidationObjectSet( iter(vos), _enumerate_certs_in_paths(interm_result) ) return AdESBasicValidationResult( ades_subindic=interm_result.ades_subindic, api_status=interm_result.update( status_cls, with_ts=False, with_attrs=False ), failure_msg=None, validation_objects=vos, ) async def _ades_process_attached_ts( signer_info, validation_context, signed: bool, tst_digest: bytes ) -> AdESBasicValidationResult: tst_signed_data = generic_cms.extract_tst_data(signer_info, signed=signed) if tst_signed_data is not None: return await _ades_timestamp_validation_from_context( tst_signed_data, validation_context, tst_digest, ) return AdESBasicValidationResult( ades_subindic=AdESIndeterminate.GENERIC, failure_msg=None, api_status=None, validation_objects=ValidationObjectSet.empty(), ) async def _process_basic_validation( signed_data: cms.SignedData, temp_status: SignatureStatus, ts_validation_context: ValidationContext, ac_validation_context: Optional[ValidationContext], signature_not_before_time: Optional[datetime], ): validation_time = temp_status.validation_time ades_trust_status: Optional[AdESSubIndic] = temp_status.trust_problem_indic signer_info = generic_cms.extract_signer_info(signed_data) ts_status: Optional[TimestampSignatureStatus] = None if ades_trust_status in ( AdESIndeterminate.REVOKED_NO_POE, AdESIndeterminate.OUT_OF_BOUNDS_NO_POE, ): # check content timestamp # TODO allow selecting one of multiple here # FIXME here and in a few other places we presume that the message # imprint algorithm agrees with the TS's algorithm. This is a fairly # safe assumption, but not an airtight one. content_ts_result = await _ades_process_attached_ts( signer_info, ts_validation_context, signed=True, tst_digest=generic_cms.find_unique_cms_attribute( signer_info['signed_attrs'], 'message_digest' ).native, ) if content_ts_result.ades_subindic == AdESPassed.OK: ts_status = content_ts_result.api_status assert ts_status is not None if signature_not_before_time is not None: signature_not_before_time = max( ts_status.timestamp, signature_not_before_time ) else: signature_not_before_time = ts_status.timestamp # now we potentially have POE to know for sure that the signer's # certificate was in fact revoked/expired. # HOWEVER, according to the spec it is _not_ within this functions # remit to check the signature timestamp to reverse a positive # X_NO_POE judgement!! perm_status: AdESSubIndic if ades_trust_status == AdESIndeterminate.REVOKED_NO_POE: revo_details = temp_status.revocation_details assert revo_details is not None cutoff = revo_details.revocation_date perm_status = AdESFailure.REVOKED else: cutoff = temp_status.signing_cert.not_valid_after perm_status = AdESIndeterminate.EXPIRED # help the typechecker assert signature_not_before_time is not None if signature_not_before_time >= cutoff: ades_trust_status = perm_status # TODO process signature policy attr once we can cert_info = generic_cms.extract_certificate_info(signed_data) # FIXME this is not entirely correct, we need past validation # for attr certs as well attr_status_kwargs = await generic_cms.collect_signer_attr_status( sd_attr_certificates=cert_info.attribute_certs, signer_cert=cert_info.signer_cert, validation_context=ac_validation_context, sd_signed_attrs=signer_info['signed_attrs'], ) ades_subindic = ades_trust_status or AdESPassed.OK return _InternalBasicValidationResult( ades_subindic=ades_subindic, trust_subindic_update=ades_trust_status, content_ts_validity=ts_status, signature_not_before_time=signature_not_before_time, signer_attr_status=SignerAttributeStatus(**attr_status_kwargs), signature_poe_time=None, validation_path=temp_status.validation_path, status_kwargs={'validation_time': validation_time}, ) def _init_vcs( validation_spec: SignatureValidationSpec, timing_info: ValidationTimingInfo, validation_data_handlers: ValidationDataHandlers, ): validation_context = ( validation_spec.cert_validation_policy.build_validation_context( timing_info=timing_info, handlers=validation_data_handlers ) ) if validation_spec.ts_cert_validation_policy is not None: ts_validation_context = ( validation_spec.ts_cert_validation_policy.build_validation_context( timing_info=timing_info, handlers=validation_data_handlers ) ) else: ts_validation_context = validation_context if validation_spec.ac_validation_policy is not None: ac_validation_context = ( validation_spec.ac_validation_policy.build_validation_context( timing_info=timing_info, handlers=validation_data_handlers ) ) else: ac_validation_context = None return validation_context, ts_validation_context, ac_validation_context # ETSI EN 319 102-1 § 5.3 @overload async def ades_basic_validation( signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, *, status_cls: Type[StatusType], timing_info: Optional[ValidationTimingInfo] = None, raw_digest: Optional[bytes] = None, validation_data_handlers: Optional[ValidationDataHandlers] = None, signature_not_before_time: Optional[datetime] = None, extra_status_kwargs: Optional[Dict[str, Any]] = None, ) -> AdESBasicValidationResult: ... @overload async def ades_basic_validation( signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, *, timing_info: Optional[ValidationTimingInfo] = None, raw_digest: Optional[bytes] = None, validation_data_handlers: Optional[ValidationDataHandlers] = None, signature_not_before_time: Optional[datetime] = None, extra_status_kwargs: Optional[Dict[str, Any]] = None, ) -> AdESBasicValidationResult: ... async def ades_basic_validation( signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, timing_info: Optional[ValidationTimingInfo] = None, raw_digest: Optional[bytes] = None, validation_data_handlers: Optional[ValidationDataHandlers] = None, signature_not_before_time: Optional[datetime] = None, extra_status_kwargs: Optional[Dict[str, Any]] = None, status_cls=StandardCMSSignatureStatus, ) -> AdESBasicValidationResult: """ Validate a CMS signature according to ETSI EN 319 102-1 § 5.3. :param signed_data: The ``SignedData`` value. :param validation_spec: Validation settings to apply. :param raw_digest: The expected message digest attribute value. :param timing_info: Data object describing the timing of the validation. Defaults to :meth:`.ValidationTimingInfo.now`. :param validation_data_handlers: Data handlers to manage validation data. :param extra_status_kwargs: Extra keyword arguments to pass to the signature status object's ``__init__`` function. :param status_cls: The class of the resulting status object in pyHanko's internal validation API. :param signature_not_before_time: Time when the signature was known _not_ to exist. :return: A :class:`.AdESBasicValidationResult`. """ timing_info = timing_info or ValidationTimingInfo.now() if validation_data_handlers is None: validation_data_handlers = bootstrap_validation_data_handlers( spec=validation_spec, timing_info=timing_info ) ( validation_context, ts_validation_context, ac_validation_context, ) = _init_vcs(validation_spec, timing_info, validation_data_handlers) interm_result = await _ades_basic_validation( signed_data=signed_data, validation_context=validation_context, ts_validation_context=ts_validation_context, ac_validation_context=ac_validation_context, key_usage_settings=validation_spec.key_usage_settings, raw_digest=raw_digest, signature_not_before_time=signature_not_before_time, extra_status_kwargs=extra_status_kwargs, status_cls=status_cls, algorithm_policy=validation_spec.signature_algorithm_policy, ) if isinstance(interm_result, AdESBasicValidationResult): return interm_result status: StandardCMSSignatureStatus = interm_result.update( StandardCMSSignatureStatus, with_ts=False, with_attrs=True ) vos = ValidationObjectSet( _enumerate_validation_objects(validation_context), _enumerate_validation_objects(ac_validation_context), _enumerate_validation_objects(ts_validation_context), _enumerate_certs_in_paths(status), ) return AdESBasicValidationResult( ades_subindic=interm_result.ades_subindic, api_status=status, failure_msg=None, validation_objects=vos, ) async def _ades_basic_validation( signed_data: cms.SignedData, validation_context: ValidationContext, ts_validation_context: ValidationContext, ac_validation_context: Optional[ValidationContext], key_usage_settings: KeyUsageConstraints, raw_digest: Optional[bytes], signature_not_before_time: Optional[datetime], extra_status_kwargs: Optional[Dict[str, Any]], algorithm_policy: Optional[CMSAlgorithmUsagePolicy], status_cls: Type[StatusType], ) -> Union[AdESBasicValidationResult, _InternalBasicValidationResult]: status_kwargs = dict(extra_status_kwargs or {}) vos = ValidationObjectSet( _enumerate_validation_objects(validation_context), _enumerate_validation_objects(ts_validation_context), _enumerate_validation_objects(ac_validation_context), ) try: status_kwargs_from_validation = await generic_cms.cms_basic_validation( signed_data, raw_digest=raw_digest, validation_context=validation_context, key_usage_settings=key_usage_settings, algorithm_policy=algorithm_policy, ) status_kwargs.update(status_kwargs_from_validation) except errors.SignatureValidationError as e: return AdESBasicValidationResult( ades_subindic=e.ades_subindication or AdESIndeterminate.GENERIC, failure_msg=e.failure_message, api_status=None, validation_objects=vos, ) # put the temp status into a SignatureStatus object for convenience status: SignatureStatus = status_cls(**status_kwargs) if not status.intact: return AdESBasicValidationResult( ades_subindic=AdESFailure.HASH_FAILURE, api_status=status, failure_msg=None, validation_objects=vos, ) elif not status.valid: return AdESBasicValidationResult( ades_subindic=AdESFailure.SIG_CRYPTO_FAILURE, api_status=status, failure_msg=None, validation_objects=vos, ) interm_result = await _process_basic_validation( signed_data, status, ts_validation_context, ac_validation_context=ac_validation_context, signature_not_before_time=signature_not_before_time, ) interm_result.status_kwargs = status_kwargs return interm_result @dataclass(frozen=True) class AdESWithTimeValidationResult(AdESBasicValidationResult): best_signature_time: datetime signature_not_before_time: Optional[datetime] _WITH_TIME_FURTHER_PROC = frozenset( { AdESPassed.OK, AdESIndeterminate.CRYPTO_CONSTRAINTS_FAILURE_NO_POE, AdESIndeterminate.REVOKED_NO_POE, AdESIndeterminate.REVOKED_CA_NO_POE, AdESIndeterminate.TRY_LATER, AdESIndeterminate.OUT_OF_BOUNDS_NO_POE, } ) @overload async def ades_with_time_validation( signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, *, timing_info: Optional[ValidationTimingInfo] = None, raw_digest: Optional[bytes] = None, validation_data_handlers: Optional[ValidationDataHandlers] = None, signature_not_before_time: Optional[datetime] = None, extra_status_kwargs: Optional[Dict[str, Any]] = None, ) -> AdESWithTimeValidationResult: ... @overload async def ades_with_time_validation( signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, *, status_cls: Type[StatusType], timing_info: Optional[ValidationTimingInfo] = None, raw_digest: Optional[bytes] = None, validation_data_handlers: Optional[ValidationDataHandlers] = None, signature_not_before_time: Optional[datetime] = None, extra_status_kwargs: Optional[Dict[str, Any]] = None, ) -> AdESWithTimeValidationResult: ... async def ades_with_time_validation( signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, timing_info: Optional[ValidationTimingInfo] = None, raw_digest: Optional[bytes] = None, validation_data_handlers: Optional[ValidationDataHandlers] = None, signature_not_before_time: Optional[datetime] = None, extra_status_kwargs: Optional[Dict[str, Any]] = None, status_cls=StandardCMSSignatureStatus, ) -> AdESWithTimeValidationResult: """ Validate a CMS signature with time according to ETSI EN 319 102-1 § 5.5. :param signed_data: The ``SignedData`` value. :param validation_spec: Validation settings to apply. :param raw_digest: The expected message digest attribute value. :param timing_info: Data object describing the timing of the validation. Defaults to :meth:`.ValidationTimingInfo.now`. :param validation_data_handlers: Data handlers to manage validation data. :param extra_status_kwargs: Extra keyword arguments to pass to the signature status object's ``__init__`` function. :param status_cls: The class of the resulting status object in pyHanko's internal validation API. :param signature_not_before_time: Time when the signature was known _not_ to exist. :return: A :class:`.AdESBasicValidationResult`. """ timing_info = timing_info or ValidationTimingInfo.now() if validation_data_handlers is None: validation_data_handlers = bootstrap_validation_data_handlers( spec=validation_spec, timing_info=timing_info ) ( validation_context, ts_validation_context, ac_validation_context, ) = _init_vcs(validation_spec, timing_info, validation_data_handlers) sig_bytes = signed_data['signer_infos'][0]['signature'].native signature_poe_time = validation_data_handlers.poe_manager[sig_bytes] interm_result = await _ades_basic_validation( signed_data, validation_context=validation_context, ts_validation_context=ts_validation_context, ac_validation_context=ac_validation_context, key_usage_settings=validation_spec.key_usage_settings, raw_digest=raw_digest, signature_not_before_time=signature_not_before_time, extra_status_kwargs=extra_status_kwargs, status_cls=status_cls, algorithm_policy=validation_spec.signature_algorithm_policy, ) if isinstance(interm_result, AdESBasicValidationResult): vos = ValidationObjectSet( _enumerate_validation_objects(validation_context), _enumerate_validation_objects(ts_validation_context), _enumerate_validation_objects(ac_validation_context), _enumerate_certs_in_paths(interm_result.api_status), ) return AdESWithTimeValidationResult( ades_subindic=interm_result.ades_subindic, api_status=interm_result.api_status, failure_msg=interm_result.failure_msg, best_signature_time=signature_poe_time, signature_not_before_time=signature_not_before_time, validation_objects=vos, ) elif interm_result.ades_subindic not in _WITH_TIME_FURTHER_PROC: assert isinstance(interm_result, _InternalBasicValidationResult) signature_not_before_time = interm_result.signature_not_before_time api_status = interm_result.update( status_cls, with_ts=True, with_attrs=True ) vos = ValidationObjectSet( _enumerate_validation_objects(validation_context), _enumerate_validation_objects(ts_validation_context), _enumerate_validation_objects(ac_validation_context), _enumerate_certs_in_paths(api_status), ) return AdESWithTimeValidationResult( ades_subindic=interm_result.ades_subindic, api_status=api_status, failure_msg=None, best_signature_time=signature_poe_time, signature_not_before_time=signature_not_before_time, validation_objects=vos, ) signer_info = generic_cms.extract_signer_info(signed_data) temp_status = interm_result.update( status_cls, with_ts=False, with_attrs=True ) # process signature timestamps # TODO allow selecting one of multiple timestamps here? tst_digest = generic_cms.compute_signature_tst_digest(signer_info) if tst_digest is None: # TODO conditionally enforce this based on policy params--- # for now we assume that someone calling this method actually cares # about timestamps vos = ValidationObjectSet( _enumerate_validation_objects(validation_context), _enumerate_validation_objects(ts_validation_context), _enumerate_validation_objects(ac_validation_context), _enumerate_certs_in_paths(interm_result), ) return AdESWithTimeValidationResult( ades_subindic=AdESIndeterminate.SIG_CONSTRAINTS_FAILURE, api_status=temp_status, failure_msg="No signature timestamp present", best_signature_time=timing_info.best_signature_time, signature_not_before_time=signature_not_before_time, validation_objects=vos, ) sig_ts_result = await _ades_process_attached_ts( signer_info, validation_context, signed=False, tst_digest=tst_digest ) vos = ValidationObjectSet( _enumerate_validation_objects(validation_context), _enumerate_validation_objects(ts_validation_context), _enumerate_validation_objects(ac_validation_context), _enumerate_certs_in_paths(interm_result), _enumerate_certs_in_paths(sig_ts_result.api_status), ) if sig_ts_result.ades_subindic != AdESPassed.OK: return AdESWithTimeValidationResult( ades_subindic=sig_ts_result.ades_subindic, api_status=temp_status, failure_msg=None, best_signature_time=signature_poe_time, signature_not_before_time=signature_not_before_time, validation_objects=vos, ) ts_status = sig_ts_result.api_status # note: we should always have a non-null status if the validation passes, assert isinstance(ts_status, TimestampSignatureStatus) if signature_poe_time is not None: signature_poe_time = min(ts_status.timestamp, signature_poe_time) else: signature_poe_time = ts_status.timestamp interm_result.signature_ts_validity = ts_status interm_result.signature_poe_time = signature_poe_time if interm_result.ades_subindic == AdESIndeterminate.REVOKED_NO_POE: revo_details: RevocationDetails = temp_status.revocation_details if signature_poe_time >= revo_details.revocation_date: # nothing we can do return AdESWithTimeValidationResult( ades_subindic=interm_result.ades_subindic, api_status=temp_status, failure_msg=None, best_signature_time=signature_poe_time, signature_not_before_time=signature_not_before_time, validation_objects=vos, ) elif interm_result.ades_subindic == AdESIndeterminate.OUT_OF_BOUNDS_NO_POE: # NOTE: we can't process expiration here since we don't have access # to _timestamped_ revocation information if signature_poe_time < temp_status.signing_cert.not_valid_before: # FIXME replace temp_status as well return AdESWithTimeValidationResult( ades_subindic=AdESIndeterminate.NOT_YET_VALID, api_status=temp_status, failure_msg=None, best_signature_time=signature_poe_time, signature_not_before_time=signature_not_before_time, validation_objects=vos, ) elif ( interm_result.ades_subindic == AdESIndeterminate.CRYPTO_CONSTRAINTS_FAILURE_NO_POE or interm_result.ades_subindic == AdESIndeterminate.TRY_LATER ): # in this case error_time_horizon is set to the point where the # constraint was triggered if signature_poe_time >= temp_status.error_time_horizon: return AdESWithTimeValidationResult( ades_subindic=interm_result.ades_subindic, api_status=temp_status, failure_msg=None, best_signature_time=signature_poe_time, signature_not_before_time=signature_not_before_time, validation_objects=vos, ) # TODO TSTInfo ordering/comparison check if ( signature_not_before_time is not None and signature_not_before_time > signature_poe_time ): return AdESWithTimeValidationResult( ades_subindic=AdESIndeterminate.TIMESTAMP_ORDER_FAILURE, api_status=temp_status, failure_msg=None, best_signature_time=signature_poe_time, signature_not_before_time=signature_not_before_time, validation_objects=vos, ) # TODO handle time-stamp delay interm_result.trust_subindic_update = None interm_result.status_kwargs['trust_problem_indic'] = None status = interm_result.update(status_cls, with_ts=True, with_attrs=True) return AdESWithTimeValidationResult( ades_subindic=AdESPassed.OK, api_status=status, failure_msg=None, best_signature_time=signature_poe_time, signature_not_before_time=signature_not_before_time, validation_objects=vos, ) class _TrustNoOne(TrustManager): def is_root(self, cert: x509.Certificate) -> bool: return False def find_potential_issuers( self, cert: x509.Certificate ) -> Iterator[TrustAnchor]: return iter(()) def _crl_issuer_cert_poe_boundary( crl: CRLOfInterest, cutoff: datetime, poe_manager: POEManager ): return any( poe_manager[prov_path.path.leaf] <= cutoff for prov_path in crl.prov_paths ) def _ocsp_issuer_cert_poe_boundary( ocsp: OCSPResponseOfInterest, cutoff: datetime, poe_manager: POEManager ): return poe_manager[ocsp.prov_path.leaf] <= cutoff async def _find_revinfo_data_for_leaf_in_past( cert: x509.Certificate, validation_data_handlers: ValidationDataHandlers, control_time: datetime, revocation_checking_rule: RevocationCheckingRule, ): # Need to find a piece of revinfo for the signing cert, for which we have # POE for the issuer cert, which must be dated before the expiration date of # the cert. (Standard is unclear as to which cert this refers to, but # it's probably the date on the signing cert. Not much point in requiring # PoE for a cert before its self-declared expiration date...) # Since our revinfo gathering logic is based on paths, we gather up all # candidate issuers and work with those "truncated" candidate paths. # Trust is not an issue at this stage. registry = validation_data_handlers.cert_registry candidate_issuers = registry.find_potential_issuers( cert=cert, trust_manager=_TrustNoOne() ) def _for_candidate_issuer(iss: x509.Certificate): truncated_path = ValidationPath( trust_anchor=CertTrustAnchor(iss), interm=[], leaf=cert ) return ades_gather_prima_facie_revinfo( path=truncated_path, revinfo_manager=validation_data_handlers.revinfo_manager, control_time=control_time, revocation_checking_rule=revocation_checking_rule, ) job_futures = asyncio.as_completed( [_for_candidate_issuer(iss) for iss in candidate_issuers] ) poe_manager = validation_data_handlers.poe_manager crls: List[CRLOfInterest] = [] ocsps: List[OCSPResponseOfInterest] = [] new_crls: Iterable[CRLOfInterest] new_ocsps: Iterable[OCSPResponseOfInterest] to_evict: Set[bytes] = set() for fut_results in job_futures: new_crls, new_ocsps = await fut_results # Collect the revinfos for which we have POE for the issuer cert # predating the expiration of the signer cert for crl_oi in new_crls: if _crl_issuer_cert_poe_boundary( crl_oi, cert.not_valid_after, poe_manager ): crls.append(crl_oi) else: revinfo_data = crl_oi.crl.crl_data.dump() to_evict.add(digest_for_poe(revinfo_data)) for ocsp_oi in new_ocsps: if _ocsp_issuer_cert_poe_boundary( ocsp_oi, cert.not_valid_after, poe_manager ): ocsps.append(ocsp_oi) else: revinfo_data = ocsp_oi.ocsp_response.ocsp_response_data.dump() to_evict.add(digest_for_poe(revinfo_data)) # we only run the eviction logic if we found at least one piece of revinfo # that we can actually use (that's what the spec says, shouldn't change # validation result, but the reported error probably makes more sense) if crls or ocsps: validation_data_handlers.revinfo_manager.evict_crls(to_evict) validation_data_handlers.revinfo_manager.evict_ocsps(to_evict) return crls, ocsps async def _build_and_past_validate_cert( cert: x509.Certificate, validation_policy_spec: CertValidationPolicySpec, validation_data_handlers: ValidationDataHandlers, ) -> Tuple[ValidationPath, datetime]: path_builder = PathBuilder( trust_manager=validation_policy_spec.trust_manager, registry=validation_data_handlers.cert_registry, ) current_subindication = None paths = path_builder.async_build_paths_lazy(cert) try: async for cand_path in paths: past_result: generic_cms.CertvalidatorOperationResult[datetime] past_result = await generic_cms.handle_certvalidator_errors( past_validate( path=cand_path, validation_policy_spec=validation_policy_spec, validation_data_handlers=validation_data_handlers, init_control_time=None, ) ) current_subindication = past_result.error_subindic validation_time = past_result.success_result if current_subindication is None: assert validation_time is not None return cand_path, validation_time finally: await paths.cancel() msg = "Unable to construct plausible past validation path" if current_subindication is not None: raise errors.SignatureValidationError( failure_message=msg, ades_subindication=current_subindication ) else: raise errors.SignatureValidationError( failure_message=f"{msg}: no prima facie paths constructed", ades_subindication=AdESIndeterminate.NO_CERTIFICATE_CHAIN_FOUND, ) async def _ades_past_signature_validation( signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, poe_manager: POEManager, current_time_sub_indic: Optional[AdESIndeterminate], init_control_time: datetime, is_timestamp: bool, ) -> ValidationPath: validation_data_handlers = bootstrap_validation_data_handlers( validation_spec, is_historical=True, poe_manager_override=poe_manager ) signature_bytes = signed_data['signer_infos'][0]['signature'].native best_signature_time = poe_manager[signature_bytes] try: cert_info = extract_certificate_info(signed_data) cert = cert_info.signer_cert validation_data_handlers.cert_registry.register_multiple( cert_info.other_certs ) except CMSExtractionError: raise errors.SignatureValidationError( 'signer certificate not included in signature', ades_subindication=AdESIndeterminate.NO_SIGNING_CERTIFICATE_FOUND, ) if is_timestamp: cert_validation_policy = ( validation_spec.ts_cert_validation_policy or validation_spec.cert_validation_policy ) else: cert_validation_policy = validation_spec.cert_validation_policy leaf_crls, leaf_ocsps = await _find_revinfo_data_for_leaf_in_past( cert, validation_data_handlers, control_time=init_control_time, revocation_checking_rule=( cert_validation_policy.revinfo_policy.revocation_checking_policy.ee_certificate_rule ), ) # Key usage for the signer is not something that varies over time, so # we delegate that to the caller. This is justified both because it's # technically simpler, and because the past signature validation block # in AdES is predicated on delegating the basic integrity checks anyhow. cert_path, validation_time = await _build_and_past_validate_cert( cert, validation_policy_spec=cert_validation_policy, validation_data_handlers=validation_data_handlers, ) # TODO revisit this once I have a clearer understanding of why this PoE # issuance check is only applied to the EE cert. def _pass_contingent_on_revinfo_issuance_poe(): if not bool(leaf_crls or leaf_ocsps): status = AdESIndeterminate.REVOCATION_OUT_OF_BOUNDS_NO_POE raise errors.SignatureValidationError( failure_message=( "POE for signature available, but could not obtain " "sufficient POE for the issuance of the " "revocation information", ), ades_subindication=status, ) if best_signature_time <= validation_time: # TODO raise an issue with ESI about TRY_LATER here if ( current_time_sub_indic == AdESIndeterminate.REVOKED_NO_POE or current_time_sub_indic == AdESIndeterminate.TRY_LATER ): _pass_contingent_on_revinfo_issuance_poe() return cert_path elif current_time_sub_indic in ( AdESIndeterminate.REVOKED_CA_NO_POE, AdESIndeterminate.CRYPTO_CONSTRAINTS_FAILURE_NO_POE, ): # This is an automatic pass given that certvalidator checks # these conditions for us as part of past_validate(...) return cert_path elif current_time_sub_indic in ( AdESIndeterminate.OUT_OF_BOUNDS_NO_POE, AdESIndeterminate.OUT_OF_BOUNDS_NOT_REVOKED, ): if best_signature_time < cert.not_valid_before: raise errors.SignatureValidationError( failure_message="Signature predates cert validity period", ades_subindication=AdESFailure.NOT_YET_VALID, ) elif best_signature_time <= cert.not_valid_after: _pass_contingent_on_revinfo_issuance_poe() return cert_path # TODO here, it would help to preserve more than the sub-indication # from before raise errors.SigSeedValueValidationError( failure_message=( "Past signature validation did not manage " "to improve current time result." ), ades_subindication=current_time_sub_indic, ) async def ades_past_signature_validation( signed_data: cms.SignedData, validation_spec: SignatureValidationSpec, poe_manager: POEManager, current_time_sub_indic: Optional[AdESIndeterminate], init_control_time: Optional[datetime] = None, ) -> AdESSubIndic: """ Validate a CMS signature in the past according to ETSI EN 319 102-1 § 5.6.2.4. This is internal API. .. danger:: The notion of "past validation" used here is only valid in the narrow technical sense in which it is used within AdES. It should _never_ be relied upon as a standalone validation routine. :param signed_data: The ``SignedData`` value. :param validation_spec: Validation settings to apply. :param poe_manager: The POE manager from which to source existence proofs. :param current_time_sub_indic: The AdES subindication from validating the signature at the current time with the relevant settings. :param init_control_time: Initial value for the control time parameter. :return: An AdES subindication indicating the validation result after going through the past validation process. """ eci = signed_data['encap_content_info'] is_timestamp = eci['content_type'].native == 'tst_info' if init_control_time is None: init_control_time = datetime.now(tz=tzlocal.get_localzone()) try: await _ades_past_signature_validation( signed_data=signed_data, validation_spec=validation_spec, poe_manager=poe_manager, current_time_sub_indic=current_time_sub_indic, init_control_time=init_control_time, is_timestamp=is_timestamp, ) return AdESPassed.OK except errors.SignatureValidationError as e: logger.warning(e) return e.ades_subindication or AdESIndeterminate.GENERIC except PathError as e: logger.warning(e) return AdESIndeterminate.CERTIFICATE_CHAIN_GENERAL_FAILURE @dataclass(frozen=True) class _PrimaFaciePOEItem: digest: bytes validation_object: ValidationObject @dataclass(frozen=True) class _PrimaFaciePOEFromTimeStamp: pdf_revision: int timestamp_dt: datetime poes_implied: FrozenSet[_PrimaFaciePOEItem] timestamp_token_signed_data: cms.SignedData doc_digest: bytes # include info from difference analysis as part of the status kwargs forensic_info: dict def add_to_poe_manager(self, manager: POEManager): for thing in self.poes_implied: manager.register_known_poe( KnownPOE( poe_type=POEType.VALIDATION, digest=thing.digest, poe_time=self.timestamp_dt, validation_object=thing.validation_object, ) ) def _extract_cert_digests_from_signed_data( sd: cms.SignedData, ) -> Generator[_PrimaFaciePOEItem, None, None]: cert_choice: cms.CertificateChoices for cert_choice in sd['certificates']: obj = cert_choice.chosen data = obj.dump() if cert_choice.name == 'certificate': vo_type = ValidationObjectType.CERTIFICATE elif cert_choice.name == 'v2_attr_cert': # There is no separate type for an attribute cert in # ETSI TS 119 102-2, so we mark it as OTHER. vo_type = ValidationObjectType.OTHER else: # skip over unsupported certificate types # since we don't want to give the impression # in the validation report that we processed them. # TODO write test to verify that these don't end up in the report continue digest = digest_for_poe(data) yield _PrimaFaciePOEItem( digest=digest, validation_object=ValidationObject(object_type=vo_type, value=obj), ) def _get_tst_timestamp(sd: cms.SignedData) -> datetime: tst_info: tsp.TSTInfo = sd['encap_content_info']['content'].parsed return tst_info['gen_time'].native def _read_validation_objects_from_revinfo_archival( revinfo_archival: asn1_pdf.RevocationInfoArchival, ) -> Generator[_PrimaFaciePOEItem, None, None]: for crl in revinfo_archival['crl']: yield _PrimaFaciePOEItem( digest=digest_for_poe(crl.dump()), validation_object=ValidationObject( object_type=ValidationObjectType.CRL, value=CRLContainer(crl), ), ) for ocsp in revinfo_archival['ocsp']: yield _PrimaFaciePOEItem( digest=digest_for_poe(ocsp.dump()), validation_object=ValidationObject( object_type=ValidationObjectType.OCSP_RESPONSE, value=OCSPContainer(ocsp), ), ) def _read_validation_objects_from_dss( dss: DocumentSecurityStore, ) -> Generator[_PrimaFaciePOEItem, None, None]: for crl_obj in dss.crls: data = crl_obj.get_object().data yield _PrimaFaciePOEItem( digest=digest_for_poe(data), validation_object=ValidationObject( object_type=ValidationObjectType.CRL, value=CRLContainer(CertificateList.load(data)), ), ) for ocsp_obj in dss.ocsps: data = ocsp_obj.get_object().data yield _PrimaFaciePOEItem( digest=digest_for_poe(data), validation_object=ValidationObject( object_type=ValidationObjectType.OCSP_RESPONSE, value=OCSPContainer(OCSPResponse.load(data)), ), ) for cert_obj in dss.certs.values(): data = cert_obj.get_object().data yield _PrimaFaciePOEItem( digest=digest_for_poe(data), validation_object=ValidationObject( object_type=ValidationObjectType.CERTIFICATE, value=x509.Certificate.load(data), ), ) def _build_prima_facie_poe_index_from_pdf_timestamps( r: PdfFileReader, include_content_ts: bool, diff_policy: Optional[DiffPolicy], ) -> List[_PrimaFaciePOEFromTimeStamp]: # This subroutine implements the POE gathering part of the evidence record # processing algorithm in AdES as applied to PDF. For the purposes of this # function, the chain of document timestamps is treated as a single evidence # record, and all document data in the revision in which a timestamp is # contained is considered fair game. # Signature timestamps are not processed as such, but POE for the timestamps # themselves will be accumulated. # Content timestamps can optionally be included. This is not standard # in AdES, but since there's no cryptographic difference (in PDF!) between # a content TS in a signature and a document timestamp signature, they # can be taken into account at the caller's discretion # TODO take algorithm usage policy into account? # TODO when ingesting OCSP responses, make an effort to register # POE for the embedded certs as well? Esp. potential responder certs. # timestamp -> hashes index. We haven't validated the chain of trust # of the timestamps yet, so we can't put them in an actual # POE manager immediately # Since the embedded signature context is necessary to validate the POE's # integrity, we do run the integrity checker for the TST data at this stage. # The actual trust validation is delegated collected_so_far: Set[_PrimaFaciePOEItem] = set() # Holds all digests of objects contained in _document_ content so far # (note: this is why it's important to traverse the revisions in order) for_next_ts: Set[_PrimaFaciePOEItem] = set() # Holds digests of objects that will be registered with POE on the next # document TS or content TS encountered. prima_facie_poe_sets: List[_PrimaFaciePOEFromTimeStamp] = [] # output array (to avoid having to work with async generators) embedded_sig: EmbeddedPdfSignature for ix, embedded_sig in enumerate(r.embedded_signatures): embedded_sig.compute_integrity_info( # different None handling convention diff_policy, skip_diff=diff_policy is None, ) hist_handler = HistoricalResolver( r, revision=embedded_sig.signed_revision ) signed_data: cms.SignedData = embedded_sig.signed_data ts_signed_data: Optional[cms.SignedData] = None is_doc_ts = False if embedded_sig.sig_object_type == '/DocTimeStamp': ts_signed_data = signed_data is_doc_ts = True elif include_content_ts: ts_signed_data = generic_cms.extract_tst_data( embedded_sig.signer_info, signed=True ) # Important remark: at this time, we do NOT consider signature # timestamps when evaluating POE data, only content timestamps & # document timestamps! # Rationale: the signature timestamp only indirectly protects # the document content, and wasn't designed for this purpose. # If we want to use signature TSes as well, we'd have to evaluate # the integrity of the signature, which requires selecting a certificate # (even if just for validation purposes), yada yada. Not doing any of # that for now. # (This approach might change in the future) if ts_signed_data is not None: # add DSS content try: dss = DocumentSecurityStore.read_dss(hist_handler) collected_so_far.update(_read_validation_objects_from_dss(dss)) except NoDSSFoundError: pass collected_so_far.update(for_next_ts) doc_digest = embedded_sig.compute_digest() coverage_normal = ( embedded_sig.evaluate_signature_coverage() >= SignatureCoverageLevel.ENTIRE_REVISION ) if coverage_normal: prima_facie_poe_sets.append( _PrimaFaciePOEFromTimeStamp( pdf_revision=embedded_sig.signed_revision, timestamp_dt=_get_tst_timestamp(ts_signed_data), poes_implied=frozenset(collected_so_far), timestamp_token_signed_data=ts_signed_data, doc_digest=doc_digest, forensic_info=embedded_sig.summarise_integrity_info(), ) ) # reset for_next_ts for_next_ts = set() for_next_ts.update( _extract_cert_digests_from_signed_data(ts_signed_data) ) # the certs in the signature container itself are not part of the # signed data in that revision, but they're covered # by whatever the next (content) TS covers -> keep 'em for_next_ts.update(_extract_cert_digests_from_signed_data(signed_data)) # same for revinfo embedded Adobe-style: # part of the signed data, but not directly timestamped # => save for next TS signed_attrs = embedded_sig.signer_info['signed_attrs'] if not is_doc_ts: try: revinfo_attr: asn1_pdf.RevocationInfoArchival = ( find_unique_cms_attribute( signed_attrs, 'adobe_revocation_info_archival' ) ) for_next_ts.update( _read_validation_objects_from_revinfo_archival(revinfo_attr) ) except (MultivaluedAttributeError, NonexistentAttributeError): pass # Prepare a POE entry for the signature itself (to be processed # with the next timestamp) sig_bytes = embedded_sig.signer_info['signature'].native for_next_ts.add( _PrimaFaciePOEItem( digest=digest_for_poe(sig_bytes), validation_object=ValidationObject( object_type=ValidationObjectType.SIGNED_DATA, # For now, we put the entire signed data object here # while we take the digest only over the signature. # This was done for expediency & ease of reasoning # given existing code, but may change in the future. value=embedded_sig.signed_data, ), ) ) # add POE entries for the timestamp(s) attached to this signature try: content_tses = find_cms_attribute( signed_attrs, 'content_time_stamp' ) except (NonexistentAttributeError, CMSStructuralError): content_tses = () try: signature_tses = find_cms_attribute( embedded_sig.signer_info['unsigned_attrs'], 'signature_time_stamp', ) except (NonexistentAttributeError, CMSStructuralError): signature_tses = () for ts_data in itertools.chain(signature_tses, content_tses): ts_data_content = ts_data['content'] for ts_signer_info in ts_data_content['signer_infos']: ts_sig_bytes = ts_signer_info['signature'].native for_next_ts.add( _PrimaFaciePOEItem( digest=digest_for_poe(ts_sig_bytes), # Same as for signedData: we take the digest over # the signature part only. # This was done for expediency & ease of reasoning # given existing code, but may change in the future. validation_object=ValidationObject( object_type=ValidationObjectType.TIMESTAMP, value=ts_data_content, ), ) ) return prima_facie_poe_sets async def _validate_prima_facie_poe( prima_facie_poe_sets: List[_PrimaFaciePOEFromTimeStamp], # we assume that the validation info extracted from the DSS # has been registered in the revinfo gathering policy object # and/or the known cert list, respectively validation_spec: SignatureValidationSpec, cur_timing_info: Optional[ValidationTimingInfo] = None, ) -> POEManager: # Sort by PDF revision, but in ascending order (!) # This is a consequence of the way the ER validation algorithm works candidate_poes = sorted(prima_facie_poe_sets, key=lambda p: p.pdf_revision) cur_timing_info = cur_timing_info or ValidationTimingInfo.now( tz=tzlocal.get_localzone() ) resulting_poes = POEManager() validation_spec.local_knowledge.add_to_poe_manager(resulting_poes) for ix, poe in enumerate(candidate_poes): temporary_poes = copy(resulting_poes) if ix < len(candidate_poes) - 1: # perform temp POE initialisation as the AdES spec requires next_poe = candidate_poes[ix + 1] next_poe.add_to_poe_manager(temporary_poes) validation_data_handlers = bootstrap_validation_data_handlers( validation_spec, timing_info=cur_timing_info, poe_manager_override=temporary_poes, ) cur_time_result = await ades_timestamp_validation( tst_signed_data=poe.timestamp_token_signed_data, validation_spec=validation_spec, timing_info=cur_timing_info, expected_tst_imprint=poe.doc_digest, validation_data_handlers=validation_data_handlers, extra_status_kwargs=poe.forensic_info, status_cls=DocumentTimestampStatus, ) sub_indic = cur_time_result.ades_subindic if sub_indic.status == AdESStatus.PASSED: # still valid at current time => ok # (AdES spec on ER validation is unclear about this, but this # should mean that we can skip the past validation block) poe.add_to_poe_manager(resulting_poes) elif sub_indic.status == AdESStatus.FAILED: # TODO more informative reporting? raise errors.SignatureValidationError( "Permanent failure while evaluating timestamp in PoE chain", ades_subindication=sub_indic, ) else: # neither pass nor fail => try past validation procedure assert isinstance(sub_indic, AdESIndeterminate) past_result = await ades_past_signature_validation( signed_data=poe.timestamp_token_signed_data, validation_spec=validation_spec, poe_manager=temporary_poes, current_time_sub_indic=sub_indic, init_control_time=cur_timing_info.validation_time, ) if past_result.status == AdESStatus.PASSED: poe.add_to_poe_manager(resulting_poes) else: raise errors.SignatureValidationError( "Could not validate timestamp in PoE chain at current " "time, and past validation also failed", ades_subindication=sub_indic, ) return resulting_poes _LTA_FURTHER_PROC = frozenset( { AdESPassed.OK, AdESIndeterminate.REVOKED_NO_POE, AdESIndeterminate.REVOKED_CA_NO_POE, AdESIndeterminate.OUT_OF_BOUNDS_NO_POE, AdESIndeterminate.OUT_OF_BOUNDS_NOT_REVOKED, AdESIndeterminate.CRYPTO_CONSTRAINTS_FAILURE_NO_POE, AdESIndeterminate.REVOCATION_OUT_OF_BOUNDS_NO_POE, AdESIndeterminate.TRY_LATER, } ) _LTA_TS_FURTHER_PROC = frozenset( { AdESIndeterminate.REVOKED_CA_NO_POE, AdESIndeterminate.OUT_OF_BOUNDS_NO_POE, AdESIndeterminate.OUT_OF_BOUNDS_NOT_REVOKED, AdESIndeterminate.CRYPTO_CONSTRAINTS_FAILURE_NO_POE, AdESIndeterminate.REVOCATION_OUT_OF_BOUNDS_NO_POE, } ) @dataclass(frozen=True) class AdESLTAValidationResult(AdESWithTimeValidationResult): """ Result of a PAdES validation for a signature providing long-term availability and integrity of validation material. See ETSI EN 319 102-1, § 5.6.3. """ oldest_evidence_record_timestamp: Optional[datetime] """ The oldest timestamp in the evidence record, after validation. .. note:: For PAdES, this refers to the chain of document timestamp signatures after signing. """ signature_timestamp_status: Optional[AdESBasicValidationResult] """ The validation result for the signature time stamp, if applicable. """ async def _process_signature_ts( embedded_sig: EmbeddedPdfSignature, validation_spec: SignatureValidationSpec, poe_manager: POEManager, timing_info: ValidationTimingInfo, ) -> Optional[AdESBasicValidationResult]: signature_bytes = embedded_sig.signer_info['signature'].native signature_ts: cms.SignedData = embedded_sig.attached_timestamp_data cert_validation_policy = ( validation_spec.ts_cert_validation_policy or validation_spec.cert_validation_policy ) algo_policy = cert_validation_policy.algorithm_usage_policy if signature_ts is None: return None expected_tst_imprint = embedded_sig.compute_tst_digest() assert expected_tst_imprint is not None signature_ts_prelim_result = await ades_timestamp_validation( tst_signed_data=signature_ts, validation_spec=validation_spec, timing_info=timing_info, expected_tst_imprint=expected_tst_imprint, validation_data_handlers=bootstrap_validation_data_handlers( validation_spec, timing_info=timing_info, poe_manager_override=poe_manager, ), ) ts_current_time_sub_indic = signature_ts_prelim_result.ades_subindic signature_ts_result: AdESBasicValidationResult if ( isinstance(ts_current_time_sub_indic, AdESIndeterminate) and ts_current_time_sub_indic in _LTA_TS_FURTHER_PROC ): try: # TODO: in principle, we should also run this if the status is # PASSED already. Ensure that that is possible. validation_path = await _ades_past_signature_validation( signed_data=signature_ts, validation_spec=validation_spec, poe_manager=poe_manager, current_time_sub_indic=ts_current_time_sub_indic, init_control_time=timing_info.validation_time, is_timestamp=True, ) signature_ts_result = AdESBasicValidationResult( ades_subindic=AdESPassed.OK, # TODO update pyHanko status object as well api_status=( dataclasses.replace( signature_ts_prelim_result.api_status, validation_path=validation_path, ) if signature_ts_prelim_result.api_status else None ), failure_msg=None, validation_objects=signature_ts_prelim_result.validation_objects, ) except errors.SignatureValidationError as e: signature_ts_result = AdESBasicValidationResult( ades_subindic=e.ades_subindication or ts_current_time_sub_indic, failure_msg=e.failure_message, api_status=signature_ts_prelim_result.api_status, validation_objects=signature_ts_prelim_result.validation_objects, ) else: signature_ts_result = signature_ts_prelim_result tst_info = signature_ts['encap_content_info']['content'].parsed if algo_policy is not None and algo_policy.digest_algorithm_allowed( tst_info['message_imprint']['hash_algorithm'], moment=timing_info.validation_time, ): signature_ts_dt = tst_info['gen_time'].native poe_manager.register(signature_bytes, signature_ts_dt) # TODO if/when we fully support signature policies, we should check # whether the policy requires a valid signature timestamp return signature_ts_result def _dss_to_local_knowledge( reader: PdfFileReader, ): try: dss = DocumentSecurityStore.read_dss(reader) dss_ocsps = [ cont for resp in dss.ocsps for cont in OCSPContainer.load_multi( OCSPResponse.load(resp.get_object().data) ) ] dss_crls = [ CRLContainer(crl_data=CertificateList.load(crl.get_object().data)) for crl in dss.crls ] dss_certs = list(dss.load_certs()) local_knowledge = LocalKnowledge( known_ocsps=dss_ocsps, known_crls=dss_crls, known_certs=dss_certs, ) except NoDSSFoundError: local_knowledge = LocalKnowledge() return local_knowledge async def ades_lta_validation( embedded_sig: EmbeddedPdfSignature, pdf_validation_spec: PdfSignatureValidationSpec, timing_info: Optional[ValidationTimingInfo] = None, signature_not_before_time: Optional[datetime] = None, ) -> AdESLTAValidationResult: """ Validate a PAdES signature providing long-term availability and integrity of validation material. See ETSI EN 319 102-1, § 5.6.3. For the purposes of PAdES validation, the chain of document time stamps in the document serves as the unique Evidence Record (ER). :param embedded_sig: The PDF signature to validate. :param pdf_validation_spec: PDF signature validation settings. :param timing_info: Data object describing the timing of the validation. Defaults to :meth:`.ValidationTimingInfo.now`. :param signature_not_before_time: Time when the signature was known _not_ to exist. :return: A validation result. """ timing_info = timing_info or ValidationTimingInfo.now( tz=tzlocal.get_localzone() ) # (1) process DocTSes as ER poe_list = _build_prima_facie_poe_index_from_pdf_timestamps( embedded_sig.reader, include_content_ts=True, diff_policy=pdf_validation_spec.diff_policy, ) validation_spec = pdf_validation_spec.signature_validation_spec init_local_knowledge = validation_spec.local_knowledge # Ingest CRLs, certs and OCSPs from the DSS # (POE info will be processed separately) dss_facts = _dss_to_local_knowledge(reader=embedded_sig.reader) local_knowledge = LocalKnowledge( known_ocsps=init_local_knowledge.known_ocsps + dss_facts.known_ocsps, known_crls=init_local_knowledge.known_crls + dss_facts.known_crls, known_certs=init_local_knowledge.known_certs + dss_facts.known_certs, known_poes=init_local_knowledge.known_poes, nonrevoked_assertions=init_local_knowledge.nonrevoked_assertions, ) augmented_validation_spec = dataclasses.replace( validation_spec, local_knowledge=local_knowledge ) updated_poe_manager = None oldest_evidence_record_timestamp = None try: updated_poe_manager = await _validate_prima_facie_poe( poe_list, validation_spec=augmented_validation_spec, cur_timing_info=timing_info, ) # The POE list has been validated at this point, # so we just pick out the oldest one oldest_docts_record = min( filter( lambda poe: poe.pdf_revision > embedded_sig.signed_revision, poe_list, ), key=lambda poe: poe.pdf_revision, default=None, ) if oldest_docts_record is not None: oldest_evidence_record_timestamp = oldest_docts_record.timestamp_dt elif not local_knowledge.known_poes: # do not show this warning if there are POEs in the local knowledge logger.warning( "No document timestamps after signature; proceeding " "without past proof of existence" ) except errors.SignatureValidationError as e: logger.warning( "Document timestamp chain failed to validate; proceeding " "without past proof of existence.", exc_info=e, ) if oldest_evidence_record_timestamp is None: updated_poe_manager = POEManager() local_knowledge.add_to_poe_manager(updated_poe_manager) assert updated_poe_manager is not None # (2) skipped, is automatic in our implementation # (3) Run validation for signatures with time with_time_data_handlers = bootstrap_validation_data_handlers( spec=augmented_validation_spec, timing_info=timing_info, poe_manager_override=copy(updated_poe_manager), ) signature_prelim_result = await ades_with_time_validation( signed_data=embedded_sig.signed_data, validation_spec=augmented_validation_spec, timing_info=timing_info, validation_data_handlers=with_time_data_handlers, raw_digest=embedded_sig.compute_digest(), signature_not_before_time=signature_not_before_time, extra_status_kwargs=embedded_sig.summarise_integrity_info(), status_cls=PdfSignatureStatus, ) # don't branch on policy here, we always continue as if archival info # is present current_time_sub_indic = signature_prelim_result.ades_subindic failure_msg: Optional[str] if current_time_sub_indic not in _LTA_FURTHER_PROC: failure_msg = ( "Validation of signature at current time failed with " f"indication {current_time_sub_indic}. Past validation not " f"applicable." ) return AdESLTAValidationResult( ades_subindic=current_time_sub_indic, api_status=signature_prelim_result.api_status, failure_msg=failure_msg, best_signature_time=signature_prelim_result.best_signature_time, signature_not_before_time=( signature_prelim_result.signature_not_before_time ), oldest_evidence_record_timestamp=oldest_evidence_record_timestamp, signature_timestamp_status=None, validation_objects=signature_prelim_result.validation_objects, ) # (4) Register PoE for the signature based on best_signature_time signature_bytes = embedded_sig.signer_info['signature'].native updated_poe_manager.register( signature_bytes, poe_type=POEType.VALIDATION, dt=signature_prelim_result.best_signature_time, ) # (5) process signature TS if present signature_ts_result = await _process_signature_ts( embedded_sig, validation_spec=augmented_validation_spec, poe_manager=copy(updated_poe_manager), timing_info=timing_info, ) # (6) past signature validation if isinstance(current_time_sub_indic, AdESIndeterminate): # TODO: in principle, we should also run this if the status is PASSED # already. Ensure that that is possible. past_sig_poe_manager = copy(updated_poe_manager) try: await _ades_past_signature_validation( signed_data=embedded_sig.signed_data, validation_spec=augmented_validation_spec, poe_manager=past_sig_poe_manager, current_time_sub_indic=current_time_sub_indic, init_control_time=timing_info.validation_time, is_timestamp=False, ) updated_poe_manager = past_sig_poe_manager except errors.SignatureValidationError as e: sig_poe = past_sig_poe_manager[signature_bytes] return AdESLTAValidationResult( ades_subindic=e.ades_subindication or current_time_sub_indic, failure_msg=e.failure_message, # FIXME rewrite api_status! api_status=signature_prelim_result.api_status, best_signature_time=sig_poe, signature_not_before_time=( signature_prelim_result.signature_not_before_time ), signature_timestamp_status=signature_ts_result, oldest_evidence_record_timestamp=( oldest_evidence_record_timestamp ), validation_objects=signature_prelim_result.validation_objects, ) # (7) get the oldest PoE for the signature signature_poe_time = updated_poe_manager[signature_bytes] # (8) perform SVA (=> only crypto checks) algo_policy = ( augmented_validation_spec.cert_validation_policy.algorithm_usage_policy ) ades_subindic: AdESSubIndic try: cert: x509.Certificate = embedded_sig.signer_cert if algo_policy is not None: _ades_signature_crypto_policy_check( embedded_sig.signer_info, algo_policy=algo_policy, control_time=signature_poe_time, public_key=cert.public_key, ) ades_subindic = AdESPassed.OK failure_msg = None except errors.SignatureValidationError as e: ades_subindic = e.ades_subindication or current_time_sub_indic failure_msg = e.failure_message return AdESLTAValidationResult( ades_subindic=ades_subindic, api_status=signature_prelim_result.api_status, failure_msg=failure_msg, best_signature_time=signature_poe_time, signature_not_before_time=( signature_prelim_result.signature_not_before_time ), signature_timestamp_status=signature_ts_result, oldest_evidence_record_timestamp=oldest_evidence_record_timestamp, validation_objects=signature_prelim_result.validation_objects, ) async def simulate_future_ades_lta_validation( embedded_sig: EmbeddedPdfSignature, pdf_validation_spec: PdfSignatureValidationSpec, future_validation_time: datetime, current_reference_time: Optional[datetime] = None, ) -> AdESLTAValidationResult: """ .. versionadded:: 0.21.0 Simulate a future LTA validation of a PDF signature, assuming perfect timestamp maintenance until the specified point in time. .. warning:: This is experimental API. The purpose of this utility function is to act as a sanity check for signers and signature archivists. It takes validation spec, a future validation time and a current reference time (defaults to the current time), and, by fiat, generates proofs of existence for all relevant objects in the PDF for that reference time. It then executes the PAdES LTA validation algorithm with that set of PoEs against the future validation time, with all remote fetching functionality disabled. The idea is that this allows the caller to assess whether a signature is "LTA maintainable", i.e. whether it contains the necessary information for the signature to remain validatable if the timestamp chain is extended properly. If this check fails but the signature validates at the current time, it may indicate a lack of contemporaneous revocation information. :param embedded_sig: The signature under scrutiny. :param pdf_validation_spec: The validation spec against which the simulated validation should be executed. :param future_validation_time: The future validation time at which the validation should be simulated. :param current_reference_time: The reference time at which all relevant objects in the PDF are presumed to have been proven to exist for the purposes of the (future) validation being simulated. Defaults to the current time. :return: An AdES LTA validation result. """ now = current_reference_time or datetime.now(tz=timezone.utc) timing_info = ValidationTimingInfo( validation_time=future_validation_time, point_in_time_validation=True, best_signature_time=future_validation_time, ) prima_facie_poes = _build_prima_facie_poe_index_from_pdf_timestamps( embedded_sig.reader, include_content_ts=True, diff_policy=None ) orig_sig_validation_spec = pdf_validation_spec.signature_validation_spec orig_local_knowledge = orig_sig_validation_spec.local_knowledge dss_knowledge = _dss_to_local_knowledge(embedded_sig.reader) new_nonrevoked_assertions = list(orig_local_knowledge.nonrevoked_assertions) # assert the nonrevoked status of the last timestamp cert, since we can't # get "future" revinfo anyway try: last_ts = embedded_sig.reader.embedded_timestamp_signatures[-1] new_nonrevoked_assertions.append( NonRevokedStatusAssertion( last_ts.signer_cert.sha256, at=future_validation_time ) ) except IndexError: pass def _poes(): # For the purposes of this test, we assert all proofs of existence # at the current time, including all the prima facie ones gathered from # the file. This simulates perfect record keeping without having to # introduce extra timestamp tokens into the validation process. yield from orig_local_knowledge.assert_existence_known_at(now) yield from dss_knowledge.assert_existence_known_at(now) for prima_facie_poe in prima_facie_poes: for item in prima_facie_poe.poes_implied: # for the prima facie ones, we only emit POEs for , since # we don't validate the would-be POEs that are embedded in the # document at this point yield KnownPOE( poe_type=POEType.PROVIDED, digest=item.digest, poe_time=now, validation_object=item.validation_object, ) updated_local_knowledge = dataclasses.replace( orig_local_knowledge, known_poes=list(_poes()), nonrevoked_assertions=new_nonrevoked_assertions, ) updated_pdf_validation_spec = dataclasses.replace( pdf_validation_spec, signature_validation_spec=dataclasses.replace( orig_sig_validation_spec, revinfo_gathering_policy=RevocationInfoGatheringSpec( RevinfoOnlineFetchingRule.LOCAL_ONLY ), local_knowledge=updated_local_knowledge, ), ) return await ades_lta_validation( embedded_sig, updated_pdf_validation_spec, timing_info=timing_info, )