authentik.stages.authenticator_sms.tests

Test SMS API

  1"""Test SMS API"""
  2
  3from unittest.mock import MagicMock, patch
  4from urllib.parse import parse_qsl
  5
  6from django.urls import reverse
  7from requests_mock import Mocker
  8
  9from authentik.core.tests.utils import create_test_admin_user, create_test_flow
 10from authentik.flows.models import FlowStageBinding
 11from authentik.flows.planner import FlowPlan
 12from authentik.flows.tests import FlowTestCase
 13from authentik.flows.views.executor import SESSION_KEY_PLAN
 14from authentik.lib.generators import generate_id
 15from authentik.stages.authenticator_sms.models import (
 16    AuthenticatorSMSStage,
 17    SMSDevice,
 18    SMSProviders,
 19    hash_phone_number,
 20)
 21from authentik.stages.authenticator_sms.stage import PLAN_CONTEXT_PHONE, PLAN_CONTEXT_SMS_DEVICE
 22from authentik.stages.prompt.stage import PLAN_CONTEXT_PROMPT
 23
 24
 25class AuthenticatorSMSStageTests(FlowTestCase):
 26    """Test SMS API"""
 27
 28    def setUp(self) -> None:
 29        super().setUp()
 30        self.flow = create_test_flow()
 31        self.stage: AuthenticatorSMSStage = AuthenticatorSMSStage.objects.create(
 32            name="foo",
 33            provider=SMSProviders.TWILIO,
 34            configure_flow=self.flow,
 35        )
 36        FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=0)
 37        self.user = create_test_admin_user()
 38        self.client.force_login(self.user)
 39
 40    def test_stage_no_prefill(self):
 41        """test stage"""
 42        self.client.get(
 43            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
 44        )
 45        response = self.client.get(
 46            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
 47        )
 48        self.assertStageResponse(
 49            response,
 50            self.flow,
 51            self.user,
 52            component="ak-stage-authenticator-sms",
 53            phone_number_required=True,
 54        )
 55
 56    def test_stage_submit(self):
 57        """test stage (submit)"""
 58        self.client.get(
 59            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
 60        )
 61        response = self.client.get(
 62            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
 63        )
 64        self.assertStageResponse(
 65            response,
 66            self.flow,
 67            self.user,
 68            component="ak-stage-authenticator-sms",
 69            phone_number_required=True,
 70        )
 71        sms_send_mock = MagicMock()
 72        with patch(
 73            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
 74            sms_send_mock,
 75        ):
 76            response = self.client.post(
 77                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
 78                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
 79            )
 80            self.assertEqual(response.status_code, 200)
 81            sms_send_mock.assert_called_once()
 82        self.assertStageResponse(
 83            response,
 84            self.flow,
 85            self.user,
 86            component="ak-stage-authenticator-sms",
 87            response_errors={},
 88            phone_number_required=False,
 89        )
 90
 91    def test_stage_submit_twilio(self):
 92        """test stage (submit) (twilio)"""
 93        self.stage.account_sid = generate_id()
 94        self.stage.auth = generate_id()
 95        self.stage.from_number = generate_id()
 96        self.stage.save()
 97        self.client.get(
 98            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
 99        )
