authentik.providers.scim.models

SCIM Provider models

View Source

  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 = "oauth", _("OAuth")
 76
 77
 78class SCIMCompatibilityMode(models.TextChoices):
 79    """SCIM compatibility mode"""
 80
 81    DEFAULT = "default", _("Default")
 82    AWS = "aws", _("AWS")
 83    SLACK = "slack", _("Slack")
 84    SALESFORCE = "sfdc", _("Salesforce")
 85    WEBEX = "webex", _("Webex")
 86
 87
 88class SCIMProvider(OutgoingSyncProvider, BackchannelProvider):
 89    """SCIM 2.0 provider to create users and groups in external applications"""
 90
 91    exclude_users_service_account = models.BooleanField(default=False)
 92
 93    group_filters = models.ManyToManyField(
 94        "authentik_core.group",
 95        default=None,
 96        blank=True,
 97        help_text=_("Group filters used to define sync-scope for groups."),
 98    )
 99
100    url = models.TextField(help_text=_("Base URL to SCIM requests, usually ends in /v2"))
101
102    auth_mode = models.TextField(
103        choices=SCIMAuthenticationMode.choices, default=SCIMAuthenticationMode.TOKEN
104    )
105
106    token = models.TextField(help_text=_("Authentication token"), blank=True)
107    auth_oauth = models.ForeignKey(
108        "authentik_sources_oauth.OAuthSource",
109        on_delete=models.SET_DEFAULT,
110        default=None,
111        null=True,
112        help_text=_("OAuth Source used for authentication"),
113    )
114    auth_oauth_params = models.JSONField(
115        blank=True, default=dict, help_text=_("Additional OAuth parameters, such as grant_type")
116    )
117    auth_oauth_user = models.ForeignKey(
118        "authentik_core.User", on_delete=models.SET_NULL, default=None, null=True
119    )
120
121    verify_certificates = models.BooleanField(default=True)
122
123    property_mappings_group = models.ManyToManyField(
124        PropertyMapping,
125        default=None,
126        blank=True,
127        help_text=_("Property mappings used for group creation/updating."),
128    )
129
130    compatibility_mode = models.CharField(
131        max_length=30,
132        choices=SCIMCompatibilityMode.choices,
133        default=SCIMCompatibilityMode.DEFAULT,
134        verbose_name=_("SCIM Compatibility Mode"),
135        help_text=_("Alter authentik behavior for vendor-specific SCIM implementations."),
136    )
137    service_provider_config_cache_timeout = models.TextField(
138        default="hours=1",
139        validators=[timedelta_string_validator],
140        help_text=_(
141            "Cache duration for ServiceProviderConfig responses. Set minutes=0 to disable."
142        ),
143    )
144
145    def scim_auth(self) -> AuthBase:
146        if self.auth_mode == SCIMAuthenticationMode.OAUTH:
147            try:
148                from authentik.enterprise.providers.scim.auth_oauth2 import SCIMOAuthAuth
149
150                return SCIMOAuthAuth(self)
151            except ImportError:
152                LOGGER.warning("Failed to import SCIM OAuth Client")
153        return SCIMTokenAuth(self)
154
155    @property
156    def icon_url(self) -> str | None:
157        return static("authentik/sources/scim.png")
158
159    @property
160    def sync_actor(self) -> Actor:
161        from authentik.providers.scim.tasks import scim_sync
162
163        return scim_sync
164
165    def client_for_model(
166        self, model: type[User | Group | SCIMProviderUser | SCIMProviderGroup]
167    ) -> BaseOutgoingSyncClient[User | Group, Any, Any, Self]:
168        if issubclass(model, User | SCIMProviderUser):
169            from authentik.providers.scim.clients.users import SCIMUserClient
170
171            return SCIMUserClient(self)
172        if issubclass(model, Group | SCIMProviderGroup):
173            from authentik.providers.scim.clients.groups import SCIMGroupClient
174
175            return SCIMGroupClient(self)
176        raise ValueError(f"Invalid model {model}")
177
178    def save(self, *args, **kwargs):
179        from django.core.cache import cache
180
181        cache_key = f"goauthentik.io/providers/scim/{self.pk}/service_provider_config"
182        cache.delete(cache_key)
183        super().save(*args, **kwargs)
184
185    def get_object_qs(self, type: type[User | Group], **kwargs) -> QuerySet[User | Group]:
186        if type == User:
187            # Get queryset of all users with consistent ordering
188            # according to the provider's settings
189            base = User.objects.all().exclude_anonymous().filter(**kwargs)
190            if self.exclude_users_service_account:
191                base = base.exclude(type=UserTypes.SERVICE_ACCOUNT).exclude(
192                    type=UserTypes.INTERNAL_SERVICE_ACCOUNT
193                )
194
195            # Filter users by their access to the backchannel application if an application is set
196            # This handles both policy bindings and group_filters
197            if self.backchannel_application:
198                pks = []
199                for user in base:
200                    engine = PolicyEngine(self.backchannel_application, user, None)
201                    engine.empty_result = AppAccessWithoutBindings.get()
202                    engine.build()
203                    if engine.passing:
204                        pks.append(user.pk)
205                base = base.filter(pk__in=pks)
206            return base.order_by("pk")
207
208        if type == Group:
209            # Get queryset of all groups with consistent ordering
210            # according to the provider's settings
211            base = Group.objects.prefetch_related("scimprovidergroup_set").all().filter(**kwargs)
212
213            # Filter groups by group_filters if set
214            if self.group_filters.exists():
215                base = base.filter(pk__in=self.group_filters.values_list("pk", flat=True))
216
217            return base.order_by("pk")
218        raise ValueError(f"Invalid type {type}")
219
220    @classmethod
221    def get_object_mappings(cls, obj: User | Group) -> list[tuple[str, str]]:
222        if isinstance(obj, User):
223            return list(obj.scimprovideruser_set.values_list("provider__pk", "scim_id"))
224        if isinstance(obj, Group):
225            return list(obj.scimprovidergroup_set.values_list("provider__pk", "scim_id"))
226        raise ValueError(f"Invalid type {type(obj)}")
227
228    @property
229    def component(self) -> str:
230        return "ak-provider-scim-form"
231
232    @property
233    def service_provider_config_cache_timeout_seconds(self) -> int:
234        return max(
235            0,
236            int(timedelta_from_string(self.service_provider_config_cache_timeout).total_seconds()),
237        )
238
239    @property
240    def serializer(self) -> type[Serializer]:
241        from authentik.providers.scim.api.providers import SCIMProviderSerializer
242
243        return SCIMProviderSerializer
244
245    def __str__(self):
246        return f"SCIM Provider {self.name}"
247
248    class Meta:
249        verbose_name = _("SCIM Provider")
250        verbose_name_plural = _("SCIM Providers")
251
252
253class SCIMMapping(PropertyMapping):
254    """Map authentik data to outgoing SCIM requests"""
255
256    @property
257    def component(self) -> str:
258        return "ak-property-mapping-provider-scim-form"
259
260    @property
261    def serializer(self) -> type[Serializer]:
262        from authentik.providers.scim.api.property_mappings import SCIMMappingSerializer
263
264        return SCIMMappingSerializer
265
266    def __str__(self):
267        return f"SCIM Provider Mapping {self.name}"
268
269    class Meta:
270        verbose_name = _("SCIM Provider Mapping")
271        verbose_name_plural = _("SCIM Provider Mappings")

