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