authentik.stages.authenticator_validate.tests.test_email

  1"""Test validator stage for Email devices"""
  2
  3from django.test.client import RequestFactory
  4from django.urls.base import reverse
  5
  6from authentik.core.tests.utils import create_test_admin_user, create_test_flow
  7from authentik.flows.models import FlowStageBinding, NotConfiguredAction
  8from authentik.flows.tests import FlowTestCase
  9from authentik.lib.generators import generate_id
 10from authentik.lib.utils.email import mask_email
 11from authentik.stages.authenticator_email.models import AuthenticatorEmailStage, EmailDevice
 12from authentik.stages.authenticator_validate.models import AuthenticatorValidateStage, DeviceClasses
 13from authentik.stages.identification.models import IdentificationStage, UserFields
 14
 15
 16class AuthenticatorValidateStageEmailTests(FlowTestCase):
 17    """Test validator stage for Email devices"""
 18
 19    def setUp(self) -> None:
 20        self.user = create_test_admin_user()
 21        self.request_factory = RequestFactory()
 22        # Create email authenticator stage
 23        self.stage = AuthenticatorEmailStage.objects.create(
 24            name="email-authenticator",
 25            use_global_settings=True,
 26            from_address="test@authentik.local",
 27        )
 28        # Create identification stage
 29        self.ident_stage = IdentificationStage.objects.create(
 30            name=generate_id(),
 31            user_fields=[UserFields.USERNAME],
 32        )
 33        # Create validation stage
 34        self.validate_stage = AuthenticatorValidateStage.objects.create(
 35            name=generate_id(),
 36            device_classes=[DeviceClasses.EMAIL],
 37        )
 38        # Create flow with both stages
 39        self.flow = create_test_flow()
 40        FlowStageBinding.objects.create(target=self.flow, stage=self.ident_stage, order=0)
 41        FlowStageBinding.objects.create(target=self.flow, stage=self.validate_stage, order=1)
 42
 43    def _identify_user(self):
 44        """Helper to identify user in flow"""
 45        response = self.client.post(
 46            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
 47            {"uid_field": self.user.username},
 48            follow=True,
 49        )
 50        self.assertEqual(response.status_code, 200)
 51        return response
 52
 53    def _send_challenge(self, device):
 54        """Helper to send challenge for device"""
 55        response = self.client.post(
 56            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
 57            {
 58                "component": "ak-stage-authenticator-validate",
 59                "selected_challenge": {
 60                    "device_class": "email",
 61                    "device_uid": str(device.pk),
 62                    "challenge": {},
 63                    "last_used": device.last_used.isoformat() if device.last_used else None,
 64                },
 65            },
 66        )
 67        self.assertEqual(response.status_code, 200)
 68        return response
 69
 70    def test_happy_path(self):
 71        """Test validator stage with valid code"""
 72        # Create a device for our user
 73        device = EmailDevice.objects.create(
 74            user=self.user,
 75            confirmed=True,
 76            stage=self.stage,
 77            email="xx@0.co",
 78        )  # Short email for testing purposes
 79
 80        # First identify the user
 81        self._identify_user()
 82
 83        # Send the challenge
 84        response = self._send_challenge(device)
 85        response_data = self.assertStageResponse(
 86            response,
 87            flow=self.flow,
 88            component="ak-stage-authenticator-validate",
 89        )
 90
 91        # Get the device challenge from the response and verify it matches
 92        device_challenge = response_data["device_challenges"][0]
 93        self.assertEqual(device_challenge["device_class"], "email")
 94        self.assertEqual(device_challenge["device_uid"], str(device.pk))
 95        self.assertEqual(device_challenge["challenge"], {"email": mask_email(device.email)})
 96
 97        # Generate a token for the device
 98        device.generate_token()
 99
100        # Submit the valid code
101        response = self.client.post(
102            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
103            {"component": "ak-stage-authenticator-validate", "code": device.token},
104        )
105        # Should redirect to root since this is the last stage
106        self.assertStageRedirects(response, "/")
107
108    def test_no_device(self):
109        """Test validator stage without configured device"""
110        configuration_stage = AuthenticatorEmailStage.objects.create(
111            name=generate_id(),
112            use_global_settings=True,
113            from_address="test@authentik.local",
114        )
115        stage = AuthenticatorValidateStage.objects.create(
116            name=generate_id(),
117            not_configured_action=NotConfiguredAction.CONFIGURE,
118            device_classes=[DeviceClasses.EMAIL],
119        )
120        stage.configuration_stages.set([configuration_stage])
121        flow = create_test_flow()
122        FlowStageBinding.objects.create(target=flow, stage=stage, order=2)
123
124        response = self.client.post(
125            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
126            {"component": "ak-stage-authenticator-validate"},
127        )
128        self.assertEqual(response.status_code, 200)
129        response_data = self.assertStageResponse(
130            response,
131            flow=flow,
132            component="ak-stage-authenticator-validate",
133        )
134        self.assertEqual(response_data["configuration_stages"], [])
135        self.assertEqual(response_data["device_challenges"], [])
136        self.assertEqual(
137            response_data["response_errors"],
138            {"non_field_errors": [{"code": "invalid", "string": "Empty response"}]},
139        )
140
141    def test_invalid_code(self):
142        """Test validator stage with invalid code"""
143        # Create a device for our user
144        device = EmailDevice.objects.create(
145            user=self.user,
146            confirmed=True,
147            stage=self.stage,
148            email="test@authentik.local",
149        )
150
151        # First identify the user
152        self._identify_user()
153
154        # Send the challenge
155        self._send_challenge(device)
156
157        # Generate a token for the device
158        device.generate_token()
159
160        # Try invalid code and verify error message
161        response = self.client.post(
162            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
163            {"component": "ak-stage-authenticator-validate", "code": "invalid"},
164        )
165        response_data = self.assertStageResponse(
166            response,
167            flow=self.flow,
168            component="ak-stage-authenticator-validate",
169        )
170        self.assertEqual(
171            response_data["response_errors"],
172            {
173                "code": [
174                    {
175                        "code": "invalid",
176                        "string": (
177                            "Invalid Token. Please ensure the time on your device "
178                            "is accurate and try again."
179                        ),
180                    }
181                ],
182            },
183        )