feat: add is_cache_disabled to CFE Identity Provider

.pre-commit-config.yaml

@@ -10,7 +10,7 @@ repos:
       - id: check-added-large-files
 
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.19.1
+    rev: v3.21.2
     hooks:
       - id: pyupgrade
         args: [--py37-plus]
@@ -59,7 +59,7 @@ repos:
           - eslint-config-standard
           - eslint-plugin-import
           - eslint-plugin-jsdoc
-          - eslint-plugin-n
+          - eslint-plugin-n@16.6.2
           - eslint-plugin-prefer-arrow
           - eslint-plugin-promise
           - '@typescript-eslint/eslint-plugin'

castlecraft/auth.py

@@ -37,7 +37,9 @@ def validate_bearer_with_introspection(token, idp):
     is_valid = False
     email = None
 
-    cached_token = get_cached_bearer_token(token)
+    cached_token = None
+    if not idp.get("is_cache_disabled"):
+        cached_token = get_cached_bearer_token(token)
     now = datetime.datetime.now()
     form_dict = frappe.local.form_dict
     token_response = {}
@@ -111,23 +113,25 @@ def validate_bearer_with_introspection(token, idp):
 
                 if user_exists:
                     email = email_from_token
-                    cache_bearer_token(token, token_response, exp, now)
-                    cache_user_from_sub(
-                        user_data.get("sub"),
-                        json.dumps({"email": email, "token": token}),
-                    )
+                    if not idp.get("is_cache_disabled"):
+                        cache_bearer_token(token, token_response, exp, now)
+                        cache_user_from_sub(
+                            user_data.get("sub"),
+                            json.dumps({"email": email, "token": token}),
+                        )
                     is_valid = True
                 # If user doesn't exist (or no email in token),
                 # check if we can create one.
                 elif idp.create_user:
                     # User does not exist, create them using the final user_data
                     user = create_and_save_user(user_data, idp)
                     email = user.email
-                    cache_bearer_token(token, token_response, exp, now)
-                    cache_user_from_sub(
-                        token_response.get("sub"),
-                        json.dumps({"email": email, "token": token}),
-                    )
+                    if not idp.get("is_cache_disabled"):
+                        cache_bearer_token(token, token_response, exp, now)
+                        cache_user_from_sub(
+                            token_response.get("sub"),
+                            json.dumps({"email": email, "token": token}),
+                        )
                     is_valid = True
 
         if is_valid:
@@ -150,7 +154,9 @@ def validate_bearer_with_jwt_verification(token, idp):
         payload = None
 
         # 1. Check for a cached, validated payload using the token itself as the key.
-        cached_payload_str = frappe.cache().get_value(f"cc_jwt_payload|{token}")
+        cached_payload_str = None
+        if not idp.get("is_cache_disabled"):
+            cached_payload_str = frappe.cache().get_value(f"cc_jwt_payload|{token}")
         now = datetime.datetime.now()
 
         if cached_payload_str:
@@ -200,7 +206,7 @@ def validate_bearer_with_jwt_verification(token, idp):
             frappe.local.form_dict = form_dict
 
             # 4. Cache the newly validated payload and other user details.
-            if payload.get("exp"):
+            if payload.get("exp") and not idp.get("is_cache_disabled"):
                 # Cache the payload against the token for the "fast path".
                 frappe.cache().set_value(
                     f"cc_jwt_payload|{token}",
@@ -211,7 +217,7 @@ def validate_bearer_with_jwt_verification(token, idp):
                     - now,
                 )
 
-            if payload.get("sub"):
+            if payload.get("sub") and not idp.get("is_cache_disabled"):
                 cache_user_from_sub(
                     payload.get("sub"),
                     json.dumps({"email": final_email, "token": token}),

castlecraft/castlecraft/doctype/cfe_identity_provider/cfe_identity_provider.json

@@ -9,6 +9,7 @@
     "identity_provider_config_section",
     "idp_name",
     "enabled",
+    "is_cache_disabled",
     "authorization_type",
     "cb_idp_0",
     "first_name_key",
@@ -47,6 +48,12 @@
       "label": "Enabled",
       "search_index": 1
     },
