refactors and adding trusted_root to Verifier and SigningContext

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

Author: javanlacerda

sigstore/_cli.py

@@ -23,7 +23,6 @@
 from textwrap import dedent
 from typing import NoReturn, Optional, TextIO, Union, cast
 
-from cryptography.x509 import load_pem_x509_certificates
 from rich.logging import RichHandler
 from sigstore_protobuf_specs.dev.sigstore.bundle.v1 import Bundle
 
@@ -651,17 +650,20 @@ def _sign(args: argparse.Namespace) -> None:
     else:
         # Assume "production" trust root if no keys are given as arguments
         trusted_root = TrustedRoot.production()
+        trusted_root.set_args(args)
         if args.ctfe_pem is not None:
             ctfe_keys = [args.ctfe_pem.read()]
         else:
             ctfe_keys = trusted_root.get_ctfe_keys()
 
-        ct_keyring = CTKeyring(Keyring(ctfe_keys))
         rekor_keyring = trusted_root.rekor_keyring(KeyringPurpose.SIGN)
 
+        ct_keyring = CTKeyring(Keyring(ctfe_keys))
+
         signing_ctx = SigningContext(
             fulcio=FulcioClient(args.fulcio_url),
             rekor=RekorClient(args.rekor_url, rekor_keyring, ct_keyring),
+            trusted_root=trusted_root,
         )
 
     # The order of precedence for identities is as follows:
@@ -809,24 +811,17 @@ def _collect_verification_state(
                 args,
                 f"Missing verification materials for {(file)}: {', '.join(missing)}",
             )
-    purpose = KeyringPurpose.VERIFY
     if args.staging:
         logger.debug("verify: staging instances requested")
-        verifier = Verifier.staging(purpose)
+        verifier = Verifier.staging()
     elif args.rekor_url == DEFAULT_REKOR_URL:
-        verifier = Verifier.production(purpose)
+        verifier = Verifier.production()
     else:
         if not args.certificate_chain:
             _die(args, "Custom Rekor URL used without specifying --certificate-chain")
 
-        try:
-            certificate_chain = load_pem_x509_certificates(
-                args.certificate_chain.read()
-            )
-        except ValueError as error:
-            _die(args, f"Invalid certificate chain: {error}")
-
         trusted_root = TrustedRoot.production()
+        trusted_root.set_args(args=args)
         ct_keys = trusted_root.get_ctfe_keys()
 
         verifier = Verifier(
@@ -835,7 +830,7 @@ def _collect_verification_state(
                 rekor_keyring=trusted_root.rekor_keyring(KeyringPurpose.VERIFY),
                 ct_keyring=CTKeyring(Keyring(ct_keys)),
             ),
-            fulcio_certificate_chain=certificate_chain,
+            trusted_root=trusted_root,
         )
 
     all_materials = []

sigstore/_internal/trustroot.py

@@ -18,12 +18,17 @@
 
 from __future__ import annotations
 
+from argparse import Namespace
 from datetime import datetime, timezone
 from enum import Enum
 from pathlib import Path
-from typing import Iterable, NewType
+from typing import Iterable, NewType, Optional
 
-from cryptography.x509 import Certificate, load_der_x509_certificate
+from cryptography.x509 import (
+    Certificate,
+    load_der_x509_certificate,
+    load_pem_x509_certificates,
+)
 from sigstore_protobuf_specs.dev.sigstore.common.v1 import TimeRange
 from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
     CertificateAuthority,
