🚧(scim) add SCIM importer base mechanics

This could allow a SCIM client to provision users and groups.

src/backend/people/settings.py

@@ -215,6 +215,7 @@ class Base(Configuration):
         "django.contrib.auth.backends.ModelBackend",
         "mailbox_oauth2.backends.MailboxModelBackend",
         "core.authentication.backends.OIDCAuthenticationBackend",
+        "scim_importer.authentication.ScimClientBackend",
     ]
 
     # Django's applications from the highest priority to the lowest
@@ -231,6 +232,7 @@ class Base(Configuration):
         "demo",
         "mailbox_manager.apps.MailboxManagerConfig",
         "mailbox_oauth2",
+        "scim_importer",
         *INSTALLED_PLUGINS,
         # Third party apps
         "django_zxcvbn_password_validator",
@@ -239,6 +241,7 @@ class Base(Configuration):
         "corsheaders",
         "django_celery_beat",
         "django_celery_results",
+        "django_scim",
         "dockerflow.django",
         "easy_thumbnails",
         "oauth2_provider",
@@ -630,6 +633,26 @@ class Base(Configuration):
         environ_prefix=None,
     )
 
+    # SCIM
+    SCIM_SERVICE_PROVIDER = {
+        "NETLOC": "localhost",
+        "AUTHENTICATION_SCHEMES": [
+            {
+                "type": "oauth2",
+                "name": "OAuth 2",
+                "description": "Oauth 2 implemented with bearer token",
+            },
+        ],
+        "USER_MODEL_GETTER": "scim_importer.utils.get_scim_importer_user_model",
+        "GROUP_MODEL": "scim_importer.models.ScimImportedGroup",
+        "SERVICE_PROVIDER_CONFIG_MODEL": "scim_importer.models.SCIMServiceProviderConfig",
+        "USER_ADAPTER": "scim_importer.adapters.SCIMUser",
+        "GROUP_ADAPTER": "scim_importer.adapters.SCIMGroup",
+        "USER_FILTER_PARSER": "scim_importer.filters.UserFilterQuery",
+        "GROUP_FILTER_PARSER": "scim_importer.filters.GroupFilterQuery",
+        "GET_EXTRA_MODEL_FILTER_KWARGS_GETTER": "scim_importer.utils.default_get_extra_model_filter_kwargs_getter",
+    }
+
     # pylint: disable=invalid-name
     @property
     def ENVIRONMENT(self):

src/backend/people/urls.py

