electron-rare/people
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🚧(scim) add SCIM importer base mechanics

Browse Source 
This could allow a SCIM client to provision users and groups.
This commit is contained in:
Quentin BEY
2025-04-25 19:36:01 +02:00
parent 0a241f0e03
commit a6882a579e
14 changed files with 705 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/people/settings.py
+23
View File
@@ -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
+1
View File
@@ -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
+1
View File
@@ -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/__init__.py
View File
src/backend/scim_importer/adapters.py
+232
View File
@@ -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
+37
View File
@@ -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
+12
View File
@@ -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
+84
View File
@@ -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
+18
View File
@@ -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
+76
View File
@@ -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/migrations/__init__.py
View File
src/backend/scim_importer/models.py
+173
View File
@@ -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
+3
View File
@@ -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
+45
View File
@@ -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