@@ -67,8 +72,8 @@ class KeyringPurpose(str, Enum):
     Keyring purpose typing
     """
 
-    SIGN = "Signing"
-    VERIFY = "Verifying"
+    SIGN = "sign"
+    VERIFY = "verify"
 
     def __str__(self) -> str:
         """Returns the purpose string value."""
@@ -78,6 +83,9 @@ def __str__(self) -> str:
 class TrustedRoot(_TrustedRoot):
     """Complete set of trusted entities for a Sigstore client"""
 
+    def __init__(self, args: Optional[Namespace] = None):
+        self.args = args
+
     @classmethod
     def from_file(cls, path: str) -> "TrustedRoot":
         """Create a new trust root from file"""
@@ -114,9 +122,10 @@ def staging(cls, offline: bool = False) -> "TrustedRoot":
 
     @staticmethod
     def _get_tlog_keys(
-        tlogs: list[TransparencyLogInstance], allow_expired: bool
+        tlogs: list[TransparencyLogInstance], purpose: KeyringPurpose
     ) -> Iterable[bytes]:
         """Return public key contents given transparency log instances."""
+        allow_expired = purpose is KeyringPurpose.VERIFY
         for key in tlogs:
             if not _is_timerange_valid(
                 key.public_key.valid_for, allow_expired=allow_expired
@@ -138,33 +147,33 @@ def _get_ca_keys(
             for cert in ca.cert_chain.certificates:
                 yield cert.raw_bytes
 
+    def set_args(self, args: Namespace) -> None:
+        self.args = args
+
     def rekor_keyring(self, purpose: KeyringPurpose) -> RekorKeyring:
         """Return public key contents given certificate authorities."""
 
-        if purpose is KeyringPurpose.VERIFY:
-            allow_expired = True
-        elif purpose is KeyringPurpose.SIGN:
-            allow_expired = False
-        else:
-            # TODO(javan): Create a specific exception
-            raise Exception()
-
-        return RekorKeyring(Keyring(self._get_rekor_keys(allow_expired)))
+        return RekorKeyring(self._get_rekor_keys(purpose))
 
     def get_ctfe_keys(self) -> list[bytes]:
         """Return the CTFE public keys contents."""
-        ctfes: list[bytes] = list(self._get_tlog_keys(self.ctlogs, True))
+        # TODO: get purpose as argument
+        purpose = KeyringPurpose.VERIFY
+        ctfes: list[bytes] = list(self._get_tlog_keys(self.ctlogs, purpose))
         if not ctfes:
             raise MetadataError("CTFE keys not found in trusted root")
         return ctfes
 
-    def _get_rekor_keys(self, allow_expired: bool) -> list[bytes]:
+    def _get_rekor_keys(self, purpose: KeyringPurpose) -> Keyring:
         """Return the rekor public key content."""
-        keys: list[bytes] = list(self._get_tlog_keys(self.tlogs, allow_expired))
-
+        keys: list[bytes]
+        if self.args and self.args.rekor_root_pubkey:
+            keys = self.args.rekor_root_pubkey.read()
+        else:
+            keys = list(self._get_tlog_keys(self.tlogs, purpose))
         if len(keys) != 1:
             raise MetadataError("Did not find one Rekor key in trusted root")
-        return keys
+        return Keyring(keys)
 
     def get_fulcio_certs(self) -> list[Certificate]:
         """Return the Fulcio certificates."""
@@ -173,11 +182,20 @@ def get_fulcio_certs(self) -> list[Certificate]:
 
         # Return expired certificates too: they are expired now but may have
         # 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)
-        ]
 
+        if self.args:
+            try:
+                certs = load_pem_x509_certificates(self.args.certificate_chain.read())
+            except ValueError as error:
+                self.args._parser.error(f"Invalid certificate chain: {error}")
+                raise ValueError("unreachable")
+        else:
+            certs = [
+                load_der_x509_certificate(c)
+                for c in self._get_ca_keys(
+                    self.certificate_authorities, allow_expired=True
+                )
+            ]
         if not certs:
             raise MetadataError("Fulcio certificates not found in trusted root")
         return certs

sigstore/sign.py

@@ -280,10 +280,7 @@ class SigningContext:
     """
 
     def __init__(
-        self,
-        *,
-        fulcio: FulcioClient,
-        rekor: RekorClient,
+        self, *, fulcio: FulcioClient, rekor: RekorClient, trusted_root: TrustedRoot
     ):
         """
         Create a new `SigningContext`.
@@ -296,29 +293,28 @@ def __init__(
         """
         self._fulcio = fulcio
         self._rekor = rekor
+        self._trusted_root = trusted_root
 
     @classmethod
     def production(cls) -> SigningContext:
         """
         Return a `SigningContext` instance configured against Sigstore's production-level services.
         """
