Coverage for pki_tools/ocsp.py: 82%

1import time 

2from functools import cache 

3 

4from loguru import logger 

5 

6from pki_tools.exceptions import ( 

7 Error, 

8 ExtensionMissing, 

9 OcspFetchFailure, 

10 OcspInvalidResponseStatus, 

11) 

12from pki_tools.types.certificate import Certificate 

13from pki_tools.types.chain import Chain 

14from pki_tools.types.extensions import ( 

15 AccessDescriptionId, 

16 AuthorityInformationAccess, 

17 UniformResourceIdentifier, 

18) 

19from pki_tools.types.ocsp import OCSPRequest, OCSPResponse 

20from pki_tools.types.signature_algorithm import HashAlgorithm 

21from pki_tools.types.utils import HTTPX_CLIENT 

22 

23OCSP_ALGORITHMS_TO_CHECK = [ 

24 HashAlgorithm(name="SHA256"), 

25 HashAlgorithm(name="SHA1"), 

26 HashAlgorithm(name="SHA512"), 

27 HashAlgorithm(name="SHA224"), 

28 HashAlgorithm(name="SHA384"), 

29] 

30 

31 

32def _is_revoked_multiple_issuers( 

33 cert: Certificate, 

34 cert_issuer: Chain, 

35 ocsp_issuer: Chain, 

36 ocsp_res_cache_seconds: int = 3600, 

37): 

38 cert_issuer.validate_trust_path(cert) 

39 issuer = cert_issuer.get_issuer(cert) 

40 

41 log = logger.bind( 

42 cert=cert.pem_string, 

43 serial=cert.serial_number, 

44 ) 

45 

46 if cert.extensions.authority_information_access is None: 

47 log.debug("OCSP extension missing") 

48 raise ExtensionMissing() 

49 

50 for i, alg in enumerate(OCSP_ALGORITHMS_TO_CHECK): 

51 try: 

52 req_path = _construct_req_path(cert, issuer, alg) 

53 

54 return _check_ocsp_status( 

55 cert.extensions.authority_information_access, 

56 req_path, 

57 cert, 

58 ocsp_issuer, 

59 ocsp_res_cache_seconds=ocsp_res_cache_seconds, 

60 ) 

61 except OcspInvalidResponseStatus: 

62 log.bind(alg=alg.name.value).debug( 

63 "OCSP check failed, trying another algorithm" 

64 ) 

65 if i + 1 == len(OCSP_ALGORITHMS_TO_CHECK): 

66 log.bind( 

67 algs=[alg.name.value for alg in OCSP_ALGORITHMS_TO_CHECK] 

68 ).debug("All algorithms check failed") 

69 raise 

70 

71 return False 

72 

73 

74def _construct_req_path(cert, issuer_cert, alg): 

75 req = OCSPRequest(hash_algorithm=alg) 

76 req.create(cert, issuer_cert) 

77 return req.request_path 

78 

79 

80def _check_ocsp_status( 

81 aia: AuthorityInformationAccess, 

82 req_path, 

83 cert: Certificate, 

84 issuer_chain: Chain, 

85 ocsp_res_cache_seconds: int = 3600, 

86): 

87 log = logger.bind(serial=cert.hex_serial) 

88 

89 checked_status = False 

90 for access_description in aia: 

91 if access_description.access_method != AccessDescriptionId.OCSP: 

92 logger.trace( 

93 "Access method is not OCSP, " 

94 "try checking next access description" 

95 ) 

96 continue 

97 

98 if not isinstance( 

99 access_description.access_location, UniformResourceIdentifier 

100 ): 

101 continue 

102 

103 checked_status = True 

104 

105 server = access_description.access_location.value 

106 

107 cache_ttl = round(time.time() / ocsp_res_cache_seconds) 

108 ocsp_res = _get_ocsp_status( 

109 f"{server}/{req_path}", cache_ttl=cache_ttl 

110 ) 

111 

112 _verify_ocsp_signature(ocsp_res, issuer_chain) 

113 

114 if ocsp_res.is_revoked: 

115 log.bind( 

116 date=str(ocsp_res._x509_obj.revocation_time_utc), 

117 ).debug("Certificate revoked") 

118 return True 

119 else: 

120 return False 

121 

122 if not checked_status: 

123 raise ExtensionMissing() 

124 

125 log.debug("Certificate valid") 

126 return False 

127 

128 

129@cache 

130def _get_ocsp_status(uri, cache_ttl=None) -> OCSPResponse: 

131 ret = HTTPX_CLIENT.get( 

132 uri, headers={"Content-Type": "application/ocsp-request"} 

133 ) 

134 

135 log = logger.bind(status=ret.status_code) 

136 if ret.status_code != 200: 

137 log.error("OCSP status fetch failed") 

138 raise OcspFetchFailure( 

139 f"Unexpected response status code: {ret.status_code}" 

140 ) 

141 

142 ocsp_res = OCSPResponse.from_der_bytes(ret.content) 

143 

144 if not ocsp_res.is_successful: 

145 log.bind(res=ocsp_res.response_status).debug("Invalid OCSP response") 

146 raise OcspInvalidResponseStatus( 

147 f"Invalid OCSP Response status: {ocsp_res.response_status}" 

148 ) 

149 

150 return ocsp_res 

151 

152 

153def _verify_ocsp_signature(ocsp_response: OCSPResponse, issuer_chain: Chain): 

154 found_issuer = None 

155 for issuer_cert in issuer_chain.certificates: 

156 public_key = issuer_cert.subject_public_key_info.algorithm 

157 cert_public_hash = ocsp_response.hash_with_alg(public_key.ocsp_bytes) 

158 

159 if cert_public_hash == ocsp_response.issuer_key_hash: 

160 found_issuer = issuer_cert 

161 

162 if found_issuer is not None: 

163 found_issuer.verify_signature(ocsp_response) 

164 # Validate the chain path for the found issuer (unless it's a root) 

165 if found_issuer.issuer != found_issuer.subject: 

166 issuer_chain.validate_trust_path(found_issuer) 

167 return 

168 

169 logger.warning( 

170 "Couldn't find issuer by key hash, " 

171 "trying to verify signature against all " 

172 "certificates in chain" 

173 ) 

174 for issuer_cert in issuer_chain.certificates: 

175 try: 

176 issuer_cert.verify_signature(ocsp_response) 

177 # Validate the chain path for the signing cert (unless it's a root) 

178 if issuer_cert.issuer != issuer_cert.subject: 

179 issuer_chain.validate_trust_path(issuer_cert) 

180 return 

181 except Exception as e: # noqa: BLE001 - try each issuer, keep going 

182 logger.error(f"Signature verification failed: {e}") 

183 continue 

184 

185 raise Error("Couldn't verify OCSP signature")