stages/email: fix email scanner voiding token (#14325)

* stages/email: fix email scanner voiding flow token

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

* misc

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

* improve consent stage error handling and testing

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

* draw the rest of the owl

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

* add e2e test

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

* fix tests

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

* fix

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

* idk why this is broken now?

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

* fix other e2e test

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

* fix the other test too

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

---------

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

authentik/core/api/users.py

@@ -84,6 +84,7 @@ from authentik.flows.views.executor import QS_KEY_TOKEN
 from authentik.lib.avatars import get_avatar
 from authentik.rbac.decorators import permission_required
 from authentik.rbac.models import get_permission_choices
+from authentik.stages.email.flow import pickle_flow_token_for_email
 from authentik.stages.email.models import EmailStage
 from authentik.stages.email.tasks import send_mails
 from authentik.stages.email.utils import TemplateEmailMessage
@@ -451,7 +452,7 @@ class UserViewSet(UsedByMixin, ModelViewSet):
     def list(self, request, *args, **kwargs):
         return super().list(request, *args, **kwargs)
 
-    def _create_recovery_link(self) -> tuple[str, Token]:
+    def _create_recovery_link(self, for_email=False) -> tuple[str, Token]:
         """Create a recovery link (when the current brand has a recovery flow set),
         that can either be shown to an admin or sent to the user directly"""
         brand: Brand = self.request._request.brand
@@ -473,12 +474,16 @@ class UserViewSet(UsedByMixin, ModelViewSet):
             raise ValidationError(
                 {"non_field_errors": "Recovery flow not applicable to user"}
             ) from None
+        _plan = FlowToken.pickle(plan)
+        if for_email:
+            _plan = pickle_flow_token_for_email(plan)
         token, __ = FlowToken.objects.update_or_create(
             identifier=f"{user.uid}-password-reset",
             defaults={
                 "user": user,
                 "flow": flow,
-                "_plan": FlowToken.pickle(plan),
+                "_plan": _plan,
+                "revoke_on_execution": not for_email,
             },
         )
         querystring = urlencode({QS_KEY_TOKEN: token.key})
@@ -648,7 +653,7 @@ class UserViewSet(UsedByMixin, ModelViewSet):
         if for_user.email == "":
             LOGGER.debug("User doesn't have an email address")
             raise ValidationError({"non_field_errors": "User does not have an email address set."})
-        link, token = self._create_recovery_link()
+        link, token = self._create_recovery_link(for_email=True)
         # Lookup the email stage to assure the current user can access it
         stages = get_objects_for_user(
             request.user, "authentik_stages_email.view_emailstage"

authentik/flows/migrations/0028_flowtoken_revoke_on_execution.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.1.9 on 2025-05-27 12:52
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_flows", "0027_auto_20231028_1424"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="flowtoken",
+            name="revoke_on_execution",
+            field=models.BooleanField(default=True),
+        ),
+    ]

authentik/flows/models.py

@@ -303,9 +303,10 @@ class FlowToken(Token):
 
     flow = models.ForeignKey(Flow, on_delete=models.CASCADE)
     _plan = models.TextField()
+    revoke_on_execution = models.BooleanField(default=True)
 
     @staticmethod
-    def pickle(plan) -> str:
+    def pickle(plan: "FlowPlan") -> str:
         """Pickle into string"""
         data = dumps(plan)
         return b64encode(data).decode()

authentik/flows/stage.py

@@ -99,9 +99,10 @@ class ChallengeStageView(StageView):
             self.logger.debug("Got StageInvalidException", exc=exc)
             return self.executor.stage_invalid()
         if not challenge.is_valid():
