replace more protobuf models

Author: woodruffw

sigstore/_cli.py

@@ -30,7 +30,7 @@
 from rich.console import Console
 from rich.logging import RichHandler
 from sigstore_models.bundle.v1 import Bundle as RawBundle
-from sigstore_protobuf_specs.dev.sigstore.common.v1 import HashAlgorithm
+from sigstore_models.common.v1 import HashAlgorithm
 from typing_extensions import TypeAlias
 
 from sigstore import __version__, dsse

sigstore/_internal/key_details.py

@@ -13,17 +13,15 @@
 # limitations under the License.
 
 """
-Utilities for getting the sigstore_protobuf_specs.dev.sigstore.common.v1.PublicKeyDetails.
+Utilities for getting PublicKeyDetails.
 """
 
-from typing import cast
-
 from cryptography.hazmat.primitives.asymmetric import ec, ed25519, padding, rsa
 from cryptography.x509 import Certificate
-from sigstore_protobuf_specs.dev.sigstore.common import v1
+from sigstore_models.common.v1 import PublicKeyDetails
 
 
-def _get_key_details(certificate: Certificate) -> v1.PublicKeyDetails:
+def _get_key_details(certificate: Certificate) -> PublicKeyDetails:
     """
     Determine PublicKeyDetails from the Certificate.
     We disclude the unrecommended types.
@@ -35,38 +33,38 @@ def _get_key_details(certificate: Certificate) -> v1.PublicKeyDetails:
     params = certificate.signature_algorithm_parameters
     if isinstance(public_key, ec.EllipticCurvePublicKey):
         if isinstance(public_key.curve, ec.SECP256R1):
-            key_details = v1.PublicKeyDetails.PKIX_ECDSA_P256_SHA_256
+            key_details = PublicKeyDetails.PKIX_ECDSA_P256_SHA_256
         elif isinstance(public_key.curve, ec.SECP384R1):
-            key_details = v1.PublicKeyDetails.PKIX_ECDSA_P384_SHA_384
+            key_details = PublicKeyDetails.PKIX_ECDSA_P384_SHA_384
         elif isinstance(public_key.curve, ec.SECP521R1):
-            key_details = v1.PublicKeyDetails.PKIX_ECDSA_P521_SHA_512
+            key_details = PublicKeyDetails.PKIX_ECDSA_P521_SHA_512
         else:
             raise ValueError(f"Unsupported EC curve: {public_key.curve.name}")
     elif isinstance(public_key, rsa.RSAPublicKey):
         if public_key.key_size == 3072:
             if isinstance(params, padding.PKCS1v15):
-                key_details = v1.PublicKeyDetails.PKIX_RSA_PKCS1V15_3072_SHA256
+                key_details = PublicKeyDetails.PKIX_RSA_PKCS1V15_3072_SHA256
             elif isinstance(params, padding.PSS):
-                key_details = v1.PublicKeyDetails.PKIX_RSA_PSS_3072_SHA256
+                key_details = PublicKeyDetails.PKIX_RSA_PSS_3072_SHA256
             else:
                 raise ValueError(
                     f"Unsupported public key type, size, and padding: {type(public_key)}, {public_key.key_size}, {params}"
                 )
         elif public_key.key_size == 4096:
             if isinstance(params, padding.PKCS1v15):
-                key_details = v1.PublicKeyDetails.PKIX_RSA_PKCS1V15_3072_SHA256
+                key_details = PublicKeyDetails.PKIX_RSA_PKCS1V15_3072_SHA256
             elif isinstance(params, padding.PSS):
-                key_details = v1.PublicKeyDetails.PKIX_RSA_PSS_3072_SHA256
+                key_details = PublicKeyDetails.PKIX_RSA_PSS_3072_SHA256
             else:
                 raise ValueError(
                     f"Unsupported public key type, size, and padding: {type(public_key)}, {public_key.key_size}, {params}"
                 )
         else:
             raise ValueError(f"Unsupported RSA key size: {public_key.key_size}")
     elif isinstance(public_key, ed25519.Ed25519PublicKey):
-        key_details = v1.PublicKeyDetails.PKIX_ED25519
+        key_details = PublicKeyDetails.PKIX_ED25519
     # There is likely no need to explicitly detect PKIX_ED25519_PH, especially since the cryptography
     # library does not yet support Ed25519ph.
     else:
         raise ValueError(f"Unsupported public key type: {type(public_key)}")
-    return cast(v1.PublicKeyDetails, key_details)
+    return key_details

sigstore/_internal/merkle.py

@@ -122,5 +122,5 @@ def verify_merkle_inclusion(entry: TransparencyLogEntry) -> None:
     if calc_hash != inclusion_proof.root_hash:
         raise VerificationError(
             f"inclusion proof contains invalid root hash: expected {inclusion_proof}, calculated "
-            f"{calc_hash}"
+            f"{calc_hash.hex()}"
         )

sigstore/_internal/rekor/__init__.py

@@ -31,7 +31,7 @@
 from sigstore.hashes import Hashed
 
 if typing.TYPE_CHECKING:
-    from sigstore.models import LogEntry
+    from sigstore.models import TransparencyLogEntry
 
 __all__ = [
     "_hashedrekord_from_parts",
@@ -72,7 +72,7 @@ class RekorLogSubmitter(ABC):
     def create_entry(
         self,
         request: EntryRequestBody,
-    ) -> LogEntry:
+    ) -> TransparencyLogEntry:
         """
         Submit the request to Rekor.
         """

sigstore/_internal/rekor/client.py

@@ -250,7 +250,7 @@ def log(self) -> RekorLog:
         """
         return RekorLog(f"{self.url}/log", session=self.session)
 
