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