stages/authenticator_validate: add ability to limit webauthn device types

authentik/stages/authenticator_validate/api.py

@@ -7,11 +7,16 @@
 from authentik.flows.api.stages import StageSerializer
 from authentik.flows.models import NotConfiguredAction
 from authentik.stages.authenticator_validate.models import AuthenticatorValidateStage
+from authentik.stages.authenticator_webauthn.api.device_types import WebAuthnDeviceTypeSerializer
 
 
 class AuthenticatorValidateStageSerializer(StageSerializer):
     """AuthenticatorValidateStage Serializer"""
 
+    webauthn_allowed_device_types_obj = WebAuthnDeviceTypeSerializer(
+        source="webauthn_allowed_device_types", many=True, read_only=True
+    )
+
     def validate_not_configured_action(self, value):
         """Ensure that a configuration stage is set when not_configured_action is configure"""
         configuration_stages = self.initial_data.get("configuration_stages", None)
@@ -31,6 +36,8 @@ class Meta:
             "configuration_stages",
             "last_auth_threshold",
             "webauthn_user_verification",
+            "webauthn_allowed_device_types",
+            "webauthn_allowed_device_types_obj",
         ]
 
 

authentik/stages/authenticator_validate/challenge.py

@@ -14,8 +14,9 @@
 from webauthn import options_to_json
 from webauthn.authentication.generate_authentication_options import generate_authentication_options
 from webauthn.authentication.verify_authentication_response import verify_authentication_response
+from webauthn.helpers import parse_authentication_credential_json
 from webauthn.helpers.base64url_to_bytes import base64url_to_bytes
-from webauthn.helpers.exceptions import InvalidAuthenticationResponse
+from webauthn.helpers.exceptions import InvalidAuthenticationResponse, InvalidJSONStructure
 from webauthn.helpers.structs import UserVerificationRequirement
 
 from authentik.core.api.utils import JSONDictField, PassiveSerializer
@@ -131,23 +132,40 @@
     """Validate WebAuthn Challenge"""
     request = stage_view.request
     challenge = request.session.get(SESSION_KEY_WEBAUTHN_CHALLENGE)
-    credential_id = data.get("id")
+    stage: AuthenticatorValidateStage = stage_view.executor.current_stage
+    try:
+        credential = parse_authentication_credential_json(data)
+    except InvalidJSONStructure as exc:
+        LOGGER.warning("Invalid WebAuthn challenge response", exc=exc)
+        raise ValidationError("Invalid device", "invalid") from None
 
-    device = WebAuthnDevice.objects.filter(credential_id=credential_id).first()
+    device = WebAuthnDevice.objects.filter(credential_id=credential.id).first()
     if not device:
-        raise ValidationError("Invalid device")
+        raise ValidationError("Invalid device", "invalid")
     # We can only check the device's user if the user we're given isn't anonymous
     # as this validation is also used for password-less login where webauthn is the very first
     # step done by a user. Only if this validation happens at a later stage we can check
     # that the device belongs to the user
     if not user.is_anonymous and device.user != user:
-        raise ValidationError("Invalid device")
-
-    stage: AuthenticatorValidateStage = stage_view.executor.current_stage
-
+        raise ValidationError("Invalid device", "invalid")
+    # When a device_type was set when creating the device (2024.4+), and we have a limitation,
+    # make sure the device type is allowed.
+    if (
+        device.device_type
+        and stage.webauthn_allowed_device_types.exists()
+        and not stage.webauthn_allowed_device_types.filter(pk=device.device_type.pk).exists()
+    ):
+        raise ValidationError(
+            _(
+                "Invalid device type. Contact your {brand} administrator for help.".format(
+                    brand=stage_view.request.brand.branding_title
+                )
+            ),
+            "invalid",
+        )
     try:
         authentication_verification = verify_authentication_response(
-            credential=data,
+            credential=credential,
             expected_challenge=challenge,
             expected_rp_id=get_rp_id(request),
             expected_origin=get_origin(request),

authentik/stages/authenticator_validate/migrations/0013_authenticatorvalidatestage_webauthn_allowed_device_types.py

@@ -0,0 +1,27 @@
+# Generated by Django 5.0.3 on 2024-04-08 18:33
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        (
+            "authentik_stages_authenticator_validate",
+            "0012_authenticatorvalidatestage_webauthn_user_verification",
+        ),
+        (
+            "authentik_stages_authenticator_webauthn",
+            "0010_webauthndevicetype_authenticatorwebauthnstage_and_more",
+        ),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="authenticatorvalidatestage",
+            name="webauthn_allowed_device_types",
+            field=models.ManyToManyField(
+                blank=True, to="authentik_stages_authenticator_webauthn.webauthndevicetype"
+            ),
+        ),
+    ]

authentik/stages/authenticator_validate/models.py

@@ -71,6 +71,9 @@ class AuthenticatorValidateStage(Stage):
         choices=UserVerification.choices,
         default=UserVerification.PREFERRED,
     )
+    webauthn_allowed_device_types = models.ManyToManyField(
+        "authentik_stages_authenticator_webauthn.WebAuthnDeviceType", blank=True
+    )
 
     @property
     def serializer(self) -> type[BaseSerializer]:

authentik/stages/authenticator_validate/stage.py

@@ -5,6 +5,7 @@
 
 from django.conf import settings
 from django.http import HttpRequest, HttpResponse
+from django.utils.translation import gettext_lazy as _
 from jwt import PyJWTError, decode, encode
 from rest_framework.fields import CharField, IntegerField, ListField, UUIDField
 from rest_framework.serializers import ValidationError
@@ -176,15 +177,30 @@ def get_device_challenges(self) -> list[dict]:
         threshold = timedelta_from_string(stage.last_auth_threshold)
         allowed_devices = []
 
+        has_webauthn_filters_set = stage.webauthn_allowed_device_types.exists()
+
         for device in user_devices:
             device_class = device.__class__.__name__.lower().replace("device", "")
             if device_class not in stage.device_classes:
                 self.logger.debug("device class not allowed", device_class=device_class)
                 continue
             if isinstance(device, SMSDevice) and device.is_hashed:
-                self.logger.debug("Hashed SMS device, skipping")
+                self.logger.debug("Hashed SMS device, skipping", device=device)
                 continue
             allowed_devices.append(device)
+            # Ignore WebAuthn devices which are not in the allowed types
+            if (
+                isinstance(device, WebAuthnDevice)
+                and device.device_type
+                and has_webauthn_filters_set
+            ):
+                if not stage.webauthn_allowed_device_types.filter(
+                    pk=device.device_type.pk
+                ).exists():
+                    self.logger.debug(
+                        "WebAuthn device type not allowed", device=device, type=device.device_type
+                    )
+                    continue
             # Ensure only one challenge per device class
             # WebAuthn does another device loop to find all WebAuthn devices
             if device_class in seen_classes:
@@ -251,7 +267,7 @@ def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:  # noqa: P
                 return self.executor.stage_ok()
             if stage.not_configured_action == NotConfiguredAction.DENY:
                 self.logger.debug("Authenticator not configured, denying")
-                return self.executor.stage_invalid()
+                return self.executor.stage_invalid(_("No (allowed) MFA authenticator configured."))
             if stage.not_configured_action == NotConfiguredAction.CONFIGURE:
                 self.logger.debug("Authenticator not configured, forcing configure")
                 return self.prepare_stages(user)