-    def create_entry(self, request: EntryRequestBody) -> LogEntry:
+    def create_entry(self, request: EntryRequestBody) -> TransparencyLogEntry:
         """
         Submit the request to Rekor.
         """

sigstore/_internal/trust.py

@@ -35,29 +35,8 @@
     Certificate,
     load_der_x509_certificate,
 )
-from sigstore_protobuf_specs.dev.sigstore.common.v1 import PublicKey as _PublicKey
-from sigstore_protobuf_specs.dev.sigstore.common.v1 import (
-    PublicKeyDetails as _PublicKeyDetails,
-)
-from sigstore_protobuf_specs.dev.sigstore.common.v1 import TimeRange
-from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
-    CertificateAuthority as _CertificateAuthority,
-)
-from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
-    ClientTrustConfig as _ClientTrustConfig,
-)
-from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
-    Service,
-    ServiceConfiguration,
-    ServiceSelector,
-    TransparencyLogInstance,
-)
-from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
-    SigningConfig as _SigningConfig,
-)
-from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
-    TrustedRoot as _TrustedRoot,
-)
+from sigstore_models.common import v1 as common_v1
+from sigstore_models.trustroot import v1 as trustroot_v1
 
 from sigstore._internal.fulcio.client import FulcioClient
 from sigstore._internal.rekor import RekorLogSubmitter
@@ -83,7 +62,9 @@
 _logger = logging.getLogger(__name__)
 
 
-def _is_timerange_valid(period: TimeRange | None, *, allow_expired: bool) -> bool:
+def _is_timerange_valid(
+    period: common_v1.TimeRange | None, *, allow_expired: bool
+) -> bool:
     """
     Given a `period`, checks that the the current time is not before `start`. If
     `allow_expired` is `False`, also checks that the current time is not after
@@ -116,19 +97,19 @@ class Key:
     key_id: KeyID
 
     _RSA_SHA_256_DETAILS: ClassVar = {
-        _PublicKeyDetails.PKCS1_RSA_PKCS1V5,
-        _PublicKeyDetails.PKIX_RSA_PKCS1V15_2048_SHA256,
-        _PublicKeyDetails.PKIX_RSA_PKCS1V15_3072_SHA256,
-        _PublicKeyDetails.PKIX_RSA_PKCS1V15_4096_SHA256,
+        common_v1.PublicKeyDetails.PKCS1_RSA_PKCS1V5,
+        common_v1.PublicKeyDetails.PKIX_RSA_PKCS1V15_2048_SHA256,
+        common_v1.PublicKeyDetails.PKIX_RSA_PKCS1V15_3072_SHA256,
+        common_v1.PublicKeyDetails.PKIX_RSA_PKCS1V15_4096_SHA256,
     }
 
     _EC_DETAILS_TO_HASH: ClassVar = {
-        _PublicKeyDetails.PKIX_ECDSA_P256_SHA_256: hashes.SHA256(),
-        _PublicKeyDetails.PKIX_ECDSA_P384_SHA_384: hashes.SHA384(),
-        _PublicKeyDetails.PKIX_ECDSA_P521_SHA_512: hashes.SHA512(),
+        common_v1.PublicKeyDetails.PKIX_ECDSA_P256_SHA_256: hashes.SHA256(),
+        common_v1.PublicKeyDetails.PKIX_ECDSA_P384_SHA_384: hashes.SHA384(),
+        common_v1.PublicKeyDetails.PKIX_ECDSA_P521_SHA_512: hashes.SHA512(),
     }
 
-    def __init__(self, public_key: _PublicKey) -> None:
+    def __init__(self, public_key: common_v1.PublicKey) -> None:
         """
         Construct a key from the given Sigstore PublicKey message.
         """
@@ -147,7 +128,7 @@ def __init__(self, public_key: _PublicKey) -> None:
             key = load_der_public_key(
                 public_key.raw_bytes, types=(ec.EllipticCurvePublicKey,)
             )
-        elif public_key.key_details == _PublicKeyDetails.PKIX_ED25519:
+        elif public_key.key_details == common_v1.PublicKeyDetails.PKIX_ED25519:
             hash_algorithm = None
             key = load_der_public_key(
                 public_key.raw_bytes, types=(ed25519.Ed25519PublicKey,)
@@ -198,7 +179,7 @@ class Keyring:
     Represents a set of keys, each of which is a potentially valid verifier.
     """
 