-            self.logger.warning(
+            self.logger.error(
                 "f(ch): Invalid challenge",
                 errors=challenge.errors,
+                challenge=challenge.data,
             )
         return HttpChallengeResponse(challenge)
 

authentik/flows/views/executor.py

@@ -146,7 +146,8 @@ class FlowExecutorView(APIView):
         except (AttributeError, EOFError, ImportError, IndexError) as exc:
             LOGGER.warning("f(exec): Failed to restore token plan", exc=exc)
         finally:
-            token.delete()
+            if token.revoke_on_execution:
+                token.delete()
         if not isinstance(plan, FlowPlan):
             return None
         plan.context[PLAN_CONTEXT_IS_RESTORED] = token

authentik/sources/saml/views.py

@@ -9,6 +9,7 @@ from django.http.response import HttpResponseBadRequest
 from django.shortcuts import get_object_or_404, redirect
 from django.utils.decorators import method_decorator
 from django.utils.http import urlencode
+from django.utils.translation import gettext as _
 from django.views import View
 from django.views.decorators.csrf import csrf_exempt
 from structlog.stdlib import get_logger
@@ -128,7 +129,9 @@ class InitiateView(View):
         # otherwise we default to POST_AUTO, with direct redirect
         if source.binding_type == SAMLBindingTypes.POST:
             injected_stages.append(in_memory_stage(ConsentStageView))
-            plan_kwargs[PLAN_CONTEXT_CONSENT_HEADER] = f"Continue to {source.name}"
+            plan_kwargs[PLAN_CONTEXT_CONSENT_HEADER] = _(
+                "Continue to {source_name}".format(source_name=source.name)
+            )
         injected_stages.append(in_memory_stage(AutosubmitStageView))
         return self.handle_login_flow(
             source,

authentik/stages/consent/stage.py

@@ -4,6 +4,8 @@ from uuid import uuid4
 
 from django.http import HttpRequest, HttpResponse
 from django.utils.timezone import now
+from django.utils.translation import gettext as _
+from rest_framework.exceptions import ValidationError
 from rest_framework.fields import CharField
 
 from authentik.core.api.utils import PassiveSerializer
@@ -47,6 +49,11 @@ class ConsentChallengeResponse(ChallengeResponse):
     component = CharField(default="ak-stage-consent")
     token = CharField(required=True)
 
+    def validate_token(self, token: str):
+        if token != self.stage.executor.request.session[SESSION_KEY_CONSENT_TOKEN]:
+            raise ValidationError(_("Invalid consent token, re-showing prompt"))
+        return token
+
 
 class ConsentStageView(ChallengeStageView):
     """Simple consent checker."""
@@ -120,9 +127,6 @@ class ConsentStageView(ChallengeStageView):
         return super().get(request, *args, **kwargs)
 
     def challenge_valid(self, response: ChallengeResponse) -> HttpResponse:
-        if response.data["token"] != self.request.session[SESSION_KEY_CONSENT_TOKEN]:
-            self.logger.info("Invalid consent token, re-showing prompt")
-            return self.get(self.request)
         if self.should_always_prompt():
             return self.executor.stage_ok()
         current_stage: ConsentStage = self.executor.current_stage

authentik/stages/consent/tests.py

@@ -17,6 +17,7 @@ from authentik.flows.views.executor import SESSION_KEY_PLAN
 from authentik.lib.generators import generate_id
 from authentik.stages.consent.models import ConsentMode, ConsentStage, UserConsent
 from authentik.stages.consent.stage import (
+    PLAN_CONTEXT_CONSENT_HEADER,
     PLAN_CONTEXT_CONSENT_PERMISSIONS,
     SESSION_KEY_CONSENT_TOKEN,
 )
@@ -33,6 +34,40 @@ class TestConsentStage(FlowTestCase):
             slug=generate_id(),
         )
 
+    def test_mismatched_token(self):
+        """Test incorrect token"""
+        flow = create_test_flow(FlowDesignation.AUTHENTICATION)
+        stage = ConsentStage.objects.create(name=generate_id(), mode=ConsentMode.ALWAYS_REQUIRE)
+        binding = FlowStageBinding.objects.create(target=flow, stage=stage, order=2)
+
+        plan = FlowPlan(flow_pk=flow.pk.hex, bindings=[binding], markers=[StageMarker()])
+        session = self.client.session
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+        response = self.client.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
+        )
+        self.assertEqual(response.status_code, 200)
+
+        session = self.client.session
+        response = self.client.post(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
+            {
+                "token": generate_id(),
+            },
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertStageResponse(
+            response,
+            flow,
+            component="ak-stage-consent",
+            response_errors={
+                "token": [{"string": "Invalid consent token, re-showing prompt", "code": "invalid"}]
+            },
+        )
+        self.assertFalse(UserConsent.objects.filter(user=self.user).exists())
+
     def test_always_required(self):
         """Test always required consent"""
         flow = create_test_flow(FlowDesignation.AUTHENTICATION)
@@ -158,6 +193,7 @@ class TestConsentStage(FlowTestCase):
             context={
                 PLAN_CONTEXT_APPLICATION: self.application,
                 PLAN_CONTEXT_CONSENT_PERMISSIONS: [PermissionDict(id="foo", name="foo-desc")],
+                PLAN_CONTEXT_CONSENT_HEADER: "test header",
             },
         )
         session = self.client.session

