Merge branch 'main' into pr/15531

authentik/stages/email/api.py

@@ -44,6 +44,8 @@ class EmailStageSerializer(StageSerializer):
             "subject",
             "template",
             "activate_user_on_success",
+            "recovery_max_attempts",
+            "recovery_cache_timeout",
         ]
         extra_kwargs = {"password": {"write_only": True}}
 

authentik/stages/email/migrations/0006_emailstage_recovery_cache_timeout_and_more.py

@@ -0,0 +1,28 @@
+# Generated by Django 5.1.11 on 2025-07-23 11:26
+
+import authentik.lib.utils.time
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_stages_email", "0005_alter_emailstage_token_expiry"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="emailstage",
+            name="recovery_cache_timeout",
+            field=models.TextField(
+                default="minutes=5",
+                help_text="The time window used to count recent account recovery attempts. If the number of attempts exceed recovery_max_attempts within this period, further attempts will be rate-limited. (Format: hours=1;minutes=2;seconds=3).",
+                validators=[authentik.lib.utils.time.timedelta_string_validator],
+            ),
+        ),
+        migrations.AddField(
+            model_name="emailstage",
+            name="recovery_max_attempts",
+            field=models.PositiveIntegerField(default=5),
+        ),
+    ]

authentik/stages/email/models.py

@@ -16,6 +16,8 @@ from authentik.flows.models import Stage
 from authentik.lib.config import CONFIG
 from authentik.lib.utils.time import timedelta_string_validator
 
+EMAIL_RECOVERY_MAX_ATTEMPTS = 5
+
 LOGGER = get_logger()
 
 
@@ -70,6 +72,17 @@ class EmailStage(Stage):
     use_ssl = models.BooleanField(default=False)
     timeout = models.IntegerField(default=10)
     from_address = models.EmailField(default="system@authentik.local")
