Coverage for pki_tools/types/certificate.py: 91%

1import base64 

2import datetime 

3import random 

4import re 

5import time 

6 

7from cryptography import x509 

8from cryptography.hazmat.primitives.asymmetric import ed448, ed25519 

9from cryptography.hazmat.primitives.asymmetric.types import ( 

10 CertificatePublicKeyTypes, 

11) 

12from loguru import logger 

13from pydantic import BaseModel, ConfigDict 

14 

15from pki_tools.exceptions import ( 

16 MissingInit, 

17 SignatureVerificationFailed, 

18) 

19from pki_tools.types.crypto_parser import ( 

20 CryptoConfig, 

21 CryptoParser, 

22 HelperFunc, 

23 InitCryptoParser, 

24) 

25from pki_tools.types.extensions import Extensions 

26from pki_tools.types.key_pair import ( 

27 CryptoKeyPair, 

28 CryptoPublicKey, 

29) 

30from pki_tools.types.name import Name 

31from pki_tools.types.signature_algorithm import ( 

32 HashAlgorithm, 

33 HashAlgorithmName, 

34 SignatureAlgorithm, 

35) 

36from pki_tools.types.utils import ( 

37 CACHE_TIME_SECONDS, 

38 CertsUri, 

39 _byte_to_hex, 

40 _download_cached, 

41 _download_server_certificate, 

42) 

43 

44PEM_CERT_REGEX = re.compile( 

45 r"\s*-+BEGIN CERTIFICATE-+[\w+/\s=]*-+END CERTIFICATE-+\s*" 

46) 

47 

48_DEFAULT_DIGEST_ALGORITHM = HashAlgorithm(name=HashAlgorithmName.SHA512) 

49 

50 

51class Validity(BaseModel): 

52 """ 

53 Describes the validity of a certificate 

54 

55 Attributes: 

56 not_before: The start date of when the certificate will be valid 

57 not_after: The date of when the certificate expires 

58 """ 

59 

60 not_before: datetime.datetime 

61 not_after: datetime.datetime 

62 

63 def __init__(self, **kwargs): 

64 super().__init__(**kwargs) 

65 

66 if self.not_before.tzinfo is None: 

67 self.not_before = self.not_before.replace( 

68 tzinfo=datetime.timezone.utc 

69 ) 

70 if self.not_after.tzinfo is None: 

71 self.not_after = self.not_after.replace( 

72 tzinfo=datetime.timezone.utc 

73 ) 

74 

75 def _string_dict(self) -> dict[str, str]: 

76 return { 

77 "Not Before": str(self.not_before), 

78 "Not After": str(self.not_after), 

79 } 

80 

81 

82class SubjectPublicKeyInfo(CryptoParser): 

83 """ 

84 Represents a certificate SubjectPublicKeyInfo. 

85 

86 Attributes: 

87 algorithm: The key algorithm in string format 

88 parameters: The dict representation of the key 

89 """ 

90 

91 algorithm: CryptoPublicKey 

92 parameters: dict[str, str] | None 

93 

94 @classmethod 

95 def from_cryptography( 

96 cls: type["SubjectPublicKeyInfo"], 

97 crypto_obj: CertificatePublicKeyTypes, 

98 ) -> "SubjectPublicKeyInfo": 

99 public_key = CryptoPublicKey.from_cryptography(crypto_obj) 

100 

101 return cls( 

102 algorithm=public_key, 

103 parameters=public_key._string_dict(), 

104 _x509_obj=crypto_obj, 

105 ) 

106 

107 def _to_cryptography(self) -> CertificatePublicKeyTypes: 

108 return self.algorithm._to_cryptography() 

109 

110 def _string_dict(self) -> dict: 

111 params = {} 

112 for k, v in self.parameters.items(): 

113 key = " ".join(ele.title() for ele in k.split("_")) 

114 if v == "None": 

115 continue 

116 

117 params[key] = v 

118 

119 return { 

120 "Public Key Algorithm": self.algorithm._string_dict(), 

121 "Parameters": params, 

122 } 