@@ -26,6 +26,7 @@ urlpatterns = (
     [
         path("admin/", admin.site.urls),
         path("o/", include(oauth2_urls)),
+        path("scim/v2/", include("django_scim.urls")),
     ]
     + api_urls.urlpatterns
     + resource_server_urls.urlpatterns

src/backend/pyproject.toml

@@ -38,6 +38,7 @@ dependencies = [
     "django-oauth-toolkit==3.0.1",
     "django-parler==2.3",
     "django-redis==5.4.0",
+    "django-scim2==0.19.1",
     "django-storages==1.14.6",
     "django-timezone-field>=5.1",
     "django-treebeard==4.7.1",

src/backend/scim_importer/adapters.py

@@ -0,0 +1,232 @@
+"""
+Adapters are used to convert the data model described by the SCIM 2.0
+specification to a data model that fits the data provided by the application
+implementing a SCIM api.
+
+For example, in a Django app, there are User and Group models that do
+not have the same attributes/fields that are defined by the SCIM 2.0
+specification. The Django User model has both ``first_name`` and ``last_name``
+attributes but the SCIM speicifcation requires this same data be sent under
+the names ``givenName`` and ``familyName`` respectively.
+
+An adapter is instantiated with a model instance. Eg::
+
+    user = get_user_model().objects.get(id=1)
+    scim_user = SCIMUser(user)
+    ...
+
+"""
+
+import uuid
+from urllib.parse import urljoin
+
+from django.core.exceptions import ValidationError
+from django.urls import reverse
+
+from django_scim import constants, exceptions
+from django_scim.adapters import SCIMGroup as BaseSCIMGroup
+from django_scim.adapters import SCIMUser as BaseSCIMUser
+from django_scim.utils import get_user_adapter, get_user_model
+
+
+class SCIMUser(BaseSCIMUser):
+    """
+    Adapter for adding SCIM functionality to a Django User object.
+
+    This adapter can be overridden; see the ``USER_ADAPTER`` setting
+    for details.
+    """
+
+    @property
+    def display_name(self):
+        """
+        Return the displayName of the user per the SCIM spec.
+        """
+        if self.obj.first_name and self.obj.last_name:
+            return f"{self.obj.first_name} {self.obj.last_name}".strip()
+        return self.obj.scim_username
+
+    @property
+    def meta(self):
+        """
+        Return the meta object of the user per the SCIM spec.
+        """
+        d = {
+            "resourceType": self.resource_type,
+            "created": self.obj.created_at.isoformat(),
+            "lastModified": self.obj.updated_at.isoformat(),
+            "location": self.location,
+        }
+
+        return d
+
+    def to_dict(self):
+        """
+        Return a ``dict`` conforming to the SCIM User Schema,
+        ready for conversion to a JSON object.
+        """
+        return {
+            "id": self.id,
+            # "externalId": self.obj.scim_external_id,
+            "schemas": [constants.SchemaURI.USER],
+            "userName": self.obj.scim_username,
+            "name": {
+                "givenName": self.obj.first_name,
+                "familyName": self.obj.last_name,
+                "formatted": self.display_name,
+            },
+            "displayName": self.display_name,
+            "emails": self.emails,
+            "active": self.obj.is_active,
+            "groups": self.groups,
+            "meta": self.meta,
+        }
+
+    def from_dict(self, d):
+        """
+        Consume a ``dict`` conforming to the SCIM User Schema, updating the
+        internal user object with data from the ``dict``.
+
+        Please note, the user object is not saved within this method. To
+        persist the changes made by this method, please call ``.save()`` on the
+        adapter. Eg::
+
+            scim_user.from_dict(d)
+            scim_user.save()
+        """
+        self.obj.client = self.request.user
+
+        scim_external_id = d.get("externalId")
+        self.obj.scim_external_id = scim_external_id or ""
+
+        if self.obj.scim_id is None:
+            self.obj.scim_id = uuid.uuid4()
+
+        username = d.get("userName")
+        self.obj.scim_username = username or ""
+
+        # self.obj.scim_username = self.obj.username
+
+        first_name = d.get("name", {}).get("givenName")
+        self.obj.first_name = first_name or ""
+
+        last_name = d.get("name", {}).get("familyName")
+        self.obj.last_name = last_name or ""
+
+        emails = d.get("emails", [])
+        self.parse_emails(emails)
+
+        # cleartext_password = d.get('password')
+        # if cleartext_password:
+        #    self.obj.set_password(cleartext_password)
+
+        active = d.get("active")
+        if active is not None:
+            self.obj.is_active = active
+
+        try:
+            self.obj.full_clean()
+        except ValidationError as exc:
+            raise exceptions.IntegrityError(str(exc)) from exc
+
+
+class SCIMGroup(BaseSCIMGroup):
+    """
+    Adapter for adding SCIM functionality to a Django Group object.
+
+    This adapter can be overridden; see the ``GROUP_ADAPTER``
+    setting for details.
+    """
+
+    @property
+    def display_name(self):
+        """
+        Return the displayName of the group per the SCIM spec.
+        """
+        return self.obj.scim_display_name
+
+    def to_dict(self):
+        """
+        Return a ``dict`` conforming to the SCIM Group Schema,
+        ready for conversion to a JSON object.
+        """
+        return {
+            "id": self.id,
+            "schemas": [constants.SchemaURI.GROUP],
+            "displayName": self.obj.scim_display_name,
+            "members": self.members,
+            "meta": self.meta,
+        }
+
+    def from_dict(self, d):
+        """
+        Consume a ``dict`` conforming to the SCIM Group Schema, updating the
+        internal group object with data from the ``dict``.
+
+        Please note, the group object is not saved within this method. To
+        persist the changes made by this method, please call ``.save()`` on the
+        adapter. Eg::
+
+            scim_group.from_dict(d)
+            scim_group.save()
+        """
+        super().from_dict(d)
+
+        if self.obj.scim_id is None:
+            self.obj.scim_id = uuid.uuid4()
+
+        display_name = d.get("displayName")
+        self.obj.scim_display_name = display_name or ""
+
+        self.obj.client = self.request.user
+
+    def handle_add(self, path, value, operation):
+        """
+        Handle add operations.
+        """
+        if path.first_path == ("members", None, None):
+            members = value or []
+            ids = [member.get("value") for member in members]
+            users = get_user_model().objects.filter(scim_id__in=ids)
+
+            if len(ids) != users.count():
+                raise exceptions.BadRequestError(
+                    "Can not add a non-existent user to group"
+                )
+
+            for user in users:
+                self.obj.user_set.add(user)
+
+        else:
+            raise exceptions.NotImplementedError
+
+    def handle_remove(self, path, value, operation):
+        """
+        Handle remove operations.
+        """
+        if path.first_path == ("members", None, None):
+            members = value or []
+            ids = [member.get("value") for member in members]
+            users = get_user_model().objects.filter(scim_id__in=ids)
+
+            if len(ids) != users.count():
+                raise exceptions.BadRequestError(
+                    "Can not remove a non-existent user from group"
+                )
+
+            for user in users:
+                self.obj.user_set.remove(user)
+
+        else:
+            raise exceptions.NotImplementedError
+
+    def handle_replace(self, path, value, operation):
+        """
+        Handle the replace operations.
+        """
+        if path.first_path == ("displayName", None, None):
+            self.obj.scim_display_name = value
+            self.save()
+
+        else:
+            raise exceptions.NotImplementedError

src/backend/scim_importer/admin.py

@@ -0,0 +1,37 @@
+"""Admin interface for SCIM importer models."""
+
+from django.contrib import admin
+
+from . import models
+
+
+@admin.register(models.ScimClient)
+class ScimClientAdmin(admin.ModelAdmin):
+    """Admin for SCIM client model."""
+
+    list_display = (
+        "name",
+        "is_active",
+    )
+    search_fields = ("name",)
+
+
+@admin.register(models.ScimImportedUser)
+class ScimImportedUserAdmin(admin.ModelAdmin):
+    """Admin for SCIM imported user model."""
+
+    list_display = ("scim_username", "client")
+    list_filter = ("client",)
+    list_select_related = ("client",)
+    autocomplete_fields = ("user", "client", "scim_groups")
+
+
+@admin.register(models.ScimImportedGroup)
+class ScimImportedGroupAdmin(admin.ModelAdmin):
+    """Admin for SCIM imported group model."""
+
+    list_display = ("scim_display_name", "client")
+    list_filter = ("client",)
+    list_select_related = ("client",)
+    autocomplete_fields = ("team", "client")
+    search_fields = ("scim_display_name",)

src/backend/scim_importer/apps.py

@@ -0,0 +1,12 @@
+"""SCIM Importer application"""
+
+from django.apps import AppConfig
+from django.utils.translation import gettext_lazy as _
+
+
+class ScimImporterConfig(AppConfig):
+    """Configuration class for the scim_importer app."""
+
+    name = "scim_importer"
+    app_label = "scim_importer"
+    verbose_name = _("SCIM Importer")

src/backend/scim_importer/authentication.py

@@ -0,0 +1,84 @@
+import logging
+
+from django.contrib.auth.backends import BaseBackend
+
+from scim_importer.models import ScimClient
+
+logger = logging.getLogger(__name__)
+
+
+class ScimClientBackend(BaseBackend):
+    """
+    Authenticates against ScimClient.
+    """
+
+    def authenticate(self, request, **kwargs):
+        auth_header = request.headers.get("Authorization", "")
+        logger.info("auth_header: %s", auth_header)
+        if not auth_header:
+            return None
+
+        try:
+            auth_mode, token = auth_header.split(" ")
+        except (IndexError, ValueError) as err:
+            logger.error("Invalid auth header: %s", err)
+            return None
+
+        if auth_mode.lower() != "bearer" or not token:
+            logger.error("Invalid auth header: %s", auth_header)
+            return None
+
+        try:
+            scim_client = ScimClient.objects.get(token=token)
+        except ScimClient.DoesNotExist:
+            logger.warning("Invalid token: %s", token)
+            return None
+
+        logger.info("scim_client: %s", scim_client)
+        return scim_client
+
+    async def aauthenticate(self, request, **kwargs):
+        auth_header = request.headers.get("Authorization", "")
+        logger.info("auth_header: %s", auth_header)
+        if not auth_header:
+            return None
+
+        try:
+            auth_mode, token = auth_header.split(" ")
+        except (IndexError, ValueError) as err:
+            logger.error("Invalid auth header: %s", err)
+            return None
+
+        if auth_mode.lower() != "bearer" or not token:
+            logger.error("Invalid auth header: %s", auth_header)
+            return None
+
+        try:
+            scim_client = await ScimClient.objects.aget(token=token)
+        except ScimClient.DoesNotExist:
+            logger.warning("Invalid token: %s", token)
+            return None
+
+        logger.info("scim_client: %s", scim_client)
+        return scim_client
+
+    def user_can_authenticate(self, user):
+        """
+        Reject users with is_active=False. Custom user models that don't have
+        that attribute are allowed.
+        """
+        return user.is_active
+
+    def get_user(self, user_id):
+        try:
+            user = ScimClient.objects.get(pk=user_id)
+        except ScimClient.DoesNotExist:
+            return None
+        return user if self.user_can_authenticate(user) else None
+
+    async def aget_user(self, user_id):
+        try:
+            user = await ScimClient.objects.aget(pk=user_id)
+        except ScimClient.DoesNotExist:
+            return None
+        return user if self.user_can_authenticate(user) else None

src/backend/scim_importer/filters.py

@@ -0,0 +1,18 @@
+from django_scim.filters import GroupFilterQuery as BaseGroupFilterQuery
+from django_scim.filters import UserFilterQuery as BaseUserFilterQuery
+
+
+class UserFilterQuery(BaseUserFilterQuery):
+    attr_map = {
+        # attr, sub attr, uri
+        ("userName", None, None): "scim_username",
+        ("name", "familyName", None): "last_name",
+        ("familyName", None, None): "last_name",
+        ("name", "givenName", None): "first_name",
+        ("givenName", None, None): "first_name",
+        ("active", None, None): "is_active",
+    }
+
+
+class GroupFilterQuery(BaseGroupFilterQuery):
+    attr_map = {}

src/backend/scim_importer/migrations/0001_initial.py

@@ -0,0 +1,76 @@
+# Generated by Django 5.2 on 2025-04-25 17:31
+
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('core', '0015_alter_accountservice_api_key'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ScimClient',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('name', models.CharField(blank=True, max_length=100, null=True, verbose_name='name')),
+                ('token', models.CharField(help_text='Authentication token for the SCIM client', max_length=100, unique=True, verbose_name='Token')),
+                ('is_active', models.BooleanField(default=True, help_text='Whether this client should be treated as active. Unselect this instead of deleting accounts.', verbose_name='active')),
+                ('organization', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='scim_clients', to='core.organization', verbose_name='organization')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='ScimImportedGroup',
+            fields=[
+                ('scim_id', models.CharField(blank=True, default=None, help_text='A unique identifier for a SCIM resource as defined by the service provider.', max_length=254, null=True, unique=True, verbose_name='SCIM ID')),
+                ('scim_external_id', models.CharField(blank=True, db_index=True, default=None, help_text='A string that is an identifier for the resource as defined by the provisioning client.', max_length=254, null=True, verbose_name='SCIM External ID')),
+                ('scim_display_name', models.CharField(blank=True, db_index=True, default=None, help_text='A human-readable name for the Group.', max_length=254, null=True, verbose_name='SCIM Display Name')),
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('client', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='groups', to='scim_importer.scimclient', verbose_name='client')),
+                ('team', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='scim_groups', to='core.team')),
+            ],
+            options={
+                'verbose_name': 'SCIM imported group',
+                'verbose_name_plural': 'SCIM imported groups',
+                'db_table': 'people_scim_importer_group',
+            },
+        ),
+        migrations.CreateModel(
+            name='ScimImportedUser',
+            fields=[
+                ('scim_id', models.CharField(blank=True, default=None, help_text='A unique identifier for a SCIM resource as defined by the service provider.', max_length=254, null=True, unique=True, verbose_name='SCIM ID')),
+                ('scim_external_id', models.CharField(blank=True, db_index=True, default=None, help_text='A string that is an identifier for the resource as defined by the provisioning client.', max_length=254, null=True, verbose_name='SCIM External ID')),
+                ('scim_username', models.CharField(blank=True, db_index=True, default=None, help_text="A service provider's unique identifier for the user", max_length=254, null=True, verbose_name='SCIM Username')),
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('first_name', models.CharField(blank=True, max_length=100, null=True, verbose_name='first name')),
+                ('last_name', models.CharField(blank=True, max_length=100, null=True, verbose_name='last name')),
+                ('email', models.EmailField(blank=True, max_length=254, null=True, verbose_name='email address')),
+                ('is_active', models.BooleanField(default=True, help_text='Whether this user should be treated as active. Unselect this instead of deleting accounts.', verbose_name='active')),
+                ('client', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='users', to='scim_importer.scimclient', verbose_name='client')),
+                ('scim_groups', models.ManyToManyField(related_name='user_set', to='scim_importer.scimimportedgroup')),
+                ('user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='scim_users', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'verbose_name': 'SCIM imported user',
+                'verbose_name_plural': 'SCIM imported users',
+                'db_table': 'people_scim_importer_user',
+                'unique_together': {('client', 'scim_username')},
+            },
+        ),
+    ]

