Expect the remote exp to be defined in time zone UTC conform rfc (Fixes #1291)

Author: wkleinheerenbrink

oauth2_provider/oauth2_validators.py

@@ -37,6 +37,7 @@
 )
 from .scopes import get_scopes_backend
 from .settings import oauth2_settings
+from .utils import get_timezone
 
 
 log = logging.getLogger("oauth2_provider")
@@ -388,7 +389,11 @@ def _get_token_from_authentication_server(
                 expires = max_caching_time
 
             scope = content.get("scope", "")
-            expires = make_aware(expires) if settings.USE_TZ else expires
+
+            if settings.USE_TZ:
+                expires = make_aware(
+                    expires, timezone=get_timezone(oauth2_settings.AUTHENTICATION_SERVER_EXP_TIME_ZONE)
+                )
 
             access_token, _created = AccessToken.objects.update_or_create(
                 token=token,

oauth2_provider/settings.py

@@ -101,6 +101,8 @@
     "RESOURCE_SERVER_AUTH_TOKEN": None,
     "RESOURCE_SERVER_INTROSPECTION_CREDENTIALS": None,
     "RESOURCE_SERVER_TOKEN_CACHING_SECONDS": 36000,
+    # Authentication Server Exp Timezone: the time zone use dby Auth Server for generate EXP
+    "AUTHENTICATION_SERVER_EXP_TIME_ZONE": "UTC",
     # Whether or not PKCE is required
     "PKCE_REQUIRED": True,
     # Whether to re-create OAuthlibCore on every request.

oauth2_provider/utils.py

@@ -1,5 +1,6 @@
 import functools
 
+from django.conf import settings
 from jwcrypto import jwk
 
 
@@ -10,3 +11,24 @@ def jwk_from_pem(pem_string):
     Converting from PEM is expensive for large keys such as those using RSA.
     """
     return jwk.JWK.from_pem(pem_string.encode("utf-8"))
+
+
+@functools.lru_cache
+def get_timezone(time_zone):
+    """
+    Return the default time zone as a tzinfo instance.
+
+    This is the time zone defined by settings.TIME_ZONE.
+    """
+    try:
+        import zoneinfo
+    except ImportError:
+        import pytz
+
+        return pytz.timezone(time_zone)
+    else:
+        if getattr(settings, "USE_DEPRECATED_PYTZ", False):
+            import pytz
+
+            return pytz.timezone(time_zone)
+        return zoneinfo.ZoneInfo(time_zone)

tests/test_introspection_auth.py

@@ -29,7 +29,7 @@
 AccessToken = get_access_token_model()
 UserModel = get_user_model()
 
-exp = datetime.datetime.now() + datetime.timedelta(days=1)
+default_exp = datetime.datetime.now() + datetime.timedelta(days=1)
 
 
 class ScopeResourceView(ScopedProtectedResourceView):
@@ -42,27 +42,28 @@ def post(self, request, *args, **kwargs):
         return HttpResponse("This is a protected resource", 200)
 
 
+class MockResponse:
+    def __init__(self, json_data, status_code):
+        self.json_data = json_data
+        self.status_code = status_code
+
+    def json(self):
+        return self.json_data
+
+
 def mocked_requests_post(url, data, *args, **kwargs):
     """
     Mock the response from the authentication server
     """
 
-    class MockResponse:
-        def __init__(self, json_data, status_code):
-            self.json_data = json_data
-            self.status_code = status_code
-
-        def json(self):
-            return self.json_data
-
     if "token" in data and data["token"] and data["token"] != "12345678900":
         return MockResponse(
             {
                 "active": True,
                 "scope": "read write dolphin",
                 "client_id": "client_id_{}".format(data["token"]),
                 "username": "{}_user".format(data["token"]),
-                "exp": int(calendar.timegm(exp.timetuple())),
+                "exp": int(calendar.timegm(default_exp.timetuple())),
             },
             200,
         )
@@ -75,6 +76,21 @@ def json(self):
     )
 
 
+def mocked_introspect_request_short_living_token(url, data, *args, **kwargs):
+    exp = datetime.datetime.now() + datetime.timedelta(minutes=30)
+
+    return MockResponse(
+        {
+            "active": True,
+            "scope": "read write dolphin",
+            "client_id": "client_id_{}".format(data["token"]),
+            "username": "{}_user".format(data["token"]),
+            "exp": int(calendar.timegm(exp.timetuple())),
+        },
+        200,
+    )
+
+
 urlpatterns = [
     path("oauth2/", include("oauth2_provider.urls")),
     path("oauth2-test-resource/", ScopeResourceView.as_view()),
@@ -156,24 +172,76 @@ def test_get_token_from_authentication_server_existing_token(self, mock_get):
         self.assertEqual(token.user.username, "foo_user")
         self.assertEqual(token.scope, "read write dolphin")
 
-    @mock.patch("requests.post", side_effect=mocked_requests_post)
-    def test_get_token_from_authentication_server_expires_timezone(self, mock_get):
+    @mock.patch("requests.post", side_effect=mocked_introspect_request_short_living_token)
+    def test_get_token_from_authentication_server_expires_no_timezone(self, mock_get):
         """
         Test method _get_token_from_authentication_server for projects with USE_TZ False
         """
         settings_use_tz_backup = settings.USE_TZ
         settings.USE_TZ = False
         try:
-            self.validator._get_token_from_authentication_server(
+            access_token = self.validator._get_token_from_authentication_server(
+                "foo",
+                oauth2_settings.RESOURCE_SERVER_INTROSPECTION_URL,
+                oauth2_settings.RESOURCE_SERVER_AUTH_TOKEN,
+                oauth2_settings.RESOURCE_SERVER_INTROSPECTION_CREDENTIALS,
+            )
+
+            self.assertFalse(access_token.is_expired())
+        except ValueError as exception:
+            self.fail(str(exception))
+        finally:
+            settings.USE_TZ = settings_use_tz_backup
+
+    @mock.patch("requests.post", side_effect=mocked_introspect_request_short_living_token)
+    def test_get_token_from_authentication_server_expires_utc_timezone(self, mock_get):
+        """
+        Test method _get_token_from_authentication_server for projects with USE_TZ True and a UTC Timezone
+        """
+        settings_use_tz_backup = settings.USE_TZ
+        settings_time_zone_backup = settings.TIME_ZONE
+        settings.USE_TZ = True
+        settings.TIME_ZONE = "UTC"
+        try:
+            access_token = self.validator._get_token_from_authentication_server(
                 "foo",
                 oauth2_settings.RESOURCE_SERVER_INTROSPECTION_URL,
                 oauth2_settings.RESOURCE_SERVER_AUTH_TOKEN,
                 oauth2_settings.RESOURCE_SERVER_INTROSPECTION_CREDENTIALS,
             )
+
+            self.assertFalse(access_token.is_expired())
+        except ValueError as exception:
+            self.fail(str(exception))
+        finally:
+            settings.USE_TZ = settings_use_tz_backup
+            settings.TIME_ZONE = settings_time_zone_backup
+
+    @mock.patch("requests.post", side_effect=mocked_introspect_request_short_living_token)
+    def test_get_token_from_authentication_server_expires_non_utc_timezone(self, mock_get):
+        """
+        Test method _get_token_from_authentication_server for projects with USE_TZ True and a non UTC Timezone
+
+        This test is important to check if the UTC Exp. date gets converted correctly
+        """
+        settings_use_tz_backup = settings.USE_TZ
+        settings_time_zone_backup = settings.TIME_ZONE
+        settings.USE_TZ = True
+        settings.TIME_ZONE = "Europe/Amsterdam"
+        try:
+            access_token = self.validator._get_token_from_authentication_server(
+                "foo",
+                oauth2_settings.RESOURCE_SERVER_INTROSPECTION_URL,
+                oauth2_settings.RESOURCE_SERVER_AUTH_TOKEN,
+                oauth2_settings.RESOURCE_SERVER_INTROSPECTION_CREDENTIALS,
+            )
+
+            self.assertFalse(access_token.is_expired())
         except ValueError as exception:
             self.fail(str(exception))
         finally:
             settings.USE_TZ = settings_use_tz_backup
+            settings.TIME_ZONE = settings_time_zone_backup
 
     @mock.patch("requests.post", side_effect=mocked_requests_post)
     def test_validate_bearer_token(self, mock_get):