123 

124 

125class Certificate(InitCryptoParser): 

126 """ 

127 An object describing a x509 Certificate 

128 

129 Attributes: 

130 issuer: Certificate issuer 

131 subject: Certificate subject 

132 validity: Contains information about NotBefore and NotAfter 

133 

134 extensions: Certificate (v3) extensions 

135 serial_number: Serial number 

136 version: The version of the certificate 

137 signature_algorithm: Describes the algorithm used to sign the 

138 certificate 

139 subject_public_key_info: The public key information 

140 

141 --8<-- "docs/examples/certificate.md" 

142 """ 

143 

144 issuer: Name 

145 validity: Validity 

146 subject: Name 

147 

148 extensions: Extensions | None = None 

149 serial_number: int | None = None 

150 version: int | None = None 

151 signature_algorithm: SignatureAlgorithm | None = None 

152 subject_public_key_info: SubjectPublicKeyInfo | None = None 

153 

154 signature_value: str | None = None 

155 

156 model_config = ConfigDict(arbitrary_types_allowed=True) 

157 

158 _key_pair: CryptoKeyPair | None 

159 

160 @classmethod 

161 def from_cryptography( 

162 cls: type["Certificate"], 

163 cert: x509.Certificate, 

164 ) -> "Certificate": 

165 """ 

166 Create a Certificate object from a `cryptography.x509.Certificate` 

167 object. 

168 

169 Args: 

170 cert: The `cryptography.x509.Certificate` object. 

171 

172 Returns: 

173 Certificate: The created Certificate object. 

174 

175 --8<-- "docs/examples/certificate_from_cryptography.md" 

176 """ 

177 extensions = None 

178 if cert.extensions: 

179 extensions = Extensions.from_cryptography(cert.extensions) 

180 

181 signature_algorithm = None 

182 if cert.signature_hash_algorithm: 

183 signature_algorithm = SignatureAlgorithm.from_cryptography( 

184 cert.signature_hash_algorithm, 

185 cert.signature_algorithm_parameters, 

186 ) 

187 

188 ret = cls( 

189 version=cert.version.value, 

190 serial_number=cert.serial_number, 

191 signature_algorithm=signature_algorithm, 

192 issuer=Name.from_cryptography(cert.issuer), 

193 validity=Validity( 

194 not_before=cert.not_valid_before_utc, 

195 not_after=cert.not_valid_after_utc, 

196 ), 

197 subject=Name.from_cryptography(cert.subject), 

198 subject_public_key_info=SubjectPublicKeyInfo.from_cryptography( 

199 cert.public_key() 

200 ), 

201 extensions=extensions, 

202 signature_value=_byte_to_hex(cert.signature), 

203 _x509_obj=cert, 

204 ) 

205 ret._x509_obj = cert 

206 return ret 

207 

208 @classmethod 

209 def from_server( 

210 cls: type["Certificate"], 

211 uri: str, 

212 cache_time_seconds: int = CACHE_TIME_SECONDS, 

213 ) -> "Certificate": 

214 """ 

215 Loads a server certificate from a URI 

216 

217 Args: 

218 uri: The https URI of the server containing the certificate 

219 cache_time_seconds: How long the request should be cached in memory 

220 

221 Returns: 

222 The loaded [Certificate][pki_tools.types.certificate] object 

223 

224 --8<-- "docs/examples/certificate_from_server.md" 

225 """ 

226 cert_uri = CertsUri(uri=uri, cache_time_seconds=cache_time_seconds) 

227 

228 cache_ttl = round(time.time() / cert_uri.cache_time_seconds) 

229 pem = _download_server_certificate(cert_uri.hostname, cache_ttl) 

230 return Certificate.from_pem_string(pem) 

231 

232 @classmethod 

233 def from_uri( 

234 cls: type["Certificate"], 

235 uri: str, 

236 cache_time_seconds: int = CACHE_TIME_SECONDS, 

237 proxy: str | None = None, 

238 ) -> "Certificate": 

