authentik.providers.scim.models

  1"""SCIM Provider models"""
  2
  3from typing import Any, Self
  4from uuid import uuid4
  5
  6from django.db import models
  7from django.db.models import QuerySet
  8from django.templatetags.static import static
  9from django.utils.translation import gettext_lazy as _
 10from dramatiq.actor import Actor
 11from requests.auth import AuthBase
 12from rest_framework.serializers import Serializer
 13from structlog.stdlib import get_logger
 14
 15from authentik.core.apps import AppAccessWithoutBindings
 16from authentik.core.models import BackchannelProvider, Group, PropertyMapping, User, UserTypes
 17from authentik.lib.models import InternallyManagedMixin, SerializerModel
 18from authentik.lib.sync.outgoing.base import BaseOutgoingSyncClient
 19from authentik.lib.sync.outgoing.models import OutgoingSyncProvider
 20from authentik.lib.utils.time import timedelta_from_string, timedelta_string_validator
 21from authentik.policies.engine import PolicyEngine
 22from authentik.providers.scim.clients.auth import SCIMTokenAuth
 23
 24LOGGER = get_logger()
 25
 26
 27class SCIMProviderUser(InternallyManagedMixin, SerializerModel):
 28    """Mapping of a user and provider to a SCIM user ID"""
 29
 30    id = models.UUIDField(primary_key=True, editable=False, default=uuid4)
 31    scim_id = models.TextField()
 32    user = models.ForeignKey(User, on_delete=models.CASCADE)
 33    provider = models.ForeignKey("SCIMProvider", on_delete=models.CASCADE)
 34    attributes = models.JSONField(default=dict)
 35
 36    @property
 37    def serializer(self) -> type[Serializer]:
 38        from authentik.providers.scim.api.users import SCIMProviderUserSerializer
 39
 40        return SCIMProviderUserSerializer
 41
 42    class Meta:
 43        unique_together = (("scim_id", "user", "provider"),)
 44
 45    def __str__(self) -> str:
 46        return f"SCIM Provider User {self.user_id} to {self.provider_id}"
 47
 48
 49class SCIMProviderGroup(InternallyManagedMixin, SerializerModel):
 50    """Mapping of a group and provider to a SCIM user ID"""
 51
 52    id = models.UUIDField(primary_key=True, editable=False, default=uuid4)
 53    scim_id = models.TextField()
 54    group = models.ForeignKey(Group, on_delete=models.CASCADE)
 55    provider = models.ForeignKey("SCIMProvider", on_delete=models.CASCADE)
 56    attributes = models.JSONField(default=dict)
 57
 58    @property
 59    def serializer(self) -> type[Serializer]:
 60        from authentik.providers.scim.api.groups import SCIMProviderGroupSerializer
 61
 62        return SCIMProviderGroupSerializer
 63
 64    class Meta:
 65        unique_together = (("scim_id", "group", "provider"),)
 66
 67    def __str__(self) -> str:
 68        return f"SCIM Provider Group {self.group_id} to {self.provider_id}"
 69
 70
 71class SCIMAuthenticationMode(models.TextChoices):
 72    """SCIM authentication modes"""
 73
 74    TOKEN = "token", _("Token")
 75    OAUTH_SILENT = "oauth", _("OAuth (Silent)")
 76    OAUTH_INTERACTIVE = "oauth_interactive", _("OAuth (interactive)")
 77
 78
 79class SCIMCompatibilityMode(models.TextChoices):
 80    """SCIM compatibility mode"""
 81
 82    DEFAULT = "default", _("Default")
 83    AWS = "aws", _("AWS")
 84    SLACK = "slack", _("Slack")
 85    SALESFORCE = "sfdc", _("Salesforce")
 86    WEBEX = "webex", _("Webex")
 87    VCENTER = "vcenter", _("vCenter")
 88
 89
 90class SCIMProvider(OutgoingSyncProvider, BackchannelProvider):
 91    """SCIM 2.0 provider to create users and groups in external applications"""
 92
 93    exclude_users_service_account = models.BooleanField(default=False)
 94
 95    group_filters = models.ManyToManyField(
 96        "authentik_core.group",
 97        default=None,
 98        blank=True,
 99        help_text=_("Group filters used to define sync-scope for groups."),
100    )
101
102    url = models.TextField(help_text=_("Base URL to SCIM requests, usually ends in /v2"))
103
104    auth_mode = models.TextField(
105        choices=SCIMAuthenticationMode.choices, default=SCIMAuthenticationMode.TOKEN
106    )
107
108    token = models.TextField(help_text=_("Authentication token"), blank=True)
109    auth_oauth = models.ForeignKey(
110        "authentik_sources_oauth.OAuthSource",
111        on_delete=models.SET_DEFAULT,
112        default=None,
113        null=True,
114        help_text=_("OAuth Source used for authentication"),
115    )
116    auth_oauth_params = models.JSONField(
117        blank=True, default=dict, help_text=_("Additional OAuth parameters, such as grant_type")
118    )
119    auth_oauth_user = models.ForeignKey(
120        "authentik_core.User", on_delete=models.SET_NULL, default=None, null=True
121    )
122
123    verify_certificates = models.BooleanField(default=True)
124
125    property_mappings_group = models.ManyToManyField(
126        PropertyMapping,
127        default=None,
128        blank=True,
129        help_text=_("Property mappings used for group creation/updating."),
130    )
131
132    compatibility_mode = models.CharField(
133        max_length=30,
134        choices=SCIMCompatibilityMode.choices,
135        default=SCIMCompatibilityMode.DEFAULT,
136        verbose_name=_("SCIM Compatibility Mode"),
137        help_text=_("Alter authentik behavior for vendor-specific SCIM implementations."),
138    )
139    service_provider_config_cache_timeout = models.TextField(
140        default="hours=1",
141        validators=[timedelta_string_validator],
142        help_text=_(
143            "Cache duration for ServiceProviderConfig responses. Set minutes=0 to disable."
144        ),
145    )
146
147    def scim_auth(self) -> AuthBase:
148        if self.auth_mode in [
149            SCIMAuthenticationMode.OAUTH_SILENT,
150            SCIMAuthenticationMode.OAUTH_INTERACTIVE,
151        ]:
152            try:
153                from authentik.enterprise.providers.scim.auth_oauth2 import SCIMOAuthAuth
154
155                return SCIMOAuthAuth(self)
156            except ImportError:
157                LOGGER.warning("Failed to import SCIM OAuth Client")
158        return SCIMTokenAuth(self)
159
160    @property
161    def icon_url(self) -> str | None:
162        return static("authentik/sources/scim.png")
163
164    @property
165    def sync_actor(self) -> Actor:
166        from authentik.providers.scim.tasks import scim_sync
167
168        return scim_sync
169
170    def client_for_model(
171        self, model: type[User | Group | SCIMProviderUser | SCIMProviderGroup]
172    ) -> BaseOutgoingSyncClient[User | Group, Any, Any, Self]:
173        if issubclass(model, User | SCIMProviderUser):
174            from authentik.providers.scim.clients.users import SCIMUserClient
175
176            return SCIMUserClient(self)
177        if issubclass(model, Group | SCIMProviderGroup):
178            from authentik.providers.scim.clients.groups import SCIMGroupClient
179
180            return SCIMGroupClient(self)
181        raise ValueError(f"Invalid model {model}")
182
183    def save(self, *args, **kwargs):
184        from django.core.cache import cache
185
186        cache_key = f"goauthentik.io/providers/scim/{self.pk}/service_provider_config"
187        cache.delete(cache_key)
188        super().save(*args, **kwargs)
189
190    def get_object_qs(self, type: type[User | Group], **kwargs) -> QuerySet[User | Group]:
191        if type == User:
192            # Get queryset of all users with consistent ordering
193            # according to the provider's settings
194            base = User.objects.all().exclude_anonymous().filter(**kwargs)
195            if self.exclude_users_service_account:
196                base = base.exclude(type=UserTypes.SERVICE_ACCOUNT).exclude(
197                    type=UserTypes.INTERNAL_SERVICE_ACCOUNT
198                )
199
200            # Filter users by their access to the backchannel application if an application is set
201            # This handles both policy bindings and group_filters
202            if self.backchannel_application:
203                pks = []
204                for user in base:
205                    engine = PolicyEngine(self.backchannel_application, user, None)
206                    engine.empty_result = AppAccessWithoutBindings.get()
207                    engine.build()
208                    if engine.passing:
209                        pks.append(user.pk)
210                base = base.filter(pk__in=pks)
211            return base.order_by("pk")
212
213        if type == Group:
214            # Get queryset of all groups with consistent ordering
215            # according to the provider's settings
216            base = Group.objects.prefetch_related("scimprovidergroup_set").all().filter(**kwargs)
217
218            # Filter groups by group_filters if set
219            if self.group_filters.exists():
220                base = base.filter(pk__in=self.group_filters.values_list("pk", flat=True))
221
222            return base.order_by("pk")
223        raise ValueError(f"Invalid type {type}")
224
225    @classmethod
226    def get_object_mappings(cls, obj: User | Group) -> list[tuple[str, str]]:
227        if isinstance(obj, User):
228            return list(obj.scimprovideruser_set.values_list("provider__pk", "scim_id"))
229        if isinstance(obj, Group):
230            return list(obj.scimprovidergroup_set.values_list("provider__pk", "scim_id"))
231        raise ValueError(f"Invalid type {type(obj)}")
232
233    @property
234    def component(self) -> str:
235        return "ak-provider-scim-form"
236
237    @property
238    def service_provider_config_cache_timeout_seconds(self) -> int:
239        return max(
240            0,
241            int(timedelta_from_string(self.service_provider_config_cache_timeout).total_seconds()),
242        )
243
244    @property
245    def serializer(self) -> type[Serializer]:
246        from authentik.providers.scim.api.providers import SCIMProviderSerializer
247
248        return SCIMProviderSerializer
249
250    def __str__(self):
251        return f"SCIM Provider {self.name}"
252
253    class Meta:
254        verbose_name = _("SCIM Provider")
255        verbose_name_plural = _("SCIM Providers")
256
257
258class SCIMMapping(PropertyMapping):
259    """Map authentik data to outgoing SCIM requests"""
260
261    @property
262    def component(self) -> str:
263        return "ak-property-mapping-provider-scim-form"
264
265    @property
266    def serializer(self) -> type[Serializer]:
267        from authentik.providers.scim.api.property_mappings import SCIMMappingSerializer
268
269        return SCIMMappingSerializer
270
271    def __str__(self):
272        return f"SCIM Provider Mapping {self.name}"
273
274    class Meta:
275        verbose_name = _("SCIM Provider Mapping")
276        verbose_name_plural = _("SCIM Provider Mappings")