stages/redirect: fix query parameter when redirecting to flow (#12750)

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

Author: BeryJu

authentik/stages/redirect/stage.py

@@ -20,7 +20,7 @@
     FlowPlanner,
 )
 from authentik.flows.stage import ChallengeStageView
-from authentik.flows.views.executor import SESSION_KEY_PLAN, InvalidStageError
+from authentik.flows.views.executor import SESSION_KEY_GET, SESSION_KEY_PLAN, InvalidStageError
 from authentik.lib.utils.urls import reverse_with_qs
 from authentik.stages.redirect.models import RedirectMode, RedirectStage
 
@@ -72,7 +72,9 @@ def switch_flow_with_context(self, flow: Flow, keep_context=True) -> str:
         self.request.session[SESSION_KEY_PLAN] = plan
         kwargs = self.executor.kwargs
         kwargs.update({"flow_slug": flow.slug})
-        return reverse_with_qs("authentik_core:if-flow", self.request.GET, kwargs=kwargs)
+        return reverse_with_qs(
+            "authentik_core:if-flow", self.request.session[SESSION_KEY_GET], kwargs=kwargs
+        )
 
     def get_challenge(self, *args, **kwargs) -> Challenge:
         """Get the redirect target. Prioritize `redirect_stage_target` if present."""

authentik/stages/redirect/tests.py

@@ -1,5 +1,7 @@
 """Test Redirect stage"""
 
+from urllib.parse import urlencode
+
 from django.urls.base import reverse
 from rest_framework.exceptions import ValidationError
 
@@ -58,6 +60,23 @@ def test_flow(self):
             response, reverse("authentik_core:if-flow", kwargs={"flow_slug": self.target_flow.slug})
         )
 
+    def test_flow_query(self):
+        self.stage.mode = RedirectMode.FLOW
+        self.stage.save()
+
+        response = self.client.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
+            + "?"
+            + urlencode({"query": urlencode({"test": "foo"})})
+        )
+
+        self.assertStageRedirects(
+            response,
+            reverse("authentik_core:if-flow", kwargs={"flow_slug": self.target_flow.slug})
+            + "?"
+            + urlencode({"test": "foo"}),
+        )
+
     def test_override_static(self):
         policy = ExpressionPolicy.objects.create(
             name=generate_id(),