239 """ 

240 Loads Certificates from a URI. 

241 

242 Args: 

243 uri: URI where the certificate can be downloaded. 

244 cache_time_seconds: Specifies how long the certificate 

245 should be cached, default is 1 month. 

246 proxy: Optional proxy URL to use for the HTTP request 

247 (e.g., 'http://proxy.example.com:8080') 

248 

249 Returns: 

250 Instance of Certificate containing the certificates 

251 fetched from the URI. 

252 

253 --8<-- "docs/examples/certificate_from_uri.md" 

254 """ 

255 

256 cache_ttl = round(time.time() / cache_time_seconds) 

257 cert_uri = CertsUri(uri=uri) 

258 res = _download_cached(cert_uri.uri, cache_ttl, proxy) 

259 

260 return Certificate.from_pem_string(res.text) 

261 

262 @property 

263 def tbs_bytes(self) -> bytes: 

264 """ 

265 Returns: 

266 The to be signed bytes of this certificate 

267 """ 

268 return self._crypto_object.tbs_certificate_bytes 

269 

270 @property 

271 def hex_serial(self) -> str: 

272 """ 

273 Parses the certificate serial into hex format 

274 

275 Returns: 

276 String representing the hex value of the certificate serial number 

277 """ 

278 hex_serial = format(self.serial_number, "x").zfill(32) 

279 return hex_serial.upper() 

280 

281 @property 

282 def public_key(self) -> bytes: 

283 """ 

284 Returns: 

285 The bytes of the public key in PEM format 

286 """ 

287 return self.subject_public_key_info.algorithm.pem_bytes 

288 

289 @property 

290 def sign_alg_oid_name(self) -> str: 

291 """ 

292 Returns: 

293 The name of the signature algorithm such as: `SHA512WITHRSA` 

294 """ 

295 name = self._crypto_object.signature_algorithm_oid._name.upper() 

296 return name.replace("ENCRYPTION", "") 

297 

298 def digest( 

299 self, 

300 algorithm: HashAlgorithm = _DEFAULT_DIGEST_ALGORITHM, 

301 ) -> str: 

302 """ 

303 Gets the base64 encoded fingerprint of the certificate 

304 

305 Args: 

306 algorithm: The algorithm to use to hash the fingerprint with 

307 

308 Returns: 

309 Hashed and base64 encoded certificate fingerprint 

310 """ 

311 fingerprint = self._crypto_object.fingerprint( 

312 algorithm._to_cryptography() 

313 ) 

314 return base64.urlsafe_b64encode(fingerprint).decode("ascii") 

315 

316 def verify_signature( 

317 self: type["Certificate"], 

318 signed: InitCryptoParser, 

319 ) -> None: 

320 """ 

321 Verifies a signature of a signed entity against this issuer certificate 

322 

323 Args: 

324 signed: The signed entity can either be a 

325 [Certificate][pki_tools.types.certificate.Certificate], 

326 [CertificateRevocationList][pki_tools.types.crl.CertificateRevocationList] 

327 or a [OCSPResponse][pki_tools.types.ocsp.OCSPResponse] 

328 

329 Raises: 

330 InvalidSignedType: When the issuer has a non-supported type 

331 SignatureVerificationFailed: When the signature verification fails 

332 """ 

333 try: 

334 self.subject_public_key_info.algorithm.verify(signed) 

335 logger.trace("Signature valid") 

336 except Exception as e: # noqa: BLE001 - re-raised below 

337 logger.bind( 

338 exceptionType=type(e).__name__, 

339 exception=str(e), 

340 ).error("Signature verification failed") 

341 raise SignatureVerificationFailed( 

342 f"signature doesn't match issuer " 

343 f"with subject: {self.subject!s}" 

344 ) 

345 

346 def sign( 

347 self, 

348 key_pair: CryptoKeyPair, 

349 signature_algorithm: SignatureAlgorithm | None = None, 

350 req_key: CryptoPublicKey | None = None, 

351 ) -> None: 