src/backend/scim_importer/models.py

@@ -0,0 +1,173 @@
+"""
+Declare and configure the models for the scim_importer application
+"""
+
+from logging import getLogger
+
+from django.contrib.auth import get_user_model
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+from django_scim.models import (
+    AbstractSCIMCommonAttributesMixin,
+    AbstractSCIMGroupMixin,
+    AbstractSCIMUserMixin,
+)
+from django_scim.models import (
+    SCIMServiceProviderConfig as BaseSCIMServiceProviderConfig,
+)
+
+from core.models import BaseModel, Organization, Team
+
+logger = getLogger(__name__)
+
+
+class ScimClient(BaseModel):
+    """SCIM client model."""
+
+    # XXX: should allow OAuth2 connection for this model?
+    name = models.CharField(_("name"), max_length=100, null=True, blank=True)
+    token = models.CharField(  # XXX: should stored as a hash?
+        _("Token"),
+        max_length=100,
+        unique=True,
+        help_text=_("Authentication token for the SCIM client"),
+    )
+    is_active = models.BooleanField(
+        _("active"),
+        default=True,
+        help_text=_(
+            "Whether this client should be treated as active. "
+            "Unselect this instead of deleting accounts."
+        ),
+    )
+
+    organization = models.ForeignKey(
+        Organization,
+        on_delete=models.CASCADE,
+        related_name="scim_clients",
+        verbose_name=_("organization"),
+    )
+
+    @property
+    def is_authenticated(self):
+        return True
+
+
+# XXX: implement the case when a ScimImportedUser exists but not the user,
+#      and the user is created via OIDC (update ScimImportedUser and create
+#      Team (?) and add UserTeamAccess).
+class ScimImportedUser(
+    AbstractSCIMUserMixin,
+    BaseModel,
+):
+    """SCIM imported user model."""
+
+    client = models.ForeignKey(
+        ScimClient,
+        on_delete=models.CASCADE,
+        related_name="users",
+        verbose_name=_("client"),
+    )
+
+    user = models.ForeignKey(
+        get_user_model(),
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name="scim_users",
+    )
+
+    first_name = models.CharField(
+        _("first name"), max_length=100, null=True, blank=True
+    )
+    last_name = models.CharField(_("last name"), max_length=100, null=True, blank=True)
+    email = models.EmailField(_("email address"), null=True, blank=True)
+
+    is_active = models.BooleanField(
+        _("active"),
+        default=True,
+        help_text=_(
+            "Whether this user should be treated as active. "
+            "Unselect this instead of deleting accounts."
+        ),
+    )
+
+    scim_groups = models.ManyToManyField(
+        "ScimImportedGroup",
+        related_name="user_set",
+    )
+
+    class Meta:
+        db_table = "people_scim_importer_user"
+        verbose_name = _("SCIM imported user")
+        verbose_name_plural = _("SCIM imported users")
+        unique_together = (("client", "scim_username"),)
+
+    @property
+    def display_name(self):
+        """Return the display name of the user."""
+        return f"{self.first_name} {self.last_name}".strip()
+
+    def __str__(self):
+        return self.display_name
+
+    def save(self, *args, **kwargs):
+        """Disable the django_scim save method."""
+        super(AbstractSCIMCommonAttributesMixin, self).save(*args, **kwargs)
+        # XXX: What should we do when a user is linked? at creation, at update?
+
+
+class ScimImportedGroup(AbstractSCIMGroupMixin, BaseModel):
+    """SCIM imported group model."""
+
+    client = models.ForeignKey(
+        ScimClient,
+        on_delete=models.CASCADE,
+        related_name="groups",
+        verbose_name=_("client"),
+    )
+
+    team = models.ForeignKey(
+        Team,
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name="scim_groups",
+    )
+
+    class Meta:
+        db_table = "people_scim_importer_group"
+        verbose_name = _("SCIM imported group")
+        verbose_name_plural = _("SCIM imported groups")
+
+    def __str__(self):
+        return self.scim_display_name or str(self.pk)
+
+    def save(self, *args, **kwargs):
+        """Disable the django_scim save method."""
+        super(AbstractSCIMCommonAttributesMixin, self).save(*args, **kwargs)
+        # XXX: What should we do when a team is linked? at creation, at update?
+        #      Should we create the team if it does not exist? Only if there is at least one user
+        #      we already know about (ScimImportedUser.user)?
+
+
+class SCIMServiceProviderConfig(BaseSCIMServiceProviderConfig):
+    """
+    SCIM Service Provider Config for SCIM compatibility mode.
+
+    Password is not supported in this mode.
+    """
+
+    def to_dict(self):
+        return (
+            super()
+            .to_dict()
+            .update(
+                {
+                    "changePassword": {
+                        "supported": False,
+                    },
+                }
+            )
+        )

