authentik.flows.tests.test_executor

  1"""flow views tests"""
  2
  3from datetime import timedelta
  4from unittest.mock import MagicMock, PropertyMock, patch
  5from urllib.parse import urlencode
  6
  7from django.http import HttpRequest, HttpResponse
  8from django.test import override_settings
  9from django.test.client import RequestFactory
 10from django.urls import reverse
 11from django.utils.timezone import now
 12from rest_framework.exceptions import ParseError
 13
 14from authentik.core.models import Group, User
 15from authentik.core.tests.utils import create_test_flow, create_test_user
 16from authentik.flows.markers import ReevaluateMarker, StageMarker
 17from authentik.flows.models import (
 18    FlowAuthenticationRequirement,
 19    FlowDeniedAction,
 20    FlowDesignation,
 21    FlowStageBinding,
 22    FlowToken,
 23    InvalidResponseAction,
 24)
 25from authentik.flows.planner import FlowPlan, FlowPlanner
 26from authentik.flows.stage import PLAN_CONTEXT_PENDING_USER_IDENTIFIER, StageView
 27from authentik.flows.tests import FlowTestCase
 28from authentik.flows.views.executor import (
 29    NEXT_ARG_NAME,
 30    QS_KEY_TOKEN,
 31    QS_QUERY,
 32    SESSION_KEY_PLAN,
 33    FlowExecutorView,
 34)
 35from authentik.lib.generators import generate_id
 36from authentik.policies.dummy.models import DummyPolicy
 37from authentik.policies.models import PolicyBinding
 38from authentik.policies.reputation.models import ReputationPolicy
 39from authentik.policies.types import PolicyResult
 40from authentik.stages.deny.models import DenyStage
 41from authentik.stages.dummy.models import DummyStage
 42from authentik.stages.identification.models import IdentificationStage, UserFields
 43from authentik.stages.password.models import PasswordStage
 44
 45POLICY_RETURN_FALSE = PropertyMock(return_value=PolicyResult(False, "foo"))
 46POLICY_RETURN_TRUE = MagicMock(return_value=PolicyResult(True))
 47
 48
 49def to_stage_response(request: HttpRequest, source: HttpResponse):
 50    """Mock for to_stage_response that returns the original response, so we can check
 51    inheritance and member attributes"""
 52    return source
 53
 54
 55TO_STAGE_RESPONSE_MOCK = MagicMock(side_effect=to_stage_response)
 56
 57
 58class TestFlowExecutor(FlowTestCase):
 59    """Test executor"""
 60
 61    def setUp(self):
 62        self.request_factory = RequestFactory()
 63
 64    @patch(
 65        "authentik.flows.views.executor.to_stage_response",
 66        TO_STAGE_RESPONSE_MOCK,
 67    )
 68    def test_existing_plan_diff_flow(self):
 69        """Check that a plan for a different flow cancels the current plan"""
 70        flow = create_test_flow(
 71            FlowDesignation.AUTHENTICATION,
 72        )
 73        stage = DummyStage.objects.create(name=generate_id())
 74        binding = FlowStageBinding(target=flow, stage=stage, order=0)
 75        plan = FlowPlan(flow_pk=flow.pk.hex + "a", bindings=[binding], markers=[StageMarker()])
 76        session = self.client.session
 77        session[SESSION_KEY_PLAN] = plan
 78        session.save()
 79
 80        cancel_mock = MagicMock()
 81        with patch("authentik.flows.views.executor.FlowExecutorView.cancel", cancel_mock):
 82            response = self.client.get(
 83                reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
 84            )
 85            self.assertEqual(response.status_code, 302)
 86            self.assertEqual(cancel_mock.call_count, 2)
 87
 88    @patch(
 89        "authentik.flows.views.executor.to_stage_response",
 90        TO_STAGE_RESPONSE_MOCK,
 91    )
 92    @patch(
 93        "authentik.policies.engine.PolicyEngine.result",
 94        POLICY_RETURN_FALSE,
 95    )
 96    def test_invalid_non_applicable_flow(self):
 97        """Tests that a non-applicable flow returns the correct error message"""
 98        flow = create_test_flow(
 99            FlowDesignation.AUTHENTICATION,
100        )
101
102        response = self.client.get(
103            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
104        )
105        self.assertStageResponse(
106            response,
107            flow=flow,
108            error_message="foo",
109            component="ak-stage-access-denied",
110        )
111
112    @patch(
113        "authentik.flows.views.executor.to_stage_response",
114        TO_STAGE_RESPONSE_MOCK,
115    )
116    @patch(
117        "authentik.policies.engine.PolicyEngine.result",
118        POLICY_RETURN_FALSE,
119    )
120    def test_invalid_non_applicable_flow_continue(self):
121        """Tests that a non-applicable flow that should redirect"""
122        flow = create_test_flow(
123            FlowDesignation.AUTHENTICATION,
124            denied_action=FlowDeniedAction.CONTINUE,
125        )
126
127        response = self.client.get(
128            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
129        )
130        self.assertEqual(response.status_code, 302)
131        self.assertEqual(response.url, reverse("authentik_core:root-redirect"))
132
133    @patch(
134        "authentik.flows.views.executor.to_stage_response",
135        TO_STAGE_RESPONSE_MOCK,
136    )
137    def test_invalid_flow_redirect(self):
138        """Test invalid flow with valid redirect destination"""
139        flow = create_test_flow(
140            FlowDesignation.AUTHENTICATION,
141        )
142
143        dest = "/unique-string"
144        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
145        response = self.client.get(url + f"?{QS_QUERY}={urlencode({NEXT_ARG_NAME: dest})}")
146        self.assertEqual(response.status_code, 302)
147        self.assertEqual(response.url, "/unique-string")
148
149    @patch(
150        "authentik.flows.views.executor.to_stage_response",
151        TO_STAGE_RESPONSE_MOCK,
152    )
153    def test_invalid_flow_invalid_redirect(self):
154        """Test invalid flow redirect with an invalid URL"""
155        flow = create_test_flow(
156            FlowDesignation.AUTHENTICATION,
157        )
158
159        dest = "http://something.example.com/unique-string"
160        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
161
162        response = self.client.get(url + f"?{QS_QUERY}={urlencode({NEXT_ARG_NAME: dest})}")
163        self.assertEqual(response.status_code, 200)
164        self.assertStageResponse(
165            response,
166            flow,
167            component="ak-stage-access-denied",
168            error_message="Invalid next URL",
169        )
170
171    @patch(
172        "authentik.flows.views.executor.to_stage_response",
173        TO_STAGE_RESPONSE_MOCK,
174    )
175    def test_valid_flow_redirect(self):
176        """Test valid flow with valid redirect destination"""
177        flow = create_test_flow()
178
179        dest = "/unique-string"
180        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
181
182        response = self.client.get(url + f"?{QS_QUERY}={urlencode({NEXT_ARG_NAME: dest})}")
183        self.assertEqual(response.status_code, 302)
184        self.assertEqual(response.url, "/unique-string")
185
186    @patch(
187        "authentik.flows.views.executor.to_stage_response",
188        TO_STAGE_RESPONSE_MOCK,
189    )
190    def test_valid_flow_redirect_authenticated(self):
191        """Test valid flow with valid redirect destination, authenticated already"""
192        flow = create_test_flow()
193        flow.designation = FlowDesignation.AUTHENTICATION
194        flow.authentication = FlowAuthenticationRequirement.REQUIRE_UNAUTHENTICATED
195        flow.save()
196        self.client.force_login(create_test_user())
197
198        dest = "/unique-string"
199        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
200
201        response = self.client.get(url + f"?{QS_QUERY}={urlencode({NEXT_ARG_NAME: dest})}")
202        self.assertEqual(response.status_code, 302)
203        self.assertEqual(response.url, "/unique-string")
204
205    @patch(
206        "authentik.flows.views.executor.to_stage_response",
207        TO_STAGE_RESPONSE_MOCK,
208    )
209    def test_valid_flow_invalid_redirect(self):
210        """Test valid flow redirect with an invalid URL"""
211        flow = create_test_flow()
212
213        dest = "http://something.example.com/unique-string"
214        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
215
216        response = self.client.get(url + f"?{QS_QUERY}={urlencode({NEXT_ARG_NAME: dest})}")
217        self.assertEqual(response.status_code, 200)
218        self.assertStageResponse(
219            response,
220            flow,
221            component="ak-stage-access-denied",
222            error_message="Invalid next URL",
223        )
224
225    @patch(
226        "authentik.flows.views.executor.to_stage_response",
227        TO_STAGE_RESPONSE_MOCK,
228    )
229    def test_invalid_empty_flow(self):
230        """Tests that an empty flow returns the correct error message"""
231        flow = create_test_flow(
232            FlowDesignation.AUTHENTICATION,
233        )
234
235        response = self.client.get(
236            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
237        )
238        self.assertEqual(response.status_code, 302)
239        self.assertEqual(response.url, reverse("authentik_core:root-redirect"))
240
241    def test_multi_stage_flow(self):
242        """Test a full flow with multiple stages"""
243        flow = create_test_flow(
244            FlowDesignation.AUTHENTICATION,
245        )
246        FlowStageBinding.objects.create(
247            target=flow, stage=DummyStage.objects.create(name=generate_id()), order=0
248        )
249        FlowStageBinding.objects.create(
250            target=flow, stage=DummyStage.objects.create(name=generate_id()), order=1
251        )
252
253        exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
254        # First Request, start planning, renders form
255        response = self.client.get(exec_url)
256        self.assertEqual(response.status_code, 200)
257        # Check that two stages are in plan
258        session = self.client.session
259        plan: FlowPlan = session[SESSION_KEY_PLAN]
260        self.assertEqual(len(plan.bindings), 2)
261        # Second request, submit form, one stage left
262        response = self.client.post(exec_url)
263        # Second request redirects to the same URL
264        self.assertEqual(response.status_code, 302)
265        self.assertEqual(response.url, exec_url)
266        # Check that two stages are in plan
267        session = self.client.session
268        plan: FlowPlan = session[SESSION_KEY_PLAN]
269        self.assertEqual(len(plan.bindings), 1)
270
271    @patch(
272        "authentik.flows.views.executor.to_stage_response",
273        TO_STAGE_RESPONSE_MOCK,
274    )
275    def test_reevaluate_remove_last(self):
276        """Test planner with re-evaluate (last stage is removed)"""
277        flow = create_test_flow(
278            FlowDesignation.AUTHENTICATION,
279        )
280        false_policy = DummyPolicy.objects.create(
281            name=generate_id(), result=False, wait_min=1, wait_max=2
282        )
283
284        binding = FlowStageBinding.objects.create(
285            target=flow,
286            stage=DummyStage.objects.create(name=generate_id()),
287            order=0,
288            evaluate_on_plan=True,
289            re_evaluate_policies=False,
290        )
291        binding2 = FlowStageBinding.objects.create(
292            target=flow,
293            stage=DummyStage.objects.create(name=generate_id()),
294            order=1,
295            re_evaluate_policies=True,
296        )
297
298        PolicyBinding.objects.create(policy=false_policy, target=binding2, order=0)
299
300        # Here we patch the dummy policy to evaluate to true so the stage is included
301        with patch("authentik.policies.dummy.models.DummyPolicy.passes", POLICY_RETURN_TRUE):
302            exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
303            # First request, run the planner
304            response = self.client.get(exec_url)
305            self.assertEqual(response.status_code, 200)
306
307            plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
308
309            self.assertEqual(plan.bindings[0], binding)
310            self.assertEqual(plan.bindings[1], binding2)
311
312            self.assertEqual(plan.markers[0].__class__, StageMarker)
313            self.assertEqual(plan.markers[1].__class__, ReevaluateMarker)
314
315            # Second request, this passes the first dummy stage
316            response = self.client.post(exec_url)
317            self.assertEqual(response.status_code, 302)
318
319        # third request, this should trigger the re-evaluate
320        # We do this request without the patch, so the policy results in false
321        response = self.client.post(exec_url)
322        self.assertEqual(response.status_code, 302)
323        self.assertEqual(response.url, reverse("authentik_core:root-redirect"))
324
325    def test_reevaluate_remove_middle(self):
326        """Test planner with re-evaluate (middle stage is removed)"""
327        flow = create_test_flow(
328            FlowDesignation.AUTHENTICATION,
329        )
330        false_policy = DummyPolicy.objects.create(
331            name=generate_id(), result=False, wait_min=1, wait_max=2
332        )
333
334        binding = FlowStageBinding.objects.create(
335            target=flow,
336            stage=DummyStage.objects.create(name=generate_id()),
337            order=0,
338            evaluate_on_plan=True,
339            re_evaluate_policies=False,
340        )
341        binding2 = FlowStageBinding.objects.create(
342            target=flow,
343            stage=DummyStage.objects.create(name=generate_id()),
344            order=1,
345            re_evaluate_policies=True,
346        )
347        binding3 = FlowStageBinding.objects.create(
348            target=flow,
349            stage=DummyStage.objects.create(name=generate_id()),
350            order=2,
351            evaluate_on_plan=True,
352            re_evaluate_policies=False,
353        )
354
355        PolicyBinding.objects.create(policy=false_policy, target=binding2, order=0)
356
357        # Here we patch the dummy policy to evaluate to true so the stage is included
358        with patch("authentik.policies.dummy.models.DummyPolicy.passes", POLICY_RETURN_TRUE):
359            exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
360            # First request, run the planner
361            response = self.client.get(exec_url)
362
363            self.assertEqual(response.status_code, 200)
364            plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
365
366            self.assertEqual(plan.bindings[0], binding)
367            self.assertEqual(plan.bindings[1], binding2)
368            self.assertEqual(plan.bindings[2], binding3)
369
370            self.assertEqual(plan.markers[0].__class__, StageMarker)
371            self.assertEqual(plan.markers[1].__class__, ReevaluateMarker)
372            self.assertEqual(plan.markers[2].__class__, StageMarker)
373
374            # Second request, this passes the first dummy stage
375            response = self.client.post(exec_url)
376            self.assertEqual(response.status_code, 302)
377
378            plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
379
380            self.assertEqual(plan.bindings[0], binding2)
381            self.assertEqual(plan.bindings[1], binding3)
382
383            self.assertEqual(plan.markers[0].__class__, ReevaluateMarker)
384            self.assertEqual(plan.markers[1].__class__, StageMarker)
385
386        # third request, this should trigger the re-evaluate
387        # We do this request without the patch, so the policy results in false
388        response = self.client.post(exec_url)
389        self.assertEqual(response.status_code, 200)
390        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
391
392    def test_reevaluate_keep(self):
393        """Test planner with re-evaluate (everything is kept)"""
394        flow = create_test_flow(
395            FlowDesignation.AUTHENTICATION,
396        )
397        true_policy = DummyPolicy.objects.create(
398            name=generate_id(), result=True, wait_min=1, wait_max=2
399        )
400
401        binding = FlowStageBinding.objects.create(
402            target=flow,
403            stage=DummyStage.objects.create(name=generate_id()),
404            order=0,
405            evaluate_on_plan=True,
406            re_evaluate_policies=False,
407        )
408        binding2 = FlowStageBinding.objects.create(
409            target=flow,
410            stage=DummyStage.objects.create(name=generate_id()),
411            order=1,
412            re_evaluate_policies=True,
413        )
414        binding3 = FlowStageBinding.objects.create(
415            target=flow,
416            stage=DummyStage.objects.create(name=generate_id()),
417            order=2,
418            evaluate_on_plan=True,
419            re_evaluate_policies=False,
420        )
421
422        PolicyBinding.objects.create(policy=true_policy, target=binding2, order=0)
423
424        # Here we patch the dummy policy to evaluate to true so the stage is included
425        with patch("authentik.policies.dummy.models.DummyPolicy.passes", POLICY_RETURN_TRUE):
426            exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
427            # First request, run the planner
428            response = self.client.get(exec_url)
429
430            self.assertEqual(response.status_code, 200)
431            plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
432
433            self.assertEqual(plan.bindings[0], binding)
434            self.assertEqual(plan.bindings[1], binding2)
435            self.assertEqual(plan.bindings[2], binding3)
436
437            self.assertEqual(plan.markers[0].__class__, StageMarker)
438            self.assertEqual(plan.markers[1].__class__, ReevaluateMarker)
439            self.assertEqual(plan.markers[2].__class__, StageMarker)
440
441            # Second request, this passes the first dummy stage
442            response = self.client.post(exec_url)
443            self.assertEqual(response.status_code, 302)
444
445            plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
446
447            self.assertEqual(plan.bindings[0], binding2)
448            self.assertEqual(plan.bindings[1], binding3)
449
450            self.assertEqual(plan.markers[0].__class__, ReevaluateMarker)
451            self.assertEqual(plan.markers[1].__class__, StageMarker)
452
453            # Third request, this passes the first dummy stage
454            response = self.client.post(exec_url)
455            self.assertEqual(response.status_code, 302)
456
457            plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
458
459            self.assertEqual(plan.bindings[0], binding3)
460
461            self.assertEqual(plan.markers[0].__class__, StageMarker)
462
463        # third request, this should trigger the re-evaluate
464        # We do this request without the patch, so the policy results in false
465        response = self.client.post(exec_url)
466        self.assertEqual(response.status_code, 200)
467        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
468
469    def test_reevaluate_remove_consecutive(self):
470        """Test planner with re-evaluate (consecutive stages are removed)"""
471        flow = create_test_flow(
472            FlowDesignation.AUTHENTICATION,
473        )
474        false_policy = DummyPolicy.objects.create(
475            name=generate_id(), result=False, wait_min=1, wait_max=2
476        )
477
478        binding = FlowStageBinding.objects.create(
479            target=flow,
480            stage=DummyStage.objects.create(name=generate_id()),
481            order=0,
482            evaluate_on_plan=True,
483            re_evaluate_policies=False,
484        )
485        binding2 = FlowStageBinding.objects.create(
486            target=flow,
487            stage=DummyStage.objects.create(name=generate_id()),
488            order=1,
489            re_evaluate_policies=True,
490        )
491        binding3 = FlowStageBinding.objects.create(
492            target=flow,
493            stage=DummyStage.objects.create(name=generate_id()),
494            order=2,
495            re_evaluate_policies=True,
496        )
497        binding4 = FlowStageBinding.objects.create(
498            target=flow,
499            stage=DummyStage.objects.create(name=generate_id()),
500            order=2,
501            evaluate_on_plan=True,
502            re_evaluate_policies=False,
503        )
504
505        PolicyBinding.objects.create(policy=false_policy, target=binding2, order=0)
506        PolicyBinding.objects.create(policy=false_policy, target=binding3, order=0)
507
508        # Here we patch the dummy policy to evaluate to true so the stage is included
509        with patch("authentik.policies.dummy.models.DummyPolicy.passes", POLICY_RETURN_TRUE):
510            exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
511            # First request, run the planner
512            response = self.client.get(exec_url)
513            self.assertEqual(response.status_code, 200)
514            self.assertStageResponse(response, flow, component="ak-stage-dummy")
515
516            plan: FlowPlan = self.client.session[SESSION_KEY_PLAN]
517
518            self.assertEqual(plan.bindings[0], binding)
519            self.assertEqual(plan.bindings[1], binding2)
520            self.assertEqual(plan.bindings[2], binding3)
521            self.assertEqual(plan.bindings[3], binding4)
522
523            self.assertEqual(plan.markers[0].__class__, StageMarker)
524            self.assertEqual(plan.markers[1].__class__, ReevaluateMarker)
525            self.assertEqual(plan.markers[2].__class__, ReevaluateMarker)
526            self.assertEqual(plan.markers[3].__class__, StageMarker)
527
528        # Second request, this passes the first dummy stage
529        response = self.client.post(exec_url)
530        self.assertEqual(response.status_code, 302)
531
532        # third request, this should trigger the re-evaluate
533        # A get request will evaluate the policies and this will return stage 4
534        # but it won't save it, hence we can't check the plan
535        response = self.client.get(exec_url)
536        self.assertStageResponse(response, flow, component="ak-stage-dummy")
537
538        # fourth request, this confirms the last stage (dummy4)
539        # We do this request without the patch, so the policy results in false
540        response = self.client.post(exec_url)
541        self.assertEqual(response.status_code, 200)
542        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
543
544    def test_stageview_user_identifier(self):
545        """Test PLAN_CONTEXT_PENDING_USER_IDENTIFIER"""
546        flow = create_test_flow(
547            FlowDesignation.AUTHENTICATION,
548        )
549        FlowStageBinding.objects.create(
550            target=flow, stage=DummyStage.objects.create(name=generate_id()), order=0
551        )
552
553        ident = "test-identifier"
554
555        user = User.objects.create(username="test-user")
556        request = self.request_factory.get(
557            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
558        )
559        request.user = user
560        planner = FlowPlanner(flow)
561        plan = planner.plan(request, default_context={PLAN_CONTEXT_PENDING_USER_IDENTIFIER: ident})
562
563        executor = FlowExecutorView()
564        executor.plan = plan
565        executor.flow = flow
566
567        stage_view = StageView(executor)
568        self.assertEqual(ident, stage_view.get_pending_user(for_display=True).username)
569
570    def test_invalid_restart(self):
571        """Test flow that restarts on invalid entry"""
572        flow = create_test_flow(
573            FlowDesignation.AUTHENTICATION,
574        )
575        # Stage 0 is a deny stage that is added dynamically
576        # when the reputation policy says so
577        deny_stage = DenyStage.objects.create(name=generate_id())
578        reputation_policy = ReputationPolicy.objects.create(
579            name=generate_id(), threshold=-1, check_ip=False
580        )
581        deny_binding = FlowStageBinding.objects.create(
582            target=flow,
583            stage=deny_stage,
584            order=0,
585            evaluate_on_plan=False,
586            re_evaluate_policies=True,
587        )
588        PolicyBinding.objects.create(policy=reputation_policy, target=deny_binding, order=0)
589
590        # Stage 1 is an identification stage
591        ident_stage = IdentificationStage.objects.create(
592            name=generate_id(),
593            user_fields=[UserFields.E_MAIL],
594            pretend_user_exists=False,
595        )
596        FlowStageBinding.objects.create(
597            target=flow,
598            stage=ident_stage,
599            order=1,
600            invalid_response_action=InvalidResponseAction.RESTART_WITH_CONTEXT,
601        )
602        exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
603        # First request, run the planner
604        response = self.client.get(exec_url)
605        self.assertStageResponse(
606            response,
607            flow,
608            component="ak-stage-identification",
609            password_fields=False,
610            primary_action="Log in",
611            sources=[],
612            show_source_labels=False,
613            user_fields=[UserFields.E_MAIL],
614        )
615        response = self.client.post(exec_url, {"uid_field": "invalid-string"}, follow=True)
616        self.assertStageResponse(response, flow, component="ak-stage-access-denied")
617
618    def test_re_evaluate_group_binding(self):
619        """Test re-evaluate stage binding that has a policy binding to a group"""
620        flow = create_test_flow()
621
622        user_group_membership = create_test_user()
623        user_direct_binding = create_test_user()
624        user_other = create_test_user()
625
626        group_a = Group.objects.create(name=generate_id())
627        user_group_membership.groups.add(group_a)
628
629        # Stage 0 is an identification stage
630        ident_stage = IdentificationStage.objects.create(
631            name=generate_id(),
632            user_fields=[UserFields.USERNAME],
633            pretend_user_exists=False,
634        )
635        FlowStageBinding.objects.create(
636            target=flow,
637            stage=ident_stage,
638            order=0,
639        )
640
641        # Stage 1 is a dummy stage that is only shown for users in group_a
642        dummy_stage = DummyStage.objects.create(name=generate_id())
643        dummy_binding = FlowStageBinding.objects.create(target=flow, stage=dummy_stage, order=1)
644        PolicyBinding.objects.create(group=group_a, target=dummy_binding, order=0)
645        PolicyBinding.objects.create(user=user_direct_binding, target=dummy_binding, order=0)
646
647        # Stage 2 is a deny stage that (in this case) only user_b will see
648        deny_stage = DenyStage.objects.create(name=generate_id())
649        FlowStageBinding.objects.create(target=flow, stage=deny_stage, order=2)
650
651        exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
652
653        with self.subTest(f"Test user access through group: {user_group_membership}"):
654            self.client.logout()
655            # First request, run the planner
656            response = self.client.get(exec_url)
657            self.assertStageResponse(response, flow, component="ak-stage-identification")
658            response = self.client.post(
659                exec_url, {"uid_field": user_group_membership.username}, follow=True
660            )
661            self.assertStageResponse(response, flow, component="ak-stage-dummy")
662        with self.subTest(f"Test user access through user: {user_direct_binding}"):
663            self.client.logout()
664            # First request, run the planner
665            response = self.client.get(exec_url)
666            self.assertStageResponse(response, flow, component="ak-stage-identification")
667            response = self.client.post(
668                exec_url, {"uid_field": user_direct_binding.username}, follow=True
669            )
670            self.assertStageResponse(response, flow, component="ak-stage-dummy")
671        with self.subTest(f"Test user has no access: {user_other}"):
672            self.client.logout()
673            # First request, run the planner
674            response = self.client.get(exec_url)
675            self.assertStageResponse(response, flow, component="ak-stage-identification")
676            response = self.client.post(exec_url, {"uid_field": user_other.username}, follow=True)
677            self.assertStageResponse(response, flow, component="ak-stage-access-denied")
678
679    @patch(
680        "authentik.flows.views.executor.to_stage_response",
681        TO_STAGE_RESPONSE_MOCK,
682    )
683    def test_invalid_json(self):
684        """Test invalid JSON body"""
685        flow = create_test_flow()
686        FlowStageBinding.objects.create(
687            target=flow, stage=DummyStage.objects.create(name=generate_id()), order=0
688        )
689        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
690
691        with override_settings(TEST=False, DEBUG=False):
692            self.client.logout()
693            response = self.client.post(url, data="{", content_type="application/json")
694            self.assertEqual(response.status_code, 200)
695
696        with self.assertRaises(ParseError):
697            self.client.logout()
698            response = self.client.post(url, data="{", content_type="application/json")
699            self.assertEqual(response.status_code, 200)
700
701    def test_cancel_next(self):
702        """Test cancel URL with ?next param set"""
703        flow = create_test_flow()
704
705        # Stage 0 is an identification stage
706        ident_stage = IdentificationStage.objects.create(
707            name=generate_id(),
708            user_fields=[UserFields.USERNAME],
709        )
710        FlowStageBinding.objects.create(
711            target=flow,
712            stage=ident_stage,
713            order=0,
714        )
715
716        # Stage 1 is a password stage
717        password_stage = PasswordStage.objects.create(name=generate_id(), backends=[])
718        FlowStageBinding.objects.create(
719            target=flow,
720            stage=password_stage,
721            order=1,
722        )
723        res = self.client.get(
724            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
725            + f"?{urlencode({QS_QUERY: urlencode({NEXT_ARG_NAME: "/foo"})})}"
726        )
727        self.assertStageResponse(res, flow, component="ak-stage-identification")
728
729        res = self.client.post(
730            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
731            + f"?{urlencode({QS_QUERY: urlencode({NEXT_ARG_NAME: "/foo"})})}",
732            data={"component": "ak-stage-identification", "uid_field": generate_id()},
733            follow=True,
734        )
735        self.assertEqual(res.status_code, 200)
736        self.assertStageResponse(
737            res,
738            flow,
739            flow_info={
740                "background": "/static/dist/assets/images/flow_background.jpg",
741                "background_themed_urls": None,
742                "cancel_url": "/flows/-/cancel/?next=%2Ffoo",
743                "layout": "stacked",
744                "title": flow.title,
745            },
746        )
747
748    @patch(
749        "authentik.flows.views.executor.to_stage_response",
750        TO_STAGE_RESPONSE_MOCK,
751    )
752    def test_expired_flow_token(self):
753        """Test that an expired flow token shows an appropriate error message"""
754        flow = create_test_flow(
755            FlowDesignation.RECOVERY,
756            authentication=FlowAuthenticationRequirement.REQUIRE_TOKEN,
757        )
758        user = create_test_user()
759        plan = FlowPlan(flow_pk=flow.pk.hex, bindings=[], markers=[])
760
761        token = FlowToken.objects.create(
762            user=user,
763            identifier=generate_id(),
764            flow=flow,
765            _plan=FlowToken.pickle(plan),
766            expires=now() - timedelta(hours=1),
767        )
768
769        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
770        response = self.client.get(
771            url + f"?{urlencode({QS_QUERY: urlencode({QS_KEY_TOKEN: token.key})})}"
772        )
773        self.assertStageResponse(
774            response,
775            flow,
776            component="ak-stage-access-denied",
777            error_message="This link is invalid or has expired. Please request a new one.",
778        )
779
780    @patch(
781        "authentik.flows.views.executor.to_stage_response",
782        TO_STAGE_RESPONSE_MOCK,
783    )
784    def test_invalid_flow_token_require_token(self):
785        """Test that an invalid/nonexistent token on a REQUIRE_TOKEN flow shows error"""
786        flow = create_test_flow(
787            FlowDesignation.RECOVERY,
788            authentication=FlowAuthenticationRequirement.REQUIRE_TOKEN,
789        )
790
791        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
792        response = self.client.get(
793            url + f"?{urlencode({QS_QUERY: urlencode({QS_KEY_TOKEN: 'invalid-token'})})}"
794        )
795        self.assertStageResponse(
796            response,
797            flow,
798            component="ak-stage-access-denied",
799            error_message="This link is invalid or has expired. Please request a new one.",
800        )
801
802    @patch(
803        "authentik.flows.views.executor.to_stage_response",
804        TO_STAGE_RESPONSE_MOCK,
805    )
806    def test_no_token_require_token(self):
807        """Test that accessing a REQUIRE_TOKEN flow without any token shows error"""
808        flow = create_test_flow(
809            FlowDesignation.RECOVERY,
810            authentication=FlowAuthenticationRequirement.REQUIRE_TOKEN,
811        )
812
813        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
814        response = self.client.get(url)
815        self.assertStageResponse(
816            response,
817            flow,
818            component="ak-stage-access-denied",
819            error_message="This link is invalid or has expired. Please request a new one.",
820        )