BB2-4781: Revoke prior tokens on new auth flow (#1605)

* BB2-4781: Revoke prior tokens on new auth flow

* Ensure token revocation only happens on v3 auth flows

* Move the revocation of prior tokens to form_valid of AuthorizationView from post of TokenView. Revoke all previous tokens as the newest one will not have been created yet at that point

* Address PR feedback, remove no longer needed test

* Address feedback - use previously existing utils function

* Fix typo

* Add unit test coverage

Author: JamesDemeryNava

apps/dot_ext/constants.py

@@ -339,9 +339,9 @@
         # Result:
         'result_has_error': False,
         'result_token_scopes_granted': APPLICATION_SCOPES_NON_DEMOGRAPHIC,
-        'result_access_token_count': 2,
-        'result_refresh_token_count': 2,
-        'result_archived_token_count': 2,
+        'result_access_token_count': 1,
+        'result_refresh_token_count': 1,
+        'result_archived_token_count': 3,
         'result_archived_data_access_grant_count': 2,
     },
     'test 5: app_requires = True bene_share = False': {
@@ -365,9 +365,9 @@
         # Result:
         'result_has_error': False,
         'result_token_scopes_granted': APPLICATION_SCOPES_FULL,
-        'result_access_token_count': 2,
-        'result_refresh_token_count': 2,
-        'result_archived_token_count': 4,
+        'result_access_token_count': 1,
+        'result_refresh_token_count': 1,
+        'result_archived_token_count': 5,
         'result_archived_data_access_grant_count': 3,
     },
     # Tests for request_app_requires_demographic = False
@@ -419,9 +419,9 @@
         # Result:
         'result_has_error': False,
         'result_token_scopes_granted': SCOPES_JUST_EOB_AND_B,
-        'result_access_token_count': 2,
-        'result_refresh_token_count': 2,
-        'result_archived_token_count': 8,
+        'result_access_token_count': 1,
+        'result_refresh_token_count': 1,
+        'result_archived_token_count': 9,
         'result_archived_data_access_grant_count': 6,
     },
     'test 11: app_requires = None bene_share = None request just PATIENT and A': {
@@ -445,9 +445,9 @@
         # Result:
         'result_has_error': False,
         'result_token_scopes_granted': SCOPES_JUST_PATIENT_AND_A,
-        'result_access_token_count': 2,
-        'result_refresh_token_count': 2,
-        'result_archived_token_count': 10,
+        'result_access_token_count': 1,
+        'result_refresh_token_count': 1,
+        'result_archived_token_count': 11,
         'result_archived_data_access_grant_count': 7,
     },
     'test 13: app_requires = False bene_share = True request just PATIENT and A': {

apps/dot_ext/tests/test_beneficiary_demographic_scope_changes.py

@@ -1,18 +1,20 @@
 import json
-from apps.test import BaseApiTest
-from django.core.management import call_command
-from django.http import HttpRequest
-from django.urls import reverse
+from http import HTTPStatus
+from unittest import mock
 
 # from oauth2_provider.compat import parse_qs, urlparse
 from urllib.parse import parse_qs, urlparse
+
+from django.core.management import call_command
+from django.http import HttpRequest
+from django.urls import reverse
 from oauth2_provider.models import AccessToken, RefreshToken
 from rest_framework.test import APIClient
 from waffle.testutils import override_switch
-from apps.authorization.models import DataAccessGrant, ArchivedDataAccessGrant
-from apps.dot_ext.models import ArchivedToken, Application
-from http import HTTPStatus
-from unittest import mock
+
+from apps.authorization.models import ArchivedDataAccessGrant, DataAccessGrant
+from apps.dot_ext.models import Application, ArchivedToken
+from apps.test import BaseApiTest
 
 
 class TestBeneficiaryDemographicScopesChanges(BaseApiTest):
@@ -127,7 +129,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         )
 
         # Assert auth request was successful
-        self.assertEqual(status_code, 200)
+        self.assertEqual(status_code, HTTPStatus.OK)
 
         # Assert scope in response content
         self.assertEqual(response_scopes, sorted(APPLICATION_SCOPES_FULL))
@@ -138,7 +140,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # Assert access to userinfo end point?
         client.credentials(HTTP_AUTHORIZATION='Bearer ' + token_1.token)
         response = client.get('/v1/connect/userinfo')
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, HTTPStatus.OK)
 
         # ------ TEST #2:  Test refresh of token_1
         refresh_request_data = {
@@ -152,7 +154,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         content = json.loads(response.content.decode('utf-8'))
 
         # Assert successful
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, HTTPStatus.OK)
 
         # Assert response scopes
         response_scopes = sorted(content['scope'].split())
