Coverage for pki_tools/types/ocsp.py: 89%

1import base64 

2import hashlib 

3import re 

4from datetime import datetime, timezone 

5from enum import Enum 

6 

7from cryptography.x509 import ocsp 

8from cryptography.x509.ocsp import OCSPCertStatus 

9from loguru import logger 

10 

11from pki_tools.exceptions import ( 

12 MissingInit, 

13) 

14from pki_tools.types.certificate import Certificate 

15from pki_tools.types.crypto_parser import ( 

16 CryptoConfig, 

17 HelperFunc, 

18 InitCryptoParser, 

19) 

20from pki_tools.types.extensions import Extensions 

21from pki_tools.types.key_pair import ( 

22 CryptoPrivateKey, 

23 Ed448PrivateKey, 

24 Ed25519PrivateKey, 

25) 

26from pki_tools.types.signature_algorithm import ( 

27 HashAlgorithm, 

28 SignatureAlgorithm, 

29) 

30from pki_tools.types.utils import _byte_to_hex 

31 

32 

33class OcspResponseStatus(Enum): 

34 """ 

35 Enumeration of OCSP response statuses. 

36 """ 

37 

38 SUCCESSFUL = "SUCCESSFUL" 

39 MALFORMED_REQUEST = "MALFORMED_REQUEST" 

40 INTERNAL_ERROR = "INTERNAL_ERROR" 

41 TRY_LATER = "TRY_LATER" 

42 SIG_REQUIRED = "SIG_REQUIRED" 

43 UNAUTHORIZED = "UNAUTHORIZED" 

44 

45 

46class OcspCertificateStatus(Enum): 

47 """ 

48 Enumeration of OCSP certificate statuses. 

49 """ 

50 

51 GOOD = "GOOD" 

52 REVOKED = "REVOKED" 

53 UNKNOWN = "UNKNOWN" 

54 

55 

56RESPONSE_REGEXP = re.compile( 

57 r"\s*-+BEGIN OCSP RESPONSE-+[\w+/\s=]*-+END OCSP RESPONSE-+\s*" 

58) 

59 

60 

61class OCSPResponse(InitCryptoParser): 

62 """ 

63 Represents an OCSP response. 

64 

65 Attributes: 

66 response_status: The OCSP response status. 

67 certificate_status: The OCSP certificate status. 

68 issuer_key_hash: The issuer key hash. 

69 revocation_time: The revocation time. 

70 

71 --8<-- "docs/examples/ocsp_response.md" 

72 """ 

73 

74 response_status: OcspResponseStatus 

75 certificate_status: OcspCertificateStatus | None = None 

76 issuer_key_hash: str | None = None 

77 revocation_time: datetime | None = None 

78 

79 @classmethod 

80 def from_cryptography( 

81 cls: type["OCSPResponse"], crypto_ocsp_response: ocsp.OCSPResponse 

82 ) -> "OCSPResponse": 

83 """ 

84 Constructs an OCSPResponse object from a cryptography 

85 OCSPResponse object. 

86 

87 Args: 

88 crypto_ocsp_response: The cryptography OCSPResponse object. 

89 

90 Returns: 

91 OCSPResponse: The constructed OCSPResponse object. 

92 

93 --8<-- "docs/examples/ocsp_response_from_cryptography.md" 

94 """ 

95 response_status = crypto_ocsp_response.response_status 

96 

97 ocsp_response_key_hash = None 

98 certificate_status = None 

99 revocation_time = None 

100 if response_status == ocsp.OCSPResponseStatus.SUCCESSFUL: 

101 certificate_status = crypto_ocsp_response.certificate_status.name 

102 try: 

103 ocsp_response_key_hash = _byte_to_hex( 

104 crypto_ocsp_response.issuer_key_hash 

105 ) 

106 except Exception as e: 

107 logger.bind( 

108 exceptionType=type(e), 

109 exception=str(e), 

110 issuerHash=crypto_ocsp_response.issuer_key_hash, 

111 ).error("Couldn't convert issuer key hash to hex") 

112 raise 

113 

114 if crypto_ocsp_response.revocation_time_utc is not None: 

115 revocation_time = crypto_ocsp_response.revocation_time_utc 

116 

117 ret = cls( 

118 response_status=getattr(OcspResponseStatus, response_status.name), 

119 certificate_status=certificate_status, 

120 issuer_key_hash=ocsp_response_key_hash, 

121 revocation_time=revocation_time, 

122 _x509_obj=crypto_ocsp_response, 

123 ) 

124 ret._x509_obj = crypto_ocsp_response 

125 return ret 

126 

127 @property 

128 def tbs_bytes(self) -> bytes: 

129 """ 

130 Returns the bytes to be singed of the OCSP response. 

131 

132 Returns: 

133 bytes: The TBS bytes. 

134 """ 

