authentik.core.models

   1"""authentik core models"""
   2
   3import re
   4import traceback
   5from datetime import datetime
   6from enum import StrEnum
   7from hashlib import sha256
   8from typing import Any, Self
   9from uuid import uuid4
  10
  11import pgtrigger
  12from deepmerge import always_merger
  13from django.contrib.auth.hashers import check_password
  14from django.contrib.auth.models import AbstractUser, Permission
  15from django.contrib.auth.models import UserManager as DjangoUserManager
  16from django.contrib.sessions.base_session import AbstractBaseSession
  17from django.core.validators import validate_slug
  18from django.db import models
  19from django.db.models import Manager, Q, QuerySet, options
  20from django.http import HttpRequest
  21from django.utils.functional import cached_property
  22from django.utils.timezone import now
  23from django.utils.translation import gettext_lazy as _
  24from guardian.conf import settings
  25from guardian.models import RoleModelPermission, RoleObjectPermission
  26from model_utils.managers import InheritanceManager
  27from psqlextra.indexes import UniqueIndex
  28from psqlextra.models import PostgresMaterializedViewModel
  29from rest_framework.serializers import Serializer
  30from structlog.stdlib import get_logger
  31
  32from authentik.admin.files.fields import FileField
  33from authentik.admin.files.manager import get_file_manager
  34from authentik.admin.files.usage import FileUsage
  35from authentik.blueprints.models import ManagedModel
  36from authentik.core.expression.exceptions import PropertyMappingExpressionException
  37from authentik.core.types import UILoginButton, UserSettingSerializer
  38from authentik.lib.avatars import get_avatar
  39from authentik.lib.expression.exceptions import ControlFlowException
  40from authentik.lib.generators import generate_id
  41from authentik.lib.merge import MERGE_LIST_UNIQUE
  42from authentik.lib.models import (
  43    CreatedUpdatedModel,
  44    DomainlessFormattedURLValidator,
  45    SerializerModel,
  46)
  47from authentik.lib.utils.inheritance import get_deepest_child
  48from authentik.lib.utils.reflection import class_to_path
  49from authentik.lib.utils.time import timedelta_from_string
  50from authentik.policies.models import PolicyBindingModel
  51from authentik.rbac.models import Role
  52from authentik.tenants.models import DEFAULT_TOKEN_DURATION, DEFAULT_TOKEN_LENGTH
  53from authentik.tenants.utils import get_current_tenant, get_unique_identifier
  54
  55LOGGER = get_logger()
  56USERNAME_MAX_LENGTH = 150
  57USER_PATH_SYSTEM_PREFIX = "goauthentik.io"
  58_USER_ATTR_PREFIX = f"{USER_PATH_SYSTEM_PREFIX}/user"
  59USER_ATTRIBUTE_DEBUG = f"{_USER_ATTR_PREFIX}/debug"
  60USER_ATTRIBUTE_GENERATED = f"{_USER_ATTR_PREFIX}/generated"
  61USER_ATTRIBUTE_EXPIRES = f"{_USER_ATTR_PREFIX}/expires"
  62USER_ATTRIBUTE_DELETE_ON_LOGOUT = f"{_USER_ATTR_PREFIX}/delete-on-logout"
  63USER_ATTRIBUTE_SOURCES = f"{_USER_ATTR_PREFIX}/sources"
  64USER_ATTRIBUTE_TOKEN_EXPIRING = f"{_USER_ATTR_PREFIX}/token-expires"  # nosec
  65USER_ATTRIBUTE_TOKEN_MAXIMUM_LIFETIME = f"{_USER_ATTR_PREFIX}/token-maximum-lifetime"  # nosec
  66USER_ATTRIBUTE_CHANGE_USERNAME = f"{_USER_ATTR_PREFIX}/can-change-username"
  67USER_ATTRIBUTE_CHANGE_NAME = f"{_USER_ATTR_PREFIX}/can-change-name"
  68USER_ATTRIBUTE_CHANGE_EMAIL = f"{_USER_ATTR_PREFIX}/can-change-email"
  69USER_PATH_SERVICE_ACCOUNT = f"{USER_PATH_SYSTEM_PREFIX}/service-accounts"
  70
  71options.DEFAULT_NAMES = options.DEFAULT_NAMES + (
  72    # used_by API that allows models to specify if they shadow an object
  73    # for example the proxy provider which is built on top of an oauth provider
  74    "authentik_used_by_shadows",
  75)
  76
  77GROUP_RECURSION_LIMIT = 20
  78
  79MANAGED_ROLE_PREFIX_USER = "ak-managed-role--user"
  80MANAGED_ROLE_PREFIX_GROUP = "ak-managed-role--group"
  81
  82
  83def managed_role_name(user_or_group: models.Model):
  84    if isinstance(user_or_group, User):
  85        return f"{MANAGED_ROLE_PREFIX_USER}-{user_or_group.pk}"
  86    if isinstance(user_or_group, Group):
  87        return f"{MANAGED_ROLE_PREFIX_GROUP}-{user_or_group.pk}"
  88    raise TypeError("Managed roles are only available for User or Group.")
  89
  90
  91def default_token_duration() -> datetime:
  92    """Default duration a Token is valid"""
  93    current_tenant = get_current_tenant()
  94    token_duration = (
  95        current_tenant.default_token_duration
  96        if hasattr(current_tenant, "default_token_duration")
  97        else DEFAULT_TOKEN_DURATION
  98    )
  99    return now() + timedelta_from_string(token_duration)
 100
 101
 102def default_token_key() -> str:
 103    """Default token key"""
 104    current_tenant = get_current_tenant()
 105    token_length = (
 106        current_tenant.default_token_length
 107        if hasattr(current_tenant, "default_token_length")
 108        else DEFAULT_TOKEN_LENGTH
 109    )
 110    # We use generate_id since the chars in the key should be easy
 111    # to use in Emails (for verification) and URLs (for recovery)
 112    return generate_id(token_length)
 113
 114
 115class UserTypes(models.TextChoices):
 116    """User types, both for grouping, licensing and permissions in the case
 117    of the internal_service_account"""
 118
 119    INTERNAL = "internal"
 120    EXTERNAL = "external"
 121
 122    # User-created service accounts
 123    SERVICE_ACCOUNT = "service_account"
 124
 125    # Special user type for internally managed and created service
 126    # accounts, such as outpost users
 127    INTERNAL_SERVICE_ACCOUNT = "internal_service_account"
 128
 129
 130class AttributesMixin(models.Model):
 131    """Adds an attributes property to a model"""
 132
 133    attributes = models.JSONField(default=dict, blank=True)
 134
 135    class Meta:
 136        abstract = True
 137
 138    def update_attributes(self, properties: dict[str, Any]):
 139        """Update fields and attributes, but correctly by merging dicts"""
 140        needs_update = False
 141        for key, value in properties.items():
 142            if key == "attributes":
 143                continue
 144            if getattr(self, key, None) != value:
 145                setattr(self, key, value)
 146                needs_update = True
 147        final_attributes = {}
 148        MERGE_LIST_UNIQUE.merge(final_attributes, self.attributes)
 149        MERGE_LIST_UNIQUE.merge(final_attributes, properties.get("attributes", {}))
 150        if self.attributes != final_attributes:
 151            self.attributes = final_attributes
 152            needs_update = True
 153        if needs_update:
 154            self.save()
 155
 156    @classmethod
 157    def update_or_create_attributes(
 158        cls, query: dict[str, Any], properties: dict[str, Any]
 159    ) -> tuple[Self, bool]:
 160        """Same as django's update_or_create but correctly updates attributes by merging dicts"""
 161        instance = cls.objects.filter(**query).first()
 162        if not instance:
 163            return cls.objects.create(**properties), True
 164        instance.update_attributes(properties)
 165        return instance, False
 166
 167
 168class GroupQuerySet(QuerySet):
 169    def with_descendants(self):
 170        pks = self.values_list("pk", flat=True)
 171        return Group.objects.filter(Q(pk__in=pks) | Q(ancestor_nodes__ancestor__in=pks)).distinct()
 172
 173    def with_ancestors(self):
 174        pks = self.values_list("pk", flat=True)
 175        return Group.objects.filter(
 176            Q(pk__in=pks) | Q(descendant_nodes__descendant__in=pks)
 177        ).distinct()
 178
 179
 180class Group(SerializerModel, AttributesMixin):
 181    """Group model which supports a hierarchy and has attributes"""
 182
 183    group_uuid = models.UUIDField(primary_key=True, editable=False, default=uuid4)
 184
 185    name = models.TextField(verbose_name=_("name"), unique=True)
 186    is_superuser = models.BooleanField(
 187        default=False, help_text=_("Users added to this group will be superusers.")
 188    )
 189
 190    roles = models.ManyToManyField("authentik_rbac.Role", related_name="groups", blank=True)
 191
 192    parents = models.ManyToManyField(
 193        "Group",
 194        blank=True,
 195        symmetrical=False,
 196        through="GroupParentageNode",
 197        related_name="children",
 198    )
 199
 200    objects = GroupQuerySet.as_manager()
 201
 202    class Meta:
 203        indexes = (
 204            models.Index(fields=["name"]),
 205            models.Index(fields=["is_superuser"]),
 206        )
 207        verbose_name = _("Group")
 208        verbose_name_plural = _("Groups")
 209        permissions = [
 210            ("add_user_to_group", _("Add user to group")),
 211            ("remove_user_from_group", _("Remove user from group")),
 212            ("enable_group_superuser", _("Enable superuser status")),
 213            ("disable_group_superuser", _("Disable superuser status")),
 214        ]
 215
 216    def __str__(self):
 217        return f"Group {self.name}"
 218
 219    @property
 220    def serializer(self) -> Serializer:
 221        from authentik.core.api.groups import GroupSerializer
 222
 223        return GroupSerializer
 224
 225    @property
 226    def num_pk(self) -> int:
 227        """Get a numerical, int32 ID for the group"""
 228        # int max is 2147483647 (10 digits) so 9 is the max usable
 229        # in the LDAP Outpost we use the last 5 chars so match here
 230        return int(str(self.pk.int)[:5])
 231
 232    def is_member(self, user: User) -> bool:
 233        """Recursively check if `user` is member of us, or any parent."""
 234        return user.all_groups().filter(group_uuid=self.group_uuid).exists()
 235
 236    def all_roles(self) -> QuerySet[Role]:
 237        """Get all roles of this group and all of its ancestors."""
 238        return Role.objects.filter(
 239            groups__in=Group.objects.filter(pk=self.pk).with_ancestors()
 240        ).distinct()
 241
 242    def get_managed_role(self, create=False):
 243        if create:
 244            name = managed_role_name(self)
 245            role, created = Role.objects.get_or_create(name=name, managed=name)
 246            if created:
 247                role.groups.add(self)
 248            return role
 249        else:
 250            return Role.objects.filter(name=managed_role_name(self)).first()
 251
 252    def assign_perms_to_managed_role(
 253        self,
 254        perms: str | list[str] | Permission | list[Permission],
 255        obj: models.Model | None = None,
 256    ):
 257        if not perms:
 258            return
 259        role = self.get_managed_role(create=True)
 260        role.assign_perms(perms, obj)
 261
 262
 263class GroupParentageNode(models.Model):
 264    uuid = models.UUIDField(primary_key=True, editable=False, default=uuid4)
 265
 266    child = models.ForeignKey(Group, related_name="parent_nodes", on_delete=models.CASCADE)
 267    parent = models.ForeignKey(Group, related_name="child_nodes", on_delete=models.CASCADE)
 268
 269    class Meta:
 270        verbose_name = _("Group Parentage Node")
 271        verbose_name_plural = _("Group Parentage Nodes")
 272
 273        db_table = "authentik_core_groupparentage"
 274
 275        triggers = [
 276            pgtrigger.Trigger(
 277                name="refresh_groupancestry",
 278                operation=pgtrigger.Insert | pgtrigger.Update | pgtrigger.Delete,
 279                when=pgtrigger.After,
 280                func="""
 281                    REFRESH MATERIALIZED VIEW CONCURRENTLY authentik_core_groupancestry;
 282                    RETURN NULL;
 283                """,
 284            ),
 285        ]
 286
 287    def __str__(self) -> str:
 288        return f"Group Parentage Node from #{self.child_id} to {self.parent_id}"
 289
 290
 291class GroupAncestryNode(PostgresMaterializedViewModel):
 292    descendant = models.ForeignKey(
 293        Group, related_name="ancestor_nodes", on_delete=models.DO_NOTHING
 294    )
 295    ancestor = models.ForeignKey(
 296        Group, related_name="descendant_nodes", on_delete=models.DO_NOTHING
 297    )
 298
 299    class Meta:
 300        # This is a transitive closure of authentik_core_groupparentage
 301        # See https://en.wikipedia.org/wiki/Transitive_closure#In_graph_theory
 302        db_table = "authentik_core_groupancestry"
 303        indexes = [
 304            models.Index(fields=["descendant"]),
 305            models.Index(fields=["ancestor"]),
 306            UniqueIndex(fields=["id"]),
 307        ]
 308
 309    class ViewMeta:
 310        query = """
 311            WITH RECURSIVE accumulator AS (
 312                SELECT
 313                child_id::text || '-' || parent_id::text as id,
 314                child_id AS descendant_id,
 315                parent_id AS ancestor_id
 316                FROM authentik_core_groupparentage
 317
 318                UNION
 319
 320                SELECT
 321                accumulator.descendant_id::text || '-' || current.parent_id::text as id,
 322                accumulator.descendant_id,
 323                current.parent_id AS ancestor_id
 324                FROM accumulator
 325                JOIN authentik_core_groupparentage current
 326                ON accumulator.ancestor_id = current.child_id
 327            )
 328            SELECT * FROM accumulator
 329        """
 330
 331    def __str__(self) -> str:
 332        return f"Group Ancestry Node from {self.descendant_id} to {self.ancestor_id}"
 333
 334
 335class UserQuerySet(models.QuerySet):
 336    """User queryset"""
 337
 338    def exclude_anonymous(self):
 339        """Exclude anonymous user"""
 340        return self.exclude(**{User.USERNAME_FIELD: settings.ANONYMOUS_USER_NAME})
 341
 342
 343class UserManager(DjangoUserManager):
 344    """User manager that doesn't assign is_superuser and is_staff"""
 345
 346    def get_queryset(self):
 347        """Create special user queryset"""
 348        return UserQuerySet(self.model, using=self._db)
 349
 350    def create_user(self, username, email=None, password=None, **extra_fields):
 351        """User manager that doesn't assign is_superuser and is_staff"""
 352        return self._create_user(username, email, password, **extra_fields)
 353
 354    def exclude_anonymous(self) -> QuerySet:
 355        """Exclude anonymous user"""
 356        return self.get_queryset().exclude_anonymous()
 357
 358
 359class User(SerializerModel, AttributesMixin, AbstractUser):
 360    """authentik User model, based on django's contrib auth user model."""
 361
 362    # Overwriting PermissionsMixin: permissions are handled by roles.
 363    # (This knowingly violates the Liskov substitution principle. It is better to fail loudly.)
 364    user_permissions = None
 365
 366    uuid = models.UUIDField(default=uuid4, editable=False, unique=True)
 367    name = models.TextField(help_text=_("User's display name."))
 368    path = models.TextField(default="users")
 369    type = models.TextField(choices=UserTypes.choices, default=UserTypes.INTERNAL)
 370
 371    sources = models.ManyToManyField("Source", through="UserSourceConnection")
 372    groups = models.ManyToManyField("Group", related_name="users")
 373    roles = models.ManyToManyField("authentik_rbac.Role", related_name="users", blank=True)
 374    password_change_date = models.DateTimeField(auto_now_add=True)
 375
 376    last_updated = models.DateTimeField(auto_now=True)
 377
 378    objects = UserManager()
 379
 380    class Meta:
 381        verbose_name = _("User")
 382        verbose_name_plural = _("Users")
 383        permissions = [
 384            ("reset_user_password", _("Reset Password")),
 385            ("impersonate", _("Can impersonate other users")),
 386            ("preview_user", _("Can preview user data sent to providers")),
 387            ("view_user_applications", _("View applications the user has access to")),
 388        ]
 389        indexes = [
 390            models.Index(fields=["last_login"]),
 391            models.Index(fields=["password_change_date"]),
 392            models.Index(fields=["uuid"]),
 393            models.Index(fields=["path"]),
 394            models.Index(fields=["type"]),
 395            models.Index(fields=["date_joined"]),
 396            models.Index(fields=["last_updated"]),
 397        ]
 398
 399    def __str__(self):
 400        return self.username
 401
 402    @staticmethod
 403    def default_path() -> str:
 404        """Get the default user path"""
 405        return User._meta.get_field("path").default
 406
 407    def all_groups(self) -> QuerySet[Group]:
 408        """Recursively get all groups this user is a member of."""
 409        return self.groups.all().with_ancestors()
 410
 411    def all_roles(self) -> QuerySet[Role]:
 412        """Get all roles of this user and all of its groups (recursively)."""
 413        return Role.objects.filter(Q(users=self) | Q(groups__in=self.all_groups())).distinct()
 414
 415    def get_managed_role(self, create=False):
 416        if create:
 417            name = managed_role_name(self)
 418            role, created = Role.objects.get_or_create(name=name, managed=name)
 419            if created:
 420                role.users.add(self)
 421            return role
 422        else:
 423            return Role.objects.filter(name=managed_role_name(self)).first()
 424
 425    def get_all_model_perms_on_managed_role(self) -> QuerySet[RoleModelPermission]:
 426        role = self.get_managed_role()
 427        if not role:
 428            return RoleModelPermission.objects.none()
 429        return RoleModelPermission.objects.filter(role=role)
 430
 431    def get_all_obj_perms_on_managed_role(self) -> QuerySet[RoleObjectPermission]:
 432        role = self.get_managed_role()
 433        if not role:
 434            return RoleObjectPermission.objects.none()
 435        return RoleObjectPermission.objects.filter(role=role)
 436
 437    def assign_perms_to_managed_role(
 438        self,
 439        perms: str | list[str] | Permission | list[Permission],
 440        obj: models.Model | None = None,
 441    ):
 442        if not perms:
 443            return
 444        role = self.get_managed_role(create=True)
 445        role.assign_perms(perms, obj)
 446
 447    def remove_perms_from_managed_role(
 448        self,
 449        perms: str | list[str] | Permission | list[Permission],
 450        obj: models.Model | None = None,
 451    ):
 452        role = self.get_managed_role()
 453        if not role:
 454            return None
 455        role.remove_perms(perms, obj)
 456
 457    def remove_all_perms_from_managed_role(self):
 458        role = self.get_managed_role()
 459        if not role:
 460            return None
 461        RoleModelPermission.objects.filter(role=role).delete()
 462        RoleObjectPermission.objects.filter(role=role).delete()
 463
 464    def group_attributes(self, request: HttpRequest | None = None) -> dict[str, Any]:
 465        """Get a dictionary containing the attributes from all groups the user belongs to,
 466        including the users attributes"""
 467        final_attributes = {}
 468        if request and hasattr(request, "brand"):
 469            always_merger.merge(final_attributes, request.brand.attributes)
 470        for group in self.all_groups().order_by("name"):
 471            always_merger.merge(final_attributes, group.attributes)
 472        always_merger.merge(final_attributes, self.attributes)
 473        return final_attributes
 474
 475    def app_entitlements(self, app: Application | None) -> QuerySet[ApplicationEntitlement]:
 476        """Get all entitlements this user has for `app`."""
 477        if not app:
 478            return []
 479        all_groups = self.all_groups()
 480        qs = app.applicationentitlement_set.filter(
 481            Q(
 482                Q(bindings__user=self) | Q(bindings__group__in=all_groups),
 483                bindings__negate=False,
 484            )
 485            | Q(
 486                Q(~Q(bindings__user=self), bindings__user__isnull=False)
 487                | Q(~Q(bindings__group__in=all_groups), bindings__group__isnull=False),
 488                bindings__negate=True,
 489            ),
 490            bindings__enabled=True,
 491        ).order_by("name")
 492        return qs
 493
 494    def app_entitlements_attributes(self, app: Application | None) -> dict:
 495        """Get a dictionary containing all merged attributes from app entitlements for `app`."""
 496        final_attributes = {}
 497        for attrs in self.app_entitlements(app).values_list("attributes", flat=True):
 498            always_merger.merge(final_attributes, attrs)
 499        return final_attributes
 500
 501    @property
 502    def serializer(self) -> Serializer:
 503        from authentik.core.api.users import UserSerializer
 504
 505        return UserSerializer
 506
 507    @cached_property
 508    def is_superuser(self) -> bool:
 509        """Get supseruser status based on membership in a group with superuser status"""
 510        return self.all_groups().filter(is_superuser=True).exists()
 511
 512    @property
 513    def is_staff(self) -> bool:
 514        """superuser == staff user"""
 515        return self.is_superuser  # type: ignore
 516
 517    # TODO: remove this after 2026.
 518    @property
 519    def ak_groups(self):
 520        """This is a proxy for a renamed, deprecated field."""
 521        from authentik.events.models import Event
 522
 523        deprecation = "authentik.core.models.User.ak_groups"
 524        replacement = "authentik.core.models.User.groups"
 525        message_logger = (
 526            f"{deprecation} is deprecated and will be removed in a future version of "
 527            f"authentik. Please use {replacement} instead."
 528        )
 529        message_event = (
 530            f"{message_logger} This event will not be repeated until it expires (by "
 531            "default: in 30 days). See authentik logs for every will invocation of this "
 532            "deprecation."
 533        )
 534        stacktrace = traceback.format_stack()
 535        # The last line is this function, the next-to-last line is its caller
 536        cause = stacktrace[-2] if len(stacktrace) > 1 else "Unknown, see stacktrace in logs"
 537        if search := re.search(r'"(.*?)"', cause):
 538            cause = f"Property mapping or Expression policy named {search.group(1)}"
 539
 540        LOGGER.warning(
 541            "deprecation used",
 542            message=message_logger,
 543            deprecation=deprecation,
 544            replacement=replacement,
 545            cause=cause,
 546            stacktrace=stacktrace,
 547        )
 548        Event.log_deprecation(
 549            deprecation, message=message_event, cause=cause, replacement=replacement
 550        )
 551        return self.groups
 552
 553    def set_password(self, raw_password, signal=True, sender=None, request=None):
 554        if self.pk and signal:
 555            from authentik.core.signals import password_changed
 556
 557            if not sender:
 558                sender = self
 559            password_changed.send(sender=sender, user=self, password=raw_password, request=request)
 560        self.password_change_date = now()
 561        return super().set_password(raw_password)
 562
 563    def check_password(self, raw_password: str) -> bool:
 564        """
 565        Return a boolean of whether the raw_password was correct. Handles
 566        hashing formats behind the scenes.
 567
 568        Slightly changed version which doesn't send a signal for such internal hash upgrades
 569        """
 570
 571        def setter(raw_password):
 572            self.set_password(raw_password, signal=False)
 573            # Password hash upgrades shouldn't be considered password changes.
 574            self._password = None
 575            self.save(update_fields=["password"])
 576
 577        return check_password(raw_password, self.password, setter)
 578
 579    @property
 580    def uid(self) -> str:
 581        """Generate a globally unique UID, based on the user ID and the hashed secret key"""
 582        return sha256(f"{self.id}-{get_unique_identifier()}".encode("ascii")).hexdigest()
 583
 584    def locale(self, request: HttpRequest | None = None) -> str:
 585        """Get the locale the user has configured"""
 586        if request and hasattr(request, "LANGUAGE_CODE"):
 587            return request.LANGUAGE_CODE
 588        try:
 589            return self.attributes.get("settings", {}).get("locale", "")
 590
 591        except Exception as exc:  # noqa
 592            LOGGER.warning("Failed to get default locale", exc=exc)
 593        if request:
 594            return request.brand.locale
 595        return ""
 596
 597    @property
 598    def avatar(self) -> str:
 599        """Get avatar, depending on authentik.avatar setting"""
 600        return get_avatar(self)
 601
 602
 603class Provider(SerializerModel):
 604    """Application-independent Provider instance. For example SAML2 Remote, OAuth2 Application"""
 605
 606    name = models.TextField(unique=True)
 607
 608    authentication_flow = models.ForeignKey(
 609        "authentik_flows.Flow",
 610        null=True,
 611        on_delete=models.SET_NULL,
 612        help_text=_(
 613            "Flow used for authentication when the associated application is accessed by an "
 614            "un-authenticated user."
 615        ),
 616        related_name="provider_authentication",
 617    )
 618    authorization_flow = models.ForeignKey(
 619        "authentik_flows.Flow",
 620        # Set to cascade even though null is allowed, since most providers
 621        # still require an authorization flow set
 622        on_delete=models.CASCADE,
 623        null=True,
 624        help_text=_("Flow used when authorizing this provider."),
 625        related_name="provider_authorization",
 626    )
 627    invalidation_flow = models.ForeignKey(
 628        "authentik_flows.Flow",
 629        on_delete=models.SET_DEFAULT,
 630        default=None,
 631        null=True,
 632        help_text=_("Flow used ending the session from a provider."),
 633        related_name="provider_invalidation",
 634    )
 635
 636    property_mappings = models.ManyToManyField("PropertyMapping", default=None, blank=True)
 637
 638    backchannel_application = models.ForeignKey(
 639        "Application",
 640        default=None,
 641        null=True,
 642        on_delete=models.CASCADE,
 643        help_text=_(
 644            "Accessed from applications; optional backchannel providers for protocols "
 645            "like LDAP and SCIM."
 646        ),
 647        related_name="backchannel_providers",
 648    )
 649
 650    is_backchannel = models.BooleanField(default=False)
 651
 652    objects = InheritanceManager()
 653
 654    @property
 655    def launch_url(self) -> str | None:
 656        """URL to this provider and initiate authorization for the user.
 657        Can return None for providers that are not URL-based"""
 658        return None
 659
 660    @property
 661    def icon_url(self) -> str | None:
 662        return None
 663
 664    @property
 665    def component(self) -> str:
 666        """Return component used to edit this object"""
 667        raise NotImplementedError
 668
 669    @property
 670    def serializer(self) -> type[Serializer]:
 671        """Get serializer for this model"""
 672        raise NotImplementedError
 673
 674    def __str__(self):
 675        return str(self.name)
 676
 677
 678class BackchannelProvider(Provider):
 679    """Base class for providers that augment other providers, for example LDAP and SCIM.
 680    Multiple of these providers can be configured per application, they may not use the application
 681    slug in URLs as an application may have multiple instances of the same
 682    type of Backchannel provider
 683
 684    They can use the application's policies and metadata"""
 685
 686    @property
 687    def component(self) -> str:
 688        raise NotImplementedError
 689
 690    @property
 691    def serializer(self) -> type[Serializer]:
 692        raise NotImplementedError
 693
 694    class Meta:
 695        abstract = True
 696
 697
 698class ApplicationQuerySet(QuerySet):
 699    def with_provider(self) -> QuerySet[Application]:
 700        qs = self.select_related("provider")
 701        for subclass in Provider.objects.get_queryset()._get_subclasses_recurse(Provider):
 702            qs = qs.select_related(f"provider__{subclass}")
 703            # Also prefetch/select through each subclass path to ensure casted instances have access
 704            qs = qs.prefetch_related(f"provider__{subclass}__property_mappings")
 705            qs = qs.select_related(f"provider__{subclass}__application")
 706            qs = qs.select_related(f"provider__{subclass}__backchannel_application")
 707        return qs
 708
 709
 710class Application(SerializerModel, PolicyBindingModel):
 711    """Every Application which uses authentik for authentication/identification/authorization
 712    needs an Application record. Other authentication types can subclass this Model to
 713    add custom fields and other properties"""
 714
 715    name = models.TextField(help_text=_("Application's display Name."))
 716    slug = models.TextField(
 717        validators=[validate_slug],
 718        help_text=_("Internal application name, used in URLs."),
 719        unique=True,
 720    )
 721    group = models.TextField(blank=True, default="")
 722
 723    provider = models.OneToOneField(
 724        "Provider", null=True, blank=True, default=None, on_delete=models.SET_DEFAULT
 725    )
 726
 727    meta_launch_url = models.TextField(
 728        default="", blank=True, validators=[DomainlessFormattedURLValidator()]
 729    )
 730
 731    open_in_new_tab = models.BooleanField(
 732        default=False, help_text=_("Open launch URL in a new browser tab or window.")
 733    )
 734
 735    meta_icon = FileField(default="", blank=True)
 736    meta_description = models.TextField(default="", blank=True)
 737    meta_publisher = models.TextField(default="", blank=True)
 738
 739    objects = ApplicationQuerySet.as_manager()
 740
 741    # Reserved slugs that would clash with OAuth2 provider endpoints
 742    reserved_slugs = ["authorize", "token", "device", "userinfo", "introspect", "revoke"]
 743
 744    @property
 745    def serializer(self) -> Serializer:
 746        from authentik.core.api.applications import ApplicationSerializer
 747
 748        return ApplicationSerializer
 749
 750    @property
 751    def get_meta_icon(self) -> str | None:
 752        """Get the URL to the App Icon image"""
 753        if not self.meta_icon:
 754            return None
 755
 756        return get_file_manager(FileUsage.MEDIA).file_url(self.meta_icon)
 757
 758    @property
 759    def get_meta_icon_themed_urls(self) -> dict[str, str] | None:
 760        """Get themed URLs for meta_icon if it contains %(theme)s"""
 761        if not self.meta_icon:
 762            return None
 763
 764        return get_file_manager(FileUsage.MEDIA).themed_urls(self.meta_icon)
 765
 766    def get_launch_url(self, user: User | None = None, user_data: dict | None = None) -> str | None:
 767        """Get launch URL if set, otherwise attempt to get launch URL based on provider.
 768
 769        Args:
 770            user: User instance for formatting the URL
 771            user_data: Pre-serialized user data to avoid re-serialization (performance optimization)
 772        """
 773        from authentik.core.api.users import UserSerializer
 774
 775        url = None
 776        if self.meta_launch_url:
 777            url = self.meta_launch_url
 778        elif provider := self.get_provider():
 779            url = provider.launch_url
 780        if user and url:
 781            try:
 782                # Use pre-serialized data if available, otherwise serialize now
 783                if user_data is None:
 784                    user_data = UserSerializer(instance=user).data
 785                return url % user_data
 786            except Exception as exc:  # noqa
 787                LOGGER.warning("Failed to format launch url", exc=exc)
 788                return url
 789        return url
 790
 791    def get_provider(self) -> Provider | None:
 792        """Get casted provider instance. Needs Application queryset with_provider"""
 793        if not self.provider:
 794            return None
 795        return get_deepest_child(self.provider)
 796
 797    def backchannel_provider_for[T: Provider](self, provider_type: type[T], **kwargs) -> T | None:
 798        """Get Backchannel provider for a specific type"""
 799        providers = self.backchannel_providers.filter(
 800            **{f"{provider_type._meta.model_name}__isnull": False},
 801            **kwargs,
 802        )
 803        return getattr(providers.first(), provider_type._meta.model_name)
 804
 805    def __str__(self):
 806        return str(self.name)
 807
 808    class Meta:
 809        verbose_name = _("Application")
 810        verbose_name_plural = _("Applications")
 811
 812
 813class ApplicationEntitlement(AttributesMixin, SerializerModel, PolicyBindingModel):
 814    """Application-scoped entitlement to control authorization in an application"""
 815
 816    name = models.TextField()
 817
 818    app = models.ForeignKey(Application, on_delete=models.CASCADE)
 819
 820    class Meta:
 821        verbose_name = _("Application Entitlement")
 822        verbose_name_plural = _("Application Entitlements")
 823        unique_together = (("app", "name"),)
 824
 825    def __str__(self):
 826        return f"Application Entitlement {self.name} for app {self.app_id}"
 827
 828    @property
 829    def serializer(self) -> type[Serializer]:
 830        from authentik.core.api.application_entitlements import ApplicationEntitlementSerializer
 831
 832        return ApplicationEntitlementSerializer
 833
 834    def supported_policy_binding_targets(self):
 835        return ["group", "user"]
 836
 837
 838class SourceUserMatchingModes(models.TextChoices):
 839    """Different modes a source can handle new/returning users"""
 840
 841    IDENTIFIER = "identifier", _("Use the source-specific identifier")
 842    EMAIL_LINK = (
 843        "email_link",
 844        _(
 845            "Link to a user with identical email address. Can have security implications "
 846            "when a source doesn't validate email addresses."
 847        ),
 848    )
 849    EMAIL_DENY = (
 850        "email_deny",
 851        _(
 852            "Use the user's email address, but deny enrollment when the email address already "
 853            "exists."
 854        ),
 855    )
 856    USERNAME_LINK = (
 857        "username_link",
 858        _(
 859            "Link to a user with identical username. Can have security implications "
 860            "when a username is used with another source."
 861        ),
 862    )
 863    USERNAME_DENY = (
 864        "username_deny",
 865        _("Use the user's username, but deny enrollment when the username already exists."),
 866    )
 867
 868
 869class SourceGroupMatchingModes(models.TextChoices):
 870    """Different modes a source can handle new/returning groups"""
 871
 872    IDENTIFIER = "identifier", _("Use the source-specific identifier")
 873    NAME_LINK = (
 874        "name_link",
 875        _(
 876            "Link to a group with identical name. Can have security implications "
 877            "when a group name is used with another source."
 878        ),
 879    )
 880    NAME_DENY = (
 881        "name_deny",
 882        _("Use the group name, but deny enrollment when the name already exists."),
 883    )
 884
 885
 886class Source(ManagedModel, SerializerModel, PolicyBindingModel):
 887    """Base Authentication source, i.e. an OAuth Provider, SAML Remote or LDAP Server"""
 888
 889    MANAGED_INBUILT = "goauthentik.io/sources/inbuilt"
 890
 891    name = models.TextField(help_text=_("Source's display Name."))
 892    slug = models.TextField(
 893        validators=[validate_slug],
 894        help_text=_("Internal source name, used in URLs."),
 895        unique=True,
 896    )
 897
 898    user_path_template = models.TextField(default="goauthentik.io/sources/%(slug)s")
 899
 900    enabled = models.BooleanField(default=True)
 901    promoted = models.BooleanField(
 902        default=False,
 903        help_text=_(
 904            "When enabled, this source will be displayed as a prominent button on the "
 905            "login page, instead of a small icon."
 906        ),
 907    )
 908    user_property_mappings = models.ManyToManyField(
 909        "PropertyMapping", default=None, blank=True, related_name="source_userpropertymappings_set"
 910    )
 911    group_property_mappings = models.ManyToManyField(
 912        "PropertyMapping", default=None, blank=True, related_name="source_grouppropertymappings_set"
 913    )
 914
 915    icon = FileField(blank=True, default="")
 916
 917    authentication_flow = models.ForeignKey(
 918        "authentik_flows.Flow",
 919        blank=True,
 920        null=True,
 921        default=None,
 922        on_delete=models.SET_NULL,
 923        help_text=_("Flow to use when authenticating existing users."),
 924        related_name="source_authentication",
 925    )
 926    enrollment_flow = models.ForeignKey(
 927        "authentik_flows.Flow",
 928        blank=True,
 929        null=True,
 930        default=None,
 931        on_delete=models.SET_NULL,
 932        help_text=_("Flow to use when enrolling new users."),
 933        related_name="source_enrollment",
 934    )
 935
 936    user_matching_mode = models.TextField(
 937        choices=SourceUserMatchingModes.choices,
 938        default=SourceUserMatchingModes.IDENTIFIER,
 939        help_text=_(
 940            "How the source determines if an existing user should be authenticated or "
 941            "a new user enrolled."
 942        ),
 943    )
 944    group_matching_mode = models.TextField(
 945        choices=SourceGroupMatchingModes.choices,
 946        default=SourceGroupMatchingModes.IDENTIFIER,
 947        help_text=_(
 948            "How the source determines if an existing group should be used or a new group created."
 949        ),
 950    )
 951
 952    objects = InheritanceManager()
 953
 954    @property
 955    def icon_url(self) -> str | None:
 956        """Get the URL to the source icon"""
 957        if not self.icon:
 958            return None
 959
 960        return get_file_manager(FileUsage.MEDIA).file_url(self.icon)
 961
 962    @property
 963    def icon_themed_urls(self) -> dict[str, str] | None:
 964        """Get themed URLs for icon if it contains %(theme)s"""
 965        if not self.icon:
 966            return None
 967
 968        return get_file_manager(FileUsage.MEDIA).themed_urls(self.icon)
 969
 970    def get_user_path(self) -> str:
 971        """Get user path, fallback to default for formatting errors"""
 972        try:
 973            return self.user_path_template % {
 974                "slug": self.slug,
 975            }
 976
 977        except Exception as exc:  # noqa
 978            LOGGER.warning("Failed to template user path", exc=exc, source=self)
 979            return User.default_path()
 980
 981    @property
 982    def component(self) -> str:
 983        """Return component used to edit this object"""
 984        if self.managed == self.MANAGED_INBUILT:
 985            return ""
 986        raise NotImplementedError
 987
 988    @property
 989    def property_mapping_type(self) -> type[PropertyMapping]:
 990        """Return property mapping type used by this object"""
 991        if self.managed == self.MANAGED_INBUILT:
 992            from authentik.core.models import PropertyMapping
 993
 994            return PropertyMapping
 995        raise NotImplementedError
 996
 997    def ui_login_button(self, request: HttpRequest) -> UILoginButton | None:
 998        """If source uses a http-based flow, return UI Information about the login
 999        button. If source doesn't use http-based flow, return None."""
