Add/update a cache entry. 'key' is of type cryptography.x509.ocsp.OCSPRequest 'value' is of type cryptography.x509.ocsp.OCSPResponse Validity of the OCSP response must be checked by caller.
(self, key: OCSPRequest, value: OCSPResponse)
| 65 | ) |
| 66 | |
| 67 | def __setitem__(self, key: OCSPRequest, value: OCSPResponse) -> None: |
| 68 | """Add/update a cache entry. |
| 69 | |
| 70 | 'key' is of type cryptography.x509.ocsp.OCSPRequest |
| 71 | 'value' is of type cryptography.x509.ocsp.OCSPResponse |
| 72 | |
| 73 | Validity of the OCSP response must be checked by caller. |
| 74 | """ |
| 75 | with self._lock: |
| 76 | cache_key = self._get_cache_key(key) |
| 77 | |
| 78 | # As per the OCSP protocol, if the response's nextUpdate field is |
| 79 | # not set, the responder is indicating that newer revocation |
| 80 | # information is available all the time. |
| 81 | next_update = _next_update(value) |
| 82 | if next_update is None: |
| 83 | self._data.pop(cache_key, None) |
| 84 | return |
| 85 | |
| 86 | this_update = _this_update(value) |
| 87 | if this_update is None: |
| 88 | return |
| 89 | now = _datetime.now(tz=timezone.utc) |
| 90 | if this_update.tzinfo is None: |
| 91 | # Make naive to match cryptography. |
| 92 | now = now.replace(tzinfo=None) |
| 93 | # Do nothing if the response is invalid. |
| 94 | if not (this_update <= now < next_update): |
| 95 | return |
| 96 | |
| 97 | # Cache new response OR update cached response if new response |
| 98 | # has longer validity. |
| 99 | cached_value = self._data.get(cache_key, None) |
| 100 | if cached_value is None: |
| 101 | self._data[cache_key] = value |
| 102 | return |
| 103 | cached_next_update = _next_update(cached_value) |
| 104 | if cached_next_update is not None and cached_next_update < next_update: |
| 105 | self._data[cache_key] = value |
| 106 | |
| 107 | def __getitem__(self, item: OCSPRequest) -> OCSPResponse: |
| 108 | """Get a cache entry if it exists. |
nothing calls this directly
no test coverage detected