ocsp.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678
  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.
  4. from __future__ import annotations
  5. import abc
  6. import datetime
  7. import typing
  8. from cryptography import utils, x509
  9. from cryptography.hazmat.bindings._rust import ocsp
  10. from cryptography.hazmat.primitives import hashes, serialization
  11. from cryptography.hazmat.primitives.asymmetric.types import (
  12. CertificateIssuerPrivateKeyTypes,
  13. )
  14. from cryptography.x509.base import (
  15. _EARLIEST_UTC_TIME,
  16. _convert_to_naive_utc_time,
  17. _reject_duplicate_extension,
  18. )
  19. class OCSPResponderEncoding(utils.Enum):
  20. HASH = "By Hash"
  21. NAME = "By Name"
  22. class OCSPResponseStatus(utils.Enum):
  23. SUCCESSFUL = 0
  24. MALFORMED_REQUEST = 1
  25. INTERNAL_ERROR = 2
  26. TRY_LATER = 3
  27. SIG_REQUIRED = 5
  28. UNAUTHORIZED = 6
  29. _ALLOWED_HASHES = (
  30. hashes.SHA1,
  31. hashes.SHA224,
  32. hashes.SHA256,
  33. hashes.SHA384,
  34. hashes.SHA512,
  35. )
  36. def _verify_algorithm(algorithm: hashes.HashAlgorithm) -> None:
  37. if not isinstance(algorithm, _ALLOWED_HASHES):
  38. raise ValueError(
  39. "Algorithm must be SHA1, SHA224, SHA256, SHA384, or SHA512"
  40. )
  41. class OCSPCertStatus(utils.Enum):
  42. GOOD = 0
  43. REVOKED = 1
  44. UNKNOWN = 2
  45. class _SingleResponse:
  46. def __init__(
  47. self,
  48. cert: x509.Certificate,
  49. issuer: x509.Certificate,
  50. algorithm: hashes.HashAlgorithm,
  51. cert_status: OCSPCertStatus,
  52. this_update: datetime.datetime,
  53. next_update: datetime.datetime | None,
  54. revocation_time: datetime.datetime | None,
  55. revocation_reason: x509.ReasonFlags | None,
  56. ):
  57. if not isinstance(cert, x509.Certificate) or not isinstance(
  58. issuer, x509.Certificate
  59. ):
  60. raise TypeError("cert and issuer must be a Certificate")
  61. _verify_algorithm(algorithm)
  62. if not isinstance(this_update, datetime.datetime):
  63. raise TypeError("this_update must be a datetime object")
  64. if next_update is not None and not isinstance(
  65. next_update, datetime.datetime
  66. ):
  67. raise TypeError("next_update must be a datetime object or None")
  68. self._cert = cert
  69. self._issuer = issuer
  70. self._algorithm = algorithm
  71. self._this_update = this_update
  72. self._next_update = next_update
  73. if not isinstance(cert_status, OCSPCertStatus):
  74. raise TypeError(
  75. "cert_status must be an item from the OCSPCertStatus enum"
  76. )
  77. if cert_status is not OCSPCertStatus.REVOKED:
  78. if revocation_time is not None:
  79. raise ValueError(
  80. "revocation_time can only be provided if the certificate "
  81. "is revoked"
  82. )
  83. if revocation_reason is not None:
  84. raise ValueError(
  85. "revocation_reason can only be provided if the certificate"
  86. " is revoked"
  87. )
  88. else:
  89. if not isinstance(revocation_time, datetime.datetime):
  90. raise TypeError("revocation_time must be a datetime object")
  91. revocation_time = _convert_to_naive_utc_time(revocation_time)
  92. if revocation_time < _EARLIEST_UTC_TIME:
  93. raise ValueError(
  94. "The revocation_time must be on or after"
  95. " 1950 January 1."
  96. )
  97. if revocation_reason is not None and not isinstance(
  98. revocation_reason, x509.ReasonFlags
  99. ):
  100. raise TypeError(
  101. "revocation_reason must be an item from the ReasonFlags "
  102. "enum or None"
  103. )
  104. self._cert_status = cert_status
  105. self._revocation_time = revocation_time
  106. self._revocation_reason = revocation_reason
  107. class OCSPRequest(metaclass=abc.ABCMeta):
  108. @property
  109. @abc.abstractmethod
  110. def issuer_key_hash(self) -> bytes:
  111. """
  112. The hash of the issuer public key
  113. """
  114. @property
  115. @abc.abstractmethod
  116. def issuer_name_hash(self) -> bytes:
  117. """
  118. The hash of the issuer name
  119. """
  120. @property
  121. @abc.abstractmethod
  122. def hash_algorithm(self) -> hashes.HashAlgorithm:
  123. """
  124. The hash algorithm used in the issuer name and key hashes
  125. """
  126. @property
  127. @abc.abstractmethod
  128. def serial_number(self) -> int:
  129. """
  130. The serial number of the cert whose status is being checked
  131. """
  132. @abc.abstractmethod
  133. def public_bytes(self, encoding: serialization.Encoding) -> bytes:
  134. """
  135. Serializes the request to DER
  136. """
  137. @property
  138. @abc.abstractmethod
  139. def extensions(self) -> x509.Extensions:
  140. """
  141. The list of request extensions. Not single request extensions.
  142. """
  143. class OCSPSingleResponse(metaclass=abc.ABCMeta):
  144. @property
  145. @abc.abstractmethod
  146. def certificate_status(self) -> OCSPCertStatus:
  147. """
  148. The status of the certificate (an element from the OCSPCertStatus enum)
  149. """
  150. @property
  151. @abc.abstractmethod
  152. def revocation_time(self) -> datetime.datetime | None:
  153. """
  154. The date of when the certificate was revoked or None if not
  155. revoked.
  156. """
  157. @property
  158. @abc.abstractmethod
  159. def revocation_time_utc(self) -> datetime.datetime | None:
  160. """
  161. The date of when the certificate was revoked or None if not
  162. revoked. Represented as a non-naive UTC datetime.
  163. """
  164. @property
  165. @abc.abstractmethod
  166. def revocation_reason(self) -> x509.ReasonFlags | None:
  167. """
  168. The reason the certificate was revoked or None if not specified or
  169. not revoked.
  170. """
  171. @property
  172. @abc.abstractmethod
  173. def this_update(self) -> datetime.datetime:
  174. """
  175. The most recent time at which the status being indicated is known by
  176. the responder to have been correct
  177. """
  178. @property
  179. @abc.abstractmethod
  180. def this_update_utc(self) -> datetime.datetime:
  181. """
  182. The most recent time at which the status being indicated is known by
  183. the responder to have been correct. Represented as a non-naive UTC
  184. datetime.
  185. """
  186. @property
  187. @abc.abstractmethod
  188. def next_update(self) -> datetime.datetime | None:
  189. """
  190. The time when newer information will be available
  191. """
  192. @property
  193. @abc.abstractmethod
  194. def next_update_utc(self) -> datetime.datetime | None:
  195. """
  196. The time when newer information will be available. Represented as a
  197. non-naive UTC datetime.
  198. """
  199. @property
  200. @abc.abstractmethod
  201. def issuer_key_hash(self) -> bytes:
  202. """
  203. The hash of the issuer public key
  204. """
  205. @property
  206. @abc.abstractmethod
  207. def issuer_name_hash(self) -> bytes:
  208. """
  209. The hash of the issuer name
  210. """
  211. @property
  212. @abc.abstractmethod
  213. def hash_algorithm(self) -> hashes.HashAlgorithm:
  214. """
  215. The hash algorithm used in the issuer name and key hashes
  216. """
  217. @property
  218. @abc.abstractmethod
  219. def serial_number(self) -> int:
  220. """
  221. The serial number of the cert whose status is being checked
  222. """
  223. class OCSPResponse(metaclass=abc.ABCMeta):
  224. @property
  225. @abc.abstractmethod
  226. def responses(self) -> typing.Iterator[OCSPSingleResponse]:
  227. """
  228. An iterator over the individual SINGLERESP structures in the
  229. response
  230. """
  231. @property
  232. @abc.abstractmethod
  233. def response_status(self) -> OCSPResponseStatus:
  234. """
  235. The status of the response. This is a value from the OCSPResponseStatus
  236. enumeration
  237. """
  238. @property
  239. @abc.abstractmethod
  240. def signature_algorithm_oid(self) -> x509.ObjectIdentifier:
  241. """
  242. The ObjectIdentifier of the signature algorithm
  243. """
  244. @property
  245. @abc.abstractmethod
  246. def signature_hash_algorithm(
  247. self,
  248. ) -> hashes.HashAlgorithm | None:
  249. """
  250. Returns a HashAlgorithm corresponding to the type of the digest signed
  251. """
  252. @property
  253. @abc.abstractmethod
  254. def signature(self) -> bytes:
  255. """
  256. The signature bytes
  257. """
  258. @property
  259. @abc.abstractmethod
  260. def tbs_response_bytes(self) -> bytes:
  261. """
  262. The tbsResponseData bytes
  263. """
  264. @property
  265. @abc.abstractmethod
  266. def certificates(self) -> list[x509.Certificate]:
  267. """
  268. A list of certificates used to help build a chain to verify the OCSP
  269. response. This situation occurs when the OCSP responder uses a delegate
  270. certificate.
  271. """
  272. @property
  273. @abc.abstractmethod
  274. def responder_key_hash(self) -> bytes | None:
  275. """
  276. The responder's key hash or None
  277. """
  278. @property
  279. @abc.abstractmethod
  280. def responder_name(self) -> x509.Name | None:
  281. """
  282. The responder's Name or None
  283. """
  284. @property
  285. @abc.abstractmethod
  286. def produced_at(self) -> datetime.datetime:
  287. """
  288. The time the response was produced
  289. """
  290. @property
  291. @abc.abstractmethod
  292. def produced_at_utc(self) -> datetime.datetime:
  293. """
  294. The time the response was produced. Represented as a non-naive UTC
  295. datetime.
  296. """
  297. @property
  298. @abc.abstractmethod
  299. def certificate_status(self) -> OCSPCertStatus:
  300. """
  301. The status of the certificate (an element from the OCSPCertStatus enum)
  302. """
  303. @property
  304. @abc.abstractmethod
  305. def revocation_time(self) -> datetime.datetime | None:
  306. """
  307. The date of when the certificate was revoked or None if not
  308. revoked.
  309. """
  310. @property
  311. @abc.abstractmethod
  312. def revocation_time_utc(self) -> datetime.datetime | None:
  313. """
  314. The date of when the certificate was revoked or None if not
  315. revoked. Represented as a non-naive UTC datetime.
  316. """
  317. @property
  318. @abc.abstractmethod
  319. def revocation_reason(self) -> x509.ReasonFlags | None:
  320. """
  321. The reason the certificate was revoked or None if not specified or
  322. not revoked.
  323. """
  324. @property
  325. @abc.abstractmethod
  326. def this_update(self) -> datetime.datetime:
  327. """
  328. The most recent time at which the status being indicated is known by
  329. the responder to have been correct
  330. """
  331. @property
  332. @abc.abstractmethod
  333. def this_update_utc(self) -> datetime.datetime:
  334. """
  335. The most recent time at which the status being indicated is known by
  336. the responder to have been correct. Represented as a non-naive UTC
  337. datetime.
  338. """
  339. @property
  340. @abc.abstractmethod
  341. def next_update(self) -> datetime.datetime | None:
  342. """
  343. The time when newer information will be available
  344. """
  345. @property
  346. @abc.abstractmethod
  347. def next_update_utc(self) -> datetime.datetime | None:
  348. """
  349. The time when newer information will be available. Represented as a
  350. non-naive UTC datetime.
  351. """
  352. @property
  353. @abc.abstractmethod
  354. def issuer_key_hash(self) -> bytes:
  355. """
  356. The hash of the issuer public key
  357. """
  358. @property
  359. @abc.abstractmethod
  360. def issuer_name_hash(self) -> bytes:
  361. """
  362. The hash of the issuer name
  363. """
  364. @property
  365. @abc.abstractmethod
  366. def hash_algorithm(self) -> hashes.HashAlgorithm:
  367. """
  368. The hash algorithm used in the issuer name and key hashes
  369. """
  370. @property
  371. @abc.abstractmethod
  372. def serial_number(self) -> int:
  373. """
  374. The serial number of the cert whose status is being checked
  375. """
  376. @property
  377. @abc.abstractmethod
  378. def extensions(self) -> x509.Extensions:
  379. """
  380. The list of response extensions. Not single response extensions.
  381. """
  382. @property
  383. @abc.abstractmethod
  384. def single_extensions(self) -> x509.Extensions:
  385. """
  386. The list of single response extensions. Not response extensions.
  387. """
  388. @abc.abstractmethod
  389. def public_bytes(self, encoding: serialization.Encoding) -> bytes:
  390. """
  391. Serializes the response to DER
  392. """
  393. OCSPRequest.register(ocsp.OCSPRequest)
  394. OCSPResponse.register(ocsp.OCSPResponse)
  395. OCSPSingleResponse.register(ocsp.OCSPSingleResponse)
  396. class OCSPRequestBuilder:
  397. def __init__(
  398. self,
  399. request: tuple[
  400. x509.Certificate, x509.Certificate, hashes.HashAlgorithm
  401. ]
  402. | None = None,
  403. request_hash: tuple[bytes, bytes, int, hashes.HashAlgorithm]
  404. | None = None,
  405. extensions: list[x509.Extension[x509.ExtensionType]] = [],
  406. ) -> None:
  407. self._request = request
  408. self._request_hash = request_hash
  409. self._extensions = extensions
  410. def add_certificate(
  411. self,
  412. cert: x509.Certificate,
  413. issuer: x509.Certificate,
  414. algorithm: hashes.HashAlgorithm,
  415. ) -> OCSPRequestBuilder:
  416. if self._request is not None or self._request_hash is not None:
  417. raise ValueError("Only one certificate can be added to a request")
  418. _verify_algorithm(algorithm)
  419. if not isinstance(cert, x509.Certificate) or not isinstance(
  420. issuer, x509.Certificate
  421. ):
  422. raise TypeError("cert and issuer must be a Certificate")
  423. return OCSPRequestBuilder(
  424. (cert, issuer, algorithm), self._request_hash, self._extensions
  425. )
  426. def add_certificate_by_hash(
  427. self,
  428. issuer_name_hash: bytes,
  429. issuer_key_hash: bytes,
  430. serial_number: int,
  431. algorithm: hashes.HashAlgorithm,
  432. ) -> OCSPRequestBuilder:
  433. if self._request is not None or self._request_hash is not None:
  434. raise ValueError("Only one certificate can be added to a request")
  435. if not isinstance(serial_number, int):
  436. raise TypeError("serial_number must be an integer")
  437. _verify_algorithm(algorithm)
  438. utils._check_bytes("issuer_name_hash", issuer_name_hash)
  439. utils._check_bytes("issuer_key_hash", issuer_key_hash)
  440. if algorithm.digest_size != len(
  441. issuer_name_hash
  442. ) or algorithm.digest_size != len(issuer_key_hash):
  443. raise ValueError(
  444. "issuer_name_hash and issuer_key_hash must be the same length "
  445. "as the digest size of the algorithm"
  446. )
  447. return OCSPRequestBuilder(
  448. self._request,
  449. (issuer_name_hash, issuer_key_hash, serial_number, algorithm),
  450. self._extensions,
  451. )
  452. def add_extension(
  453. self, extval: x509.ExtensionType, critical: bool
  454. ) -> OCSPRequestBuilder:
  455. if not isinstance(extval, x509.ExtensionType):
  456. raise TypeError("extension must be an ExtensionType")
  457. extension = x509.Extension(extval.oid, critical, extval)
  458. _reject_duplicate_extension(extension, self._extensions)
  459. return OCSPRequestBuilder(
  460. self._request, self._request_hash, [*self._extensions, extension]
  461. )
  462. def build(self) -> OCSPRequest:
  463. if self._request is None and self._request_hash is None:
  464. raise ValueError("You must add a certificate before building")
  465. return ocsp.create_ocsp_request(self)
  466. class OCSPResponseBuilder:
  467. def __init__(
  468. self,
  469. response: _SingleResponse | None = None,
  470. responder_id: tuple[x509.Certificate, OCSPResponderEncoding]
  471. | None = None,
  472. certs: list[x509.Certificate] | None = None,
  473. extensions: list[x509.Extension[x509.ExtensionType]] = [],
  474. ):
  475. self._response = response
  476. self._responder_id = responder_id
  477. self._certs = certs
  478. self._extensions = extensions
  479. def add_response(
  480. self,
  481. cert: x509.Certificate,
  482. issuer: x509.Certificate,
  483. algorithm: hashes.HashAlgorithm,
  484. cert_status: OCSPCertStatus,
  485. this_update: datetime.datetime,
  486. next_update: datetime.datetime | None,
  487. revocation_time: datetime.datetime | None,
  488. revocation_reason: x509.ReasonFlags | None,
  489. ) -> OCSPResponseBuilder:
  490. if self._response is not None:
  491. raise ValueError("Only one response per OCSPResponse.")
  492. singleresp = _SingleResponse(
  493. cert,
  494. issuer,
  495. algorithm,
  496. cert_status,
  497. this_update,
  498. next_update,
  499. revocation_time,
  500. revocation_reason,
  501. )
  502. return OCSPResponseBuilder(
  503. singleresp,
  504. self._responder_id,
  505. self._certs,
  506. self._extensions,
  507. )
  508. def responder_id(
  509. self, encoding: OCSPResponderEncoding, responder_cert: x509.Certificate
  510. ) -> OCSPResponseBuilder:
  511. if self._responder_id is not None:
  512. raise ValueError("responder_id can only be set once")
  513. if not isinstance(responder_cert, x509.Certificate):
  514. raise TypeError("responder_cert must be a Certificate")
  515. if not isinstance(encoding, OCSPResponderEncoding):
  516. raise TypeError(
  517. "encoding must be an element from OCSPResponderEncoding"
  518. )
  519. return OCSPResponseBuilder(
  520. self._response,
  521. (responder_cert, encoding),
  522. self._certs,
  523. self._extensions,
  524. )
  525. def certificates(
  526. self, certs: typing.Iterable[x509.Certificate]
  527. ) -> OCSPResponseBuilder:
  528. if self._certs is not None:
  529. raise ValueError("certificates may only be set once")
  530. certs = list(certs)
  531. if len(certs) == 0:
  532. raise ValueError("certs must not be an empty list")
  533. if not all(isinstance(x, x509.Certificate) for x in certs):
  534. raise TypeError("certs must be a list of Certificates")
  535. return OCSPResponseBuilder(
  536. self._response,
  537. self._responder_id,
  538. certs,
  539. self._extensions,
  540. )
  541. def add_extension(
  542. self, extval: x509.ExtensionType, critical: bool
  543. ) -> OCSPResponseBuilder:
  544. if not isinstance(extval, x509.ExtensionType):
  545. raise TypeError("extension must be an ExtensionType")
  546. extension = x509.Extension(extval.oid, critical, extval)
  547. _reject_duplicate_extension(extension, self._extensions)
  548. return OCSPResponseBuilder(
  549. self._response,
  550. self._responder_id,
  551. self._certs,
  552. [*self._extensions, extension],
  553. )
  554. def sign(
  555. self,
  556. private_key: CertificateIssuerPrivateKeyTypes,
  557. algorithm: hashes.HashAlgorithm | None,
  558. ) -> OCSPResponse:
  559. if self._response is None:
  560. raise ValueError("You must add a response before signing")
  561. if self._responder_id is None:
  562. raise ValueError("You must add a responder_id before signing")
  563. return ocsp.create_ocsp_response(
  564. OCSPResponseStatus.SUCCESSFUL, self, private_key, algorithm
  565. )
  566. @classmethod
  567. def build_unsuccessful(
  568. cls, response_status: OCSPResponseStatus
  569. ) -> OCSPResponse:
  570. if not isinstance(response_status, OCSPResponseStatus):
  571. raise TypeError(
  572. "response_status must be an item from OCSPResponseStatus"
  573. )
  574. if response_status is OCSPResponseStatus.SUCCESSFUL:
  575. raise ValueError("response_status cannot be SUCCESSFUL")
  576. return ocsp.create_ocsp_response(response_status, None, None, None)
  577. load_der_ocsp_request = ocsp.load_der_ocsp_request
  578. load_der_ocsp_response = ocsp.load_der_ocsp_response