HP-2491 | return kc_action_status from Keycloak on authentication

tunnistamo/endpoints.py

@@ -33,6 +33,22 @@ def is_tunnistamo_session_active(self):
 
 
 class TunnistamoAuthorizeEndpoint(TunnistamoSessionEndpointMixin, AuthorizeEndpoint):
+
+    def __init__(self, request):
+        super().__init__(request)
+
+        if kc_value := self.request.session.get("kc_action_status"):
+            self.params["kc_action_status"] = kc_value
+
+    def create_response_uri(self):
+        """Add the status of the Keycloak action the callback url."""
+        uri = super().create_response_uri()
+
+        if kc_value := self.params.get("kc_action_status"):
+            uri = add_params_to_url(uri, {"kc_action_status": kc_value})
+
+        return uri
+
     def _get_tunnistamo_session(self):
         tunnistamo_session_id = self.request.session.get("tunnistamo_session_id")
         if not tunnistamo_session_id:

tunnistamo/settings.py

@@ -514,6 +514,9 @@
 
     # Add loa to Tunnistamo Session
     'users.pipeline.add_loa_to_tunnistamo_session',
+
+    # Save status of the Keycloak action to session
+    'users.pipeline.save_kc_action_status_to_session',
 )
 
 ALLOW_DUPLICATE_EMAILS = env("ALLOW_DUPLICATE_EMAILS")

tunnistamo/tests/test_tunnistamo_session_checks.py

@@ -58,6 +58,38 @@ def test_authorize_endpoint(user, response_type, ended):
             assert 'code=' in response['Location']
 
 
+@pytest.mark.django_db
+def test_authorize_endpoint_redirect_url_contains_kc_action_status_parameter(user):
+    django_test_client = DjangoTestClient()
+    django_test_client.force_login(user)
+    oidc_client = create_oidc_clients_and_api()
+
+    authorize_url = reverse("authorize")
+
+    authorize_request_data = {
+        "client_id": oidc_client.client_id,
+        "redirect_uri": oidc_client.redirect_uris[0],
+        "scope": "openid profile",
+        "response_type": "code",
+        "response_mode": "form_post",
+        "nonce": get_random_string(12),
+    }
+
+    # Add the expected parameters to the session which would be added in the social
+    # auth pipeline.
+    session = django_test_client.session
+    session.update({"kc_action_status": "success"})
+    session.save()
+
+    response = django_test_client.get(
+        authorize_url, authorize_request_data, follow=False
+    )
+
+    assert response.status_code == 302, response["Location"]
+
+    assert "kc_action_status=success" in response["Location"]
+
+
 @pytest.mark.django_db
 @pytest.mark.parametrize('response_type', [
     'code',

users/pipeline.py

@@ -2,7 +2,7 @@
 import uuid
 
 from django.conf import settings
-from django.contrib.auth import get_user_model, logout
+from django.contrib.auth import REDIRECT_FIELD_NAME, get_user_model, logout
 from django.shortcuts import render
 from django.urls import reverse
 from helusers.utils import uuid_to_username
@@ -256,10 +256,10 @@
                      ' with a different user. Log out and fail the login.')
         # Save the existing next value from the session and add it back after log out
         # so that InterruptedSocialAuthMiddleware can redirect back to the OIDC client
-        next_url = strategy.session.get('next')
+        next_url = strategy.session_get(REDIRECT_FIELD_NAME)
         logout(request)
         if next_url:
-            strategy.session_set('next', next_url)
+            strategy.session_set(REDIRECT_FIELD_NAME, next_url)
 
         raise AuthFailed(backend, 'Duplicate login')
 
@@ -337,3 +337,18 @@
         return {
             'user': existing_social_auth.user,
         }
+
+
+def save_kc_action_status_to_session(backend, strategy, *args, **kwargs):
+    """When a Keycloak action status is present save it to the session."""
+    if isinstance(backend, (HelsinkiTunnus, HelsinkiTunnistus)) and (
+        kc_action_status := strategy.request_data().get("kc_action_status")
+    ):
+        logger.debug(
+            f"kc_action_status {kc_action_status} in pipeline. Backend: {backend.name}."
+        )
+        strategy.session_set("kc_action_status", kc_action_status)
+    else:
+        strategy.session_pop("kc_action_status")
+
+    return

users/tests/test_pipeline.py

@@ -1,13 +1,16 @@
+from urllib.parse import urlencode
+
 import pytest
 from django.contrib import auth
+from django.contrib.sessions.middleware import SessionMiddleware
 from django.urls import reverse
 from django.utils.crypto import get_random_string
 from social_core.exceptions import AuthFailed
 from social_django.models import UserSocialAuth
 from social_django.utils import load_backend, load_strategy
 
 from tunnistamo.tests.conftest import create_rsa_key, reload_social_django_utils
-from users.pipeline import association_by_keycloak_uuid, update_ad_groups
+from users.pipeline import association_by_keycloak_uuid, save_kc_action_status_to_session, update_ad_groups
 from users.tests.conftest import DummyFixedOidcBackend
 
 
@@ -103,3 +106,31 @@ def test_association_by_keycloak_uuid(login_backend, existing_backend, user, exp
         assert pipeline_data == {"user": user}
     else:
         assert pipeline_data is None
+
+
+@pytest.mark.parametrize(
+    "backend_name,expected",
+    (
+        ("heltunnistussuomifi", "success"),
+        ("helsinki_tunnus", "success"),
+        ("helsinki_adfs", None),
+    ),
+    ids=repr,
+)
+def test_save_kc_action_status_to_session(backend_name, expected, rf):
+    uri = "{}?{}".format(
+        reverse("social:complete", kwargs={"backend": backend_name}),
+        urlencode(
+            {
+                "kc_action_status": "success",
+            }
+        ),
+    )
+    request = rf.get(uri)
+    SessionMiddleware(get_response=lambda response: response).process_request(request)
+    strategy = load_strategy(request)
+    backend = load_backend(strategy, backend_name, uri)
+
+    save_kc_action_status_to_session(backend, strategy)
+
+    assert strategy.session_get("kc_action_status") == expected