Coverage for pki_tools/types/crl.py: 86%

1import re 

2import time 

3from datetime import datetime 

4 

5from cryptography import x509 

6from loguru import logger 

7 

8from pki_tools.exceptions import LoadError, MissingInit 

9from pki_tools.types.crypto_parser import ( 

10 CryptoConfig, 

11 CryptoParser, 

12 HelperFunc, 

13 InitCryptoParser, 

14) 

15from pki_tools.types.extensions import Extensions 

16from pki_tools.types.key_pair import ( 

17 CryptoPrivateKey, 

18 Ed448PrivateKey, 

19 Ed25519PrivateKey, 

20) 

21from pki_tools.types.name import Name 

22from pki_tools.types.signature_algorithm import SignatureAlgorithm 

23from pki_tools.types.utils import ( 

24 CACHE_TIME_SECONDS, 

25 CertsUri, 

26 _download_cached, 

27) 

28 

29 

30class RevokedCertificate(CryptoParser): 

31 """ 

32 Represents a revoked certificate. 

33 

34 Attributes: 

35 serial: The serial number of the certificate. 

36 date: The revocation date of the certificate. 

37 extensions: Extensions associated with the certificate. 

38 """ 

39 

40 serial: int 

41 date: datetime 

42 extensions: Extensions | None = None 

43 

44 @classmethod 

45 def from_cryptography( 

46 cls: type["RevokedCertificate"], crypto_obj: x509.RevokedCertificate 

47 ) -> "RevokedCertificate": 

48 """ 

49 Create a RevokedCertificate object from a cryptography 

50 RevokedCertificate. 

51 

52 Args: 

53 crypto_obj: Cryptography RevokedCertificate. 

54 

55 Returns: 

56 RevokedCertificate: Instance of RevokedCertificate. 

57 """ 

58 extensions = None 

59 if crypto_obj.extensions: 

60 extensions = Extensions.from_cryptography(crypto_obj.extensions) 

61 

62 ret = cls( 

63 serial=crypto_obj.serial_number, 

64 date=crypto_obj.revocation_date_utc, 

65 extensions=extensions, 

66 _x509_obj=crypto_obj, 

67 ) 

68 

69 return ret 

70 

71 def _to_cryptography(self) -> x509.RevokedCertificate: 

72 builder = ( 

73 x509.RevokedCertificateBuilder() 

74 .serial_number(self.serial) 

75 .revocation_date(self.date) 

76 ) 

77 

78 if self.extensions is not None: 

79 for extension in self.extensions: 

80 builder = builder.add_extension( 

81 extval=extension._to_cryptography(), 

82 critical=extension.critical, 

83 ) 

84 

85 return builder.build() 

86 

87 def _string_dict(self) -> dict: 

88 return { 

89 "Serial Number": self.serial, 

90 "Revocation Date": str(self.date), 

91 "Extensions": self.extensions._string_dict(), 

92 } 

93 

94 

95CRL_REGEXP = re.compile(r"\s*-+BEGIN X509 CRL-+[\w+/\s=]*-+END X509 CRL-+\s*") 

96 

97 

98class CertificateRevocationList(InitCryptoParser): 

99 """ 

100 Represents a certificate revocation list (CRL). 

101 

102 Attributes: 

103 issuer: The name of the issuer. 

104 last_update: The last update time of the CRL. 

105 next_update: The next update time of the CRL. 

106 revoked_certs: List of revoked certificates. 

107 """ 

108 

109 issuer: Name 

110 last_update: datetime 

111 next_update: datetime 

112 revoked_certs: list[RevokedCertificate] | None = None 

113 extensions: Extensions | None = None 

114 

115 _private_key: CryptoPrivateKey 

116 

117 @classmethod 

118 def from_cryptography( 

119 cls: type["CertificateRevocationList"], 

120 crypto_crl: x509.CertificateRevocationList, 

121 ) -> "CertificateRevocationList": 

122 """ 

123 Create a CertificateRevocationList object from a cryptography 

124 CertificateRevocationList. 

125 

126 Args: 

127 crypto_crl: Cryptography CertificateRevocationList. 

128 

129 Returns: 

130 Instance of CertificateRevocationList. 

131 """ 

132 

133 extensions = None 

134 if crypto_crl.extensions is not None: 

135 extensions = Extensions.from_cryptography(crypto_crl.extensions) 

136 