135 return self._crypto_object.tbs_response_bytes 

136 

137 @property 

138 def pem_bytes(self) -> bytes: 

139 """ 

140 Returns the PEM bytes of the object 

141 

142 Returns: 

143 The PEM bytes. 

144 """ 

145 ocsp_request_b64 = base64.b64encode(self.der_bytes).decode() 

146 

147 # Create the PEM string with headers and footers 

148 return ( 

149 "-----BEGIN OCSP RESPONSE-----\n" 

150 + "\n".join( 

151 [ 

152 ocsp_request_b64[i : i + 64] 

153 for i in range(0, len(ocsp_request_b64), 64) 

154 ] 

155 ) 

156 + "\n-----END OCSP RESPONSE-----\n" 

157 ).encode() 

158 

159 def hash_with_alg(self, der_key: bytes) -> str: 

160 """ 

161 Hashes a DER key bytes with the algorithm of the OCSP response. 

162 

163 Args: 

164 der_key: The DER key. 

165 

166 Returns: 

167 str: The hashed key. 

168 """ 

169 

170 hash_algorithm = hashlib.new(self._crypto_object.hash_algorithm.name) 

171 hash_algorithm.update(der_key) 

172 return hash_algorithm.hexdigest().upper() 

173 

174 @property 

175 def is_successful(self) -> bool: 

176 """ 

177 Checks if the OCSP response is successful. 

178 

179 Returns: 

180 True if the response is successful, False otherwise. 

181 """ 

182 return self.response_status == OcspResponseStatus.SUCCESSFUL 

183 

184 @property 

185 def is_revoked(self) -> bool: 

186 """ 

187 Checks if the certificate is revoked. 

188 

189 Returns: 

190 True if the certificate is revoked, False otherwise. 

191 """ 

192 return self.certificate_status == OcspCertificateStatus.REVOKED 

193 

194 def sign( 

195 self, 

196 cert: Certificate, 

197 issuer: Certificate, 

198 response_algorithm: SignatureAlgorithm, 

199 private_key: CryptoPrivateKey, 

200 signature_algorithm: SignatureAlgorithm | None = None, 

201 ): 

202 """ 

203 Signs the OCSP response. 

204 

205 Args: 

206 cert: The certificate. 

207 issuer: The issuer certificate. 

208 response_algorithm: The signature algorithm for the response. 

209 private_key: The private key to sign the response. 

210 signature_algorithm: The signature algorithm. 

211 """ 

212 self._cert = cert 

213 self._issuer = issuer 

214 self._response_algorithm = response_algorithm 

215 self._private_key = private_key 

216 self._signature_algorithm = signature_algorithm 

217 self._x509_obj = self._to_cryptography() 

218 

219 def _to_cryptography(self) -> ocsp.OCSPResponse: 

220 if not hasattr(self, "_private_key"): 

221 raise MissingInit( 

222 f"Please use OCSPResponse.{self._init_func} function" 

223 ) 

224 

225 builder = ocsp.OCSPResponseBuilder() 

226 cert_status = None 

227 if self.certificate_status is not None: 

228 cert_status = getattr(OCSPCertStatus, self.certificate_status.name) 

229 

230 builder = builder.add_response( 

231 cert=self._cert._to_cryptography(), 

232 issuer=self._issuer._to_cryptography(), 

233 algorithm=self._response_algorithm.algorithm._to_cryptography(), 

234 cert_status=cert_status, 

235 this_update=datetime.now(timezone.utc), 

236 next_update=datetime.now(timezone.utc), 

237 revocation_reason=None, 

238 revocation_time=self.revocation_time, 

239 ).responder_id( 

240 ocsp.OCSPResponderEncoding.HASH, self._cert._to_cryptography() 

241 ) 

242 

243 if isinstance(self._private_key, (Ed25519PrivateKey, Ed448PrivateKey)): 

244 alg = None 

245 else: 

246 alg = self._signature_algorithm.algorithm._to_cryptography() 

247 

248 return builder.sign( 

249 self._private_key._to_cryptography(), 

250 alg, 

251 ) 

252 

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

254 ret = { 

255 "Response Status": self.response_status.value, 

256 } 

257 if self.certificate_status is not None: 

258 ret["Certificate Status"] = self.certificate_status.value 

259 if self.issuer_key_hash is not None: 

260 ret["Issuer Key Hash"] = self.issuer_key_hash 

261 if self.revocation_time is not None: 

262 ret["Revocation Time"] = self.revocation_time 

263 return ret 

264 

265 @classmethod 

266 def _load_pem(cls, pem: bytes): 

267 pem_lines = pem.decode().strip().split("\n") 

268 base64_data = "".join(pem_lines[1:-1]) 

269 der_bytes = base64.b64decode(base64_data) 

270 

271 return ocsp.load_der_ocsp_response(der_bytes) 

272 