+    recovery_max_attempts = models.PositiveIntegerField(default=EMAIL_RECOVERY_MAX_ATTEMPTS)
+    recovery_cache_timeout = models.TextField(
+        default="minutes=5",
+        validators=[timedelta_string_validator],
+        help_text=_(
+            "The time window used to count recent account recovery attempts. "
+            "If the number of attempts exceed recovery_max_attempts within "
+            "this period, further attempts will be rate-limited. "
+            "(Format: hours=1;minutes=2;seconds=3)."
+        ),
+    )
 
     activate_user_on_success = models.BooleanField(
         default=False, help_text=_("Activate users upon completion of stage.")

authentik/stages/email/stage.py

@@ -1,9 +1,12 @@
 """authentik multi-stage authentication engine"""
 
-from datetime import timedelta
+import math
+from datetime import UTC, datetime, timedelta
+from hashlib import sha256
 from uuid import uuid4
 
 from django.contrib import messages
+from django.core.cache import cache
 from django.http import HttpRequest, HttpResponse
 from django.http.request import QueryDict
 from django.template.exceptions import TemplateSyntaxError
@@ -27,6 +30,8 @@ from authentik.stages.email.models import EmailStage
 from authentik.stages.email.tasks import send_mails
 from authentik.stages.email.utils import TemplateEmailMessage
 
+EMAIL_RECOVERY_CACHE_KEY = "goauthentik.io/stages/email/stage/"
+
 PLAN_CONTEXT_EMAIL_SENT = "email_sent"
 PLAN_CONTEXT_EMAIL_OVERRIDE = "email"
 
@@ -170,10 +175,66 @@ class EmailStageView(ChallengeStageView):
     def challenge_valid(self, response: ChallengeResponse) -> HttpResponse:
         return super().challenge_invalid(response)
 
+    def _get_cache_key(self) -> str:
+        """Return the cache key used for rate limiting email recovery attempts."""
+        user = self.get_pending_user()
+        user_email_hashed = sha256(user.email.lower().encode("utf-8")).hexdigest()
+        return EMAIL_RECOVERY_CACHE_KEY + user_email_hashed
+
+    def _is_rate_limited(self) -> int | None:
+        """Check whether the email recovery attempt should be rate limited.
+
+        If the request should be rate limited, update the cache and return the
+        remaining time in minutes before the user is allowed to try again.
+        Otherwise, return None."""
+        cache_key = self._get_cache_key()
+        attempts = cache.get(cache_key, [])
+
+        stage = self.executor.current_stage
+        stage.refresh_from_db()
+        max_attempts = stage.recovery_max_attempts
+        cache_timeout_delta = timedelta_from_string(stage.recovery_cache_timeout)
+
+        _now = now()
+        start_window = _now - cache_timeout_delta
+
+        # Convert unix timestamps to datetime objects for comparison
+        recent_attempts_in_window = [
+            datetime.fromtimestamp(attempt, UTC)
+            for attempt in attempts
+            if datetime.fromtimestamp(attempt, UTC) > start_window
+        ]
+
+        if len(recent_attempts_in_window) >= max_attempts:
+            retry_after = (min(recent_attempts_in_window) + cache_timeout_delta) - _now
+            minutes_left = max(1, math.ceil(retry_after.total_seconds() / 60))
+            return minutes_left
+
+        recent_attempts_in_window.append(_now)
+
+        # Convert datetime objects back to unix timestamps to update cache
+        recent_attempts_in_window = [attempt.timestamp() for attempt in recent_attempts_in_window]
+
+        cache.set(
+            cache_key,
+            recent_attempts_in_window,
+            int(cache_timeout_delta.total_seconds()),
+        )
+
+        return None
+
     def challenge_invalid(self, response: ChallengeResponse) -> HttpResponse:
+        if minutes_left := self._is_rate_limited():
+            error = _(
+                "Too many account verification attempts. Please try again after {minutes} minutes."
+            ).format(minutes=minutes_left)
+            messages.error(self.request, error)
+            return super().challenge_invalid(response)
+
         if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:
             messages.error(self.request, _("No pending user."))
             return super().challenge_invalid(response)
+
         self.send_email()
         messages.success(self.request, _("Email Successfully sent."))
         # We can't call stage_ok yet, as we're still waiting

authentik/stages/email/tests/test_stage.py

@@ -1,7 +1,9 @@
 """email tests"""
 
+from hashlib import sha256
 from unittest.mock import MagicMock, PropertyMock, patch
 
+from django.contrib import messages
 from django.core import mail
 from django.core.mail.backends.locmem import EmailBackend
 from django.core.mail.backends.smtp import EmailBackend as SMTPEmailBackend
@@ -9,6 +11,7 @@ from django.test import RequestFactory
 from django.urls import reverse
 from django.utils.http import urlencode
 
+from authentik.brands.models import Brand
 from authentik.core.tests.utils import create_test_admin_user, create_test_flow
 from authentik.flows.markers import StageMarker
 from authentik.flows.models import FlowDesignation, FlowStageBinding, FlowToken
@@ -17,6 +20,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.lib.tests.utils import get_request
 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
@@ -291,3 +295,173 @@ class TestEmailStage(FlowTestCase):
             stage_view.get_full_url(**{QS_KEY_TOKEN: token}),
             f"http://testserver/if/flow/{self.flow.slug}/?foo=bar&flow_token={token}",
         )
