Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sigstore/verify/verifier.py: 24%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

221 statements  

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 )