352 """ 

353 Signs a created [Certificate][pki_tools.types.certificate.Certificate] 

354 object with a given 

355 [CryptoKeyPair][pki_tools.types.key_pair.CryptoKeyPair] 

356 

357 Args: 

358 key_pair: Keypair containing the private key to sing the 

359 certificate with 

360 signature_algorithm: Algorithm to use for the signature 

361 req_key: Can be used to sign another public key, defaults to the 

362 public key part in `key_pair` 

363 """ 

364 self._key_pair = key_pair 

365 self.serial_number = random.randint(1, 2**32 - 1) 

366 self.signature_algorithm = signature_algorithm 

367 

368 if req_key is None: 

369 req_key = key_pair.public_key 

370 

371 self.subject_public_key_info = SubjectPublicKeyInfo( 

372 algorithm=req_key, 

373 parameters=req_key._string_dict(), 

374 ) 

375 

376 self._x509_obj = self._to_cryptography() 

377 

378 def _to_cryptography(self) -> x509.Certificate: 

379 if hasattr(self, "_x509_obj"): 

380 return self._x509_obj 

381 

382 if not hasattr(self, "_key_pair"): 

383 raise MissingInit( 

384 f"Please use Certificate.{self._init_func} function" 

385 ) 

386 

387 subject = self.subject._to_cryptography() 

388 issuer = self.issuer._to_cryptography() 

389 crypto_key = self._key_pair.private_key._to_cryptography() 

390 

391 alg = self.subject_public_key_info.algorithm 

392 public_key = alg._to_cryptography() 

393 

394 cert_builder = ( 

395 x509.CertificateBuilder() 

396 .subject_name( 

397 subject, 

398 ) 

399 .issuer_name( 

400 issuer, 

401 ) 

402 .serial_number( 

403 x509.random_serial_number(), 

404 ) 

405 .public_key(public_key) 

406 .not_valid_before( 

407 self.validity.not_before, 

408 ) 

409 .not_valid_after( 

410 self.validity.not_after, 

411 ) 

412 ) 

413 

414 if self.extensions is not None: 

415 for extension in self.extensions: 

416 cert_builder = cert_builder.add_extension( 

417 extension._to_cryptography(), extension.critical 

418 ) 

419 

420 if isinstance( 

421 crypto_key, 

422 (ed25519.Ed25519PrivateKey, ed448.Ed448PrivateKey), 

423 ): 

424 # Ed25519/Ed448 signing keys must use a None algorithm. This is 

425 # decided by the *signing* key, not the subject public key (which 

426 # can differ when a CA issues a cert carrying another key). 

427 alg = None 

428 else: 

429 alg = self.signature_algorithm.algorithm._to_cryptography() 

430 

431 cert = cert_builder.sign(crypto_key, alg) 

432 

433 return cert 

434 

435 def _string_dict(self): 

436 ret = { 

437 "Issuer": str(self.issuer), 

438 "Validity": self.validity._string_dict(), 

439 "Subject": str(self.subject), 

440 } 

441 if self.version is not None: 

442 ret["Version"] = self.version 

443 if self.extensions: 

444 ret["Extensions"] = self.extensions._string_dict() 

445 if self.serial_number is not None: 

446 ret["Serial Number"] = self.hex_serial 

447 if self.signature_value is not None: 

448 ret["Signature Value"] = self.signature_value 

449 if self.subject_public_key_info is not None: 

450 subject_key_info = self.subject_public_key_info._string_dict() 

451 ret["Subject Public Key Info"] = subject_key_info 

452 if self.signature_algorithm is not None: 

453 signature_alg = self.signature_algorithm.algorithm.name.value 

454 ret["Signature Algorithm"] = signature_alg 

455 

456 return ret 

457 

458 @classmethod 

459 def _crypto_config(cls) -> CryptoConfig: 

460 return CryptoConfig( 

461 load_pem=HelperFunc(func=x509.load_pem_x509_certificate), 

462 load_der=HelperFunc(func=x509.load_der_x509_certificate), 

463 pem_regexp=PEM_CERT_REGEX, 

464 )