authentik.stages.authenticator_duo.tests

  1"""Test duo stage"""
  2
  3from unittest.mock import MagicMock, patch
  4from uuid import uuid4
  5
  6from django.test.client import RequestFactory
  7from django.urls import reverse
  8
  9from authentik.core.tests.utils import create_test_admin_user, create_test_flow
 10from authentik.flows.models import FlowStageBinding
 11from authentik.flows.tests import FlowTestCase
 12from authentik.lib.generators import generate_id
 13from authentik.stages.authenticator_duo.models import AuthenticatorDuoStage, DuoDevice
 14from authentik.stages.identification.models import IdentificationStage, UserFields
 15
 16
 17class AuthenticatorDuoStageTests(FlowTestCase):
 18    """Test duo stage"""
 19
 20    def setUp(self) -> None:
 21        self.user = create_test_admin_user()
 22        self.request_factory = RequestFactory()
 23
 24    def test_client(self):
 25        """Test Duo client setup"""
 26        stage = AuthenticatorDuoStage(
 27            name=generate_id(),
 28            client_id=generate_id(),
 29            client_secret=generate_id(),
 30            admin_integration_key=generate_id(),
 31            admin_secret_key=generate_id(),
 32            api_hostname=generate_id(),
 33        )
 34        self.assertEqual(stage.auth_client().ikey, stage.client_id)
 35        self.assertEqual(stage.admin_client().ikey, stage.admin_integration_key)
 36        stage.admin_integration_key = ""
 37        with self.assertRaises(ValueError):
 38            self.assertEqual(stage.admin_client().ikey, stage.admin_integration_key)
 39
 40    def test_api_enrollment_invalid(self):
 41        """Test `enrollment_status`"""
 42        self.client.force_login(self.user)
 43        response = self.client.post(
 44            reverse(
 45                "authentik_api:authenticatorduostage-enrollment-status",
 46                kwargs={
 47                    "pk": str(uuid4()),
 48                },
 49            )
 50        )
 51        self.assertEqual(response.status_code, 404)
 52
 53    def test_api_import_manual_invalid_username(self):
 54        """Test `import_device_manual`"""
 55        self.client.force_login(self.user)
 56        stage = AuthenticatorDuoStage.objects.create(
 57            name=generate_id(),
 58            client_id=generate_id(),
 59            client_secret=generate_id(),
 60            api_hostname=generate_id(),
 61        )
 62        response = self.client.post(
 63            reverse(
 64                "authentik_api:authenticatorduostage-import-device-manual",
 65                kwargs={
 66                    "pk": str(stage.pk),
 67                },
 68            ),
 69            data={
 70                "username": generate_id(),
 71            },
 72        )
 73        self.assertEqual(response.status_code, 400)
 74
 75    def test_api_import_manual_duplicate_device(self):
 76        """Test `import_device_manual`"""
 77        self.client.force_login(self.user)
 78        stage = AuthenticatorDuoStage.objects.create(
 79            name=generate_id(),
 80            client_id=generate_id(),
 81            client_secret=generate_id(),
 82            api_hostname=generate_id(),
 83        )
 84        device = DuoDevice.objects.create(
 85            name="foo",
 86            duo_user_id=generate_id(),
 87            user=self.user,
 88            stage=stage,
 89        )
 90        response = self.client.post(
 91            reverse(
 92                "authentik_api:authenticatorduostage-import-device-manual",
 93                kwargs={
 94                    "pk": str(stage.pk),
 95                },
 96            ),
 97            data={
 98                "username": self.user.username,
 99                "duo_user_id": device.duo_user_id,
100            },
101        )
102        self.assertEqual(response.status_code, 400)
103
104    def test_api_import_manual(self):
105        """Test `import_device_manual`"""
106        self.client.force_login(self.user)
107        stage = AuthenticatorDuoStage.objects.create(
108            name=generate_id(),
109            client_id=generate_id(),
110            client_secret=generate_id(),
111            api_hostname=generate_id(),
112        )
113        response = self.client.post(
114            reverse(
115                "authentik_api:authenticatorduostage-import-device-manual",
116                kwargs={
117                    "pk": str(stage.pk),
118                },
119            ),
120            data={
121                "username": self.user.username,
122                "duo_user_id": "foo",
123            },
124        )
125        self.assertEqual(response.status_code, 204)
126
127    def test_api_import_automatic_invalid(self):
128        """test `import_devices_automatic`"""
129        self.client.force_login(self.user)
130        stage = AuthenticatorDuoStage.objects.create(
131            name=generate_id(),
132            client_id=generate_id(),
133            client_secret=generate_id(),
134            api_hostname=generate_id(),
135        )
136
137        # Test missing admin credentials
138        response = self.client.post(
139            reverse(
140                "authentik_api:authenticatorduostage-import-devices-automatic",
141                kwargs={
142                    "pk": str(stage.pk),
143                },
144            ),
145        )
146        self.assertEqual(response.status_code, 400)
147
148        # Test internal error handling
149        stage.admin_integration_key = generate_id()
150        stage.admin_secret_key = generate_id()
151        stage.save()
152        with patch(
153            "duo_client.admin.Admin.get_users_iterator",
154            MagicMock(side_effect=RuntimeError("Duo API error")),
155        ):
156            response = self.client.post(
157                reverse(
158                    "authentik_api:authenticatorduostage-import-devices-automatic",
159                    kwargs={
160                        "pk": str(stage.pk),
161                    },
162                ),
163            )
164            self.assertEqual(response.status_code, 400)
165            self.assertJSONEqual(
166                response.content,
167                {
168                    "error": "An internal error occurred while importing devices.",
169                    "count": 0,
170                },
171            )
172
173    def test_api_import_automatic(self):
174        """test `import_devices_automatic`"""
175        self.client.force_login(self.user)
176        stage = AuthenticatorDuoStage.objects.create(
177            name=generate_id(),
178            client_id=generate_id(),
179            client_secret=generate_id(),
180            admin_integration_key=generate_id(),
181            admin_secret_key=generate_id(),
182            api_hostname=generate_id(),
183        )
184        device = DuoDevice.objects.create(
185            name="foo",
186            duo_user_id=generate_id(),
187            user=self.user,
188            stage=stage,
189        )
190        with patch(
191            "duo_client.admin.Admin.get_users_iterator",
192            MagicMock(
193                return_value=[
194                    {
195                        "user_id": "foo",
196                        "username": "bar",
197                    },
198                    {
199                        "user_id": device.duo_user_id,
200                        "username": self.user.username,
201                    },
202                    {
203                        "user_id": generate_id(),
204                        "username": self.user.username,
205                    },
206                ]
207            ),
208        ):
209            response = self.client.post(
210                reverse(
211                    "authentik_api:authenticatorduostage-import-devices-automatic",
212                    kwargs={
213                        "pk": str(stage.pk),
214                    },
215                ),
216            )
217            self.assertEqual(response.status_code, 200)
218            self.assertEqual(response.content.decode(), '{"error":"","count":1}')
219
220    def test_stage_enroll_basic(self):
221        """Test stage"""
222        conf_stage = IdentificationStage.objects.create(
223            name=generate_id(),
224            user_fields=[
225                UserFields.USERNAME,
226            ],
227        )
228        stage = AuthenticatorDuoStage.objects.create(
229            name=generate_id(),
230            client_id=generate_id(),
231            client_secret=generate_id(),
232            api_hostname=generate_id(),
233        )
234        flow = create_test_flow()
235        FlowStageBinding.objects.create(target=flow, stage=conf_stage, order=0)
236        FlowStageBinding.objects.create(target=flow, stage=stage, order=1)
237
238        response = self.client.post(
239            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
240            {"uid_field": self.user.username},
241        )
242        self.assertEqual(response.status_code, 302)
243
244        enroll_mock = MagicMock(
245            return_value={
246                "user_id": "foo",
247                "activation_barcode": "bar",
248                "activation_code": "bar",
249            }
250        )
251        with patch("duo_client.auth.Auth.enroll", enroll_mock):
252            response = self.client.get(
253                reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
254                follow=True,
255            )
256            self.assertStageResponse(
257                response,
258                flow,
259                component="ak-stage-authenticator-duo",
260                pending_user=self.user.username,
261                activation_barcode="bar",
262                activation_code="bar",
263            )
264
265            response = self.client.get(
266                reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
267                follow=True,
268            )
269            self.assertStageResponse(
270                response,
271                flow,
272                component="ak-stage-authenticator-duo",
273                pending_user=self.user.username,
274                activation_barcode="bar",
275                activation_code="bar",
276            )
277            self.assertEqual(enroll_mock.call_count, 1)
278
279            with patch("duo_client.auth.Auth.enroll_status", MagicMock(return_value="success")):
280                response = self.client.post(
281                    reverse(
282                        "authentik_api:authenticatorduostage-enrollment-status",
283                        kwargs={
284                            "pk": str(stage.pk),
285                        },
286                    )
287                )
288                self.assertEqual(response.status_code, 200)
289                self.assertJSONEqual(response.content, {"duo_response": "success"})
290
291                response = self.client.post(
292                    reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), {}
293                )
294                self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))