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