BB2-3321: Oauth revoke endpoint

apps/dot_ext/tests/test_authorization.py

@@ -578,6 +578,61 @@ def test_dag_expiration_exists(self):
         expiration_date_string = strftime('%Y-%m-%d %H:%M:%SZ', expiration_date.timetuple())
         self.assertEqual(tkn["access_grant_expiration"][:-4], expiration_date_string[:-4])
 
+    def test_revoke_endpoint(self):
+        redirect_uri = 'http://localhost'
+        # create a user
+        self._create_user('anna', '123456')
+        capability_a = self._create_capability('Capability A', [])
+        capability_b = self._create_capability('Capability B', [])
+        # create an application and add capabilities
+        application = self._create_application(
+            'an app',
+            grant_type=Application.GRANT_AUTHORIZATION_CODE,
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            redirect_uris=redirect_uri)
+        application.scope.add(capability_a, capability_b)
+        # user logs in
+        request = HttpRequest()
+        self.client.login(request=request, username='anna', password='123456')
+        # post the authorization form with only one scope selected
+        payload = {
+            'client_id': application.client_id,
+            'response_type': 'code',
+            'redirect_uri': redirect_uri,
+            'scope': ['capability-a'],
+            'expires_in': 86400,
+            'allow': True,
+        }
+        response = self.client.post(reverse('oauth2_provider:authorize'), data=payload)
+        self.client.logout()
+        self.assertEqual(response.status_code, 302)
+        # now extract the authorization code and use it to request an access_token
+        query_dict = parse_qs(urlparse(response['Location']).query)
+        authorization_code = query_dict.pop('code')
+        token_request_data = {
+            'grant_type': 'authorization_code',
+            'code': authorization_code,
+            'redirect_uri': redirect_uri,
+            'client_id': application.client_id,
+            'client_secret': application.client_secret_plain,
+        }
+        c = Client()
+        response = c.post('/v1/o/token/', data=token_request_data)
+        self.assertEqual(response.status_code, 200)
+        # extract token and use it to make a revoke request
+        tkn = response.json()['access_token']
+        revoke_request_data = {
+            'token': tkn,
+            'client_id': application.client_id,
+            'client_secret': application.client_secret_plain,
+        }
+        c = Client()
+        rev_response = c.post('/v1/o/revoke/', data=revoke_request_data)
+        self.assertEqual(rev_response.status_code, 200)
+        # check DAG deletion
+        dags_count = DataAccessGrant.objects.count()
+        self.assertEqual(dags_count, 0)
+
     def test_refresh_with_revoked_token(self):
         redirect_uri = 'http://localhost'
         # create a user

apps/dot_ext/urls.py

@@ -15,6 +15,7 @@
     ),
     path("token/", views.TokenView.as_view(), name="token"),
     path("revoke_token/", views.RevokeTokenView.as_view(), name="revoke-token"),
+    path("revoke/", views.RevokeView.as_view(), name="revoke"),
     path("introspect/", views.IntrospectTokenView.as_view(), name="introspect"),
 ]
 

apps/dot_ext/v2/urls.py

@@ -15,6 +15,7 @@
     ),
     path("token/", views.TokenView.as_view(), name="token-v2"),
     path("revoke_token/", views.RevokeTokenView.as_view(), name="revoke-token-v2"),
+    path("revoke/", views.RevokeView.as_view(), name="revoke-v2"),
     path("introspect/", views.IntrospectTokenView.as_view(), name="introspect-v2"),
 ]
 

apps/dot_ext/views/__init__.py

@@ -1,2 +1,2 @@
 from .application import ApplicationRegistration, ApplicationUpdate, ApplicationDelete  # NOQA
-from .authorization import AuthorizationView, ApprovalView, TokenView, RevokeTokenView, IntrospectTokenView  # NOQA
+from .authorization import AuthorizationView, ApprovalView, TokenView, RevokeTokenView, RevokeView, IntrospectTokenView  # NOQA

apps/dot_ext/views/authorization.py

@@ -120,12 +120,12 @@ def sensitive_info_check(self, request):
     def get_template_names(self):
         flag = get_waffle_flag_model().get("limit_data_access")
         if waffle.switch_is_active('require-scopes'):
-            if flag.rollout or (flag.id is not None and flag.is_active_for_user(self.application.user)):
+            if flag.rollout or (flag.id is not None and self.application and flag.is_active_for_user(self.application.user)):
                 return ["design_system/new_authorize_v2.html"]
             else:
                 return ["design_system/authorize_v2.html"]
         else:
-            if flag.rollout or (flag.id is not None and flag.is_active_for_user(self.user)):
+            if flag.rollout or (flag.id is not None and self.user and flag.is_active_for_user(self.user)):
                 return ["design_system/new_authorize_v2.html"]
             else:
                 return ["design_system/authorize.html"]
@@ -354,6 +354,30 @@ def post(self, request, *args, **kwargs):
         return super().post(request, args, kwargs)
 
 
+@method_decorator(csrf_exempt, name="dispatch")
+class RevokeView(DotRevokeTokenView):
+
+    @method_decorator(sensitive_post_parameters("password"))
+    def post(self, request, *args, **kwargs):
+        try:
+            app = validate_app_is_active(request)
+        except (InvalidClientError, InvalidGrantError) as error:
+            return json_response_from_oauth2_error(error)
+
+        token = get_access_token_model().objects.get(
+            token=request.POST.get("token"))
+        try:
+            dag = DataAccessGrant.objects.get(
+                beneficiary=token.user,
+                application=app
+            )
+            dag.delete()
+        except DataAccessGrant.DoesNotExist:
+            pass
+
+        return super().post(request, args, kwargs)
+
+
 @method_decorator(csrf_exempt, name="dispatch")
 class IntrospectTokenView(DotIntrospectTokenView):