+
+    def test_get_cache_key(self):
+        """Test to ensure that the correct cache key is returned."""
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
+        session = self.client.session
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+
+        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
+        request = self.factory.post(url)
+        request.user = self.user
+        request.session = session
+
+        executor = FlowExecutorView(request=request, flow=self.flow)
+        executor.plan = plan
+
+        stage_view = EmailStageView(executor, request=request)
+
+        cache_key = stage_view._get_cache_key()
+
+        expected_hash = sha256(self.user.email.lower().encode("utf-8")).hexdigest()
+        expected_cache_key = "goauthentik.io/stages/email/stage/" + expected_hash
+
+        self.assertEqual(cache_key, expected_cache_key)
+
+    def test_is_rate_limited_returns_none(self):
+        """Test to ensure None is returned if the request shouldn't be rate limited."""
+        self.stage.recovery_max_attempts = 2
+        self.stage.recovery_cache_timeout = "minutes=10"
+        self.stage.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
+        session = self.client.session
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+
+        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
+        request = self.factory.post(url)
+        request.user = self.user
+        request.session = session
+
+        executor = FlowExecutorView(request=request, flow=self.flow)
+        executor.current_stage = self.stage
+        executor.plan = plan
+
+        stage_view = EmailStageView(executor, request=request)
+
+        result = stage_view._is_rate_limited()
+        self.assertIsNone(result)
+
+    def test_is_rate_limited_returns_remaining_time(self):
+        """Test to ensure the remaining time is returned if the request
+        should be rate limited."""
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
+        session = self.client.session
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+
+        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
+        request = self.factory.post(url)
+        request.user = self.user
+        request.session = session
+
+        executor = FlowExecutorView(request=request, flow=self.flow)
+        executor.current_stage = self.stage
+        executor.plan = plan
+
+        stage_view = EmailStageView(executor, request=request)
+
+        test_cases = [
+            # 2 attempts within 2 minutes
+            (2, "seconds=120", 2),
+            # 4 attempts within 5 minutes
+            (4, "minutes=5", 5),
+            # 6 attempts within 5 minutes. Although 299 seconds is less than
+            # 5 minutes, the user is intentionally shown "5 minutes". This is
+            # because an initial rate limiting message like "Try again after 4 minutes"
+            # can be confusing.
+            (6, "seconds=299", 5),
+        ]
+        for test_case in test_cases:
+            max_attempts, cache_timeout, minutes_remaining = test_case
+            with self.subTest(
+                f"Test recovery with {max_attempts} max attempts and "
+                f"{cache_timeout} cache timeout seconds"
+            ):
+                self.stage.recovery_max_attempts = max_attempts
+                self.stage.recovery_cache_timeout = cache_timeout
+                self.stage.save()
+
+                # Simulate multiple requests
+                for _ in range(max_attempts):
+                    stage_view._is_rate_limited()
+
+                # The following request should be rate-limited
+                result = stage_view._is_rate_limited()
+
+                self.assertEqual(result, minutes_remaining)
+
+    def _challenge_invalid_helper(self):
+        """Helper to test the challenge_invalid() method."""
+        self.stage.recovery_max_attempts = 1
+        self.stage.recovery_cache_timeout = "seconds=300"
+        self.stage.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
+        session = self.client.session
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+
+        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
+        request = get_request(url, user=self.user)
+        request.session = session
+
+        request.brand = Brand.objects.create(domain="foo-domain.com", default=True)
+
+        executor = FlowExecutorView(request=request, flow=self.flow)
+        executor.current_stage = self.stage
+        executor.plan = plan
+
+        stage_view = EmailStageView(executor, request=request)
+        challenge_response = stage_view.get_response_instance(data={})
+        challenge_response.is_valid()
+
+        return challenge_response, stage_view, request
+
+    def test_challenge_invalid_not_rate_limited(self):
+        """Tests that the request is not rate limited and email is sent."""
+        challenge_response, stage_view, request = self._challenge_invalid_helper()
+
+        with patch.object(stage_view, "send_email") as mock_send_email:
+            result = stage_view.challenge_invalid(challenge_response)
+
+            self.assertEqual(result.status_code, 200)
+
+            mock_send_email.assert_called_once()
+
+            message_list = list(messages.get_messages(request))
+            self.assertEqual(len(message_list), 1)
+            self.assertEqual(
+                "Email Successfully sent.",
+                message_list[-1].message,
+            )
+
+    def test_challenge_invalid_returns_error_if_rate_limited(self):
+        """Tests that an error is returned if the request is rate limited. Ensure
+        that an email is not sent."""
+        challenge_response, stage_view, request = self._challenge_invalid_helper()
+
+        # Initial request that shouldn't be rate limited
+        stage_view.challenge_invalid(challenge_response)
+
+        with patch.object(stage_view, "send_email") as mock_send_email:
+            # This next request should be rate limited
+            result = stage_view.challenge_invalid(challenge_response)
+
+            self.assertEqual(result.status_code, 200)
+
+            mock_send_email.assert_not_called()
+
+            message_list = list(messages.get_messages(request))
+            self.assertEqual(len(message_list), 2)
+            self.assertEqual(
+                "Too many account verification attempts. Please try again after 5 minutes.",
+                message_list[-1].message,
+            )

blueprints/example/flows-recovery-email-verification.yaml

@@ -61,6 +61,8 @@ entries:
       subject: authentik
       template: email/password_reset.html
       activate_user_on_success: true
+      recovery_max_attempts: 5
+      recovery_cache_timeout: minutes=5
   - identifiers:
       name: default-recovery-user-write
     id: default-recovery-user-write

blueprints/schema.json

@@ -14336,6 +14336,18 @@
                     "type": "boolean",
                     "title": "Activate user on success",
                     "description": "Activate users upon completion of stage."
