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

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

115 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 ): 

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 )