@@ -166,7 +168,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # Assert access to userinfo end point?
         client.credentials(HTTP_AUTHORIZATION='Bearer ' + token.token)
         response = client.get('/v1/connect/userinfo')
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, HTTPStatus.OK)
 
         # Verify token counts expected.
         self.assertEqual(AccessToken.objects.count(), 1)
@@ -197,7 +199,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # Assert NO access to userinfo end point?
         client.credentials(HTTP_AUTHORIZATION='Bearer ' + token_3.token)
         response = client.get('/v1/connect/userinfo')
-        self.assertEqual(response.status_code, 403)
+        self.assertEqual(response.status_code, HTTPStatus.FORBIDDEN)
 
         # Verify token counts expected.
         self.assertEqual(AccessToken.objects.count(), 1)
@@ -215,7 +217,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # Test access to userinfo end point? NO ACCESS!
         response = client.get('/v1/connect/userinfo')
         content = json.loads(response.content)
-        self.assertEqual(response.status_code, 401)
+        self.assertEqual(response.status_code, HTTPStatus.UNAUTHORIZED)
         self.assertEqual(content.get('detail', None), 'Authentication credentials were not provided.')
 
         # ------ TEST #5:  Test token_1 from TEST #1 token refresh? NO ACCESS!
@@ -241,7 +243,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         )
 
         # Assert auth request was successful
-        self.assertEqual(status_code, 200)
+        self.assertEqual(status_code, HTTPStatus.OK)
 
         # Assert scope in response content
         self.assertEqual(response_scopes, sorted(APPLICATION_SCOPES_FULL))
@@ -252,7 +254,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # Assert access to userinfo end point?
         client.credentials(HTTP_AUTHORIZATION='Bearer ' + token_6.token)
         response = client.get('/v1/connect/userinfo')
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, HTTPStatus.OK)
 
         # ------ TEST #7: Test token_3 from TEST #3 again. It should still have access, but no permission with status=403.
 
@@ -262,13 +264,13 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # Test access to userinfo end point?
         response = client.get('/v1/connect/userinfo')
         content = json.loads(response.content)
-        self.assertEqual(response.status_code, 403)
-        self.assertEqual(content.get('detail', None), 'You do not have permission to perform this action.')
+        self.assertEqual(response.status_code, HTTPStatus.UNAUTHORIZED)
+        self.assertEqual(content.get('detail', None), 'Authentication credentials were not provided.')
 
         # Verify token counts expected.
-        self.assertEqual(AccessToken.objects.count(), 2)
-        self.assertEqual(RefreshToken.objects.count(), 2)
-        self.assertEqual(ArchivedToken.objects.count(), 2)
+        self.assertEqual(AccessToken.objects.count(), 1)
+        self.assertEqual(RefreshToken.objects.count(), 1)
+        self.assertEqual(ArchivedToken.objects.count(), 3)
 
         # Verify grant counts expected.
         self.assertEqual(DataAccessGrant.objects.count(), 1)
@@ -280,15 +282,15 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
 
         # Perform partial authorization request, with out application getting an access token.
         response = self.client.post(reverse('oauth2_provider:authorize'), data=payload)
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, HTTPStatus.FOUND)
 
         # Setup token_3 in APIClient from previous step. It should be removed now?
         client.credentials(HTTP_AUTHORIZATION='Bearer ' + token_3.token)
 
         # Test access to userinfo end point?
         response = client.get('/v1/connect/userinfo')
         content = json.loads(response.content)
-        self.assertEqual(response.status_code, 401)
+        self.assertEqual(response.status_code, HTTPStatus.UNAUTHORIZED)
         self.assertEqual(content.get('detail', None), 'Authentication credentials were not provided.')
 
         # Verify token counts expected.
@@ -309,7 +311,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         )
 
         # Assert auth request was successful
-        self.assertEqual(status_code, 200)
+        self.assertEqual(status_code, HTTPStatus.OK)
 
         # Verify token counts expected.
         self.assertEqual(AccessToken.objects.count(), 1)
@@ -323,14 +325,14 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # Assert access to userinfo end point?
         client.credentials(HTTP_AUTHORIZATION='Bearer ' + token_9.token)
         response = client.get('/v1/connect/userinfo')
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, HTTPStatus.OK)
 
         # Beneficiary chooses the DENY button choice on consent page
         payload['allow'] = False
 
         # Perform partial authorization request, with out application getting an access token.
         response = self.client.post(reverse('oauth2_provider:authorize'), data=payload)
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, HTTPStatus.FOUND)
 
         # Verify token counts expected.
         self.assertEqual(AccessToken.objects.count(), 1)
