authentik.flows.views.inspector

  1"""Flow Inspector"""
  2
  3from hashlib import sha256
  4from typing import Any
  5
  6from django.conf import settings
  7from django.http import Http404
  8from django.http.request import HttpRequest
  9from django.http.response import HttpResponse
 10from django.shortcuts import get_object_or_404
 11from django.utils.decorators import method_decorator
 12from django.views.decorators.clickjacking import xframe_options_sameorigin
 13from drf_spectacular.types import OpenApiTypes
 14from drf_spectacular.utils import OpenApiResponse, extend_schema
 15from rest_framework.fields import BooleanField, ListField, SerializerMethodField
 16from rest_framework.permissions import IsAuthenticated
 17from rest_framework.request import Request
 18from rest_framework.response import Response
 19from rest_framework.views import APIView
 20from structlog.stdlib import BoundLogger, get_logger
 21
 22from authentik.core.api.utils import PassiveSerializer
 23from authentik.events.utils import sanitize_dict
 24from authentik.flows.api.bindings import FlowStageBindingSerializer
 25from authentik.flows.models import Flow
 26from authentik.flows.planner import FlowPlan
 27from authentik.flows.views.executor import SESSION_KEY_HISTORY, SESSION_KEY_PLAN
 28
 29MIN_FLOW_LENGTH = 2
 30
 31
 32class FlowInspectorPlanSerializer(PassiveSerializer):
 33    """Serializer for an active FlowPlan"""
 34
 35    current_stage = SerializerMethodField()
 36    next_planned_stage = SerializerMethodField(required=False)
 37    plan_context = SerializerMethodField()
 38    session_id = SerializerMethodField()
 39
 40    def get_current_stage(self, plan: FlowPlan) -> FlowStageBindingSerializer:
 41        """Get the current stage"""
 42        return FlowStageBindingSerializer(instance=plan.bindings[0]).data
 43
 44    def get_next_planned_stage(self, plan: FlowPlan) -> FlowStageBindingSerializer:
 45        """Get the next planned stage"""
 46        if len(plan.bindings) < MIN_FLOW_LENGTH:
 47            return FlowStageBindingSerializer().data
 48        return FlowStageBindingSerializer(instance=plan.bindings[1]).data
 49
 50    def get_plan_context(self, plan: FlowPlan) -> dict[str, Any]:
 51        """Get the plan's context, sanitized"""
 52        return sanitize_dict(plan.context)
 53
 54    def get_session_id(self, _plan: FlowPlan) -> str:
 55        """Get a unique session ID"""
 56        request: Request = self.context["request"]
 57        return sha256(request._request.session.session_key.encode("ascii")).hexdigest()
 58
 59
 60class FlowInspectionSerializer(PassiveSerializer):
 61    """Serializer for inspect endpoint"""
 62
 63    plans = ListField(child=FlowInspectorPlanSerializer())
 64    current_plan = FlowInspectorPlanSerializer(required=False)
 65    is_completed = BooleanField()
 66
 67
 68@method_decorator(xframe_options_sameorigin, name="dispatch")
 69class FlowInspectorView(APIView):
 70    """Flow inspector API"""
 71
 72    flow: Flow
 73    _logger: BoundLogger
 74    permission_classes = [IsAuthenticated]
 75
 76    def setup(self, request: HttpRequest, flow_slug: str):
 77        super().setup(request, flow_slug=flow_slug)
 78        self._logger = get_logger().bind(flow_slug=flow_slug)
 79        self.flow = get_object_or_404(Flow.objects.select_related(), slug=flow_slug)
 80        if settings.DEBUG:
 81            return
 82        if request.user.has_perm(
 83            "authentik_flows.inspect_flow", self.flow
 84        ) or request.user.has_perm("authentik_flows.inspect_flow"):
 85            return
 86        raise Http404
 87
 88    @extend_schema(
 89        responses={
 90            200: FlowInspectionSerializer(),
 91            400: OpenApiResponse(description="No flow plan in session."),
 92        },
 93        request=OpenApiTypes.NONE,
 94        operation_id="flows_inspector_get",
 95    )
 96    def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
 97        """Get current flow state and record it"""
 98        plans = []
 99        for plan in request.session.get(SESSION_KEY_HISTORY, []):
100            plan: FlowPlan
101            if plan.flow_pk != self.flow.pk.hex:
102                continue
103            plan_serializer = FlowInspectorPlanSerializer(
104                instance=plan, context={"request": request}
105            )
106            plans.append(plan_serializer.data)
107        is_completed = False
108        if SESSION_KEY_PLAN in request.session:
109            current_plan: FlowPlan = request.session[SESSION_KEY_PLAN]
110        else:
111            try:
112                current_plan = request.session.get(SESSION_KEY_HISTORY, [])[-1]
113            except IndexError:
114                return Response(status=400)
115            is_completed = True
116        current_serializer = FlowInspectorPlanSerializer(
117            instance=current_plan, context={"request": request}
118        )
119        response = {
120            "plans": plans,
121            "current_plan": current_serializer.data,
122            "is_completed": is_completed,
123        }
124        return Response(response)