authentik.stages.authenticator_validate.tests.test_webauthn

  1"""Test validator stage"""
  2
  3from time import sleep
  4
  5from django.urls.base import reverse
  6from rest_framework.serializers import ValidationError
  7from webauthn.helpers.base64url_to_bytes import base64url_to_bytes
  8from webauthn.helpers.bytes_to_base64url import bytes_to_base64url
  9
 10from authentik.core.tests.utils import RequestFactory, create_test_admin_user, create_test_flow
 11from authentik.flows.models import FlowStageBinding, NotConfiguredAction
 12from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
 13from authentik.flows.stage import StageView
 14from authentik.flows.tests import FlowTestCase
 15from authentik.flows.views.executor import SESSION_KEY_PLAN, FlowExecutorView
 16from authentik.lib.generators import generate_id
 17from authentik.stages.authenticator_validate.challenge import (
 18    get_challenge_for_device,
 19    get_webauthn_challenge_without_user,
 20    validate_challenge_webauthn,
 21)
 22from authentik.stages.authenticator_validate.models import AuthenticatorValidateStage, DeviceClasses
 23from authentik.stages.authenticator_validate.stage import (
 24    PLAN_CONTEXT_DEVICE_CHALLENGES,
 25    AuthenticatorValidateStageView,
 26)
 27from authentik.stages.authenticator_webauthn.models import (
 28    UserVerification,
 29    WebAuthnDevice,
 30    WebAuthnDeviceType,
 31    WebAuthnHint,
 32)
 33from authentik.stages.authenticator_webauthn.stage import PLAN_CONTEXT_WEBAUTHN_CHALLENGE
 34from authentik.stages.authenticator_webauthn.tasks import webauthn_mds_import
 35from authentik.stages.identification.models import IdentificationStage, UserFields
 36from authentik.stages.user_login.models import UserLoginStage
 37
 38
 39class AuthenticatorValidateStageWebAuthnTests(FlowTestCase):
 40    """Test validator stage"""
 41
 42    def setUp(self) -> None:
 43        self.user = create_test_admin_user()
 44        self.request_factory = RequestFactory()
 45
 46    def test_last_auth_threshold(self):
 47        """Test last_auth_threshold"""
 48        ident_stage = IdentificationStage.objects.create(
 49            name=generate_id(),
 50            user_fields=[
 51                UserFields.USERNAME,
 52            ],
 53        )
 54        device: WebAuthnDevice = WebAuthnDevice.objects.create(
 55            user=self.user,
 56            confirmed=True,
 57        )
 58        device.set_sign_count(device.sign_count + 1)
 59        stage = AuthenticatorValidateStage.objects.create(
 60            name=generate_id(),
 61            last_auth_threshold="milliseconds=0",
 62            not_configured_action=NotConfiguredAction.CONFIGURE,
 63            device_classes=[DeviceClasses.WEBAUTHN],
 64        )
 65        sleep(1)
 66        stage.configuration_stages.set([ident_stage])
 67        flow = create_test_flow()
 68        FlowStageBinding.objects.create(target=flow, stage=ident_stage, order=0)
 69        FlowStageBinding.objects.create(target=flow, stage=stage, order=1)
 70
 71        response = self.client.post(
 72            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
 73            {"uid_field": self.user.username},
 74        )
 75        self.assertEqual(response.status_code, 302)
 76        response = self.client.get(
 77            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
 78            follow=True,
 79        )
 80        self.assertStageResponse(
 81            response,
 82            flow,
 83            component="ak-stage-authenticator-validate",
 84        )
 85
 86    def test_device_challenge_webauthn(self):
 87        """Test webauthn"""
 88        request = self.request_factory.get("/")
 89        request.user = self.user
 90
 91        webauthn_device = WebAuthnDevice.objects.create(
 92            user=self.user,
 93            public_key=bytes_to_base64url(b"qwerqwerqre"),
 94            credential_id=bytes_to_base64url(b"foobarbaz"),
 95            sign_count=0,
 96            rp_id=generate_id(),
 97        )
 98        stage = AuthenticatorValidateStage.objects.create(
 99            name=generate_id(),
100            last_auth_threshold="milliseconds=0",
101            not_configured_action=NotConfiguredAction.CONFIGURE,
102            device_classes=[DeviceClasses.WEBAUTHN],
103            webauthn_user_verification=UserVerification.PREFERRED,
104        )
105        plan = FlowPlan("")
106        stage_view = AuthenticatorValidateStageView(
107            FlowExecutorView(flow=None, current_stage=stage, plan=plan), request=request
108        )
109        challenge = get_challenge_for_device(stage_view, stage, webauthn_device)
110        del challenge["challenge"]
111        self.assertEqual(
112            challenge,
113            {
114                "allowCredentials": [
115                    {
116                        "id": "Zm9vYmFyYmF6",
117                        "type": "public-key",
118                    }
119                ],
120                "rpId": "testserver",
121                "timeout": 60000,
122                "userVerification": "preferred",
123            },
124        )
125
126        with self.assertRaises(ValidationError):
127            validate_challenge_webauthn(
128                {},
129                StageView(FlowExecutorView(current_stage=stage, plan=plan), request=request),
130                self.user,
131            )
132
133    def test_device_challenge_webauthn_restricted(self):
134        """Test webauthn (getting device challenges with a webauthn
135        device that is not allowed due to aaguid restrictions)"""
136        webauthn_mds_import.send(force=True).get_result()
137        request = self.request_factory.get("/")
138        request.user = self.user
139
140        WebAuthnDevice.objects.create(
141            user=self.user,
142            public_key=bytes_to_base64url(b"qwerqwerqre"),
143            credential_id=bytes_to_base64url(b"foobarbaz"),
144            sign_count=0,
145            rp_id=generate_id(),
146            device_type=WebAuthnDeviceType.objects.get(
147                aaguid="2fc0579f-8113-47ea-b116-bb5a8db9202a"
148            ),
149        )
150        flow = create_test_flow()
151        stage = AuthenticatorValidateStage.objects.create(
152            name=generate_id(),
153            last_auth_threshold="milliseconds=0",
154            not_configured_action=NotConfiguredAction.DENY,
155            device_classes=[DeviceClasses.WEBAUTHN],
156            webauthn_user_verification=UserVerification.PREFERRED,
157        )
158        stage.webauthn_allowed_device_types.set(
159            WebAuthnDeviceType.objects.filter(description="YubiKey 5 Series")
160        )
161        session = self.client.session
162        plan = FlowPlan(flow_pk=flow.pk.hex)
163        plan.append_stage(stage)
164        plan.append_stage(UserLoginStage.objects.create(name=generate_id()))
165        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
166        session[SESSION_KEY_PLAN] = plan
167        session.save()
168
169        response = self.client.get(
170            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
171        )
172        self.assertStageResponse(
173            response,
174            flow,
175            component="ak-stage-access-denied",
176            error_message="No (allowed) MFA authenticator configured.",
177        )
178
179    def test_raw_get_challenge(self):
180        """Test webauthn"""
181        request = self.request_factory.get("/")
182        request.user = self.user
183
184        stage = AuthenticatorValidateStage.objects.create(
185            name=generate_id(),
186            last_auth_threshold="milliseconds=0",
187            not_configured_action=NotConfiguredAction.CONFIGURE,
188            device_classes=[DeviceClasses.WEBAUTHN],
189            webauthn_user_verification=UserVerification.PREFERRED,
190        )
191        webauthn_device = WebAuthnDevice.objects.create(
192            user=self.user,
193            public_key=(
194                "pQECAyYgASFYIGsBLkklToCQkT7qJT_bJYN1sEc1oJdbnmoOc43i0J"
195                "H6IlggLTXytuhzFVYYAK4PQNj8_coGrbbzSfUxdiPAcZTQCyU"
196            ),
197            credential_id="QKZ97ASJAOIDyipAs6mKUxDUZgDrWrbAsUb5leL7-oU",
198            sign_count=0,
199            rp_id=generate_id(),
200        )
201        plan = FlowPlan("")
202        plan.context[PLAN_CONTEXT_WEBAUTHN_CHALLENGE] = base64url_to_bytes(
203            "g98I51mQvZXo5lxLfhrD2zfolhZbLRyCgqkkYap1jwSaJ13BguoJWCF9_Lg3AgO4Wh-Bqa556JE20oKsYbl6RA"
204        )
205        stage_view = AuthenticatorValidateStageView(
206            FlowExecutorView(flow=None, current_stage=stage, plan=plan), request=request
207        )
208        challenge = get_challenge_for_device(stage_view, stage, webauthn_device)
209        self.assertEqual(
210            challenge["allowCredentials"],
211            [
212                {
213                    "id": "QKZ97ASJAOIDyipAs6mKUxDUZgDrWrbAsUb5leL7-oU",
214                    "type": "public-key",
215                }
216            ],
217        )
218        self.assertIsNotNone(challenge["challenge"])
219        self.assertEqual(
220            challenge["rpId"],
221            "testserver",
222        )
223        self.assertEqual(
224            challenge["timeout"],
225            60000,
226        )
227        self.assertEqual(
228            challenge["userVerification"],
229            "preferred",
230        )
231
232    def test_get_challenge_userless(self):
233        """Test webauthn (userless)"""
234        request = self.request_factory.get("/")
235        stage = AuthenticatorValidateStage.objects.create(
236            name=generate_id(), webauthn_user_verification=UserVerification.PREFERRED
237        )
238        stage.refresh_from_db()
239        WebAuthnDevice.objects.create(
240            user=self.user,
241            public_key=(
242                "pQECAyYgASFYIGsBLkklToCQkT7qJT_bJYN1sEc1oJdbnmoOc43i0J"
243                "H6IlggLTXytuhzFVYYAK4PQNj8_coGrbbzSfUxdiPAcZTQCyU"
244            ),
245            credential_id="QKZ97ASJAOIDyipAs6mKUxDUZgDrWrbAsUb5leL7-oU",
246            sign_count=0,
247            rp_id=generate_id(),
248        )
249        plan = FlowPlan("")
250        stage_view = AuthenticatorValidateStageView(
251            FlowExecutorView(flow=None, current_stage=stage, plan=plan), request=request
252        )
253        challenge = get_webauthn_challenge_without_user(stage_view, stage)
254        self.assertEqual(challenge["allowCredentials"], [])
255        self.assertIsNotNone(challenge["challenge"])
256        self.assertEqual(challenge["rpId"], "testserver")
257        self.assertEqual(challenge["timeout"], 60000)
258        self.assertEqual(challenge["userVerification"], "preferred")
259
260    def test_device_challenge_webauthn_with_hints(self):
261        """Test that webauthn hints are included in authentication challenge"""
262        request = self.request_factory.get("/")
263        request.user = self.user
264
265        webauthn_device = WebAuthnDevice.objects.create(
266            user=self.user,
267            public_key=bytes_to_base64url(b"qwerqwerqre"),
268            credential_id=bytes_to_base64url(b"foobarbaz"),
269            sign_count=0,
270            rp_id=generate_id(),
271        )
272        stage = AuthenticatorValidateStage.objects.create(
273            name=generate_id(),
274            last_auth_threshold="milliseconds=0",
275            not_configured_action=NotConfiguredAction.CONFIGURE,
276            device_classes=[DeviceClasses.WEBAUTHN],
277            webauthn_user_verification=UserVerification.PREFERRED,
278            webauthn_hints=[WebAuthnHint.CLIENT_DEVICE, WebAuthnHint.HYBRID],
279        )
280        plan = FlowPlan("")
281        stage_view = AuthenticatorValidateStageView(
282            FlowExecutorView(flow=None, current_stage=stage, plan=plan), request=request
283        )
284        challenge = get_challenge_for_device(stage_view, stage, webauthn_device)
285        self.assertEqual(challenge["hints"], ["client-device", "hybrid"])
286
287    def test_device_challenge_webauthn_no_hints(self):
288        """Test that hints key is absent when no hints configured"""
289        request = self.request_factory.get("/")
290        request.user = self.user
291
292        webauthn_device = WebAuthnDevice.objects.create(
293            user=self.user,
294            public_key=bytes_to_base64url(b"qwerqwerqre"),
295            credential_id=bytes_to_base64url(b"foobarbaz"),
296            sign_count=0,
297            rp_id=generate_id(),
298        )
299        stage = AuthenticatorValidateStage.objects.create(
300            name=generate_id(),
301            last_auth_threshold="milliseconds=0",
302            not_configured_action=NotConfiguredAction.CONFIGURE,
303            device_classes=[DeviceClasses.WEBAUTHN],
304            webauthn_user_verification=UserVerification.PREFERRED,
305        )
306        plan = FlowPlan("")
307        stage_view = AuthenticatorValidateStageView(
308            FlowExecutorView(flow=None, current_stage=stage, plan=plan), request=request
309        )
310        challenge = get_challenge_for_device(stage_view, stage, webauthn_device)
311        self.assertNotIn("hints", challenge)
312
313    def test_get_challenge_userless_with_hints(self):
314        """Test that hints are included in userless/passwordless challenge"""
315        request = self.request_factory.get("/")
316        stage = AuthenticatorValidateStage.objects.create(
317            name=generate_id(),
318            webauthn_user_verification=UserVerification.PREFERRED,
319            webauthn_hints=[WebAuthnHint.SECURITY_KEY, WebAuthnHint.CLIENT_DEVICE],
320        )
321        plan = FlowPlan("")
322        stage_view = AuthenticatorValidateStageView(
323            FlowExecutorView(flow=None, current_stage=stage, plan=plan), request=request
324        )
325        challenge = get_webauthn_challenge_without_user(stage_view, stage)
326        self.assertEqual(challenge["hints"], ["security-key", "client-device"])
327
328    def test_device_challenge_webauthn_hints_order_preserved(self):
329        """Test that hint order is preserved in authentication challenge"""
330        request = self.request_factory.get("/")
331        request.user = self.user
332
333        webauthn_device = WebAuthnDevice.objects.create(
334            user=self.user,
335            public_key=bytes_to_base64url(b"qwerqwerqre"),
336            credential_id=bytes_to_base64url(b"foobarbaz"),
337            sign_count=0,
338            rp_id=generate_id(),
339        )
340        stage = AuthenticatorValidateStage.objects.create(
341            name=generate_id(),
342            last_auth_threshold="milliseconds=0",
343            not_configured_action=NotConfiguredAction.CONFIGURE,
344            device_classes=[DeviceClasses.WEBAUTHN],
345            webauthn_user_verification=UserVerification.PREFERRED,
346            webauthn_hints=[
347                WebAuthnHint.HYBRID,
348                WebAuthnHint.SECURITY_KEY,
349                WebAuthnHint.CLIENT_DEVICE,
350            ],
351        )
352        plan = FlowPlan("")
353        stage_view = AuthenticatorValidateStageView(
354            FlowExecutorView(flow=None, current_stage=stage, plan=plan), request=request
355        )
356        challenge = get_challenge_for_device(stage_view, stage, webauthn_device)
357        self.assertEqual(challenge["hints"], ["hybrid", "security-key", "client-device"])
358
359    def test_validate_challenge_unrestricted(self):
360        """Test webauthn authentication (unrestricted webauthn device)"""
361        webauthn_mds_import.send(force=True).get_result()
362        device = WebAuthnDevice.objects.create(
363            user=self.user,
364            public_key=(
365                "pQECAyYgASFYIF-N4GvQJdTJMAmTOxFX9_boL00zBiSrP0DY9xvJl_FFIlggnyZloVSVofdJNTLMeMdjQHgW2Rzmd5_Xt5AWtNztcdo"
366            ),
367            credential_id="X43ga9Al1MkwCZM7EXD1r8Sxj7aXnNsuR013XM7he4kZ-GS9TaA-u3i36wsswjPm",
368            sign_count=2,
369            rp_id=generate_id(),
370            device_type=WebAuthnDeviceType.objects.get(
371                aaguid="2fc0579f-8113-47ea-b116-bb5a8db9202a"
372            ),
373        )
374        flow = create_test_flow()
375        stage = AuthenticatorValidateStage.objects.create(
376            name=generate_id(),
377            not_configured_action=NotConfiguredAction.CONFIGURE,
378            device_classes=[DeviceClasses.WEBAUTHN],
379        )
380        session = self.client.session
381        plan = FlowPlan(flow_pk=flow.pk.hex)
382        plan.append_stage(stage)
383        plan.append_stage(UserLoginStage.objects.create(name=generate_id()))
384        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
385        plan.context[PLAN_CONTEXT_DEVICE_CHALLENGES] = [
386            {
387                "device_class": device.__class__.__name__.lower().replace("device", ""),
388                "device_uid": device.pk,
389                "challenge": {},
390                "last_used": None,
391            }
392        ]
393        plan.context[PLAN_CONTEXT_WEBAUTHN_CHALLENGE] = base64url_to_bytes(
394            "aCC6ak_DP45xMH1qyxzUM5iC2xc4QthQb09v7m4qDBmY8FvWvhxFzSuFlDYQmclrh5fWS5q0TPxgJGF4vimcFQ"
395        )
396        session[SESSION_KEY_PLAN] = plan
397        session.save()
398
399        response = self.client.post(
400            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
401            data={
402                "webauthn": {
403                    "id": "X43ga9Al1MkwCZM7EXD1r8Sxj7aXnNsuR013XM7he4kZ-GS9TaA-u3i36wsswjPm",
404                    "rawId": "X43ga9Al1MkwCZM7EXD1r8Sxj7aXnNsuR013XM7he4kZ-GS9TaA-u3i36wsswjPm",
405                    "type": "public-key",
406                    "assertionClientExtensions": "{}",
407                    "response": {
408                        "clientDataJSON": (
409                            "eyJ0eXBlIjoid2ViYXV0aG4uZ2V0IiwiY2hhbGxlbmdlIjoiYUNDN"
410                            "mFrX0RQNDV4TUgxcXl4elVNNWlDMnhjNFF0aFFiMDl2N200cURCbV"
411                            "k4RnZXdmh4RnpTdUZsRFlRbWNscmg1ZldTNXEwVFB4Z0pHRjR2aW1"
412                            "jRlEiLCJvcmlnaW4iOiJodHRwOi8vbG9jYWxob3N0OjkwMDAiLCJj"
413                            "cm9zc09yaWdpbiI6ZmFsc2V9"
414                        ),
415                        "signature": (
416                            "MEQCIAHQCGfE_PX1z6mBDaXUNqK_NrllhXylNOmETUD3Khv9AiBTl"
417                            "rX3GDRj5OaOfTToOwUwAhtd74tu0T6DZAVHPb_hlQ=="
418                        ),
419                        "authenticatorData": "SZYN5YgOjGh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2MFAAAABg==",
420                        "userHandle": None,
421                    },
422                },
423            },
424            SERVER_NAME="localhost",
425            SERVER_PORT="9000",
426        )
427        self.assertEqual(response.status_code, 302)
428        response = self.client.get(
429            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
430        )
431        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
432
433    def test_validate_challenge_restricted(self):
434        """Test webauthn authentication (restricted device type, failure)"""
435        webauthn_mds_import.send(force=True).get_result()
436        device = WebAuthnDevice.objects.create(
437            user=self.user,
438            public_key=(
439                "pQECAyYgASFYIF-N4GvQJdTJMAmTOxFX9_boL00zBiSrP0DY9xvJl_FFIlggnyZloVSVofdJNTLMeMdjQHgW2Rzmd5_Xt5AWtNztcdo"
440            ),
441            credential_id="X43ga9Al1MkwCZM7EXD1r8Sxj7aXnNsuR013XM7he4kZ-GS9TaA-u3i36wsswjPm",
442            sign_count=2,
443            rp_id=generate_id(),
444            device_type=WebAuthnDeviceType.objects.get(
445                aaguid="2fc0579f-8113-47ea-b116-bb5a8db9202a"
446            ),
447        )
448        flow = create_test_flow()
449        stage = AuthenticatorValidateStage.objects.create(
450            name=generate_id(),
451            not_configured_action=NotConfiguredAction.CONFIGURE,
452            device_classes=[DeviceClasses.WEBAUTHN],
453        )
454        stage.webauthn_allowed_device_types.set(
455            WebAuthnDeviceType.objects.filter(description="YubiKey 5 Series")
456        )
457        session = self.client.session
458        plan = FlowPlan(flow_pk=flow.pk.hex)
459        plan.append_stage(stage)
460        plan.append_stage(UserLoginStage.objects.create(name=generate_id()))
461        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
462        plan.context[PLAN_CONTEXT_DEVICE_CHALLENGES] = [
463            {
464                "device_class": device.__class__.__name__.lower().replace("device", ""),
465                "device_uid": device.pk,
466                "challenge": {},
467                "last_used": None,
468            }
469        ]
470        plan.context[PLAN_CONTEXT_WEBAUTHN_CHALLENGE] = base64url_to_bytes(
471            "aCC6ak_DP45xMH1qyxzUM5iC2xc4QthQb09v7m4qDBmY8FvWvhxFzSuFlDYQmclrh5fWS5q0TPxgJGF4vimcFQ"
472        )
473        session[SESSION_KEY_PLAN] = plan
474        session.save()
475
476        response = self.client.post(
477            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
478            data={
479                "webauthn": {
480                    "id": "X43ga9Al1MkwCZM7EXD1r8Sxj7aXnNsuR013XM7he4kZ-GS9TaA-u3i36wsswjPm",
481                    "rawId": "X43ga9Al1MkwCZM7EXD1r8Sxj7aXnNsuR013XM7he4kZ-GS9TaA-u3i36wsswjPm",
482                    "type": "public-key",
483                    "assertionClientExtensions": "{}",
484                    "response": {
485                        "clientDataJSON": (
486                            "eyJ0eXBlIjoid2ViYXV0aG4uZ2V0IiwiY2hhbGxlbmdlIjoiYUNDN"
487                            "mFrX0RQNDV4TUgxcXl4elVNNWlDMnhjNFF0aFFiMDl2N200cURCbV"
488                            "k4RnZXdmh4RnpTdUZsRFlRbWNscmg1ZldTNXEwVFB4Z0pHRjR2aW1"
489                            "jRlEiLCJvcmlnaW4iOiJodHRwOi8vbG9jYWxob3N0OjkwMDAiLCJj"
490                            "cm9zc09yaWdpbiI6ZmFsc2V9"
491                        ),
492                        "signature": (
493                            "MEQCIAHQCGfE_PX1z6mBDaXUNqK_NrllhXylNOmETUD3Khv9AiBTl"
494                            "rX3GDRj5OaOfTToOwUwAhtd74tu0T6DZAVHPb_hlQ=="
495                        ),
496                        "authenticatorData": "SZYN5YgOjGh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2MFAAAABg==",
497                        "userHandle": None,
498                    },
499                }
500            },
501            SERVER_NAME="localhost",
502            SERVER_PORT="9000",
503        )
504        self.assertEqual(response.status_code, 200)
505        self.assertStageResponse(
506            response,
507            flow,
508            component="ak-stage-authenticator-validate",
509            response_errors={
510                "webauthn": [
511                    {
512                        "string": (
513                            "Invalid device type. Contact your authentik administrator for help."
514                        ),
515                        "code": "invalid",
516                    }
517                ]
518            },
519        )
520
521    def test_validate_challenge_userless(self):
522        """Test webauthn"""
523        device = WebAuthnDevice.objects.create(
524            user=self.user,
525            public_key=(
526                "pQECAyYgASFYIGsBLkklToCQkT7qJT_bJYN1sEc1oJdbnmoOc43i0J"
527                "H6IlggLTXytuhzFVYYAK4PQNj8_coGrbbzSfUxdiPAcZTQCyU"
528            ),
529            credential_id="QKZ97ASJAOIDyipAs6mKUxDUZgDrWrbAsUb5leL7-oU",
530            sign_count=4,
531            rp_id=generate_id(),
532        )
533        flow = create_test_flow()
534        stage = AuthenticatorValidateStage.objects.create(
535            name=generate_id(),
536            not_configured_action=NotConfiguredAction.CONFIGURE,
537            device_classes=[DeviceClasses.WEBAUTHN],
538        )
539        session = self.client.session
540        plan = FlowPlan(flow_pk=flow.pk.hex)
541        plan.append_stage(stage)
542        plan.append_stage(UserLoginStage.objects.create(name=generate_id()))
543        plan.context[PLAN_CONTEXT_DEVICE_CHALLENGES] = [
544            {
545                "device_class": device.__class__.__name__.lower().replace("device", ""),
546                "device_uid": device.pk,
547                "challenge": {},
548                "last_used": None,
549            }
550        ]
551        plan.context[PLAN_CONTEXT_WEBAUTHN_CHALLENGE] = base64url_to_bytes(
552            "g98I51mQvZXo5lxLfhrD2zfolhZbLRyCgqkkYap1jwSaJ13BguoJWCF9_Lg3AgO4Wh-Bqa556JE20oKsYbl6RA"
553        )
554        session[SESSION_KEY_PLAN] = plan
555        session.save()
556
557        response = self.client.post(
558            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
559            data={
560                "webauthn": {
561                    "id": "QKZ97ASJAOIDyipAs6mKUxDUZgDrWrbAsUb5leL7-oU",
562                    "rawId": "QKZ97ASJAOIDyipAs6mKUxDUZgDrWrbAsUb5leL7-oU",
563                    "type": "public-key",
564                    "assertionClientExtensions": "{}",
565                    "response": {
566                        "clientDataJSON": (
567                            "eyJ0eXBlIjoid2ViYXV0aG4uZ2V0IiwiY2hhbGxlbmdlIjoiZzk4STUxbVF2Wlhv"
568                            "NWx4TGZockQyemZvbGhaYkxSeUNncWtrWWFwMWp3U2FKMTNCZ3VvSldDRjlfTGcz"
569                            "QWdPNFdoLUJxYTU1NkpFMjBvS3NZYmw2UkEiLCJvcmlnaW4iOiJodHRwOi8vbG9j"
570                            "YWxob3N0OjkwMDAiLCJjcm9zc09yaWdpbiI6ZmFsc2UsIm90aGVyX2tleXNfY2Fu"
571                            "X2JlX2FkZGVkX2hlcmUiOiJkbyBub3QgY29tcGFyZSBjbGllbnREYXRhSlNPTiBh"
572                            "Z2FpbnN0IGEgdGVtcGxhdGUuIFNlZSBodHRwczovL2dvby5nbC95YWJQZXgifQ=="
573                        ),
574                        "signature": (
575                            "MEQCIFNlrHf9ablJAalXLWkrqvHB8oIu8kwvRpH3X3rbJVpI"
576                            "AiAqtOK6mIZPk62kZN0OzFsHfuvu_RlOl7zlqSNzDdz_Ag=="
577                        ),
578                        "authenticatorData": "SZYN5YgOjGh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2MFAAAABQ==",
579                        "userHandle": None,
580                    },
581                },
582            },
583            SERVER_NAME="localhost",
584            SERVER_PORT="9000",
585        )
586        self.assertEqual(response.status_code, 302)
587        response = self.client.get(
588            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
589        )
590        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
591
592    def test_validate_challenge_invalid(self):
593        """Test webauthn"""
594        request = self.request_factory.get("/")
595        request.user = self.user
596
597        WebAuthnDevice.objects.create(
598            user=self.user,
599            public_key=(
600                "pQECAyYgASFYIGsBLkklToCQkT7qJT_bJYN1sEc1oJdbnmoOc4"
601                "3i0JH6IlggLTXytuhzFVYYAK4PQNj8_coGrbbzSfUxdiPAcZTQCyU"
602            ),
603            credential_id="QKZ97ASJAOIDyipAs6mKUxDUZgDrWrbAsUb5leL7-oU",
604            # One more sign count than above, make it invalid
605            sign_count=5,
606            rp_id=generate_id(),
607        )
608        flow = create_test_flow()
609        stage = AuthenticatorValidateStage.objects.create(
610            name=generate_id(),
611            not_configured_action=NotConfiguredAction.CONFIGURE,
612            device_classes=[DeviceClasses.WEBAUTHN],
613        )
614        plan = FlowPlan(flow.pk.hex)
615        plan.context[PLAN_CONTEXT_WEBAUTHN_CHALLENGE] = base64url_to_bytes(
616            "g98I51mQvZXo5lxLfhrD2zfolhZbLRyCgqkkYap1jwSaJ13BguoJWCF9_Lg3AgO4Wh-Bqa556JE20oKsYbl6RA"
617        )
618        request = self.request_factory.get("/")
619
620        stage_view = AuthenticatorValidateStageView(
621            FlowExecutorView(flow=flow, current_stage=stage, plan=plan), request=request
622        )
623        request.META["SERVER_NAME"] = "localhost"
624        request.META["SERVER_PORT"] = "9000"
625        with self.assertRaises(ValidationError):
626            validate_challenge_webauthn(
627                {
628                    "id": "QKZ97ASJAOIDyipAs6mKUxDUZgDrWrbAsUb5leL7-oU",
629                    "rawId": "QKZ97ASJAOIDyipAs6mKUxDUZgDrWrbAsUb5leL7-oU",
630                    "type": "public-key",
631                    "assertionClientExtensions": "{}",
632                    "response": {
633                        "clientDataJSON": (
634                            "eyJ0eXBlIjoid2ViYXV0aG4uZ2V0IiwiY2hhbGxlbmdlIjoiZzk4STUxbVF2WlhvNWx4"
635                            "TGZockQyemZvbGhaYkxSeUNncWtrWWFwMWp3U2FKMTNCZ3VvSldDRjlfTGczQWdPNFdo"
636                            "LUJxYTU1NkpFMjBvS3NZYmw2UkEiLCJvcmlnaW4iOiJodHRwOi8vbG9jYWxob3N0Ojkw"
637                            "MDAiLCJjcm9zc09yaWdpbiI6ZmFsc2UsIm90aGVyX2tleXNfY2FuX2JlX2FkZGVkX2hl"
638                            "cmUiOiJkbyBub3QgY29tcGFyZSBjbGllbnREYXRhSlNPTiBhZ2FpbnN0IGEgdGVtcGxh"
639                            "dGUuIFNlZSBodHRwczovL2dvby5nbC95YWJQZXgifQ=="
640                        ),
641                        "signature": (
642                            "MEQCIFNlrHf9ablJAalXLWkrqvHB8oIu8kwvRpH3X3rbJVpI"
643                            "AiAqtOK6mIZPk62kZN0OzFsHfuvu_RlOl7zlqSNzDdz_Ag=="
644                        ),
645                        "authenticatorData": "SZYN5YgOjGh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2MFAAAABQ==",
646                        "userHandle": None,
647                    },
648                },
649                stage_view,
650                self.user,
651            )