Merge pull request #1 from javanlacerda/javan.refactor.ct-key-management

move ct keyring responsibilities to trustedroot

sigstore/_cli.py

@@ -27,7 +27,6 @@
 from sigstore_protobuf_specs.dev.sigstore.bundle.v1 import Bundle
 
 from sigstore import __version__
-from sigstore._internal.ctfe import CTKeyring
 from sigstore._internal.fulcio.client import (
     DEFAULT_FULCIO_URL,
     ExpiredCertificate,
@@ -37,7 +36,7 @@
     DEFAULT_REKOR_URL,
     RekorClient,
 )
-from sigstore._internal.trustroot import Keyring, KeyringPurpose, TrustedRoot
+from sigstore._internal.trustroot import KeyringPurpose, TrustedRoot
 from sigstore._utils import PEMCert, cert_der_to_pem, sha256_digest
 from sigstore.errors import Error
 from sigstore.oidc import (
@@ -649,15 +648,10 @@ def _sign(args: argparse.Namespace) -> None:
     else:
         # Assume "production" trust root if no keys are given as arguments
         trusted_root = TrustedRoot.production(args=args, purpose=KeyringPurpose.SIGN)
-        if args.ctfe_pem is not None:
-            ctfe_keys = [args.ctfe_pem.read()]
-        else:
-            ctfe_keys = trusted_root.get_ctfe_keys()
 
+        ct_keyring = trusted_root.ct_keyring()
         rekor_keyring = trusted_root.rekor_keyring()
 
-        ct_keyring = CTKeyring(Keyring(ctfe_keys))
-
         signing_ctx = SigningContext(
             fulcio=FulcioClient(args.fulcio_url),
             rekor=RekorClient(args.rekor_url, rekor_keyring, ct_keyring),
@@ -819,13 +813,12 @@ def _collect_verification_state(
             _die(args, "Custom Rekor URL used without specifying --certificate-chain")
 
         trusted_root = TrustedRoot.production(args=args, purpose=KeyringPurpose.VERIFY)
-        ct_keys = trusted_root.get_ctfe_keys()
 
         verifier = Verifier(
             rekor=RekorClient(
                 url=args.rekor_url,
                 rekor_keyring=trusted_root.rekor_keyring(),
-                ct_keyring=CTKeyring(Keyring(ct_keys)),
+                ct_keyring=trusted_root.ct_keyring(),
             ),
             trusted_root=trusted_root,
         )

sigstore/_internal/rekor/client.py

@@ -28,8 +28,7 @@
 import rekor_types
 import requests
 
-from sigstore._internal.ctfe import CTKeyring
-from sigstore._internal.trustroot import Keyring, RekorKeyring, TrustedRoot
+from sigstore._internal.trustroot import CTKeyring, RekorKeyring, TrustedRoot
 from sigstore.transparency import LogEntry
 
 logger = logging.getLogger(__name__)
@@ -251,13 +250,10 @@ def production(cls, trust_root: TrustedRoot) -> RekorClient:
 
         trust_root must be a `TrustedRoot` for the production TUF repository.
         """
-        rekor_keyring = trust_root.rekor_keyring()
-        ctfe_keys = trust_root.get_ctfe_keys()
-
         return cls(
             DEFAULT_REKOR_URL,
-            rekor_keyring,
-            CTKeyring(Keyring(ctfe_keys)),
+            rekor_keyring=trust_root.rekor_keyring(),
+            ct_keyring=trust_root.ct_keyring(),
         )
 
     @classmethod
@@ -268,12 +264,12 @@ def staging(cls, trust_root: TrustedRoot) -> RekorClient:
         trust_root must be a `TrustedRoot` for the staging TUF repository.
         """
         rekor_keyring = trust_root.rekor_keyring()
-        ctfe_keys = trust_root.get_ctfe_keys()
+        ctfe_keys = trust_root.ct_keyring()
 
         return cls(
             STAGING_REKOR_URL,
             rekor_keyring,
-            CTKeyring(Keyring(ctfe_keys)),
+            ctfe_keys,
         )
 
     @property

sigstore/_internal/sct.py

@@ -36,8 +36,8 @@
 )
 from cryptography.x509.oid import ExtendedKeyUsageOID
 
-from sigstore._internal.ctfe import CTKeyring
 from sigstore._internal.trustroot import (
+    CTKeyring,
     KeyringError,
     KeyringLookupError,
     KeyringSignatureError,

sigstore/_internal/trustroot.py

@@ -173,6 +173,7 @@ def verify(self, *, key_id: KeyID, signature: bytes, data: bytes) -> None:
 
 
 RekorKeyring = NewType("RekorKeyring", Keyring)
+CTKeyring = NewType("CTKeyring", Keyring)
 
 
 class KeyringPurpose(str, Enum):
@@ -292,14 +293,20 @@ def rekor_keyring(self) -> RekorKeyring:
 
         return RekorKeyring(self._get_rekor_keys())
 
-    def get_ctfe_keys(self) -> list[bytes]:
+    def ct_keyring(self) -> CTKeyring:
+        """Return public key contents given certificate authorities."""
+
+        return CTKeyring(self._get_ctfe_keys())
+
+    def _get_ctfe_keys(self) -> Keyring:
         """Return the CTFE public keys contents."""
-        # TODO: get purpose as argument
-        purpose = KeyringPurpose.VERIFY
-        ctfes: list[bytes] = list(self._get_tlog_keys(self.ctlogs, purpose))
+        if self.args and self.args.ctfe_pem:
+            ctfes = [self.args.ctfe_pem.read()]
+        else:
+            ctfes = list(self._get_tlog_keys(self.ctlogs, self.purpose))
         if not ctfes:
             raise MetadataError("CTFE keys not found in trusted root")
-        return ctfes
+        return Keyring(ctfes)
 
     def _get_rekor_keys(self) -> Keyring:
         """Return the rekor public key content."""

test/unit/internal/test_ctfe.py

@@ -15,8 +15,7 @@
 import pretend
 import pytest
 
-from sigstore._internal.ctfe import CTKeyring
-from sigstore._internal.trustroot import Keyring, KeyringLookupError
+from sigstore._internal.trustroot import CTKeyring, Keyring, KeyringLookupError
 
 
 class TestCTKeyring:

test/unit/internal/test_trust_root.py

@@ -53,7 +53,7 @@ def test_trust_root_tuf_caches_and_requests(mock_staging_tuf, tuf_dirs):
     assert reqs == expected_requests
     assert fail_reqs == expected_fail_reqs
 
-    trust_root.get_ctfe_keys()
+    trust_root.ct_keyring()
     trust_root.rekor_keyring()
 
     # no new requests
@@ -69,7 +69,7 @@ def test_trust_root_tuf_caches_and_requests(mock_staging_tuf, tuf_dirs):
     assert reqs == expected_requests
     assert fail_reqs == expected_fail_reqs
 
-    trust_root.get_ctfe_keys()
+    trust_root.ct_keyring()
     trust_root.rekor_keyring()
     # Expect no requests
     assert reqs == expected_requests
@@ -91,7 +91,7 @@ def test_trust_root_tuf_offline(mock_staging_tuf, tuf_dirs):
     assert reqs == {}
     assert fail_reqs == {}
 
-    trust_root.get_ctfe_keys()
+    trust_root.ct_keyring()
     trust_root.rekor_keyring()
 
     # Still no requests
@@ -159,20 +159,20 @@ def _pem_keys(keys):
 
     # Assert that trust root from TUF contains the expected keys/certs
     trust_root = TrustedRoot.staging(purpose=KeyringPurpose.VERIFY)
-    assert ctfe_keys[0] in _der_keys(trust_root.get_ctfe_keys())
+    assert ctfe_keys[0] in get_public_bytes(trust_root.ct_keyring()._keyring.values())
     assert get_public_bytes(trust_root.rekor_keyring()._keyring.values()) == rekor_keys
     assert trust_root.get_fulcio_certs() == fulcio_certs
 
     # Assert that trust root from offline TUF contains the expected keys/certs
     trust_root = TrustedRoot.staging(offline=True, purpose=KeyringPurpose.VERIFY)
-    assert ctfe_keys[0] in _der_keys(trust_root.get_ctfe_keys())
+    assert ctfe_keys[0] in get_public_bytes(trust_root.ct_keyring()._keyring.values())
     assert get_public_bytes(trust_root.rekor_keyring()._keyring.values()) == rekor_keys
     assert trust_root.get_fulcio_certs() == fulcio_certs
 
     # Assert that trust root from file contains the expected keys/certs
     path = tuf_asset.target_path("trusted_root.json")
     trust_root = TrustedRoot.from_file(path)
-    assert ctfe_keys[0] in _der_keys(trust_root.get_ctfe_keys())
+    assert ctfe_keys[0] in get_public_bytes(trust_root.ct_keyring()._keyring.values())
     assert get_public_bytes(trust_root.rekor_keyring()._keyring.values()) == rekor_keys
     assert trust_root.get_fulcio_certs() == fulcio_certs
 
@@ -186,7 +186,7 @@ def test_trust_root_tuf_ctfe_keys_error(monkeypatch):
     trust_root = TrustedRoot.staging(offline=True)
     monkeypatch.setattr(trust_root, "ctlogs", [])
     with pytest.raises(Exception, match="CTFE keys not found in trusted root"):
-        trust_root.get_ctfe_keys()
+        trust_root.ct_keyring()
 
 
 def test_trust_root_fulcio_certs_error(tuf_asset, monkeypatch):