authentik.stages.identification.tests

  1"""identification tests"""
  2
  3from django.urls import reverse
  4from requests_mock import Mocker
  5from rest_framework.exceptions import ValidationError
  6
  7from authentik.core.tests.utils import create_test_admin_user, create_test_flow
  8from authentik.flows.models import FlowDesignation, FlowStageBinding
  9from authentik.flows.stage import PLAN_CONTEXT_PENDING_USER_IDENTIFIER
 10from authentik.flows.tests import FlowTestCase
 11from authentik.lib.generators import generate_id
 12from authentik.sources.oauth.models import OAuthSource
 13from authentik.stages.authenticator_validate.models import AuthenticatorValidateStage, DeviceClasses
 14from authentik.stages.authenticator_webauthn.models import WebAuthnDevice
 15from authentik.stages.captcha.models import CaptchaStage
 16from authentik.stages.captcha.tests import RECAPTCHA_PRIVATE_KEY, RECAPTCHA_PUBLIC_KEY
 17from authentik.stages.identification.api import IdentificationStageSerializer
 18from authentik.stages.identification.models import IdentificationStage, UserFields
 19from authentik.stages.password import BACKEND_INBUILT
 20from authentik.stages.password.models import PasswordStage
 21
 22
 23class TestIdentificationStagePasskey(FlowTestCase):
 24    """Passkey authentication tests"""
 25
 26    def setUp(self):
 27        super().setUp()
 28        self.user = create_test_admin_user()
 29        self.flow = create_test_flow(FlowDesignation.AUTHENTICATION)
 30        self.webauthn_stage = AuthenticatorValidateStage.objects.create(
 31            name="webauthn-validate",
 32            device_classes=[DeviceClasses.WEBAUTHN],
 33        )
 34        self.stage = IdentificationStage.objects.create(
 35            name="identification",
 36            user_fields=[UserFields.E_MAIL],
 37            webauthn_stage=self.webauthn_stage,
 38        )
 39        FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=0)
 40        self.device = WebAuthnDevice.objects.create(
 41            user=self.user,
 42            name="Test Passkey",
 43            credential_id="test-credential-id",
 44            public_key="test-public-key",
 45            sign_count=0,
 46            rp_id="testserver",
 47        )
 48
 49    def test_passkey_auth_success(self):
 50        """Test passkey sets device, user, backend and updates last_used"""
 51        from unittest.mock import patch
 52
 53        # Get challenge to initialize session
 54        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 55        self.client.get(url)
 56
 57        with patch(
 58            "authentik.stages.identification.stage.validate_challenge_webauthn",
 59            return_value=self.device,
 60        ):
 61            response = self.client.post(
 62                url, {"passkey": {"id": "test"}}, content_type="application/json"
 63            )
 64
 65        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
 66        # Verify device last_used was updated
 67        self.device.refresh_from_db()
 68        self.assertIsNotNone(self.device.last_used)
 69
 70    def test_passkey_challenge_disabled(self):
 71        """Test that passkey challenge is not included when webauthn_stage is not set"""
 72        self.stage.webauthn_stage = None
 73        self.stage.save()
 74        response = self.client.get(
 75            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 76        )
 77        self.assertEqual(response.status_code, 200)
 78        data = response.json()
 79        self.assertIsNone(data.get("passkey_challenge"))
 80
 81    def test_passkey_challenge_enabled(self):
 82        """Test that passkey challenge is included when webauthn_stage is set"""
 83        response = self.client.get(
 84            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 85        )
 86        self.assertEqual(response.status_code, 200)
 87        data = response.json()
 88        self.assertIsNotNone(data.get("passkey_challenge"))
 89        passkey_challenge = data["passkey_challenge"]
 90        self.assertIn("challenge", passkey_challenge)
 91        self.assertIn("rpId", passkey_challenge)
 92        self.assertEqual(passkey_challenge["allowCredentials"], [])
 93
 94    def test_passkey_challenge_generation(self):
 95        """Test passkey challenge is generated correctly"""
 96        response = self.client.get(
 97            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 98        )
 99        self.assertEqual(response.status_code, 200)
