Better cache implem, add sso metada in shared cache.

Author: regilero

django_pyoidc/utils.py

@@ -1,4 +1,11 @@
+import hashlib
+import logging
+from typing import Dict, Union
+
 from django.conf import settings
+from django.core.cache import BaseCache, caches
+
+logger = logging.getLogger(__name__)
 
 
 def get_setting_for_sso_op(op_name: str, key: str, default=None):
@@ -10,3 +17,42 @@ def get_setting_for_sso_op(op_name: str, key: str, default=None):
 
 def get_settings_for_sso_op(op_name: str):
     return settings.DJANGO_PYOIDC[op_name]
+
+
+class OIDCCacheBackendForDjango:
+    """Implement General cache for OIDC using django cache"""
+
+    def __init__(self, op_name):
+        self.op_name = op_name
+        self.enabled = get_setting_for_sso_op(
+            self.op_name, "OIDC_CACHE_PROVIDER_METADATA", False
+        )
+        if self.enabled:
+            self.storage: BaseCache = caches[
+                get_setting_for_sso_op(self.op_name, "CACHE_DJANGO_BACKEND")
+            ]
+
+    def generate_hashed_cache_key(self, value: str) -> str:
+        h = hashlib.new("sha256")
+        h.update(value.encode())
+        cache_key = h.hexdigest()
+        return cache_key
+
+    def clear(self):
+        return self.storage.clear()
+
+    def get_key(self, key):
+        return f"oidc-{self.op_name}-{key}"
+
+    def set(self, key: str, value: Dict[str, Union[str, bool]], expiry: int) -> None:
+        if self.enabled:
+            self.storage.set(self.get_key(key), value, expiry)
+
+    def __getitem__(self, key: str) -> Dict[str, Union[str, bool]]:
+        if self.enabled:
+            data = self.storage.get(self.get_key(key))
+            if data is None:
+                raise KeyError  # Makes __getItem__ handle like Python dict
+            return data
+        else:
+            raise KeyError

django_pyoidc/views.py

@@ -1,12 +1,10 @@
 import datetime
-import hashlib
 import logging
 from importlib import import_module
 
 # import oic
 from django.conf import settings
 from django.contrib import auth, messages
-from django.core.cache import caches
 from django.core.exceptions import PermissionDenied, SuspiciousOperation
 from django.http import HttpResponse
 from django.shortcuts import redirect, resolve_url
@@ -23,7 +21,11 @@
 from django_pyoidc import get_user_by_email
 from django_pyoidc.models import OIDCSession
 from django_pyoidc.session import OIDCCacheSessionBackendForDjango
-from django_pyoidc.utils import get_setting_for_sso_op, get_settings_for_sso_op
+from django_pyoidc.utils import (
+    OIDCCacheBackendForDjango,
+    get_setting_for_sso_op,
+    get_settings_for_sso_op,
+)
 
 SessionStore = import_module(settings.SESSION_ENGINE).SessionStore
 
@@ -48,7 +50,8 @@ class OIDClient:
     def __init__(self, op_name, session_id=None):
         self._op_name = op_name
 
-        self.cache_backend = OIDCCacheSessionBackendForDjango(self._op_name)
+        self.session_cache_backend = OIDCCacheSessionBackendForDjango(self._op_name)
+        self.general_cache_backend = OIDCCacheBackendForDjango(self._op_name)
 
         consumer_config = {
             # "debug": True,
@@ -61,7 +64,7 @@ def __init__(self, op_name, session_id=None):
         }
 
         self.consumer = Consumer(
-            session_db=self.cache_backend,
+            session_db=self.session_cache_backend,
             consumer_config=consumer_config,
             client_config=client_config,
         )
@@ -72,13 +75,22 @@ def __init__(self, op_name, session_id=None):
             op_name, "OIDC_PROVIDER_DISCOVERY_URI"
         )
         client_secret = get_setting_for_sso_op(op_name, "OIDC_CLIENT_SECRET")
-        # self.client_extension.provider_config(provider_info_uri)
         self.client_extension.client_secret = client_secret
 
         if session_id:
             self.consumer.restore(session_id)
         else:
-            self.consumer.provider_config(provider_info_uri)
+
+            cache_key = self.general_cache_backend.generate_hashed_cache_key(
+                provider_info_uri
+            )
+            try:
+                config = self.general_cache_backend[cache_key]
+            except KeyError:
+                config = self.consumer.provider_config(provider_info_uri)
+                # shared microcache for provider config
+                # FIXME: Setting for duration
+                self.general_cache_backend.set(cache_key, config, 60)
             self.consumer.client_secret = client_secret
 
 
@@ -367,6 +379,10 @@ class OIDCCallbackView(OIDCView):
 
     http_method_names = ["get"]
 
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.general_cache_backend = OIDCCacheBackendForDjango(self.op_name)
+
     def success_url(self, request):
         # Pull the next url from the session or settings --we don't need to
         # sanitize here because it should already have been sanitized.
@@ -390,14 +406,12 @@ def _introspect_access_token(self, access_token_jwt):
         # FIXME: in what case could we not have an access token available?
         # should we raise an error then?
         if access_token_jwt is not None:
-            cache = caches[self.get_setting("CACHE_DJANGO_BACKEND")]
-            h = hashlib.new("sha256")
-            h.update(access_token_jwt.encode())
-            cache_key = h.hexdigest()
-
-            access_token_claims = cache.get("cache_key")
-
-            if access_token_claims is None:
+            cache_key = self.general_cache_backend.generate_hashed_cache_key(
+                access_token_jwt
+            )
+            try:
+                access_token_claims = self.general_cache_backend["cache_key"]
+            except KeyError:
                 # CACHE MISS
 
                 # RFC 7662: token introspection: ask SSO to validate and render the jwt as json
@@ -426,7 +440,7 @@ def _introspect_access_token(self, access_token_jwt):
                 print(access_token_expiry)
                 exp = int(access_token_expiry) - int(current)
                 print(exp)
-                cache.set(cache_key, access_token_claims, exp)
+                self.general_cache_backend.set(cache_key, access_token_claims, exp)
         return access_token_claims
 
     def get(self, request, *args, **kwargs):

tests/e2e/tests_keycloak.py

@@ -16,8 +16,6 @@
 
 from tests.e2e.utils import OIDCE2EKeycloakTestCase
 
-#  , wrap_class
-
 # HTTP debug for requests
 http_client.HTTPConnection.debuglevel = 1
 
@@ -526,98 +524,3 @@ def test_06_selenium_minimal_audience_checks(self, *args):
         # Check the session message is shown
         self.assertTrue("message: [email protected] is logged in." in bodyText)
         self._selenium_logout(end_url)
-
-
-#    @override_settings(
-#        DJANGO_PYOIDC={
-#            "sso1": {
-#                "OIDC_CLIENT_ID": "app1",
-#                "CACHE_DJANGO_BACKEND": "default",
-#                "OIDC_PROVIDER_DISCOVERY_URI": "http://localhost:8080/auth/realms/realm1",
-#                "OIDC_CLIENT_SECRET": "secret_app1",
-#                "OIDC_CALLBACK_PATH": "/callback",
-#                "LOGIN_URIS_REDIRECT_ALLOWED_HOSTS": ["testserver"],
-#                "REDIRECT_REQUIRES_HTTPS": False,
-#                "POST_LOGOUT_REDIRECT_URI": "/test-logout-done",
-#                "POST_LOGIN_URI_SUCCESS": "/test-success",
-#                "POST_LOGIN_URI_FAILURE": "/test-failure",
-#                "HOOK_GET_USER": "tests.e2e.test_app.callback:get_user",
-#            },
-#        },
-#    )
-#    def test_10_selenium_pyoidc_provider_config_calls(self, *args):
-#        """
-#        FIXME:
-#        Ensure calls to SSO provider configuration is managed with a cache.
-#
-#        Several logins should not generates a call to provider config each time.
-#        Provider configuration should be shared between users, at least for a
-#        short cache lifetime duration.
-#
-#        FIXME:
-#        For pyoidc this means the Client must be feed with provider_info data
-#        to avoid re-requesting it in auto discovery mode.
-#        i.e. keys client_registration and provider_info in
-#        ["srv_discovery_url", "client_info", "client_registration", "provider_info"].
-#        "srv_discovery_url" should only be used when no cache data is available.
-#        """
-#        timeout = 5
-#
-#        login_url = reverse("test_login")
-#        success_url = reverse("test_success")
-#        post_logout_url = reverse("test_logout_done")
-#        start_url = f"{self.live_server_url}{login_url}"
-#        middle_url = f"{self.live_server_url}{success_url}"
-#        end_url = f"{self.live_server_url}{post_logout_url}"
-#        from oic.oauth2 import Client
-#
-#        with wrap_class(Client, "provider_config") as mocked_provider_config:
-#
-#            self.wait = WebDriverWait(self.selenium, timeout)
-#            self._selenium_sso_login(
-#                start_url,
-#                middle_url,
-#                "user_limit_app2",
-#                "passwd2",
-#                active_sso_session=False,
-#            )
-#
-#            bodyText = self.selenium.find_element(By.TAG_NAME, "body").text
-#
-#            # Check the session message is shown
-#            self.assertTrue(
-#                "message: [email protected] is logged in." in bodyText
-#            )
-#
-#            self._selenium_logout(end_url)
-#
-#            self._selenium_sso_login(
-#                start_url,
-#                middle_url,
-#                "user_limit_app1",
-#                "passwd1",
-#                active_sso_session=False,
-#            )
-#
-#            bodyText = self.selenium.find_element(By.TAG_NAME, "body").text
-#
-#            # Check the session message is shown
-#            self.assertTrue(
-#                "message: [email protected] is logged in." in bodyText
-#            )
-#
-#            self._selenium_logout(end_url)
-#
-#            # oic.oauth2.base.PBase.http_request:
-#            # expected_call= [
-#            #     call('http://localhost:8080/auth/realms/realm1/.well-known/openid-configuration', allow_redirects=True)
-#            # ]
-#            print("==================================")
-#            print(mocked_provider_config.call_count)
-#            print(mocked_provider_config.call_args_list)
-#            # mocked_provider_config.assert_any_call(expected_calls)
-#
-#            # Need to set this test alone, maybe with a simplier Keycloak setUp...
-#            # self.assertEquals( mocked_provider_config.call_count, 1)
-#            self.assertEquals(mocked_provider_config.call_count, 2)
-#

