Revert "Share flow executor planning"

This reverts commit 2ceb643478.

authentik/enterprise/stages/account_lockdown/api.py

@@ -27,7 +27,8 @@ from authentik.enterprise.stages.account_lockdown.stage import (
 from authentik.flows.api.stages import StageSerializer
 from authentik.flows.challenge import RedirectChallenge
 from authentik.flows.exceptions import EmptyFlowException, FlowNonApplicableException
-from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, plan_flow_for_executor
+from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlanner
+from authentik.flows.views.executor import SESSION_KEY_HISTORY, SESSION_KEY_PLAN
 
 LOGGER = get_logger()
 
@@ -80,17 +81,16 @@ class UserAccountLockdownMixin:
         flow = request._request.brand.flow_lockdown
         if flow is None:
             raise ValidationError({"non_field_errors": [_("No lockdown flow configured.")]})
+        planner = FlowPlanner(flow)
+        planner.use_cache = False
         try:
-            plan_flow_for_executor(
-                request._request,
-                flow,
-                {PLAN_CONTEXT_PENDING_USER: user},
-                use_cache=False,
-            )
-        except EmptyFlowException, FlowNonApplicableException:
+            plan = planner.plan(request._request, {PLAN_CONTEXT_PENDING_USER: user})
+        except (EmptyFlowException, FlowNonApplicableException):
             raise ValidationError(
                 {"non_field_errors": [_("Lockdown flow is not applicable.")]}
             ) from None
+        request.session[SESSION_KEY_HISTORY] = []
+        request.session[SESSION_KEY_PLAN] = plan
         return request.build_absolute_uri(
             reverse("authentik_core:if-flow", kwargs={"flow_slug": flow.slug})
         )

authentik/flows/api/flows.py

@@ -30,12 +30,8 @@ from authentik.core.api.utils import (
 from authentik.flows.api.flows_diagram import FlowDiagram, FlowDiagramSerializer
 from authentik.flows.exceptions import FlowNonApplicableException
 from authentik.flows.models import Flow
-from authentik.flows.planner import (
-    CACHE_PREFIX,
-    PLAN_CONTEXT_PENDING_USER,
-    cache_key,
-    plan_flow_for_executor,
-)
+from authentik.flows.planner import CACHE_PREFIX, PLAN_CONTEXT_PENDING_USER, FlowPlanner, cache_key
+from authentik.flows.views.executor import SESSION_KEY_HISTORY, SESSION_KEY_PLAN
 from authentik.lib.views import bad_request_message
 from authentik.rbac.decorators import permission_required
 from authentik.rbac.filters import ObjectFilter
@@ -187,14 +183,15 @@ class FlowViewSet(UsedByMixin, ModelViewSet):
     @action(detail=True, pagination_class=None, filter_backends=[ObjectFilter])
     def execute(self, request: Request, slug: str):
         """Execute flow for current user"""
+        # Because we pre-plan the flow here, and not in the planner, we need to manually clear
+        # the history of the inspector
+        request.session[SESSION_KEY_HISTORY] = []
         flow: Flow = self.get_object()
+        planner = FlowPlanner(flow)
+        planner.use_cache = False
         try:
-            plan_flow_for_executor(
-                self.request,
-                flow,
-                {PLAN_CONTEXT_PENDING_USER: request.user},
-                use_cache=False,
-            )
+            plan = planner.plan(self.request, {PLAN_CONTEXT_PENDING_USER: request.user})
+            self.request.session[SESSION_KEY_PLAN] = plan
         except FlowNonApplicableException as exc:
             return bad_request_message(
                 request,

authentik/flows/planner.py

@@ -60,26 +60,6 @@ def cache_key(flow: Flow, user: User | None = None) -> str:
     return prefix
 
 
-def plan_flow_for_executor(
-    request: HttpRequest,
-    flow: Flow,
-    default_context: dict[str, Any] | None = None,
-    *,
-    use_cache: bool = True,
-    allow_empty_flows: bool = False,
-) -> FlowPlan:
-    """Plan a flow and store it in the executor session."""
-    from authentik.flows.views.executor import SESSION_KEY_HISTORY, SESSION_KEY_PLAN
-
-    request.session[SESSION_KEY_HISTORY] = []
-    planner = FlowPlanner(flow)
-    planner.use_cache = use_cache
-    planner.allow_empty_flows = allow_empty_flows
-    plan = planner.plan(request, default_context)
-    request.session[SESSION_KEY_PLAN] = plan
-    return plan
-
-
 @dataclass(slots=True)
 class FlowPlan:
     """This data-class is the output of a FlowPlanner. It holds a flat list