100        data = response.json()
101        self.assertIsNotNone(data.get("passkey_challenge"))
102
103    def test_passkey_no_uid_field_required(self):
104        """Test that uid_field is not required when passkey is provided"""
105        # Get the challenge first to set up the session
106        response = self.client.get(
107            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
108        )
109        self.assertEqual(response.status_code, 200)
110
111        # Submit without uid_field but with passkey (invalid passkey will fail validation)
112        form_data = {
113            "passkey": {
114                "id": "invalid",
115                "rawId": "invalid",
116                "type": "public-key",
117                "response": {
118                    "clientDataJSON": "invalid",
119                    "authenticatorData": "invalid",
120                    "signature": "invalid",
121                },
122            }
123        }
124        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
125        response = self.client.post(url, form_data, content_type="application/json")
126        self.assertEqual(response.status_code, 200)
127        data = response.json()
128        self.assertIn("response_errors", data)
129        errors = data.get("response_errors", {})
130        self.assertNotIn("uid_field", errors)
131
132
133class TestIdentificationStage(FlowTestCase):
134    """Identification tests"""
135
136    def setUp(self):
137        super().setUp()
138        self.user = create_test_admin_user()
139
140        # OAuthSource for the login view
141        self.source = OAuthSource.objects.create(name=generate_id(), slug=generate_id())
142
143        self.flow = create_test_flow(FlowDesignation.AUTHENTICATION)
144        self.stage = IdentificationStage.objects.create(
145            name="identification",
146            user_fields=[UserFields.E_MAIL],
147            pretend_user_exists=False,
148        )
149        self.stage.sources.set([self.source])
150        self.stage.save()
151        FlowStageBinding.objects.create(
152            target=self.flow,
153            stage=self.stage,
154            order=0,
155        )
156
157    def test_valid_render(self):
158        """Test that View renders correctly"""
159        response = self.client.get(
160            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
161        )
162        self.assertEqual(response.status_code, 200)
163
164    def test_valid_with_email(self):
165        """Test with valid email, check that URL redirects back to itself"""
166        form_data = {"uid_field": self.user.email}
167        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
168        response = self.client.post(url, form_data)
169        self.assertEqual(response.status_code, 200)
170        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
171
172    def test_valid_with_password(self):
173        """Test with valid email and password in single step"""
174        pw_stage = PasswordStage.objects.create(name="password", backends=[BACKEND_INBUILT])
175        self.stage.password_stage = pw_stage
176        self.stage.save()
177        form_data = {"uid_field": self.user.email, "password": self.user.username}
178        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
179        response = self.client.post(url, form_data)
180        self.assertEqual(response.status_code, 200)
181        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
182
183    def test_invalid_with_password(self):
184        """Test with valid email and invalid password in single step"""
185        pw_stage = PasswordStage.objects.create(name="password", backends=[BACKEND_INBUILT])
186        self.stage.password_stage = pw_stage
187        self.stage.save()
188        form_data = {
189            "uid_field": self.user.email,
190            "password": self.user.username + "test",
191        }
192        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
193        response = self.client.post(url, form_data)
194        self.assertStageResponse(
195            response,
196            self.flow,
197            component="ak-stage-identification",
198            password_fields=True,
199            primary_action="Log in",
200            response_errors={
201                "non_field_errors": [{"code": "invalid", "string": "Failed to authenticate."}]
202            },
203            sources=[
204                {
205                    "challenge": {
206                        "component": "xak-flow-redirect",
207                        "to": f"/source/oauth/login/{self.source.slug}/",
208                    },
209                    "icon_url": "/static/authentik/sources/default.svg",
210                    "name": self.source.name,
211                    "promoted": False,
212                }
213            ],
214            show_source_labels=False,
215            user_fields=["email"],
216        )
217
218    def test_invalid_with_password_pretend(self):
219        """Test with invalid email and invalid password in single step (with pretend_user_exists)"""
220        self.stage.pretend_user_exists = True
221        pw_stage = PasswordStage.objects.create(name="password", backends=[BACKEND_INBUILT])
222        self.stage.password_stage = pw_stage
223        self.stage.save()
224        form_data = {
225            "uid_field": self.user.email + "test",
226            "password": self.user.username + "test",
227        }
228        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
229        response = self.client.post(url, form_data)
230        self.assertStageResponse(
231            response,
232            self.flow,
233            component="ak-stage-identification",
234            password_fields=True,
235            primary_action="Log in",
236            response_errors={
237                "non_field_errors": [{"code": "invalid", "string": "Failed to authenticate."}]
238            },
239            sources=[
240                {
241                    "challenge": {
242                        "component": "xak-flow-redirect",
243                        "to": f"/source/oauth/login/{self.source.slug}/",
244                    },
245                    "icon_url": "/static/authentik/sources/default.svg",
246                    "name": self.source.name,
247                    "promoted": False,
248                }
249            ],
250            show_source_labels=False,
251            user_fields=["email"],
252        )
253
254    @Mocker()
255    def test_valid_with_captcha(self, mock: Mocker):
256        """Test with valid email and captcha token in single step"""
257        mock.post(
258            "https://www.recaptcha.net/recaptcha/api/siteverify",
259            json={
260                "success": True,
261                "score": 0.5,
262            },
263        )
264
265        captcha_stage = CaptchaStage.objects.create(
266            name="captcha",
267            public_key=RECAPTCHA_PUBLIC_KEY,
268            private_key=RECAPTCHA_PRIVATE_KEY,
269        )
270        self.stage.captcha_stage = captcha_stage
271        self.stage.save()
272
273        form_data = {"uid_field": self.user.email, "captcha_token": "PASSED"}
274        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
275        response = self.client.post(url, form_data)
276        self.assertEqual(response.status_code, 200)
277        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
278
279    @Mocker()
280    def test_invalid_with_captcha(self, mock: Mocker):
281        """Test with valid email and invalid captcha token in single step"""
282        mock.post(
283            "https://www.recaptcha.net/recaptcha/api/siteverify",
284            json={
285                "success": False,
286                "score": 0.5,
287            },
288        )
289
290        captcha_stage = CaptchaStage.objects.create(
291            name="captcha",
292            public_key=RECAPTCHA_PUBLIC_KEY,
293            private_key=RECAPTCHA_PRIVATE_KEY,
294        )
295
296        self.stage.captcha_stage = captcha_stage
297        self.stage.save()
298
299        form_data = {
300            "uid_field": self.user.email,
301            "captcha_token": "FAILED",
302        }
303        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
304        response = self.client.post(url, form_data)
305        self.assertStageResponse(
306            response,
307            self.flow,
308            component="ak-stage-identification",
309            password_fields=False,
310            primary_action="Log in",
311            response_errors={
312                "non_field_errors": [{"code": "invalid", "string": "Failed to authenticate."}]
313            },
314            sources=[
315                {
316                    "challenge": {
317                        "component": "xak-flow-redirect",
318                        "to": f"/source/oauth/login/{self.source.slug}/",
319                    },
320                    "icon_url": "/static/authentik/sources/default.svg",
321                    "name": self.source.name,
322                    "promoted": False,
323                }
324            ],
325            show_source_labels=False,
326            user_fields=["email"],
327        )
328
329    @Mocker()
330    def test_invalid_with_captcha_retriable(self, mock: Mocker):
331        """Test with valid email and invalid captcha token in single step"""
332        mock.post(
333            "https://www.recaptcha.net/recaptcha/api/siteverify",
334            json={
335                "success": False,
336                "score": 0.5,
337                "error-codes": ["timeout-or-duplicate"],
338            },
339        )
340
341        captcha_stage = CaptchaStage.objects.create(
342            name="captcha",
343            public_key=RECAPTCHA_PUBLIC_KEY,
344            private_key=RECAPTCHA_PRIVATE_KEY,
345        )
346
347        self.stage.captcha_stage = captcha_stage
348        self.stage.save()
349
350        form_data = {
351            "uid_field": self.user.email,
352            "captcha_token": "FAILED",
353        }
354        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
355        response = self.client.post(url, form_data)
356        self.assertStageResponse(
357            response,
358            self.flow,
359            component="ak-stage-identification",
360            password_fields=False,
361            primary_action="Log in",
362            response_errors={
363                "non_field_errors": [
364                    {
365                        "code": "invalid",
366                        "string": "Failed to authenticate.",
367                    }
368                ]
369            },
370            sources=[
371                {
372                    "challenge": {
373                        "component": "xak-flow-redirect",
374                        "to": f"/source/oauth/login/{self.source.slug}/",
375                    },
376                    "icon_url": "/static/authentik/sources/default.svg",
377                    "name": self.source.name,
378                    "promoted": False,
379                }
380            ],
381            show_source_labels=False,
382            user_fields=["email"],
383        )
384
385    def test_invalid_with_username(self):
386        """Test invalid with username (user exists but stage only allows email)"""
387        form_data = {"uid_field": self.user.username}
388        response = self.client.post(
389            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
390            form_data,
391        )
392        self.assertEqual(response.status_code, 200)
393        self.assertStageResponse(
394            response,
395            self.flow,
396            component="ak-stage-identification",
397            response_errors={
398                "non_field_errors": [{"string": "Failed to authenticate.", "code": "invalid"}]
399            },
400        )
401
402    def test_invalid_with_username_pretend(self):
403        """Test invalid with username (user exists but stage only allows email)"""
404        self.stage.pretend_user_exists = True
405        self.stage.save()
406        form_data = {"uid_field": self.user.username}
407        response = self.client.post(
408            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
409            form_data,
410        )
411        self.assertEqual(response.status_code, 200)
412        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
413
414    def test_invalid_no_fields(self):
415        """Test invalid with username (no user fields are enabled)"""
416        self.stage.user_fields = []
417        self.stage.save()
418        form_data = {"uid_field": self.user.username}
419        response = self.client.post(
420            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
421            form_data,
422        )
423        self.assertStageResponse(
424            response,
425            self.flow,
426            component="ak-stage-identification",
427            password_fields=False,
428            primary_action="Log in",
429            response_errors={
430                "non_field_errors": [{"code": "invalid", "string": "Failed to authenticate."}]
431            },
432            show_source_labels=False,
433            sources=[
434                {
435                    "challenge": {
436                        "component": "xak-flow-redirect",
437                        "to": f"/source/oauth/login/{self.source.slug}/",
438                    },
439                    "icon_url": "/static/authentik/sources/default.svg",
440                    "name": self.source.name,
441                    "promoted": False,
442                }
443            ],
444            user_fields=[],
445        )
446
447    def test_invalid_with_invalid_email(self):
448        """Test with invalid email (user doesn't exist) -> Will return to login form"""
449        form_data = {"uid_field": self.user.email + "test"}
450        response = self.client.post(
451            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
452            form_data,
453        )
454        self.assertEqual(response.status_code, 200)
455
456    def test_enrollment_flow(self):
457        """Test that enrollment flow is linked correctly"""
458        flow = create_test_flow()
459        self.stage.enrollment_flow = flow
460        self.stage.save()
461        FlowStageBinding.objects.create(
462            target=flow,
463            stage=self.stage,
464            order=0,
465        )
466
467        response = self.client.get(
468            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
469        )
470        self.assertStageResponse(
471            response,
472            self.flow,
473            component="ak-stage-identification",
474            user_fields=["email"],
475            password_fields=False,
476            enroll_url=reverse(
477                "authentik_core:if-flow",
478                kwargs={"flow_slug": flow.slug},
479            ),
480            show_source_labels=False,
481            primary_action="Log in",
482            sources=[
483                {
484                    "icon_url": "/static/authentik/sources/default.svg",
485                    "name": self.source.name,
486                    "challenge": {
487                        "component": "xak-flow-redirect",
488                        "to": f"/source/oauth/login/{self.source.slug}/",
489                    },
490                    "promoted": False,
491                }
492            ],
493        )
494
495    def test_link_recovery_flow(self):
496        """Test that recovery flow is linked correctly"""
497        flow = create_test_flow()
498        self.stage.recovery_flow = flow
499        self.stage.save()
500        FlowStageBinding.objects.create(
501            target=flow,
502            stage=self.stage,
503            order=0,
504        )
505        response = self.client.get(
506            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
507        )
508        self.assertStageResponse(
509            response,
510            self.flow,
511            component="ak-stage-identification",
512            user_fields=["email"],
513            password_fields=False,
514            recovery_url=reverse(
515                "authentik_core:if-flow",
516                kwargs={"flow_slug": flow.slug},
517            ),
518            show_source_labels=False,
519            primary_action="Log in",
520            sources=[
521                {
522                    "challenge": {
523                        "component": "xak-flow-redirect",
524                        "to": f"/source/oauth/login/{self.source.slug}/",
525                    },
526                    "icon_url": "/static/authentik/sources/default.svg",
527                    "name": self.source.name,
528                    "promoted": False,
529                }
530            ],
531        )
532
533    def test_recovery_flow_invalid_user(self):
534        """Test that an invalid user can proceed in a recovery flow"""
535        self.flow.designation = FlowDesignation.RECOVERY
536        self.flow.save()
537        response = self.client.get(
538            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
539        )
540        self.assertStageResponse(
541            response,
542            self.flow,
543            component="ak-stage-identification",
544            user_fields=["email"],
545            password_fields=False,
546            show_source_labels=False,
547            primary_action="Continue",
548            sources=[
549                {
550                    "challenge": {
551                        "component": "xak-flow-redirect",
552                        "to": f"/source/oauth/login/{self.source.slug}/",
553                    },
554                    "icon_url": "/static/authentik/sources/default.svg",
555                    "name": self.source.name,
556                    "promoted": False,
557                }
558            ],
559        )
560        form_data = {"uid_field": generate_id()}
561        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
562        response = self.client.post(url, form_data)
563        self.assertEqual(response.status_code, 200)
564
565    def test_api_validate(self):
566        """Test API validation"""
567        self.assertTrue(
568            IdentificationStageSerializer(
569                data={
570                    "name": generate_id(),
571                    "user_fields": [UserFields.E_MAIL, UserFields.USERNAME],
572                }
573            ).is_valid(raise_exception=True)
574        )
575        with self.assertRaises(ValidationError):
576            IdentificationStageSerializer(
577                data={
578                    "name": generate_id(),
579                    "user_fields": [],
580                    "sources": [],
581                }
582            ).is_valid(raise_exception=True)
583
584    def test_prefill(self):
585        """Username prefill from existing flow context"""
586        pw_stage = PasswordStage.objects.create(name=generate_id(), backends=[BACKEND_INBUILT])
587        self.stage.password_stage = pw_stage
588        self.stage.save()
589
590        self.client.get(
591            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
592        )
593
594        plan = self.get_flow_plan()
595        plan.context[PLAN_CONTEXT_PENDING_USER_IDENTIFIER] = "foo"
596        self.set_flow_plan(plan)
597        with self.assertFlowFinishes() as plan:
598            response = self.client.get(
599                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
600            )
601            self.assertEqual(response.status_code, 200)
602            self.assertStageResponse(
603                response,
604                self.flow,
605                component="ak-stage-identification",
606                pending_user_identifier="foo",
607            )
608
609    def test_prefill_simple(self):
610        """Username prefill from existing flow context"""
611        self.stage.pretend_user_exists = True
612        self.stage.save()
613
614        self.client.get(
615            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
616        )
617        plan = self.get_flow_plan()
618        plan.context[PLAN_CONTEXT_PENDING_USER_IDENTIFIER] = "foo"
619        self.set_flow_plan(plan)
620        response = self.client.get(
621            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
622        )
623        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))