flows: allow splicing a planned flow into an active plan

authentik/flows/planner.py

@@ -88,6 +88,11 @@ def insert_stage(self, stage: Stage, marker: StageMarker | None = None, index=1)
         self.bindings.insert(index, FlowStageBinding(stage=stage, order=0))
         self.markers.insert(index, marker or StageMarker())
 
+    def insert_plan(self, plan: FlowPlan, index=1):
+        """Insert all bindings and markers of `plan` into this plan, as immediate next stages"""
+        self.bindings[index:index] = plan.bindings
+        self.markers[index:index] = plan.markers
+
     def redirect(self, destination: str):
         """Insert a redirect stage as next stage"""
         from authentik.flows.stage import RedirectStage
@@ -198,6 +203,7 @@ class FlowPlanner:
 
     use_cache: bool
     allow_empty_flows: bool
+    check_authentication: bool
 
     flow: Flow
 
@@ -206,6 +212,7 @@ class FlowPlanner:
     def __init__(self, flow: Flow):
         self.use_cache = True
         self.allow_empty_flows = False
+        self.check_authentication = True
         self.flow = flow
         self._logger = get_logger().bind(flow_slug=flow.slug)
 
@@ -277,7 +284,10 @@ def plan(self, request: HttpRequest, default_context: dict[str, Any] | None = No
             else:
                 user = request.user
 
-            context.update(self._check_authentication(request, context))
+            # Skipping the authentication check also skips outpost detection,
+            # callers disabling it must have established the user's identity already
+            if self.check_authentication:
+                context.update(self._check_authentication(request, context))
             # First off, check the flow's direct policy bindings
             # to make sure the user even has access to the flow
             engine = PolicyEngine(self.flow, user, request)

authentik/flows/tests/test_planner.py

@@ -28,6 +28,7 @@
     PLAN_CONTEXT_IS_REDIRECTED,
     PLAN_CONTEXT_IS_RESTORED,
     PLAN_CONTEXT_PENDING_USER,
+    FlowPlan,
     FlowPlanner,
     cache_key,
 )
@@ -317,6 +318,41 @@ def dispatch(self, request: HttpRequest, *args, **kwargs):
         plan.append_stage(in_memory_stage(TStageView))
         self.assertTrue(plan.requires_flow_executor(allowed_silent_types=[TStageView]))
 
+    def test_insert_plan(self):
+        """Test that insert_plan splices bindings and markers as immediate next stages"""
+        plan = FlowPlan(flow_pk=generate_id())
+        first = FlowStageBinding(stage=DummyStage(name="first"))
+        last = FlowStageBinding(stage=DummyStage(name="last"))
+        plan.append(first)
+        plan.append(last)
+
+        other = FlowPlan(flow_pk=generate_id())
+        inserted = FlowStageBinding(stage=DummyStage(name="inserted"))
+        marker = ReevaluateMarker(None)
+        other.append(inserted, marker)
+
+        plan.insert_plan(other)
+
+        self.assertEqual(plan.bindings, [first, inserted, last])
+        self.assertEqual(plan.markers[1], marker)
+        self.assertEqual(len(plan.bindings), len(plan.markers))
+
+    def test_check_authentication_disabled(self):
+        """Test that disabling check_authentication skips the authentication requirement"""
+        flow = create_test_flow()
+        flow.authentication = FlowAuthenticationRequirement.REQUIRE_AUTHENTICATED
+        request = self.request_factory.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
+        )
+
+        with self.assertRaises(FlowNonApplicableException):
+            FlowPlanner(flow).plan(request)
+
+        planner = FlowPlanner(flow)
+        planner.allow_empty_flows = True
+        planner.check_authentication = False
+        planner.plan(request)
+
     def test_to_redirect_skip_policies(self):
         """Test to_redirect and skipping the flow executor
         (with a marker on the stage view type that can be skipped)