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

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

111 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 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 )