authentik/stages/email/flow.py

@@ -0,0 +1,38 @@
+from base64 import b64encode
+from copy import deepcopy
+from pickle import dumps  # nosec
+
+from django.utils.translation import gettext as _
+
+from authentik.flows.models import FlowToken, in_memory_stage
+from authentik.flows.planner import PLAN_CONTEXT_IS_RESTORED, FlowPlan
+from authentik.stages.consent.stage import PLAN_CONTEXT_CONSENT_HEADER, ConsentStageView
+
+
+def pickle_flow_token_for_email(plan: FlowPlan):
+    """Insert a consent stage into the flow plan and pickle it for a FlowToken,
+    to be sent via Email. This is to prevent automated email scanners, which sometimes
+    open links in emails in a full browser from breaking the link."""
+    plan_copy = deepcopy(plan)
+    plan_copy.insert_stage(in_memory_stage(EmailTokenRevocationConsentStageView), index=0)
+    plan_copy.context[PLAN_CONTEXT_CONSENT_HEADER] = _("Continue to confirm this email address.")
+    data = dumps(plan_copy)
+    return b64encode(data).decode()
+
+
+class EmailTokenRevocationConsentStageView(ConsentStageView):
+
+    def get(self, request, *args, **kwargs):
+        token: FlowToken = self.executor.plan.context[PLAN_CONTEXT_IS_RESTORED]
+        try:
+            token.refresh_from_db()
+        except FlowToken.DoesNotExist:
+            return self.executor.stage_invalid(
+                _("Link was already used, please request a new link.")
+            )
+        return super().get(request, *args, **kwargs)
+
+    def challenge_valid(self, response):
+        token: FlowToken = self.executor.plan.context[PLAN_CONTEXT_IS_RESTORED]
+        token.delete()
+        return super().challenge_valid(response)

authentik/stages/email/stage.py

@@ -23,6 +23,7 @@ from authentik.flows.stage import ChallengeStageView
 from authentik.flows.views.executor import QS_KEY_TOKEN, QS_QUERY
 from authentik.lib.utils.errors import exception_to_string
 from authentik.lib.utils.time import timedelta_from_string
+from authentik.stages.email.flow import pickle_flow_token_for_email
 from authentik.stages.email.models import EmailStage
 from authentik.stages.email.tasks import send_mails
 from authentik.stages.email.utils import TemplateEmailMessage
@@ -86,7 +87,8 @@ class EmailStageView(ChallengeStageView):
                 user=pending_user,
                 identifier=identifier,
                 flow=self.executor.flow,
-                _plan=FlowToken.pickle(self.executor.plan),
+                _plan=pickle_flow_token_for_email(self.executor.plan),
+                revoke_on_execution=False,
             )
         token = tokens.first()
         # Check if token is expired and rotate key if so

authentik/stages/email/tests/test_sending.py

@@ -174,5 +174,5 @@ class TestEmailStageSending(FlowTestCase):
                 response = self.client.post(url)
             response = self.client.post(url)
             self.assertEqual(response.status_code, 200)
