authentik.sources.scim.models

  1"""SCIM Source"""
  2
  3from typing import Any
  4from uuid import uuid4
  5
  6from django.db import models
  7from django.templatetags.static import static
  8from django.utils.translation import gettext_lazy as _
  9from rest_framework.serializers import BaseSerializer, Serializer
 10
 11from authentik.core.models import Group, PropertyMapping, Source, Token, User
 12from authentik.lib.models import InternallyManagedMixin, SerializerModel
 13
 14
 15class SCIMSource(Source):
 16    """System for Cross-domain Identity Management Source, allows for
 17    cross-system user provisioning"""
 18
 19    token = models.ForeignKey(Token, on_delete=models.CASCADE, null=True, default=None)
 20
 21    @property
 22    def service_account_identifier(self) -> str:
 23        return f"ak-source-scim-{self.pk}"
 24
 25    @property
 26    def component(self) -> str:
 27        """Return component used to edit this object"""
 28        return "ak-source-scim-form"
 29
 30    @property
 31    def icon_url(self) -> str:
 32        return static("authentik/sources/scim.png")
 33
 34    @property
 35    def serializer(self) -> BaseSerializer:
 36        from authentik.sources.scim.api.sources import SCIMSourceSerializer
 37
 38        return SCIMSourceSerializer
 39
 40    @property
 41    def property_mapping_type(self) -> type[PropertyMapping]:
 42        return SCIMSourcePropertyMapping
 43
 44    def get_base_user_properties(self, data: dict[str, Any]) -> dict[str, Any | dict[str, Any]]:
 45        properties = {}
 46
 47        def get_email(data: list[dict]) -> str:
 48            """Wrapper to get primary email or first email"""
 49            for email in data:
 50                if email.get("primary", False):
 51                    return email.get("value")
 52            if len(data) < 1:
 53                return ""
 54            return data[0].get("value")
 55
 56        if "userName" in data:
 57            properties["username"] = data.get("userName")
 58        if "name" in data:
 59            properties["name"] = data.get("name", {}).get("formatted", data.get("displayName"))
 60        if "emails" in data:
 61            properties["email"] = get_email(data.get("emails"))
 62        if "active" in data:
 63            properties["is_active"] = data.get("active")
 64
 65        return properties
 66
 67    def get_base_group_properties(self, data: dict[str, Any]) -> dict[str, Any | dict[str, Any]]:
 68        properties = {}
 69
 70        if "displayName" in data:
 71            properties["name"] = data.get("displayName")
 72
 73        return properties
 74
 75    def __str__(self) -> str:
 76        return f"SCIM Source {self.name}"
 77
 78    class Meta:
 79
 80        verbose_name = _("SCIM Source")
 81        verbose_name_plural = _("SCIM Sources")
 82
 83
 84class SCIMSourcePropertyMapping(PropertyMapping):
 85    """Map SCIM properties to User or Group object attributes"""
 86
 87    @property
 88    def component(self) -> str:
 89        return "ak-property-mapping-source-scim-form"
 90
 91    @property
 92    def serializer(self) -> type[Serializer]:
 93        from authentik.sources.scim.api.property_mappings import (
 94            SCIMSourcePropertyMappingSerializer,
 95        )
 96
 97        return SCIMSourcePropertyMappingSerializer
 98
 99    class Meta:
100        verbose_name = _("SCIM Source Property Mapping")
101        verbose_name_plural = _("SCIM Source Property Mappings")
102
103
104class SCIMSourceUser(InternallyManagedMixin, SerializerModel):
105    """Mapping of a user and source to a SCIM user ID"""
106
107    id = models.TextField(primary_key=True, default=uuid4)
108    external_id = models.TextField()
109    user = models.ForeignKey(User, on_delete=models.CASCADE)
110    source = models.ForeignKey(SCIMSource, on_delete=models.CASCADE)
111    attributes = models.JSONField(default=dict)
112    last_update = models.DateTimeField(auto_now=True)
113
114    @property
115    def serializer(self) -> BaseSerializer:
116        from authentik.sources.scim.api.users import SCIMSourceUserSerializer
117
118        return SCIMSourceUserSerializer
119
120    class Meta:
121        unique_together = (("external_id", "source"),)
122        indexes = [
123            models.Index(fields=["external_id"]),
124        ]
125
126    def __str__(self) -> str:
127        return f"SCIM User {self.user_id} to {self.source_id}"
128
129
130class SCIMSourceGroup(InternallyManagedMixin, SerializerModel):
131    """Mapping of a group and source to a SCIM user ID"""
132
133    id = models.TextField(primary_key=True, default=uuid4)
134    external_id = models.TextField()
135    group = models.ForeignKey(Group, on_delete=models.CASCADE)
136    source = models.ForeignKey(SCIMSource, on_delete=models.CASCADE)
137    attributes = models.JSONField(default=dict)
138    last_update = models.DateTimeField(auto_now=True)
139
140    @property
141    def serializer(self) -> BaseSerializer:
142        from authentik.sources.scim.api.groups import SCIMSourceGroupSerializer
143
144        return SCIMSourceGroupSerializer
145
146    class Meta:
147        unique_together = (("external_id", "source"),)
148        indexes = [
149            models.Index(fields=["external_id"]),
150        ]
151
152    def __str__(self) -> str:
153        return f"SCIM Group {self.group_id} to {self.source_id}"