authentik.flows.planner

  1"""Flows Planner"""
  2
  3from dataclasses import dataclass, field
  4from typing import TYPE_CHECKING, Any
  5
  6from django.core.cache import cache
  7from django.http import HttpRequest, HttpResponse
  8from django.utils.translation import gettext as _
  9from sentry_sdk import start_span
 10from sentry_sdk.tracing import Span
 11from structlog.stdlib import BoundLogger, get_logger
 12
 13from authentik.core.models import User
 14from authentik.events.models import cleanse_dict
 15from authentik.flows.apps import HIST_FLOWS_PLAN_TIME
 16from authentik.flows.exceptions import EmptyFlowException, FlowNonApplicableException
 17from authentik.flows.markers import ReevaluateMarker, StageMarker
 18from authentik.flows.models import (
 19    Flow,
 20    FlowAuthenticationRequirement,
 21    FlowDesignation,
 22    FlowStageBinding,
 23    Stage,
 24    in_memory_stage,
 25)
 26from authentik.lib.config import CONFIG
 27from authentik.lib.utils.urls import redirect_with_qs
 28from authentik.outposts.models import Outpost
 29from authentik.policies.engine import PolicyEngine
 30from authentik.policies.types import PolicyResult
 31from authentik.root.middleware import ClientIPMiddleware
 32
 33if TYPE_CHECKING:
 34    from authentik.flows.stage import StageView
 35
 36
 37LOGGER = get_logger()
 38PLAN_CONTEXT_PENDING_USER = "pending_user"
 39PLAN_CONTEXT_SSO = "is_sso"
 40PLAN_CONTEXT_REDIRECT = "redirect"
 41PLAN_CONTEXT_APPLICATION = "application"
 42PLAN_CONTEXT_DEVICE = "device"
 43PLAN_CONTEXT_SOURCE = "source"
 44PLAN_CONTEXT_OUTPOST = "outpost"
 45PLAN_CONTEXT_POST = "goauthentik.io/http/post"
 46# Is set by the Flow Planner when a FlowToken was used, and the currently active flow plan
 47# was restored.
 48PLAN_CONTEXT_IS_RESTORED = "is_restored"
 49PLAN_CONTEXT_IS_REDIRECTED = "is_redirected"
 50PLAN_CONTEXT_REDIRECT_STAGE_TARGET = "redirect_stage_target"
 51CACHE_TIMEOUT = CONFIG.get_int("cache.timeout_flows")
 52CACHE_PREFIX = "goauthentik.io/flows/planner/"
 53
 54
 55def cache_key(flow: Flow, user: User | None = None) -> str:
 56    """Generate Cache key for flow"""
 57    prefix = CACHE_PREFIX + str(flow.pk)
 58    if user:
 59        prefix += f"#{user.pk}"
 60    return prefix
 61
 62
 63@dataclass(slots=True)
 64class FlowPlan:
 65    """This data-class is the output of a FlowPlanner. It holds a flat list
 66    of all Stages that should be run."""
 67
 68    flow_pk: str
 69
 70    bindings: list[FlowStageBinding] = field(default_factory=list)
 71    context: dict[str, Any] = field(default_factory=dict)
 72    markers: list[StageMarker] = field(default_factory=list)
 73
 74    def append_stage(self, stage: Stage, marker: StageMarker | None = None):
 75        """Append `stage` to the end of the plan, optionally with stage marker"""
 76        return self.append(FlowStageBinding(stage=stage), marker)
 77
 78    def append(self, binding: FlowStageBinding, marker: StageMarker | None = None):
 79        """Append `stage` to the end of the plan, optionally with stage marker"""
 80        self.bindings.append(binding)
 81        self.markers.append(marker or StageMarker())
 82
 83    def insert_stage(self, stage: Stage, marker: StageMarker | None = None, index=1):
 84        """Insert stage into plan, as immediate next stage"""
 85        self.bindings.insert(index, FlowStageBinding(stage=stage, order=0))
 86        self.markers.insert(index, marker or StageMarker())
 87
 88    def redirect(self, destination: str):
 89        """Insert a redirect stage as next stage"""
 90        from authentik.flows.stage import RedirectStage
 91
 92        self.insert_stage(in_memory_stage(RedirectStage, destination=destination))
 93
 94    def next(self, http_request: HttpRequest | None) -> FlowStageBinding | None:
 95        """Return next pending stage from the bottom of the list"""
 96        if not self.has_stages:
 97            return None
 98        binding = self.bindings[0]
 99        marker = self.markers[0]
