Repository: apple/app-store-server-library-python

Issue:
Feature request: OCSP response caching
We see a lot of repeated network requests to the same URLs on ocsp.apple.com. This isn’t a _big_ issue, but it would be very helpful if this could be cached for a very short amount of time. 

## Details

Each decode+verify seems to require 2 requests (made serially). On our server processing a single HTTP request (purchase and/or notifications) can often require verifying/decoding multiple tokens.

The only reason this isn’t a huge issue for us is that ocsp.apple.com responds just absurdly quickly (usually <50ms). But if we’re verifying a pair of transaction + renewal info tokens, then the result is 50ms * 4 = 200ms delay added to our server’s response time. For processing a server notification, that could be 600ms.

I’m not a security engineer (so tell me if I’m wrong!), but it seems to me that an OCSP response for a given certificate could reasonably be cached for a short time. Even a 5s cache would cover processing an HTTP request, so processing a notification or purchase request would be reduced from 4-6 requests in serial to 2 requests (usually there’s the same 2 certs which are used).

## Example

Purchase example:
1. Create a verifier with enable_online_checks=True
1. Receive a new purchase from iOS app with transaction + renewal info
2. Verify the signed transaction info: sends 2 requests to ocsp.apple.com, processed serially one after the other
3. Verify the renewal info: sends 2 requests to ocsp.apple.com, processed serially one after the other

Notification Example:
1. Create a verifier with enable_online_checks=True
1. Receive a server notification
2. Verify the notification: sends 2 requests to ocsp.apple.com, processed serially one after the other
3. Verify the signed transaction info: sends 2 requests to ocsp.apple.com, processed serially one after the other
4. Verify the renewal info: sends 2 requests to ocsp.apple.com, processed serially one after the other

For this reason, we’ve opted to create 2 verifiers: `enable_online_checks=True` for the notification, then `enable_online_checks=False` for the signed tokens inside the notification. We’re reasoning that if the outer token has a non-revoked certificate, the inner ones are approximately vouched for.

<img width="517" alt="Image" src="https://github.com/user-attachments/assets/f4310759-c748-4044-a4f1-b26cb64791f9" />

## Possible implementation

I'm not that familiar with the codebase, but I think the issue is here:
https://github.com/apple/app-store-server-library-python/blob/1202058ddf4eac67c66f3947827e042e9ae6eee1/appstoreserverlibrary/signed_data_verifier.py#L213-L219

One could potentially just add a response cache on `_ChainVerifier` keyed on the `public_bytes` + `accesss_location`.

Discussion:
@alexanderjordanbaker: @WFT Yeah, we implemented a cache in the Java version of the library for these calls recently, but need to bring that to the rest of the languages to improve performance@michaelkariv: I am on python version. Would love to see this done sooner rather than later. This latensy is the reason I will stick to receipt verification for as long as I can@alexanderjordanbaker: @michaelkariv @WFT this has been adopted in 1.7.0 of the library, check it out and let me know how it works for you@WFT: @alexanderjordanbaker Pushed it out yesterday and we're already enjoying the benefits.

For example, decoding a notification (when the cache is full) can now be under 2ms rather than 100ms - 300ms. And even when the cache isn't full, we're still reducing the time by a fair amount by skipping extra work.

<img width="690" alt="Image" src="https://github.com/user-attachments/assets/c34fbabb-2c68-4d9b-9c1a-b7238449b682" />

Solution patch:
diff --git a/appstoreserverlibrary/signed_data_verifier.py b/appstoreserverlibrary/signed_data_verifier.py
index 56bda598..3e52b7ca 100644
--- a/appstoreserverlibrary/signed_data_verifier.py
+++ b/appstoreserverlibrary/signed_data_verifier.py
@@ -1,6 +1,6 @@
 # Copyright (c) 2023 Apple Inc. Licensed under MIT License.
 
-from typing import List, Optional
+from typing import List, Optional, Dict
 from base64 import b64decode
 from enum import IntEnum
 import time
