Ensure token and auth flows function or throw 403s depending on values in the flag. Add 403 handling for userinfo v3

JamesDemeryNava · JamesDemeryNava · commit 423bb30e11f6 · 2025-11-24T13:25:30.000-05:00
diff --git a/apps/accounts/views/oauth2_profile.py b/apps/accounts/views/oauth2_profile.py
@@ -1,19 +1,16 @@
 from collections import OrderedDict
-from django.conf import settings
-from django.contrib.auth import get_user_model
 from django.http import JsonResponse
 from oauth2_provider.contrib.rest_framework import OAuth2Authentication
 from oauth2_provider.decorators import protected_resource
-from oauth2_provider.models import get_access_token_model, get_application_model
 from rest_framework.decorators import api_view, permission_classes, authentication_classes
-from waffle import get_waffle_flag_model
 
 from apps.authorization.permissions import DataAccessGrantPermission
 from apps.capabilities.permissions import TokenHasProtectedCapability
 from apps.fhir.bluebutton.models import Crosswalk
 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):
@@ -45,23 +42,11 @@ 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.
-    # 4250: Handling to ensure this only returns successfully if the flag is enabled for the application
-    # associated with the user making the call
-    if version == Versions.V3:
-        user = get_user_model().objects.get(username=request.resource_owner)
-        access_token = get_access_token_model().objects.get(user_id=user.id)
-        application = get_application_model().objects.get(id=access_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 not flag.is_active_for_user(application_user):
-            return JsonResponse(
-                {'status_code': 403, 'message': settings.APPLICATION_DOES_NOT_HAVE_V3_ENABLED_YET.format(application.name)},
-                status=403,
-            )
 
     return JsonResponse(_get_userinfo(request.resource_owner, version))
 
@@ -75,8 +60,6 @@ def openidconnect_userinfo_v2(request):
 
 
 def openidconnect_userinfo_v3(request):
-    print("openidconnect_userinfo_v3 request: ", request.__dict__)
-    print("openidconnect_userinfo_v3 user: ", request.user.__dict__)
     return _openidconnect_userinfo(request, version=Versions.V3)
 
 
diff --git a/apps/dot_ext/views/authorization.py b/apps/dot_ext/views/authorization.py
@@ -10,11 +10,12 @@
 from django.http import JsonResponse
 from django.http.response import HttpResponse, HttpResponseBadRequest
 from django.template.response import TemplateResponse
-from django.core.exceptions import ObjectDoesNotExist
+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
 from oauth2_provider.models import get_refresh_token_model, get_access_token_model
@@ -79,50 +80,8 @@ def _wrapped(request, *args, **kwargs):
     return _wrapped
 
 
-def check_v3_endpoint_access(view_func):
-    @wraps(view_func)
-    def _wrapped(request, *args, **kwargs):
-        # 4250-TODO how do we not call this so many times?
-        path_info = request.__dict__.get('path_info')
-        version = get_api_version_number_from_url(path_info)
-        if version != Versions.V3:
-            return view_func(request, *args, **kwargs)
-
-        flag = get_waffle_flag_model().get('v3_early_adopter')
-        req_meta = request.META
-        url_query = parse_qs(req_meta.get('QUERY_STRING'))
-        client_id = url_query.get('client_id', [None])
-        try:
-            if client_id[0]:
-                application = get_application_model().objects.get(client_id=client_id[0])
-            else:
-                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 not None and flag.is_active_for_user(application_user):
-                return view_func(request, *args, **kwargs)
-            else:
-                return JsonResponse(
-                    {'status_code': 403, 'message': settings.APPLICATION_DOES_NOT_HAVE_V3_ENABLED_YET.format(application.name)},
-                    status=403,
-                )
-        except ObjectDoesNotExist:
-            # 4250-TODO Do we need this?
-            return JsonResponse(
-                {'status_code': 500, 'message': 'Error retrieving data'},
-                status=500,
-            )
-
-    return _wrapped
-
-
 @method_decorator(csrf_exempt, name="dispatch")
 @method_decorator(require_post_state_decorator, name="dispatch")
-@method_decorator(check_v3_endpoint_access, name="dispatch")
 class AuthorizationView(DotAuthorizationView):
     """
     Override the base authorization view from dot to
@@ -274,6 +233,27 @@ def get(self, request, *args, **kwargs):
         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 not None and flag.is_active_for_user(application_user):
+                return
+            else:
+                raise AccessDeniedTokenCustomError(
+                    description=settings.APPLICATION_DOES_NOT_HAVE_V3_ENABLED_YET.format(application.name)
+                )
+        except ObjectDoesNotExist:
+            # 4250-TODO Do we need this?
+            return JsonResponse(
+                {'status_code': 500, 'message': 'Error retrieving data'},
+                status=500,
+            )
+
     def form_valid(self, form):
         client_id = form.cleaned_data["client_id"]
         application = get_application_model().objects.get(client_id=client_id)
@@ -312,6 +292,12 @@ def form_valid(self, form):
         refresh_token_delete_cnt = 0
 
         try:
+            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:
+                self.validate_v3_authorization_request()
+
             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
@@ -441,7 +427,6 @@ def dispatch(self, request, uuid, *args, **kwargs):
         return result
 
 
-# @method_decorator(check_v3_endpoint_access, name="dispatch")
 @method_decorator(csrf_exempt, name="dispatch")
 class TokenView(DotTokenView):
 
@@ -461,13 +446,47 @@ def validate_token_endpoint_request_body(self, request):
                 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')
+        req_meta = request.META
+        url_query = parse_qs(req_meta.get('QUERY_STRING'))
+        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 not None and flag.is_active_for_user(application_user):
+                return
+            else:
+                raise PermissionDenied(
+                    settings.APPLICATION_DOES_NOT_HAVE_V3_ENABLED_YET.format(application.name)
+                )
+        except ObjectDoesNotExist:
+            # 4250-TODO Do we need this?
+            return JsonResponse(
+                {'status_code': 500, 'message': 'Error retrieving data'},
+                status=500,
+            )
+
     @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 as e:
+            return JsonResponse(
+                {'status_code': 403, 'message': str(e)},
+                status=403,
+            )
 
         url, headers, body, status = self.create_token_response(request)
 
diff --git a/apps/fhir/bluebutton/permissions.py b/apps/fhir/bluebutton/permissions.py
@@ -113,9 +113,6 @@ def has_permission(self, request, view):
 
 class V3EarlyAdopterPermission(permissions.BasePermission):
     def has_permission(self, request, view):
-        print("IN HAS_PERMISSION OF V3EarlyAdopterPermission")
-        print("V3EarlyAdopterPermission request: ", request.__dict__)
-        print("V3EarlyAdopterPermission view: ", view.__dict__)
         # if it is not version 3, we do not need to check the waffle switch or flag
         if view.version < Versions.V3:
             return True
diff --git a/apps/wellknown/permissions.py b/apps/wellknown/permissions.py
@@ -0,0 +1,36 @@
+import logging
+
+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
+from rest_framework.exceptions import PermissionDenied
+from waffle import get_waffle_flag_model
+from apps.versions import Versions
+
+import apps.logging.request_logger as bb2logging
+
+logger = logging.getLogger(bb2logging.HHS_SERVER_LOGNAME_FMT.format(__name__))
+
+
+class V3EarlyAdopterWellKnownPermission(permissions.BasePermission):
+    # BB2-4250: Handling to ensure this only returns successfully if the flag is enabled for the application
+    # associated with the user making the call
+    def has_permission(self, request, view):
+        version = view.kwargs.get('version')
+        # if it is not version 3, we do not need to check the waffle switch or flag
+        if 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 not None and flag.is_active_for_user(application_user):
+            return True
+        else:
+            raise PermissionDenied(
+                settings.APPLICATION_DOES_NOT_HAVE_V3_ENABLED_YET.format(application.name)
+            )