100        response = self.client.get(
101            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
102        )
103        self.assertStageResponse(
104            response,
105            self.flow,
106            self.user,
107            component="ak-stage-authenticator-sms",
108            phone_number_required=True,
109        )
110        number = generate_id()
111
112        with Mocker() as mocker:
113            mocker.post(
114                (
115                    "https://api.twilio.com/2010-04-01/Accounts/"
116                    f"{self.stage.account_sid}/Messages.json"
117                ),
118                json={},
119            )
120            response = self.client.post(
121                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
122                data={"component": "ak-stage-authenticator-sms", "phone_number": number},
123            )
124            self.assertEqual(response.status_code, 200)
125            self.assertEqual(mocker.call_count, 1)
126            self.assertEqual(mocker.request_history[0].method, "POST")
127            request_body = dict(parse_qsl(mocker.request_history[0].body))
128            device: SMSDevice = self.get_flow_plan().context[PLAN_CONTEXT_SMS_DEVICE]
129            self.assertEqual(
130                request_body,
131                {
132                    "To": number,
133                    "From": self.stage.from_number,
134                    "Body": f"Use this code to authenticate in authentik: {device.token}",
135                },
136            )
137        self.assertStageResponse(
138            response,
139            self.flow,
140            self.user,
141            component="ak-stage-authenticator-sms",
142            response_errors={},
143            phone_number_required=False,
144        )
145
146    def test_stage_context_data(self):
147        """test stage context data"""
148        self.client.get(
149            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
150        )
151        sms_send_mock = MagicMock()
152        with (
153            patch(
154                (
155                    "authentik.stages.authenticator_sms.stage."
156                    "AuthenticatorSMSStageView._has_phone_number"
157                ),
158                MagicMock(
159                    return_value="1234",
160                ),
161            ),
162            patch(
163                "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
164                sms_send_mock,
165            ),
166        ):
167            response = self.client.get(
168                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
169            )
170            sms_send_mock.assert_called_once()
171        self.assertStageResponse(
172            response,
173            self.flow,
174            self.user,
175            component="ak-stage-authenticator-sms",
176            phone_number_required=False,
177        )
178
179    def test_stage_context_data_duplicate(self):
180        """test stage context data (phone number exists already)"""
181        self.client.get(
182            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
183        )
184        plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
185        plan.context[PLAN_CONTEXT_PROMPT] = {
186            PLAN_CONTEXT_PHONE: "1234",
187        }
188        session = self.client.session
189        session[SESSION_KEY_PLAN] = plan
190        session.save()
191        SMSDevice.objects.create(
192            phone_number="1234",
193            user=self.user,
194            stage=self.stage,
195        )
196        sms_send_mock = MagicMock()
197        with (
198            patch(
199                "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
200                sms_send_mock,
201            ),
202        ):
203            response = self.client.get(
204                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
205            )
206        self.assertStageResponse(
207            response,
208            self.flow,
209            self.user,
210            component="ak-stage-authenticator-sms",
211            phone_number_required=True,
212        )
213        plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
214        self.assertEqual(plan.context[PLAN_CONTEXT_PROMPT], {})
215
216    def test_stage_submit_full(self):
217        """test stage (submit)"""
218        self.client.get(
219            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
220        )
221        response = self.client.get(
222            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
223        )
224        self.assertStageResponse(
225            response,
226            self.flow,
227            self.user,
228            component="ak-stage-authenticator-sms",
229            phone_number_required=True,
230        )
231        sms_send_mock = MagicMock()
232        with patch(
233            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
234            sms_send_mock,
235        ):
236            response = self.client.post(
237                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
238                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
239            )
240            self.assertEqual(response.status_code, 200)
241            sms_send_mock.assert_called_once()
242        self.assertStageResponse(
243            response,
244            self.flow,
245            self.user,
246            component="ak-stage-authenticator-sms",
247            response_errors={},
248            phone_number_required=False,
249        )
250        with patch(
251            "authentik.stages.authenticator_sms.models.SMSDevice.verify_token",
252            MagicMock(return_value=True),
253        ):
254            response = self.client.post(
255                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
256                data={
257                    "component": "ak-stage-authenticator-sms",
258                    "phone_number": "foo",
259                    "code": "123456",
260                },
261            )
262        self.assertEqual(response.status_code, 200)
263        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
264
265    def test_stage_hash(self):
266        """test stage (verify_only)"""
267        self.stage.verify_only = True
268        self.stage.save()
269        self.client.get(
270            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
271        )
272        response = self.client.get(
273            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
274        )
275        self.assertStageResponse(
276            response,
277            self.flow,
278            self.user,
279            component="ak-stage-authenticator-sms",
280            phone_number_required=True,
281        )
282        sms_send_mock = MagicMock()
283        with patch(
284            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
285            sms_send_mock,
286        ):
287            response = self.client.post(
288                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
289                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
290            )
291            self.assertEqual(response.status_code, 200)
292            sms_send_mock.assert_called_once()
293        self.assertStageResponse(
294            response,
295            self.flow,
296            self.user,
297            component="ak-stage-authenticator-sms",
298            response_errors={},
299            phone_number_required=False,
300        )
301        with patch(
302            "authentik.stages.authenticator_sms.models.SMSDevice.verify_token",
303            MagicMock(return_value=True),
304        ):
305            response = self.client.post(
306                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
307                data={
308                    "component": "ak-stage-authenticator-sms",
309                    "phone_number": "foo",
310                    "code": "123456",
311                },
312            )
313        self.assertEqual(response.status_code, 200)
314        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
315        device: SMSDevice = SMSDevice.objects.filter(user=self.user).first()
316        self.assertTrue(device.is_hashed)
317
318    def test_stage_hash_twice(self):
319        """test stage (hash + duplicate)"""
320        SMSDevice.objects.create(
321            user=create_test_admin_user(),
322            stage=self.stage,
323            phone_number=hash_phone_number("foo"),
324        )
325        self.stage.verify_only = True
326        self.stage.save()
327        self.client.get(
328            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
329        )
330        response = self.client.get(
331            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
332        )
333        self.assertStageResponse(
334            response,
335            self.flow,
336            self.user,
337            component="ak-stage-authenticator-sms",
338            phone_number_required=True,
339        )
340        sms_send_mock = MagicMock()
341        with patch(
342            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
343            sms_send_mock,
344        ):
345            response = self.client.post(
346                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
347                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
348            )
349            self.assertEqual(response.status_code, 200)
350        self.assertStageResponse(
351            response,
352            self.flow,
353            self.user,
354            component="ak-stage-authenticator-sms",
355            response_errors={
356                "non_field_errors": [{"code": "invalid", "string": "Invalid phone number"}]
357            },
358            phone_number_required=False,
359        )
class AuthenticatorSMSStageTests(authentik.flows.tests.FlowTestCase): 
 26class AuthenticatorSMSStageTests(FlowTestCase):
 27    """Test SMS API"""
 28
 29    def setUp(self) -> None:
 30        super().setUp()
 31        self.flow = create_test_flow()
 32        self.stage: AuthenticatorSMSStage = AuthenticatorSMSStage.objects.create(
 33            name="foo",
 34            provider=SMSProviders.TWILIO,
 35            configure_flow=self.flow,
 36        )
 37        FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=0)
 38        self.user = create_test_admin_user()
 39        self.client.force_login(self.user)
 40
 41    def test_stage_no_prefill(self):
 42        """test stage"""
 43        self.client.get(
 44            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
 45        )
 46        response = self.client.get(
 47            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
 48        )
 49        self.assertStageResponse(
 50            response,
 51            self.flow,
 52            self.user,
 53            component="ak-stage-authenticator-sms",
 54            phone_number_required=True,
 55        )
 56
 57    def test_stage_submit(self):
 58        """test stage (submit)"""
 59        self.client.get(
 60            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
 61        )
 62        response = self.client.get(
 63            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
 64        )
 65        self.assertStageResponse(
 66            response,
 67            self.flow,
 68            self.user,
 69            component="ak-stage-authenticator-sms",
 70            phone_number_required=True,
 71        )
 72        sms_send_mock = MagicMock()
 73        with patch(
 74            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
 75            sms_send_mock,
 76        ):
 77            response = self.client.post(
 78                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
 79                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
 80            )
 81            self.assertEqual(response.status_code, 200)
 82            sms_send_mock.assert_called_once()
 83        self.assertStageResponse(
 84            response,
 85            self.flow,
 86            self.user,
 87            component="ak-stage-authenticator-sms",
 88            response_errors={},
 89            phone_number_required=False,
 90        )
 91
 92    def test_stage_submit_twilio(self):
 93        """test stage (submit) (twilio)"""
 94        self.stage.account_sid = generate_id()
 95        self.stage.auth = generate_id()
 96        self.stage.from_number = generate_id()
 97        self.stage.save()
 98        self.client.get(
 99            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
100        )
101        response = self.client.get(
102            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
103        )
104        self.assertStageResponse(
105            response,
106            self.flow,
107            self.user,
108            component="ak-stage-authenticator-sms",
109            phone_number_required=True,
110        )
111        number = generate_id()
112
113        with Mocker() as mocker:
114            mocker.post(
115                (
116                    "https://api.twilio.com/2010-04-01/Accounts/"
117                    f"{self.stage.account_sid}/Messages.json"
118                ),
119                json={},
120            )
121            response = self.client.post(
122                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
123                data={"component": "ak-stage-authenticator-sms", "phone_number": number},
124            )
125            self.assertEqual(response.status_code, 200)
126            self.assertEqual(mocker.call_count, 1)
127            self.assertEqual(mocker.request_history[0].method, "POST")
128            request_body = dict(parse_qsl(mocker.request_history[0].body))
129            device: SMSDevice = self.get_flow_plan().context[PLAN_CONTEXT_SMS_DEVICE]
130            self.assertEqual(
131                request_body,
132                {
133                    "To": number,
134                    "From": self.stage.from_number,
135                    "Body": f"Use this code to authenticate in authentik: {device.token}",
136                },
137            )
138        self.assertStageResponse(
139            response,
140            self.flow,
141            self.user,
142            component="ak-stage-authenticator-sms",
143            response_errors={},
144            phone_number_required=False,
145        )
146
147    def test_stage_context_data(self):
148        """test stage context data"""
149        self.client.get(
150            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
151        )
152        sms_send_mock = MagicMock()
153        with (
154            patch(
155                (
156                    "authentik.stages.authenticator_sms.stage."
157                    "AuthenticatorSMSStageView._has_phone_number"
158                ),
159                MagicMock(
160                    return_value="1234",
161                ),
162            ),
163            patch(
164                "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
165                sms_send_mock,
166            ),
167        ):
168            response = self.client.get(
169                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
170            )
171            sms_send_mock.assert_called_once()
172        self.assertStageResponse(
173            response,
174            self.flow,
175            self.user,
176            component="ak-stage-authenticator-sms",
177            phone_number_required=False,
178        )
179
180    def test_stage_context_data_duplicate(self):
181        """test stage context data (phone number exists already)"""
182        self.client.get(
183            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
184        )
185        plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
186        plan.context[PLAN_CONTEXT_PROMPT] = {
187            PLAN_CONTEXT_PHONE: "1234",
188        }
189        session = self.client.session
190        session[SESSION_KEY_PLAN] = plan
191        session.save()
192        SMSDevice.objects.create(
193            phone_number="1234",
194            user=self.user,
195            stage=self.stage,
196        )
197        sms_send_mock = MagicMock()
198        with (
199            patch(
200                "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
201                sms_send_mock,
202            ),
203        ):
204            response = self.client.get(
205                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
206            )
207        self.assertStageResponse(
208            response,
209            self.flow,
210            self.user,
211            component="ak-stage-authenticator-sms",
212            phone_number_required=True,
213        )
214        plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
215        self.assertEqual(plan.context[PLAN_CONTEXT_PROMPT], {})
216
217    def test_stage_submit_full(self):
218        """test stage (submit)"""
219        self.client.get(
220            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
221        )
222        response = self.client.get(
223            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
224        )
225        self.assertStageResponse(
226            response,
227            self.flow,
228            self.user,
229            component="ak-stage-authenticator-sms",
230            phone_number_required=True,
231        )
232        sms_send_mock = MagicMock()
233        with patch(
234            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
235            sms_send_mock,
236        ):
237            response = self.client.post(
238                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
239                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
240            )
241            self.assertEqual(response.status_code, 200)
242            sms_send_mock.assert_called_once()
243        self.assertStageResponse(
244            response,
245            self.flow,
246            self.user,
247            component="ak-stage-authenticator-sms",
248            response_errors={},
249            phone_number_required=False,
250        )
251        with patch(
252            "authentik.stages.authenticator_sms.models.SMSDevice.verify_token",
253            MagicMock(return_value=True),
254        ):
255            response = self.client.post(
256                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
257                data={
258                    "component": "ak-stage-authenticator-sms",
259                    "phone_number": "foo",
260                    "code": "123456",
261                },
262            )
263        self.assertEqual(response.status_code, 200)
264        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
265
266    def test_stage_hash(self):
267        """test stage (verify_only)"""
268        self.stage.verify_only = True
269        self.stage.save()
270        self.client.get(
271            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
272        )
273        response = self.client.get(
274            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
275        )
276        self.assertStageResponse(
277            response,
278            self.flow,
279            self.user,
280            component="ak-stage-authenticator-sms",
281            phone_number_required=True,
282        )
283        sms_send_mock = MagicMock()
284        with patch(
285            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
286            sms_send_mock,
287        ):
288            response = self.client.post(
289                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
290                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
291            )
292            self.assertEqual(response.status_code, 200)
293            sms_send_mock.assert_called_once()
294        self.assertStageResponse(
295            response,
296            self.flow,
297            self.user,
298            component="ak-stage-authenticator-sms",
299            response_errors={},
300            phone_number_required=False,
301        )
302        with patch(
303            "authentik.stages.authenticator_sms.models.SMSDevice.verify_token",
304            MagicMock(return_value=True),
305        ):
306            response = self.client.post(
307                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
308                data={
309                    "component": "ak-stage-authenticator-sms",
310                    "phone_number": "foo",
311                    "code": "123456",
312                },
313            )
314        self.assertEqual(response.status_code, 200)
315        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
316        device: SMSDevice = SMSDevice.objects.filter(user=self.user).first()
317        self.assertTrue(device.is_hashed)
318
319    def test_stage_hash_twice(self):
320        """test stage (hash + duplicate)"""
321        SMSDevice.objects.create(
322            user=create_test_admin_user(),
323            stage=self.stage,
324            phone_number=hash_phone_number("foo"),
325        )
326        self.stage.verify_only = True
327        self.stage.save()
328        self.client.get(
329            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
330        )
331        response = self.client.get(
332            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
333        )
334        self.assertStageResponse(
335            response,
336            self.flow,
337            self.user,
338            component="ak-stage-authenticator-sms",
339            phone_number_required=True,
340        )
341        sms_send_mock = MagicMock()
342        with patch(
343            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
344            sms_send_mock,
345        ):
346            response = self.client.post(
347                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
348                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
349            )
350            self.assertEqual(response.status_code, 200)
351        self.assertStageResponse(
352            response,
353            self.flow,
354            self.user,
355            component="ak-stage-authenticator-sms",
356            response_errors={
357                "non_field_errors": [{"code": "invalid", "string": "Invalid phone number"}]
358            },
359            phone_number_required=False,
360        )

