Added middleware to refresh access tokens

Author: daggaz

django_auth_adfs/backend.py

@@ -396,6 +396,11 @@ def authenticate(self, request=None, authorization_code=None, **kwargs):
             logger.debug("Authentication backend was called but no authorization code was received")
             return
 
+        # If there's no request object, we pass control to the next authentication backend
+        if request is None:
+            logger.debug("Authentication backend was called without request")
+            return
+
         # If loaded data is too old, reload it again
         provider_config.load_config()
 

django_auth_adfs/middleware.py

@@ -63,5 +63,5 @@ def middleware(request):
             backend = auth.load_backend(backend_str)
             if isinstance(backend, AdfsAuthCodeBackend):
                 backend.process_request(request)
-        return get_response()
+        return get_response(request)
     return middleware

tests/settings.py

@@ -35,6 +35,7 @@
     'django.contrib.messages.middleware.MessageMiddleware',
     'django.middleware.clickjacking.XFrameOptionsMiddleware',
 
+    'django_auth_adfs.middleware.adfs_refresh_middleware',
     'django_auth_adfs.middleware.LoginRequiredMiddleware',
 )
 

tests/test_authentication.py

@@ -1,5 +1,7 @@
 import base64
 
+from datetime import datetime, timedelta
+
 from django.urls import reverse
 
 from django_auth_adfs.exceptions import MFARequired
@@ -12,9 +14,9 @@
 from copy import deepcopy
 
 from django.contrib.auth.models import Group, User
-from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
+from django.core.exceptions import ObjectDoesNotExist
 from django.db.models.signals import post_save
-from django.test import RequestFactory, TestCase
+from django.test import TestCase
 from mock import Mock, patch
 
 from django_auth_adfs import signals
@@ -29,13 +31,12 @@ def setUp(self):
         Group.objects.create(name='group1')
         Group.objects.create(name='group2')
         Group.objects.create(name='group3')
-        self.request = RequestFactory().get('/oauth2/callback')
         self.signal_handler = Mock()
         signals.post_authenticate.connect(self.signal_handler)
 
     @mock_adfs("2012")
     def test_post_authenticate_signal_send(self):
-        response = self.client.get(reverse('django_auth_adfs:callback'), data={'code': "dummycode"})
+        self.client.get(reverse('django_auth_adfs:callback'), data={'code': "dummycode"})
         self.assertEqual(self.signal_handler.call_count, 1)
 
     @mock_adfs("2012")
@@ -495,3 +496,34 @@ def test_nonexisting_user(self):
                 patch("django_auth_adfs.backend.settings", Settings()):
             response = self.client.get(reverse('django_auth_adfs:callback'), data={'code': "testcode"})
             self.assertEqual(response.status_code, 401)
+
+    @mock_adfs("2016")
+    def test_access_token_unexpired(self):
+        response = self.client.get(reverse('django_auth_adfs:callback'), data={'code': "testcode"})
+        self.assertFalse(response.wsgi_request.user.is_anonymous)
+        response = self.client.get(reverse('test'))
+        self.assertEqual(response.status_code, 200)
+
+    @mock_adfs("2016")
+    def test_access_token_expired(self):
+        response = self.client.get(reverse('django_auth_adfs:callback'), data={'code': "testcode"})
+        self.assertFalse(response.wsgi_request.user.is_anonymous)
+        fromisoformat = datetime.fromisoformat
+        with patch('django_auth_adfs.backend.datetime') as dt:
+            dt.fromisoformat = fromisoformat
+            dt.now.return_value = datetime.now() + timedelta(hours=1)
+            response = self.client.get(reverse('test'))
+        self.assertEqual(response.status_code, 200)
+
+    @mock_adfs("2016", refresh_token_expired=True)
+    def test_refresh_token_expired(self):
+        response = self.client.get(reverse('django_auth_adfs:callback'), data={'code': "testcode"})
+        self.assertFalse(response.wsgi_request.user.is_anonymous)
+        fromisoformat = datetime.fromisoformat
+        with patch('django_auth_adfs.backend.datetime') as dt:
+            dt.fromisoformat = fromisoformat
+            dt.now.return_value = datetime.now() + timedelta(hours=1)
+            response = self.client.get(reverse('test'))
+        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response['Location'], f"{reverse('django_auth_adfs:login')}?next=/")
+        self.assertTrue(response.wsgi_request.user.is_anonymous)

tests/urls.py

@@ -1,6 +1,9 @@
-from django.urls import include, re_path
+from django.urls import include, re_path, path
+
+from tests.views import TestView
 
 urlpatterns = [
+    path('', TestView.as_view(), name='test'),
     re_path(r'^oauth2/', include('django_auth_adfs.urls')),
     re_path(r'^oauth2/', include('django_auth_adfs.drf_urls')),
 ]

tests/utils.py

@@ -5,6 +5,7 @@
 import time
 from datetime import datetime, tzinfo, timedelta
 from functools import partial
