Merge branch 'main' into kindversion

Author: ramonpetgrave64

CHANGELOG.md

@@ -23,6 +23,11 @@ All versions prior to 0.9.0 are untracked.
 * API: Make Rekor APIs compatible with Rekor v2 by removing trailing slashes
   from endpoints ([#1366](https://github.com/sigstore/sigstore-python/pull/1366))
 
+### Changed
+
+* `--trust-config` now requires a file with SigningConfig v0.2, and is able to fully
+  configure the used Sigstore instance [#1358]/(https://github.com/sigstore/sigstore-python/pull/1358)
+
 ## [3.6.2]
 
 ### Fixed

install/requirements.txt

@@ -372,9 +372,9 @@ multidict==6.4.3 \
     --hash=sha256:fbd8d737867912b6c5f99f56782b8cb81f978a97b4437a1c476de90a3e41c9a1 \
     --hash=sha256:fbf226ac85f7d6b6b9ba77db4ec0704fde88463dc17717aec78ec3c8546c70ad
     # via grpclib
-platformdirs==4.3.7 \
-    --hash=sha256:a03875334331946f13c549dbd8f4bac7a13a50a895a0eb1e8c6a8ace80d40a94 \
-    --hash=sha256:eb437d586b6a0986388f0d6f74aa0cde27b48d0e3d66843640bfb6bdcdb6e351
+platformdirs==4.3.8 \
+    --hash=sha256:3d512d96e16bcb959a814c9f348431070822a6496326a4be0911c40b5a74c2bc \
+    --hash=sha256:ff7059bb7eb1179e2685604f4aaf157cfd9535242bd23742eadc3c13542139b4
     # via sigstore
 pyasn1==0.6.1 \
     --hash=sha256:0d632f46f2ba09143da3a8afe9e33fb6f92fa2320ab7e886e2d0f7672af84629 \

pyproject.toml

@@ -38,7 +38,7 @@ dependencies = [
   "rfc8785 ~= 0.1.2",
   "rfc3161-client >= 0.1.2,< 1.1.0",
   # NOTE(ww): Both under active development, so strictly pinned.
-  "sigstore-protobuf-specs == 0.3.2",
+  "sigstore-protobuf-specs == 0.4.1",
   "sigstore-rekor-types == 0.0.18",
   "tuf ~= 6.0",
   "platformdirs ~= 4.2",

sigstore/_cli.py

@@ -713,9 +713,7 @@ def _sign_common(
             else:
                 sig_output = sys.stdout
 
-            signature = base64.b64encode(
-                result._inner.message_signature.signature
-            ).decode()
+            signature = base64.b64encode(result.signature).decode()
             print(signature, file=sig_output)
             if outputs.signature is not None:
                 print(f"Signature written to {outputs.signature}")
@@ -1202,7 +1200,7 @@ def _fix_bundle(args: argparse.Namespace) -> None:
     # for custom Rekor instances.
     rekor = RekorClient.staging() if args.staging else RekorClient.production()
 
-    raw_bundle = RawBundle().from_json(args.bundle.read_text())
+    raw_bundle = RawBundle.from_dict(json.loads(args.bundle.read_bytes()))
 
     if len(raw_bundle.verification_material.tlog_entries) != 1:
         _fatal("unfixable bundle: must have exactly one log entry")

sigstore/_internal/trust.py

@@ -45,8 +45,13 @@
     ClientTrustConfig as _ClientTrustConfig,
 )
 from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
+    Service,
+    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,
 )
@@ -93,14 +98,14 @@ class Key:
     key: PublicKey
     key_id: KeyID
 
-    _RSA_SHA_256_DETAILS: ClassVar[set[_PublicKeyDetails]] = {
+    _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,
     }
 
-    _EC_DETAILS_TO_HASH: ClassVar[dict[_PublicKeyDetails, hashes.HashAlgorithm]] = {
+    _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(),
@@ -278,6 +283,114 @@ def certificates(self, *, allow_expired: bool) -> list[Certificate]:
         return self._certificates
 
 
+class SigningConfig:
+    """
+    Signing configuration for a Sigstore instance.
+    """
+
+    class SigningConfigType(str, Enum):
+        """
+        Known Sigstore signing config media types.
+        """
+
+        SIGNING_CONFIG_0_2 = "application/vnd.dev.sigstore.signingconfig.v0.2+json"
+
+        def __str__(self) -> str:
+            """Returns the variant's string value."""
+            return self.value
+
+    def __init__(self, inner: _SigningConfig):
+        """
+        Construct a new `SigningConfig`.
+
+        @api private
+        """
+        self._inner = inner
+        self._verify()
+
+    def _verify(self) -> None:
+        """
+        Performs various feats of heroism to ensure that the signing config
+        is well-formed.
+        """
+
+        # must have a recognized media type.
+        try:
+            SigningConfig.SigningConfigType(self._inner.media_type)
+        except ValueError:
+            raise Error(f"unsupported signing config format: {self._inner.media_type}")
+
+        # currently not supporting other select modes
+        # TODO: Support other modes ensuring tsa_urls() and tlog_urls() work
+        if self._inner.rekor_tlog_config.selector != ServiceSelector.ANY:
+            raise Error(
+                f"unsupported tlog selector {self._inner.rekor_tlog_config.selector}"
+            )
+        if self._inner.tsa_config.selector != ServiceSelector.ANY:
+            raise Error(f"unsupported TSA selector {self._inner.tsa_config.selector}")
+
+    @classmethod
+    def from_file(
+        cls,
+        path: str,
+    ) -> SigningConfig:
+        """Create a new signing config from file"""
+        inner = _SigningConfig().from_json(Path(path).read_bytes())
+        return cls(inner)
+
+    @staticmethod
+    def _get_valid_service_url(services: list[Service]) -> str | None:
+        for service in services:
+            if service.major_api_version != 1:
+                continue
+
+            if not _is_timerange_valid(service.valid_for, allow_expired=False):
+                continue
+            return service.url
+        return None
+
+    def get_tlog_urls(self) -> list[str]:
+        """
+        Returns the rekor transparency logs that client should sign with.
+        Currently only returns a single one but could in future return several
+        """
+
+        url = self._get_valid_service_url(self._inner.rekor_tlog_urls)
+        if not url:
+            raise Error("No valid Rekor transparency log found in signing config")
+        return [url]
+
+    def get_fulcio_url(self) -> str:
+        """
+        Returns url for the fulcio instance that client should use to get a
+        signing certificate from
+        """
+        url = self._get_valid_service_url(self._inner.ca_urls)
+        if not url:
+            raise Error("No valid Fulcio CA found in signing config")
+        return url
+
+    def get_oidc_url(self) -> str:
+        """
+        Returns url for the OIDC provider that client should use to interactively
+        authenticate.
+        """
+        url = self._get_valid_service_url(self._inner.oidc_urls)
+        if not url:
+            raise Error("No valid OIDC provider found in signing config")
+        return url
+
+    def get_tsa_urls(self) -> list[str]:
+        """
+        Returns timestamp authority API end points. Currently returns a single one
+        but may return more in future.
+        """
+        url = self._get_valid_service_url(self._inner.tsa_urls)
+        if not url:
+            raise Error("No valid Timestamp Authority found in signing config")
+        return [url]
+
+
 class TrustedRoot:
     """
     The cryptographic root(s) of trust for a Sigstore instance.
@@ -473,3 +586,10 @@ def trusted_root(self) -> TrustedRoot:
         Return the interior root of trust, as a `TrustedRoot`.
         """
         return TrustedRoot(self._inner.trusted_root)
+
+    @property
+    def signing_config(self) -> SigningConfig:
+        """
+        Return the interior root of trust, as a `SigningConfig`.
+        """
+        return SigningConfig(self._inner.signing_config)

sigstore/hashes.py

@@ -60,4 +60,4 @@ def __str__(self) -> str:
         """
         Returns a str representation of this `Hashed`.
         """
-        return f"{self.algorithm.name}:{self.digest.hex()}"
+        return f"{HashAlgorithm(self.algorithm)}:{self.digest.hex()}"

sigstore/models.py

@@ -19,6 +19,7 @@
 from __future__ import annotations
 
 import base64
+import json
 import logging
 import typing
 from enum import Enum
@@ -470,18 +471,18 @@ def _verify(self) -> None:
             Bundle.BundleType.BUNDLE_0_3_ALT,
         ):
             # For "v3" bundles, the signing certificate is the only one present.
+            if not self._inner.verification_material.certificate:
+                raise InvalidBundle("expected certificate in bundle")
+
             leaf_cert = load_der_x509_certificate(
                 self._inner.verification_material.certificate.raw_bytes
             )
         else:
             # In older bundles, there is an entire pool (misleadingly called
             # a chain) of certificates, the first of which is the signing
             # certificate.
-            certs = (
-                self._inner.verification_material.x509_certificate_chain.certificates
-            )
-
-            if len(certs) == 0:
+            chain = self._inner.verification_material.x509_certificate_chain
+            if not chain or not chain.certificates:
                 raise InvalidBundle("expected non-empty certificate chain in bundle")
 
             # Per client policy in protobuf-specs: the first entry in the chain
@@ -493,7 +494,7 @@ def _verify(self) -> None:
             # and intermediate CAs, so we issue warnings and not hard errors
             # in those cases.
             leaf_cert, *chain_certs = (
-                load_der_x509_certificate(cert.raw_bytes) for cert in certs
+                load_der_x509_certificate(cert.raw_bytes) for cert in chain.certificates
             )
             if not cert_is_leaf(leaf_cert):
                 raise InvalidBundle(
@@ -580,8 +581,8 @@ def _dsse_envelope(self) -> dsse.Envelope | None:
 
         @private
         """
-        if self._inner.dsse_envelope:
-            return dsse.Envelope(self._inner.dsse_envelope)
+        if self._inner.is_set("dsse_envelope"):
+            return dsse.Envelope(self._inner.dsse_envelope)  # type: ignore[arg-type]
         return None
 
     @property
@@ -593,7 +594,7 @@ def signature(self) -> bytes:
         return (
             self._dsse_envelope.signature
             if self._dsse_envelope
-            else self._inner.message_signature.signature
+            else self._inner.message_signature.signature  # type: ignore[union-attr]
         )
 
     @property
@@ -608,7 +609,7 @@ def from_json(cls, raw: bytes | str) -> Bundle:
         """
         Deserialize the given Sigstore bundle.
         """
-        inner = _Bundle().from_json(raw)
+        inner = _Bundle.from_dict(json.loads(raw))
         return cls(inner)
 
     def to_json(self) -> str:
@@ -627,7 +628,10 @@ def _to_parts(
         """
 
         content: common_v1.MessageSignature | dsse.Envelope
-        content = self._dsse_envelope or self._inner.message_signature
+        if self._dsse_envelope:
+            content = self._dsse_envelope
+        else:
+            content = self._inner.message_signature  # type: ignore[assignment]
 
         return (self.signing_certificate, content, self.log_entry)
 
@@ -654,30 +658,32 @@ def _from_parts(
         @private
         """
 
-        inner = _Bundle(
-            media_type=Bundle.BundleType.BUNDLE_0_3.value,
-            verification_material=bundle_v1.VerificationMaterial(
-                certificate=common_v1.X509Certificate(cert.public_bytes(Encoding.DER)),
-            ),
+        timestamp_verifcation_data = bundle_v1.TimestampVerificationData(
+            rfc3161_timestamps=[]
         )
+        if signed_timestamp is not None:
+            timestamp_verifcation_data.rfc3161_timestamps.extend(
+                [
+                    Rfc3161SignedTimestamp(signed_timestamp=response.as_bytes())
+                    for response in signed_timestamp
+                ]
+            )
 
         # Fill in the appropriate variants.
         if isinstance(content, common_v1.MessageSignature):
-            inner.message_signature = content
+            # mypy will be mystified if types are specified here
+            content_dict: dict[str, Any] = {"message_signature": content}
         else:
-            inner.dsse_envelope = content._inner
+            content_dict = {"dsse_envelope": content._inner}
 
-        tlog_entry = log_entry._to_rekor()
-        inner.verification_material.tlog_entries = [tlog_entry]
-
-        if signed_timestamp is not None:
-            inner.verification_material.timestamp_verification_data = (
-                bundle_v1.TimestampVerificationData(
-                    rfc3161_timestamps=[
-                        Rfc3161SignedTimestamp(signed_timestamp=response.as_bytes())
-                        for response in signed_timestamp
-                    ]
-                )
-            )
+        inner = _Bundle(
+            media_type=Bundle.BundleType.BUNDLE_0_3.value,
+            verification_material=bundle_v1.VerificationMaterial(
+                certificate=common_v1.X509Certificate(cert.public_bytes(Encoding.DER)),
+                tlog_entries=[log_entry._to_rekor()],
+                timestamp_verification_data=timestamp_verifcation_data,
+            ),
+            **content_dict,
+        )
 
         return cls(inner)

sigstore/sign.py

@@ -352,13 +352,13 @@ def _from_trust_config(cls, trust_config: ClientTrustConfig) -> SigningContext:
 
         @api private
         """
+        signing_config = trust_config.signing_config
         return cls(
-            fulcio=FulcioClient(trust_config._inner.signing_config.ca_url),
-            rekor=RekorClient(trust_config._inner.signing_config.tlog_urls[0]),
+            fulcio=FulcioClient(signing_config.get_fulcio_url()),
+            rekor=RekorClient(signing_config.get_tlog_urls()[0]),
             trusted_root=trust_config.trusted_root,
             tsa_clients=[
-                TimestampAuthorityClient(tsa_url)
-                for tsa_url in trust_config._inner.signing_config.tsa_urls
+                TimestampAuthorityClient(url) for url in signing_config.get_tsa_urls()
             ],
         )