tests/tests_session.py

@@ -1,6 +1,7 @@
 from unittest import mock
-from unittest.mock import ANY, call
+from unittest.mock import call
 
+from django_pyoidc.utils import OIDCCacheBackendForDjango
 from django_pyoidc.views import OIDClient
 from tests.utils import OIDCTestCase
 
@@ -14,10 +15,62 @@ def test_session_isolation_between_providers(self, mocked_provider_config):
         sso1 = OIDClient(op_name="sso1")
         sso2 = OIDClient(op_name="sso2")
 
+        mocked_provider_config.assert_has_calls([call(""), call("")])
+        assert 2 == mocked_provider_config.call_count
+
+        sso1.consumer._backup(sid="1234")
+        sso2.consumer._backup(sid="1234")
+
+        # no more calls
+        mocked_provider_config.assert_has_calls([call(""), call("")])
+        assert 2 == mocked_provider_config.call_count
+
+        client1 = OIDClient(op_name="sso1", session_id="1234")
+        self.assertEqual(client1.consumer.client_id, "1")
+
+        client2 = OIDClient(op_name="sso2", session_id="1234")
+
+        # no more calls
+        mocked_provider_config.assert_has_calls([call(""), call("")])
+        assert 2 == mocked_provider_config.call_count
+
+        self.assertEqual(client2.consumer.client_id, "2")
+
+    @mock.patch(
+        "django_pyoidc.views.Consumer.provider_config",
+        return_value=('[{"foo": "bar"}]'),
+    )
+    def test_session_isolation_between_providers_cached(self, mocked_provider_config):
+        """
+        Test that different SSO providers with active cache using same SID do not conflict
+        """
+
+        # empty the caches
+        cache1 = OIDCCacheBackendForDjango("sso3")
+        cache2 = OIDCCacheBackendForDjango("sso4")
+        cache1.clear()
+        cache2.clear()
+
+        sso1 = OIDClient(op_name="sso3")
+        mocked_provider_config.assert_has_calls([call("http://sso3/uri")])
+        assert 1 == mocked_provider_config.call_count
+
+        sso2 = OIDClient(op_name="sso4")
+        mocked_provider_config.assert_has_calls(
+            [call("http://sso3/uri"), call("http://sso4/uri")]
+        )
+        assert 2 == mocked_provider_config.call_count
+
         sso1.consumer._backup(sid="1234")
         sso2.consumer._backup(sid="1234")
 
-        client_new = OIDClient(op_name="sso1", session_id="1234")
+        client1 = OIDClient(op_name="sso3", session_id="1234")
+        self.assertEqual(client1.consumer.client_id, "3")
+
+        client2 = OIDClient(op_name="sso4", session_id="1234")
+        self.assertEqual(client2.consumer.client_id, "4")
 
-        mocked_provider_config.assert_has_calls([call(ANY), call(ANY)])
-        self.assertEqual(client_new.consumer.client_id, "1")
+        mocked_provider_config.assert_has_calls(
+            [call("http://sso3/uri"), call("http://sso4/uri")]
+        )
+        assert 2 == mocked_provider_config.call_count

tests/utils.py

@@ -8,6 +8,7 @@
 @override_settings(
     DJANGO_PYOIDC={
         "sso1": {
+            "OIDC_CACHE_PROVIDER_METADATA": False,
             "OIDC_CLIENT_ID": "1",
             "CACHE_DJANGO_BACKEND": "default",
             "OIDC_PROVIDER_DISCOVERY_URI": "",
@@ -20,11 +21,26 @@
             "POST_LOGIN_URI_FAILURE": "/logout_failure",
         },
         "sso2": {
+            "OIDC_CACHE_PROVIDER_METADATA": False,
             "OIDC_CLIENT_ID": "2",
             "CACHE_DJANGO_BACKEND": "default",
             "OIDC_PROVIDER_DISCOVERY_URI": "",
             "OIDC_CLIENT_SECRET": "",
         },
+        "sso3": {
+            "OIDC_CACHE_PROVIDER_METADATA": True,
+            "OIDC_CLIENT_ID": "3",
+            "CACHE_DJANGO_BACKEND": "default",
+            "OIDC_PROVIDER_DISCOVERY_URI": "http://sso3/uri",
+            "OIDC_CLIENT_SECRET": "",
+        },
+        "sso4": {
+            "OIDC_CACHE_PROVIDER_METADATA": True,
+            "OIDC_CLIENT_ID": "4",
+            "CACHE_DJANGO_BACKEND": "default",
+            "OIDC_PROVIDER_DISCOVERY_URI": "http://sso4/uri",
+            "OIDC_CLIENT_SECRET": "",
+        },
     }
 )
 class OIDCTestCase(TestCase):