LOGGER = <BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>

class SCIMProviderUser(authentik.lib.models.InternallyManagedMixin, authentik.lib.models.SerializerModel): View Source

28class SCIMProviderUser(InternallyManagedMixin, SerializerModel):
29    """Mapping of a user and provider to a SCIM user ID"""
30
31    id = models.UUIDField(primary_key=True, editable=False, default=uuid4)
32    scim_id = models.TextField()
33    user = models.ForeignKey(User, on_delete=models.CASCADE)
34    provider = models.ForeignKey("SCIMProvider", on_delete=models.CASCADE)
35    attributes = models.JSONField(default=dict)
36
37    @property
38    def serializer(self) -> type[Serializer]:
39        from authentik.providers.scim.api.users import SCIMProviderUserSerializer
40
41        return SCIMProviderUserSerializer
42
43    class Meta:
44        unique_together = (("scim_id", "user", "provider"),)
45
46    def __str__(self) -> str:
47        return f"SCIM Provider User {self.user_id} to {self.provider_id}"

Mapping of a user and provider to a SCIM user ID

def id(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def scim_id(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

user

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

provider

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

def attributes(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

serializer: type[rest_framework.serializers.Serializer] View Source

37    @property
38    def serializer(self) -> type[Serializer]:
39        from authentik.providers.scim.api.users import SCIMProviderUserSerializer
40
41        return SCIMProviderUserSerializer

Get serializer for this model

user_id

provider_id

def objects(unknown):

The type of the None singleton.

Inherited Members

authentik.lib.models.SerializerModel: Meta

class SCIMProviderUser.DoesNotExist(django.core.exceptions.ObjectDoesNotExist):

The requested object does not exist

class SCIMProviderUser.MultipleObjectsReturned(django.core.exceptions.MultipleObjectsReturned):

The query returned multiple objects when only one was expected.

class SCIMProviderGroup(authentik.lib.models.InternallyManagedMixin, authentik.lib.models.SerializerModel): View Source

50class SCIMProviderGroup(InternallyManagedMixin, SerializerModel):
51    """Mapping of a group and provider to a SCIM user ID"""
52
53    id = models.UUIDField(primary_key=True, editable=False, default=uuid4)
54    scim_id = models.TextField()
55    group = models.ForeignKey(Group, on_delete=models.CASCADE)
56    provider = models.ForeignKey("SCIMProvider", on_delete=models.CASCADE)
57    attributes = models.JSONField(default=dict)
58
59    @property
60    def serializer(self) -> type[Serializer]:
61        from authentik.providers.scim.api.groups import SCIMProviderGroupSerializer
62
63        return SCIMProviderGroupSerializer
64
65    class Meta:
66        unique_together = (("scim_id", "group", "provider"),)
67
68    def __str__(self) -> str:
69        return f"SCIM Provider Group {self.group_id} to {self.provider_id}"

Mapping of a group and provider to a SCIM user ID

def id(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def scim_id(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

group

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

provider

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

def attributes(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

serializer: type[rest_framework.serializers.Serializer] View Source

59    @property
60    def serializer(self) -> type[Serializer]:
61        from authentik.providers.scim.api.groups import SCIMProviderGroupSerializer
62
63        return SCIMProviderGroupSerializer

Get serializer for this model

group_id

provider_id

def objects(unknown):

The type of the None singleton.

Inherited Members

authentik.lib.models.SerializerModel: Meta

class SCIMProviderGroup.DoesNotExist(django.core.exceptions.ObjectDoesNotExist):

The requested object does not exist

class SCIMProviderGroup.MultipleObjectsReturned(django.core.exceptions.MultipleObjectsReturned):

The query returned multiple objects when only one was expected.

class SCIMAuthenticationMode(django.db.models.enums.TextChoices): View Source

72class SCIMAuthenticationMode(models.TextChoices):
73    """SCIM authentication modes"""
74
75    TOKEN = "token", _("Token")
76    OAUTH = "oauth", _("OAuth")

SCIM authentication modes

TOKEN = SCIMAuthenticationMode.TOKEN

OAUTH = SCIMAuthenticationMode.OAUTH

class SCIMCompatibilityMode(django.db.models.enums.TextChoices): View Source

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")

SCIM compatibility mode

DEFAULT = SCIMCompatibilityMode.DEFAULT

AWS = SCIMCompatibilityMode.AWS

SLACK = SCIMCompatibilityMode.SLACK

SALESFORCE = SCIMCompatibilityMode.SALESFORCE

WEBEX = SCIMCompatibilityMode.WEBEX

class SCIMProvider(authentik.lib.sync.outgoing.models.OutgoingSyncProvider, authentik.core.models.BackchannelProvider): View Source

 89class SCIMProvider(OutgoingSyncProvider, BackchannelProvider):
 90    """SCIM 2.0 provider to create users and groups in external applications"""
 91
 92    exclude_users_service_account = models.BooleanField(default=False)
 93
 94    group_filters = models.ManyToManyField(
 95        "authentik_core.group",
 96        default=None,
 97        blank=True,
 98        help_text=_("Group filters used to define sync-scope for groups."),
 99    )
100
101    url = models.TextField(help_text=_("Base URL to SCIM requests, usually ends in /v2"))
102
103    auth_mode = models.TextField(
104        choices=SCIMAuthenticationMode.choices, default=SCIMAuthenticationMode.TOKEN
105    )
106
107    token = models.TextField(help_text=_("Authentication token"), blank=True)
108    auth_oauth = models.ForeignKey(
109        "authentik_sources_oauth.OAuthSource",
110        on_delete=models.SET_DEFAULT,
111        default=None,
112        null=True,
113        help_text=_("OAuth Source used for authentication"),
114    )
115    auth_oauth_params = models.JSONField(
116        blank=True, default=dict, help_text=_("Additional OAuth parameters, such as grant_type")
117    )
118    auth_oauth_user = models.ForeignKey(
119        "authentik_core.User", on_delete=models.SET_NULL, default=None, null=True
120    )
121
122    verify_certificates = models.BooleanField(default=True)
123
124    property_mappings_group = models.ManyToManyField(
125        PropertyMapping,
126        default=None,
127        blank=True,
128        help_text=_("Property mappings used for group creation/updating."),
129    )
130
131    compatibility_mode = models.CharField(
132        max_length=30,
133        choices=SCIMCompatibilityMode.choices,
134        default=SCIMCompatibilityMode.DEFAULT,
135        verbose_name=_("SCIM Compatibility Mode"),
136        help_text=_("Alter authentik behavior for vendor-specific SCIM implementations."),
137    )
138    service_provider_config_cache_timeout = models.TextField(
139        default="hours=1",
140        validators=[timedelta_string_validator],
141        help_text=_(
142            "Cache duration for ServiceProviderConfig responses. Set minutes=0 to disable."
143        ),
144    )
145
146    def scim_auth(self) -> AuthBase:
147        if self.auth_mode == SCIMAuthenticationMode.OAUTH:
148            try:
149                from authentik.enterprise.providers.scim.auth_oauth2 import SCIMOAuthAuth
150
151                return SCIMOAuthAuth(self)
152            except ImportError:
153                LOGGER.warning("Failed to import SCIM OAuth Client")
154        return SCIMTokenAuth(self)
155
156    @property
157    def icon_url(self) -> str | None:
158        return static("authentik/sources/scim.png")
159
160    @property
161    def sync_actor(self) -> Actor:
162        from authentik.providers.scim.tasks import scim_sync
163
164        return scim_sync
165
166    def client_for_model(
167        self, model: type[User | Group | SCIMProviderUser | SCIMProviderGroup]
168    ) -> BaseOutgoingSyncClient[User | Group, Any, Any, Self]:
169        if issubclass(model, User | SCIMProviderUser):
170            from authentik.providers.scim.clients.users import SCIMUserClient
171
172            return SCIMUserClient(self)
173        if issubclass(model, Group | SCIMProviderGroup):
174            from authentik.providers.scim.clients.groups import SCIMGroupClient
175
176            return SCIMGroupClient(self)
177        raise ValueError(f"Invalid model {model}")
178
179    def save(self, *args, **kwargs):
180        from django.core.cache import cache
181
182        cache_key = f"goauthentik.io/providers/scim/{self.pk}/service_provider_config"
183        cache.delete(cache_key)
184        super().save(*args, **kwargs)
185
186    def get_object_qs(self, type: type[User | Group], **kwargs) -> QuerySet[User | Group]:
187        if type == User:
188            # Get queryset of all users with consistent ordering
189            # according to the provider's settings
190            base = User.objects.all().exclude_anonymous().filter(**kwargs)
191            if self.exclude_users_service_account:
192                base = base.exclude(type=UserTypes.SERVICE_ACCOUNT).exclude(
193                    type=UserTypes.INTERNAL_SERVICE_ACCOUNT
194                )
195
196            # Filter users by their access to the backchannel application if an application is set
197            # This handles both policy bindings and group_filters
198            if self.backchannel_application:
199                pks = []
200                for user in base:
201                    engine = PolicyEngine(self.backchannel_application, user, None)
202                    engine.empty_result = AppAccessWithoutBindings.get()
203                    engine.build()
204                    if engine.passing:
205                        pks.append(user.pk)
206                base = base.filter(pk__in=pks)
207            return base.order_by("pk")
208
209        if type == Group:
210            # Get queryset of all groups with consistent ordering
211            # according to the provider's settings
212            base = Group.objects.prefetch_related("scimprovidergroup_set").all().filter(**kwargs)
213
214            # Filter groups by group_filters if set
215            if self.group_filters.exists():
216                base = base.filter(pk__in=self.group_filters.values_list("pk", flat=True))
217
218            return base.order_by("pk")
219        raise ValueError(f"Invalid type {type}")
220
221    @classmethod
222    def get_object_mappings(cls, obj: User | Group) -> list[tuple[str, str]]:
223        if isinstance(obj, User):
224            return list(obj.scimprovideruser_set.values_list("provider__pk", "scim_id"))
225        if isinstance(obj, Group):
226            return list(obj.scimprovidergroup_set.values_list("provider__pk", "scim_id"))
227        raise ValueError(f"Invalid type {type(obj)}")
228
229    @property
230    def component(self) -> str:
231        return "ak-provider-scim-form"
232
233    @property
234    def service_provider_config_cache_timeout_seconds(self) -> int:
235        return max(
236            0,
237            int(timedelta_from_string(self.service_provider_config_cache_timeout).total_seconds()),
238        )
239
240    @property
241    def serializer(self) -> type[Serializer]:
242        from authentik.providers.scim.api.providers import SCIMProviderSerializer
243
244        return SCIMProviderSerializer
245
246    def __str__(self):
247        return f"SCIM Provider {self.name}"
248
249    class Meta:
250        verbose_name = _("SCIM Provider")
251        verbose_name_plural = _("SCIM Providers")

SCIM 2.0 provider to create users and groups in external applications

def exclude_users_service_account(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

group_filters

Accessor to the related objects manager on the forward and reverse sides of a many-to-many relation.

In the example::

class Pizza(Model):
    toppings = ManyToManyField(Topping, related_name='pizzas')

Pizza.toppings and Topping.pizzas are ManyToManyDescriptor instances.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

def url(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def auth_mode(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def token(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

auth_oauth

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

def auth_oauth_params(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

auth_oauth_user

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

def verify_certificates(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

property_mappings_group

Accessor to the related objects manager on the forward and reverse sides of a many-to-many relation.

In the example::

class Pizza(Model):
    toppings = ManyToManyField(Topping, related_name='pizzas')

Pizza.toppings and Topping.pizzas are ManyToManyDescriptor instances.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

def compatibility_mode(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def service_provider_config_cache_timeout(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def scim_auth(self) -> requests.auth.AuthBase: View Source

146    def scim_auth(self) -> AuthBase:
147        if self.auth_mode == SCIMAuthenticationMode.OAUTH:
148            try:
149                from authentik.enterprise.providers.scim.auth_oauth2 import SCIMOAuthAuth
150
151                return SCIMOAuthAuth(self)
152            except ImportError:
153                LOGGER.warning("Failed to import SCIM OAuth Client")
154        return SCIMTokenAuth(self)

icon_url: str | None View Source

156    @property
157    def icon_url(self) -> str | None:
158        return static("authentik/sources/scim.png")

sync_actor: dramatiq.actor.Actor View Source

160    @property
161    def sync_actor(self) -> Actor:
162        from authentik.providers.scim.tasks import scim_sync
163
164        return scim_sync

def client_for_model( self, model: type[authentik.core.models.User | authentik.core.models.Group | SCIMProviderUser | SCIMProviderGroup]) -> authentik.lib.sync.outgoing.base.BaseOutgoingSyncClient[authentik.core.models.User | authentik.core.models.Group, typing.Any, typing.Any, typing.Self]: View Source

166    def client_for_model(
167        self, model: type[User | Group | SCIMProviderUser | SCIMProviderGroup]
168    ) -> BaseOutgoingSyncClient[User | Group, Any, Any, Self]:
169        if issubclass(model, User | SCIMProviderUser):
170            from authentik.providers.scim.clients.users import SCIMUserClient
171
172            return SCIMUserClient(self)
173        if issubclass(model, Group | SCIMProviderGroup):
174            from authentik.providers.scim.clients.groups import SCIMGroupClient
175
176            return SCIMGroupClient(self)
177        raise ValueError(f"Invalid model {model}")

def save(self, *args, **kwargs): View Source

179    def save(self, *args, **kwargs):
180        from django.core.cache import cache
181
182        cache_key = f"goauthentik.io/providers/scim/{self.pk}/service_provider_config"
183        cache.delete(cache_key)
184        super().save(*args, **kwargs)

Save the current instance. Override this in a subclass if you want to control the saving process.

The 'force_insert' and 'force_update' parameters can be used to insist that the "save" must be an SQL insert or update (or equivalent for non-SQL backends), respectively. Normally, they should not be set.

def get_object_qs( self, type: type[authentik.core.models.User | authentik.core.models.Group], **kwargs) -> django.db.models.query.QuerySet: View Source

186    def get_object_qs(self, type: type[User | Group], **kwargs) -> QuerySet[User | Group]:
187        if type == User:
188            # Get queryset of all users with consistent ordering
189            # according to the provider's settings
190            base = User.objects.all().exclude_anonymous().filter(**kwargs)
191            if self.exclude_users_service_account:
192                base = base.exclude(type=UserTypes.SERVICE_ACCOUNT).exclude(
193                    type=UserTypes.INTERNAL_SERVICE_ACCOUNT
194                )
195
196            # Filter users by their access to the backchannel application if an application is set
197            # This handles both policy bindings and group_filters
198            if self.backchannel_application:
199                pks = []
200                for user in base:
201                    engine = PolicyEngine(self.backchannel_application, user, None)
202                    engine.empty_result = AppAccessWithoutBindings.get()
203                    engine.build()
204                    if engine.passing:
205                        pks.append(user.pk)
206                base = base.filter(pk__in=pks)
207            return base.order_by("pk")
208
209        if type == Group:
210            # Get queryset of all groups with consistent ordering
211            # according to the provider's settings
212            base = Group.objects.prefetch_related("scimprovidergroup_set").all().filter(**kwargs)
213
214            # Filter groups by group_filters if set
215            if self.group_filters.exists():
216                base = base.filter(pk__in=self.group_filters.values_list("pk", flat=True))
217
218            return base.order_by("pk")
219        raise ValueError(f"Invalid type {type}")

@classmethod

def get_object_mappings( cls, obj: authentik.core.models.User | authentik.core.models.Group) -> list[tuple[str, str]]: View Source

221    @classmethod
222    def get_object_mappings(cls, obj: User | Group) -> list[tuple[str, str]]:
223        if isinstance(obj, User):
224            return list(obj.scimprovideruser_set.values_list("provider__pk", "scim_id"))
225        if isinstance(obj, Group):
226            return list(obj.scimprovidergroup_set.values_list("provider__pk", "scim_id"))
227        raise ValueError(f"Invalid type {type(obj)}")

Get a list of mapping between User/Group and ProviderUser/Group: [("provider_pk", "obj_pk")]

component: str View Source

229    @property
230    def component(self) -> str:
231        return "ak-provider-scim-form"

Return component used to edit this object

service_provider_config_cache_timeout_seconds: int View Source

233    @property
234    def service_provider_config_cache_timeout_seconds(self) -> int:
235        return max(
236            0,
237            int(timedelta_from_string(self.service_provider_config_cache_timeout).total_seconds()),
238        )

serializer: type[rest_framework.serializers.Serializer] View Source

240    @property
241    def serializer(self) -> type[Serializer]:
242        from authentik.providers.scim.api.providers import SCIMProviderSerializer
243
244        return SCIMProviderSerializer

Get serializer for this model

def sync_page_size(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def sync_page_timeout(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def dry_run(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

schedules

Accessor to the related objects manager on the one-to-many relation created by GenericRelation.

In the example::

class Post(Model):
    comments = GenericRelation(Comment)

post.comments is a ReverseGenericManyToOneDescriptor instance.

tasks

Accessor to the related objects manager on the one-to-many relation created by GenericRelation.

In the example::

class Post(Model):
    comments = GenericRelation(Comment)

post.comments is a ReverseGenericManyToOneDescriptor instance.

def get_auth_mode_display(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

auth_oauth_id

auth_oauth_user_id

def get_compatibility_mode_display(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

provider_ptr_id

provider_ptr

Accessor to the related object on the forward side of a one-to-one relation.

In the example::

class Restaurant(Model):
    place = OneToOneField(Place, related_name='restaurant')

Restaurant.place is a ForwardOneToOneDescriptor instance.

scimprovideruser_set

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

scimprovidergroup_set

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

Inherited Members

authentik.lib.sync.outgoing.models.OutgoingSyncProvider: Meta; get_paginator; get_object_sync_time_limit_ms; get_sync_time_limit_ms; sync_lock; sync_dispatch; schedule_specs
authentik.tasks.schedules.models.ScheduledModel: models
authentik.core.models.Provider: name; authentication_flow; authorization_flow; invalidation_flow; property_mappings; backchannel_application; is_backchannel; objects; launch_url; authentication_flow_id; authorization_flow_id; invalidation_flow_id; backchannel_application_id; id; application; outpost_set; oauth2provider; ldapprovider; racprovider; radiusprovider; samlprovider; scimprovider; googleworkspaceprovider; microsoftentraprovider; ssfprovider

class SCIMProvider.DoesNotExist(django.core.exceptions.ObjectDoesNotExist):

The requested object does not exist

class SCIMProvider.MultipleObjectsReturned(django.core.exceptions.MultipleObjectsReturned):

The query returned multiple objects when only one was expected.

class SCIMMapping(authentik.core.models.PropertyMapping): View Source

254class SCIMMapping(PropertyMapping):
255    """Map authentik data to outgoing SCIM requests"""
256
257    @property
258    def component(self) -> str:
259        return "ak-property-mapping-provider-scim-form"
260
261    @property
262    def serializer(self) -> type[Serializer]:
263        from authentik.providers.scim.api.property_mappings import SCIMMappingSerializer
264
265        return SCIMMappingSerializer
266
267    def __str__(self):
268        return f"SCIM Provider Mapping {self.name}"
269
270    class Meta:
271        verbose_name = _("SCIM Provider Mapping")
272        verbose_name_plural = _("SCIM Provider Mappings")

Map authentik data to outgoing SCIM requests

component: str View Source

257    @property
258    def component(self) -> str:
259        return "ak-property-mapping-provider-scim-form"

Return component used to edit this object

serializer: type[rest_framework.serializers.Serializer] View Source

261    @property
262    def serializer(self) -> type[Serializer]:
263        from authentik.providers.scim.api.property_mappings import SCIMMappingSerializer
264
265        return SCIMMappingSerializer

Get serializer for this model

propertymapping_ptr_id

propertymapping_ptr

Accessor to the related object on the forward side of a one-to-one relation.

In the example::

class Restaurant(Model):
    place = OneToOneField(Place, related_name='restaurant')

Restaurant.place is a ForwardOneToOneDescriptor instance.

Inherited Members

authentik.lib.models.SerializerModel: Meta
authentik.core.models.PropertyMapping: pm_uuid; name; expression; objects; evaluate; managed; provider_set; source_userpropertymappings_set; source_grouppropertymappings_set; notificationwebhookmapping; oauthsourcepropertymapping; scopemapping; endpoint_set; racpropertymapping; radiusproviderpropertymapping; samlsourcepropertymapping; samlpropertymapping; scimprovider_set; scimmapping; kerberossourcepropertymapping; ldapsourcepropertymapping; plexsourcepropertymapping; scimsourcepropertymapping; telegramsourcepropertymapping; googleworkspaceprovider_set; googleworkspaceprovidermapping; microsoftentraprovider_set; microsoftentraprovidermapping

class SCIMMapping.DoesNotExist(authentik.core.models.PropertyMapping.DoesNotExist):

The requested object does not exist

class SCIMMapping.MultipleObjectsReturned(authentik.core.models.PropertyMapping.MultipleObjectsReturned):

The query returned multiple objects when only one was expected.