sigstore: flatten models into sigstore.models

Signed-off-by: William Woodruff <william@trailofbits.com>

Author: woodruffw

sigstore/_cli.py

@@ -41,6 +41,7 @@
 from sigstore._utils import sha256_digest
 from sigstore.errors import Error, VerificationError
 from sigstore.hashes import Hashed
+from sigstore.models import Bundle
 from sigstore.oidc import (
     DEFAULT_OAUTH_ISSUER_URL,
     STAGING_OAUTH_ISSUER_URL,
@@ -54,7 +55,6 @@
     Verifier,
     policy,
 )
-from sigstore.verify.models import Bundle
 
 logging.basicConfig(format="%(message)s", datefmt="[%X]", handlers=[RichHandler()])
 _logger = logging.getLogger(__name__)

sigstore/_internal/merkle.py

@@ -33,7 +33,7 @@
 from sigstore.errors import VerificationError
 
 if typing.TYPE_CHECKING:
-    from sigstore.transparency import LogEntry
+    from sigstore.models import LogEntry
 
 
 _LEAF_HASH_PREFIX = 0

sigstore/_internal/rekor/__init__.py

@@ -24,12 +24,7 @@
 from sigstore._utils import base64_encode_pem_cert
 from sigstore.hashes import Hashed
 
-from .checkpoint import SignedCheckpoint
-from .client import RekorClient
-
 __all__ = [
-    "RekorClient",
-    "SignedCheckpoint",
     "_hashedrekord_from_parts",
 ]
 

sigstore/_internal/rekor/checkpoint.py

@@ -32,7 +32,7 @@
 from sigstore.errors import VerificationError
 
 if typing.TYPE_CHECKING:
-    from sigstore.transparency import LogEntry
+    from sigstore.models import LogEntry
 
 
 @dataclass(frozen=True)

sigstore/_internal/rekor/client.py

@@ -28,7 +28,7 @@
 import rekor_types
 import requests
 
-from sigstore.transparency import LogEntry
+from sigstore.models import LogEntry
 
 _logger = logging.getLogger(__name__)
 

sigstore/verify/models.py renamed to sigstore/models.py

@@ -13,20 +13,33 @@
 # limitations under the License.
 
 """
-Common (base) models for the verification APIs.
+Common models shared between signing and verification.
 """
 
 from __future__ import annotations
 
 import base64
 import logging
+import typing
 from textwrap import dedent
+from typing import Any, List, Optional
 
+import rfc8785
 from cryptography.hazmat.primitives.serialization import Encoding
 from cryptography.x509 import (
     Certificate,
     load_der_x509_certificate,
 )
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    StrictInt,
+    StrictStr,
+    ValidationInfo,
+    field_validator,
+)
+from pydantic.dataclasses import dataclass
 from sigstore_protobuf_specs.dev.sigstore.bundle import v1 as bundle_v1
 from sigstore_protobuf_specs.dev.sigstore.bundle.v1 import (
     Bundle as _Bundle,
@@ -39,18 +52,203 @@
 )
 
 from sigstore import dsse
+from sigstore._internal.merkle import verify_merkle_inclusion
+from sigstore._internal.rekor.checkpoint import verify_checkpoint
 from sigstore._utils import (
     B64Str,
     BundleType,
+    KeyID,
     cert_is_leaf,
     cert_is_root_ca,
 )
-from sigstore.errors import Error
-from sigstore.transparency import LogEntry, LogInclusionProof
+from sigstore.errors import Error, VerificationError
+
+if typing.TYPE_CHECKING:
+    from sigstore._internal.trustroot import RekorKeyring
+
 
 _logger = logging.getLogger(__name__)
 
 
+class LogInclusionProof(BaseModel):
+    """
+    Represents an inclusion proof for a transparency log entry.
+    """
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    checkpoint: StrictStr = Field(..., alias="checkpoint")
+    hashes: List[StrictStr] = Field(..., alias="hashes")
+    log_index: StrictInt = Field(..., alias="logIndex")
+    root_hash: StrictStr = Field(..., alias="rootHash")
+    tree_size: StrictInt = Field(..., alias="treeSize")
+
+    @field_validator("log_index")
+    def _log_index_positive(cls, v: int) -> int:
+        if v < 0:
+            raise ValueError(f"Inclusion proof has invalid log index: {v} < 0")
+        return v
+
+    @field_validator("tree_size")
+    def _tree_size_positive(cls, v: int) -> int:
+        if v < 0:
+            raise ValueError(f"Inclusion proof has invalid tree size: {v} < 0")
+        return v
+
+    @field_validator("tree_size")
+    def _log_index_within_tree_size(
+        cls, v: int, info: ValidationInfo, **kwargs: Any
+    ) -> int:
+        if "log_index" in info.data and v <= info.data["log_index"]:
+            raise ValueError(
+                "Inclusion proof has log index greater than or equal to tree size: "
+                f"{v} <= {info.data['log_index']}"
+            )
+        return v
+
+
+@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]
+    """
+    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]
+    """
+    An inclusion promise for this log entry, if present.
+
+    Internally, this is a base64-encoded Signed Entry Timestamp (SET) for this
+    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]
+        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"],
+        )
+
+    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 InvalidBundle(Error):
     """
     Raised when the associated `Bundle` is invalid in some way.
@@ -72,20 +270,6 @@ def diagnostics(self) -> str:
         )
 
 
-class InvalidRekorEntry(InvalidBundle):
-    """
-    Raised if the effective Rekor entry in `VerificationMaterials.rekor_entry()`
-    does not match the other materials in `VerificationMaterials`.
-
-    This can only happen in two scenarios:
-
-    * A user has supplied the wrong offline entry, potentially maliciously;
-    * The Rekor log responded with the wrong entry, suggesting a server error.
-    """
-
-    pass
-
-
 class Bundle:
     """
     Represents a Sigstore bundle.

sigstore/sign.py

@@ -64,8 +64,8 @@
 from sigstore._internal.sct import verify_sct
 from sigstore._internal.trustroot import KeyringPurpose, TrustedRoot
 from sigstore._utils import sha256_digest
+from sigstore.models import Bundle
 from sigstore.oidc import ExpiredIdentity, IdentityToken
-from sigstore.verify.models import Bundle
 
 _logger = logging.getLogger(__name__)