Test SMS API

def setUp(self) -> None: 
29    def setUp(self) -> None:
30        super().setUp()
31        self.flow = create_test_flow()
32        self.stage: AuthenticatorSMSStage = AuthenticatorSMSStage.objects.create(
33            name="foo",
34            provider=SMSProviders.TWILIO,
35            configure_flow=self.flow,
36        )
37        FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=0)
38        self.user = create_test_admin_user()
39        self.client.force_login(self.user)

Hook method for setting up the test fixture before exercising it.

def test_stage_no_prefill(self): 
41    def test_stage_no_prefill(self):
42        """test stage"""
43        self.client.get(
44            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
45        )
46        response = self.client.get(
47            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
48        )
49        self.assertStageResponse(
50            response,
51            self.flow,
52            self.user,
53            component="ak-stage-authenticator-sms",
54            phone_number_required=True,
55        )

test stage

def test_stage_submit(self): 
57    def test_stage_submit(self):
58        """test stage (submit)"""
59        self.client.get(
60            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
61        )
62        response = self.client.get(
63            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
64        )
65        self.assertStageResponse(
66            response,
67            self.flow,
68            self.user,
69            component="ak-stage-authenticator-sms",
70            phone_number_required=True,
71        )
72        sms_send_mock = MagicMock()
73        with patch(
74            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
75            sms_send_mock,
76        ):
77            response = self.client.post(
78                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
79                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
80            )
81            self.assertEqual(response.status_code, 200)
82            sms_send_mock.assert_called_once()
83        self.assertStageResponse(
84            response,
85            self.flow,
86            self.user,
87            component="ak-stage-authenticator-sms",
88            response_errors={},
89            phone_number_required=False,
90        )

