authentik.stages.consent.stage

  1"""authentik consent stage"""
  2
  3from hmac import compare_digest
  4from uuid import uuid4
  5
  6from django.http import HttpRequest, HttpResponse
  7from django.utils.timezone import now
  8from django.utils.translation import gettext as _
  9from rest_framework.exceptions import ValidationError
 10from rest_framework.fields import CharField
 11
 12from authentik.core.api.utils import PassiveSerializer
 13from authentik.flows.challenge import (
 14    Challenge,
 15    ChallengeResponse,
 16    WithUserInfoChallenge,
 17)
 18from authentik.flows.planner import PLAN_CONTEXT_APPLICATION, PLAN_CONTEXT_PENDING_USER
 19from authentik.flows.stage import ChallengeStageView
 20from authentik.lib.utils.time import timedelta_from_string
 21from authentik.stages.consent.models import ConsentMode, ConsentStage, UserConsent
 22
 23PLAN_CONTEXT_CONSENT = "consent"
 24PLAN_CONTEXT_CONSENT_HEADER = "consent_header"
 25PLAN_CONTEXT_CONSENT_PERMISSIONS = "consent_permissions"
 26PLAN_CONTEXT_CONSENT_EXTRA_PERMISSIONS = "consent_additional_permissions"
 27PLAN_CONTEXT_CONSENT_TOKEN = "goauthentik.io/stages/consent/token"  # nosec
 28
 29
 30class ConsentPermissionSerializer(PassiveSerializer):
 31    """Permission used for consent"""
 32
 33    name = CharField(allow_blank=True)
 34    id = CharField()
 35
 36
 37class ConsentChallenge(WithUserInfoChallenge):
 38    """Challenge info for consent screens"""
 39
 40    header_text = CharField(required=False)
 41    permissions = ConsentPermissionSerializer(many=True)
 42    additional_permissions = ConsentPermissionSerializer(many=True)
 43    component = CharField(default="ak-stage-consent")
 44    token = CharField(required=True)
 45
 46
 47class ConsentChallengeResponse(ChallengeResponse):
 48    """Consent challenge response, any valid response request is valid"""
 49
 50    component = CharField(default="ak-stage-consent")
 51    token = CharField(required=True)
 52
 53    def validate_token(self, token: str):
 54        if not compare_digest(
 55            token, self.stage.executor.plan.context.get(PLAN_CONTEXT_CONSENT_TOKEN, "")
 56        ):
 57            raise ValidationError(_("Invalid consent token, re-showing prompt"))
 58        return token
 59
 60
 61class ConsentStageView(ChallengeStageView):
 62    """Simple consent checker."""
 63
 64    response_class = ConsentChallengeResponse
 65
 66    def get_challenge(self) -> Challenge:
 67        token = str(uuid4())
 68        self.executor.plan.context[PLAN_CONTEXT_CONSENT_TOKEN] = token
 69        data = {
 70            "permissions": self.executor.plan.context.get(PLAN_CONTEXT_CONSENT_PERMISSIONS, []),
 71            "additional_permissions": self.executor.plan.context.get(
 72                PLAN_CONTEXT_CONSENT_EXTRA_PERMISSIONS, []
 73            ),
 74            "token": token,
 75        }
 76        if PLAN_CONTEXT_CONSENT_HEADER in self.executor.plan.context:
 77            data["header_text"] = self.executor.plan.context[PLAN_CONTEXT_CONSENT_HEADER]
 78        challenge = ConsentChallenge(data=data)
 79        return challenge
 80
 81    def should_always_prompt(self) -> bool:
 82        """Check if the current request should require a prompt for non consent reasons,
 83        i.e. this stage injected from another stage, mode is always required or no application
 84        is set."""
 85        current_stage: ConsentStage = self.executor.current_stage
 86        # Make this StageView work when injected, in which case `current_stage` is an instance
 87        # of the base class, and we don't save any consent, as it is assumed to be a one-time
 88        # prompt
 89        if not isinstance(current_stage, ConsentStage):
 90            return True
 91        # For always require, we always return the challenge
 92        if current_stage.mode == ConsentMode.ALWAYS_REQUIRE:
 93            return True
 94        # at this point we need to check consent from database
 95        if PLAN_CONTEXT_APPLICATION not in self.executor.plan.context:
 96            # No application in this plan, hence we can't check DB and require user consent
 97            return True
 98        return None
 99
100    def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
101        if self.should_always_prompt():
102            return super().get(request, *args, **kwargs)
103        application = self.executor.plan.context[PLAN_CONTEXT_APPLICATION]
104
105        user = self.request.user
106        if PLAN_CONTEXT_PENDING_USER in self.executor.plan.context:
107            user = self.executor.plan.context[PLAN_CONTEXT_PENDING_USER]
108
109        consent: UserConsent | None = UserConsent.objects.filter(
110            user=user, application=application
111        ).first()
112        self.executor.plan.context[PLAN_CONTEXT_CONSENT] = consent
113
114        if consent:
115            perms = self.executor.plan.context.get(PLAN_CONTEXT_CONSENT_PERMISSIONS, [])
116            allowed_perms = set(consent.permissions.split(" ") if consent.permissions != "" else [])
117            requested_perms = set(x["id"] for x in perms)
118
119            if allowed_perms != requested_perms:
120                self.executor.plan.context[PLAN_CONTEXT_CONSENT_PERMISSIONS] = [
121                    x for x in perms if x["id"] in allowed_perms
122                ]
123                self.executor.plan.context[PLAN_CONTEXT_CONSENT_EXTRA_PERMISSIONS] = [
124                    x for x in perms if x["id"] in requested_perms.difference(allowed_perms)
125                ]
126                return super().get(request, *args, **kwargs)
127            return self.executor.stage_ok()
128
129        # No consent found, return consent prompt
130        return super().get(request, *args, **kwargs)
131
132    def challenge_valid(self, response: ChallengeResponse) -> HttpResponse:
133        if self.should_always_prompt():
134            return self.executor.stage_ok()
135        current_stage: ConsentStage = self.executor.current_stage
136        application = self.executor.plan.context[PLAN_CONTEXT_APPLICATION]
137        permissions = self.executor.plan.context.get(
138            PLAN_CONTEXT_CONSENT_PERMISSIONS, []
139        ) + self.executor.plan.context.get(PLAN_CONTEXT_CONSENT_EXTRA_PERMISSIONS, [])
140        permissions_string = " ".join(x["id"] for x in permissions)
141
142        if not self.executor.plan.context.get(PLAN_CONTEXT_CONSENT, None):
143            self.executor.plan.context[PLAN_CONTEXT_CONSENT] = UserConsent(
144                user=self.request.user,
145                application=application,
146            )
147        consent: UserConsent = self.executor.plan.context[PLAN_CONTEXT_CONSENT]
148        consent.permissions = permissions_string
149        if current_stage.mode == ConsentMode.PERMANENT:
150            consent.expiring = False
151        if current_stage.mode == ConsentMode.EXPIRING:
152            consent.expires = now() + timedelta_from_string(current_stage.consent_expire_in)
153        consent.save()
154        return self.executor.stage_ok()