+                },
+                "recovery_max_attempts": {
+                    "type": "integer",
+                    "minimum": 0,
+                    "maximum": 2147483647,
+                    "title": "Recovery max attempts"
+                },
+                "recovery_cache_timeout": {
+                    "type": "string",
+                    "minLength": 1,
+                    "title": "Recovery cache timeout",
+                    "description": "The time window used to count recent account recovery attempts. If the number of attempts exceed recovery_max_attempts within this period, further attempts will be rate-limited. (Format: hours=1;minutes=2;seconds=3)."
                 }
             },
             "required": []

schema.yml

@@ -44845,6 +44845,15 @@ components:
         activate_user_on_success:
           type: boolean
           description: Activate users upon completion of stage.
+        recovery_max_attempts:
+          type: integer
+          maximum: 2147483647
+          minimum: 0
+        recovery_cache_timeout:
+          type: string
+          description: 'The time window used to count recent account recovery attempts.
+            If the number of attempts exceed recovery_max_attempts within this period,
+            further attempts will be rate-limited. (Format: hours=1;minutes=2;seconds=3).'
       required:
       - component
       - meta_model_name
@@ -44905,6 +44914,16 @@ components:
         activate_user_on_success:
           type: boolean
           description: Activate users upon completion of stage.
+        recovery_max_attempts:
+          type: integer
+          maximum: 2147483647
+          minimum: 0
+        recovery_cache_timeout:
+          type: string
+          minLength: 1
+          description: 'The time window used to count recent account recovery attempts.
+            If the number of attempts exceed recovery_max_attempts within this period,
+            further attempts will be rate-limited. (Format: hours=1;minutes=2;seconds=3).'
       required:
       - name
     Endpoint:
@@ -53394,6 +53413,16 @@ components:
         activate_user_on_success:
           type: boolean
           description: Activate users upon completion of stage.
+        recovery_max_attempts:
+          type: integer
+          maximum: 2147483647
+          minimum: 0
+        recovery_cache_timeout:
+          type: string
+          minLength: 1
+          description: 'The time window used to count recent account recovery attempts.
+            If the number of attempts exceed recovery_max_attempts within this period,
+            further attempts will be rate-limited. (Format: hours=1;minutes=2;seconds=3).'
     PatchedEndpointDeviceRequest:
       type: object
       description: Serializer for Endpoint authenticator devices

web/src/admin/stages/email/EmailStageForm.ts

@@ -232,6 +232,36 @@ export class EmailStageForm extends BaseStageForm<EmailStage> {
                             })}
                         </select>
                     </ak-form-element-horizontal>
+                    <ak-form-element-horizontal
+                        label=${msg("Account Recovery Max Attempts")}
+                        required
+                        name="recoveryMaxAttempts"
+                    >
+                        <input
+                            type="number"
+                            value="${this.instance?.recoveryMaxAttempts ?? 5}"
+                            class="pf-c-form-control"
+                            required
+                        />
+                    </ak-form-element-horizontal>
+                    <ak-form-element-horizontal
+                        label=${msg("Account Recovery Cache Timeout")}
+                        required
+                        name="recoveryCacheTimeout"
+                    >
+                        <input
+                            type="text"
+                            value="${ifDefined(this.instance?.recoveryCacheTimeout || "minutes=5")}"
+                            class="pf-c-form-control"
+                            required
+                        />
+                        <p class="pf-c-form__helper-text">
+                            ${msg(
+                                "The time window used to count recent account recovery attempts.",
+                            )}
+                        </p>
+                        <ak-utils-time-delta-help></ak-utils-time-delta-help>
+                    </ak-form-element-horizontal>
                 </div>
             </ak-form-group>
             ${this.renderConnectionSettings()}`;

website/docs/developer-docs/translation.md

@@ -22,6 +22,8 @@ To simplify translation you can use https://www.transifex.com/authentik/authenti
 - Make (again, any recent version should work)
 - Docker
 
+### Frontend
+
 Run `npm i` in the `/web` folder to install all dependencies.
 
 Ensure the language code is in the `lit-localize.json` file in `web/`:
@@ -42,3 +44,17 @@ Afterwards, run `make web-i18n-extract` to generate a base .xlf file.
 The .xlf files can be edited by any text editor, or using a tool such as [POEdit](https://poedit.net/).
 
 To see the change, run `make web-watch` in the root directory of the repository.
+
+### Backend
+
+Backend translations are handled by `core-i18n-extract`.
+
+Use Django's translation utility to declare the string, e.g.:
+
+```python
+from django.utils.translation import gettext as _
+
+_("New text to be translated.")
+```
+
+Afterwards, run `make core-i18n-extract` to generate the updated translation files.