+    {
+      "default": "0",
+      "fieldname": "is_cache_disabled",
+      "fieldtype": "Check",
+      "label": "Is Cache Disabled"
+    },
     {
       "default": "Introspection",
       "fieldname": "authorization_type",

tests/test_auth_cache_disabling.py

@@ -0,0 +1,143 @@
+import datetime
+import json
+import unittest
+from unittest.mock import MagicMock, patch
+
+from castlecraft import auth
+from tests.conftest import MockDoc
+
+
+class TestAuthCacheDisabling(unittest.TestCase):
+    def setUp(self):
+        self.test_user_email = "test@example.com"
+        self.access_token = "valid-token-123"
+
+    @patch("castlecraft.auth.get_idp")
+    @patch("castlecraft.auth.requests")
+    def test_introspection_cache_disabled(self, mock_requests, mock_get_idp):
+        """
+        Verify that introspection bypasses cache and doesn't store in cache
+        when is_cache_disabled is set.
+        """
+        # --- Arrange ---
+        mock_idp = MockDoc(
+            dict(
+                doctype="CFE Identity Provider",
+                idp_name="test-idp",
+                enabled=1,
+                is_cache_disabled=1,  # CACHE DISABLED
+                authorization_type="Introspection",
+                email_key="email",
+                introspection_endpoint="https://idp.example.com/introspect",
+                token_key="token",
+                auth_header_enabled=1,
+                client_id="test-client",
+                client_secret="test-secret",
+                create_user=False,
+                fetch_user_info=False,
+                user_roles=[],
+            )
+        )
+        auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}"
+        mock_get_idp.return_value = mock_idp
+        auth.frappe.db.exists.return_value = True
+
+        # Even if there is a cached token, it should be ignored
+        auth.frappe.cache().get_value.return_value = json.dumps(
+            {
+                "active": True,
+                "email": self.test_user_email,
+                "exp": (
+                    datetime.datetime.now() + datetime.timedelta(hours=1)
+                ).timestamp(),
+            }
+        )
+
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        introspection_payload = {
+            "active": True,
+            "email": self.test_user_email,
+            "sub": "user-subject-123",
+            "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(),
+        }
+        mock_response.json.return_value = introspection_payload
+        mock_requests.post.return_value = mock_response
+
+        # --- Act ---
+        auth.validate()
+
+        # --- Assert ---
+        # 1. Ensure it still made a request (meaning it didn't use the cache)
+        mock_requests.post.assert_called_once()
+
+        # 2. Ensure it didn't call get_value on cache (though my implementation skips it)
+        # Actually, if it skips it, get_value won't be called.
+        auth.frappe.cache().get_value.assert_not_called()
+
+        # 3. Ensure it didn't try to SAVE to cache
+        auth.frappe.cache().set_value.assert_not_called()
+
+        auth.frappe.set_user.assert_called_once_with(self.test_user_email)
+
+    @patch("castlecraft.auth.requests")
+    @patch("castlecraft.auth.frappe.get_value")
+    @patch("castlecraft.auth.jwt")
+    @patch("castlecraft.auth.get_idp")
+    def test_jwt_cache_disabled(
+        self, mock_get_idp, mock_jwt, mock_get_value, mock_requests
+    ):
+        """
+        Verify that JWT verification bypasses cache and doesn't store in cache
+        when is_cache_disabled is set.
+        """
+        # --- Arrange ---
+        mock_jwt_idp = MockDoc(
+            dict(
+                doctype="CFE Identity Provider",
+                authorization_type="JWT Verification",
+                is_cache_disabled=1,  # CACHE DISABLED
+                email_key="email",
+                create_user=False,
+                fetch_user_info=False,
+                jwks_endpoint="https://idp.example.com/.well-known/jwks.json",
+                audience_claim_key="aud",
+                allowed_audience=[MockDoc(dict(aud="test-audience"))],
+            )
+        )
+        auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}"
+        mock_get_idp.return_value = mock_jwt_idp
+        mock_get_value.return_value = self.test_user_email
+
+        jwt_payload = {
+            "sub": "user-sub-123",
+            "exp": datetime.datetime.now().timestamp() + 3600,
+            "aud": "test-audience",
+            "email": self.test_user_email,
+        }
+
+        # Even if there is a cached payload, it should be ignored
+        auth.frappe.cache().get_value.return_value = json.dumps(jwt_payload)
+
+        # Mock the internal dependencies of validate_signature
+        mock_requests.get.return_value.json.return_value = {
+            "keys": [{"kid": "test-kid"}]
+        }
+        mock_jwt.get_unverified_header.return_value = {"kid": "test-kid"}
+        mock_jwt.algorithms.RSAAlgorithm.from_jwk.return_value = MagicMock()
+        mock_jwt.decode.return_value = jwt_payload
+
+        # --- Act ---
+        auth.validate()
+
+        # --- Assert ---
+        # 1. Ensure it still decoded the JWT (meaning it didn't use the cache)
+        mock_jwt.decode.assert_called_once()
+
+        # 2. Ensure it didn't call get_value on cache
+        auth.frappe.cache().get_value.assert_not_called()
+
+        # 3. Ensure it didn't try to SAVE to cache
+        auth.frappe.cache().set_value.assert_not_called()
+
+        auth.frappe.set_user.assert_called_once_with(self.test_user_email)