Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/model_signing/_signing/sign_certificate.py: 26%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

120 statements  

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 )