1# Copyright 2025 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"""Signers and verifiers using certificates."""
16
17import base64
18from collections.abc import Iterable
19import logging
20import pathlib
21
22import certifi
23from cryptography import exceptions
24from cryptography import x509
25from cryptography.hazmat.primitives import hashes
26from cryptography.hazmat.primitives import serialization
27from cryptography.hazmat.primitives.asymmetric import ec
28from cryptography.x509 import oid
29from OpenSSL import crypto
30from sigstore_models.bundle import v1 as bundle_pb
31from sigstore_models.common import v1 as common_pb
32from typing_extensions import override
33
34from model_signing._signing import sign_ec_key as ec_key
35from model_signing._signing import sign_sigstore_pb as sigstore_pb
36
37
38logger = logging.getLogger(__name__)
39
40
41class Signer(ec_key.Signer):
42 """Signer using certificates."""
43
44 def __init__(
45 self,
46 private_key_path: pathlib.Path,
47 signing_certificate_path: pathlib.Path,
48 certificate_chain_paths: Iterable[pathlib.Path],
49 ):
50 """Initializes the signer with the key, certificate and trust chain.
51
52 Args:
53 private_key_path: The path to the PEM encoded private key.
54 signing_certificate_path: The path to the signing certificate.
55 certificate_chain_paths: Paths to other certificates used to
56 establish chain of trust.
57
58 Raises:
59 ValueError: Signing certificate's public key does not match the
60 private key's public pair.
61 """
62 super().__init__(private_key_path)
63 self._signing_certificate = x509.load_pem_x509_certificate(
64 signing_certificate_path.read_bytes()
65 )
66
67 public_key_from_key = self._private_key.public_key()
68 public_key_from_certificate = self._signing_certificate.public_key()
69 if public_key_from_key != public_key_from_certificate:
70 raise ValueError(
71 "The public key from the certificate does not match "
72 "the public key paired with the private key"
73 )
74
75 chain_bytes = b"".join(
76 [path.read_bytes() for path in certificate_chain_paths]
77 )
78 self._trust_chain = (
79 x509.load_pem_x509_certificates(chain_bytes) if chain_bytes else []
80 )
81
82 @override
83 def _get_verification_material(self) -> bundle_pb.VerificationMaterial:
84 def _to_protobuf_certificate(certificate):
85 return common_pb.X509Certificate(
86 raw_bytes=base64.b64encode(
87 certificate.public_bytes(
88 encoding=serialization.Encoding.DER
89 )
90 )
91 )
92
93 chain = [_to_protobuf_certificate(self._signing_certificate)]
94 chain.extend(
95 [
96 _to_protobuf_certificate(certificate)
97 for certificate in self._trust_chain
98 ]
99 )
100
101 return bundle_pb.VerificationMaterial(
102 x509_certificate_chain=common_pb.X509CertificateChain(
103 certificates=chain
104 ),
105 tlog_entries=[],
106 )
107
108
109def _log_certificate_fingerprint(
110 where: str, certificate: x509.Certificate, hash_algorithm: hashes.Hash
111) -> None:
112 """Log the fingerprint of a certificate, for debugging.
113
114 Args:
115 where: Location of where this gets called from, useful for debugging.
116 certificate: Certificate to compute fingerprint of and log.
117 hash_algorithm: The algorithm used to compute the fingerprint.
118 """
119 fp = certificate.fingerprint(hash_algorithm)
120 logger.info(
121 f"[{where:^8}] {hash_algorithm.name} "
122 f"Fingerprint: {':'.join(f'{b:02X}' for b in fp)}"
123 )
124
125
126class Verifier(sigstore_pb.Verifier):
127 """Verifier for signatures generated via signing with certificates."""
128
129 def __init__(
130 self,
131 certificate_chain_paths: Iterable[pathlib.Path] = frozenset(),
132 log_fingerprints: bool = False,
133 expected_san_uris: Iterable[str] = frozenset(),
134 ):
135 """Initializes the verifier with the list of certificates to use.
136
137 Args:
138 certificate_chain_paths: Paths to certificates used to verify
139 signature and establish chain of trust. By default this is empty,
140 in which case we would use the root certificates from the
141 operating system, as per `certifi.where()`.
142 log_fingerprints: Log the fingerprints of certificates
143 expected_san_uris: If non-empty, verification additionally requires
144 that every listed URI appear in the leaf certificate's
145 SubjectAltName URI entries. This binds the signature to a signer
146 identity (e.g. a SPIFFE ID, which per RFC-compliant SVIDs is
147 always carried in the URI SAN) rather than trusting any
148 certificate issued under the CA.
149 """
150 self._log_fingerprints = log_fingerprints
151 self._expected_san_uris = frozenset(expected_san_uris)
152
153 if not certificate_chain_paths:
154 certificate_chain_paths = [pathlib.Path(certifi.where())]
155
156 certificates = x509.load_pem_x509_certificates(
157 b"".join([path.read_bytes() for path in certificate_chain_paths])
158 )
159
160 self._store = crypto.X509Store()
161 for certificate in certificates:
162 if self._log_fingerprints:
163 _log_certificate_fingerprint(
164 "init", certificate, hashes.SHA256()
165 )
166 self._store.add_cert(crypto.X509.from_cryptography(certificate))
167
168 @override
169 def _verify_bundle(self, bundle: bundle_pb.Bundle) -> tuple[str, bytes]:
170 public_key = self._verify_certificates(bundle.verification_material)
171 envelope = bundle.dsse_envelope
172 try:
173 public_key.verify(
174 envelope.signatures[0].sig,
175 sigstore_pb.pae(envelope.payload),
176 ec.ECDSA(ec_key.get_ec_key_hash(public_key)),
177 )
178 except exceptions.InvalidSignature:
179 # Compatibility layer with pre 1.0 release
180 # Here, we patch over a bug in `pae` which mixed unicode `str` and
181 # `bytes`. As a result, additional escape characters were added to
182 # the material that got signed over.
183 public_key.verify(
184 envelope.signatures[0].sig,
185 sigstore_pb.pae_compat(envelope.payload),
186 # Note another bug here: the v0.2 signatures were generated with
187 # hardcoded SHA256 hash, instead of the one that matches the
188 # key type. To verify those signatures, we have to hardcode this
189 # here too (instead of `ec_key.get_ec_key_hash(public_key)`).
190 # For the hardcode path see:
191 # https://github.com/sigstore/model-transparency/blob/9737f0e28349bf43897857ada7beaa22ec18e9a6/src/model_signing/signature/key.py#L103
192 ec.ECDSA(hashes.SHA256()),
193 )
194
195 return envelope.payload_type, envelope.payload
196
197 def _verify_certificates(
198 self,
199 verification_material: bundle_pb.VerificationMaterial,
200 log_fingerprints: bool = False,
201 ) -> ec.EllipticCurvePublicKey:
202 """Verifies the certificate chain and returns the public key.
203
204 The public key is extracted from the signing certificate from the chain
205 of trust, after the chain is validated. It must match the public key
206 from the key used during signing.
207 """
208
209 def _to_openssl_certificate(certificate_bytes, log_fingerprints):
210 cert = x509.load_der_x509_certificate(certificate_bytes)
211 if log_fingerprints:
212 _log_certificate_fingerprint("verify", cert, hashes.SHA256())
213 return crypto.X509.from_cryptography(cert)
214
215 signing_chain = verification_material.x509_certificate_chain
216 signing_certificate = x509.load_der_x509_certificate(
217 signing_chain.certificates[0].raw_bytes
218 )
219
220 max_signing_time = signing_certificate.not_valid_before_utc
221 self._store.set_time(max_signing_time)
222
223 trust_chain_ssl = [
224 _to_openssl_certificate(
225 certificate.raw_bytes, self._log_fingerprints
226 )
227 for certificate in signing_chain.certificates[1:]
228 ]
229 signing_certificate_ssl = _to_openssl_certificate(
230 signing_chain.certificates[0].raw_bytes, self._log_fingerprints
231 )
232
233 store_context = crypto.X509StoreContext(
234 self._store, signing_certificate_ssl, trust_chain_ssl
235 )
236 store_context.verify_certificate()
237
238 extensions = signing_certificate.extensions
239 can_use_for_signing = False
240 try:
241 usage = extensions.get_extension_for_class(x509.KeyUsage)
242 if usage.value.digital_signature:
243 can_use_for_signing = True
244 except x509.ExtensionNotFound:
245 logger.warning("Certificate does not specify 'KeyUsage'.")
246
247 if not can_use_for_signing:
248 try:
249 usage = extensions.get_extension_for_class(
250 x509.ExtendedKeyUsage
251 )
252 if oid.ExtendedKeyUsageOID.CODE_SIGNING in usage.value:
253 can_use_for_signing = True
254 except x509.ExtensionNotFound:
255 logger.warning(
256 "Certificate does not specify 'ExtendedKeyUsage'."
257 )
258
259 # An extended key usage, when present, restricts the certificate to the
260 # listed purposes (RFC 5280 4.2.1.12). A certificate not marked for code
261 # signing must be rejected even when the digitalSignature key usage bit
262 # is set, otherwise a TLS (serverAuth) certificate chaining to a trusted
263 # root would be accepted for model signing.
264 try:
265 eku = extensions.get_extension_for_class(
266 x509.ExtendedKeyUsage
267 ).value
268 if (
269 oid.ExtendedKeyUsageOID.CODE_SIGNING not in eku
270 and oid.ExtendedKeyUsageOID.ANY_EXTENDED_KEY_USAGE not in eku
271 ):
272 can_use_for_signing = False
273 except x509.ExtensionNotFound:
274 pass
275
276 if not can_use_for_signing:
277 raise ValueError("Signing certificate cannot be used for signing")
278
279 self._verify_san_identity(signing_certificate)
280
281 return signing_certificate.public_key()
282
283 def _verify_san_identity(
284 self, signing_certificate: x509.Certificate
285 ) -> None:
286 """Assert the leaf's SubjectAltName carries every expected URI.
287
288 Chain-of-trust proves the CA vouched for *some* leaf; it does not tell
289 us *which* leaf. If the caller declared expected SAN URIs (e.g. a
290 SPIFFE ID), the signing certificate embedded in the bundle must carry
291 them, otherwise a different (but still CA-issued) key could produce
292 accepted signatures.
293 """
294 if not self._expected_san_uris:
295 return
296
297 try:
298 san = signing_certificate.extensions.get_extension_for_class(
299 x509.SubjectAlternativeName
300 ).value
301 except x509.ExtensionNotFound as err:
302 raise ValueError(
303 "Signing certificate has no SubjectAlternativeName; cannot "
304 "verify expected signer identity."
305 ) from err
306
307 actual_uris = frozenset(
308 san.get_values_for_type(x509.UniformResourceIdentifier)
309 )
310 missing_uris = self._expected_san_uris - actual_uris
311 if missing_uris:
312 raise ValueError(
313 "Signing certificate SubjectAltName is missing expected "
314 f"URI(s): {sorted(missing_uris)} "
315 f"(present: {sorted(actual_uris)})"
316 )