BB2-4961: Switch Account Link

apps/dot_ext/tests/test_authorization.py

@@ -1671,6 +1671,7 @@ def test_v3_sets_scopes_in_kwargs(self, mock_pc):
         view = AuthorizationView(version=Versions.V3)
         mock_application = MagicMock()
         request = HttpRequest()
+        request.session = {}
         request.beneficiary_name = 'Test A User'
         view.request = request
         view.application = mock_application
@@ -1705,6 +1706,7 @@ def test_v2_does_not_scopes_in_kwargs(self, mock_pc):
         view = AuthorizationView(version=Versions.V2)
         mock_application = MagicMock()
         request = HttpRequest()
+        request.session = {}
         view.request = request
         view.application = mock_application
 
@@ -1736,6 +1738,7 @@ def test_v3_form_scopes_are_intersection_of_app_and_requested(self, mock_pc):
         view = AuthorizationView(version=Versions.V3)
         mock_application = MagicMock()
         request = HttpRequest()
+        request.session = {}
         request.beneficiary_name = 'Test A User'
         view.request = request
         view.application = mock_application

apps/dot_ext/utils.py

@@ -58,6 +58,23 @@ def validate_client_id(client_id: str) -> None:
         )
 
 
+def get_oauth_param(request: HttpRequest, parameter: str, fallback_session_parameter: str = None) -> str | None:
+    """Resolve an OAuth parameter from GET, then session, then session['oauth_params'].
+
+    Args:
+        request: Django HttpRequest object
+        parameter: The parameter name to look up
+        fallback_session_parameter: Optional alternate session key to try (e.g. 'auth_client_id' for 'client_id')
+
+    Returns:
+        The first truthy value found, or None
+    """
+    result = request.GET.get(parameter) or request.session.get(parameter)
+    if not result and fallback_session_parameter:
+        result = request.session.get(fallback_session_parameter)
+    return result or request.session.get('oauth_params', {}).get(parameter)
+
+
 def remove_application_user_pair_tokens_data_access(
     application, user, delete_data_access_grant: bool, delete_access_tokens: bool
 ):

apps/dot_ext/v3/urls.py

@@ -19,6 +19,7 @@
     path('revoke_token/', waffle_switch('v3_endpoints')(views.RevokeTokenView.as_view()), name='revoke-token-v3'),
     path('revoke/', waffle_switch('v3_endpoints')(views.RevokeView.as_view()), name='revoke-v3'),
     path('introspect/', waffle_switch('v3_endpoints')(views.IntrospectTokenView.as_view()), name='introspect-v3'),
+    path('permission-logout', views.PermissionScreenLogoutView.as_view(), name='permission-logout'),
 ]
 
 

apps/dot_ext/views/__init__.py

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

apps/dot_ext/views/authorization.py

@@ -9,22 +9,24 @@
 from functools import wraps
 from http import HTTPStatus
 from typing import Any
-from urllib.parse import parse_qs, urlparse
+from urllib.parse import parse_qs, urlencode, urlparse
 
 import jwt
 import waffle
 from django.conf import settings
-from django.contrib.auth import get_user_model
+from django.contrib.auth import get_user_model, logout
 from django.contrib.auth.models import User
 from django.contrib.auth.views import redirect_to_login
 from django.core.cache import cache
 from django.core.exceptions import ObjectDoesNotExist
 from django.db.models import Q
 from django.http import HttpRequest, JsonResponse
 from django.http.response import HttpResponse, HttpResponseBadRequest
+from django.shortcuts import redirect
 from django.template.response import TemplateResponse
 from django.urls import reverse
 from django.utils.decorators import method_decorator
+from django.views import View
 from django.views.decorators.csrf import csrf_exempt
 from django.views.decorators.debug import sensitive_post_parameters
 from fhir.resources.R4B.address import Address
@@ -69,6 +71,7 @@
     AUDIT_EVENT_SEARCH_SCOPE,
     CLIENT_CREDENTIALS,
     CLIENT_CREDENTIALS_ACCEPTED_JWT_ALGORITHMS,
