sigstore: refactor purpose handling

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

Author: woodruffw

sigstore/_internal/trust.py

@@ -217,71 +217,69 @@ def __str__(self) -> str:
         return self.value
 
 
-class TrustedRoot(_TrustedRoot):
-    """Complete set of trusted entities for a Sigstore client"""
+class TrustedRoot:
+    """
+    The cryptographic root(s) of trust for a Sigstore instance.
+    """
 
-    purpose: KeyringPurpose
+    def __init__(self, inner: _TrustedRoot):
+        self._inner = inner
 
     @classmethod
     def from_file(
         cls,
         path: str,
-        purpose: KeyringPurpose = KeyringPurpose.VERIFY,
     ) -> TrustedRoot:
         """Create a new trust root from file"""
-        trusted_root: TrustedRoot = cls().from_json(Path(path).read_bytes())
-        trusted_root.purpose = purpose
-        return trusted_root
+        inner = _TrustedRoot().from_json(Path(path).read_bytes())
+        return cls(inner)
 
     @classmethod
     def from_tuf(
         cls,
         url: str,
         offline: bool = False,
-        purpose: KeyringPurpose = KeyringPurpose.VERIFY,
     ) -> TrustedRoot:
         """Create a new trust root from a TUF repository.
 
         If `offline`, will use trust root in local TUF cache. Otherwise will
         update the trust root from remote TUF repository.
         """
         path = TrustUpdater(url, offline).get_trusted_root_path()
-        return cls.from_file(path, purpose)
+        return cls.from_file(path)
 
     @classmethod
     def production(
         cls,
         offline: bool = False,
-        purpose: KeyringPurpose = KeyringPurpose.VERIFY,
     ) -> TrustedRoot:
         """Create new trust root from Sigstore production TUF repository.
 
         If `offline`, will use trust root in local TUF cache. Otherwise will
         update the trust root from remote TUF repository.
         """
-        return cls.from_tuf(DEFAULT_TUF_URL, offline, purpose)
+        return cls.from_tuf(DEFAULT_TUF_URL, offline)
 
     @classmethod
     def staging(
         cls,
         offline: bool = False,
-        purpose: KeyringPurpose = KeyringPurpose.VERIFY,
     ) -> TrustedRoot:
         """Create new trust root from Sigstore staging TUF repository.
 
         If `offline`, will use trust root in local TUF cache. Otherwise will
         update the trust root from remote TUF repository.
         """
-        return cls.from_tuf(STAGING_TUF_URL, offline, purpose)
+        return cls.from_tuf(STAGING_TUF_URL, offline)
 
     def _get_tlog_keys(
-        self, tlogs: list[TransparencyLogInstance]
+        self, tlogs: list[TransparencyLogInstance], purpose: KeyringPurpose
     ) -> Iterable[_PublicKey]:
         """
         Yields an iterator of public keys for transparency log instances that
         are suitable for `purpose`.
         """
