habsi/authentik
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

flows: more tests (#11587)

Browse Source 
Signed-off-by: Jens Langhammer <jens@goauthentik.io>
This commit is contained in:
Jens L.
2024-12-20 21:18:09 +01:00
committed by GitHub
parent 15be3f2461
commit ee648269f7
3 changed files with 144 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
authentik/flows/tests/test_executor.py
View File

@ -7,8 +7,8 @@ from django.http import HttpRequest, HttpResponse
from django.test.client import RequestFactory
from django.urls import reverse
from authentik.core.models import User
from authentik.core.tests.utils import create_test_flow
from authentik.core.models import Group, User
from authentik.core.tests.utils import create_test_flow, create_test_user
from authentik.flows.markers import ReevaluateMarker, StageMarker
from authentik.flows.models import (
FlowDeniedAction,
@ -255,7 +255,11 @@ class TestFlowExecutor(FlowTestCase):
)
binding = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name=generate_id()), order=0
target=flow,
stage=DummyStage.objects.create(name=generate_id()),
order=0,
evaluate_on_plan=True,
re_evaluate_policies=False,
)
binding2 = FlowStageBinding.objects.create(
target=flow,
@ -278,8 +282,8 @@ class TestFlowExecutor(FlowTestCase):
self.assertEqual(plan.bindings[0], binding)
self.assertEqual(plan.bindings[1], binding2)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], ReevaluateMarker)
self.assertEqual(plan.markers[0].__class__, StageMarker)
self.assertEqual(plan.markers[1].__class__, ReevaluateMarker)
# Second request, this passes the first dummy stage
response = self.client.post(exec_url)
@ -301,7 +305,11 @@ class TestFlowExecutor(FlowTestCase):
)
binding = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name=generate_id()), order=0
target=flow,
stage=DummyStage.objects.create(name=generate_id()),
order=0,
evaluate_on_plan=True,
re_evaluate_policies=False,
)
binding2 = FlowStageBinding.objects.create(
target=flow,
@ -310,7 +318,11 @@ class TestFlowExecutor(FlowTestCase):
re_evaluate_policies=True,
)
binding3 = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name=generate_id()), order=2
target=flow,
stage=DummyStage.objects.create(name=generate_id()),
order=2,
evaluate_on_plan=True,
re_evaluate_policies=False,
)
PolicyBinding.objects.create(policy=false_policy, target=binding2, order=0)
@ -328,9 +340,9 @@ class TestFlowExecutor(FlowTestCase):
self.assertEqual(plan.bindings[1], binding2)
self.assertEqual(plan.bindings[2], binding3)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], ReevaluateMarker)
self.assertIsInstance(plan.markers[2], StageMarker)
self.assertEqual(plan.markers[0].__class__, StageMarker)
self.assertEqual(plan.markers[1].__class__, ReevaluateMarker)
self.assertEqual(plan.markers[2].__class__, StageMarker)
# Second request, this passes the first dummy stage
response = self.client.post(exec_url)
@ -341,8 +353,8 @@ class TestFlowExecutor(FlowTestCase):
self.assertEqual(plan.bindings[0], binding2)
self.assertEqual(plan.bindings[1], binding3)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], StageMarker)
self.assertEqual(plan.markers[0].__class__, ReevaluateMarker)
self.assertEqual(plan.markers[1].__class__, StageMarker)
# third request, this should trigger the re-evaluate
# We do this request without the patch, so the policy results in false
@ -360,7 +372,11 @@ class TestFlowExecutor(FlowTestCase):
)
binding = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name=generate_id()), order=0
target=flow,
stage=DummyStage.objects.create(name=generate_id()),
order=0,
evaluate_on_plan=True,
re_evaluate_policies=False,
)
binding2 = FlowStageBinding.objects.create(
target=flow,
@ -369,7 +385,11 @@ class TestFlowExecutor(FlowTestCase):
re_evaluate_policies=True,
)
binding3 = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name=generate_id()), order=2
target=flow,
stage=DummyStage.objects.create(name=generate_id()),
order=2,
evaluate_on_plan=True,
re_evaluate_policies=False,
)
PolicyBinding.objects.create(policy=true_policy, target=binding2, order=0)
@ -387,9 +407,9 @@ class TestFlowExecutor(FlowTestCase):
self.assertEqual(plan.bindings[1], binding2)
self.assertEqual(plan.bindings[2], binding3)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], ReevaluateMarker)
self.assertIsInstance(plan.markers[2], StageMarker)
self.assertEqual(plan.markers[0].__class__, StageMarker)
self.assertEqual(plan.markers[1].__class__, ReevaluateMarker)
self.assertEqual(plan.markers[2].__class__, StageMarker)
# Second request, this passes the first dummy stage
response = self.client.post(exec_url)
@ -400,8 +420,8 @@ class TestFlowExecutor(FlowTestCase):
self.assertEqual(plan.bindings[0], binding2)
self.assertEqual(plan.bindings[1], binding3)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], StageMarker)
self.assertEqual(plan.markers[0].__class__, ReevaluateMarker)
self.assertEqual(plan.markers[1].__class__, StageMarker)
# Third request, this passes the first dummy stage
response = self.client.post(exec_url)
@ -411,7 +431,7 @@ class TestFlowExecutor(FlowTestCase):
self.assertEqual(plan.bindings[0], binding3)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertEqual(plan.markers[0].__class__, StageMarker)
# third request, this should trigger the re-evaluate
# We do this request without the patch, so the policy results in false
@ -429,7 +449,11 @@ class TestFlowExecutor(FlowTestCase):
)
binding = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name=generate_id()), order=0
target=flow,
stage=DummyStage.objects.create(name=generate_id()),
order=0,
evaluate_on_plan=True,
re_evaluate_policies=False,
)
binding2 = FlowStageBinding.objects.create(
target=flow,
@ -444,7 +468,11 @@ class TestFlowExecutor(FlowTestCase):
re_evaluate_policies=True,
)
binding4 = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name=generate_id()), order=2
target=flow,
stage=DummyStage.objects.create(name=generate_id()),
order=2,
evaluate_on_plan=True,
re_evaluate_policies=False,
)
PolicyBinding.objects.create(policy=false_policy, target=binding2, order=0)
@ -465,10 +493,10 @@ class TestFlowExecutor(FlowTestCase):
self.assertEqual(plan.bindings[2], binding3)
self.assertEqual(plan.bindings[3], binding4)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], ReevaluateMarker)
self.assertIsInstance(plan.markers[2], ReevaluateMarker)
self.assertIsInstance(plan.markers[3], StageMarker)
self.assertEqual(plan.markers[0].__class__, StageMarker)
self.assertEqual(plan.markers[1].__class__, ReevaluateMarker)
self.assertEqual(plan.markers[2].__class__, ReevaluateMarker)
self.assertEqual(plan.markers[3].__class__, StageMarker)
# Second request, this passes the first dummy stage
response = self.client.post(exec_url)
@ -519,9 +547,9 @@ class TestFlowExecutor(FlowTestCase):
)
# Stage 0 is a deny stage that is added dynamically
# when the reputation policy says so
deny_stage = DenyStage.objects.create(name="deny")
deny_stage = DenyStage.objects.create(name=generate_id())
reputation_policy = ReputationPolicy.objects.create(
name="reputation", threshold=-1, check_ip=False
name=generate_id(), threshold=-1, check_ip=False
)
deny_binding = FlowStageBinding.objects.create(
target=flow,
@ -534,7 +562,7 @@ class TestFlowExecutor(FlowTestCase):
# Stage 1 is an identification stage
ident_stage = IdentificationStage.objects.create(
name="ident",
name=generate_id(),
user_fields=[UserFields.E_MAIL],
pretend_user_exists=False,
)
@ -559,3 +587,64 @@ class TestFlowExecutor(FlowTestCase):
)
response = self.client.post(exec_url, {"uid_field": "invalid-string"}, follow=True)
self.assertStageResponse(response, flow, component="ak-stage-access-denied")
def test_re_evaluate_group_binding(self):
"""Test re-evaluate stage binding that has a policy binding to a group"""
flow = create_test_flow()
user_group_membership = create_test_user()
user_direct_binding = create_test_user()
user_other = create_test_user()
group_a = Group.objects.create(name=generate_id())
user_group_membership.ak_groups.add(group_a)
# Stage 0 is an identification stage
ident_stage = IdentificationStage.objects.create(
name=generate_id(),
user_fields=[UserFields.USERNAME],
pretend_user_exists=False,
)
FlowStageBinding.objects.create(
target=flow,
stage=ident_stage,
order=0,
)
# Stage 1 is a dummy stage that is only shown for users in group_a
dummy_stage = DummyStage.objects.create(name=generate_id())
dummy_binding = FlowStageBinding.objects.create(target=flow, stage=dummy_stage, order=1)
PolicyBinding.objects.create(group=group_a, target=dummy_binding, order=0)
PolicyBinding.objects.create(user=user_direct_binding, target=dummy_binding, order=0)
# Stage 2 is a deny stage that (in this case) only user_b will see
deny_stage = DenyStage.objects.create(name=generate_id())
FlowStageBinding.objects.create(target=flow, stage=deny_stage, order=2)
exec_url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
with self.subTest(f"Test user access through group: {user_group_membership}"):
self.client.logout()
# First request, run the planner
response = self.client.get(exec_url)
self.assertStageResponse(response, flow, component="ak-stage-identification")
response = self.client.post(
exec_url, {"uid_field": user_group_membership.username}, follow=True
)
self.assertStageResponse(response, flow, component="ak-stage-dummy")
with self.subTest(f"Test user access through user: {user_direct_binding}"):
self.client.logout()
# First request, run the planner
response = self.client.get(exec_url)
self.assertStageResponse(response, flow, component="ak-stage-identification")
response = self.client.post(
exec_url, {"uid_field": user_direct_binding.username}, follow=True
)
self.assertStageResponse(response, flow, component="ak-stage-dummy")
with self.subTest(f"Test user has no access: {user_other}"):
self.client.logout()
# First request, run the planner
response = self.client.get(exec_url)
self.assertStageResponse(response, flow, component="ak-stage-identification")
response = self.client.post(exec_url, {"uid_field": user_other.username}, follow=True)
self.assertStageResponse(response, flow, component="ak-stage-access-denied")

