authentik.enterprise.stages.source.tests
Source stage tests
1"""Source stage tests""" 2 3from django.urls import reverse 4 5from authentik.core.tests.utils import create_test_flow, create_test_user 6from authentik.enterprise.stages.source.models import SourceStage 7from authentik.enterprise.stages.source.stage import SourceStageFinal 8from authentik.flows.models import FlowDesignation, FlowStageBinding, FlowToken, in_memory_stage 9from authentik.flows.planner import PLAN_CONTEXT_IS_RESTORED, FlowPlan 10from authentik.flows.tests import FlowTestCase 11from authentik.flows.views.executor import SESSION_KEY_PLAN 12from authentik.lib.generators import generate_id 13from authentik.sources.saml.models import SAMLSource 14from authentik.stages.identification.models import IdentificationStage, UserFields 15from authentik.stages.password import BACKEND_INBUILT 16from authentik.stages.password.models import PasswordStage 17from authentik.stages.user_login.models import UserLoginStage 18 19 20class TestSourceStage(FlowTestCase): 21 """Source stage tests""" 22 23 def setUp(self): 24 self.source = SAMLSource.objects.create( 25 slug=generate_id(), 26 issuer="authentik", 27 allow_idp_initiated=True, 28 pre_authentication_flow=create_test_flow(), 29 ) 30 31 def test_source_success(self): 32 """Test""" 33 user = create_test_user() 34 flow = create_test_flow(FlowDesignation.AUTHENTICATION) 35 stage = SourceStage.objects.create(name=generate_id(), source=self.source) 36 FlowStageBinding.objects.create( 37 target=flow, 38 stage=IdentificationStage.objects.create( 39 name=generate_id(), 40 user_fields=[UserFields.USERNAME], 41 ), 42 order=0, 43 ) 44 FlowStageBinding.objects.create( 45 target=flow, 46 stage=PasswordStage.objects.create(name=generate_id(), backends=[BACKEND_INBUILT]), 47 order=5, 48 ) 49 FlowStageBinding.objects.create(target=flow, stage=stage, order=10) 50 FlowStageBinding.objects.create( 51 target=flow, 52 stage=UserLoginStage.objects.create( 53 name=generate_id(), 54 ), 55 order=15, 56 ) 57 58 # Get user identification stage 59 response = self.client.get( 60 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), 61 ) 62 self.assertEqual(response.status_code, 200) 63 self.assertStageResponse(response, flow, component="ak-stage-identification") 64 # Send username 65 response = self.client.post( 66 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), 67 data={"uid_field": user.username}, 68 follow=True, 69 ) 70 self.assertEqual(response.status_code, 200) 71 self.assertStageResponse(response, flow, component="ak-stage-password") 72 # Send password 73 response = self.client.post( 74 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), 75 data={"password": user.username}, 76 follow=True, 77 ) 78 self.assertEqual(response.status_code, 200) 79 self.assertStageRedirects( 80 response, 81 reverse("authentik_sources_saml:login", kwargs={"source_slug": self.source.slug}), 82 ) 83 84 # Hijack flow plan so we don't have to emulate the source 85 flow_token = FlowToken.objects.filter( 86 identifier__startswith=f"ak-source-stage-{stage.name.lower()}" 87 ).first() 88 self.assertIsNotNone(flow_token) 89 session = self.client.session 90 plan: FlowPlan = session[SESSION_KEY_PLAN] 91 plan.insert_stage(in_memory_stage(SourceStageFinal), index=0) 92 plan.context[PLAN_CONTEXT_IS_RESTORED] = flow_token 93 plan.context["foo"] = "bar" 94 session[SESSION_KEY_PLAN] = plan 95 session.save() 96 97 # Pretend we've just returned from the source 98 with self.assertFlowFinishes() as ff: 99 response = self.client.get( 100 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), follow=True 101 ) 102 self.assertEqual(response.status_code, 200) 103 self.assertStageRedirects( 104 response, reverse("authentik_core:if-flow", kwargs={"flow_slug": flow.slug}) 105 ) 106 self.assertEqual(ff().context["foo"], "bar")
21class TestSourceStage(FlowTestCase): 22 """Source stage tests""" 23 24 def setUp(self): 25 self.source = SAMLSource.objects.create( 26 slug=generate_id(), 27 issuer="authentik", 28 allow_idp_initiated=True, 29 pre_authentication_flow=create_test_flow(), 30 ) 31 32 def test_source_success(self): 33 """Test""" 34 user = create_test_user() 35 flow = create_test_flow(FlowDesignation.AUTHENTICATION) 36 stage = SourceStage.objects.create(name=generate_id(), source=self.source) 37 FlowStageBinding.objects.create( 38 target=flow, 39 stage=IdentificationStage.objects.create( 40 name=generate_id(), 41 user_fields=[UserFields.USERNAME], 42 ), 43 order=0, 44 ) 45 FlowStageBinding.objects.create( 46 target=flow, 47 stage=PasswordStage.objects.create(name=generate_id(), backends=[BACKEND_INBUILT]), 48 order=5, 49 ) 50 FlowStageBinding.objects.create(target=flow, stage=stage, order=10) 51 FlowStageBinding.objects.create( 52 target=flow, 53 stage=UserLoginStage.objects.create( 54 name=generate_id(), 55 ), 56 order=15, 57 ) 58 59 # Get user identification stage 60 response = self.client.get( 61 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), 62 ) 63 self.assertEqual(response.status_code, 200) 64 self.assertStageResponse(response, flow, component="ak-stage-identification") 65 # Send username 66 response = self.client.post( 67 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), 68 data={"uid_field": user.username}, 69 follow=True, 70 ) 71 self.assertEqual(response.status_code, 200) 72 self.assertStageResponse(response, flow, component="ak-stage-password") 73 # Send password 74 response = self.client.post( 75 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), 76 data={"password": user.username}, 77 follow=True, 78 ) 79 self.assertEqual(response.status_code, 200) 80 self.assertStageRedirects( 81 response, 82 reverse("authentik_sources_saml:login", kwargs={"source_slug": self.source.slug}), 83 ) 84 85 # Hijack flow plan so we don't have to emulate the source 86 flow_token = FlowToken.objects.filter( 87 identifier__startswith=f"ak-source-stage-{stage.name.lower()}" 88 ).first() 89 self.assertIsNotNone(flow_token) 90 session = self.client.session 91 plan: FlowPlan = session[SESSION_KEY_PLAN] 92 plan.insert_stage(in_memory_stage(SourceStageFinal), index=0) 93 plan.context[PLAN_CONTEXT_IS_RESTORED] = flow_token 94 plan.context["foo"] = "bar" 95 session[SESSION_KEY_PLAN] = plan 96 session.save() 97 98 # Pretend we've just returned from the source 99 with self.assertFlowFinishes() as ff: 100 response = self.client.get( 101 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), follow=True 102 ) 103 self.assertEqual(response.status_code, 200) 104 self.assertStageRedirects( 105 response, reverse("authentik_core:if-flow", kwargs={"flow_slug": flow.slug}) 106 ) 107 self.assertEqual(ff().context["foo"], "bar")
Source stage tests
def
setUp(self):
24 def setUp(self): 25 self.source = SAMLSource.objects.create( 26 slug=generate_id(), 27 issuer="authentik", 28 allow_idp_initiated=True, 29 pre_authentication_flow=create_test_flow(), 30 )
Hook method for setting up the test fixture before exercising it.
def
test_source_success(self):
32 def test_source_success(self): 33 """Test""" 34 user = create_test_user() 35 flow = create_test_flow(FlowDesignation.AUTHENTICATION) 36 stage = SourceStage.objects.create(name=generate_id(), source=self.source) 37 FlowStageBinding.objects.create( 38 target=flow, 39 stage=IdentificationStage.objects.create( 40 name=generate_id(), 41 user_fields=[UserFields.USERNAME], 42 ), 43 order=0, 44 ) 45 FlowStageBinding.objects.create( 46 target=flow, 47 stage=PasswordStage.objects.create(name=generate_id(), backends=[BACKEND_INBUILT]), 48 order=5, 49 ) 50 FlowStageBinding.objects.create(target=flow, stage=stage, order=10) 51 FlowStageBinding.objects.create( 52 target=flow, 53 stage=UserLoginStage.objects.create( 54 name=generate_id(), 55 ), 56 order=15, 57 ) 58 59 # Get user identification stage 60 response = self.client.get( 61 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), 62 ) 63 self.assertEqual(response.status_code, 200) 64 self.assertStageResponse(response, flow, component="ak-stage-identification") 65 # Send username 66 response = self.client.post( 67 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), 68 data={"uid_field": user.username}, 69 follow=True, 70 ) 71 self.assertEqual(response.status_code, 200) 72 self.assertStageResponse(response, flow, component="ak-stage-password") 73 # Send password 74 response = self.client.post( 75 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), 76 data={"password": user.username}, 77 follow=True, 78 ) 79 self.assertEqual(response.status_code, 200) 80 self.assertStageRedirects( 81 response, 82 reverse("authentik_sources_saml:login", kwargs={"source_slug": self.source.slug}), 83 ) 84 85 # Hijack flow plan so we don't have to emulate the source 86 flow_token = FlowToken.objects.filter( 87 identifier__startswith=f"ak-source-stage-{stage.name.lower()}" 88 ).first() 89 self.assertIsNotNone(flow_token) 90 session = self.client.session 91 plan: FlowPlan = session[SESSION_KEY_PLAN] 92 plan.insert_stage(in_memory_stage(SourceStageFinal), index=0) 93 plan.context[PLAN_CONTEXT_IS_RESTORED] = flow_token 94 plan.context["foo"] = "bar" 95 session[SESSION_KEY_PLAN] = plan 96 session.save() 97 98 # Pretend we've just returned from the source 99 with self.assertFlowFinishes() as ff: 100 response = self.client.get( 101 reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}), follow=True 102 ) 103 self.assertEqual(response.status_code, 200) 104 self.assertStageRedirects( 105 response, reverse("authentik_core:if-flow", kwargs={"flow_slug": flow.slug}) 106 ) 107 self.assertEqual(ff().context["foo"], "bar")
Test