-            self.assertTrue(len(mail.outbox) >= 1)
+            self.assertGreaterEqual(len(mail.outbox), 1)
             self.assertEqual(mail.outbox[0].subject, "authentik")

authentik/stages/email/tests/test_stage.py

@@ -17,6 +17,7 @@ from authentik.flows.tests import FlowTestCase
 from authentik.flows.views.executor import QS_KEY_TOKEN, SESSION_KEY_PLAN, FlowExecutorView
 from authentik.lib.config import CONFIG
 from authentik.lib.generators import generate_id
+from authentik.stages.consent.stage import SESSION_KEY_CONSENT_TOKEN
 from authentik.stages.email.models import EmailStage
 from authentik.stages.email.stage import PLAN_CONTEXT_EMAIL_OVERRIDE, EmailStageView
 
@@ -160,6 +161,17 @@ class TestEmailStage(FlowTestCase):
                     kwargs={"flow_slug": self.flow.slug},
                 )
             )
+            self.assertStageResponse(response, self.flow, component="ak-stage-consent")
+            response = self.client.post(
+                reverse(
+                    "authentik_api:flow-executor",
+                    kwargs={"flow_slug": self.flow.slug},
+                ),
+                data={
+                    "token": self.client.session[SESSION_KEY_CONSENT_TOKEN],
+                },
+                follow=True,
+            )
 
             self.assertEqual(response.status_code, 200)
             self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
@@ -182,6 +194,7 @@ class TestEmailStage(FlowTestCase):
         # Set flow token user to a different user
         token: FlowToken = FlowToken.objects.get(user=self.user)
         token.user = create_test_admin_user()
+        token.revoke_on_execution = True
         token.save()
 
         with patch("authentik.flows.views.executor.FlowExecutorView.cancel", MagicMock()):

tests/e2e/test_flows_enroll.py

@@ -10,6 +10,7 @@ from authentik.blueprints.tests import apply_blueprint
 from authentik.core.models import User
 from authentik.flows.models import Flow
 from authentik.lib.config import CONFIG
+from authentik.lib.generators import generate_id
 from authentik.stages.identification.models import IdentificationStage
 from tests.e2e.utils import SeleniumTestCase, retry
 
@@ -17,6 +18,10 @@ from tests.e2e.utils import SeleniumTestCase, retry
 class TestFlowsEnroll(SeleniumTestCase):
     """Test Enroll flow"""
 
+    def setUp(self):
+        super().setUp()
+        self.username = generate_id()
+
     @retry()
     @apply_blueprint(
         "default/flow-default-authentication-flow.yaml",
@@ -39,8 +44,8 @@ class TestFlowsEnroll(SeleniumTestCase):
         self.initial_stages()
         sleep(2)
 
-        user = User.objects.get(username="foo")
+        user = User.objects.get(username=self.username)
-        self.assertEqual(user.username, "foo")
+        self.assertEqual(user.username, self.username)
         self.assertEqual(user.name, "some name")
         self.assertEqual(user.email, "foo@bar.baz")
 
@@ -87,7 +92,16 @@ class TestFlowsEnroll(SeleniumTestCase):
 
         sleep(2)
 
-        self.assert_user(User.objects.get(username="foo"))
+        flow_executor = self.get_shadow_root("ak-flow-executor")
+        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
+        consent_stage.find_element(
+            By.CSS_SELECTOR,
+            "[type=submit]",
+        ).click()
+
+        self.wait_for_url(self.if_user_url())
+
+        self.assert_user(User.objects.get(username=self.username))
 
     def initial_stages(self):
         """Fill out initial stages"""
@@ -105,7 +119,7 @@ class TestFlowsEnroll(SeleniumTestCase):
         wait = WebDriverWait(prompt_stage, self.wait_timeout)
 
         wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "input[name=username]")))
-        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=username]").send_keys("foo")
+        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=username]").send_keys(self.username)
         prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
             self.user.username
         )