test stage (submit)

def test_stage_submit_twilio(self): 
 92    def test_stage_submit_twilio(self):
 93        """test stage (submit) (twilio)"""
 94        self.stage.account_sid = generate_id()
 95        self.stage.auth = generate_id()
 96        self.stage.from_number = generate_id()
 97        self.stage.save()
 98        self.client.get(
 99            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
100        )
101        response = self.client.get(
102            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
103        )
104        self.assertStageResponse(
105            response,
106            self.flow,
107            self.user,
108            component="ak-stage-authenticator-sms",
109            phone_number_required=True,
110        )
111        number = generate_id()
112
113        with Mocker() as mocker:
114            mocker.post(
115                (
116                    "https://api.twilio.com/2010-04-01/Accounts/"
117                    f"{self.stage.account_sid}/Messages.json"
118                ),
119                json={},
120            )
121            response = self.client.post(
122                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
123                data={"component": "ak-stage-authenticator-sms", "phone_number": number},
124            )
125            self.assertEqual(response.status_code, 200)
126            self.assertEqual(mocker.call_count, 1)
127            self.assertEqual(mocker.request_history[0].method, "POST")
128            request_body = dict(parse_qsl(mocker.request_history[0].body))
129            device: SMSDevice = self.get_flow_plan().context[PLAN_CONTEXT_SMS_DEVICE]
130            self.assertEqual(
131                request_body,
132                {
133                    "To": number,
134                    "From": self.stage.from_number,
135                    "Body": f"Use this code to authenticate in authentik: {device.token}",
136                },
137            )
138        self.assertStageResponse(
139            response,
140            self.flow,
141            self.user,
142            component="ak-stage-authenticator-sms",
143            response_errors={},
144            phone_number_required=False,
145        )