+    CODE_CHALLENGE_METHOD_S256,
     HHS_SERVER_LOGNAME_FMT,
     OPENID_SCOPE,
     REFRESH_TOKEN,
@@ -110,6 +113,7 @@
     check_auth_tracking_and_create_access_token_extension,
     check_can_token_scope_for_audit_event_scopes,
     get_api_version_number_from_url,
+    get_oauth_param,
     json_response_from_oauth2_error,
     remove_application_user_pair_tokens_data_access,
     validate_app_is_active,
@@ -267,6 +271,19 @@ def get_context_data(self, **kwargs):
         context['permission_end_date_text'] = self.application.access_end_date_text()
         context['permission_end_date'] = self.application.access_end_date()
 
+        params = [
+            'client_id',
+            'redirect_uri',
+            'response_type',
+            'scope',
+            'state',
+            'code_challenge',
+            'code_challenge_method',
+        ]
+        oauth_params = {}
+        for param in params:
+            oauth_params[param] = self.request.GET.get(param)
+        self.request.session['oauth_params'] = oauth_params
         if 'form' in context and self.version == Versions.V3:
             # By setting this to matching_scopes instead of application_scopes, we ensure that the scopes
             # for the access token are in the intersection of what the application is allowed to have and
@@ -1138,6 +1155,7 @@ def _retrieve_prior_include_samhsa_and_part_d_eob_only_values(
     def post(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
         version = get_api_version_number_from_url(self.request.path_info)
         grant_type = request.POST.get('grant_type')
+
         try:
             # If it is not version 3, we don't need to check that the application is in the v3_early_adopter flag,
             # just continue with standard validation.
@@ -1431,3 +1449,47 @@ def post(self, request, *args, **kwargs):
             return json_response_from_oauth2_error(error)
 
         return super(IntrospectTokenView, self).post(request, args, kwargs)
+
+
+class PermissionScreenLogoutView(View):
+    def post(self, request, *args, **kwargs):
+        # Save version before logout clears the session
+        version = request.session.get('version', None)
+
+        # Save the original OAuth params before logout
+        oauth_params = {
+            'client_id': get_oauth_param(request, 'client_id', 'auth_client_id'),
+            'redirect_uri': get_oauth_param(request, 'redirect_uri'),
+            'response_type': get_oauth_param(request, 'response_type'),
+            'state': get_oauth_param(request, 'state'),
+            'code_challenge_method': CODE_CHALLENGE_METHOD_S256,
+            'code_challenge': request.session.get('oauth_params', {}).get('code_challenge'),
+            'scope': request.session.get('oauth_params', {}).get('scope'),
+            'code_verifier': request.session.get('code_verifier'),
+        }
+
+        # Remove None values
+        oauth_params = {k: v for k, v in oauth_params.items() if v is not None}
+
+        # Clear token and log out
+        request.session.pop('token', None)
+        logout(request)
+
+        # this is to avoid a corrupted session warning - though I am still seeing that warning
+        request.session.cycle_key()
+
+        # Restore version after logout
+        if version is not None:
+            request.session['version'] = version
+
+            if 'testclient' in oauth_params.get('redirect_uri', ''):
+                # this is the key that the testclient looks for
+                request.session['api_ver'] = version
+                request.session['code_verifier'] = oauth_params.get('code_verifier')
+                request.session['client_id'] = oauth_params.get('client_id')
+
+        # Rebuild the authorize URL with original params
+        base_url = request.build_absolute_uri('/').rstrip('/')
+        authorize_url = f'{base_url}/v3/o/authorize?{urlencode(oauth_params)}'
+
+        return redirect(authorize_url)

apps/testclient/tests.py

@@ -1,5 +1,6 @@
 import json
 import os
+from typing import List
 
 import pytest
 from django.core.management import call_command
@@ -467,3 +468,61 @@ def test_get_oidc_discovery(self):
         response = self.client.get(reverse('openid-configuration'))
         self.assertEqual(response.status_code, 200)
         self.assertContains(response, 'userinfo_endpoint')
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    'post_switch_account_link, expected_keys, expected_excluded_keys',
+    [
+        (
+            True,
+            [
+                'resource_uri',
+                'coverage_uri',
+                'authorization_uri',
+                'token_uri',
+                'userinfo_uri',
+                'patient_uri',
+                'eob_uri',
+                'coverage_uri',
+                'digital_insurance_card_uri',
+            ],
+            ['code_challenge_method', 'code_verifier', 'code_challenge', 'state', 'redirect_uri'],
+        ),
+        (
+            False,
+            [
+                'resource_uri',
+                'coverage_uri',
+                'authorization_uri',
+                'token_uri',
+                'userinfo_uri',
+                'patient_uri',
+                'eob_uri',
+                'coverage_uri',
+                'digital_insurance_card_uri',
+                'code_challenge_method',
+                'code_verifier',
+                'code_challenge',
+                'state',
+                'redirect_uri',
+            ],
+            [],
+        ),
+    ],
+)
+def test_setup_testclient_http_response_is_post_switch_account(
+    post_switch_account_link: bool,
+    expected_keys: List[str],
+    expected_excluded_keys: List[str],
+) -> None:
+    # TODO: replace with fixtures once 4964/4965 branch is merged
+    call_command('create_blue_button_scopes')
+    call_command('create_test_user_and_application')
+    result = setup_testclient_http_response(version=3, post_switch_account_link=post_switch_account_link)
+
+    for key in expected_excluded_keys:
+        assert key not in result.keys()
+
+    for key in expected_keys:
+        assert key in result.keys()

apps/testclient/utils.py

@@ -36,7 +36,9 @@ def _start_url_with_http_or_https(host: str) -> str:
 
 
 def setup_testclient_http_response(
-    include_client_secret: bool = True, version: int = Versions.NOT_AN_API_VERSION
+    include_client_secret: bool = True,
+    version: int = Versions.NOT_AN_API_VERSION,
+    post_switch_account_link: bool = True,
 ) -> OrderedDict:
     """Prepare testclient response environment
 
@@ -48,6 +50,8 @@ def setup_testclient_http_response(
     Args:
         include_client_secret (bool) : What it says.
         version (Version): Which version of the API are we navigating through.
+        post_switch_account_link (bool): Flag for if we are hitting this function after the Switch account link
+        on the v3 permissions screen was clicked. If so, we only want to set the resource uri values
 
     Returns:
         OrderedDict: A dictionary used to prepare/extend the Django session.
@@ -66,14 +70,15 @@ def setup_testclient_http_response(
     host = _start_url_with_http_or_https(host)
 
     response['resource_uri'] = host
-    response['redirect_uri'] = '{}{}'.format(host, TESTCLIENT_REDIRECT_URI)
     response['coverage_uri'] = '{}/v{}/fhir/Coverage/'.format(host, version)
 
-    auth_data = __generate_auth_data()
-    response['code_challenge_method'] = 'S256'
-    response['code_verifier'] = auth_data['code_verifier']
-    response['code_challenge'] = auth_data['code_challenge']
-    response['state'] = auth_data['state']
+    if not post_switch_account_link:
+        auth_data = __generate_auth_data()
+        response['code_challenge_method'] = 'S256'
+        response['code_verifier'] = auth_data['code_verifier']
+        response['code_challenge'] = auth_data['code_challenge']
+        response['state'] = auth_data['state']
+        response['redirect_uri'] = '{}{}'.format(host, TESTCLIENT_REDIRECT_URI)
 
     response['authorization_uri'] = f'{host}/v{version}/o/authorize/'
     response['token_uri'] = f'{host}/v{version}/o/token/'

apps/testclient/views.py

@@ -17,6 +17,7 @@
 
 from apps.constants import HHS_SERVER_LOGNAME_FMT
 from apps.dot_ext.loggers import cleanup_session_auth_flow_trace
+from apps.dot_ext.utils import get_oauth_param
 from apps.fhir.bluebutton.views.home import fhir_conformance_v1, fhir_conformance_v2, fhir_conformance_v3
 from apps.testclient.constants import HOME_PAGE, RESULTS_PAGE, EndpointUrl, ResponseErrors
 from apps.testclient.utils import (
@@ -101,8 +102,14 @@ def _convert_response_string_to_json(json_response: str) -> Dict[str, object]:
 
 
 def _get_oauth2_session_with_redirect(request: HttpRequest) -> OAuth2Session:
-    client_id = request.session['client_id']
-    redirect_uri = request.session['redirect_uri']
+    client_id = get_oauth_param(request, 'client_id', 'auth_client_id')
+    redirect_uri = get_oauth_param(request, 'redirect_uri')
+    if not redirect_uri:
+        # Default to /testclient/callback if there is no redirect_uri attribute. Confirmed that HOSTNAME_URL
+        # has no trailing slash in any deployed env.
+        host = _start_url_with_http_or_https(settings.HOSTNAME_URL)
+        redirect_uri = host + '/testclient/callback'
+
     return OAuth2Session(client_id, redirect_uri=redirect_uri)
 
 
@@ -142,7 +149,7 @@ def _is_synthetic_patient_id(patient_id: str) -> bool:
 def callback(request: HttpRequest):
     """Called when returning from authorizing as a beneficiary.
 
-    When using the text client, users can authorize as a beneficiary as part of the auth workflow.
+    When using the test client, users can authorize as a beneficiary as part of the auth workflow.
 
     https://bluebutton.cms.gov/developers/#authorization:~:text=Click-,Authorize%20as%20a%20Beneficiary,-.
 
@@ -177,6 +184,7 @@ def callback(request: HttpRequest):
     # However, the default is `v0`, which is an invalid version. This keeps it of the
     # same type as valid values, but does not allow us to proceed if something has broken.
     version = request.session.get('api_ver', Versions.NOT_AN_API_VERSION)
+
     match version:
         case Versions.V1:
             token_uri += reverse('oauth2_provider:token')
@@ -196,6 +204,10 @@ def callback(request: HttpRequest):
         # Perhaps oas.fetch_token fails (and raises a `MissingTokenError`) if the code verifier
         # cannot be pulled from the session.
         cv = request.session.get('code_verifier', '')
+        if not cv:
+            session_cache = getattr(request.session, '_session_cache', {})
+            cv = session_cache.get('code_verifier', '')
+
         token = oas.fetch_token(
             token_uri, client_secret=get_client_secret(), authorization_response=auth_uri, code_verifier=cv
         )
@@ -226,7 +238,12 @@ def callback(request: HttpRequest):
 
     # We are guaranteed at this point that the patient_id is not None, and it is a synthetic user id.
     request.session['token'] = token
-    userinfo_uri = request.session['userinfo_uri']
+
+    userinfo_uri = request.session.get('userinfo_uri')
+    # If there is no userinfo_uri in the session, that means the Switch account link on the v3 permissions screen
+    # was clicked, and we need to reset certain uri values, so the testclient works correctly
+    if not userinfo_uri:
+        request.session.update(setup_testclient_http_response(version=version, post_switch_account_link=True))
 
     try:
         userinfo = oas.get(userinfo_uri).json()
@@ -313,8 +330,7 @@ def _link_session_or_version_is_bad(session, version):
 
 
 def _authorize_link(request: HttpRequest, version=Versions.NOT_AN_API_VERSION):
-    request.session.update(setup_testclient_http_response(version=version))
-
+    request.session.update(setup_testclient_http_response(version=version, post_switch_account_link=False))
     oas = _get_oauth2_session_with_redirect(request)
 
     # We need scopes in V3.