100
101        if marker.__class__ is not StageMarker:
102            LOGGER.debug("f(plan_inst): stage has marker", binding=binding, marker=marker)
103        marked_stage = marker.process(self, binding, http_request)
104        if not marked_stage:
105            LOGGER.debug("f(plan_inst): marker returned none, next stage", binding=binding)
106            self.bindings.remove(binding)
107            self.markers.remove(marker)
108            if not self.has_stages:
109                return None
110
111            return self.next(http_request)
112        return marked_stage
113
114    def pop(self):
115        """Pop next pending stage from bottom of list"""
116        if not self.markers and not self.bindings:
117            return
118        self.markers.pop(0)
119        self.bindings.pop(0)
120
121    @property
122    def has_stages(self) -> bool:
123        """Check if there are any stages left in this plan"""
124        return len(self.markers) + len(self.bindings) > 0
125
126    def requires_flow_executor(
127        self,
128        allowed_silent_types: list[StageView] | None = None,
129    ):
130        # Check if we actually need to show the Flow executor, or if we can jump straight to the end
131        found_unskippable = True
132        if allowed_silent_types:
133            LOGGER.debug("Checking if we can skip the flow executor...")
134            # Policies applied to the flow have already been evaluated, so we're checking for stages
135            # allow-listed or bindings that require a policy re-eval
136            found_unskippable = False
137            for binding, marker in zip(self.bindings, self.markers, strict=True):
138                if binding.stage.view not in allowed_silent_types:
139                    found_unskippable = True
140                if marker and isinstance(marker, ReevaluateMarker):
141                    found_unskippable = True
142        LOGGER.debug("Required flow executor status", status=found_unskippable)
143        return found_unskippable
144
145    def to_redirect(
146        self,
147        request: HttpRequest,
148        flow: Flow,
149        next: str | None = None,
150        allowed_silent_types: list[StageView] | None = None,
151    ) -> HttpResponse:
152        """Redirect to the flow executor for this flow plan"""
153        from authentik.flows.views.executor import (
154            NEXT_ARG_NAME,
155            SESSION_KEY_PLAN,
156            FlowExecutorView,
157        )
158
159        request.session[SESSION_KEY_PLAN] = self
160        requires_flow_executor = self.requires_flow_executor(allowed_silent_types)
161
162        if not requires_flow_executor:
163            # No unskippable stages found, so we can directly return the response of the last stage
164            final_stage: type[StageView] = self.bindings[-1].stage.view
165            temp_exec = FlowExecutorView(flow=flow, request=request, plan=self)
166            temp_exec.current_stage = self.bindings[-1].stage
167            temp_exec.current_stage_view = final_stage
168            temp_exec.setup(request, flow.slug)
169            stage = final_stage(request=request, executor=temp_exec)
170            response = stage.dispatch(request)
171            # Ensure we clean the flow state we have in the session before we redirect away
172            temp_exec.stage_ok()
173            return response
174
175        get_qs = request.GET.copy()
176        if request.user.is_authenticated and (
177            # Object-scoped permission or global permission
178            request.user.has_perm("authentik_flows.inspect_flow", flow)
179            or request.user.has_perm("authentik_flows.inspect_flow")
180        ):
181            get_qs["inspector"] = "available"
182        if next:
183            get_qs[NEXT_ARG_NAME] = next
184
185        return redirect_with_qs(
186            "authentik_core:if-flow",
187            get_qs,
188            flow_slug=flow.slug,
189        )
190
191
192class FlowPlanner:
193    """Execute all policies to plan out a flat list of all Stages
194    that should be applied."""
195
196    use_cache: bool
197    allow_empty_flows: bool
198
199    flow: Flow
200
201    _logger: BoundLogger
202
203    def __init__(self, flow: Flow):
204        self.use_cache = True
205        self.allow_empty_flows = False
206        self.flow = flow
207        self._logger = get_logger().bind(flow_slug=flow.slug)
208
209    def _check_authentication(self, request: HttpRequest, context: dict[str, Any]):
210        """Check the flow's authentication level is matched by `request`"""
211        if (
212            self.flow.authentication == FlowAuthenticationRequirement.REQUIRE_AUTHENTICATED
213            and not request.user.is_authenticated
214        ):
215            raise FlowNonApplicableException()
216        if (
217            self.flow.authentication == FlowAuthenticationRequirement.REQUIRE_UNAUTHENTICATED
218            and request.user.is_authenticated
219        ):
220            raise FlowNonApplicableException()
221        if (
222            self.flow.authentication == FlowAuthenticationRequirement.REQUIRE_SUPERUSER
223            and not request.user.is_superuser
224        ):
225            raise FlowNonApplicableException()
226        if (
227            self.flow.authentication == FlowAuthenticationRequirement.REQUIRE_REDIRECT
228            and context.get(PLAN_CONTEXT_IS_REDIRECTED) is None
229        ):
230            raise FlowNonApplicableException()
231        if (
232            self.flow.authentication == FlowAuthenticationRequirement.REQUIRE_TOKEN
233            and context.get(PLAN_CONTEXT_IS_RESTORED) is None
234        ):
235            raise FlowNonApplicableException(
236                PolicyResult(
237                    False, _("This link is invalid or has expired. Please request a new one.")
238                )
239            )
240        outpost_user = ClientIPMiddleware.get_outpost_user(request)
241        if self.flow.authentication == FlowAuthenticationRequirement.REQUIRE_OUTPOST:
242            if not outpost_user:
243                raise FlowNonApplicableException()
244        if outpost_user:
245            outpost = Outpost.objects.filter(
246                # TODO: Since Outpost and user are not directly connected, we have to look up a user
247                # like this. This should ideally by in authentik/outposts/models.py
248                pk=outpost_user.username.replace("ak-outpost-", "")
249            ).first()
250            if outpost:
251                return {
252                    PLAN_CONTEXT_OUTPOST: {
253                        "instance": outpost,
254                    }
255                }
256        return {}
257
258    def plan(self, request: HttpRequest, default_context: dict[str, Any] | None = None) -> FlowPlan:
259        """Check each of the flows' policies, check policies for each stage with PolicyBinding
260        and return ordered list"""
261        with start_span(op="authentik.flow.planner.plan", name=self.flow.slug) as span:
262            span: Span
263            span.set_data("flow", self.flow)
264            span.set_data("request", request)
265
266            self._logger.debug(
267                "f(plan): starting planning process",
268            )
269            context = default_context or {}
270            # Bit of a workaround here, if there is a pending user set in the default context
271            # we use that user for our cache key to make sure they don't get the generic response
272            if context and PLAN_CONTEXT_PENDING_USER in context:
273                user = context[PLAN_CONTEXT_PENDING_USER]
274            else:
275                user = request.user
276
277            context.update(self._check_authentication(request, context))
278            # First off, check the flow's direct policy bindings
279            # to make sure the user even has access to the flow
280            engine = PolicyEngine(self.flow, user, request)
281            engine.use_cache = self.use_cache
282            span.set_data("context", cleanse_dict(context))
283            engine.request.context.update(context)
284            engine.build()
285            result = engine.result
286            if not result.passing:
287                raise FlowNonApplicableException(result)
288            # User is passing so far, check if we have a cached plan
289            cached_plan_key = cache_key(self.flow, user)
290            cached_plan = cache.get(cached_plan_key, None)
291            if self.flow.designation not in [FlowDesignation.STAGE_CONFIGURATION]:
292                if cached_plan and self.use_cache:
293                    self._logger.debug(
294                        "f(plan): taking plan from cache",
295                        key=cached_plan_key,
296                    )
297                    # Reset the context as this isn't factored into caching
298                    cached_plan.context = context
299                    return cached_plan
300            self._logger.debug(
301                "f(plan): building plan",
302            )
303            plan = self._build_plan(user, request, context)
304            if self.use_cache:
305                cache.set(cache_key(self.flow, user), plan, CACHE_TIMEOUT)
306            if not plan.bindings and not self.allow_empty_flows:
307                raise EmptyFlowException()
308            return plan
309
310    def _build_plan(
311        self,
312        user: User,
313        request: HttpRequest,
314        default_context: dict[str, Any] | None,
315    ) -> FlowPlan:
316        """Build flow plan by checking each stage in their respective
317        order and checking the applied policies"""
318        with (
319            start_span(
320                op="authentik.flow.planner.build_plan",
321                name=self.flow.slug,
322            ) as span,
323            HIST_FLOWS_PLAN_TIME.labels(flow_slug=self.flow.slug).time(),
324        ):
325            span: Span
326            span.set_data("flow", self.flow)
327            span.set_data("user", user)
328            span.set_data("request", request)
329
330            plan = FlowPlan(flow_pk=self.flow.pk.hex)
331            if default_context:
332                plan.context = default_context
333            # Check Flow policies
334            bindings = list(
335                FlowStageBinding.objects.filter(target__pk=self.flow.pk).order_by("order")
336            )
337            stages = Stage.objects.filter(flowstagebinding__in=[binding.pk for binding in bindings])
338            for binding in bindings:
339                binding: FlowStageBinding
340                stage = [stage for stage in stages if stage.pk == binding.stage_id][0]
341                marker = StageMarker()
342                if binding.evaluate_on_plan:
343                    self._logger.debug(
344                        "f(plan): evaluating on plan",
345                        stage=stage,
346                    )
347                    engine = PolicyEngine(binding, user, request)
348                    engine.use_cache = self.use_cache
349                    engine.request.context["flow_plan"] = plan
350                    engine.request.context.update(plan.context)
351                    engine.build()
352                    if engine.passing:
353                        self._logger.debug(
354                            "f(plan): stage passing",
355                            stage=stage,
356                        )
357                    else:
358                        stage = None
359                else:
360                    self._logger.debug(
361                        "f(plan): not evaluating on plan",
362                        stage=stage,
363                    )
364                if binding.re_evaluate_policies and stage:
365                    self._logger.debug(
366                        "f(plan): stage has re-evaluate marker",
367                        stage=stage,
368                    )
369                    marker = ReevaluateMarker(binding=binding)
370                if stage:
371                    plan.append(binding, marker)
372        self._logger.debug(
373            "f(plan): finished building",
374        )
375        return plan