feat(idp): handle prompt=login

Author: pennersr

allauth/core/internal/httpkit.py

@@ -48,6 +48,25 @@ def redirect(to):
         return shortcuts.redirect(f"/{to}")
 
 
+def del_query_params(url: str, *params: str) -> str:
+    parsed_url = urlparse(url)
+    query_params = parse_qs(parsed_url.query, keep_blank_values=True)
+    for param in params:
+        query_params.pop(param, None)
+    encoded_query = urlencode(query_params, doseq=True)
+    new_url = urlunparse(
+        (
+            parsed_url.scheme,
+            parsed_url.netloc,
+            parsed_url.path,
+            parsed_url.params,
+            encoded_query,
+            parsed_url.fragment,
+        )
+    )
+    return new_url
+
+
 def add_query_params(url: str, params: dict) -> str:
     parsed_url = urlparse(url)
     query_params = parse_qs(parsed_url.query)

allauth/idp/oidc/tests/test_authorization.py

@@ -366,3 +366,37 @@ def test_authorization_code_flow_with_pkce(
         "scope",
         "refresh_token",
     }
+
+
+@pytest.mark.parametrize("client_fixture", ["auth_client", "client"])
+@pytest.mark.parametrize(
+    "prompt,next_prompt", [("login", None), ("login consent", "consent")]
+)
+def test_redirect_to_login_with_prompt_login(
+    request, client_fixture, oidc_client, prompt, next_prompt
+):
+    client = request.getfixturevalue(client_fixture)
+    redirect_uri = oidc_client.get_redirect_uris()[0]
+    resp = client.get(
+        reverse("idp:oidc:authorization")
+        + "?"
+        + urlencode(
+            {
+                "client_id": oidc_client.id,
+                "response_type": "code",
+                "redirect_uri": redirect_uri,
+                "prompt": prompt,
+            }
+        )
+    )
+    assert resp.status_code == HTTPStatus.FOUND
+    parts = urlparse(resp["location"])
+    assert parts.path == reverse(
+        "account_login" if client_fixture == "client" else "account_reauthenticate"
+    )
+    params = parse_qs(parts.query)
+    params = parse_qs(urlparse(params["next"][0]).query)
+    if next_prompt:
+        assert params["prompt"][0] == next_prompt
+    else:
+        assert "prompt" not in params

allauth/idp/oidc/views.py

@@ -1,10 +1,13 @@
-from typing import List
+from typing import List, Optional
 
+from django.contrib.auth import REDIRECT_FIELD_NAME
 from django.contrib.auth.decorators import login_required
 from django.contrib.sites.shortcuts import get_current_site
 from django.core.exceptions import PermissionDenied
 from django.core.signing import BadSignature, Signer
 from django.http import (
+    HttpRequest,
+    HttpResponse,
     HttpResponseForbidden,
     HttpResponseRedirect,
     JsonResponse,
@@ -23,7 +26,7 @@
 from allauth.account import app_settings as account_settings
 from allauth.account.internal.decorators import login_not_required
 from allauth.core.internal import jwkkit
-from allauth.core.internal.httpkit import add_query_params
+from allauth.core.internal.httpkit import add_query_params, del_query_params
 from allauth.idp.oidc import app_settings
 from allauth.idp.oidc.adapter import get_adapter
 from allauth.idp.oidc.forms import AuthorizationForm
@@ -124,10 +127,34 @@ def post(self, request, *args, **kwargs):
             return self._respond_with_access_denied()
         return super().post(request, *args, **kwargs)
 
-    def _login_required(self, request):
+    def _login_required(self, request) -> Optional[HttpResponse]:
+        prompts = []
+        prompt = request.GET.get("prompt")
+        if prompt:
+            prompts = prompt.split()
+        if "login" in prompts:
+            return self._handle_login_prompt(request, prompts)
         if request.user.is_authenticated:
             return None
-        return login_required()(None)(request)
+        return login_required()(None)(request)  # type:ignore[misc,type-var]
+
+    def _handle_login_prompt(
+        self, request: HttpRequest, prompts: List[str]
+    ) -> HttpResponse:
+        prompts.remove("login")
+        next_url = request.get_full_path()
+        if prompts:
+            next_url = add_query_params(next_url, {"prompt": " ".join(prompts)})
+        else:
+            next_url = del_query_params(next_url, "prompt")
+        params = {}
+        params[REDIRECT_FIELD_NAME] = next_url
+        path = reverse(
+            "account_reauthenticate"
+            if request.user.is_authenticated
+            else "account_login"
+        )
+        return HttpResponseRedirect(add_query_params(path, params))
 
     def _skip_consent(self):
         scopes = self._request_info["request"].scopes