test stage (submit) (twilio)

def test_stage_context_data(self): 
147    def test_stage_context_data(self):
148        """test stage context data"""
149        self.client.get(
150            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
151        )
152        sms_send_mock = MagicMock()
153        with (
154            patch(
155                (
156                    "authentik.stages.authenticator_sms.stage."
157                    "AuthenticatorSMSStageView._has_phone_number"
158                ),
159                MagicMock(
160                    return_value="1234",
161                ),
162            ),
163            patch(
164                "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
165                sms_send_mock,
166            ),
167        ):
168            response = self.client.get(
169                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
170            )
171            sms_send_mock.assert_called_once()
172        self.assertStageResponse(
173            response,
174            self.flow,
175            self.user,
176            component="ak-stage-authenticator-sms",
177            phone_number_required=False,
178        )

test stage context data

def test_stage_context_data_duplicate(self): 
180    def test_stage_context_data_duplicate(self):
181        """test stage context data (phone number exists already)"""
182        self.client.get(
183            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
184        )
185        plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
186        plan.context[PLAN_CONTEXT_PROMPT] = {
187            PLAN_CONTEXT_PHONE: "1234",
188        }
189        session = self.client.session
190        session[SESSION_KEY_PLAN] = plan
191        session.save()
192        SMSDevice.objects.create(
193            phone_number="1234",
194            user=self.user,
195            stage=self.stage,
196        )
197        sms_send_mock = MagicMock()
198        with (
199            patch(
200                "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
201                sms_send_mock,
202            ),
203        ):
204            response = self.client.get(
205                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
206            )
207        self.assertStageResponse(
208            response,
209            self.flow,
210            self.user,
211            component="ak-stage-authenticator-sms",
212            phone_number_required=True,
213        )
214        plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
215        self.assertEqual(plan.context[PLAN_CONTEXT_PROMPT], {})

