BB2-4250: Make v3_endpoints waffle switch app specific

apps/accounts/views/oauth2_profile.py

@@ -10,6 +10,7 @@
 from apps.fhir.bluebutton.permissions import ApplicationActivePermission
 
 from apps.versions import Versions
+from apps.wellknown.permissions import V3EarlyAdopterWellKnownPermission
 
 
 def _get_userinfo(user, version=Versions.NOT_AN_API_VERSION):
@@ -41,7 +42,8 @@ def _get_userinfo(user, version=Versions.NOT_AN_API_VERSION):
 @authentication_classes([OAuth2Authentication])
 @permission_classes([ApplicationActivePermission,
                      TokenHasProtectedCapability,
-                     DataAccessGrantPermission])
+                     DataAccessGrantPermission,
+                     V3EarlyAdopterWellKnownPermission])
 @protected_resource()  # Django OAuth Toolkit -> resource_owner = AccessToken
 def _openidconnect_userinfo(request, version=Versions.NOT_AN_API_VERSION):
     # NOTE: The **kwargs are not used anywhere down the callchain, and are being ignored.

apps/authorization/tests/test_data_access_grant_permissions.py

@@ -13,7 +13,7 @@
 from apps.authorization.models import (
     DataAccessGrant,
 )
-from waffle.testutils import override_switch
+from waffle.testutils import override_switch, override_flag
 from apps.fhir.bluebutton.tests.test_fhir_resources_read_search_w_validation import (
     get_response_json,
 )
@@ -145,6 +145,7 @@ def setUp(self):
         self.client = Client()
 
     @override_switch('v3_endpoints', active=True)