22
authentik/flows/tests/test_inspector.py
View File

@ -8,6 +8,7 @@ from rest_framework.test import APITestCase
from authentik.core.tests.utils import create_test_admin_user, create_test_flow
from authentik.flows.models import FlowDesignation, FlowStageBinding, InvalidResponseAction
from authentik.lib.generators import generate_id
from authentik.stages.dummy.models import DummyStage
from authentik.stages.identification.models import IdentificationStage, UserFields
@ -26,7 +27,7 @@ class TestFlowInspector(APITestCase):
# Stage 1 is an identification stage
ident_stage = IdentificationStage.objects.create(
name="ident",
name=generate_id(),
user_fields=[UserFields.USERNAME],
)
FlowStageBinding.objects.create(
@ -35,9 +36,8 @@ class TestFlowInspector(APITestCase):
order=1,
invalid_response_action=InvalidResponseAction.RESTART_WITH_CONTEXT,
)
FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy2"), order=1
)
dummy_stage = DummyStage.objects.create(name=generate_id())
FlowStageBinding.objects.create(target=flow, stage=dummy_stage, order=1)
res = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
@ -68,9 +68,11 @@ class TestFlowInspector(APITestCase):
)
content = loads(ins.content)
self.assertEqual(content["is_completed"], False)
self.assertEqual(content["current_plan"]["current_stage"]["stage_obj"]["name"], "ident")
self.assertEqual(
content["current_plan"]["next_planned_stage"]["stage_obj"]["name"], "dummy2"
content["current_plan"]["current_stage"]["stage_obj"]["name"], ident_stage.name
)
self.assertEqual(
content["current_plan"]["next_planned_stage"]["stage_obj"]["name"], dummy_stage.name
)
self.client.post(
@ -84,8 +86,12 @@ class TestFlowInspector(APITestCase):
)
content = loads(ins.content)
self.assertEqual(content["is_completed"], False)
self.assertEqual(content["plans"][0]["current_stage"]["stage_obj"]["name"], "ident")
self.assertEqual(content["current_plan"]["current_stage"]["stage_obj"]["name"], "dummy2")
self.assertEqual(
content["plans"][0]["current_stage"]["stage_obj"]["name"], ident_stage.name
)
self.assertEqual(
content["current_plan"]["current_stage"]["stage_obj"]["name"], dummy_stage.name
)
self.assertEqual(
content["current_plan"]["plan_context"]["pending_user"]["username"], self.admin.username
)

