Add option to sign multiple artifacts with the same key and certificate

CHANGELOG.md

@@ -41,6 +41,14 @@ All versions prior to 0.9.0 are untracked.
   `sigstore verify identity`, as it was during the 1.0 release series
   ([#642](https://github.com/sigstore/sigstore-python/pull/642))
 
+* API change: the `Signer` API has been broken up into `SigningContext`
+  and `Signer`, allowing a `SigningContext` to create individual `Signer`
+  instances that correspond to a single `IdentityToken`. This new API
+  also enables ephemeral key and certificate reuse across multiple inputs,
+  reducing the number of cryptographic operations and network roundtrips
+  required when signing more than one input
+  ([#645](https://github.com/sigstore/sigstore-python/pull/645))
+
 ### Fixed
 
 * Fixed a case where `sigstore verify` would fail to verify an otherwise valid

sigstore/_cli.py

@@ -28,7 +28,11 @@
 
 from sigstore import __version__
 from sigstore._internal.ctfe import CTKeyring
-from sigstore._internal.fulcio.client import DEFAULT_FULCIO_URL, FulcioClient
+from sigstore._internal.fulcio.client import (
+    DEFAULT_FULCIO_URL,
+    ExpiredCertificate,
+    FulcioClient,
+)
 from sigstore._internal.keyring import Keyring
 from sigstore._internal.rekor.client import (
     DEFAULT_REKOR_URL,
@@ -41,11 +45,12 @@
 from sigstore.oidc import (
     DEFAULT_OAUTH_ISSUER_URL,
     STAGING_OAUTH_ISSUER_URL,
+    ExpiredIdentity,
     IdentityToken,
     Issuer,
     detect_credential,
 )
-from sigstore.sign import Signer
+from sigstore.sign import SigningContext
 from sigstore.transparency import LogEntry
 from sigstore.verify import (
     CertificateVerificationFailure,
@@ -620,13 +625,13 @@ def _sign(args: argparse.Namespace) -> None:
             "bundle": bundle,
         }
 
-    # Select the signer to use.
+    # Select the signing context to use.
     if args.staging:
         logger.debug("sign: staging instances requested")
-        signer = Signer.staging()
+        signing_ctx = SigningContext.staging()
         args.oidc_issuer = STAGING_OAUTH_ISSUER_URL
     elif args.fulcio_url == DEFAULT_FULCIO_URL and args.rekor_url == DEFAULT_REKOR_URL:
-        signer = Signer.production()
+        signing_ctx = SigningContext.production()
     else:
         # Assume "production" keys if none are given as arguments
         updater = TrustUpdater.production()
@@ -642,7 +647,7 @@ def _sign(args: argparse.Namespace) -> None:
         ct_keyring = CTKeyring(Keyring(ctfe_keys))
         rekor_keyring = RekorKeyring(Keyring(rekor_keys))
 
-        signer = Signer(
+        signing_ctx = SigningContext(
             fulcio=FulcioClient(args.fulcio_url),
             rekor=RekorClient(args.rekor_url, rekor_keyring, ct_keyring),
         )
@@ -661,38 +666,46 @@ def _sign(args: argparse.Namespace) -> None:
     if not identity:
         _die(args, "No identity token supplied or detected!")
 
-    for file, outputs in output_map.items():
-        logger.debug(f"signing for {file.name}")
-        with file.open(mode="rb", buffering=0) as io:
-            result = signer.sign(
-                input_=io,
-                identity=identity,
-            )
+    with signing_ctx.signer(identity) as signer:
+        for file, outputs in output_map.items():
+            logger.debug(f"signing for {file.name}")
+            with file.open(mode="rb", buffering=0) as io:
+                try:
+                    result = signer.sign(input_=io)
+                except ExpiredIdentity as exp_identity:
+                    print("Signature failed: identity token has expired")
+                    raise exp_identity
 
-        print("Using ephemeral certificate:")
-        print(result.cert_pem)
+                except ExpiredCertificate as exp_certificate:
+                    print("Signature failed: Fulcio signing certificate has expired")
+                    raise exp_certificate
 
-        print(f"Transparency log entry created at index: {result.log_entry.log_index}")
+            print("Using ephemeral certificate:")
+            print(result.cert_pem)
 
-        sig_output: TextIO
-        if outputs["sig"] is not None:
-            sig_output = outputs["sig"].open("w")
-        else:
-            sig_output = sys.stdout
+            print(
+                f"Transparency log entry created at index: {result.log_entry.log_index}"
+            )
+
+            sig_output: TextIO
+            if outputs["sig"] is not None:
+                sig_output = outputs["sig"].open("w")
+            else:
+                sig_output = sys.stdout
 
-        print(result.b64_signature, file=sig_output)
-        if outputs["sig"] is not None:
-            print(f"Signature written to {outputs['sig']}")
+            print(result.b64_signature, file=sig_output)
+            if outputs["sig"] is not None:
+                print(f"Signature written to {outputs['sig']}")
 
-        if outputs["cert"] is not None:
-            with outputs["cert"].open(mode="w") as io:
-                print(result.cert_pem, file=io)
-            print(f"Certificate written to {outputs['cert']}")
+            if outputs["cert"] is not None:
+                with outputs["cert"].open(mode="w") as io:
+                    print(result.cert_pem, file=io)
+                print(f"Certificate written to {outputs['cert']}")
 
-        if outputs["bundle"] is not None:
-            with outputs["bundle"].open(mode="w") as io:
-                print(result._to_bundle().to_json(), file=io)
-            print(f"Sigstore bundle written to {outputs['bundle']}")
+            if outputs["bundle"] is not None:
+                with outputs["bundle"].open(mode="w") as io:
+                    print(result._to_bundle().to_json(), file=io)
+                print(f"Sigstore bundle written to {outputs['bundle']}")
 
 
 def _collect_verification_state(

sigstore/_internal/fulcio/__init__.py

@@ -19,12 +19,14 @@
 
 from .client import (
     DetachedFulcioSCT,
+    ExpiredCertificate,
     FulcioCertificateSigningResponse,
     FulcioClient,
 )
 
 __all__ = [
     "DetachedFulcioSCT",
+    "ExpiredCertificate",
     "FulcioCertificateSigningResponse",
     "FulcioClient",
 ]

sigstore/_internal/fulcio/client.py

@@ -164,6 +164,10 @@ def signature(self) -> bytes:
 SignedCertificateTimestamp.register(DetachedFulcioSCT)
 
 
+class ExpiredCertificate(Exception):
+    """An error raised when the Certificate is expired."""
+
+
 @dataclass(frozen=True)
 class FulcioCertificateSigningResponse:
     """Certificate response"""

sigstore/oidc.py

@@ -24,6 +24,7 @@
 import time
 import urllib.parse
 import webbrowser
+from datetime import datetime, timezone
 from typing import NoReturn, Optional, cast
 
 import id
@@ -58,6 +59,20 @@ class _OpenIDConfiguration(BaseModel):
     token_endpoint: StrictStr
 
 
+# See: https://github.com/sigstore/fulcio/blob/b2186c0/pkg/config/config.go#L182-L201
+_KNOWN_OIDC_ISSUERS = {
+    "https://accounts.google.com": "email",
+    "https://oauth2.sigstore.dev/auth": "email",
+    "https://oauth2.sigstage.dev/auth": "email",
+    "https://token.actions.githubusercontent.com": "sub",
+}
+DEFAULT_AUDIENCE = "sigstore"
+
+
+class ExpiredIdentity(Exception):
+    """An error raised when an identity token is expired."""
+
+
 class IdentityToken:
     """
     An OIDC "identity", corresponding to an underlying OIDC token with
@@ -77,22 +92,28 @@ def __init__(self, raw_token: str) -> None:
         # certificate binding and issuance.
         try:
             self._unverified_claims = jwt.decode(
-                self._raw_token, options={"verify_signature": False}
+                raw_token,
+                options={
+                    "verify_signature": False,
+                    "verify_aud": True,
+                    "verify_iat": True,
+                    "verify_exp": True,
+                    "require": ["aud", "iat", "exp", "iss"],
+                },
+                audience=DEFAULT_AUDIENCE,
             )
-        except jwt.InvalidTokenError as exc:
-            raise IdentityError("invalid identity token") from exc
+        except Exception as exc:
+            raise IdentityError(
+                "Identity token is malformed or missing claims"
+            ) from exc
 
-        self._issuer: str = self._unverified_claims.get("iss")
-        if self._issuer is None:
-            raise IdentityError("Identity token missing the required `iss` claim")
+        self._iss: str = self._unverified_claims["iss"]
+        self._nbf: int | None = self._unverified_claims.get("nbf")
+        self._exp: int = self._unverified_claims["exp"]
 
-        aud = self._unverified_claims.get("aud")
-        if aud is None:
-            raise IdentityError("Identity token missing the required `aud` claim")
-        if aud != _DEFAULT_AUDIENCE:
-            raise IdentityError(
-                f"Audience should be {_DEFAULT_AUDIENCE!r}, not {aud!r}"
-            )
+        # Fail early if this token isn't within its validity period.
+        if not self.in_validity_period():
+            raise IdentityError("Identity token is not within its validity period")
 
         # When verifying the private key possession proof, Fulcio uses
         # different claims depending on the token's issuer.
@@ -102,15 +123,17 @@ def __init__(self, raw_token: str) -> None:
         if identity_claim is not None:
             if identity_claim not in self._unverified_claims:
                 raise IdentityError(
-                    f"Identity token missing the required {identity_claim!r} claim"
+                    f"Identity token is missing the required {identity_claim!r} claim"
                 )
 
             self._identity = str(self._unverified_claims.get(identity_claim))
         else:
             try:
                 self._identity = str(self._unverified_claims["sub"])
             except KeyError:
-                raise IdentityError("Identity token missing the required 'sub' claim")
+                raise IdentityError(
+                    "Identity token is missing the required 'sub' claim"
+                )
 
         # This identity token might have been retrieved directly from
         # an identity provider, or it might be a "federated" identity token
@@ -139,6 +162,22 @@ def __init__(self, raw_token: str) -> None:
 
                 self._federated_issuer = federated_issuer
 
+    def in_validity_period(self) -> bool:
+        """
+        Returns whether or not this `Identity` is currently within its self-stated validity period.
+
+        NOTE: As noted in `Identity.__init__`, this is not a verifying wrapper;
+        the check here only asserts whether the *unverified* identity's claims
+        are within their validity period.
+        """
+
+        now = datetime.now(timezone.utc).timestamp()
+
+        if self._nbf is not None:
+            return self._nbf <= now < self._exp
+        else:
+            return now < self._exp
+
     @property
     def identity(self) -> str:
         """
@@ -156,7 +195,7 @@ def issuer(self) -> str:
         """
         Returns a URL identifying this `IdentityToken`'s issuer.
         """
-        return self._issuer
+        return self._iss
 
     @property
     def expected_certificate_subject(self) -> str:
@@ -178,7 +217,7 @@ def expected_certificate_subject(self) -> str:
         if self._federated_issuer is not None:
             return self._federated_issuer
 
-        return self._issuer
+        return self.issuer
 
     def __str__(self) -> str:
         """