+    @override_flag('v3_early_adopter', active=True)
     def _assert_call_all_fhir_endpoints(
         self,
         access_token=None,

apps/dot_ext/tests/test_authorization.py

@@ -6,15 +6,19 @@
 from datetime import datetime
 from dateutil.relativedelta import relativedelta
 # from oauth2_provider.compat import parse_qs, urlparse
-from urllib.parse import parse_qs, urlparse
+from oauthlib.oauth2.rfc6749.errors import AccessDeniedError as AccessDeniedTokenCustomError
 from oauth2_provider.models import get_access_token_model, get_refresh_token_model
+from django.core.exceptions import PermissionDenied
 from django.http import HttpRequest
 from django.urls import reverse
 from django.test import Client
+from unittest.mock import patch, MagicMock
+from urllib.parse import parse_qs, urlencode, urlparse
 from waffle.testutils import override_switch
 
 from apps.test import BaseApiTest
 from ..models import Application, ArchivedToken
+from apps.dot_ext.views import AuthorizationView, TokenView
 from apps.authorization.models import DataAccessGrant, ArchivedDataAccessGrant
 from http import HTTPStatus
 
@@ -1020,3 +1024,103 @@ def _execute_token_endpoint(self, token_path):
         c = Client()
         response = c.post(token_path, data=token_request_data)
         self.assertEqual(response.status_code, 200)
+
+    @patch('apps.dot_ext.views.authorization.get_user_model')
+    @patch('apps.dot_ext.views.authorization.get_application_model')
+    @patch('apps.dot_ext.views.authorization.get_waffle_flag_model')
+    def test_permission_denied_raised_for_authorize_app_not_in_flag(
+        self,
+        mock_get_flag_model,
+        mock_get_application_model,
+        mock_get_user_model
+    ):
+        # Unit test to show that we will raise an AccessDeniedTokenCustomError
+        # when the validate_v3_authorization_request of AuthorizationView function is called
+        # when the v3_early_adopter flag is not active for an application_user
+        # set up a fake request
+        # TODO: When we enable v3 endpoints for all applications, remove this test
+        query_params = urlencode({'client_id': 'FAKE_CLIENT_ID'})
+        request = HttpRequest()
+        request.META['QUERY_STRING'] = query_params
+
+        # Mock the required objects/queries around flag/application/user
+        fake_flag = MagicMock()
+        fake_flag.id = 123
+        fake_flag.name = 'v3_early_adopter'
+        fake_flag.is_active_for_user.return_value = False
+        mock_get_flag_model.return_value.get.return_value = fake_flag
+
+        fake_application = MagicMock()
+        fake_application.id = 42
+        fake_application.user_id = 999
+        fake_application.name = 'TestApp'
+        mock_manager_app = MagicMock()
+        mock_manager_app.get.return_value = fake_application
+        mock_get_application_model.return_value.objects = mock_manager_app
+
+        fake_user = MagicMock()
+        fake_user.id = 999
+        mock_manager_user = MagicMock()
+        mock_manager_user.get.return_value = fake_user
+        mock_get_user_model.return_value.objects = mock_manager_user
+
+        # Create an instance of the view
+        view_instance = AuthorizationView()
+        view_instance.request = request
+
+        with self.assertRaises(AccessDeniedTokenCustomError):
+            view_instance.validate_v3_authorization_request()
+
+    @patch('apps.dot_ext.views.authorization.get_user_model')
+    @patch('apps.dot_ext.views.authorization.get_application_model')
+    @patch('apps.dot_ext.views.authorization.get_refresh_token_model')
+    @patch('apps.dot_ext.views.authorization.get_waffle_flag_model')
+    def test_permission_denied_raised_for_refresh_token_app_not_in_flag(
+        self,
+        mock_get_flag_model,
+        mock_get_refresh_token_model,
+        mock_get_application_model,
+        mock_get_user_model
+    ):
+        # BB2-4250Unit test to show that we will raise an PermissionDenied
+        # when the validate_v3_token_call of TokenView function is called
+        # when the v3_early_adopter flag is not active for an application_user
+        # TODO: When we enable v3 endpoints for all applications, remove this test
+        token_value = 'FAKE_REFRESH_TOKEN'
+        body = urlencode({'refresh_token': token_value}).encode('utf-8')
+        request = HttpRequest()
+        request._body = body
+
+        # Mock the required objects/queries around flag/refresh_token/application/user
+        fake_flag = MagicMock()
+        fake_flag.id = 123
+        fake_flag.name = 'v3_early_adopter'
+        fake_flag.is_active_for_user.return_value = False
+        mock_get_flag_model.return_value.get.return_value = fake_flag
+
+        fake_refresh_token = MagicMock()
+        fake_refresh_token.token = token_value
+        fake_refresh_token.application_id = 42
+        mock_manager_refresh = MagicMock()
+        mock_manager_refresh.get.return_value = fake_refresh_token
+        mock_get_refresh_token_model.return_value.objects = mock_manager_refresh
+
+        fake_application = MagicMock()
+        fake_application.id = 42
+        fake_application.user_id = 999
+        fake_application.name = 'TestApp'
+        mock_manager_app = MagicMock()
+        mock_manager_app.get.return_value = fake_application
+        mock_get_application_model.return_value.objects = mock_manager_app
+
+        fake_user = MagicMock()
+        fake_user.id = 999
+        mock_manager_user = MagicMock()
+        mock_manager_user.get.return_value = fake_user
+        mock_get_user_model.return_value.objects = mock_manager_user
+
+        # Create an instance of the view
+        view_instance = TokenView()
+
+        with self.assertRaises(PermissionDenied):
+            view_instance.validate_v3_token_call(request)

apps/dot_ext/views/authorization.py

@@ -4,30 +4,36 @@
 from functools import wraps
 from time import strftime
 
+from django.conf import settings
+from django.contrib.auth import get_user_model
 from django.contrib.auth.views import redirect_to_login
-from django.http import JsonResponse
+from django.http import HttpResponseForbidden, JsonResponse
 from django.http.response import HttpResponse, HttpResponseBadRequest
 from django.template.response import TemplateResponse
+from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
 from django.utils.decorators import method_decorator
 from django.views.decorators.csrf import csrf_exempt
 from django.views.decorators.debug import sensitive_post_parameters
 from apps.dot_ext.constants import TOKEN_ENDPOINT_V3_KEY
+from oauthlib.oauth2.rfc6749.errors import AccessDeniedError as AccessDeniedTokenCustomError
 from oauth2_provider.exceptions import OAuthToolkitError
-from oauth2_provider.views.base import app_authorized, get_access_token_model
+from oauth2_provider.views.base import app_authorized
+from oauth2_provider.models import get_refresh_token_model, get_access_token_model
 from oauth2_provider.views.base import AuthorizationView as DotAuthorizationView
 from oauth2_provider.views.base import TokenView as DotTokenView
 from oauth2_provider.views.base import RevokeTokenView as DotRevokeTokenView
 from oauth2_provider.views.introspect import (
     IntrospectTokenView as DotIntrospectTokenView,
 )
-from waffle import switch_is_active
+from waffle import switch_is_active, get_waffle_flag_model
 from oauth2_provider.models import get_application_model
 from oauthlib.oauth2 import AccessDeniedError
 from oauthlib.oauth2.rfc6749.errors import InvalidClientError, InvalidGrantError, InvalidRequestError
 from urllib.parse import urlparse, parse_qs
 import html
 from apps.dot_ext.scopes import CapabilitiesScopes
 import apps.logging.request_logger as bb2logging
+from apps.versions import Versions
 
 from ..signals import beneficiary_authorized_application
 from ..forms import SimpleAllowForm
@@ -41,6 +47,7 @@
 )
 from ..models import Approval
 from ..utils import (
+    get_api_version_number_from_url,
     remove_application_user_pair_tokens_data_access,
     validate_app_is_active,
     json_response_from_oauth2_error,
@@ -84,6 +91,7 @@
     # TODO: rename this so that it isn't the same as self.version (works but confusing)
     # this needs to be here for urls.py as_view(version) calls, but don't use it
     version = 0
+    # Variable to help reduce the amount of times validate_v3_authorization_request is called
     form_class = SimpleAllowForm
     login_url = "/mymedicare/login"
 
@@ -116,6 +124,8 @@
                 error_message = "State parameter should have a minimum of 16 characters"
                 return JsonResponse({"status_code": 400, "message": error_message}, status=400)
 
+        # BB2-4250: This code will not execute if the application is not in the v3_early_adopter flag
+        # so it will not be modified as part of BB2-4250
         if switch_is_active('v3_endpoints') and v3:
             if 'scope' not in request.GET:
                 missing_params.append("scope")
@@ -140,6 +150,15 @@
         initially create an AuthFlowUuid object for authorization
         flow tracing in logs.
         """
+        path_info = self.request.__dict__.get('path_info')
+        version = get_api_version_number_from_url(path_info)
+        # If it is not version 3, we don't need to check anything, just return
+        if version == Versions.V3:
+            try:
+                self.validate_v3_authorization_request()
+            except AccessDeniedTokenCustomError as e:
+                return HttpResponseForbidden(e)
+
         # TODO: Should the client_id match a valid application here before continuing, instead of after matching to FHIR_ID?
         if not kwargs.get('is_subclass_approvalview', False):
             # Create new authorization flow trace UUID in session and AuthFlowUuid instance, if subclass is not ApprovalView
@@ -224,6 +243,28 @@
         kwargs['code_challenge_method'] = request.GET.get('code_challenge_method', None)
         return super().get(request, *args, **kwargs)
 
+    def validate_v3_authorization_request(self):
+        flag = get_waffle_flag_model().get('v3_early_adopter')
+        req_meta = self.request.META
+        url_query = parse_qs(req_meta.get('QUERY_STRING'))
+        client_id = url_query.get('client_id', [None])
+        try:
+            application = get_application_model().objects.get(client_id=client_id[0])
+            application_user = get_user_model().objects.get(id=application.user_id)
+
+            if flag.id is None or flag.is_active_for_user(application_user):
+                # Update the class variable to ensure subsequent calls to dispatch don't call this function
+                # more times than is needed
+                return
+            else:
+                raise AccessDeniedTokenCustomError(
+                    description=settings.APPLICATION_DOES_NOT_HAVE_V3_ENABLED_YET.format(application.name)
+                )
+        except ObjectDoesNotExist:
+            raise AccessDeniedTokenCustomError(
+                description='Unable to verify permission.'
+            )
+
     def form_valid(self, form):
         client_id = form.cleaned_data["client_id"]
         application = get_application_model().objects.get(client_id=client_id)
@@ -262,6 +303,7 @@
         refresh_token_delete_cnt = 0
 
         try:
+
             if not scopes:
                 # Since the create_authorization_response will re-inject scopes even when none are
                 # valid, we want to pre-emptively treat this as an error case
@@ -410,13 +452,48 @@
                 description=f"Invalid parameters in request: {invalid_parameters}"
             )
 
+    def validate_v3_token_call(self, request) -> None:
+        flag = get_waffle_flag_model().get('v3_early_adopter')
+
+        try:
+            url_query = parse_qs(request._body.decode('utf-8'))
+            refresh_token_from_request = url_query.get('refresh_token', [None])
+            refresh_token = get_refresh_token_model().objects.get(token=refresh_token_from_request[0])
+            application = get_application_model().objects.get(id=refresh_token.application_id)
+            application_user = get_user_model().objects.get(id=application.user_id)
+
+            if flag.id is None or flag.is_active_for_user(application_user):
+                return
+            else:
+                raise PermissionDenied(
+                    settings.APPLICATION_DOES_NOT_HAVE_V3_ENABLED_YET.format(application.name)
+                )
+        except ObjectDoesNotExist:
+            return JsonResponse(
+                {'status_code': 403, 'message': 'Unable to verify permission.'},
+                status=403,
+            )
+
     @method_decorator(sensitive_post_parameters("password"))
     def post(self, request, *args, **kwargs):
+        path_info = self.request.__dict__.get('path_info')
+        version = get_api_version_number_from_url(path_info)
+        # If it is not version 3, we don't need to check anything, just return
         try:
+            if version == Versions.V3:
+                self.validate_v3_token_call(request)
             self.validate_token_endpoint_request_body(request)
             app = validate_app_is_active(request)
         except (InvalidClientError, InvalidGrantError, InvalidRequestError) as error:
             return json_response_from_oauth2_error(error)
+        except PermissionDenied:
+            log.exception('Permission denied during token endpoint processing.')
+            # This error will not match other errors thrown by this waffle_flag as Github raised
+            # a security concern about it, but only here.
+            return JsonResponse(
+                {'status_code': 403, 'message': 'You do not have permission to perform this action.'},
+                status=403,
+            )
 
         url, headers, body, status = self.create_token_response(request)
 

apps/fhir/bluebutton/permissions.py

@@ -2,8 +2,11 @@
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
+from oauth2_provider.views.base import get_access_token_model
+from oauth2_provider.models import get_application_model
 from rest_framework import permissions, exceptions
-from rest_framework.exceptions import AuthenticationFailed
+from rest_framework.exceptions import AuthenticationFailed, PermissionDenied
+from waffle import get_waffle_flag_model
 from .constants import ALLOWED_RESOURCE_TYPES
 from apps.versions import Versions, VersionNotMatched
 
@@ -106,3 +109,22 @@ def has_permission(self, request, view):
             )
 
         return True
+
+
+class V3EarlyAdopterPermission(permissions.BasePermission):
+    def has_permission(self, request, view):
+        # if it is not version 3, we do not need to check the waffle switch or flag
+        if view.version < Versions.V3:
+            return True
+
+        token = get_access_token_model().objects.get(token=request._auth)
+        application = get_application_model().objects.get(id=token.application_id)
+        application_user = get_user_model().objects.get(id=application.user_id)
+        flag = get_waffle_flag_model().get('v3_early_adopter')
+
+        if flag.id is None or flag.is_active_for_user(application_user):
+            return True
+        else:
+            raise PermissionDenied(
+                settings.APPLICATION_DOES_NOT_HAVE_V3_ENABLED_YET.format(application.name)
+            )