18
authentik/flows/tests/test_planner.py
View File

@ -29,6 +29,7 @@ from authentik.flows.planner import (
cache_key,
)
from authentik.flows.stage import StageView
from authentik.lib.generators import generate_id
from authentik.lib.tests.utils import dummy_get_response
from authentik.outposts.apps import MANAGED_OUTPOST
from authentik.outposts.models import Outpost
@ -153,7 +154,7 @@ class TestFlowPlanner(TestCase):
"""Test planner cache"""
flow = create_test_flow(FlowDesignation.AUTHENTICATION)
FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy"), order=0
target=flow, stage=DummyStage.objects.create(name=generate_id()), order=0
)
request = self.request_factory.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
@ -172,7 +173,7 @@ class TestFlowPlanner(TestCase):
"""Test planner with default_context"""
flow = create_test_flow()
FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy"), order=0
target=flow, stage=DummyStage.objects.create(name=generate_id()), order=0
)
user = User.objects.create(username="test-user")
@ -191,7 +192,7 @@ class TestFlowPlanner(TestCase):
FlowStageBinding.objects.create(
target=flow,
stage=DummyStage.objects.create(name="dummy1"),
stage=DummyStage.objects.create(name=generate_id()),
order=0,
re_evaluate_policies=True,
)
@ -204,7 +205,7 @@ class TestFlowPlanner(TestCase):
planner = FlowPlanner(flow)
plan = planner.plan(request)
self.assertIsInstance(plan.markers[0], ReevaluateMarker)
self.assertEqual(plan.markers[0].__class__, ReevaluateMarker)
def test_planner_reevaluate_actual(self):
"""Test planner with re-evaluate"""
@ -212,11 +213,14 @@ class TestFlowPlanner(TestCase):
false_policy = DummyPolicy.objects.create(result=False, wait_min=1, wait_max=2)
binding = FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy1"), order=0
target=flow,
stage=DummyStage.objects.create(name=generate_id()),
order=0,
re_evaluate_policies=False,
)
binding2 = FlowStageBinding.objects.create(
target=flow,
stage=DummyStage.objects.create(name="dummy2"),
stage=DummyStage.objects.create(name=generate_id()),
order=1,
re_evaluate_policies=True,
)
@ -240,6 +244,8 @@ class TestFlowPlanner(TestCase):
self.assertEqual(plan.bindings[0], binding)
self.assertEqual(plan.bindings[1], binding2)
self.assertEqual(plan.markers[0].__class__, StageMarker)
self.assertEqual(plan.markers[1].__class__, ReevaluateMarker)
self.assertIsInstance(plan.markers[0], StageMarker)
self.assertIsInstance(plan.markers[1], ReevaluateMarker)