From 26be6bb946140a9f79b673f7cf3cd69594fa8719 Mon Sep 17 00:00:00 2001 From: ECD5A Date: Sat, 8 Aug 2026 16:29:26 +0500 Subject: [PATCH] rekor: use checkpoint key IDs Select Rekor checkpoint verifiers by signed-note name and key ID. Fall back to the first four bytes of log_id for older trust roots. Closes #1364 Signed-off-by: ECD5A --- CHANGELOG.md | 6 ++ sigstore/_internal/rekor/checkpoint.py | 33 ++++----- sigstore/_internal/trust.py | 81 ++++++++++++++++++++- sigstore/models.py | 15 ++-- test/unit/internal/rekor/test_checkpoint.py | 48 +++++++++++- test/unit/internal/test_trust.py | 14 +++- 6 files changed, 170 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e6baab377..56ffa7932 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,12 @@ All versions prior to 0.9.0 are untracked. ## [Unreleased] +### Fixed + +* Rekor checkpoint verification now uses `checkpoint_key_id` and the log name + from the trusted root, falling back to `log_id` for older trust roots. + ([#1364](https://github.com/sigstore/sigstore-python/issues/1364)) + ## [4.5.0] ### Fixed diff --git a/sigstore/_internal/rekor/checkpoint.py b/sigstore/_internal/rekor/checkpoint.py index c07703787..fd80c86a1 100644 --- a/sigstore/_internal/rekor/checkpoint.py +++ b/sigstore/_internal/rekor/checkpoint.py @@ -171,28 +171,26 @@ def from_text(cls, text: str) -> SignedNote: return cls(note=header, signatures=signatures) - def verify(self, rekor_keyring: RekorKeyring, key_id: KeyID) -> None: + def verify(self, rekor_keyring: RekorKeyring) -> None: """ - Verify the `SignedNote` using the given RekorKeyring and KeyID. + Verify the `SignedNote` using the given RekorKeyring. """ note = str.encode(self.note) for sig in self.signatures: - if sig.sig_hash == key_id[:4]: - try: - rekor_keyring.verify( - key_id=key_id, - signature=base64.b64decode(sig.signature), - data=note, - ) - return - except VerificationError as sig_err: - raise VerificationError(f"checkpoint: invalid signature: {sig_err}") + try: + rekor_keyring.verify_checkpoint( + name=sig.name, + key_id=KeyID(sig.sig_hash), + signature=base64.b64decode(sig.signature), + data=note, + ) + return + except VerificationError: + pass - raise VerificationError( - f"checkpoint: Signature not found for log ID {key_id.hex()}" - ) + raise VerificationError("checkpoint: no valid signature found") @dataclass(frozen=True) @@ -228,10 +226,7 @@ def verify_checkpoint(rekor_keyring: RekorKeyring, entry: TransparencyLogEntry) # 1) verify the signature on the checkpoint # 2) verify the root hash in the checkpoint matches the root hash from the inclusion proof. signed_checkpoint = SignedCheckpoint.from_text(inclusion_proof.checkpoint.envelope) - signed_checkpoint.signed_note.verify( - rekor_keyring, - KeyID(entry._inner.log_id.key_id), - ) + signed_checkpoint.signed_note.verify(rekor_keyring) checkpoint_hash = signed_checkpoint.checkpoint.log_hash root_hash = inclusion_proof.root_hash.hex() diff --git a/sigstore/_internal/trust.py b/sigstore/_internal/trust.py index 238a8e3ef..2c2de2acd 100644 --- a/sigstore/_internal/trust.py +++ b/sigstore/_internal/trust.py @@ -24,6 +24,7 @@ from enum import Enum from pathlib import Path from typing import ClassVar, NewType +from urllib.parse import urlparse import cryptography.hazmat.primitives.asymmetric.padding as padding from cryptography.exceptions import InvalidSignature @@ -190,7 +191,85 @@ def verify(self, *, key_id: KeyID, signature: bytes, data: bytes) -> None: raise VerificationError("keyring: invalid signature") -RekorKeyring = NewType("RekorKeyring", Keyring) +@dataclass(frozen=True) +class _RekorVerifier: + """A Rekor key and the signed-note names that identify its log.""" + + key: Key + names: frozenset[str] + + +def _rekor_log_names(base_url: str) -> frozenset[str]: + """Return the signed-note names represented by a Rekor base URL.""" + + base_url = base_url.rstrip("/") + parsed = urlparse(base_url if "://" in base_url else f"//{base_url}") + names = {base_url} + if parsed.netloc: + names.add(parsed.netloc) + if parsed.hostname: + names.add(parsed.hostname) + return frozenset(names) + + +class RekorKeyring: + """A keyring that tracks Rekor log and checkpoint key identifiers.""" + + def __init__(self, tlogs: list[trustroot_v1.TransparencyLogInstance]): + self._keyring: dict[KeyID, list[Key]] = {} + self._checkpoint_keyring: dict[KeyID, list[_RekorVerifier]] = {} + self._keys: list[Key] = [] + + for tlog in tlogs: + try: + key = Key(tlog.public_key) + except VerificationError as e: + _logger.warning(f"Failed to load a trusted root key: {e}") + continue + + self._keys.append(key) + log_id = KeyID(tlog.log_id.key_id) + self._keyring.setdefault(log_id, []).append(key) + + checkpoint_key_id = ( + KeyID(tlog.checkpoint_key_id.key_id) + if tlog.checkpoint_key_id is not None + else KeyID(log_id[:4]) + ) + verifier = _RekorVerifier(key=key, names=_rekor_log_names(tlog.base_url)) + self._checkpoint_keyring.setdefault(checkpoint_key_id, []).append(verifier) + + def verify(self, *, key_id: KeyID, signature: bytes, data: bytes) -> None: + """Verify a Rekor inclusion promise, using its log ID as a hint.""" + + candidates = self._keyring.get(key_id, self._keys) + for candidate in candidates: + try: + candidate.verify(signature, data) + return + except InvalidSignature: + pass + + raise VerificationError("keyring: invalid signature") + + def verify_checkpoint( + self, *, name: str, key_id: KeyID, signature: bytes, data: bytes + ) -> None: + """Verify a checkpoint signature identified by signed-note name and key ID.""" + + candidates = self._checkpoint_keyring.get(key_id, []) + for candidate in candidates: + if name not in candidate.names: + continue + try: + candidate.key.verify(signature, data) + return + except InvalidSignature: + pass + + raise VerificationError("keyring: invalid checkpoint signature") + + CTKeyring = NewType("CTKeyring", Keyring) diff --git a/sigstore/models.py b/sigstore/models.py index 2237b772e..3514669c5 100644 --- a/sigstore/models.py +++ b/sigstore/models.py @@ -855,12 +855,17 @@ def _get_tlog_keys( def rekor_keyring(self, purpose: KeyringPurpose) -> RekorKeyring: """Return keyring with keys for Rekor.""" - keys: list[common_v1.PublicKey] = list( - self._get_tlog_keys(self._inner.tlogs, purpose) - ) - if len(keys) == 0: + allow_expired = purpose is KeyringPurpose.VERIFY + tlogs = [ + tlog + for tlog in self._inner.tlogs + if is_timerange_valid( + tlog.public_key.valid_for, allow_expired=allow_expired + ) + ] + if not tlogs: raise MetadataError("Did not find any Rekor keys in trusted root") - return RekorKeyring(Keyring(keys)) + return RekorKeyring(tlogs) def ct_keyring(self, purpose: KeyringPurpose) -> CTKeyring: """Return keyring with key for CTFE.""" diff --git a/test/unit/internal/rekor/test_checkpoint.py b/test/unit/internal/rekor/test_checkpoint.py index 757566245..24da83342 100644 --- a/test/unit/internal/rekor/test_checkpoint.py +++ b/test/unit/internal/rekor/test_checkpoint.py @@ -15,8 +15,13 @@ import base64 import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from sigstore_models.common import v1 as common_v1 +from sigstore_models.trustroot import v1 as trustroot_v1 -from sigstore._internal.rekor.checkpoint import LogCheckpoint +from sigstore._internal.rekor.checkpoint import LogCheckpoint, SignedNote +from sigstore._internal.trust import RekorKeyring from sigstore.errors import VerificationError @@ -45,3 +50,44 @@ def test_from_text_invalid_root_hash(self): # An undecodable base64 root hash must also surface as a VerificationError. with pytest.raises(VerificationError, match="invalid root hash"): LogCheckpoint.from_text("rekor.example - 123\n42\n!!!notbase64!!!\n") + + +class TestSignedNote: + @staticmethod + def _signed_note(name: str, checkpoint_key_id: bytes): + private_key = Ed25519PrivateKey.generate() + public_key = common_v1.PublicKey( + raw_bytes=base64.b64encode( + private_key.public_key().public_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + ), + key_details=common_v1.PublicKeyDetails.PKIX_ED25519, + ) + tlog = trustroot_v1.TransparencyLogInstance( + base_url="https://rekor.example", + hash_algorithm=common_v1.HashAlgorithm.SHA2_256, + public_key=public_key, + log_id=common_v1.LogId(key_id=base64.b64encode(b"legacy-log-id")), + checkpoint_key_id=common_v1.LogId( + key_id=base64.b64encode(checkpoint_key_id) + ), + ) + + root_hash = base64.b64encode(b"\x00" * 32).decode() + note = f"rekor.example\n42\n{root_hash}\n" + signature = checkpoint_key_id + private_key.sign(note.encode()) + envelope = f"{note}\n— {name} {base64.b64encode(signature).decode()}\n" + return SignedNote.from_text(envelope), RekorKeyring([tlog]) + + def test_verify_uses_checkpoint_key_id(self): + signed_note, keyring = self._signed_note("rekor.example", b"\x01\x02\x03\x04") + + signed_note.verify(keyring) + + def test_verify_rejects_wrong_log_name(self): + signed_note, keyring = self._signed_note("other.example", b"\x01\x02\x03\x04") + + with pytest.raises(VerificationError, match="no valid signature"): + signed_note.verify(keyring) diff --git a/test/unit/internal/test_trust.py b/test/unit/internal/test_trust.py index 4340ee007..8763e46ac 100644 --- a/test/unit/internal/test_trust.py +++ b/test/unit/internal/test_trust.py @@ -13,11 +13,12 @@ # limitations under the License. +import base64 import os from datetime import datetime, timedelta, timezone import pytest -from sigstore_models.common.v1 import TimeRange +from sigstore_models.common.v1 import LogId, TimeRange from sigstore_models.trustroot.v1 import ( Service, ServiceConfiguration, @@ -229,6 +230,17 @@ def test_good(self, asset, file): assert root.get_fulcio_certs() is not None assert root.get_timestamp_authorities() is not None + def test_rekor_keyring_tracks_checkpoint_key_id(self, asset): + root = TrustedRoot.from_file(asset("trusted_root/trustedroot.v1.json")) + checkpoint_key_id = b"\x01\x02\x03\x04" + root._inner.tlogs[0].checkpoint_key_id = LogId( + key_id=base64.b64encode(checkpoint_key_id) + ) + + keyring = root.rekor_keyring(KeyringPurpose.VERIFY) + + assert checkpoint_key_id in keyring._checkpoint_keyring + def test_bad_media_type(self, asset): path = asset("trusted_root/trustedroot.badtype.json")