sigstore.sign

  1# Copyright 2022 The Sigstore Authors
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#      http://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""
 16API for signing artifacts.
 17
 18Example:
 19
 20```python
 21from pathlib import Path
 22
 23from sigstore.sign import SigningContext
 24from sigstore.oidc import Issuer
 25
 26issuer = Issuer.production()
 27identity = issuer.identity_token()
 28
 29# The artifact to sign
 30artifact = Path("foo.txt").read_bytes()
 31
 32signing_ctx = SigningContext.production()
 33with signing_ctx.signer(identity, cache=True) as signer:
 34    result = signer.sign_artifact(artifact)
 35    print(result)
 36```
 37"""
 38
 39from __future__ import annotations
 40
 41import base64
 42import logging
 43from contextlib import contextmanager
 44from datetime import datetime, timezone
 45from typing import Iterator, Optional
 46
 47import cryptography.x509 as x509
 48import rekor_types
 49from cryptography.hazmat.primitives import hashes, serialization
 50from cryptography.hazmat.primitives.asymmetric import ec
 51from cryptography.x509.oid import NameOID
 52from sigstore_protobuf_specs.dev.sigstore.common.v1 import (
 53    HashOutput,
 54    MessageSignature,
 55)
 56
 57from sigstore import dsse
 58from sigstore import hashes as sigstore_hashes
 59from sigstore._internal.fulcio import (
 60    ExpiredCertificate,
 61    FulcioClient,
 62)
 63from sigstore._internal.rekor.client import RekorClient
 64from sigstore._internal.sct import verify_sct
 65from sigstore._internal.trust import ClientTrustConfig, KeyringPurpose, TrustedRoot
 66from sigstore._utils import sha256_digest
 67from sigstore.models import Bundle
 68from sigstore.oidc import ExpiredIdentity, IdentityToken
 69
 70_logger = logging.getLogger(__name__)
 71
 72
 73class Signer:
 74    """
 75    The primary API for signing operations.
 76    """
 77
 78    def __init__(
 79        self,
 80        identity_token: IdentityToken,
 81        signing_ctx: SigningContext,
 82        cache: bool = True,
 83    ) -> None:
 84        """
 85        Create a new `Signer`.
 86
 87        `identity_token` is the identity token used to request a signing certificate
 88        from Fulcio.
 89
 90        `signing_ctx` is a `SigningContext` that keeps information about the signing
 91        configuration.
 92
 93        `cache` determines whether the signing certificate and ephemeral private key
 94        should be reused (until the certificate expires) to sign different artifacts.
 95        Default is `True`.
 96        """
 97        self._identity_token = identity_token
 98        self._signing_ctx: SigningContext = signing_ctx
 99        self.__cached_private_key: Optional[ec.EllipticCurvePrivateKey] = None
100        self.__cached_signing_certificate: Optional[x509.Certificate] = None
101        if cache:
102            _logger.debug("Generating ephemeral keys...")
103            self.__cached_private_key = ec.generate_private_key(ec.SECP256R1())
104            _logger.debug("Requesting ephemeral certificate...")
105            self.__cached_signing_certificate = self._signing_cert()
106
107    @property
108    def _private_key(self) -> ec.EllipticCurvePrivateKey:
109        """Get or generate a signing key."""
110        if self.__cached_private_key is None:
111            _logger.debug("no cached key; generating ephemeral key")
112            return ec.generate_private_key(ec.SECP256R1())
113        return self.__cached_private_key
114
115    def _signing_cert(
116        self,
117    ) -> x509.Certificate:
118        """
119        Get or request a signing certificate from Fulcio.
120
121        Internally, this performs a CSR against Fulcio and verifies that
122        the returned certificate is present in Fulcio's CT log.
123        """
124
125        # Our CSR cannot possibly succeed if our underlying identity token
126        # is expired.
127        if not self._identity_token.in_validity_period():
128            raise ExpiredIdentity
129
130        # If it exists, verify if the current certificate is expired
131        if self.__cached_signing_certificate:
132            not_valid_after = self.__cached_signing_certificate.not_valid_after_utc
133            if datetime.now(timezone.utc) > not_valid_after:
134                raise ExpiredCertificate
135            return self.__cached_signing_certificate
136
137        else:
138            _logger.debug("Retrieving signed certificate...")
139
140            # Build an X.509 Certificiate Signing Request
141            builder = (
142                x509.CertificateSigningRequestBuilder()
143                .subject_name(
144                    x509.Name(
145                        [
146                            x509.NameAttribute(
147                                NameOID.EMAIL_ADDRESS, self._identity_token._identity
148                            ),
149                        ]
150                    )
151                )
152                .add_extension(
153                    x509.BasicConstraints(ca=False, path_length=None),
154                    critical=True,
155                )
156            )
157            certificate_request = builder.sign(self._private_key, hashes.SHA256())
158
159            certificate_response = self._signing_ctx._fulcio.signing_cert.post(
160                certificate_request, self._identity_token
161            )
162
163            # Verify the SCT
164            sct = certificate_response.sct
165            cert = certificate_response.cert
166            chain = certificate_response.chain
167
168            verify_sct(
169                sct,
170                cert,
171                chain,
172                self._signing_ctx._trusted_root.ct_keyring(KeyringPurpose.SIGN),
173            )
174
175            _logger.debug("Successfully verified SCT...")
176
177            return cert
178
179    def _finalize_sign(
180        self,
181        cert: x509.Certificate,
182        content: MessageSignature | dsse.Envelope,
183        proposed_entry: rekor_types.Hashedrekord | rekor_types.Dsse,
184    ) -> Bundle:
185        """
186        Perform the common "finalizing" steps in a Sigstore signing flow.
187        """
188        # Submit the proposed entry to the transparency log
189        entry = self._signing_ctx._rekor.log.entries.post(proposed_entry)
190
191        _logger.debug(f"Transparency log entry created with index: {entry.log_index}")
192
193        return Bundle._from_parts(cert, content, entry)
194
195    def sign_dsse(
196        self,
197        input_: dsse.Statement,
198    ) -> Bundle:
199        """
200        Sign the given in-toto statement as a DSSE envelope, and return a
201        `Bundle` containing the signed result.
202
203        This API is **only** for in-toto statements; to sign arbitrary artifacts,
204        use `sign_artifact` instead.
205        """
206        cert = self._signing_cert()
207
208        # Prepare inputs
209        b64_cert = base64.b64encode(
210            cert.public_bytes(encoding=serialization.Encoding.PEM)
211        )
212
213        # Sign the statement, producing a DSSE envelope
214        content = dsse._sign(self._private_key, input_)
215
216        # Create the proposed DSSE log entry
217        proposed_entry = rekor_types.Dsse(
218            spec=rekor_types.dsse.DsseSchema(
219                # NOTE: mypy can't see that this kwarg is correct due to two interacting
220                # behaviors/bugs (one pydantic, one datamodel-codegen):
221                # See: <https://github.com/pydantic/pydantic/discussions/7418#discussioncomment-9024927>
222                # See: <https://github.com/koxudaxi/datamodel-code-generator/issues/1903>
223                proposed_content=rekor_types.dsse.ProposedContent(  # type: ignore[call-arg]
224                    envelope=content.to_json(),
225                    verifiers=[b64_cert.decode()],
226                ),
227            ),
228        )
229
230        return self._finalize_sign(cert, content, proposed_entry)
231
232    def sign_artifact(
233        self,
234        input_: bytes | sigstore_hashes.Hashed,
235    ) -> Bundle:
236        """
237        Sign an artifact, and return a `Bundle` corresponding to the signed result.
238
239        The input can be one of two forms:
240
241        1. A `bytes` buffer;
242        2. A `Hashed` object, containing a pre-hashed input (e.g., for inputs
243           that are too large to buffer into memory).
244
245        Regardless of the input format, the signing operation will produce a
246        `hashedrekord` entry within the bundle. No other entry types
247        are supported by this API.
248        """
249
250        cert = self._signing_cert()
251
252        # Prepare inputs
253        b64_cert = base64.b64encode(
254            cert.public_bytes(encoding=serialization.Encoding.PEM)
255        )
256
257        # Sign artifact
258        hashed_input = sha256_digest(input_)
259
260        artifact_signature = self._private_key.sign(
261            hashed_input.digest, ec.ECDSA(hashed_input._as_prehashed())
262        )
263
264        content = MessageSignature(
265            message_digest=HashOutput(
266                algorithm=hashed_input.algorithm,
267                digest=hashed_input.digest,
268            ),
269            signature=artifact_signature,
270        )
271
272        # Create the proposed hashedrekord entry
273        proposed_entry = rekor_types.Hashedrekord(
274            spec=rekor_types.hashedrekord.HashedrekordV001Schema(
275                signature=rekor_types.hashedrekord.Signature(
276                    content=base64.b64encode(artifact_signature).decode(),
277                    public_key=rekor_types.hashedrekord.PublicKey(
278                        content=b64_cert.decode()
279                    ),
280                ),
281                data=rekor_types.hashedrekord.Data(
282                    hash=rekor_types.hashedrekord.Hash(
283                        algorithm=hashed_input._as_hashedrekord_algorithm(),
284                        value=hashed_input.digest.hex(),
285                    )
286                ),
287            ),
288        )
289
290        return self._finalize_sign(cert, content, proposed_entry)
291
292
293class SigningContext:
294    """
295    Keep a context between signing operations.
296    """
297
298    def __init__(
299        self, *, fulcio: FulcioClient, rekor: RekorClient, trusted_root: TrustedRoot
300    ):
301        """
302        Create a new `SigningContext`.
303
304        `fulcio` is a `FulcioClient` capable of connecting to a Fulcio instance
305        and returning signing certificates.
306
307        `rekor` is a `RekorClient` capable of connecting to a Rekor instance
308        and creating transparency log entries.
309        """
310        self._fulcio = fulcio
311        self._rekor = rekor
312        self._trusted_root = trusted_root
313
314    @classmethod
315    def production(cls) -> SigningContext:
316        """
317        Return a `SigningContext` instance configured against Sigstore's production-level services.
318        """
319        return cls(
320            fulcio=FulcioClient.production(),
321            rekor=RekorClient.production(),
322            trusted_root=TrustedRoot.production(),
323        )
324
325    @classmethod
326    def staging(cls) -> SigningContext:
327        """
328        Return a `SignerContext` instance configured against Sigstore's staging-level services.
329        """
330        return cls(
331            fulcio=FulcioClient.staging(),
332            rekor=RekorClient.staging(),
333            trusted_root=TrustedRoot.staging(),
334        )
335
336    @classmethod
337    def _from_trust_config(cls, trust_config: ClientTrustConfig) -> SigningContext:
338        """
339        Create a `SigningContext` from the given `ClientTrustConfig`.
340
341        @api private
342        """
343        return cls(
344            fulcio=FulcioClient(trust_config._inner.signing_config.ca_url),
345            rekor=RekorClient(trust_config._inner.signing_config.tlog_urls[0]),
346            trusted_root=trust_config.trusted_root,
347        )
348
349    @contextmanager
350    def signer(
351        self, identity_token: IdentityToken, *, cache: bool = True
352    ) -> Iterator[Signer]:
353        """
354        A context manager for signing operations.
355
356        `identity_token` is the identity token passed to the `Signer` instance
357        and used to request a signing certificate from Fulcio.
358
359        `cache` determines whether the signing certificate and ephemeral private key
360        generated by the `Signer` instance should be reused (until the certificate expires)
361        to sign different artifacts.
362        Default is `True`.
363        """
364        yield Signer(identity_token, self, cache)