add captcha to identification stage

authentik/flows/tests/test_inspector.py

@@ -46,6 +46,7 @@ def test(self):
             res.content,
             {
                 "allow_show_password": False,
+                "captcha_stage": None,
                 "component": "ak-stage-identification",
                 "flow_info": {
                     "background": flow.background_url,

authentik/stages/captcha/stage.py

@@ -27,6 +27,43 @@ class CaptchaChallenge(WithUserInfoChallenge):
     component = CharField(default="ak-stage-captcha")
 
 
+def verify_captcha_token(stage: CaptchaStage, token: str, remote_ip: str):
+    """Validate captcha token"""
+    try:
+        response = get_http_session().post(
+            stage.api_url,
+            headers={
+                "Content-type": "application/x-www-form-urlencoded",
+            },
+            data={
+                "secret": stage.private_key,
+                "response": token,
+                "remoteip": remote_ip,
+            },
+        )
+        response.raise_for_status()
+        data = response.json()
+        if stage.error_on_invalid_score:
+            if not data.get("success", False):
+                raise ValidationError(
+                    _(
+                        "Failed to validate token: {error}".format(
+                            error=data.get("error-codes", _("Unknown error"))
+                        )
+                    )
+                )
+            if "score" in data:
+                score = float(data.get("score"))
+                if stage.score_max_threshold > -1 and score > stage.score_max_threshold:
+                    raise ValidationError(_("Invalid captcha response"))
+                if stage.score_min_threshold > -1 and score < stage.score_min_threshold:
+                    raise ValidationError(_("Invalid captcha response"))
+    except (RequestException, TypeError) as exc:
+        raise ValidationError(_("Failed to validate token")) from exc
+
+    return data
+
+
 class CaptchaChallengeResponse(ChallengeResponse):
     """Validate captcha token"""
 
@@ -36,38 +73,9 @@ class CaptchaChallengeResponse(ChallengeResponse):
     def validate_token(self, token: str) -> str:
         """Validate captcha token"""
         stage: CaptchaStage = self.stage.executor.current_stage
-        try:
-            response = get_http_session().post(
-                stage.api_url,
-                headers={
-                    "Content-type": "application/x-www-form-urlencoded",
-                },
-                data={
-                    "secret": stage.private_key,
-                    "response": token,
-                    "remoteip": ClientIPMiddleware.get_client_ip(self.stage.request),
-                },
-            )
-            response.raise_for_status()
-            data = response.json()
-            if stage.error_on_invalid_score:
-                if not data.get("success", False):
-                    raise ValidationError(
-                        _(
-                            "Failed to validate token: {error}".format(
-                                error=data.get("error-codes", _("Unknown error"))
-                            )
-                        )
-                    )
-                if "score" in data:
-                    score = float(data.get("score"))
-                    if stage.score_max_threshold > -1 and score > stage.score_max_threshold:
-                        raise ValidationError(_("Invalid captcha response"))
-                    if stage.score_min_threshold > -1 and score < stage.score_min_threshold:
-                        raise ValidationError(_("Invalid captcha response"))
-        except (RequestException, TypeError) as exc:
-            raise ValidationError(_("Failed to validate token")) from exc
-        return data
+        client_ip = ClientIPMiddleware.get_client_ip(self.stage.request)
+
+        return verify_captcha_token(stage, token, client_ip)
 
 
 class CaptchaStageView(ChallengeStageView):

authentik/stages/identification/api.py

@@ -27,6 +27,7 @@ class Meta:
         fields = StageSerializer.Meta.fields + [
             "user_fields",
             "password_stage",
+            "captcha_stage",
             "case_insensitive_matching",
             "show_matched_user",
             "enrollment_flow",
@@ -46,6 +47,7 @@ class IdentificationStageViewSet(UsedByMixin, ModelViewSet):
     filterset_fields = [
         "name",
         "password_stage",
+        "captcha_stage",
         "case_insensitive_matching",
         "show_matched_user",
         "enrollment_flow",

authentik/stages/identification/migrations/0015_identificationstage_captcha_stage.py

@@ -0,0 +1,26 @@
+# Generated by Django 5.0.8 on 2024-08-29 11:31
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_stages_captcha", "0003_captchastage_error_on_invalid_score_and_more"),
+        ("authentik_stages_identification", "0014_identificationstage_pretend"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="identificationstage",
+            name="captcha_stage",
+            field=models.ForeignKey(
+                default=None,
+                help_text="When set, adds functionality exactly like a Captcha stage, but baked into the Identification stage.",
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                to="authentik_stages_captcha.captchastage",
+            ),
+        ),
+    ]

authentik/stages/identification/models.py

@@ -8,6 +8,7 @@
 
 from authentik.core.models import Source
 from authentik.flows.models import Flow, Stage
+from authentik.stages.captcha.models import CaptchaStage
 from authentik.stages.password.models import PasswordStage
 
 
@@ -43,6 +44,19 @@ class IdentificationStage(Stage):
         ),
     )
 
+    captcha_stage = models.ForeignKey(
+        CaptchaStage,
+        null=True,
+        default=None,
+        on_delete=models.SET_NULL,
+        help_text=_(
+            (
+                "When set, adds functionality exactly like a Captcha stage, but baked into the "
+                "Identification stage."
+            ),
+        ),
+    )
+
     case_insensitive_matching = models.BooleanField(
         default=True,
         help_text=_("When enabled, user fields are matched regardless of their casing."),

authentik/stages/identification/stage.py

@@ -29,6 +29,7 @@
 from authentik.lib.utils.reflection import all_subclasses
 from authentik.lib.utils.urls import reverse_with_qs
 from authentik.root.middleware import ClientIPMiddleware
+from authentik.stages.captcha.stage import CaptchaChallenge, verify_captcha_token
 from authentik.stages.identification.models import IdentificationStage
 from authentik.stages.identification.signals import identification_failed
 from authentik.stages.password.stage import authenticate
@@ -75,6 +76,7 @@ class IdentificationChallenge(Challenge):
     allow_show_password = BooleanField(default=False)
     application_pre = CharField(required=False)
     flow_designation = ChoiceField(FlowDesignation.choices)
+    captcha_stage = CaptchaChallenge(required=False)
 
     enroll_url = CharField(required=False)
     recovery_url = CharField(required=False)
@@ -91,14 +93,16 @@ class IdentificationChallengeResponse(ChallengeResponse):
 
     uid_field = CharField()
     password = CharField(required=False, allow_blank=True, allow_null=True)
+    token = CharField(required=False, allow_blank=True, allow_null=True)
     component = CharField(default="ak-stage-identification")
 
     pre_user: User | None = None
 
     def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
-        """Validate that user exists, and optionally their password"""
+        """Validate that user exists, and optionally their password and captcha token"""
         uid_field = attrs["uid_field"]
         current_stage: IdentificationStage = self.stage.executor.current_stage
+        client_ip = ClientIPMiddleware.get_client_ip(self.stage.request)
 
         pre_user = self.stage.get_user(uid_field)
         if not pre_user:
@@ -113,7 +117,7 @@ def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
             self.stage.logger.info(
                 "invalid_login",
                 identifier=uid_field,
-                client_ip=ClientIPMiddleware.get_client_ip(self.stage.request),
+                client_ip=client_ip,
                 action="invalid_identifier",
                 context={
                     "stage": sanitize_item(self.stage),
@@ -136,6 +140,15 @@ def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
                 return attrs
             raise ValidationError("Failed to authenticate.")
         self.pre_user = pre_user
+
+        # Captcha check
+        if captcha_stage := current_stage.captcha_stage:
+            token = attrs.get("token", None)
+            if not token:
+                self.stage.logger.warning("Token not set for captcha attempt")
+            verify_captcha_token(captcha_stage, token, client_ip)
+
+        # Password check
         if not current_stage.password_stage:
             # No password stage select, don't validate the password
             return attrs
@@ -206,6 +219,14 @@ def get_challenge(self) -> Challenge:
                 "primary_action": self.get_primary_action(),
                 "user_fields": current_stage.user_fields,
                 "password_fields": bool(current_stage.password_stage),
+                "captcha_stage": (
+                    {
+                        "js_url": current_stage.captcha_stage.js_url,
+                        "site_key": current_stage.captcha_stage.public_key,
+                    }
+                    if current_stage.captcha_stage
+                    else None
+                ),
                 "allow_show_password": bool(current_stage.password_stage)
                 and current_stage.password_stage.allow_show_password,
                 "show_source_labels": current_stage.show_source_labels,

authentik/stages/identification/tests.py

@@ -1,13 +1,16 @@
 """identification tests"""
 
 from django.urls import reverse
+from requests_mock import Mocker
 from rest_framework.exceptions import ValidationError
 
 from authentik.core.tests.utils import create_test_admin_user, create_test_flow
 from authentik.flows.models import FlowDesignation, FlowStageBinding
 from authentik.flows.tests import FlowTestCase
 from authentik.lib.generators import generate_id
 from authentik.sources.oauth.models import OAuthSource
+from authentik.stages.captcha.models import CaptchaStage
+from authentik.stages.captcha.tests import RECAPTCHA_PRIVATE_KEY, RECAPTCHA_PUBLIC_KEY
 from authentik.stages.identification.api import IdentificationStageSerializer
 from authentik.stages.identification.models import IdentificationStage, UserFields
 from authentik.stages.password import BACKEND_INBUILT
@@ -133,6 +136,82 @@ def test_invalid_with_password_pretend(self):
             user_fields=["email"],
         )
 
+    @Mocker()
+    def test_valid_with_captcha(self, mock: Mocker):
+        """Test with valid email and captcha token in single step"""
+        mock.post(
+            "https://www.recaptcha.net/recaptcha/api/siteverify",
+            json={
+                "success": True,
+                "score": 0.5,
+            },
+        )
+
+        captcha_stage = CaptchaStage.objects.create(
+            name="captcha",
+            public_key=RECAPTCHA_PUBLIC_KEY,
+            private_key=RECAPTCHA_PRIVATE_KEY,
+        )
+        self.stage.captcha_stage = captcha_stage
+        self.stage.save()
+
+        form_data = {"uid_field": self.user.email, "token": "PASSED"}
+        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
+        response = self.client.post(url, form_data)
+        self.assertEqual(response.status_code, 200)
+        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
+
+    @Mocker()
+    def test_invalid_with_captcha(self, mock: Mocker):
+        """Test with valid email and invalid captcha token in single step"""
+        mock.post(
+            "https://www.recaptcha.net/recaptcha/api/siteverify",
+            json={
+                "success": False,
+                "score": 0.5,
+            },
+        )
+
+        captcha_stage = CaptchaStage.objects.create(
+            name="captcha",
+            public_key=RECAPTCHA_PUBLIC_KEY,
+            private_key=RECAPTCHA_PRIVATE_KEY,
+        )
+
+        self.stage.captcha_stage = captcha_stage
+        self.stage.save()
+
+        form_data = {
+            "uid_field": self.user.email,
+            "token": "FAILED",
+        }
+        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
+        response = self.client.post(url, form_data)
+        self.assertStageResponse(
+            response,
+            self.flow,
+            component="ak-stage-identification",
+            password_fields=False,
+            primary_action="Log in",
+            response_errors={
+                "non_field_errors": [
+                    {"code": "invalid", "string": "Failed to validate token: Unknown error"}
+                ]
+            },
+            sources=[
+                {
+                    "challenge": {
+                        "component": "xak-flow-redirect",
+                        "to": "/source/oauth/login/test/",
+                    },
+                    "icon_url": "/static/authentik/sources/default.svg",
+                    "name": "test",
+                }
+            ],
+            show_source_labels=False,
+            user_fields=["email"],
+        )
+
     def test_invalid_with_username(self):
         """Test invalid with username (user exists but stage only allows email)"""
         form_data = {"uid_field": self.user.username}

blueprints/schema.json

@@ -10131,6 +10131,11 @@
                     "title": "Password stage",
                     "description": "When set, shows a password field, instead of showing the password field as separate step."
                 },
+                "captcha_stage": {
+                    "type": "integer",
+                    "title": "Captcha stage",
+                    "description": "When set, adds functionality exactly like a Captcha stage, but baked into the Identification stage."
+                },
                 "case_insensitive_matching": {
                     "type": "boolean",
                     "title": "Case insensitive matching",