137 ret = cls( 

138 issuer=Name.from_cryptography(crypto_crl.issuer), 

139 last_update=crypto_crl.last_update_utc, 

140 next_update=crypto_crl.next_update_utc, 

141 extensions=extensions, 

142 ) 

143 ret._x509_obj = crypto_crl 

144 return ret 

145 

146 @classmethod 

147 def from_uri( 

148 cls: type["CertificateRevocationList"], 

149 uri: str, 

150 cache_time_seconds: int = CACHE_TIME_SECONDS, 

151 proxy: str | None = None, 

152 ) -> "CertificateRevocationList": 

153 """ 

154 Loads CertificateRevocationList from a URI. 

155 

156 Args: 

157 uri: URI where the CRL can be downloaded. 

158 cache_time_seconds: Specifies how long the CRL 

159 should be cached, default is 1 month. 

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

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

162 

163 Returns: 

164 Instance of CertificateRevocationList containing the revoked 

165 certificates fetched from the URI. 

166 """ 

167 

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

169 crl_uri = CertsUri(uri=uri) 

170 res = _download_cached(crl_uri.uri, cache_ttl, proxy) 

171 

172 try: 

173 return CertificateRevocationList.from_der_bytes(res.content) 

174 except (TypeError, ValueError) as e: 

175 logger.bind(error=str(e)).trace("Error during loading of CRL DER") 

176 

177 try: 

178 return CertificateRevocationList.from_pem_string( 

179 res.content.decode() 

180 ) 

181 except TypeError as e: 

182 logger.bind(crl=res.content).error("Failed to load CRL") 

183 raise LoadError(e) from None 

184 

185 @property 

186 def tbs_bytes(self) -> bytes: 

187 """ 

188 Return the bytes to be signed of the CRL. 

189 

190 Returns: 

191 TBS bytes of the CRL. 

192 """ 

193 return self._crypto_object.tbs_certlist_bytes 

194 

195 def sign( 

196 self, 

197 private_key: CryptoPrivateKey, 

198 algorithm: SignatureAlgorithm | None = None, 

199 ) -> None: 

200 """ 

201 Sign the CRL with the provided private key and algorithm. 

202 

203 Args: 

204 private_key: Private key used to sign the CRL. 

205 algorithm: Signature algorithm to use for signing. 

206 """ 

207 self._private_key = private_key 

208 self._signature_algorithm = algorithm 

209 self._x509_obj = self._to_cryptography() 

210 

211 def get_revoked(self, cert_serial: int) -> RevokedCertificate | None: 

212 """ 

213 Get a revoked certificate by serial number. 

214 

215 Args: 

216 cert_serial: Serial number of the certificate. 

217 

218 Returns: 

219 RevokedCertificate object if found, else None. 

220 """ 

221 crypto_revoked = ( 

222 self._crypto_object.get_revoked_certificate_by_serial_number( 

223 cert_serial 

224 ) 

225 ) 

226 if crypto_revoked is not None: 

227 return RevokedCertificate.from_cryptography(crypto_revoked) 

228 return None 

229 

230 def _to_cryptography(self) -> x509.CertificateRevocationList: 

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

232 raise MissingInit("Please use 'sign' function") 

233 

234 builder = x509.CertificateRevocationListBuilder() 

235 builder = builder.issuer_name(self.issuer._to_cryptography()) 

236 builder = builder.last_update(self.last_update) 

237 builder = builder.next_update(self.next_update) 

238 

239 if self.extensions is not None: 

240 for extension in self.extensions: 

241 builder = builder.add_extension( 

242 extval=extension._to_cryptography(), 

243 critical=extension.critical, 

244 ) 

245 

246 for cert in self.revoked_certs: 

247 builder = builder.add_revoked_certificate(cert._to_cryptography()) 

248 

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

250 alg = None 

251 else: 

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

253 

254 return builder.sign( 

255 private_key=self._private_key._to_cryptography(), 

256 algorithm=alg, 

257 ) 

258 

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

260 certs = [] 

261 for cert in self.revoked_certs: 

262 certs.append(cert._string_dict()) 

263 

264 return {"Issuer": self.name._string_dict(), "Certs": certs} 

265 

266 @classmethod 

267 def _crypto_config(cls) -> CryptoConfig: 

268 return CryptoConfig( 

269 load_pem=HelperFunc(func=x509.load_pem_x509_crl), 

270 load_der=HelperFunc(func=x509.load_der_x509_crl), 

271 pem_regexp=CRL_REGEXP, 

272 )