-        allow_expired = self.purpose is KeyringPurpose.VERIFY
+        allow_expired = purpose is KeyringPurpose.VERIFY
         for tlog in tlogs:
             if not _is_timerange_valid(
                 tlog.public_key.valid_for, allow_expired=allow_expired
@@ -302,17 +300,17 @@ def _get_ca_keys(
             for cert in ca.cert_chain.certificates:
                 yield cert.raw_bytes
 
-    def rekor_keyring(self) -> RekorKeyring:
+    def rekor_keyring(self, purpose: KeyringPurpose) -> RekorKeyring:
         """Return keyring with keys for Rekor."""
 
-        keys: list[_PublicKey] = list(self._get_tlog_keys(self.tlogs))
+        keys: list[_PublicKey] = list(self._get_tlog_keys(self._inner.tlogs, purpose))
         if len(keys) != 1:
             raise MetadataError("Did not find one Rekor key in trusted root")
         return RekorKeyring(Keyring(keys))
 
-    def ct_keyring(self) -> CTKeyring:
+    def ct_keyring(self, purpose: KeyringPurpose) -> CTKeyring:
         """Return keyring with key for CTFE."""
-        ctfes: list[_PublicKey] = list(self._get_tlog_keys(self.ctlogs))
+        ctfes: list[_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))
@@ -326,7 +324,9 @@ def get_fulcio_certs(self) -> list[Certificate]:
         # been active when the certificate was used to sign.
         certs = [
             load_der_x509_certificate(c)
-            for c in self._get_ca_keys(self.certificate_authorities, allow_expired=True)
+            for c in self._get_ca_keys(
+                self._inner.certificate_authorities, allow_expired=True
+            )
         ]
         if not certs:
             raise MetadataError("Fulcio certificates not found in trusted root")

sigstore/sign.py

@@ -165,7 +165,12 @@ def _signing_cert(
             cert = certificate_response.cert
             chain = certificate_response.chain
 
-            verify_sct(sct, cert, chain, self._signing_ctx._trusted_root.ct_keyring())
+            verify_sct(
+                sct,
+                cert,
+                chain,
+                self._signing_ctx._trusted_root.ct_keyring(KeyringPurpose.SIGN),
+            )
 
             _logger.debug("Successfully verified SCT...")
 
@@ -311,21 +316,21 @@ def production(cls) -> SigningContext:
         """
         Return a `SigningContext` instance configured against Sigstore's production-level services.
         """
-        trusted_root = TrustedRoot.production(purpose=KeyringPurpose.SIGN)
-        rekor = RekorClient.production()
         return cls(
-            fulcio=FulcioClient.production(), rekor=rekor, trusted_root=trusted_root
+            fulcio=FulcioClient.production(),
+            rekor=RekorClient.production(),
+            trusted_root=TrustedRoot.production(),
         )
 
     @classmethod
     def staging(cls) -> SigningContext:
         """
         Return a `SignerContext` instance configured against Sigstore's staging-level services.
         """
-        trusted_root = TrustedRoot.staging(purpose=KeyringPurpose.SIGN)
-        rekor = RekorClient.staging()
         return cls(
-            fulcio=FulcioClient.staging(), rekor=rekor, trusted_root=trusted_root
+            fulcio=FulcioClient.staging(),
+            rekor=RekorClient.staging(),
+            trusted_root=TrustedRoot.staging(),
         )
 
     @contextmanager

sigstore/verify/verifier.py

@@ -81,21 +81,19 @@ def production(cls) -> Verifier:
         """
         Return a `Verifier` instance configured against Sigstore's production-level services.
         """
-        trusted_root = TrustedRoot.production(purpose=KeyringPurpose.VERIFY)
         return cls(
             rekor=RekorClient.production(),
-            trusted_root=trusted_root,
+            trusted_root=TrustedRoot.production(),
         )
 
     @classmethod
     def staging(cls) -> Verifier:
         """
         Return a `Verifier` instance configured against Sigstore's staging-level services.
         """
-        trusted_root = TrustedRoot.staging(purpose=KeyringPurpose.VERIFY)
         return cls(
             rekor=RekorClient.staging(),
-            trusted_root=trusted_root,
+            trusted_root=TrustedRoot.staging(),
         )
 
     def _verify_common_signing_cert(
@@ -166,7 +164,7 @@ def _verify_common_signing_cert(
                 sct,
                 cert,
                 [parent_cert.to_cryptography() for parent_cert in chain],
-                self._trusted_root.ct_keyring(),
+                self._trusted_root.ct_keyring(KeyringPurpose.VERIFY),
             )
         except VerificationError as e:
             raise VerificationError(f"failed to verify SCT on signing certificate: {e}")
@@ -190,7 +188,7 @@ def _verify_common_signing_cert(
         # (5): verify the inclusion promise for the log entry, if present.
         entry = bundle.log_entry
         try:
-            entry._verify(self._trusted_root.rekor_keyring())
+            entry._verify(self._trusted_root.rekor_keyring(KeyringPurpose.VERIFY))
         except VerificationError as exc:
             raise VerificationError(f"invalid log entry: {exc}")