removing keys from rekor client

Signed-off-by: Javan lacerda <javanlacerda@google.com>

Author: javanlacerda

sigstore/_cli.py

@@ -649,12 +649,9 @@ def _sign(args: argparse.Namespace) -> None:
         # Assume "production" trust root if no keys are given as arguments
         trusted_root = TrustedRoot.production(args=args, purpose=KeyringPurpose.SIGN)
 
-        ct_keyring = trusted_root.ct_keyring()
-        rekor_keyring = trusted_root.rekor_keyring()
-
         signing_ctx = SigningContext(
             fulcio=FulcioClient(args.fulcio_url),
-            rekor=RekorClient(args.rekor_url, rekor_keyring, ct_keyring),
+            rekor=RekorClient(args.rekor_url),
             trusted_root=trusted_root,
         )
 
@@ -817,8 +814,6 @@ def _collect_verification_state(
         verifier = Verifier(
             rekor=RekorClient(
                 url=args.rekor_url,
-                rekor_keyring=trusted_root.rekor_keyring(),
-                ct_keyring=trusted_root.ct_keyring(),
             ),
             trusted_root=trusted_root,
         )

sigstore/_internal/rekor/checkpoint.py

@@ -26,8 +26,7 @@
 
 from pydantic import BaseModel, Field, StrictStr
 
-from sigstore._internal.rekor.client import RekorClient
-from sigstore._internal.trustroot import KeyringSignatureError
+from sigstore._internal.trustroot import KeyringSignatureError, RekorKeyring
 from sigstore._utils import KeyID
 from sigstore.transparency import LogEntry
 
@@ -163,7 +162,7 @@ def from_text(cls, text: str) -> SignedNote:
 
         return cls(note=header, signatures=signatures)
 
-    def verify(self, client: RekorClient, key_id: KeyID) -> None:
+    def verify(self, rekor_keyring: RekorKeyring, key_id: KeyID) -> None:
         """
         Verify the `SignedNote` with using the given RekorClient by verifying each contained signature.
         """
@@ -175,7 +174,7 @@ def verify(self, client: RekorClient, key_id: KeyID) -> None:
                 raise CheckpointError("sig_hash hint does not match expected key_id")
 
             try:
-                client._rekor_keyring.verify(
+                rekor_keyring.verify(
                     key_id=key_id, signature=base64.b64decode(sig.signature), data=note
                 )
             except KeyringSignatureError as sig_err:
@@ -202,7 +201,7 @@ def from_text(cls, text: str) -> SignedCheckpoint:
         return cls(signed_note=signed_note, checkpoint=checkpoint)
 
 
-def verify_checkpoint(client: RekorClient, entry: LogEntry) -> None:
+def verify_checkpoint(rekor_keyring: RekorKeyring, entry: LogEntry) -> None:
     """
     Verify the inclusion proof's checkpoint.
     """
@@ -215,7 +214,9 @@ def verify_checkpoint(client: RekorClient, entry: LogEntry) -> None:
     # 1) verify the signature on the checkpoint
     # 2) verify the root hash in the checkpoint matches the root hash from the inclusion proof.
     signed_checkpoint = SignedCheckpoint.from_text(inclusion_proof.checkpoint)
-    signed_checkpoint.signed_note.verify(client, KeyID(bytes.fromhex(entry.log_id)))
+    signed_checkpoint.signed_note.verify(
+        rekor_keyring, KeyID(bytes.fromhex(entry.log_id))
+    )
 
     checkpoint_hash = signed_checkpoint.checkpoint.log_hash
     root_hash = inclusion_proof.root_hash

sigstore/_internal/rekor/client.py

@@ -28,7 +28,6 @@
 import rekor_types
 import requests
 
-from sigstore._internal.trustroot import CTKeyring, RekorKeyring, TrustedRoot
 from sigstore.transparency import LogEntry
 
 logger = logging.getLogger(__name__)
@@ -222,9 +221,7 @@ def post(
 class RekorClient:
     """The internal Rekor client"""
 
-    def __init__(
-        self, url: str, rekor_keyring: RekorKeyring, ct_keyring: CTKeyring
-    ) -> None:
+    def __init__(self, url: str) -> None:
         """
         Create a new `RekorClient` from the given URL.
         """
@@ -234,43 +231,31 @@ def __init__(
             {"Content-Type": "application/json", "Accept": "application/json"}
         )
 
-        self._ct_keyring = ct_keyring
-        self._rekor_keyring = rekor_keyring
-
     def __del__(self) -> None:
         """
         Terminates the underlying network session.
         """
         self.session.close()
 
     @classmethod
-    def production(cls, trust_root: TrustedRoot) -> RekorClient:
+    def production(cls) -> RekorClient:
         """
         Returns a `RekorClient` populated with the default Rekor production instance.
 
         trust_root must be a `TrustedRoot` for the production TUF repository.
         """
         return cls(
             DEFAULT_REKOR_URL,
-            rekor_keyring=trust_root.rekor_keyring(),
-            ct_keyring=trust_root.ct_keyring(),
         )
 
     @classmethod
-    def staging(cls, trust_root: TrustedRoot) -> RekorClient:
+    def staging(cls) -> RekorClient:
         """
         Returns a `RekorClient` populated with the default Rekor staging instance.
 
         trust_root must be a `TrustedRoot` for the staging TUF repository.
         """
-        rekor_keyring = trust_root.rekor_keyring()
-        ctfe_keys = trust_root.ct_keyring()
-
-        return cls(
-            STAGING_REKOR_URL,
-            rekor_keyring,
-            ctfe_keys,
-        )
+        return cls(STAGING_REKOR_URL)
 
     @property
     def log(self) -> RekorLog:

sigstore/_internal/set.py

@@ -20,7 +20,7 @@
 
 from cryptography.exceptions import InvalidSignature
 
-from sigstore._internal.rekor import RekorClient
+from sigstore._internal.trustroot import RekorKeyring
 from sigstore._utils import KeyID
 from sigstore.transparency import LogEntry
 
@@ -33,7 +33,7 @@ class InvalidSETError(Exception):
     pass
 
 
-def verify_set(client: RekorClient, entry: LogEntry) -> None:
+def verify_set(rekor_keyring: RekorKeyring, entry: LogEntry) -> None:
     """
     Verify the inclusion promise (Signed Entry Timestamp) for a given transparency log
     `entry` using the given `client`.
@@ -46,7 +46,7 @@ def verify_set(client: RekorClient, entry: LogEntry) -> None:
     signed_entry_ts = base64.b64decode(entry.inclusion_promise)
 
     try:
-        client._rekor_keyring.verify(
+        rekor_keyring.verify(
             key_id=KeyID(bytes.fromhex(entry.log_id)),
             signature=signed_entry_ts,
             data=entry.encode_canonical(),

sigstore/sign.py

@@ -203,7 +203,7 @@ def sign(
         cert = certificate_response.cert
         chain = certificate_response.chain
 
-        verify_sct(sct, cert, chain, self._signing_ctx._rekor._ct_keyring)
+        verify_sct(sct, cert, chain, self._signing_ctx._trusted_root.ct_keyring())
 
         _logger.debug("Successfully verified SCT...")
 
@@ -301,7 +301,7 @@ 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(trusted_root)
+        rekor = RekorClient.production()
         return cls(
             fulcio=FulcioClient.production(), rekor=rekor, trusted_root=trusted_root
         )
@@ -312,7 +312,7 @@ 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(trusted_root)
+        rekor = RekorClient.staging()
         return cls(
             fulcio=FulcioClient.staging(), rekor=rekor, trusted_root=trusted_root
         )

sigstore/verify/verifier.py

@@ -124,6 +124,7 @@ def __init__(self, *, rekor: RekorClient, trusted_root: TrustedRoot):
             X509.from_cryptography(parent_cert)
             for parent_cert in trusted_root.get_fulcio_certs()
         ]
+        self.trusted_root = trusted_root
 
     @classmethod
     def production(cls) -> Verifier:
@@ -132,7 +133,7 @@ def production(cls) -> Verifier:
         """
         trusted_root = TrustedRoot.production(purpose=KeyringPurpose.VERIFY)
         return cls(
-            rekor=RekorClient.production(trusted_root),
+            rekor=RekorClient.production(),
             trusted_root=trusted_root,
         )
 
@@ -143,7 +144,7 @@ def staging(cls) -> Verifier:
         """
         trusted_root = TrustedRoot.staging(purpose=KeyringPurpose.VERIFY)
         return cls(
-            rekor=RekorClient.staging(trusted_root),
+            rekor=RekorClient.staging(),
             trusted_root=trusted_root,
         )
 
@@ -225,7 +226,7 @@ def verify(
             sct,
             materials.certificate,
             [parent_cert.to_cryptography() for parent_cert in chain],
-            self._rekor._ct_keyring,
+            self.trusted_root.ct_keyring(),
         )
 
         # 3) Check that the signing certificate contains the proof claim as the subject
@@ -293,7 +294,7 @@ def verify(
                 )
 
             try:
-                verify_checkpoint(self._rekor, entry)
+                verify_checkpoint(self.trusted_root.rekor_keyring(), entry)
             except CheckpointError as exc:
                 return VerificationFailure(reason=f"invalid Rekor root hash: {exc}")
 
@@ -313,7 +314,7 @@ def verify(
         # 7) Verify the Signed Entry Timestamp (SET) supplied by Rekor for this artifact
         if entry.inclusion_promise:
             try:
-                verify_set(self._rekor, entry)
+                verify_set(self.trusted_root.rekor_keyring(), entry)
                 _logger.debug(
                     f"successfully verified inclusion promise: index={entry.log_index}"
                 )

test/unit/test_sign.py

@@ -69,15 +69,18 @@ def test_sct_verify_keyring_lookup_error(signer_and_ident, monkeypatch):
 
     # a signer whose keyring always fails to lookup a given key.
     ctx: SigningContext = ctx()
-    ctx._rekor._ct_keyring = pretend.stub(verify=pretend.raiser(KeyringLookupError))
+    mock = pretend.stub(
+        ct_keyring=lambda: pretend.stub(verify=pretend.raiser(KeyringLookupError))
+    )
+    ctx._trusted_root = mock
     assert identity is not None
 
     payload = secrets.token_bytes(32)
-
     with pytest.raises(
         InvalidSCTError,
     ) as excinfo:
         with ctx.signer(identity) as signer:
+            print(signer.sign(payload))
             signer.sign(payload)
 
     # The exception subclass is the one we expect.
@@ -91,6 +94,10 @@ def test_sct_verify_keyring_error(signer_and_ident, monkeypatch):
 
     # a signer whose keyring throws an internal error.
     ctx: SigningContext = ctx()
+    mock = pretend.stub(
+        ct_keyring=lambda: pretend.stub(verify=pretend.raiser(KeyringLookupError))
+    )
+    ctx._trusted_root = mock
     ctx._rekor._ct_keyring = pretend.stub(verify=pretend.raiser(KeyringError))
     assert identity is not None
 
@@ -132,6 +139,8 @@ def test_sign_prehashed(staging):
     sign_ctx: SigningContext = sign_ctx()
     verifier: Verifier = verifier()
 
+    # mock = pretend.stub(ct_keyring=lambda: pretend.stub(verify=pretend.raiser(KeyringLookupError)))
+    # sign_ctx._trusted_root = mock
     input_ = secrets.token_bytes(32)
     hashed = Hashed(
         digest=hashlib.sha256(input_).digest(), algorithm=HashAlgorithm.SHA2_256

test/unit/verify/test_models.py

@@ -17,7 +17,6 @@
 from sigstore_protobuf_specs.dev.sigstore.common.v1 import HashAlgorithm
 
 from sigstore._internal.rekor.client import RekorClient
-from sigstore._internal.trustroot import KeyringPurpose, TrustedRoot
 from sigstore._utils import _sha256_streaming
 from sigstore.hashes import Hashed
 from sigstore.verify.models import (
@@ -52,8 +51,7 @@ def test_verification_materials_retrieves_rekor_entry(self, signing_materials):
         file, materials = signing_materials("a.txt")
         assert materials._rekor_entry is None
 
-        trust_root = TrustedRoot.staging(purpose=KeyringPurpose.VERIFY)
-        client = RekorClient.staging(trust_root)
+        client = RekorClient.staging()
 
         with file.open(mode="rb", buffering=0) as input_:
             digest = _sha256_streaming(input_)