Only allow x-trusted-proxy header to be valid for so long (ansible#574)

Author: john-westcott-iv

ansible_base/jwt_consumer/common/util.py

@@ -1,11 +1,13 @@
 import logging
 import time
+from base64 import b64encode
 
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import padding
 
 from ansible_base.jwt_consumer.common.cert import JWTCert, JWTCertException
+from ansible_base.lib.utils.settings import get_setting
 
 logger = logging.getLogger('ansible_base.jwt_consumer.common.util')
 
@@ -42,6 +44,15 @@ def validate_x_trusted_proxy_header(header_value: str, ignore_cache=False) -> bo
         logger.warning("Failed to validate x-trusted-proxy-header, malformed, expected value to contain a -")
         return False
 
+    # Validate that the header has been cut within the last 300ms (by default)
+    try:
+        if time.time_ns() - int(timestamp) > get_setting('trusted_header_timeout_in_ns', 300000000):
+            logger.warning(f"Timestamp {timestamp} was too old to be valid alter trusted_header_timeout_in_ns if needed")
+            return False
+    except ValueError:
+        logger.warning(f"Unable to convert timestamp (base64) {b64encode(timestamp.encode('UTF-8'))} into an integer")
+        return False
+
     try:
         signature_bytes = bytes.fromhex(signature)
     except ValueError:

test_app/tests/jwt_consumer/common/test_util.py

@@ -1,3 +1,4 @@
+import time
 from unittest import mock
 
 from django.test.utils import override_settings
@@ -30,14 +31,43 @@ def test_validate_trusted_proxy_header_bad_public_key(self, random_public_key):
         with override_settings(ANSIBLE_BASE_JWT_KEY=random_public_key):
             assert not validate_x_trusted_proxy_header("0-12345123451234512345")
 
-    def test_validate_x_trusted_proxy_header_invalid_signature(self, random_public_key, expected_log):
+    def test_header_timeout(self, expected_log, rsa_keypair):
+        header = generate_x_trusted_proxy_header(rsa_keypair.private)
+        with override_settings(ANSIBLE_BASE_JWT_KEY=rsa_keypair.public):
+            # Assert this header is valid if used right away
+            assert validate_x_trusted_proxy_header(header) is True
+
+            # By default the header is only valid for 300ms so a 1/2 second sleep will expire it
+            time.sleep(0.5)
+            with expected_log(
+                'ansible_base.jwt_consumer.common.util.logger',
+                'warning',
+                'was too old to be valid alter trusted_header_timeout_in_ns if needed',
+            ):
+                assert validate_x_trusted_proxy_header(header) is False
+
+    def test_invalid_header_timestamp(self, expected_log, rsa_keypair):
+        header = generate_x_trusted_proxy_header(rsa_keypair.private)
+        _, signed_part = header.split('-')
+        header = f'asdf-{signed_part}'
+        with override_settings(ANSIBLE_BASE_JWT_KEY=rsa_keypair.public):
+            with expected_log(
+                'ansible_base.jwt_consumer.common.util.logger',
+                'warning',
+                'Unable to convert timestamp (base64)',
+            ):
+                assert validate_x_trusted_proxy_header(header) is False
+
+    def test_validate_x_trusted_proxy_header_invalid_signature(self, random_public_key, expected_log, rsa_keypair):
         with override_settings(ANSIBLE_BASE_JWT_KEY=random_public_key):
-            # Idealy we would mock match bytes.fromhex but I couldn't get that to work
+            # Ideally we would mock match bytes.fromhex but I couldn't get that to work
             # with mock.patch('ansible_base.jwt_consumer.common.util.validate_x_trusted_proxy_header.bytes.fromhex', side_effect=ValueError()):
+            header = generate_x_trusted_proxy_header(rsa_keypair.private)
             with expected_log(
                 'ansible_base.jwt_consumer.common.util.logger',
                 'warning',
                 'Failed to validate x-trusted-proxy-header, malformed, expected signature to well-formed base64',
             ):
                 # 0 is invalid bytes
-                assert validate_x_trusted_proxy_header("0-0") is False
+                timestamp, junk = header.split('-')
+                assert validate_x_trusted_proxy_header(f"{timestamp}-0") is False