-        trust_root = TrustedRoot.production()
-        rekor = RekorClient.production(trust_root, purpose=KeyringPurpose.SIGN)
+        trusted_root = TrustedRoot.production()
+        rekor = RekorClient.production(trusted_root, purpose=KeyringPurpose.SIGN)
         return cls(
-            fulcio=FulcioClient.production(),
-            rekor=rekor,
+            fulcio=FulcioClient.production(), rekor=rekor, trusted_root=trusted_root
         )
 
     @classmethod
     def staging(cls) -> SigningContext:
         """
         Return a `SignerContext` instance configured against Sigstore's staging-level services.
         """
-        trust_root = TrustedRoot.staging()
-        rekor = RekorClient.staging(trust_root, purpose=KeyringPurpose.SIGN)
+        trusted_root = TrustedRoot.staging()
+        rekor = RekorClient.staging(trusted_root, purpose=KeyringPurpose.SIGN)
         return cls(
-            fulcio=FulcioClient.staging(),
-            rekor=rekor,
+            fulcio=FulcioClient.staging(), rekor=rekor, trusted_root=trusted_root
         )
 
     @contextmanager

sigstore/verify/verifier.py

@@ -26,7 +26,6 @@
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.x509 import (
-    Certificate,
     ExtendedKeyUsage,
     KeyUsage,
 )
@@ -110,9 +109,7 @@ class Verifier:
     The primary API for verification operations.
     """
 
-    def __init__(
-        self, *, rekor: RekorClient, fulcio_certificate_chain: List[Certificate]
-    ):
+    def __init__(self, *, rekor: RekorClient, trusted_root: TrustedRoot):
         """
         Create a new `Verifier`.
 
@@ -125,29 +122,29 @@ def __init__(
         self._rekor = rekor
         self._fulcio_certificate_chain: List[X509] = [
             X509.from_cryptography(parent_cert)
-            for parent_cert in fulcio_certificate_chain
+            for parent_cert in trusted_root.get_fulcio_certs()
         ]
 
     @classmethod
-    def production(cls, purpose: KeyringPurpose) -> Verifier:
+    def production(cls) -> Verifier:
         """
         Return a `Verifier` instance configured against Sigstore's production-level services.
         """
-        trust_root = TrustedRoot.production()
+        trusted_root = TrustedRoot.production()
         return cls(
-            rekor=RekorClient.production(trust_root, purpose),
-            fulcio_certificate_chain=trust_root.get_fulcio_certs(),
+            rekor=RekorClient.production(trusted_root, KeyringPurpose.VERIFY),
+            trusted_root=trusted_root,
         )
 
     @classmethod
-    def staging(cls, purpose: KeyringPurpose) -> Verifier:
+    def staging(cls) -> Verifier:
         """
         Return a `Verifier` instance configured against Sigstore's staging-level services.
         """
-        trust_root = TrustedRoot.staging()
+        trusted_root = TrustedRoot.staging()
         return cls(
-            rekor=RekorClient.staging(trust_root, purpose),
-            fulcio_certificate_chain=trust_root.get_fulcio_certs(),
+            rekor=RekorClient.staging(trusted_root, KeyringPurpose.VERIFY),
+            trusted_root=trusted_root,
         )
 
     def verify(

test/unit/test_sign.py

@@ -23,7 +23,6 @@
 import sigstore.oidc
 from sigstore._internal.keyring import KeyringError, KeyringLookupError
 from sigstore._internal.sct import InvalidSCTError, InvalidSCTKeyError
-from sigstore._internal.trustroot import KeyringPurpose
 from sigstore.dsse import _StatementBuilder, _Subject
 from sigstore.hashes import Hashed
 from sigstore.sign import SigningContext
@@ -131,7 +130,7 @@ def test_sign_prehashed(staging):
     sign_ctx, verifier, identity = staging
 
     sign_ctx: SigningContext = sign_ctx()
-    verifier: Verifier = verifier(KeyringPurpose.VERIFY)
+    verifier: Verifier = verifier()
 
     input_ = secrets.token_bytes(32)
     hashed = Hashed(