-    def __init__(self, public_keys: list[_PublicKey] = []):
+    def __init__(self, public_keys: list[common_v1.PublicKey] = []):
         """
         Create a new `Keyring`, with `keys` as the initial set of verifying keys.
         """
@@ -263,7 +244,7 @@ class CertificateAuthority:
     Certificate Authority used in a Trusted Root configuration.
     """
 
-    def __init__(self, inner: _CertificateAuthority):
+    def __init__(self, inner: trustroot_v1.CertificateAuthority):
         """
         Construct a new `CertificateAuthority`.
 
@@ -278,7 +259,7 @@ def from_json(cls, path: str) -> CertificateAuthority:
         """
         Create a CertificateAuthority directly from JSON.
         """
-        inner = _CertificateAuthority().from_json(Path(path).read_bytes())
+        inner = trustroot_v1.CertificateAuthority.from_json(Path(path).read_bytes())
         return cls(inner)
 
     def _verify(self) -> None:
@@ -335,7 +316,7 @@ def __str__(self) -> str:
             """Returns the variant's string value."""
             return self.value
 
-    def __init__(self, inner: _SigningConfig):
+    def __init__(self, inner: trustroot_v1.SigningConfig):
         """
         Construct a new `SigningConfig`.
 
@@ -377,19 +358,19 @@ def from_file(
         path: str,
     ) -> SigningConfig:
         """Create a new signing config from file"""
-        inner = _SigningConfig().from_json(Path(path).read_bytes())
+        inner = trustroot_v1.SigningConfig.from_json(Path(path).read_bytes())
         return cls(inner)
 
     @staticmethod
     def _get_valid_services(
-        services: list[Service],
+        services: list[trustroot_v1.Service],
         supported_versions: list[int],
-        config: ServiceConfiguration | None,
-    ) -> list[Service]:
+        config: trustroot_v1.ServiceConfiguration | None,
+    ) -> list[trustroot_v1.Service]:
         """Return supported services, taking SigningConfig restrictions into account"""
 
         # split services by operator, only include valid services
-        services_by_operator: dict[str, list[Service]] = defaultdict(list)
+        services_by_operator: dict[str, list[trustroot_v1.Service]] = defaultdict(list)
         for service in services:
             if service.major_api_version not in supported_versions:
                 continue
@@ -401,20 +382,19 @@ def _get_valid_services(
 
         # build a list of services but make sure we only include one service per operator
         # and use the highest version available for that operator
-        result: list[Service] = []
+        result: list[trustroot_v1.Service] = []
         for op_services in services_by_operator.values():
             op_services.sort(key=lambda s: s.major_api_version)
             result.append(op_services[-1])
 
         # Depending on ServiceSelector, prune the result list
-        if not config or config.selector == ServiceSelector.ALL:
+        if not config or config.selector == trustroot_v1.ServiceSelector.ALL:
             return result
 
-        if config.selector == ServiceSelector.UNDEFINED:
-            raise ValueError("Undefined is not a valid signing config ServiceSelector")
-
         # handle EXACT and ANY selectors
-        count = config.count if config.selector == ServiceSelector.EXACT else 1
+        count = (
+            config.count if config.selector == trustroot_v1.ServiceSelector.EXACT else 1
+        )
         if len(result) < count:
             raise ValueError(
                 f"Expected {count} services in signing config, found {len(result)}"
@@ -474,7 +454,7 @@ def __str__(self) -> str:
             """Returns the variant's string value."""
             return self.value
 