1000        return None
1001
1002    def ui_user_settings(self) -> UserSettingSerializer | None:
1003        """Entrypoint to integrate with User settings. Can either return None if no
1004        user settings are available, or UserSettingSerializer."""
1005        return None
1006
1007    def get_base_user_properties(self, **kwargs) -> dict[str, Any | dict[str, Any]]:
1008        """Get base properties for a user to build final properties upon."""
1009        if self.managed == self.MANAGED_INBUILT:
1010            return {}
1011        raise NotImplementedError
1012
1013    def get_base_group_properties(self, **kwargs) -> dict[str, Any | dict[str, Any]]:
1014        """Get base properties for a group to build final properties upon."""
1015        if self.managed == self.MANAGED_INBUILT:
1016            return {}
1017        raise NotImplementedError
1018
1019    def __str__(self):
1020        return str(self.name)
1021
1022    class Meta:
1023        indexes = [
1024            models.Index(
1025                fields=[
1026                    "slug",
1027                ]
1028            ),
1029            models.Index(
1030                fields=[
1031                    "name",
1032                ]
1033            ),
1034            models.Index(
1035                fields=[
1036                    "enabled",
1037                ]
1038            ),
1039        ]
1040
1041
1042class UserSourceConnection(SerializerModel, CreatedUpdatedModel):
1043    """Connection between User and Source."""
1044
1045    user = models.ForeignKey(User, on_delete=models.CASCADE)
1046    source = models.ForeignKey(Source, on_delete=models.CASCADE)
1047    identifier = models.TextField()
1048
1049    objects = InheritanceManager()
1050
1051    @property
1052    def serializer(self) -> type[Serializer]:
1053        """Get serializer for this model"""
1054        raise NotImplementedError
1055
1056    def __str__(self) -> str:
1057        return f"User-source connection (user={self.user_id}, source={self.source_id})"
1058
1059    class Meta:
1060        unique_together = (("user", "source"),)
1061        indexes = (
1062            models.Index(fields=("identifier",)),
1063            models.Index(fields=("source", "identifier")),
1064        )
1065
1066
1067class GroupSourceConnection(SerializerModel, CreatedUpdatedModel):
1068    """Connection between Group and Source."""
1069
1070    group = models.ForeignKey(Group, on_delete=models.CASCADE)
1071    source = models.ForeignKey(Source, on_delete=models.CASCADE)
1072    identifier = models.TextField()
1073
1074    objects = InheritanceManager()
1075
1076    @property
1077    def serializer(self) -> type[Serializer]:
1078        """Get serializer for this model"""
1079        raise NotImplementedError
1080
1081    def __str__(self) -> str:
1082        return f"Group-source connection (group={self.group_id}, source={self.source_id})"
1083
1084    class Meta:
1085        unique_together = (("group", "source"),)
1086
1087
1088class ExpiringManager(Manager):
1089    """Manager for expiring objects which filters out expired objects by default"""
1090
1091    def get_queryset(self):
1092        return QuerySet(self.model, using=self._db).exclude(expires__lt=now(), expiring=True)
1093
1094    def including_expired(self):
1095        return QuerySet(self.model, using=self._db)
1096
1097
1098class ExpiringModel(models.Model):
1099    """Base Model which can expire, and is automatically cleaned up."""
1100
1101    expires = models.DateTimeField(default=None, null=True)
1102    expiring = models.BooleanField(default=True)
1103
1104    objects = ExpiringManager()
1105
1106    class Meta:
1107        abstract = True
1108        indexes = [
1109            models.Index(fields=["expires"]),
1110            models.Index(fields=["expiring"]),
1111            models.Index(fields=["expiring", "expires"]),
1112        ]
1113
1114    def expire_action(self, *args, **kwargs):
1115        """Handler which is called when this object is expired. By
1116        default the object is deleted. This is less efficient compared
1117        to bulk deleting objects, but classes like Token() need to change
1118        values instead of being deleted."""
1119        try:
1120            return self.delete(*args, **kwargs)
1121        except self.DoesNotExist:
1122            # Object has already been deleted, so this should be fine
1123            return None
1124
1125    @classmethod
1126    def filter_not_expired(cls, **kwargs) -> QuerySet[Self]:
1127        """Filer for tokens which are not expired yet or are not expiring,
1128        and match filters in `kwargs`"""
1129        from authentik.events.models import Event
1130
1131        deprecation_id = f"{class_to_path(cls)}.filter_not_expired"
1132
1133        Event.log_deprecation(
1134            deprecation_id,
1135            message=(
1136                ".filter_not_expired() is deprecated as the default lookup now excludes "
1137                "expired objects."
1138            ),
1139        )
1140
1141        for obj in (
1142            cls.objects.including_expired()
1143            .filter(**kwargs)
1144            .filter(Q(expires__lt=now(), expiring=True))
1145        ):
1146            obj.delete()
1147        return cls.objects.filter(**kwargs)
1148
1149    @property
1150    def is_expired(self) -> bool:
1151        """Check if token is expired yet."""
1152        if not self.expiring:
1153            return False
1154        return now() > self.expires
1155
1156
1157class TokenIntents(models.TextChoices):
1158    """Intents a Token can be created for."""
1159
1160    # Single use token
1161    INTENT_VERIFICATION = "verification"
1162
1163    # Allow access to API
1164    INTENT_API = "api"
1165
1166    # Recovery use for the recovery app
1167    INTENT_RECOVERY = "recovery"
1168
1169    # App-specific passwords
1170    INTENT_APP_PASSWORD = "app_password"  # nosec
1171
1172
1173class Token(SerializerModel, ManagedModel, ExpiringModel):
1174    """Token used to authenticate the User for API Access or confirm another Stage like Email."""
1175
1176    token_uuid = models.UUIDField(primary_key=True, editable=False, default=uuid4)
1177    identifier = models.SlugField(max_length=255, unique=True)
1178    key = models.TextField(default=default_token_key)
1179    intent = models.TextField(
1180        choices=TokenIntents.choices, default=TokenIntents.INTENT_VERIFICATION
1181    )
1182    user = models.ForeignKey("User", on_delete=models.CASCADE, related_name="+")
1183    description = models.TextField(default="", blank=True)
1184
1185    class Meta:
1186        verbose_name = _("Token")
1187        verbose_name_plural = _("Tokens")
1188        indexes = ExpiringModel.Meta.indexes + [
1189            models.Index(fields=["identifier"]),
1190            models.Index(fields=["key"]),
1191        ]
1192        permissions = [
1193            ("view_token_key", _("View token's key")),
1194            ("set_token_key", _("Set a token's key")),
1195        ]
1196
1197    def __str__(self):
1198        description = f"{self.identifier}"
1199        if self.expiring:
1200            description += f" (expires={self.expires})"
1201        return description
1202
1203    @property
1204    def serializer(self) -> type[Serializer]:
1205        from authentik.core.api.tokens import TokenSerializer
1206
1207        return TokenSerializer
1208
1209    def expire_action(self, *args, **kwargs):
1210        """Handler which is called when this object is expired."""
1211        from authentik.events.models import Event, EventAction
1212
1213        if self.intent in [
1214            TokenIntents.INTENT_RECOVERY,
1215            TokenIntents.INTENT_VERIFICATION,
1216            TokenIntents.INTENT_APP_PASSWORD,
1217        ]:
1218            super().expire_action(*args, **kwargs)
1219            return
1220
1221        self.key = default_token_key()
1222        self.expires = default_token_duration()
1223        self.save(*args, **kwargs)
1224        Event.new(
1225            action=EventAction.SECRET_ROTATE,
1226            token=self,
1227            message=f"Token {self.identifier}'s secret was rotated.",
1228        ).save()
1229
1230
1231class PropertyMapping(SerializerModel, ManagedModel):
1232    """User-defined key -> x mapping which can be used by providers to expose extra data."""
1233
1234    pm_uuid = models.UUIDField(primary_key=True, editable=False, default=uuid4)
1235    name = models.TextField(unique=True)
1236    expression = models.TextField()
1237
1238    objects = InheritanceManager()
1239
1240    @property
1241    def component(self) -> str:
1242        """Return component used to edit this object"""
1243        raise NotImplementedError
1244
1245    @property
1246    def serializer(self) -> type[Serializer]:
1247        """Get serializer for this model"""
1248        raise NotImplementedError
1249
1250    def evaluate(self, user: User | None, request: HttpRequest | None, **kwargs) -> Any:
1251        """Evaluate `self.expression` using `**kwargs` as Context."""
1252        from authentik.core.expression.evaluator import PropertyMappingEvaluator
1253
1254        evaluator = PropertyMappingEvaluator(self, user, request, **kwargs)
1255        try:
1256            return evaluator.evaluate(self.expression)
1257        except ControlFlowException as exc:
1258            raise exc
1259        except Exception as exc:
1260            raise PropertyMappingExpressionException(exc, self) from exc
1261
1262    def __str__(self):
1263        return f"Property Mapping {self.name}"
1264
1265    class Meta:
1266        verbose_name = _("Property Mapping")
1267        verbose_name_plural = _("Property Mappings")
1268
1269
1270class Session(ExpiringModel, AbstractBaseSession):
1271    """User session with extra fields for fast access"""
1272
1273    # Remove upstream field because we're using our own ExpiringModel
1274    expire_date = None
1275    session_data = models.BinaryField(_("session data"))
1276
1277    # Keep in sync with Session.Keys
1278    last_ip = models.GenericIPAddressField()
1279    last_user_agent = models.TextField(blank=True)
1280    last_used = models.DateTimeField(auto_now=True)
1281
1282    class Meta:
1283        verbose_name = _("Session")
1284        verbose_name_plural = _("Sessions")
1285        indexes = ExpiringModel.Meta.indexes + [
1286            models.Index(fields=["expires", "session_key"]),
1287        ]
1288        default_permissions = []
1289
1290    def __str__(self):
1291        return self.session_key
1292
1293    class Keys(StrEnum):
1294        """
1295        Keys to be set with the session interface for the fields above to be updated.
1296
1297        If a field is added here that needs to be initialized when the session is initialized,
1298        it must also be reflected in authentik.root.middleware.SessionMiddleware.process_request
1299        and in authentik.core.sessions.SessionStore.__init__
1300        """
1301
1302        LAST_IP = "last_ip"
1303        LAST_USER_AGENT = "last_user_agent"
1304        LAST_USED = "last_used"
1305
1306    @classmethod
1307    def get_session_store_class(cls):
1308        from authentik.core.sessions import SessionStore
1309
1310        return SessionStore
1311
1312    def get_decoded(self):
1313        raise NotImplementedError
1314
1315
1316class AuthenticatedSession(SerializerModel):
1317    session = models.OneToOneField(Session, on_delete=models.CASCADE, primary_key=True)
1318    # We use the session as primary key, but we need the API to be able to reference
1319    # this object uniquely without exposing the session key
1320    uuid = models.UUIDField(default=uuid4, unique=True)
1321
1322    user = models.ForeignKey(User, on_delete=models.CASCADE)
1323
1324    @property
1325    def serializer(self) -> type[Serializer]:
1326        from authentik.core.api.authenticated_sessions import AuthenticatedSessionSerializer
1327
1328        return AuthenticatedSessionSerializer
1329
1330    class Meta:
1331        verbose_name = _("Authenticated Session")
1332        verbose_name_plural = _("Authenticated Sessions")
1333
1334    def __str__(self) -> str:
1335        return f"Authenticated Session {str(self.pk)[:10]}"
1336
1337    @staticmethod
1338    def from_request(request: HttpRequest, user: User) -> AuthenticatedSession | None:
1339        """Create a new session from a http request"""
1340        if not hasattr(request, "session") or not request.session.exists(
1341            request.session.session_key
1342        ):
1343            return None
1344        return AuthenticatedSession(
1345            session=Session.objects.filter(session_key=request.session.session_key).first(),
1346            user=user,
1347        )