@@ -155,11 +155,25 @@ def _decode_signed_object(self, signed_obj: str) -> dict:
             raise VerificationException(VerificationStatus.VERIFICATION_FAILURE) from e
 
 class _ChainVerifier:
+    MAXIMUM_CACHE_SIZE = 32 # There are unlikely to be more than a couple keys at once
+    CACHE_TIME_LIMIT = 15 * 60 # 15 minutes
+
     def __init__(self, root_certificates: List[bytes], enable_strict_checks=True):
         self.enable_strict_checks = enable_strict_checks
         self.root_certificates = root_certificates
+        self.verified_certificates_cache: Dict[tuple[str, ...], (str, int)] = {}
 
     def verify_chain(self, certificates: List[str], perform_online_checks: bool, effective_date: int) -> str:
+        if perform_online_checks and len(certificates) > 0:
+            cached_public_key = self.get_cached_public_key(certificates)
+            if cached_public_key is not None:
+                return cached_public_key
+        verified_public_key = self._verify_chain_without_caching(certificates=certificates, perform_online_checks=perform_online_checks, effective_date=effective_date)
+        if perform_online_checks:
+            self.put_verified_public_key(certificates, verified_public_key)
+        return verified_public_key
+    
+    def _verify_chain_without_caching(self, certificates: List[str], perform_online_checks: bool, effective_date: int) -> str:
         if len(self.root_certificates) == 0:
             raise VerificationException(VerificationStatus.INVALID_CERTIFICATE)
         if len(certificates) != 3:
@@ -295,7 +309,22 @@ def check_ocsp_status(self, cert: crypto.X509, issuer: crypto.X509, root: crypto
                             return
 
         raise VerificationException(VerificationStatus.VERIFICATION_FAILURE)
+    
+    def get_cached_public_key(self, certificates: List[str]) -> Optional[str]:
+        verified_public_key = self.verified_certificates_cache.get(tuple(certificates))
+        if verified_public_key is None:
+            return None
+        if verified_public_key[1] <= time.time():
+            return None
+        return verified_public_key[0]
 
+    def put_verified_public_key(self, certificates: List[str], verified_public_key: str):
+        cache_expiration = time.time() + _ChainVerifier.CACHE_TIME_LIMIT
+        self.verified_certificates_cache[tuple(certificates)] = (verified_public_key, cache_expiration)
+        if len(self.verified_certificates_cache) > _ChainVerifier.MAXIMUM_CACHE_SIZE:
+            for k, v in list(self.verified_certificates_cache.items()):
+                if v[1] <= time.time():
+                    del self.verified_certificates_cache[k]
 
 class VerificationStatus(IntEnum):
     OK = 0


Test patch:
diff --git a/tests/test_x509_verifiction.py b/tests/test_x509_verifiction.py
index 461e612f..98d1af99 100644
--- a/tests/test_x509_verifiction.py
+++ b/tests/test_x509_verifiction.py
@@ -1,6 +1,8 @@
 # Copyright (c) 2023 Apple Inc. Licensed under MIT License.
 
 import unittest
+from unittest import mock
+from unittest.mock import MagicMock, patch
 
 from appstoreserverlibrary.signed_data_verifier import _ChainVerifier, VerificationException, VerificationStatus
 from base64 import b64decode, b64encode
@@ -20,6 +22,7 @@
 REAL_APPLE_SIGNING_CERTIFICATE_BASE64_ENCODED = "MIIEMDCCA7agAwIBAgIQaPoPldvpSoEH0lBrjDPv9jAKBggqhkjOPQQDAzB1MUQwQgYDVQQDDDtBcHBsZSBXb3JsZHdpZGUgRGV2ZWxvcGVyIFJlbGF0aW9ucyBDZXJ0aWZpY2F0aW9uIEF1dGhvcml0eTELMAkGA1UECwwCRzYxEzARBgNVBAoMCkFwcGxlIEluYy4xCzAJBgNVBAYTAlVTMB4XDTIxMDgyNTAyNTAzNFoXDTIzMDkyNDAyNTAzM1owgZIxQDA+BgNVBAMMN1Byb2QgRUNDIE1hYyBBcHAgU3RvcmUgYW5kIGlUdW5lcyBTdG9yZSBSZWNlaXB0IFNpZ25pbmcxLDAqBgNVBAsMI0FwcGxlIFdvcmxkd2lkZSBEZXZlbG9wZXIgUmVsYXRpb25zMRMwEQYDVQQKDApBcHBsZSBJbmMuMQswCQYDVQQGEwJVUzBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABOoTcaPcpeipNL9eQ06tCu7pUcwdCXdN8vGqaUjd58Z8tLxiUC0dBeA+euMYggh1/5iAk+FMxUFmA2a1r4aCZ8SjggIIMIICBDAMBgNVHRMBAf8EAjAAMB8GA1UdIwQYMBaAFD8vlCNR01DJmig97bB85c+lkGKZMHAGCCsGAQUFBwEBBGQwYjAtBggrBgEFBQcwAoYhaHR0cDovL2NlcnRzLmFwcGxlLmNvbS93d2RyZzYuZGVyMDEGCCsGAQUFBzABhiVodHRwOi8vb2NzcC5hcHBsZS5jb20vb2NzcDAzLXd3ZHJnNjAyMIIBHgYDVR0gBIIBFTCCAREwggENBgoqhkiG92NkBQYBMIH+MIHDBggrBgEFBQcCAjCBtgyBs1JlbGlhbmNlIG9uIHRoaXMgY2VydGlmaWNhdGUgYnkgYW55IHBhcnR5IGFzc3VtZXMgYWNjZXB0YW5jZSBvZiB0aGUgdGhlbiBhcHBsaWNhYmxlIHN0YW5kYXJkIHRlcm1zIGFuZCBjb25kaXRpb25zIG9mIHVzZSwgY2VydGlmaWNhdGUgcG9saWN5IGFuZCBjZXJ0aWZpY2F0aW9uIHByYWN0aWNlIHN0YXRlbWVudHMuMDYGCCsGAQUFBwIBFipodHRwOi8vd3d3LmFwcGxlLmNvbS9jZXJ0aWZpY2F0ZWF1dGhvcml0eS8wHQYDVR0OBBYEFCOCmMBq//1L5imvVmqX1oCYeqrMMA4GA1UdDwEB/wQEAwIHgDAQBgoqhkiG92NkBgsBBAIFADAKBggqhkjOPQQDAwNoADBlAjEAl4JB9GJHixP2nuibyU1k3wri5psGIxPME05sFKq7hQuzvbeyBu82FozzxmbzpogoAjBLSFl0dZWIYl2ejPV+Di5fBnKPu8mymBQtoE/H2bES0qAs8bNueU3CBjjh1lwnDsI="
 
 EFFECTIVE_DATE = 1681312846
+CLOCK_DATE = 41231
 
 class X509Verification(unittest.TestCase):
     def test_valid_chain_without_ocsp(self):
@@ -115,6 +118,82 @@ def test_apple_chain_is_valid_with_ocsp_and_strict(self):
             REAL_APPLE_ROOT_BASE64_ENCODED
         ], True, EFFECTIVE_DATE)
 
