authentik.flows.challenge

  1"""Challenge helpers"""
  2
  3from dataclasses import asdict, is_dataclass
  4from enum import Enum
  5from typing import TYPE_CHECKING, TypedDict
  6from uuid import UUID
  7
  8from django.core.serializers.json import DjangoJSONEncoder
  9from django.db import models
 10from django.http import JsonResponse
 11from rest_framework.fields import BooleanField, CharField, ChoiceField, DictField
 12from rest_framework.request import Request
 13
 14from authentik.core.api.utils import PassiveSerializer, ThemedUrlsSerializer
 15from authentik.lib.utils.errors import exception_to_string
 16
 17if TYPE_CHECKING:
 18    from authentik.flows.stage import StageView
 19
 20PLAN_CONTEXT_TITLE = "title"
 21PLAN_CONTEXT_URL = "url"
 22PLAN_CONTEXT_ATTRS = "attrs"
 23
 24
 25class FlowLayout(models.TextChoices):
 26    """Flow layouts"""
 27
 28    STACKED = "stacked"
 29    CONTENT_LEFT = "content_left"
 30    CONTENT_RIGHT = "content_right"
 31    SIDEBAR_LEFT = "sidebar_left"
 32    SIDEBAR_RIGHT = "sidebar_right"
 33
 34    SIDEBAR_LEFT_FRAME_BACKGROUND = "sidebar_left_frame_background"
 35    SIDEBAR_RIGHT_FRAME_BACKGROUND = "sidebar_right_frame_background"
 36
 37
 38class ErrorDetailSerializer(PassiveSerializer):
 39    """Serializer for rest_framework's error messages"""
 40
 41    string = CharField()
 42    code = CharField()
 43
 44
 45class ContextualFlowInfo(PassiveSerializer):
 46    """Contextual flow information for a challenge"""
 47
 48    title = CharField(required=False, allow_blank=True)
 49    background = CharField(required=False)
 50    background_themed_urls = ThemedUrlsSerializer(required=False, allow_null=True)
 51    cancel_url = CharField()
 52    layout = ChoiceField(choices=[(x.value, x.name) for x in FlowLayout])
 53
 54
 55class Challenge(PassiveSerializer):
 56    """Challenge that gets sent to the client based on which stage
 57    is currently active"""
 58
 59    flow_info = ContextualFlowInfo(required=False)
 60    component = CharField(default="")
 61
 62    response_errors = DictField(
 63        child=ErrorDetailSerializer(many=True), allow_empty=True, required=False
 64    )
 65
 66
 67class RedirectChallenge(Challenge):
 68    """Challenge type to redirect the client"""
 69
 70    to = CharField()
 71    component = CharField(default="xak-flow-redirect")
 72
 73
 74class ShellChallenge(Challenge):
 75    """challenge type to render HTML as-is"""
 76
 77    body = CharField()
 78    component = CharField(default="xak-flow-shell")
 79
 80
 81class WithUserInfoChallenge(Challenge):
 82    """Challenge base which shows some user info"""
 83
 84    pending_user = CharField(allow_blank=True)
 85    pending_user_avatar = CharField()
 86
 87
 88class FlowErrorChallenge(Challenge):
 89    """Challenge class when an unhandled error occurs during a stage. Normal users
 90    are shown an error message, superusers are shown a full stacktrace."""
 91
 92    component = CharField(default="ak-stage-flow-error")
 93
 94    request_id = CharField()
 95
 96    error = CharField(required=False)
 97    traceback = CharField(required=False)
 98
 99    def __init__(self, request: Request | None = None, error: Exception | None = None):
100        super().__init__(data={})
101        if not request or not error:
102            return
103        self.initial_data["request_id"] = request.request_id
104        from authentik.core.models import USER_ATTRIBUTE_DEBUG
105
106        if request.user and request.user.is_authenticated:
107            if request.user.is_superuser or request.user.group_attributes(request).get(
108                USER_ATTRIBUTE_DEBUG, False
109            ):
110                self.initial_data["error"] = str(error)
111                self.initial_data["traceback"] = exception_to_string(error)
112
113
114class AccessDeniedChallenge(WithUserInfoChallenge):
115    """Challenge when a flow's active stage calls `stage_invalid()`."""
116
117    component = CharField(default="ak-stage-access-denied")
118
119    error_message = CharField(required=False)
120
121
122class SessionEndChallenge(WithUserInfoChallenge):
123    """Challenge for ending a session"""
124
125    component = CharField(default="ak-stage-session-end")
126
127    application_name = CharField(required=False)
128    application_launch_url = CharField(required=False)
129
130    invalidation_flow_url = CharField(required=False)
131    brand_name = CharField(required=True)
132
133
134class PermissionDict(TypedDict):
135    """Consent Permission"""
136
137    id: str
138    name: str
139
140
141class ChallengeResponse(PassiveSerializer):
142    """Base class for all challenge responses"""
143
144    stage: StageView | None
145    component = CharField(default="xak-flow-response-default")
146
147    def __init__(self, instance=None, data=None, **kwargs):
148        self.stage = kwargs.pop("stage", None)
149        super().__init__(instance=instance, data=data, **kwargs)
150
151
152class AutosubmitChallenge(Challenge):
153    """Autosubmit challenge used to send and navigate a POST request"""
154
155    url = CharField()
156    attrs = DictField(child=CharField(allow_blank=True), allow_empty=True)
157    title = CharField(required=False)
158    component = CharField(default="ak-stage-autosubmit")
159
160
161class AutoSubmitChallengeResponse(ChallengeResponse):
162    """Pseudo class for autosubmit response"""
163
164    component = CharField(default="ak-stage-autosubmit")
165
166
167class FrameChallenge(Challenge):
168    """Challenge type to render a frame"""
169
170    component = CharField(default="xak-flow-frame")
171    url = CharField()
172    loading_overlay = BooleanField(default=False)
173    loading_text = CharField()
174
175
176class FrameChallengeResponse(ChallengeResponse):
177
178    component = CharField(default="xak-flow-frame")
179
180
181class DataclassEncoder(DjangoJSONEncoder):
182    """Convert any dataclass to json"""
183
184    def default(self, o):
185        if is_dataclass(o):
186            return asdict(o)
187        if isinstance(o, UUID):
188            return str(o)
189        if isinstance(o, Enum):
190            return o.value
191        return super().default(o)  # pragma: no cover
192
193
194class HttpChallengeResponse(JsonResponse):
195    """Subclass of JsonResponse that uses the `DataclassEncoder`"""
196
197    def __init__(self, challenge, **kwargs) -> None:
198        super().__init__(challenge.data, encoder=DataclassEncoder, **kwargs)