+from urllib.parse import parse_qs
 
 import jwt
 import responses
@@ -98,9 +99,14 @@ def build_access_token_azure_groups_in_claim_source(request):
     return do_build_access_token(request, issuer, groups_in_claim_names=True)
 
 
+def build_access_token_adfs_expired(request):
+    issuer = "http://adfs.example.com/adfs/services/trust"
+    return do_build_access_token(request, issuer, refresh_token_expired=True)
+
+
 def do_build_mfa_error(request):
     response = {'error_description': 'AADSTS50076'}
-    return 400, [], json.dumps(response)
+    return 400, {}, json.dumps(response)
 
 
 def do_build_graph_response(request):
@@ -111,7 +117,11 @@ def do_build_graph_response_no_group_perm(request):
     return do_build_ms_graph_groups(request, missing_group_names=True)
 
 
-def do_build_access_token(request, issuer, schema=None, no_upn=False, idp=None, groups_in_claim_names=False):
+def do_build_access_token(request, issuer, schema=None, no_upn=False, idp=None, groups_in_claim_names=False,
+                          refresh_token_expired=False):
+    data = parse_qs(request.body)
+    if data.get('grant_type') == ['refresh_token'] and data.get('refresh_token') == ['expired_refresh_token']:
+        return 401, {}, None
     issued_at = int(time.time())
     expires = issued_at + 3600
     auth_time = datetime.utcnow()
@@ -159,16 +169,20 @@ def do_build_access_token(request, issuer, schema=None, no_upn=False, idp=None,
             }
         }
     token = jwt.encode(claims, signing_key_b, algorithm="RS256")
+    if refresh_token_expired:
+        refresh_token = 'expired_refresh_token'
+    else:
+        refresh_token = 'random_refresh_token'
     response = {
         'resource': 'django_website.adfs.relying_party_id',
         'token_type': 'bearer',
         'refresh_token_expires_in': 28799,
-        'refresh_token': 'random_refresh_token',
+        'refresh_token': refresh_token,
         'expires_in': 3600,
         'id_token': 'not_used',
         'access_token': token.decode() if isinstance(token, bytes) else token  # PyJWT>=2 returns a str instead of bytes
     }
-    return 200, [], json.dumps(response)
+    return 200, {}, json.dumps(response)
 
 
 def do_build_obo_access_token(request):
@@ -228,7 +242,7 @@ def do_build_obo_access_token(request):
         'refresh_token': 'not_used',
         'access_token': token.decode() if isinstance(token, bytes) else token  # PyJWT>=2 returns a str instead of bytes
     }
-    return 200, [], json.dumps(response)
+    return 200, {}, json.dumps(response)
 
 
 def do_build_ms_graph_groups(request, missing_group_names=False):
@@ -308,7 +322,7 @@ def do_build_ms_graph_groups(request, missing_group_names=False):
     if missing_group_names:
         for group in response["value"]:
             group["displayName"] = None
-    return 200, [], json.dumps(response)
+    return 200, {}, json.dumps(response)
 
 
 def build_openid_keys(request, empty_keys=False):
@@ -337,15 +351,15 @@ def build_openid_keys(request, empty_keys=False):
                 },
             ]
         }
-    return 200, [], json.dumps(keys)
+    return 200, {}, json.dumps(keys)
 
 
 def build_adfs_meta(request):
     with open(os.path.join(os.path.dirname(__file__), "mock_files/FederationMetadata.xml"), mode="r") as f:
         data = "".join(f.readlines())
     data = data.replace("REPLACE_WITH_CERT_A", base64.b64encode(signing_cert_a).decode())
     data = data.replace("REPLACE_WITH_CERT_B", base64.b64encode(signing_cert_b).decode())
-    return 200, [], data
+    return 200, {}, data
 
 
 def mock_adfs(
@@ -356,6 +370,7 @@ def mock_adfs(
     version=None,
     requires_obo=False,
     missing_graph_group_perm=False,
+    refresh_token_expired=False,
 ):
     if adfs_version not in ["2012", "2016", "azure"]:
         raise NotImplementedError("This version of ADFS is not implemented")
@@ -465,6 +480,12 @@ def wrapper(*original_args, **original_kwargs):
                             callback=do_build_mfa_error,
                             content_type='application/json',
                         )
+                    elif refresh_token_expired:
+                        rsps.add_callback(
+                            rsps.POST, token_endpoint,
+                            callback=build_access_token_adfs_expired,
+                            content_type='application/json',
+                        )
                     else:
                         rsps.add_callback(
                             rsps.POST, token_endpoint,

tests/views.py

@@ -1,2 +1,11 @@
+from django.http import HttpResponse
+from django.views import View
+
+
 def test_failed_response(request, error_message, status):
     pass
+
+
+class TestView(View):
+    def get(self, request):
+        return HttpResponse('okay')