rekor: use checkpoint_key_id when available

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

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()

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)
 
 

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."""

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)

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")