273 @classmethod 

274 def _crypto_config(cls) -> CryptoConfig: 

275 return CryptoConfig( 

276 load_pem=HelperFunc(func=cls._load_pem), 

277 load_der=HelperFunc(func=ocsp.load_der_ocsp_response), 

278 pem_regexp=RESPONSE_REGEXP, 

279 ) 

280 

281 

282REQUEST_REGEXP = re.compile( 

283 r"\s*-+BEGIN OCSP REQUEST-+[\w+/\s=]*-+END OCSP REQUEST-+\s*" 

284) 

285 

286 

287class OCSPRequest(InitCryptoParser): 

288 """ 

289 Represents an OCSP request. 

290 

291 Attributes: 

292 hash_algorithm: The hash algorithm. 

293 serial_number: The serial number. 

294 extensions: The extensions. 

295 

296 --8<-- "docs/examples/ocsp_request.md" 

297 """ 

298 

299 hash_algorithm: HashAlgorithm 

300 

301 serial_number: int | None = None 

302 extensions: Extensions | None = None 

303 

304 _init_func = "create" 

305 

306 @classmethod 

307 def from_cryptography( 

308 cls: type["OCSPRequest"], crypto_obj: ocsp.OCSPRequest 

309 ) -> "OCSPRequest": 

310 """ 

311 Constructs an OCSPRequest object from a cryptography OCSPRequest 

312 object. 

313 

314 Args: 

315 crypto_obj: The cryptography OCSPRequest object. 

316 

317 Returns: 

318 The constructed OCSPRequest object. 

319 

320 --8<-- "docs/examples/ocsp_request_from_cryptography.md" 

321 """ 

322 alg = HashAlgorithm.from_cryptography(crypto_obj.hash_algorithm) 

323 ret = cls( 

324 hash_algorithm=alg, 

325 serial_number=crypto_obj.serial_number, 

326 extensions=Extensions.from_cryptography(crypto_obj.extensions), 

327 _x590_obj=crypto_obj, 

328 ) 

329 ret._x509_obj = crypto_obj 

330 return ret 

331 

332 @property 

333 def pem_bytes(self) -> bytes: 

334 """ 

335 Returns the PEM bytes of the object 

336 

337 Returns: 

338 The PEM bytes. 

339 """ 

340 ocsp_request_b64 = base64.b64encode(self.der_bytes).decode() 

341 

342 return ( 

343 "-----BEGIN OCSP REQUEST-----\n" 

344 + "\n".join( 

345 [ 

346 ocsp_request_b64[i : i + 64] 

347 for i in range(0, len(ocsp_request_b64), 64) 

348 ] 

349 ) 

350 + "\n-----END OCSP REQUEST-----\n" 

351 ).encode() 

352 

353 @property 

354 def request_path(self) -> str: 

355 """ 

356 The request path of the OCSP Response. 

357 

358 Returns: 

359 The request path. 

360 """ 

361 return base64.b64encode(self.der_bytes).decode() 

362 

363 def create(self, cert: Certificate, issuer_cert: Certificate): 

364 """ 

365 Creates an OCSP request. 

366 

367 Args: 

368 cert: The certificate. 

369 issuer_cert: The issuer of the OCSP Response. 

370 """ 

371 self._cert = cert 

372 self._issuer = issuer_cert 

373 self._x509_obj = self._to_cryptography() 

374 

375 def _to_cryptography(self) -> ocsp.OCSPRequest: 

376 if not hasattr(self, "_cert"): 

377 raise MissingInit( 

378 f"Please use OCSPRequest.{self._init_func} function" 

379 ) 

380 

381 builder = ocsp.OCSPRequestBuilder() 

382 builder = builder.add_certificate( 

383 self._cert._to_cryptography(), 

384 self._issuer._to_cryptography(), 

385 self.hash_algorithm._to_cryptography(), 

386 ) 

387 return builder.build() 

388 

389 def _string_dict(self) -> dict: 

390 ret = self.hash_algorithm._string_dict() 

391 

392 if self.serial_number is not None: 

393 ret["Serial Number"] = str(self.serial_number) 

394 if self.extensions is not None: 

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

396 

397 return ret 

398 

399 @classmethod 

400 def _load_pem(cls, pem: bytes): 

401 pem_lines = pem.decode().strip().split("\n") 

402 base64_data = "".join(pem_lines[1:-1]) 

403 der_bytes = base64.b64decode(base64_data) 

404 

405 return ocsp.load_der_ocsp_request(der_bytes) 

406 

407 @classmethod 

408 def _crypto_config(cls) -> CryptoConfig: 

409 return CryptoConfig( 

410 load_pem=HelperFunc(func=cls._load_pem), 

411 load_der=HelperFunc(func=ocsp.load_der_ocsp_request), 

412 pem_regexp=REQUEST_REGEXP, 

413 )