sigstore._internal.trust

  1# Copyright 2023 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"""
 16Client trust configuration and trust root management for sigstore-python.
 17"""
 18
 19from __future__ import annotations
 20
 21from dataclasses import dataclass
 22from datetime import datetime, timezone
 23from enum import Enum
 24from pathlib import Path
 25from typing import ClassVar, Iterable, List, NewType
 26
 27import cryptography.hazmat.primitives.asymmetric.padding as padding
 28from cryptography.exceptions import InvalidSignature
 29from cryptography.hazmat.primitives import hashes
 30from cryptography.hazmat.primitives.asymmetric import ec, rsa
 31from cryptography.x509 import (
 32    Certificate,
 33    load_der_x509_certificate,
 34)
 35from sigstore_protobuf_specs.dev.sigstore.common.v1 import PublicKey as _PublicKey
 36from sigstore_protobuf_specs.dev.sigstore.common.v1 import (
 37    PublicKeyDetails as _PublicKeyDetails,
 38)
 39from sigstore_protobuf_specs.dev.sigstore.common.v1 import TimeRange
 40from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
 41    CertificateAuthority,
 42    TransparencyLogInstance,
 43)
 44from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
 45    ClientTrustConfig as _ClientTrustConfig,
 46)
 47from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
 48    TrustedRoot as _TrustedRoot,
 49)
 50
 51from sigstore._internal.tuf import DEFAULT_TUF_URL, STAGING_TUF_URL, TrustUpdater
 52from sigstore._utils import (
 53    KeyID,
 54    PublicKey,
 55    key_id,
 56    load_der_public_key,
 57)
 58from sigstore.errors import Error, MetadataError, VerificationError
 59
 60
 61def _is_timerange_valid(period: TimeRange | None, *, allow_expired: bool) -> bool:
 62    """
 63    Given a `period`, checks that the the current time is not before `start`. If
 64    `allow_expired` is `False`, also checks that the current time is not after
 65    `end`.
 66    """
 67    now = datetime.now(timezone.utc)
 68
 69    # If there was no validity period specified, the key is always valid.
 70    if not period:
 71        return True
 72
 73    # Active: if the current time is before the starting period, we are not yet
 74    # valid.
 75    if now < period.start:
 76        return False
 77
 78    # If we want Expired keys, the key is valid at this point. Otherwise, check
 79    # that we are within range.
 80    return allow_expired or (period.end is None or now <= period.end)
 81
 82
 83@dataclass(init=False)
 84class Key:
 85    """
 86    Represents a key in a `Keyring`.
 87    """
 88
 89    hash_algorithm: hashes.HashAlgorithm
 90    key: PublicKey
 91    key_id: KeyID
 92
 93    _RSA_SHA_256_DETAILS: ClassVar[set[_PublicKeyDetails]] = {
 94        _PublicKeyDetails.PKCS1_RSA_PKCS1V5,
 95        _PublicKeyDetails.PKIX_RSA_PKCS1V15_2048_SHA256,
 96        _PublicKeyDetails.PKIX_RSA_PKCS1V15_3072_SHA256,
 97        _PublicKeyDetails.PKIX_RSA_PKCS1V15_4096_SHA256,
 98    }
 99
100    _EC_DETAILS_TO_HASH: ClassVar[dict[_PublicKeyDetails, hashes.HashAlgorithm]] = {
101        _PublicKeyDetails.PKIX_ECDSA_P256_SHA_256: hashes.SHA256(),
102        _PublicKeyDetails.PKIX_ECDSA_P384_SHA_384: hashes.SHA384(),
103        _PublicKeyDetails.PKIX_ECDSA_P521_SHA_512: hashes.SHA512(),
104    }
105
106    def __init__(self, public_key: _PublicKey) -> None:
107        """
108        Construct a key from the given Sigstore PublicKey message.
109        """
110
111        # NOTE: `raw_bytes` is marked as `optional` in the `PublicKey` message,
112        # for unclear reasons.
113        if not public_key.raw_bytes:
114            raise VerificationError("public key is empty")
115
116        hash_algorithm: hashes.HashAlgorithm
117        if public_key.key_details in self._RSA_SHA_256_DETAILS:
118            hash_algorithm = hashes.SHA256()
119            key = load_der_public_key(public_key.raw_bytes, types=(rsa.RSAPublicKey,))
120        elif public_key.key_details in self._EC_DETAILS_TO_HASH:
121            hash_algorithm = self._EC_DETAILS_TO_HASH[public_key.key_details]
122            key = load_der_public_key(
123                public_key.raw_bytes, types=(ec.EllipticCurvePublicKey,)
124            )
125        else:
126            raise VerificationError(f"unsupported key type: {public_key.key_details}")
127
128        self.hash_algorithm = hash_algorithm
129        self.key = key
130        self.key_id = key_id(key)
131
132    def verify(self, signature: bytes, data: bytes) -> None:
133        """
134        Verifies the given `data` against `signature` using the current key.
135        """
136        if isinstance(self.key, rsa.RSAPublicKey):
137            self.key.verify(
138                signature=signature,
139                data=data,
140                # TODO: Parametrize this as well, for PSS.
141                padding=padding.PKCS1v15(),
142                algorithm=self.hash_algorithm,
143            )
144        elif isinstance(self.key, ec.EllipticCurvePublicKey):
145            self.key.verify(
146                signature=signature,
147                data=data,
148                signature_algorithm=ec.ECDSA(self.hash_algorithm),
149            )
150        else:
151            # Unreachable without API misuse.
152            raise VerificationError(f"keyring: unsupported key: {self.key}")
153
154
155class Keyring:
156    """
157    Represents a set of keys, each of which is a potentially valid verifier.
158    """
159
160    def __init__(self, public_keys: List[_PublicKey] = []):
161        """
162        Create a new `Keyring`, with `keys` as the initial set of verifying keys.
163        """
164        self._keyring: dict[KeyID, Key] = {}
165
166        for public_key in public_keys:
167            key = Key(public_key)
168            self._keyring[key.key_id] = key
169
170    def verify(self, *, key_id: KeyID, signature: bytes, data: bytes) -> None:
171        """
172        Verify that `signature` is a valid signature for `data`, using the
173        key identified by `key_id`.
174
175        `key_id` is an unauthenticated hint; if no key matches the given key ID,
176        all keys in the keyring are tried.
177
178        Raises if the signature is invalid, i.e. is not valid for any of the
179        keys in the keyring.
180        """
181
182        key = self._keyring.get(key_id)
183        if key is not None:
184            candidates = [key]
185        else:
186            candidates = list(self._keyring.values())
187
188        # Try to verify each candidate key. In the happy case, this will
189        # be exactly one candidate.
190        valid = False
191        for candidate in candidates:
192            try:
193                candidate.verify(signature, data)
194                valid = True
195                break
196            except InvalidSignature:
197                pass
198
199        if not valid:
200            raise VerificationError("keyring: invalid signature")
201
202
203RekorKeyring = NewType("RekorKeyring", Keyring)
204CTKeyring = NewType("CTKeyring", Keyring)
205
206
207class KeyringPurpose(str, Enum):
208    """
209    Keyring purpose typing
210    """
211
212    SIGN = "sign"
213    VERIFY = "verify"
214
215    def __str__(self) -> str:
216        """Returns the purpose string value."""
217        return self.value
218
219
220class TrustedRoot:
221    """
222    The cryptographic root(s) of trust for a Sigstore instance.
223    """
224
225    class TrustedRootType(str, Enum):
226        """
227        Known Sigstore trusted root media types.
228        """
229
230        TRUSTED_ROOT_0_1 = "application/vnd.dev.sigstore.trustedroot+json;version=0.1"
231
232        def __str__(self) -> str:
233            """Returns the variant's string value."""
234            return self.value
235
236    def __init__(self, inner: _TrustedRoot):
237        """
238        Construct a new `TrustedRoot`.
239
240        @api private
241        """
242        self._inner = inner
243        self._verify()
244
245    def _verify(self) -> None:
246        """
247        Performs various feats of heroism to ensure that the trusted root
248        is well-formed.
249        """
250
251        # The trusted root must have a recognized media type.
252        try:
253            TrustedRoot.TrustedRootType(self._inner.media_type)
254        except ValueError:
255            raise Error(f"unsupported trusted root format: {self._inner.media_type}")
256
257    @classmethod
258    def from_file(
259        cls,
260        path: str,
261    ) -> TrustedRoot:
262        """Create a new trust root from file"""
263        inner = _TrustedRoot().from_json(Path(path).read_bytes())
264        return cls(inner)
265
266    @classmethod
267    def from_tuf(
268        cls,
269        url: str,
270        offline: bool = False,
271    ) -> TrustedRoot:
272        """Create a new trust root from a TUF repository.
273
274        If `offline`, will use trust root in local TUF cache. Otherwise will
275        update the trust root from remote TUF repository.
276        """
277        path = TrustUpdater(url, offline).get_trusted_root_path()
278        return cls.from_file(path)
279
280    @classmethod
281    def production(
282        cls,
283        offline: bool = False,
284    ) -> TrustedRoot:
285        """Create new trust root from Sigstore production TUF repository.
286
287        If `offline`, will use trust root in local TUF cache. Otherwise will
288        update the trust root from remote TUF repository.
289        """
290        return cls.from_tuf(DEFAULT_TUF_URL, offline)
291
292    @classmethod
293    def staging(
294        cls,
295        offline: bool = False,
296    ) -> TrustedRoot:
297        """Create new trust root from Sigstore staging TUF repository.
298
299        If `offline`, will use trust root in local TUF cache. Otherwise will
300        update the trust root from remote TUF repository.
301        """
302        return cls.from_tuf(STAGING_TUF_URL, offline)
303
304    def _get_tlog_keys(
305        self, tlogs: list[TransparencyLogInstance], purpose: KeyringPurpose
306    ) -> Iterable[_PublicKey]:
307        """
308        Yields an iterator of public keys for transparency log instances that
309        are suitable for `purpose`.
310        """
311        allow_expired = purpose is KeyringPurpose.VERIFY
312        for tlog in tlogs:
313            if not _is_timerange_valid(
314                tlog.public_key.valid_for, allow_expired=allow_expired
315            ):
316                continue
317
318            yield tlog.public_key
319
320    @staticmethod
321    def _get_ca_keys(
322        cas: list[CertificateAuthority], *, allow_expired: bool
323    ) -> Iterable[bytes]:
324        """Return public key contents given certificate authorities."""
325
326        for ca in cas:
327            if not _is_timerange_valid(ca.valid_for, allow_expired=allow_expired):
328                continue
329            for cert in ca.cert_chain.certificates:
330                yield cert.raw_bytes
331
332    def rekor_keyring(self, purpose: KeyringPurpose) -> RekorKeyring:
333        """Return keyring with keys for Rekor."""
334
335        keys: list[_PublicKey] = list(self._get_tlog_keys(self._inner.tlogs, purpose))
336        if len(keys) != 1:
337            raise MetadataError("Did not find one Rekor key in trusted root")
338        return RekorKeyring(Keyring(keys))
339
340    def ct_keyring(self, purpose: KeyringPurpose) -> CTKeyring:
341        """Return keyring with key for CTFE."""
342        ctfes: list[_PublicKey] = list(self._get_tlog_keys(self._inner.ctlogs, purpose))
343        if not ctfes:
344            raise MetadataError("CTFE keys not found in trusted root")
345        return CTKeyring(Keyring(ctfes))
346
347    def get_fulcio_certs(self) -> list[Certificate]:
348        """Return the Fulcio certificates."""
349
350        certs: list[Certificate]
351
352        # Return expired certificates too: they are expired now but may have
353        # been active when the certificate was used to sign.
354        certs = [
355            load_der_x509_certificate(c)
356            for c in self._get_ca_keys(
357                self._inner.certificate_authorities, allow_expired=True
358            )
359        ]
360        if not certs:
361            raise MetadataError("Fulcio certificates not found in trusted root")
362        return certs
363
364
365class ClientTrustConfig:
366    """
367    Represents a Sigstore client's trust configuration, including a root of trust.
368    """
369
370    class ClientTrustConfigType(str, Enum):
371        """
372        Known Sigstore client trust config media types.
373        """
374
375        CONFIG_0_1 = "application/vnd.dev.sigstore.clienttrustconfig.v0.1+json"
376
377        def __str__(self) -> str:
378            """Returns the variant's string value."""
379            return self.value
380
381    @classmethod
382    def from_json(cls, raw: str) -> ClientTrustConfig:
383        """
384        Deserialize the given client trust config.
385        """
386        inner = _ClientTrustConfig().from_json(raw)
387        return cls(inner)
388
389    def __init__(self, inner: _ClientTrustConfig) -> None:
390        """
391        @api private
392        """
393        self._inner = inner
394        self._verify()
395
396    def _verify(self) -> None:
397        """
398        Performs various feats of heroism to ensure that the client trust config
399        is well-formed.
400        """
401
402        # The client trust config must have a recognized media type.
403        try:
404            ClientTrustConfig.ClientTrustConfigType(self._inner.media_type)
405        except ValueError:
406            raise Error(
407                f"unsupported client trust config format: {self._inner.media_type}"
408            )
409
410    @property
411    def trusted_root(self) -> TrustedRoot:
412        """
413        Return the interior root of trust, as a `TrustedRoot`.
414        """
415        return TrustedRoot(self._inner.trusted_root)