authentik.providers.rac.views

  1"""RAC Views"""
  2
  3from typing import Any
  4
  5from django.http import Http404, HttpRequest, HttpResponse
  6from django.shortcuts import get_object_or_404, redirect
  7from django.urls import reverse
  8from django.utils.timezone import now
  9from django.utils.translation import gettext as _
 10
 11from authentik.core.models import Application
 12from authentik.core.views.interface import InterfaceView
 13from authentik.events.models import Event, EventAction
 14from authentik.flows.challenge import RedirectChallenge
 15from authentik.flows.exceptions import FlowNonApplicableException
 16from authentik.flows.models import in_memory_stage
 17from authentik.flows.planner import PLAN_CONTEXT_APPLICATION, FlowPlanner
 18from authentik.flows.stage import RedirectStage
 19from authentik.lib.utils.time import timedelta_from_string
 20from authentik.policies.engine import PolicyEngine
 21from authentik.policies.views import PolicyAccessView
 22from authentik.providers.rac.models import ConnectionToken, Endpoint, RACProvider
 23from authentik.stages.prompt.stage import PLAN_CONTEXT_PROMPT
 24
 25PLAN_CONNECTION_SETTINGS = "connection_settings"
 26
 27
 28class RACStartView(PolicyAccessView):
 29    """Start a RAC connection by checking access and creating a connection token"""
 30
 31    endpoint: Endpoint
 32
 33    def resolve_provider_application(self):
 34        self.application = get_object_or_404(Application, slug=self.kwargs["app"])
 35        # Endpoint permissions are validated in the RACFinalStage below
 36        self.endpoint = get_object_or_404(Endpoint, pk=self.kwargs["endpoint"])
 37        self.provider = RACProvider.objects.get(application=self.application)
 38
 39    def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
 40        """Start flow planner for RAC provider"""
 41        planner = FlowPlanner(self.provider.authorization_flow)
 42        planner.allow_empty_flows = True
 43        try:
 44            plan = planner.plan(
 45                self.request,
 46                {
 47                    PLAN_CONTEXT_APPLICATION: self.application,
 48                },
 49            )
 50        except FlowNonApplicableException:
 51            raise Http404 from None
 52        plan.append_stage(
 53            in_memory_stage(
 54                RACFinalStage,
 55                application=self.application,
 56                endpoint=self.endpoint,
 57                provider=self.provider,
 58            )
 59        )
 60        return plan.to_redirect(request, self.provider.authorization_flow)
 61
 62
 63class RACInterface(InterfaceView):
 64    """Start RAC connection"""
 65
 66    template_name = "if/rac.html"
 67    token: ConnectionToken
 68
 69    def dispatch(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
 70        # Early sanity check to ensure token still exists
 71        token = ConnectionToken.objects.filter(
 72            token=self.kwargs["token"],
 73            session__session__session_key=request.session.session_key,
 74        ).first()
 75        if not token:
 76            return redirect("authentik_core:if-user")
 77        self.token = token
 78        return super().dispatch(request, *args, **kwargs)
 79
 80    def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
 81        kwargs["token"] = self.token
 82        return super().get_context_data(**kwargs)
 83
 84
 85class RACFinalStage(RedirectStage):
 86    """RAC Connection final stage, set the connection token in the stage"""
 87
 88    endpoint: Endpoint
 89    provider: RACProvider
 90    application: Application
 91
 92    def dispatch(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
 93        self.endpoint = self.executor.current_stage.endpoint
 94        self.provider = self.executor.current_stage.provider
 95        self.application = self.executor.current_stage.application
 96        # Check policies bound to endpoint directly
 97        engine = PolicyEngine(self.endpoint, self.request.user, self.request)
 98        engine.use_cache = False
 99        engine.build()
100        passing = engine.result
101        if not passing.passing:
102            return self.executor.stage_invalid(", ".join(passing.messages))
103        # Check if we're already at the maximum connection limit
104        all_tokens = ConnectionToken.objects.filter(
105            endpoint=self.endpoint,
106        )
107        if self.endpoint.maximum_connections > -1:
108            if all_tokens.count() >= self.endpoint.maximum_connections:
109                msg = [_("Maximum connection limit reached.")]
110                # Check if any other tokens exist for the current user, and inform them
111                # they are already connected
112                if all_tokens.filter(session__user=self.request.user).exists():
113                    msg.append(_("(You are already connected in another tab/window)"))
114                return self.executor.stage_invalid(" ".join(msg))
115        return super().dispatch(request, *args, **kwargs)
116
117    def get_challenge(self, *args, **kwargs) -> RedirectChallenge:
118        settings = self.executor.plan.context.get(PLAN_CONNECTION_SETTINGS)
119        if not settings:
120            settings = self.executor.plan.context.get(PLAN_CONTEXT_PROMPT, {}).get(
121                PLAN_CONNECTION_SETTINGS
122            )
123        token = ConnectionToken.objects.create(
124            provider=self.provider,
125            endpoint=self.endpoint,
126            settings=settings or {},
127            session=self.request.session["authenticatedsession"],
128            expires=now() + timedelta_from_string(self.provider.connection_expiry),
129            expiring=True,
130        )
131        Event.new(
132            EventAction.AUTHORIZE_APPLICATION,
133            authorized_application=self.application,
134            flow=self.executor.plan.flow_pk,
135            endpoint=self.endpoint.name,
136        ).from_http(self.request)
137        self.executor.current_stage.destination = self.request.build_absolute_uri(
138            reverse("authentik_providers_rac:if-rac", kwargs={"token": str(token.token)})
139        )
140        return super().get_challenge(*args, **kwargs)