authentik.flows.tests.test_executor

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