@@ -124,3 +138,82 @@ class TestFlowsEnroll(SeleniumTestCase):
         prompt_stage.find_element(By.CSS_SELECTOR, "input[name=name]").send_keys("some name")
         prompt_stage.find_element(By.CSS_SELECTOR, "input[name=email]").send_keys("foo@bar.baz")
         prompt_stage.find_element(By.CSS_SELECTOR, ".pf-c-button").click()
+
+    @retry()
+    @apply_blueprint(
+        "default/flow-default-authentication-flow.yaml",
+        "default/flow-default-invalidation-flow.yaml",
+    )
+    @apply_blueprint(
+        "example/flows-enrollment-email-verification.yaml",
+    )
+    @CONFIG.patch("email.port", 1025)
+    def test_enroll_email_pretend_email_scanner(self):
+        """Test enroll with Email verification. Open the email link twice to pretend we have an
+        email scanner that clicks on links"""
+        # Attach enrollment flow to identification stage
+        ident_stage: IdentificationStage = IdentificationStage.objects.get(
+            name="default-authentication-identification"
+        )
+        ident_stage.enrollment_flow = Flow.objects.get(slug="default-enrollment-flow")
+        ident_stage.save()
+
+        self.driver.get(self.live_server_url)
+        self.initial_stages()
+
+        # Email stage
+        flow_executor = self.get_shadow_root("ak-flow-executor")
+        email_stage = self.get_shadow_root("ak-stage-email", flow_executor)
+
+        wait = WebDriverWait(email_stage, self.wait_timeout)
+
+        # Wait for the success message so we know the email is sent
+        wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-form p")))
+
+        # Open Mailpit
+        self.driver.get("http://localhost:8025")
+
+        # Click on first message
+        self.wait.until(ec.presence_of_element_located((By.CLASS_NAME, "message")))
+        self.driver.find_element(By.CLASS_NAME, "message").click()
+        self.driver.switch_to.frame(self.driver.find_element(By.ID, "preview-html"))
+        confirmation_link = self.driver.find_element(By.ID, "confirm").get_attribute("href")
+
+        main_tab = self.driver.current_window_handle
+
+        self.driver.switch_to.new_window("tab")
+        confirm_tab = self.driver.current_window_handle
+
+        # On the new tab, check that we have the confirmation screen
+        self.driver.get(confirmation_link)
+        self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "ak-flow-executor")))
+
+        flow_executor = self.get_shadow_root("ak-flow-executor")
+        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
+
+        self.assertEqual(
+            "Continue to confirm this email address.",
+            consent_stage.find_element(By.CSS_SELECTOR, "#header-text").text,
+        )
+
+        # Back on the main tab, confirm
+        self.driver.switch_to.window(main_tab)
+        self.driver.get(confirmation_link)
+
+        flow_executor = self.get_shadow_root("ak-flow-executor")
+        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
+        consent_stage.find_element(
+            By.CSS_SELECTOR,
+            "[type=submit]",
+        ).click()
+
+        self.wait_for_url(self.if_user_url())
+        sleep(2)
+
+        self.assert_user(User.objects.get(username=self.username))
+
+        self.driver.switch_to.window(confirm_tab)
+        self.driver.refresh()
+        flow_executor = self.get_shadow_root("ak-flow-executor")
+        wait = WebDriverWait(flow_executor, self.wait_timeout)
+        wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "ak-stage-access-denied")))

tests/e2e/test_flows_recovery.py

@@ -84,6 +84,14 @@ class TestFlowsRecovery(SeleniumTestCase):
         self.driver.switch_to.window(self.driver.window_handles[0])
 
         sleep(2)
+
+        flow_executor = self.get_shadow_root("ak-flow-executor")
+        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
+        consent_stage.find_element(
+            By.CSS_SELECTOR,
+            "[type=submit]",
+        ).click()
+
         # We can now enter the new password
         flow_executor = self.get_shadow_root("ak-flow-executor")
         prompt_stage = self.get_shadow_root("ak-stage-prompt", flow_executor)