test stage context data (phone number exists already)

def test_stage_submit_full(self): 
217    def test_stage_submit_full(self):
218        """test stage (submit)"""
219        self.client.get(
220            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
221        )
222        response = self.client.get(
223            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
224        )
225        self.assertStageResponse(
226            response,
227            self.flow,
228            self.user,
229            component="ak-stage-authenticator-sms",
230            phone_number_required=True,
231        )
232        sms_send_mock = MagicMock()
233        with patch(
234            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
235            sms_send_mock,
236        ):
237            response = self.client.post(
238                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
239                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
240            )
241            self.assertEqual(response.status_code, 200)
242            sms_send_mock.assert_called_once()
243        self.assertStageResponse(
244            response,
245            self.flow,
246            self.user,
247            component="ak-stage-authenticator-sms",
248            response_errors={},
249            phone_number_required=False,
250        )
251        with patch(
252            "authentik.stages.authenticator_sms.models.SMSDevice.verify_token",
253            MagicMock(return_value=True),
254        ):
255            response = self.client.post(
256                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
257                data={
258                    "component": "ak-stage-authenticator-sms",
259                    "phone_number": "foo",
260                    "code": "123456",
261                },
262            )
263        self.assertEqual(response.status_code, 200)
264        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))

test stage (submit)