src/backend/scim_importer/tests/__init__.py

@@ -0,0 +1,3 @@
+"""Tests for the SCIM importer module."""
+# XXX: implement tests for the SCIM importer module,
+#      probably using https://verify.scim.dev/ requests as a base

src/backend/scim_importer/utils.py

@@ -0,0 +1,45 @@
+from django.apps import apps as django_apps
+from django.contrib.auth import get_user_model
+
+
+def get_scim_importer_user_model():
+    """
+    Return the user model that is used for SCIM import.
+    """
+    return django_apps.get_model("scim_importer.ScimImportedUser", require_ready=False)
+
+
+def default_get_extra_model_filter_kwargs_getter(model):
+    """
+    Return a **method** that will return extra model filter kwargs for the passed in model.
+
+    :param model:
+    """
+
+    if model is get_user_model():
+
+        def get_extra_filter_kwargs(request, *args, **kwargs):
+            """
+            Return extra filter kwargs for the given model.
+            :param request:
+            :param args:
+            :param kwargs:
+            :rtype: dict
+            """
+            return {
+                "scim_id__isnull": False,
+            }
+
+        return get_extra_filter_kwargs
+
+    def get_extra_filter_kwargs(request, *args, **kwargs):
+        """
+        Return extra filter kwargs for the given model.
+        :param request:
+        :param args:
+        :param kwargs:
+        :rtype: dict
+        """
+        return {}
+
+    return get_extra_filter_kwargs