stages/email: add wrapper view to accept queryargs and redirects to flow if

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

authentik/stages/email/apps.py

@@ -16,6 +16,7 @@ class AuthentikStageEmailConfig(AppConfig):
     name = "authentik.stages.email"
     label = "authentik_stages_email"
     verbose_name = "authentik Stages.Email"
+    mountpoint = "stages/email/"
 
     def ready(self):
         import_module("authentik.stages.email.tasks")

authentik/stages/email/stage.py

@@ -45,7 +45,7 @@ class EmailStageView(ChallengeStageView):
     def get_full_url(self, **kwargs) -> str:
         """Get full URL to be used in template"""
         base_url = reverse(
-            "authentik_core:if-flow",
+            "authentik_stages_email:from-email",
             kwargs={"flow_slug": self.executor.flow.slug},
         )
         relative_url = f"{base_url}?{urlencode(kwargs)}"

authentik/stages/email/tests/test_stage.py

@@ -7,7 +7,6 @@ from django.urls import reverse
 from django.utils.encoding import force_str
 
 from authentik.core.models import Token, User
-from authentik.flows.challenge import ChallengeTypes
 from authentik.flows.markers import StageMarker
 from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
 from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
@@ -110,7 +109,7 @@ class TestEmailStage(TestCase):
         with patch("authentik.flows.views.FlowExecutorView.cancel", MagicMock()):
             # Call the executor shell to preseed the session
             url = reverse(
-                "authentik_core:if-flow",
+                "authentik_stages_email:from-email",
                 kwargs={"flow_slug": self.flow.slug},
             )
             token = Token.objects.get(user=self.user)
@@ -127,12 +126,7 @@ class TestEmailStage(TestCase):
             self.assertEqual(response.status_code, 200)
             self.assertJSONEqual(
                 force_str(response.content),
-                {
-                    "component": "ak-stage-access-denied",
-                    "error_message": None,
-                    "title": "",
-                    "type": ChallengeTypes.native.value,
-                },
+                {"to": reverse("authentik_core:root-redirect"), "type": "redirect"},
             )
 
             session = self.client.session

authentik/stages/email/urls.py

@@ -0,0 +1,8 @@
+"""Email stage url patterns"""
+from django.urls import path
+
+from authentik.stages.email.views import FromEmailView
+
+urlpatterns = [
+    path("from-email/<slug:flow_slug>/", FromEmailView.as_view(), name="from-email"),
+]

authentik/stages/email/views.py

@@ -0,0 +1,31 @@
+"""Email stage views"""
+from django.http.request import HttpRequest
+from django.http.response import HttpResponse, HttpResponseBadRequest
+from django.shortcuts import get_object_or_404, redirect
+from django.views import View
+from structlog.stdlib import get_logger
+
+from authentik.core.models import Token
+from authentik.flows.views import SESSION_KEY_GET
+from authentik.stages.email.stage import QS_KEY_TOKEN
+
+LOGGER = get_logger()
+
+
+class FromEmailView(View):
+    """FromEmailView, this view is linked in the email confirmation link.
+    It is required because the flow executor does not pass query args to the API,
+    so this view gets called, checks for a Querystring and updates the plan
+    if everything is valid."""
+
+    def get(self, request: HttpRequest, flow_slug: str) -> HttpResponse:
+        """Check for ?token param and validate it."""
+        if QS_KEY_TOKEN not in request.GET:
+            LOGGER.debug("No token set")
+            return HttpResponseBadRequest()
+        # Lookup token here to quickly fail for invalid input
+        get_object_or_404(Token, pk=request.GET[QS_KEY_TOKEN])
+        if SESSION_KEY_GET not in request.session:
+            request.session[SESSION_KEY_GET] = {}
+        request.session[SESSION_KEY_GET][QS_KEY_TOKEN] = request.GET[QS_KEY_TOKEN]
+        return redirect("authentik_core:if-flow", flow_slug=flow_slug)