def test_stage_hash(self): 
266    def test_stage_hash(self):
267        """test stage (verify_only)"""
268        self.stage.verify_only = True
269        self.stage.save()
270        self.client.get(
271            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
272        )
273        response = self.client.get(
274            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
275        )
276        self.assertStageResponse(
277            response,
278            self.flow,
279            self.user,
280            component="ak-stage-authenticator-sms",
281            phone_number_required=True,
282        )
283        sms_send_mock = MagicMock()
284        with patch(
285            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
286            sms_send_mock,
287        ):
288            response = self.client.post(
289                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
290                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
291            )
292            self.assertEqual(response.status_code, 200)
293            sms_send_mock.assert_called_once()
294        self.assertStageResponse(
295            response,
296            self.flow,
297            self.user,
298            component="ak-stage-authenticator-sms",
299            response_errors={},
300            phone_number_required=False,
301        )
302        with patch(
303            "authentik.stages.authenticator_sms.models.SMSDevice.verify_token",
304            MagicMock(return_value=True),
305        ):
306            response = self.client.post(
307                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
308                data={
309                    "component": "ak-stage-authenticator-sms",
310                    "phone_number": "foo",
311                    "code": "123456",
312                },
313            )
314        self.assertEqual(response.status_code, 200)
315        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
316        device: SMSDevice = SMSDevice.objects.filter(user=self.user).first()
317        self.assertTrue(device.is_hashed)

test stage (verify_only)

def test_stage_hash_twice(self): 
319    def test_stage_hash_twice(self):
320        """test stage (hash + duplicate)"""
321        SMSDevice.objects.create(
322            user=create_test_admin_user(),
323            stage=self.stage,
324            phone_number=hash_phone_number("foo"),
325        )
326        self.stage.verify_only = True
327        self.stage.save()
328        self.client.get(
329            reverse("authentik_flows:configure", kwargs={"stage_uuid": self.stage.stage_uuid}),
330        )
331        response = self.client.get(
332            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
333        )
334        self.assertStageResponse(
335            response,
336            self.flow,
337            self.user,
338            component="ak-stage-authenticator-sms",
339            phone_number_required=True,
340        )
341        sms_send_mock = MagicMock()
342        with patch(
343            "authentik.stages.authenticator_sms.models.AuthenticatorSMSStage.send",
344            sms_send_mock,
345        ):
346            response = self.client.post(
347                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
348                data={"component": "ak-stage-authenticator-sms", "phone_number": "foo"},
349            )
350            self.assertEqual(response.status_code, 200)
351        self.assertStageResponse(
352            response,
353            self.flow,
354            self.user,
355            component="ak-stage-authenticator-sms",
356            response_errors={
357                "non_field_errors": [{"code": "invalid", "string": "Invalid phone number"}]
358            },
359            phone_number_required=False,
360        )

test stage (hash + duplicate)

Inherited Members
authentik.flows.tests.FlowTestCase
assertStageResponse
get_flow_plan
set_flow_plan
assertStageRedirects
assertFlowFinishes