+    def test_ocsp_response_caching(self):
+        verifier = _ChainVerifier([b64decode(REAL_APPLE_ROOT_BASE64_ENCODED)])
+        magic_mock = MagicMock(return_value=LEAF_CERT_BASE64_ENCODED)
+        verifier._verify_chain_without_caching = magic_mock
+        with patch('time.time', mock.MagicMock(return_value=CLOCK_DATE)):
+            verifier.verify_chain([
+                REAL_APPLE_SIGNING_CERTIFICATE_BASE64_ENCODED,
+                REAL_APPLE_INTERMEDIATE_BASE64_ENCODED,
+                REAL_APPLE_ROOT_BASE64_ENCODED
+            ], True, EFFECTIVE_DATE)
+        self.assertEqual(1, magic_mock.call_count)
+        with patch('time.time', mock.MagicMock(return_value=CLOCK_DATE + 1)): # 1 second
+            verifier.verify_chain([
+                REAL_APPLE_SIGNING_CERTIFICATE_BASE64_ENCODED,
+                REAL_APPLE_INTERMEDIATE_BASE64_ENCODED,
+                REAL_APPLE_ROOT_BASE64_ENCODED
+            ], True, EFFECTIVE_DATE)
+        self.assertEqual(1, magic_mock.call_count)
+
+    def test_ocsp_response_caching_has_expiration(self):
+        verifier = _ChainVerifier([b64decode(REAL_APPLE_ROOT_BASE64_ENCODED)])
+        magic_mock = MagicMock(return_value=LEAF_CERT_BASE64_ENCODED)
+        verifier._verify_chain_without_caching = magic_mock
+        with patch('time.time', mock.MagicMock(return_value=CLOCK_DATE)):
+            verifier.verify_chain([
+                REAL_APPLE_SIGNING_CERTIFICATE_BASE64_ENCODED,
+                REAL_APPLE_INTERMEDIATE_BASE64_ENCODED,
+                REAL_APPLE_ROOT_BASE64_ENCODED
+            ], True, EFFECTIVE_DATE)
+        self.assertEqual(1, magic_mock.call_count)
+        with patch('time.time', mock.MagicMock(return_value=CLOCK_DATE + 900)): # 15 minutes
+            verifier.verify_chain([
+                REAL_APPLE_SIGNING_CERTIFICATE_BASE64_ENCODED,
+                REAL_APPLE_INTERMEDIATE_BASE64_ENCODED,
+                REAL_APPLE_ROOT_BASE64_ENCODED
+            ], True, EFFECTIVE_DATE)
+        self.assertEqual(2, magic_mock.call_count)
+
+    def test_ocsp_response_caching_with_different_chain(self):
+        verifier = _ChainVerifier([b64decode(REAL_APPLE_ROOT_BASE64_ENCODED)])
+        magic_mock = MagicMock(return_value=LEAF_CERT_BASE64_ENCODED)
+        verifier._verify_chain_without_caching = magic_mock
+        with patch('time.time', mock.MagicMock(return_value=CLOCK_DATE)):
+            verifier.verify_chain([
+                LEAF_CERT_BASE64_ENCODED,
+                INTERMEDIATE_CA_BASE64_ENCODED,
+                ROOT_CA_BASE64_ENCODED
+            ], True, EFFECTIVE_DATE)
+        self.assertEqual(1, magic_mock.call_count)
+        with patch('time.time', mock.MagicMock(return_value=CLOCK_DATE)): # Same
+            verifier.verify_chain([
+                REAL_APPLE_SIGNING_CERTIFICATE_BASE64_ENCODED,
+                REAL_APPLE_INTERMEDIATE_BASE64_ENCODED,
+                REAL_APPLE_ROOT_BASE64_ENCODED
+            ], True, EFFECTIVE_DATE)
+        self.assertEqual(2, magic_mock.call_count)
+
+    def test_ocsp_response_caching_with_slightly_different_chain(self):
+        verifier = _ChainVerifier([b64decode(REAL_APPLE_ROOT_BASE64_ENCODED)])
+        magic_mock = MagicMock(return_value=LEAF_CERT_BASE64_ENCODED)
+        verifier._verify_chain_without_caching = magic_mock
+        with patch('time.time', mock.MagicMock(return_value=CLOCK_DATE)):
+            verifier.verify_chain([
+                LEAF_CERT_BASE64_ENCODED,
+                INTERMEDIATE_CA_BASE64_ENCODED,
+                ROOT_CA_BASE64_ENCODED
+            ], True, EFFECTIVE_DATE)
+        self.assertEqual(1, magic_mock.call_count)
+        with patch('time.time', mock.MagicMock(return_value=CLOCK_DATE)): # Same
+            verifier.verify_chain([
+                LEAF_CERT_BASE64_ENCODED,
+                INTERMEDIATE_CA_BASE64_ENCODED,
+                REAL_APPLE_ROOT_BASE64_ENCODED
+            ], True, EFFECTIVE_DATE)
+        self.assertEqual(2, magic_mock.call_count)
+
 
 if __name__ == '__main__':
     unittest.main()