@@ -346,7 +348,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # when the allow parameter is false
         client.credentials(HTTP_AUTHORIZATION='Bearer ' + token_9.token)
         response = client.get('/v1/connect/userinfo')
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, HTTPStatus.OK)
 
         # BB2-4270: Remove prior active tokens so tests below are not looking for multiple active tokens
         # which is an impossible state
@@ -360,12 +362,12 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         payload['allow'] = True
 
         # Perform authorization request
-        token_10, refresh_token_10, status_code, response_scopes, access_token_scopes = self._authorize_and_request_token(
-            payload, application
+        token_10, refresh_token_10, status_code, response_scopes, access_token_scopes = (
+            self._authorize_and_request_token(payload, application)
         )
 
         # Assert auth request was successful
-        self.assertEqual(status_code, 200)
+        self.assertEqual(status_code, HTTPStatus.OK)
 
         # Verify token counts expected.
         self.assertEqual(AccessToken.objects.count(), 1)
@@ -379,15 +381,15 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # Assert access to userinfo end point?
         client.credentials(HTTP_AUTHORIZATION='Bearer ' + token_10.token)
         response = client.get('/v1/connect/userinfo')
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, HTTPStatus.OK)
 
         # Application changes choice to require demographic scopes
         application.require_demographic_scopes = False
         application.save()
 
         # Perform partial authorization request, with out application getting an access token.
         response = self.client.post(reverse('oauth2_provider:authorize'), data=payload)
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, HTTPStatus.FOUND)
 
         # Verify token counts expected.
         self.assertEqual(AccessToken.objects.count(), 0)
@@ -400,7 +402,7 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
 
         # Perform partial authorization request, with out application getting an access token.
         response = self.client.post(reverse('oauth2_provider:authorize'), data=payload)
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, HTTPStatus.FOUND)
 
         # Verify token counts expected.
         self.assertEqual(AccessToken.objects.count(), 0)
@@ -414,5 +416,5 @@ def test_bene_demo_scopes_change(self, mock_get_and_update):
         # Assert access to userinfo end point?
         client.credentials(HTTP_AUTHORIZATION='Bearer ' + token_10.token)
         response = client.get('/v1/connect/userinfo')
-        self.assertEqual(response.status_code, 401)
+        self.assertEqual(response.status_code, HTTPStatus.UNAUTHORIZED)
         self.assertEqual(content.get('detail', None), 'Authentication credentials were not provided.')

apps/dot_ext/tests/test_utils.py

@@ -1,7 +1,17 @@
+from unittest.mock import MagicMock, patch
+
 from django.test import TestCase
 
 from apps.dot_ext.constants import SUPPORTED_VERSION_TEST_CASES
+<<<<<<< HEAD
 from apps.dot_ext.utils import get_api_version_number_from_url, validate_latin_extended_string
+=======
+from apps.dot_ext.utils import (
+    get_api_version_number_from_url,
+    remove_application_user_pair_tokens_data_access,
+    validate_latin_extended_string,
+)
+>>>>>>> 2e2d4da2 (BB2-4781: Revoke prior tokens on new auth flow (#1605))
 from apps.versions import VersionNotMatched
 
 
@@ -33,3 +43,28 @@ def test_latin_extended_failure(self):
 
         for text in invalid_inputs:
             assert not validate_latin_extended_string(text)
+
+    @patch('apps.dot_ext.utils.AccessToken')
+    @patch('apps.dot_ext.utils.DataAccessGrant')
+    @patch('apps.dot_ext.utils.RefreshToken')
+    def test_remove_application_user_pair_tokens_data_access_delete_access_tokens_not_grant(
+        self, mock_refresh_token, mock_data_access_grant, mock_access_token
+    ) -> None:
+        application = MagicMock()
+        user = MagicMock()
+        access_token_queryset = MagicMock()
+        mock_access_token.objects.filter.return_value = access_token_queryset
+        data_access_grant_queryset = MagicMock()
+        mock_data_access_grant.objects.filter.return_value = data_access_grant_queryset
+        refresh_token_queryset = MagicMock()
+        mock_refresh_token.objects.filter.return_value = refresh_token_queryset
+
+        remove_application_user_pair_tokens_data_access(application, user, False, True)
+
+        mock_access_token.objects.filter.assert_called_once_with(application=application, user=user)
+        access_token_queryset.delete.assert_called_once()
+
+        data_access_grant_queryset.delete.assert_not_called()
+
+        mock_refresh_token.objects.filter.assert_called_once_with(application=application, user=user)
+        refresh_token_queryset.delete.assert_called_once()