1# Copyright 2022 The Sigstore Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15"""
16Verification API machinery.
17"""
18
19from __future__ import annotations
20
21import base64
22import logging
23from datetime import datetime, timezone
24from typing import cast
25
26import rekor_types
27from cryptography.exceptions import InvalidSignature
28from cryptography.hazmat.primitives import serialization
29from cryptography.hazmat.primitives.asymmetric import ec
30from cryptography.x509 import (
31 Certificate,
32 ExtendedKeyUsage,
33 KeyUsage,
34 UnsupportedGeneralNameType,
35)
36from cryptography.x509.oid import ExtendedKeyUsageOID
37from cryptography.x509.verification import (
38 Criticality,
39 ExtensionPolicy,
40 PolicyBuilder,
41 Store,
42)
43from cryptography.x509.verification import (
44 VerificationError as X509VerificationError,
45)
46from pydantic import ValidationError
47from rfc3161_client import TimeStampResponse, VerifierBuilder
48from rfc3161_client import VerificationError as Rfc3161VerificationError
49from sigstore_models.common import v1
50from sigstore_models.rekor import v2
51
52from sigstore import dsse
53from sigstore._internal.key_details import _get_key_details, _get_prehash
54from sigstore._internal.rekor import _hashedrekord_from_parts
55from sigstore._internal.rekor.client import RekorClient
56from sigstore._internal.sct import (
57 verify_sct,
58)
59from sigstore._internal.timestamp import TimestampSource, TimestampVerificationResult
60from sigstore._internal.trust import KeyringPurpose
61from sigstore._utils import base64_encode_pem_cert, sha256_digest
62from sigstore.errors import CertValidationError, VerificationError
63from sigstore.hashes import Hashed
64from sigstore.models import Bundle, ClientTrustConfig, TrustedRoot
65from sigstore.verify.policy import VerificationPolicy
66
67_logger = logging.getLogger(__name__)
68
69# Limit the number of timestamps to prevent DoS
70# From https://github.com/sigstore/sigstore-go/blob/e92142f0734064ebf6001f188b7330a1212245fe/pkg/verify/tsa.go#L29
71MAX_ALLOWED_TIMESTAMP: int = 32
72
73# When verifying an entry, this threshold represents the minimum number of required
74# verified times to consider a signature valid.
75VERIFIED_TIME_THRESHOLD: int = 1
76
77
78class Verifier:
79 """
80 The primary API for verification operations.
81 """
82
83 def __init__(self, *, trusted_root: TrustedRoot):
84 """
85 Create a new `Verifier`.
86
87 `trusted_root` is the `TrustedRoot` object containing the root of trust
88 for the verification process.
89 """
90 self._fulcio_certificate_chain = trusted_root.get_fulcio_certs()
91 self._trusted_root = trusted_root
92
93 # this is an ugly hack needed for verifying "detached" materials
94 # In reality we should be choosing the rekor instance based on the logid
95 tlogs = trusted_root._inner.tlogs
96 if not tlogs:
97 raise VerificationError(
98 "trusted root contains no transparency log instances"
99 )
100 url = tlogs[0].base_url
101 self._rekor = RekorClient(url)
102
103 @classmethod
104 def production(cls, *, offline: bool = False) -> Verifier:
105 """
106 Return a `Verifier` instance configured against Sigstore's production-level services.
107
108 `offline` controls the Trusted Root refresh behavior: if `True`,
109 the verifier uses the Trusted Root in the local TUF cache. If `False`,
110 a TUF repository refresh is attempted.
111 """
112 config = ClientTrustConfig.production(offline=offline)
113 return cls(
114 trusted_root=config.trusted_root,
115 )
116
117 @classmethod
118 def staging(cls, *, offline: bool = False) -> Verifier:
119 """
120 Return a `Verifier` instance configured against Sigstore's staging-level services.
121
122 `offline` controls the Trusted Root refresh behavior: if `True`,
123 the verifier uses the Trusted Root in the local TUF cache. If `False`,
124 a TUF repository refresh is attempted.
125 """
126 config = ClientTrustConfig.staging(offline=offline)
127 return cls(
128 trusted_root=config.trusted_root,
129 )
130
131 def _verify_signed_timestamp(
132 self, timestamp_response: TimeStampResponse, message: bytes
133 ) -> TimestampVerificationResult | None:
134 """
135 Verify a Signed Timestamp using the TSA provided by the Trusted Root.
136 """
137 cert_authorities = self._trusted_root.get_timestamp_authorities()
138 for certificate_authority in cert_authorities:
139 certificates = certificate_authority.certificates(allow_expired=True)
140
141 # We expect at least a signing cert and a root cert but there may be intermediates
142 if len(certificates) < 2:
143 _logger.debug("Unable to verify Timestamp: cert chain is incomplete")
144 continue
145
146 builder = (
147 VerifierBuilder()
148 .tsa_certificate(certificates[0])
149 .add_root_certificate(certificates[-1])
150 )
151 for certificate in certificates[1:-1]:
152 builder = builder.add_intermediate_certificate(certificate)
153
154 verifier = builder.build()
155 try:
156 verifier.verify_message(timestamp_response, message)
157 except Rfc3161VerificationError:
158 _logger.debug("Unable to verify Timestamp with CA.", exc_info=True)
159 continue
160
161 if (
162 certificate_authority.validity_period_start
163 <= timestamp_response.tst_info.gen_time
164 ) and (
165 not certificate_authority.validity_period_end
166 or timestamp_response.tst_info.gen_time
167 < certificate_authority.validity_period_end
168 ):
169 return TimestampVerificationResult(
170 source=TimestampSource.TIMESTAMP_AUTHORITY,
171 time=timestamp_response.tst_info.gen_time,
172 )
173
174 _logger.debug("Unable to verify Timestamp because not in CA time range.")
175
176 return None
177
178 def _verify_timestamp_authority(
179 self, bundle: Bundle
180 ) -> list[TimestampVerificationResult]:
181 """
182 Verify that the given bundle has been timestamped by a trusted timestamp authority
183 and that the timestamp is valid.
184
185 Returns the number of valid signed timestamp in the bundle.
186 """
187 timestamp_responses = []
188 if (
189 timestamp_verification_data
190 := bundle.verification_material.timestamp_verification_data
191 ):
192 timestamp_responses = timestamp_verification_data.rfc3161_timestamps
193
194 if len(timestamp_responses) > MAX_ALLOWED_TIMESTAMP:
195 msg = f"too many signed timestamp: {len(timestamp_responses)} > {MAX_ALLOWED_TIMESTAMP}"
196 raise VerificationError(msg)
197
198 if len(set(timestamp_responses)) != len(timestamp_responses):
199 msg = "duplicate timestamp found"
200 raise VerificationError(msg)
201
202 verified_timestamps = [
203 result
204 for tsr in timestamp_responses
205 if (result := self._verify_signed_timestamp(tsr, bundle.signature))
206 ]
207
208 return verified_timestamps
209
210 def _establish_time(self, bundle: Bundle) -> list[TimestampVerificationResult]:
211 """
212 Establish the time for bundle verification.
213
214 This method uses timestamps from two possible sources:
215 1. RFC3161 signed timestamps from a Timestamping Authority (TSA)
216 2. Transparency Log timestamps
217 """
218 verified_timestamps = []
219
220 # If a timestamp from the timestamping service is available, the Verifier MUST
221 # perform path validation using the timestamp from the Timestamping Service.
222 if bundle.verification_material.timestamp_verification_data:
223 if not self._trusted_root.get_timestamp_authorities():
224 msg = (
225 "no Timestamp Authorities have been provided to validate this "
226 "bundle but it contains a signed timestamp"
227 )
228 raise VerificationError(msg)
229
230 timestamp_from_tsa = self._verify_timestamp_authority(bundle)
231 verified_timestamps.extend(timestamp_from_tsa)
232
233 # If a timestamp from the Transparency Service is available, the Verifier MUST
234 # perform path validation using the timestamp from the Transparency Service.
235 # NOTE: We only include this timestamp if it's accompanied by an inclusion
236 # promise that cryptographically binds it. We verify the inclusion promise
237 # itself later, as part of log entry verification.
238 if (
239 timestamp := bundle.log_entry._inner.integrated_time
240 ) and bundle.log_entry._inner.inclusion_promise:
241 kv = bundle.log_entry._inner.kind_version
242 if not (kv.kind in ["dsse", "hashedrekord"] and kv.version == "0.0.1"):
243 raise VerificationError(
244 "Integrated time only supported for dsse/hashedrekord 0.0.1 types"
245 )
246
247 verified_timestamps.append(
248 TimestampVerificationResult(
249 source=TimestampSource.TRANSPARENCY_SERVICE,
250 time=datetime.fromtimestamp(timestamp, tz=timezone.utc),
251 )
252 )
253 return verified_timestamps
254
255 def _verify_chain_at_time(
256 self, certificate: Certificate, timestamp_result: TimestampVerificationResult
257 ) -> list[Certificate]:
258 """
259 Verify the validity of the certificate chain at the given time.
260
261 Raises a VerificationError if the chain can't be built or be verified.
262 """
263 # Client verifiers normally require the client-auth EKU. Fulcio certificates
264 # instead use code-signing, which is checked separately below; overriding
265 # only the EKU validators preserves the remaining default extension policies.
266 ca_policy = ExtensionPolicy.webpki_defaults_ca().may_be_present(
267 ExtendedKeyUsage, Criticality.NON_CRITICAL, None
268 )
269 ee_policy = ExtensionPolicy.webpki_defaults_ee().may_be_present(
270 ExtendedKeyUsage, Criticality.NON_CRITICAL, None
271 )
272 verifier = (
273 PolicyBuilder()
274 .store(Store(self._fulcio_certificate_chain))
275 .time(timestamp_result.time)
276 .extension_policies(ca_policy=ca_policy, ee_policy=ee_policy)
277 .build_client_verifier()
278 )
279
280 try:
281 # The verified chain includes the end-entity certificate, which callers omit.
282 return verifier.verify(certificate, []).chain[1:]
283 except (X509VerificationError, UnsupportedGeneralNameType) as e:
284 raise CertValidationError(
285 f"failed to build timestamp certificate chain: {e}"
286 )
287
288 def _verify_common_signing_cert(
289 self, bundle: Bundle, policy: VerificationPolicy
290 ) -> None:
291 """
292 Performs the signing certificate verification steps that are shared between
293 `verify_dsse` and `verify_artifact`.
294
295 Raises `VerificationError` on all failures.
296 """
297
298 # In order to verify an artifact, we need to achieve the following:
299 #
300 # 0. Establish a time for the signature.
301 # 1. Verify that the signing certificate chains to the root of trust
302 # and is valid at the time of signing.
303 # 2. Verify the signing certificate's SCT.
304 # 3. Verify that the signing certificate conforms to the Sigstore
305 # X.509 profile as well as the passed-in `VerificationPolicy`.
306 # 4. Verify the inclusion proof and signed checkpoint for the log
307 # entry.
308 # 5. Verify the inclusion promise for the log entry, if present.
309 # 6. Verify the timely insertion of the log entry against the validity
310 # period for the signing certificate.
311 # 7. Verify the signature and input against the signing certificate's
312 # public key.
313 # 8. Verify the transparency log entry's consistency against the other
314 # materials, to prevent variants of CVE-2022-36056.
315 #
316 # This method performs steps (0) through (6) above. Its caller
317 # MUST perform steps (7) and (8) separately, since they vary based on
318 # the kind of verification being performed (i.e. hashedrekord, DSSE, etc.)
319
320 cert = bundle.signing_certificate
321
322 # (0): Establishing a Time for the Signature
323 # First, establish verified times for the signature. This is required to
324 # validate the certificate chain, so this step comes first.
325 # These include TSA timestamps and (in the case of rekor v1 entries)
326 # rekor log integrated time.
327 verified_timestamps = self._establish_time(bundle)
328 if len(verified_timestamps) < VERIFIED_TIME_THRESHOLD:
329 raise VerificationError("not enough sources of verified time")
330
331 # (1): verify that the signing certificate is signed by the root
332 # certificate and that the signing certificate was valid at the
333 # time of signing.
334 chain: list[Certificate] = []
335 for vts in verified_timestamps:
336 chain = self._verify_chain_at_time(cert, vts)
337
338 # (2): verify the signing certificate's SCT.
339 try:
340 verify_sct(
341 cert,
342 chain,
343 self._trusted_root.ct_keyring(KeyringPurpose.VERIFY),
344 )
345 except VerificationError as e:
346 raise VerificationError(f"failed to verify SCT on signing certificate: {e}")
347
348 # (3): verify the signing certificate against the Sigstore
349 # X.509 profile and verify against the given `VerificationPolicy`.
350 usage_ext = cert.extensions.get_extension_for_class(KeyUsage)
351 if not usage_ext.value.digital_signature:
352 raise VerificationError("Key usage is not of type `digital signature`")
353
354 extended_usage_ext = cert.extensions.get_extension_for_class(ExtendedKeyUsage)
355 if ExtendedKeyUsageOID.CODE_SIGNING not in extended_usage_ext.value:
356 raise VerificationError("Extended usage does not contain `code signing`")
357
358 policy.verify(cert)
359
360 _logger.debug("Successfully verified signing certificate validity...")
361
362 # (4): verify the inclusion proof and signed checkpoint for the
363 # log entry.
364 # (5): verify the inclusion promise for the log entry, if present.
365 entry = bundle.log_entry
366 try:
367 entry._verify(self._trusted_root.rekor_keyring(KeyringPurpose.VERIFY))
368 except VerificationError as exc:
369 raise VerificationError(f"invalid log entry: {exc}")
370
371 # (6): verify our established times (timestamps or the log integration time) are
372 # within signing certificate validity period.
373 for vts in verified_timestamps:
374 if not (
375 bundle.signing_certificate.not_valid_before_utc
376 <= vts.time
377 <= bundle.signing_certificate.not_valid_after_utc
378 ):
379 raise VerificationError(
380 f"invalid signing cert: expired at time of signing, time via {vts}"
381 )
382
383 def verify_dsse(
384 self, bundle: Bundle, policy: VerificationPolicy
385 ) -> tuple[str, bytes]:
386 """
387 Verifies an bundle's DSSE envelope, returning the encapsulated payload
388 and its content type.
389
390 This method is only for DSSE-enveloped payloads. To verify
391 an arbitrary input against a bundle, use the `verify_artifact`
392 method.
393
394 `bundle` is the Sigstore `Bundle` to both verify and verify against.
395
396 `policy` is the `VerificationPolicy` to verify against.
397
398 Returns a tuple of `(type, payload)`, where `type` is the payload's
399 type as encoded in the DSSE envelope and `payload` is the raw `bytes`
400 of the payload. No validation of either `type` or `payload` is
401 performed; users of this API **must** assert that `type` is known
402 to them before proceeding to handle `payload` in an application-dependent
403 manner.
404 """
405
406 # (1) through (6) are performed by `_verify_common_signing_cert`.
407 self._verify_common_signing_cert(bundle, policy)
408
409 # (7): verify the bundle's signature and DSSE envelope against the
410 # signing certificate's public key.
411 envelope = bundle._dsse_envelope
412 if envelope is None:
413 raise VerificationError(
414 "cannot perform DSSE verification on a bundle without a DSSE envelope"
415 )
416
417 signing_key = bundle.signing_certificate.public_key()
418 signing_key = cast(ec.EllipticCurvePublicKey, signing_key)
419 dsse._verify(signing_key, envelope)
420
421 # (8): verify the consistency of the log entry's body against
422 # the other bundle materials.
423 # Rekor v2 records DSSE envelopes as hashedrekord/0.0.2 entries whose
424 # digest covers PAE(payloadType, payload) and whose signature.content
425 # equals envelope.signatures[0].sig (rekor-v2-spec §6.1.4). Rekor v1
426 # used a dsse/0.0.1 entry, which is slightly weaker than the
427 # hashedrekord consistency check: dsse entries record an envelope
428 # hash that we *cannot* verify (the envelope is uncanonicalized JSON),
429 # so we manually pick apart the entry body and verify the parts we
430 # can (payload hash and signature list).
431 entry = bundle.log_entry
432 kind = entry._inner.kind_version.kind
433 version = entry._inner.kind_version.version
434 if kind == "hashedrekord" and version == "0.0.2":
435 _validate_hashedrekord_v002_dsse_entry_body(bundle)
436 elif kind == "dsse" and version == "0.0.1":
437 _validate_dsse_v001_entry_body(bundle)
438 else:
439 raise VerificationError(
440 f"Unsupported DSSE log entry type: {kind}/{version}"
441 )
442
443 return (envelope._inner.payload_type, envelope._inner.payload)
444
445 def verify_artifact(
446 self,
447 input_: bytes | Hashed,
448 bundle: Bundle,
449 policy: VerificationPolicy,
450 ) -> None:
451 """
452 Public API for verifying.
453
454 `input_` is the input to verify, either as a buffer of contents or as
455 a prehashed `Hashed` object.
456
457 `bundle` is the Sigstore `Bundle` to verify against.
458
459 `policy` is the `VerificationPolicy` to verify against.
460
461 On failure, this method raises `VerificationError`.
462 """
463
464 # (1) through (6) are performed by `_verify_common_signing_cert`.
465 self._verify_common_signing_cert(bundle, policy)
466
467 hashed_input = sha256_digest(input_)
468 bundle_signature = bundle._inner.message_signature
469 if bundle_signature is None:
470 raise VerificationError("Missing bundle message signature")
471
472 # signature is verified over input digest, but if the bundle documents the digest we still
473 # want to ensure it matches the input digest:
474 if (
475 bundle_signature.message_digest is not None
476 and hashed_input.digest != bundle_signature.message_digest.digest
477 ):
478 raise VerificationError("Bundle message digest mismatch")
479
480 # (7): verify that the signature was signed by the public key in the signing certificate.
481 try:
482 signing_key = bundle.signing_certificate.public_key()
483 signing_key = cast(ec.EllipticCurvePublicKey, signing_key)
484 signing_key.verify(
485 bundle_signature.signature,
486 hashed_input.digest,
487 ec.ECDSA(hashed_input._as_prehashed()),
488 )
489 except InvalidSignature:
490 raise VerificationError("Signature is invalid for input")
491
492 _logger.debug("Successfully verified signature...")
493
494 # (8): verify the consistency of the log entry's body against
495 # the other bundle materials (and input being verified).
496 entry = bundle.log_entry
497 if entry._inner.kind_version.kind != "hashedrekord":
498 raise VerificationError(
499 f"Expected entry type hashedrekord, got {entry._inner.kind_version.kind}"
500 )
501
502 if entry._inner.kind_version.version == "0.0.2":
503 _validate_hashedrekord_v002_entry_body(bundle, hashed_input)
504 elif entry._inner.kind_version.version == "0.0.1":
505 _validate_hashedrekord_v001_entry_body(bundle, hashed_input)
506 else:
507 raise VerificationError(
508 f"Unsupported hashedrekord version {entry._inner.kind_version.version}"
509 )
510
511
512def _validate_dsse_v001_entry_body(bundle: Bundle) -> None:
513 """
514 Validate the Entry body for dsse v001.
515 """
516 entry = bundle.log_entry
517 envelope = bundle._dsse_envelope
518 if envelope is None:
519 raise VerificationError(
520 "cannot perform DSSE verification on a bundle without a DSSE envelope"
521 )
522 try:
523 entry_body = rekor_types.Dsse.model_validate_json(
524 entry._inner.canonicalized_body
525 )
526 except ValidationError as exc:
527 raise VerificationError(f"invalid DSSE log entry: {exc}")
528
529 payload_hash = sha256_digest(envelope._inner.payload).digest.hex()
530 if (
531 entry_body.spec.root.payload_hash.algorithm # type: ignore[union-attr]
532 != rekor_types.dsse.Algorithm.SHA256
533 ):
534 raise VerificationError("expected SHA256 payload hash in DSSE log entry")
535 if payload_hash != entry_body.spec.root.payload_hash.value: # type: ignore[union-attr]
536 raise VerificationError("log entry payload hash does not match bundle")
537
538 # NOTE: Like `dsse._verify`: multiple signatures would be frivolous here,
539 # but we handle them just in case the signer has somehow produced multiple
540 # signatures for their envelope with the same signing key.
541 signatures = [
542 rekor_types.dsse.Signature(
543 signature=base64.b64encode(signature.sig).decode(),
544 verifier=base64_encode_pem_cert(bundle.signing_certificate),
545 )
546 for signature in envelope._inner.signatures
547 ]
548 if signatures != entry_body.spec.root.signatures:
549 raise VerificationError("log entry signatures do not match bundle")
550
551
552def _validate_hashedrekord_v001_entry_body(
553 bundle: Bundle, hashed_input: Hashed
554) -> None:
555 """
556 Validate the Entry body for hashedrekord v001.
557 """
558 entry = bundle.log_entry
559 expected_body = _hashedrekord_from_parts(
560 bundle.signing_certificate,
561 bundle._inner.message_signature.signature, # type: ignore[union-attr]
562 hashed_input,
563 )
564 actual_body = rekor_types.Hashedrekord.model_validate_json(
565 entry._inner.canonicalized_body
566 )
567 if expected_body != actual_body:
568 raise VerificationError(
569 "transparency log entry is inconsistent with other materials"
570 )
571
572
573def _validate_hashedrekord_v002_dsse_entry_body(bundle: Bundle) -> None:
574 """
575 Validate Entry body for a Rekor v2 DSSE envelope encoded as a
576 hashedrekord/0.0.2 entry (rekor-v2-spec §6.1.4).
577
578 The expected entry body has:
579 - data.digest = Hash(PAE(payloadType, payload)), where Hash is the
580 externalized hash function of the entry's signing algorithm.
581 - data.algorithm = the matching HashAlgorithm.
582 - signature.content = envelope.signatures[0].sig.
583 - signature.verifier = the bundle's signing certificate.
584 """
585 entry = bundle.log_entry
586 envelope = bundle._dsse_envelope
587 if envelope is None:
588 raise VerificationError(
589 "cannot perform DSSE verification on a bundle without a DSSE envelope"
590 )
591 if len(envelope._inner.signatures) != 1:
592 raise VerificationError(
593 "DSSE envelope must have exactly one signature for hashedrekord encoding"
594 )
595
596 expected_verifier = _v2_verifier_from_certificate(bundle.signing_certificate)
597 algorithm, hash_func = _get_prehash(expected_verifier.key_details)
598 pae_digest = hash_func(envelope.pae()).digest()
599
600 expected_body = v2.entry.Entry(
601 kind=entry._inner.kind_version.kind,
602 api_version=entry._inner.kind_version.version,
603 spec=v2.entry.Spec(
604 hashed_rekord_v002=v2.hashedrekord.HashedRekordLogEntryV002(
605 data=v1.HashOutput(
606 algorithm=algorithm,
607 digest=base64.b64encode(pae_digest),
608 ),
609 signature=v2.verifier.Signature(
610 content=base64.b64encode(envelope.signature),
611 verifier=expected_verifier,
612 ),
613 )
614 ),
615 )
616 actual_body = v2.entry.Entry.from_json(entry._inner.canonicalized_body)
617 if expected_body != actual_body:
618 raise VerificationError(
619 "transparency log entry is inconsistent with other materials"
620 )
621
622
623def _validate_hashedrekord_v002_entry_body(
624 bundle: Bundle, hashed_input: Hashed
625) -> None:
626 """
627 Validate Entry body for hashedrekord v002.
628 """
629 entry = bundle.log_entry
630 if bundle._inner.message_signature is None:
631 raise VerificationError(
632 "invalid hashedrekord log entry: missing message signature"
633 )
634 v2_expected_body = v2.entry.Entry(
635 kind=entry._inner.kind_version.kind,
636 api_version=entry._inner.kind_version.version,
637 spec=v2.entry.Spec(
638 hashed_rekord_v002=v2.hashedrekord.HashedRekordLogEntryV002(
639 data=v1.HashOutput(
640 algorithm=hashed_input.algorithm,
641 digest=base64.b64encode(hashed_input.digest),
642 ),
643 signature=v2.verifier.Signature(
644 content=base64.b64encode(bundle._inner.message_signature.signature),
645 verifier=_v2_verifier_from_certificate(bundle.signing_certificate),
646 ),
647 )
648 ),
649 )
650 v2_actual_body = v2.entry.Entry.from_json(entry._inner.canonicalized_body)
651 if v2_expected_body != v2_actual_body:
652 raise VerificationError(
653 "transparency log entry is inconsistent with other materials"
654 )
655
656
657def _v2_verifier_from_certificate(certificate: Certificate) -> v2.verifier.Verifier:
658 """
659 Return a Rekor v2 Verifier for the signing certificate.
660
661 Key-to-algorithm mapping is handled by the algorithm registry via
662 `_get_key_details`.
663 """
664 return v2.verifier.Verifier(
665 x509_certificate=v1.X509Certificate(
666 raw_bytes=base64.b64encode(
667 certificate.public_bytes(encoding=serialization.Encoding.DER)
668 )
669 ),
670 key_details=_get_key_details(certificate),
671 )