port rekor v2 models, get tests passing locally

Author: woodruffw

sigstore/_internal/rekor/client_v2.py

@@ -18,16 +18,17 @@
 
 from __future__ import annotations
 
+import base64
 import json
 import logging
 
 import requests
 from cryptography.hazmat.primitives import serialization
 from cryptography.x509 import Certificate
+from sigstore_models import intoto
+from sigstore_models.common import v1 as common_v1
+from sigstore_models.rekor import v2 as rekor_v2
 from sigstore_models.rekor.v1 import TransparencyLogEntry as _TransparencyLogEntry
-from sigstore_protobuf_specs.dev.sigstore.common import v1 as common_v1
-from sigstore_protobuf_specs.dev.sigstore.rekor import v2
-from sigstore_protobuf_specs.io import intoto
 
 from sigstore._internal import USER_AGENT
 from sigstore._internal.key_details import _get_key_details
@@ -104,15 +105,17 @@ def _build_hashed_rekord_request(
         """
         Construct a hashed rekord request to submit to Rekor.
         """
-        req = v2.CreateEntryRequest(
-            hashed_rekord_request_v002=v2.HashedRekordRequestV002(
+        req = rekor_v2.entry.CreateEntryRequest(
+            hashed_rekord_request_v002=rekor_v2.hashedrekord.HashedRekordRequestV002(
                 digest=hashed_input.digest,
-                signature=v2.Signature(
+                signature=rekor_v2.verifier.Signature(
                     content=signature,
-                    verifier=v2.Verifier(
+                    verifier=rekor_v2.verifier.Verifier(
                         x509_certificate=common_v1.X509Certificate(
-                            raw_bytes=certificate.public_bytes(
-                                encoding=serialization.Encoding.DER
+                            raw_bytes=base64.b64encode(
+                                certificate.public_bytes(
+                                    encoding=serialization.Encoding.DER
+                                )
                             )
                         ),
                         key_details=_get_key_details(certificate),
@@ -129,24 +132,16 @@ def _build_dsse_request(
         """
         Construct a dsse request to submit to Rekor.
         """
-        req = v2.CreateEntryRequest(
-            dsse_request_v002=v2.DsseRequestV002(
-                envelope=intoto.Envelope(
-                    payload=envelope._inner.payload,
-                    payload_type=envelope._inner.payload_type,
-                    signatures=[
-                        intoto.Signature(
-                            keyid=signature.keyid,  # type: ignore[arg-type]
-                            sig=signature.sig,
-                        )
-                        for signature in envelope._inner.signatures
-                    ],
-                ),
+        req = rekor_v2.entry.CreateEntryRequest(
+            dsse_request_v002=rekor_v2.dsse.DSSERequestV002(
+                envelope=envelope._inner,
                 verifiers=[
-                    v2.Verifier(
+                    rekor_v2.verifier.Verifier(
                         x509_certificate=common_v1.X509Certificate(
-                            raw_bytes=certificate.public_bytes(
-                                encoding=serialization.Encoding.DER
+                            raw_bytes=base64.b64encode(
+                                certificate.public_bytes(
+                                    encoding=serialization.Encoding.DER
+                                )
                             )
                         ),
                         key_details=_get_key_details(certificate),

sigstore/models.py

@@ -218,233 +218,6 @@ def _verify(self, keyring: RekorKeyring) -> None:
             )
 
 
-# @dataclass(frozen=True)
-# class LogEntry:
-#     """
-#     Represents a transparency log entry.
-
-#     Log entries are retrieved from the transparency log after signing or verification events,
-#     or loaded from "Sigstore" bundles provided by the user.
-
-#     This representation allows for either a missing inclusion promise or a missing
-#     inclusion proof, but not both: attempting to construct a `LogEntry` without
-#     at least one will fail.
-#     """
-
-#     uuid: Optional[str]  # noqa: UP045
-#     """
-#     This entry's unique ID in the log instance it was retrieved from.
-
-#     For sharded log deployments, IDs are unique per-shard.
-
-#     Not present for `LogEntry` instances loaded from Sigstore bundles.
-#     """
-
-#     body: B64Str
-#     """
-#     The base64-encoded body of the transparency log entry.
-#     """
-
-#     integrated_time: int
-#     """
-#     The UNIX time at which this entry was integrated into the transparency log.
-#     """
-
-#     log_id: str
-#     """
-#     The log's ID (as the SHA256 hash of the DER-encoded public key for the log
-#     at the time of entry inclusion).
-#     """
-
-#     log_index: int
-#     """
-#     The index of this entry within the log.
-#     """
-
-#     inclusion_proof: LogInclusionProof
-#     """
-#     An inclusion proof for this log entry.
-#     """
-
-#     inclusion_promise: Optional[B64Str]  # noqa: UP045
-#     """
-#     An inclusion promise for this log entry, if present.
-
-#     Internally, this is a base64-encoded Signed Entry Timestamp (SET) for this
-#     log entry.
-#     """
-
-#     _kind_version: KindVersion
-#     """
-#     The kind and version of the log entry.
-#     """
-
-#     @classmethod
-#     def _from_response(cls, dict_: dict[str, Any]) -> LogEntry:
-#         """
-#         Create a new `LogEntry` from the given API response.
-#         """
-
-#         # Assumes we only get one entry back
-#         entries = list(dict_.items())
-#         if len(entries) != 1:
-#             raise ValueError("Received multiple entries in response")
-#         uuid, entry = entries[0]
-
-#         # Fill in the appropriate kind
-#         body_entry: ProposedEntry = TypeAdapter(ProposedEntry).validate_json(
-#             base64.b64decode(entry["body"])
-#         )
-#         if not isinstance(body_entry, (Hashedrekord, Dsse)):
-#             raise InvalidBundle("log entry is not of expected type")
-
-#         return LogEntry(
-#             uuid=uuid,
-#             body=entry["body"],
-#             integrated_time=entry["integratedTime"],
-#             log_id=entry["logID"],
-#             log_index=entry["logIndex"],
-#             inclusion_proof=LogInclusionProof.model_validate(
-#                 entry["verification"]["inclusionProof"]
-#             ),
-#             inclusion_promise=entry["verification"]["signedEntryTimestamp"],
-#             _kind_version=KindVersion(
-#                 kind=body_entry.kind, version=body_entry.api_version
-#             ),
-#         )
-
-#     @classmethod
-#     def _from_dict_rekor(cls, dict_: dict[str, Any]) -> LogEntry:
-#         """
-#         Create a new `LogEntry` from the given Rekor TransparencyLogEntry.
-#         """
-#         tlog_entry = rekor_v1.TransparencyLogEntry.from_dict(dict_)
-
-#         inclusion_proof: InclusionProof | None = tlog_entry.inclusion_proof
-#         # This check is required by us as the client, not the
-#         # protobuf-specs themselves.
-#         if not inclusion_proof or not inclusion_proof.checkpoint.envelope:
-#             raise InvalidBundle("entry must contain inclusion proof, with checkpoint")
-
-#         parsed_inclusion_proof = LogInclusionProof(
-#             checkpoint=inclusion_proof.checkpoint.envelope,
-#             hashes=[h.hex() for h in inclusion_proof.hashes],
-#             log_index=inclusion_proof.log_index,
-#             root_hash=inclusion_proof.root_hash.hex(),
-#             tree_size=inclusion_proof.tree_size,
-#         )
-
-#         inclusion_promise: B64Str | None = None
-#         if tlog_entry.inclusion_promise:
-#             inclusion_promise = B64Str(
-#                 base64.b64encode(
-#                     tlog_entry.inclusion_promise.signed_entry_timestamp
-#                 ).decode()
-#             )
-
-#         return LogEntry(
-#             uuid=None,
-#             body=B64Str(base64.b64encode(tlog_entry.canonicalized_body).decode()),
-#             integrated_time=tlog_entry.integrated_time,
-#             log_id=tlog_entry.log_id.key_id.hex(),
-#             log_index=tlog_entry.log_index,
-#             inclusion_proof=parsed_inclusion_proof,
-#             _kind_version=tlog_entry.kind_version,
-#             inclusion_promise=inclusion_promise,
-#         )
-
-#     def _to_rekor(self) -> rekor_v1.TransparencyLogEntry:
-#         """
-#         Create a new protobuf-level `TransparencyLogEntry` from this `LogEntry`.
-
-#         @private
-#         """
-#         inclusion_proof = rekor_v1.InclusionProof(
-#             log_index=self.inclusion_proof.log_index,
-#             root_hash=bytes.fromhex(self.inclusion_proof.root_hash),
-#             tree_size=self.inclusion_proof.tree_size,
-#             hashes=[bytes.fromhex(hash_) for hash_ in self.inclusion_proof.hashes],
-#             checkpoint=rekor_v1.Checkpoint(envelope=self.inclusion_proof.checkpoint),
-#         )
-
-#         tlog_entry = rekor_v1.TransparencyLogEntry(
-#             log_index=self.log_index,
-#             log_id=common_v1.LogId(key_id=bytes.fromhex(self.log_id)),
-#             integrated_time=self.integrated_time,
-#             inclusion_proof=inclusion_proof,
-#             kind_version=self._kind_version,
-#             canonicalized_body=base64.b64decode(self.body),
-#         )
-#         if self.inclusion_promise:
-#             inclusion_promise = rekor_v1.InclusionPromise(
-#                 signed_entry_timestamp=base64.b64decode(self.inclusion_promise)
-#             )
-#             tlog_entry.inclusion_promise = inclusion_promise
-
-#         return tlog_entry
-
-#     def encode_canonical(self) -> bytes:
-#         """
-#         Returns a canonicalized JSON (RFC 8785) representation of the transparency log entry.
-
-#         This encoded representation is suitable for verification against
-#         the Signed Entry Timestamp.
-#         """
-#         payload: dict[str, int | str] = {
-#             "body": self.body,
-#             "integratedTime": self.integrated_time,
-#             "logID": self.log_id,
-#             "logIndex": self.log_index,
-#         }
-
-#         return rfc8785.dumps(payload)
-
-#     def _verify_set(self, keyring: RekorKeyring) -> None:
-#         """
-#         Verify the inclusion promise (Signed Entry Timestamp) for a given transparency log
-#         `entry` using the given `keyring`.
-
-#         Fails if the given log entry does not contain an inclusion promise.
-#         """
-
-#         if self.inclusion_promise is None:
-#             raise VerificationError("SET: invalid inclusion promise: missing")
-
-#         signed_entry_ts = base64.b64decode(self.inclusion_promise)
-
-#         try:
-#             keyring.verify(
-#                 key_id=KeyID(bytes.fromhex(self.log_id)),
-#                 signature=signed_entry_ts,
-#                 data=self.encode_canonical(),
-#             )
-#         except VerificationError as exc:
-#             raise VerificationError(f"SET: invalid inclusion promise: {exc}")
-
-#     def _verify(self, keyring: RekorKeyring) -> None:
-#         """
-#         Verifies this log entry.
-
-#         This method performs steps (5), (6), and optionally (7) in
-#         the top-level verify API:
-
-#         * Verifies the consistency of the entry with the given bundle;
-#         * Verifies the Merkle inclusion proof and its signed checkpoint;
-#         * Verifies the inclusion promise, if present.
-#         """
-
-#         verify_merkle_inclusion(self)
-#         verify_checkpoint(keyring, self)
-
-#         _logger.debug(f"successfully verified inclusion proof: index={self.log_index}")
-
-#         if self.inclusion_promise:
-#             self._verify_set(keyring)
-#             _logger.debug(
-#                 f"successfully verified inclusion promise: index={self.log_index}"
-#             )
-
-
 class TimestampVerificationData:
     """
     Represents a TimestampVerificationData structure.

sigstore/verify/verifier.py

@@ -39,8 +39,8 @@
 from pydantic import ValidationError
 from rfc3161_client import TimeStampResponse, VerifierBuilder
 from rfc3161_client import VerificationError as Rfc3161VerificationError
-from sigstore_protobuf_specs.dev.sigstore.common import v1
-from sigstore_protobuf_specs.dev.sigstore.rekor import v2
+from sigstore_models.common import v1
+from sigstore_models.rekor import v2
 
 from sigstore import dsse
 from sigstore._internal.rekor import _hashedrekord_from_parts
@@ -546,7 +546,7 @@ def _validate_dsse_v002_entry_body(bundle: Bundle) -> None:
             "cannot perform DSSE verification on a bundle without a DSSE envelope"
         )
     try:
-        v2_body = v2.Entry().from_json(entry._inner.canonicalized_body)
+        v2_body = v2.entry.Entry.from_json(entry._inner.canonicalized_body)
     except ValidationError as exc:
         raise VerificationError(f"invalid DSSE log entry: {exc}")
 
@@ -561,8 +561,8 @@ def _validate_dsse_v002_entry_body(bundle: Bundle) -> None:
         raise VerificationError("DSSE entry payload hash does not match bundle")
 
     v2_signatures = [
-        v2.Signature(
-            content=signature.sig,
+        v2.verifier.Signature(
+            content=base64.b64encode(signature.sig),
             verifier=_v2_verifier_from_certificate(bundle.signing_certificate),
         )
         for signature in envelope._inner.signatures
@@ -601,30 +601,32 @@ def _validate_hashedrekord_v002_entry_body(bundle: Bundle) -> None:
         raise VerificationError(
             "invalid hashedrekord log entry: missing message signature"
         )
-    v2_expected_body = v2.Entry(
+    v2_expected_body = v2.entry.Entry(
         kind=entry._inner.kind_version.kind,
         api_version=entry._inner.kind_version.version,
-        spec=v2.Spec(
-            hashed_rekord_v002=v2.HashedRekordLogEntryV002(
+        spec=v2.entry.Spec(
+            hashed_rekord_v002=v2.hashedrekord.HashedRekordLogEntryV002(
                 data=v1.HashOutput(
                     algorithm=bundle._inner.message_signature.message_digest.algorithm,
-                    digest=bundle._inner.message_signature.message_digest.digest,
+                    digest=base64.b64encode(
+                        bundle._inner.message_signature.message_digest.digest
+                    ),
                 ),
-                signature=v2.Signature(
-                    content=bundle._inner.message_signature.signature,
+                signature=v2.verifier.Signature(
+                    content=base64.b64encode(bundle._inner.message_signature.signature),
                     verifier=_v2_verifier_from_certificate(bundle.signing_certificate),
                 ),
             )
         ),
     )
-    v2_actual_body = v2.Entry().from_json(entry._inner.canonicalized_body)
+    v2_actual_body = v2.entry.Entry.from_json(entry._inner.canonicalized_body)
     if v2_expected_body != v2_actual_body:
         raise VerificationError(
             "transparency log entry is inconsistent with other materials"
         )
 
 
-def _v2_verifier_from_certificate(certificate: Certificate) -> v2.Verifier:
+def _v2_verifier_from_certificate(certificate: Certificate) -> v2.verifier.Verifier:
     """
     Return a Rekor v2 protobuf Verifier for the signing certificate.
 
@@ -648,9 +650,11 @@ def _v2_verifier_from_certificate(certificate: Certificate) -> v2.Verifier:
     else:
         raise ValueError(f"Unsupported public key type: {type(public_key)}")
 
-    return v2.Verifier(
+    return v2.verifier.Verifier(
         x509_certificate=v1.X509Certificate(
-            certificate.public_bytes(encoding=serialization.Encoding.DER)
+            raw_bytes=base64.b64encode(
+                certificate.public_bytes(encoding=serialization.Encoding.DER)
+            )
         ),
         key_details=cast(v1.PublicKeyDetails, key_details),
     )