[AAP-50642] memoize pem private key to avoid openssl slowdown (#791)

For 2.6, we've moved to RHEL9 for the base container, which has openssl
3.2 in it. Apparently , its is a known issue that the RSA key validation
is more secure and more slow in this version vs. Openssl 1 that RHEL8
uses. We were loading and validating the same jwt private key on every
request, at a cost of ~ 38 ms, vs previously this took ~ 3 ms. We
noticed this slowdown in the benchmarks and after deep dive/profiling we
found this was the difference and memoizing the validated private key
saves us from this broad slowdown. See
https://issues.redhat.com/browse/AAP-50642 for more details

[pyca/cryptography#7236
(comment)](pyca/cryptography#7236 (comment))
<--- report that details Openssl 3 slowdown

Author: chrismeyersfsu

ansible_base/jwt_consumer/common/util.py

@@ -1,6 +1,7 @@
 import logging
 import time
 from base64 import b64encode
+from functools import lru_cache
 
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import hashes, serialization
@@ -14,8 +15,16 @@
 _SHARED_SECRET = 'trusted_proxy'
 
 
+@lru_cache
+def _load_pem_private_key(key: str):
+    # Loading and validating the private key is more expensive in OpenSSL 3.2 (from RHEL9) than in Openssl 1.1 (from RHEL8)
+    # For that reason, we will memoize the result of this function, and only re-execute it if the key changes
+    # This is stored in memory local to the process
+    return serialization.load_pem_private_key(bytes(key, 'utf-8'), password=None)
+
+
 def generate_x_trusted_proxy_header(key: str) -> str:
-    private_key = serialization.load_pem_private_key(bytes(key, 'utf-8'), password=None)
+    private_key = _load_pem_private_key(key)
     timestamp = time.time_ns()
     message = f'{_SHARED_SECRET}-{timestamp}'
     signature = private_key.sign(bytes(message, 'utf-8'), padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH), hashes.SHA256())

test_app/tests/jwt_consumer/common/test_util.py

@@ -3,7 +3,7 @@
 
 from django.test.utils import override_settings
 
-from ansible_base.jwt_consumer.common.util import generate_x_trusted_proxy_header, validate_x_trusted_proxy_header
+from ansible_base.jwt_consumer.common.util import _load_pem_private_key, generate_x_trusted_proxy_header, validate_x_trusted_proxy_header
 
 
 class TestValidateTrustedProxy:
@@ -71,3 +71,26 @@ def test_validate_x_trusted_proxy_header_invalid_signature(self, random_public_k
                 # 0 is invalid bytes
                 timestamp, junk = header.split('-')
                 assert validate_x_trusted_proxy_header(f"{timestamp}-0") is False
+
+    def test_generate_x_trusted_proxy_header(self, rsa_keypair, rsa_keypair_factory):
+        """
+        This test ensures that, for the same key, the function is called only once.
+        Otherwise, the function is not called and the return value is returned from the cache.
+        """
+        _load_pem_private_key.cache_clear()
+        new_rsa_keypair = rsa_keypair_factory()
+
+        # Create a mock private key that has the sign method
+        mock_private_key = mock.Mock()
+        mock_private_key.sign.return_value = b'fake_signature'
+
+        for keypair in [rsa_keypair, new_rsa_keypair]:
+            with mock.patch("cryptography.hazmat.primitives.serialization.load_pem_private_key", return_value=mock_private_key) as mock_load_pem:
+                # Call the function multiple times
+                generate_x_trusted_proxy_header(keypair.private)
+                generate_x_trusted_proxy_header(keypair.private)
+                generate_x_trusted_proxy_header(keypair.private)
+
+                # Verify the function is called only once due to caching
+                assert mock_load_pem.call_count == 1
+                mock_load_pem.assert_called_with(bytes(keypair.private, 'utf-8'), password=None)