authentik.stages.prompt.stage

  1"""Prompt Stage Logic"""
  2
  3from collections.abc import Callable
  4from email.policy import Policy
  5from types import MethodType
  6from typing import Any
  7
  8from django.contrib.messages import INFO, add_message
  9from django.db.models.query import QuerySet
 10from django.http import HttpRequest, HttpResponse
 11from django.http.request import QueryDict
 12from django.utils.translation import gettext_lazy as _
 13from rest_framework.fields import (
 14    BooleanField,
 15    CharField,
 16    ChoiceField,
 17    IntegerField,
 18    ListField,
 19    empty,
 20)
 21from rest_framework.serializers import ValidationError
 22
 23from authentik.core.api.utils import PassiveSerializer
 24from authentik.core.models import User
 25from authentik.flows.challenge import Challenge, ChallengeResponse
 26from authentik.flows.planner import FlowPlan
 27from authentik.flows.stage import ChallengeStageView
 28from authentik.policies.engine import PolicyEngine
 29from authentik.policies.models import PolicyBinding, PolicyBindingModel, PolicyEngineMode
 30from authentik.stages.prompt.models import FieldTypes, Prompt, PromptStage
 31from authentik.stages.prompt.signals import password_validate
 32
 33PLAN_CONTEXT_PROMPT = "prompt_data"
 34
 35
 36class PromptChoiceSerializer(PassiveSerializer):
 37    """Serializer for a single Choice field"""
 38
 39    value = CharField(required=True)
 40    label = CharField(required=True)
 41
 42
 43class StagePromptSerializer(PassiveSerializer):
 44    """Serializer for a single Prompt field"""
 45
 46    field_key = CharField()
 47    label = CharField(allow_blank=True)
 48    type = ChoiceField(choices=FieldTypes.choices)
 49    required = BooleanField()
 50    placeholder = CharField(allow_blank=True)
 51    initial_value = CharField(allow_blank=True)
 52    order = IntegerField()
 53    sub_text = CharField(allow_blank=True)
 54    choices = ListField(child=PromptChoiceSerializer(), allow_empty=True, allow_null=True)
 55
 56
 57class PromptChallenge(Challenge):
 58    """Initial challenge being sent, define fields"""
 59
 60    fields = StagePromptSerializer(many=True)
 61    component = CharField(default="ak-stage-prompt")
 62
 63
 64class PromptChallengeResponse(ChallengeResponse):
 65    """Validate response, fields are dynamically created based
 66    on the stage"""
 67
 68    stage_instance: PromptStage
 69
 70    component = CharField(default="ak-stage-prompt")
 71
 72    def __init__(self, *args, **kwargs):
 73        stage: PromptStage = kwargs.pop("stage_instance", None)
 74        plan: FlowPlan = kwargs.pop("plan", None)
 75        request: HttpRequest = kwargs.pop("request", None)
 76        user: User = kwargs.pop("user", None)
 77        super().__init__(*args, **kwargs)
 78        self.stage_instance = stage
 79        self.plan = plan
 80        self.request = request
 81        if not self.stage_instance:
 82            return
 83        # list() is called so we only load the fields once
 84        fields = list(self.stage_instance.fields.all())
 85        for field in fields:
 86            field: Prompt
 87            choices = field.get_choices(
 88                plan.context.get(PLAN_CONTEXT_PROMPT, {}), user, self.request
 89            )
 90            current = field.get_initial_value(
 91                plan.context.get(PLAN_CONTEXT_PROMPT, {}), user, self.request
 92            )
 93            self.fields[field.field_key] = field.field(current, choices)
 94            # Special handling for fields with username type
 95            # these check for existing users with the same username
 96            if field.type == FieldTypes.USERNAME:
 97                setattr(
 98                    self,
 99                    f"validate_{field.field_key}",
100                    MethodType(username_field_validator_factory(), self),
101                )
102            # Check if we have a password field, add a handler that sends a signal
103            # to validate it
104            if field.type == FieldTypes.PASSWORD:
105                setattr(
106                    self,
107                    f"validate_{field.field_key}",
108                    MethodType(password_single_validator_factory(), self),
109                )
110
111        self.field_order = sorted(fields, key=lambda x: x.order)
112
113    def _validate_password_fields(self, *field_names):
114        """Check if the value of all password fields match by merging them into a set
115        and checking the length"""
116        all_passwords = {self.initial_data[x] for x in field_names}
117        if len(all_passwords) > 1:
118            raise ValidationError(_("Passwords don't match."))
119
120    def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
121        # Check if we have any static or hidden fields, and ensure they
122        # still have the same value
123        static_hidden_fields: QuerySet[Prompt] = self.stage_instance.fields.filter(
124            type__in=[
125                FieldTypes.HIDDEN,
126                FieldTypes.STATIC,
127                FieldTypes.ALERT_INFO,
128                FieldTypes.ALERT_WARNING,
129                FieldTypes.ALERT_DANGER,
130                FieldTypes.TEXT_READ_ONLY,
131                FieldTypes.TEXT_AREA_READ_ONLY,
132            ]
133        )
134        for static_hidden in static_hidden_fields:
135            field = self.fields[static_hidden.field_key]
136            default = field.default
137            # Prevent rest_framework.fields.empty from ending up in policies and events
138            if default == empty:
139                default = ""
140            attrs[static_hidden.field_key] = default
141
142        # Check if we have two password fields, and make sure they are the same
143        password_fields: QuerySet[Prompt] = self.stage_instance.fields.filter(
144            type=FieldTypes.PASSWORD
145        )
146        if password_fields.exists() and password_fields.count() == 2:  # noqa: PLR2004
147            self._validate_password_fields(*[field.field_key for field in password_fields])
148
149        engine = ListPolicyEngine(
150            self.stage_instance.validation_policies.all(),
151            self.stage.get_pending_user(),
152            self.request,
153        )
154        engine.mode = PolicyEngineMode.MODE_ALL
155        engine.request.context[PLAN_CONTEXT_PROMPT] = attrs
156        engine.use_cache = False
157        engine.build()
158        result = engine.result
159        if not result.passing:
160            raise ValidationError(list(result.messages))
161        else:
162            for msg in result.messages:
163                add_message(self.request, INFO, msg)
164        return attrs
165
166
167def username_field_validator_factory() -> Callable[[PromptChallengeResponse, str], Any]:
168    """Return a `clean_` method for `field`. Clean method checks if username is taken already."""
169
170    def username_field_validator(self: PromptChallengeResponse, value: str) -> Any:
171        """Check for duplicate usernames"""
172        pending_user = self.stage.get_pending_user()
173        query = User.objects.all()
174        if pending_user.pk:
175            query = query.exclude(username=pending_user.username)
176        if query.filter(username=value).exists():
177            raise ValidationError("Username is already taken.")
178        return value
179
180    return username_field_validator
181
182
183def password_single_validator_factory() -> Callable[[PromptChallengeResponse, str], Any]:
184    """Return a `clean_` method for `field`. Clean method checks if the password meets configured
185    PasswordPolicy."""
186
187    def password_single_clean(self: PromptChallengeResponse, value: str) -> Any:
188        """Send password validation signals for e.g. LDAP Source"""
189        password_validate.send(sender=self, password=value, plan_context=self.plan.context)
190        return value
191
192    return password_single_clean
193
194
195class ListPolicyEngine(PolicyEngine):
196    """Slightly modified policy engine, which uses a list instead of a PolicyBindingModel"""
197
198    def __init__(self, policies: list[Policy], user: User, request: HttpRequest = None) -> None:
199        super().__init__(PolicyBindingModel(), user, request)
200        self.__list = policies
201        self.use_cache = False
202
203    def bindings(self):
204        for idx, policy in enumerate(self.__list):
205            yield PolicyBinding(
206                policy=policy,
207                order=idx,
208            )
209
210
211class PromptStageView(ChallengeStageView):
212    """Prompt Stage, save form data in plan context."""
213
214    response_class = PromptChallengeResponse
215
216    def get_prompt_challenge_fields(self, fields: list[Prompt], context: dict, dry_run=False):
217        """Get serializers for all fields in `fields`, using the context `context`.
218        If `dry_run` is set, property mapping expression errors are raised, otherwise they
219        are logged and events are created"""
220        serializers = []
221        for field in fields:
222            data = StagePromptSerializer(field).data
223            # Ensure all placeholders and initial values are str, as
224            # otherwise further in we can fail serializer validation if we return
225            # some types such as bool
226            # choices can be a dict with value and label
227            choices = field.get_choices(context, self.get_pending_user(), self.request, dry_run)
228            if choices:
229                data["choices"] = list(self.clean_choices(choices))
230            else:
231                data["choices"] = None
232            data["placeholder"] = str(
233                field.get_placeholder(context, self.get_pending_user(), self.request, dry_run)
234            )
235            data["initial_value"] = str(
236                field.get_initial_value(context, self.get_pending_user(), self.request, dry_run)
237            )
238            serializers.append(data)
239        return serializers
240
241    def clean_choices(self, choices):
242        for choice in choices:
243            label, value = choice, choice
244            if isinstance(choice, dict):
245                label = choice.get("label", "")
246                value = choice.get("value", "")
247            yield {"label": str(label), "value": str(value)}
248
249    def get_challenge(self, *args, **kwargs) -> Challenge:
250        fields: list[Prompt] = list(self.executor.current_stage.fields.all().order_by("order"))
251        context_prompt = self.executor.plan.context.get(PLAN_CONTEXT_PROMPT, {})
252        serializers = self.get_prompt_challenge_fields(fields, context_prompt)
253        challenge = PromptChallenge(
254            data={
255                "fields": serializers,
256            },
257        )
258        return challenge
259
260    def get_response_instance(self, data: QueryDict) -> ChallengeResponse:
261        if not self.executor.plan:  # pragma: no cover
262            raise ValueError
263        return PromptChallengeResponse(
264            instance=None,
265            data=data,
266            request=self.request,
267            stage_instance=self.executor.current_stage,
268            stage=self,
269            plan=self.executor.plan,
270            user=self.get_pending_user(),
271        )
272
273    def challenge_valid(self, response: ChallengeResponse) -> HttpResponse:
274        if PLAN_CONTEXT_PROMPT not in self.executor.plan.context:
275            self.executor.plan.context[PLAN_CONTEXT_PROMPT] = {}
276        self.executor.plan.context[PLAN_CONTEXT_PROMPT].update(response.validated_data)
277        return self.executor.stage_ok()