-    def __init__(self, inner: _TrustedRoot):
+    def __init__(self, inner: trustroot_v1.TrustedRoot):
         """
         Construct a new `TrustedRoot`.
 
@@ -501,12 +481,12 @@ def from_file(
         path: str,
     ) -> TrustedRoot:
         """Create a new trust root from file"""
-        inner = _TrustedRoot().from_json(Path(path).read_bytes())
+        inner = trustroot_v1.TrustedRoot.from_json(Path(path).read_bytes())
         return cls(inner)
 
     def _get_tlog_keys(
-        self, tlogs: list[TransparencyLogInstance], purpose: KeyringPurpose
-    ) -> Iterable[_PublicKey]:
+        self, tlogs: list[trustroot_v1.TransparencyLogInstance], purpose: KeyringPurpose
+    ) -> Iterable[common_v1.PublicKey]:
         """
         Yields an iterator of public keys for transparency log instances that
         are suitable for `purpose`.
@@ -523,14 +503,18 @@ def _get_tlog_keys(
     def rekor_keyring(self, purpose: KeyringPurpose) -> RekorKeyring:
         """Return keyring with keys for Rekor."""
 
-        keys: list[_PublicKey] = list(self._get_tlog_keys(self._inner.tlogs, purpose))
+        keys: list[common_v1.PublicKey] = list(
+            self._get_tlog_keys(self._inner.tlogs, purpose)
+        )
         if len(keys) == 0:
             raise MetadataError("Did not find any Rekor keys in trusted root")
         return RekorKeyring(Keyring(keys))
 
     def ct_keyring(self, purpose: KeyringPurpose) -> CTKeyring:
         """Return keyring with key for CTFE."""
-        ctfes: list[_PublicKey] = list(self._get_tlog_keys(self._inner.ctlogs, purpose))
+        ctfes: list[common_v1.PublicKey] = list(
+            self._get_tlog_keys(self._inner.ctlogs, purpose)
+        )
         if not ctfes:
             raise MetadataError("CTFE keys not found in trusted root")
         return CTKeyring(Keyring(ctfes))
@@ -585,7 +569,7 @@ def from_json(cls, raw: str) -> ClientTrustConfig:
         """
         Deserialize the given client trust config.
         """
-        inner = _ClientTrustConfig().from_json(raw)
+        inner = trustroot_v1.ClientTrustConfig.from_json(raw)
         return cls(inner)
 
     @classmethod
@@ -626,48 +610,33 @@ def from_tuf(
         updater = TrustUpdater(url, offline)
 
         tr_path = updater.get_trusted_root_path()
-        inner_tr = _TrustedRoot().from_json(Path(tr_path).read_bytes())
+        inner_tr = trustroot_v1.TrustedRoot.from_json(Path(tr_path).read_bytes())
 
         try:
             sc_path = updater.get_signing_config_path()
-            inner_sc = _SigningConfig().from_json(Path(sc_path).read_bytes())
+            inner_sc = trustroot_v1.SigningConfig.from_json(Path(sc_path).read_bytes())
         except TUFError as e:
             # TUF repo may not have signing config yet: hard code values for prod:
             # https://github.com/sigstore/sigstore-python/issues/1388
             if url == DEFAULT_TUF_URL:
                 embedded = read_embedded("signing_config.v0.2.json", url)
-                inner_sc = _SigningConfig().from_json(embedded)
+                inner_sc = trustroot_v1.SigningConfig.from_json(embedded)
             else:
                 raise e
 
         return cls(
-            _ClientTrustConfig(
-                ClientTrustConfig.ClientTrustConfigType.CONFIG_0_1,
-                inner_tr,
-                inner_sc,
+            trustroot_v1.ClientTrustConfig(
+                media_type=ClientTrustConfig.ClientTrustConfigType.CONFIG_0_1.value,
+                trusted_root=inner_tr,
+                signing_config=inner_sc,
             )
         )
 
-    def __init__(self, inner: _ClientTrustConfig) -> None:
+    def __init__(self, inner: trustroot_v1.ClientTrustConfig) -> None:
         """
         @api private
         """
         self._inner = inner
-        self._verify()
-
-    def _verify(self) -> None:
-        """
-        Performs various feats of heroism to ensure that the client trust config
-        is well-formed.
-        """
-
-        # The client trust config must have a recognized media type.
-        try:
-            ClientTrustConfig.ClientTrustConfigType(self._inner.media_type)
-        except ValueError:
-            raise Error(
-                f"unsupported client trust config format: {self._inner.media_type}"
-            )
 
     @property
     def trusted_root(self) -> TrustedRoot: