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
1# This file is dual licensed under the terms of the Apache License, Version
2# 2.0, and the BSD License. See the LICENSE file in the root of this repository
3# for complete details.
5from __future__ import annotations
7import datetime
8from collections.abc import Iterable
10from cryptography import utils, x509
11from cryptography.hazmat.bindings._rust import ocsp
12from cryptography.hazmat.primitives import hashes
13from cryptography.hazmat.primitives.asymmetric.types import (
14 CertificateIssuerPrivateKeyTypes,
15)
16from cryptography.x509.base import _reject_duplicate_extension
19class OCSPResponderEncoding(utils.Enum):
20 HASH = "By Hash"
21 NAME = "By Name"
24class OCSPResponseStatus(utils.Enum):
25 SUCCESSFUL = 0
26 MALFORMED_REQUEST = 1
27 INTERNAL_ERROR = 2
28 TRY_LATER = 3
29 SIG_REQUIRED = 5
30 UNAUTHORIZED = 6
33_ALLOWED_HASHES = (
34 hashes.SHA1,
35 hashes.SHA224,
36 hashes.SHA256,
37 hashes.SHA384,
38 hashes.SHA512,
39)
42def _verify_algorithm(algorithm: hashes.HashAlgorithm) -> None:
43 if not isinstance(algorithm, _ALLOWED_HASHES):
44 raise ValueError(
45 "Algorithm must be SHA1, SHA224, SHA256, SHA384, or SHA512"
46 )
49class OCSPCertStatus(utils.Enum):
50 GOOD = 0
51 REVOKED = 1
52 UNKNOWN = 2
55class _SingleResponse:
56 def __init__(
57 self,
58 resp: tuple[x509.Certificate, x509.Certificate] | None,
59 resp_hash: tuple[bytes, bytes, int] | None,
60 algorithm: hashes.HashAlgorithm,
61 cert_status: OCSPCertStatus,
62 this_update: datetime.datetime,
63 next_update: datetime.datetime | None,
64 revocation_time: datetime.datetime | None,
65 revocation_reason: x509.ReasonFlags | None,
66 ):
67 _verify_algorithm(algorithm)
68 if not isinstance(this_update, datetime.datetime):
69 raise TypeError("this_update must be a datetime object")
70 if next_update is not None and not isinstance(
71 next_update, datetime.datetime
72 ):
73 raise TypeError("next_update must be a datetime object or None")
75 self._resp = resp
76 self._resp_hash = resp_hash
77 self._algorithm = algorithm
78 self._this_update = this_update
79 self._next_update = next_update
81 if not isinstance(cert_status, OCSPCertStatus):
82 raise TypeError(
83 "cert_status must be an item from the OCSPCertStatus enum"
84 )
85 if cert_status is not OCSPCertStatus.REVOKED:
86 if revocation_time is not None:
87 raise ValueError(
88 "revocation_time can only be provided if the certificate "
89 "is revoked"
90 )
91 if revocation_reason is not None:
92 raise ValueError(
93 "revocation_reason can only be provided if the certificate"
94 " is revoked"
95 )
96 else:
97 if not isinstance(revocation_time, datetime.datetime):
98 raise TypeError("revocation_time must be a datetime object")
100 if revocation_reason is not None and not isinstance(
101 revocation_reason, x509.ReasonFlags
102 ):
103 raise TypeError(
104 "revocation_reason must be an item from the ReasonFlags "
105 "enum or None"
106 )
108 self._cert_status = cert_status
109 self._revocation_time = revocation_time
110 self._revocation_reason = revocation_reason
113OCSPRequest = ocsp.OCSPRequest
114OCSPResponse = ocsp.OCSPResponse
115OCSPSingleResponse = ocsp.OCSPSingleResponse
118class OCSPRequestBuilder:
119 def __init__(
120 self,
121 request: tuple[
122 x509.Certificate, x509.Certificate, hashes.HashAlgorithm
123 ]
124 | None = None,
125 request_hash: tuple[bytes, bytes, int, hashes.HashAlgorithm]
126 | None = None,
127 extensions: list[x509.Extension[x509.ExtensionType]] = [],
128 ) -> None:
129 self._request = request
130 self._request_hash = request_hash
131 self._extensions = extensions
133 def add_certificate(
134 self,
135 cert: x509.Certificate,
136 issuer: x509.Certificate,
137 algorithm: hashes.HashAlgorithm,
138 ) -> OCSPRequestBuilder:
139 if self._request is not None or self._request_hash is not None:
140 raise ValueError("Only one certificate can be added to a request")
142 _verify_algorithm(algorithm)
143 if not isinstance(cert, x509.Certificate) or not isinstance(
144 issuer, x509.Certificate
145 ):
146 raise TypeError("cert and issuer must be a Certificate")
148 return OCSPRequestBuilder(
149 (cert, issuer, algorithm), self._request_hash, self._extensions
150 )
152 def add_certificate_by_hash(
153 self,
154 issuer_name_hash: bytes,
155 issuer_key_hash: bytes,
156 serial_number: int,
157 algorithm: hashes.HashAlgorithm,
158 ) -> OCSPRequestBuilder:
159 if self._request is not None or self._request_hash is not None:
160 raise ValueError("Only one certificate can be added to a request")
162 if not isinstance(serial_number, int):
163 raise TypeError("serial_number must be an integer")
165 _verify_algorithm(algorithm)
166 utils._check_bytes("issuer_name_hash", issuer_name_hash)
167 utils._check_bytes("issuer_key_hash", issuer_key_hash)
168 if algorithm.digest_size != len(
169 issuer_name_hash
170 ) or algorithm.digest_size != len(issuer_key_hash):
171 raise ValueError(
172 "issuer_name_hash and issuer_key_hash must be the same length "
173 "as the digest size of the algorithm"
174 )
176 return OCSPRequestBuilder(
177 self._request,
178 (issuer_name_hash, issuer_key_hash, serial_number, algorithm),
179 self._extensions,
180 )
182 def add_extension(
183 self, extval: x509.ExtensionType, critical: bool
184 ) -> OCSPRequestBuilder:
185 if not isinstance(extval, x509.ExtensionType):
186 raise TypeError("extension must be an ExtensionType")
188 extension = x509.Extension(extval.oid, critical, extval)
189 _reject_duplicate_extension(extension, self._extensions)
191 return OCSPRequestBuilder(
192 self._request, self._request_hash, [*self._extensions, extension]
193 )
195 def build(self) -> OCSPRequest:
196 if self._request is None and self._request_hash is None:
197 raise ValueError("You must add a certificate before building")
199 return ocsp.create_ocsp_request(self)
202class OCSPResponseBuilder:
203 def __init__(
204 self,
205 response: _SingleResponse | None = None,
206 responder_id: tuple[x509.Certificate, OCSPResponderEncoding]
207 | None = None,
208 certs: list[x509.Certificate] | None = None,
209 extensions: list[x509.Extension[x509.ExtensionType]] = [],
210 ):
211 self._response = response
212 self._responder_id = responder_id
213 self._certs = certs
214 self._extensions = extensions
216 def add_response(
217 self,
218 cert: x509.Certificate,
219 issuer: x509.Certificate,
220 algorithm: hashes.HashAlgorithm,
221 cert_status: OCSPCertStatus,
222 this_update: datetime.datetime,
223 next_update: datetime.datetime | None,
224 revocation_time: datetime.datetime | None,
225 revocation_reason: x509.ReasonFlags | None,
226 ) -> OCSPResponseBuilder:
227 if self._response is not None:
228 raise ValueError("Only one response per OCSPResponse.")
230 if not isinstance(cert, x509.Certificate) or not isinstance(
231 issuer, x509.Certificate
232 ):
233 raise TypeError("cert and issuer must be a Certificate")
235 singleresp = _SingleResponse(
236 (cert, issuer),
237 None,
238 algorithm,
239 cert_status,
240 this_update,
241 next_update,
242 revocation_time,
243 revocation_reason,
244 )
245 return OCSPResponseBuilder(
246 singleresp,
247 self._responder_id,
248 self._certs,
249 self._extensions,
250 )
252 def add_response_by_hash(
253 self,
254 issuer_name_hash: bytes,
255 issuer_key_hash: bytes,
256 serial_number: int,
257 algorithm: hashes.HashAlgorithm,
258 cert_status: OCSPCertStatus,
259 this_update: datetime.datetime,
260 next_update: datetime.datetime | None,
261 revocation_time: datetime.datetime | None,
262 revocation_reason: x509.ReasonFlags | None,
263 ) -> OCSPResponseBuilder:
264 if self._response is not None:
265 raise ValueError("Only one response per OCSPResponse.")
267 if not isinstance(serial_number, int):
268 raise TypeError("serial_number must be an integer")
270 utils._check_bytes("issuer_name_hash", issuer_name_hash)
271 utils._check_bytes("issuer_key_hash", issuer_key_hash)
272 _verify_algorithm(algorithm)
273 if algorithm.digest_size != len(
274 issuer_name_hash
275 ) or algorithm.digest_size != len(issuer_key_hash):
276 raise ValueError(
277 "issuer_name_hash and issuer_key_hash must be the same length "
278 "as the digest size of the algorithm"
279 )
281 singleresp = _SingleResponse(
282 None,
283 (issuer_name_hash, issuer_key_hash, serial_number),
284 algorithm,
285 cert_status,
286 this_update,
287 next_update,
288 revocation_time,
289 revocation_reason,
290 )
291 return OCSPResponseBuilder(
292 singleresp,
293 self._responder_id,
294 self._certs,
295 self._extensions,
296 )
298 def responder_id(
299 self, encoding: OCSPResponderEncoding, responder_cert: x509.Certificate
300 ) -> OCSPResponseBuilder:
301 if self._responder_id is not None:
302 raise ValueError("responder_id can only be set once")
303 if not isinstance(responder_cert, x509.Certificate):
304 raise TypeError("responder_cert must be a Certificate")
305 if not isinstance(encoding, OCSPResponderEncoding):
306 raise TypeError(
307 "encoding must be an element from OCSPResponderEncoding"
308 )
310 return OCSPResponseBuilder(
311 self._response,
312 (responder_cert, encoding),
313 self._certs,
314 self._extensions,
315 )
317 def certificates(
318 self, certs: Iterable[x509.Certificate]
319 ) -> OCSPResponseBuilder:
320 if self._certs is not None:
321 raise ValueError("certificates may only be set once")
322 certs = list(certs)
323 if len(certs) == 0:
324 raise ValueError("certs must not be an empty list")
325 if not all(isinstance(x, x509.Certificate) for x in certs):
326 raise TypeError("certs must be a list of Certificates")
327 return OCSPResponseBuilder(
328 self._response,
329 self._responder_id,
330 certs,
331 self._extensions,
332 )
334 def add_extension(
335 self, extval: x509.ExtensionType, critical: bool
336 ) -> OCSPResponseBuilder:
337 if not isinstance(extval, x509.ExtensionType):
338 raise TypeError("extension must be an ExtensionType")
340 extension = x509.Extension(extval.oid, critical, extval)
341 _reject_duplicate_extension(extension, self._extensions)
343 return OCSPResponseBuilder(
344 self._response,
345 self._responder_id,
346 self._certs,
347 [*self._extensions, extension],
348 )
350 def sign(
351 self,
352 private_key: CertificateIssuerPrivateKeyTypes,
353 algorithm: hashes.HashAlgorithm | None,
354 ) -> OCSPResponse:
355 if self._response is None:
356 raise ValueError("You must add a response before signing")
357 if self._responder_id is None:
358 raise ValueError("You must add a responder_id before signing")
360 return ocsp.create_ocsp_response(
361 OCSPResponseStatus.SUCCESSFUL, self, private_key, algorithm
362 )
364 @classmethod
365 def build_unsuccessful(
366 cls, response_status: OCSPResponseStatus
367 ) -> OCSPResponse:
368 if not isinstance(response_status, OCSPResponseStatus):
369 raise TypeError(
370 "response_status must be an item from OCSPResponseStatus"
371 )
372 if response_status is OCSPResponseStatus.SUCCESSFUL:
373 raise ValueError("response_status cannot be SUCCESSFUL")
375 return ocsp.create_ocsp_response(response_status, None, None, None)
378load_der_ocsp_request = ocsp.load_der_ocsp_request
379load_der_ocsp_response = ocsp.load_der_ocsp_response