sources/scim: add managed_objects_only option for stricter provisioning

authentik/sources/scim/api/sources.py

@@ -42,6 +42,7 @@ class Meta:
             "meta_model_name",
             "managed",
             "user_path_template",
+            "managed_objects_only",
             "root_url",
             "token_obj",
         ]

authentik/sources/scim/managed.py

@@ -0,0 +1,56 @@
+"""Helpers for SCIM source managed-objects-only mode."""
+
+from django.db.models import QuerySet
+
+from authentik.core.models import Group, User
+from authentik.sources.scim.models import SCIMSource, SCIMSourceGroup, SCIMSourceUser
+from authentik.sources.scim.views.v2.exceptions import (
+    SCIMConflictError,
+    SCIMError,
+    SCIMErrorTypes,
+    SCIMValidationError,
+)
+
+
+def resolve_user(
+    source: SCIMSource, connection: SCIMSourceUser | None, username: str
+) -> User:
+    """Resolve the user to create or update for a SCIM request."""
+    if connection:
+        return connection.user
+    if source.managed_objects_only:
+        if User.objects.filter(username=username).exists():
+            raise SCIMConflictError("User with this userName already exists.")
+        return User()
+    if existing := User.objects.filter(username=username).first():
+        return existing
+    return User()
+
+
+def resolve_group(source: SCIMSource, connection: SCIMSourceGroup | None, name: str) -> Group:
+    """Resolve the group to create or update for a SCIM request."""
+    if connection:
+        return connection.group
+    if source.managed_objects_only:
+        if Group.objects.filter(name=name).exists():
+            raise SCIMConflictError("Group with this displayName already exists.")
+        return Group()
+    if existing := Group.objects.filter(name=name).first():
+        return existing
+    return Group()
+
+
+def filter_group_members(source: SCIMSource, users: QuerySet[User]) -> QuerySet[User]:
+    """Filter group members to users managed by this source when required."""
+    if not source.managed_objects_only:
+        return users
+    managed = users.filter(scimsourceuser__source=source).distinct()
+    if users.count() != managed.count():
+        raise SCIMValidationError(
+            SCIMError(
+                detail="One or more group members are not managed by this SCIM source.",
+                scimType=SCIMErrorTypes.invalid_value,
+                status=400,
+            )
+        )
+    return managed

authentik/sources/scim/migrations/0004_scimsource_managed_objects_only.py

@@ -0,0 +1,22 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("authentik_sources_scim", "0003_alter_scimsourcegroup_unique_together_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="scimsource",
+            name="managed_objects_only",
+            field=models.BooleanField(
+                default=False,
+                help_text=(
+                    "Only manage users and groups created by this SCIM source. "
+                    "Disables tenant-wide correlation, restricts group membership, "
+                    "and unlinks objects on DELETE instead of deleting them."
+                ),
+            ),
+        ),
+    ]

authentik/sources/scim/models.py

@@ -17,6 +17,14 @@ class SCIMSource(Source):
     cross-system user provisioning"""
 
     token = models.ForeignKey(Token, on_delete=models.CASCADE, null=True, default=None)
+    managed_objects_only = models.BooleanField(
+        default=False,
+        help_text=_(
+            "Only manage users and groups created by this SCIM source. "
+            "Disables tenant-wide correlation, restricts group membership, "
+            "and unlinks objects on DELETE instead of deleting them."
+        ),
+    )
 
     @property
     def service_account_identifier(self) -> str:

authentik/sources/scim/tests/test_managed_objects_only.py

@@ -0,0 +1,173 @@
+"""Tests for SCIM source managed-objects-only mode."""
+
+from json import dumps
+from uuid import uuid4
+
+from django.urls import reverse
+from rest_framework.test import APITestCase
+
+from authentik.core.models import Group, User
+from authentik.core.tests.utils import create_test_user
+from authentik.lib.generators import generate_id
+from authentik.sources.scim.models import SCIMSource, SCIMSourceGroup, SCIMSourceUser
+from authentik.sources.scim.views.v2.base import SCIM_CONTENT_TYPE
+
+
+class TestSCIMManagedObjectsOnly(APITestCase):
+    """Test SCIM source managed_objects_only behavior."""
+
+    def setUp(self) -> None:
+        self.source = SCIMSource.objects.create(
+            name=generate_id(),
+            slug=generate_id(),
+            managed_objects_only=True,
+        )
+
+    def test_user_create_conflict_existing_username(self):
+        """Cannot correlate to an existing user when managed_objects_only is enabled."""
+        existing = create_test_user()
+        response = self.client.post(
+            reverse(
+                "authentik_sources_scim:v2-users",
+                kwargs={"source_slug": self.source.slug},
+            ),
+            data=dumps({"userName": existing.username, "name": {"formatted": "Test"}}),
+            content_type=SCIM_CONTENT_TYPE,
+            HTTP_AUTHORIZATION=f"Bearer {self.source.token.key}",
+        )
+        self.assertEqual(response.status_code, 409)
+
+    def test_group_create_conflict_existing_name(self):
+        """Cannot correlate to an existing group when managed_objects_only is enabled."""
+        admin_group = Group.objects.create(name="authentik Admins", is_superuser=True)
+        response = self.client.post(
+            reverse(
+                "authentik_sources_scim:v2-groups",
+                kwargs={"source_slug": self.source.slug},
+            ),
+            data=dumps({"displayName": admin_group.name, "externalId": generate_id()}),
+            content_type=SCIM_CONTENT_TYPE,
+            HTTP_AUTHORIZATION=f"Bearer {self.source.token.key}",
+        )
+        self.assertEqual(response.status_code, 409)
+        self.assertFalse(SCIMSourceGroup.objects.filter(source=self.source, group=admin_group).exists())
+
+    def test_group_members_require_managed_users(self):
+        """Group membership is limited to users managed by this SCIM source."""
+        unmanaged = create_test_user()
+        scim_user = create_test_user()
+        SCIMSourceUser.objects.create(
+            source=self.source,
+            user=scim_user,
+            external_id=str(uuid4()),
+        )
+        response = self.client.post(
+            reverse(
+                "authentik_sources_scim:v2-groups",
+                kwargs={"source_slug": self.source.slug},
+            ),
+            data=dumps(
+                {
+                    "displayName": generate_id(),
+                    "externalId": generate_id(),
+                    "members": [{"value": str(unmanaged.uuid)}],
+                }
+            ),
+            content_type=SCIM_CONTENT_TYPE,
+            HTTP_AUTHORIZATION=f"Bearer {self.source.token.key}",
+        )
+        self.assertEqual(response.status_code, 400)
+
+    def test_escalation_to_admin_group_blocked(self):
+        """Full escalation path is blocked when managed_objects_only is enabled."""
+        admin_group = Group.objects.create(name="authentik Admins", is_superuser=True)
+        attacker_username = generate_id()
+        create_response = self.client.post(
+            reverse(
+                "authentik_sources_scim:v2-users",
+                kwargs={"source_slug": self.source.slug},
+            ),
+            data=dumps(
+                {
+                    "userName": attacker_username,
+                    "name": {"formatted": "Attacker"},
+                    "active": True,
+                }
+            ),
+            content_type=SCIM_CONTENT_TYPE,
+            HTTP_AUTHORIZATION=f"Bearer {self.source.token.key}",
+        )
+        self.assertEqual(create_response.status_code, 201)
+        attacker_uuid = create_response.json()["id"]
+
+        group_response = self.client.post(
+            reverse(
+                "authentik_sources_scim:v2-groups",
+                kwargs={"source_slug": self.source.slug},
+            ),
+            data=dumps(
+                {
+                    "displayName": admin_group.name,
+                    "externalId": generate_id(),
+                    "members": [{"value": attacker_uuid}],
+                }
+            ),
+            content_type=SCIM_CONTENT_TYPE,
+            HTTP_AUTHORIZATION=f"Bearer {self.source.token.key}",
+        )
+        self.assertEqual(group_response.status_code, 409)
+        attacker = User.objects.get(uuid=attacker_uuid)
+        self.assertFalse(attacker.is_superuser)
+        self.assertEqual(admin_group.users.count(), 0)
+
+    def test_delete_user_unlinks_only(self):
+        """DELETE removes the SCIM link but keeps the user when managed_objects_only is enabled."""
+        username = generate_id()
+        create_response = self.client.post(
+            reverse(
+                "authentik_sources_scim:v2-users",
+                kwargs={"source_slug": self.source.slug},
+            ),
+            data=dumps({"userName": username, "name": {"formatted": "Test"}}),
+            content_type=SCIM_CONTENT_TYPE,
+            HTTP_AUTHORIZATION=f"Bearer {self.source.token.key}",
+        )
+        self.assertEqual(create_response.status_code, 201)
+        user_id = create_response.json()["id"]
+        delete_response = self.client.delete(
+            reverse(
+                "authentik_sources_scim:v2-users",
+                kwargs={"source_slug": self.source.slug, "user_id": user_id},
+            ),
+            HTTP_AUTHORIZATION=f"Bearer {self.source.token.key}",
+        )
+        self.assertEqual(delete_response.status_code, 204)
+        self.assertTrue(User.objects.filter(username=username).exists())
+        self.assertFalse(SCIMSourceUser.objects.filter(source=self.source, user__uuid=user_id).exists())
+
+    def test_delete_group_unlinks_only(self):
+        """DELETE removes the SCIM link but keeps the group when managed_objects_only is enabled."""
+        name = generate_id()
+        create_response = self.client.post(
+            reverse(
+                "authentik_sources_scim:v2-groups",
+                kwargs={"source_slug": self.source.slug},
+            ),
+            data=dumps({"displayName": name, "externalId": generate_id()}),
+            content_type=SCIM_CONTENT_TYPE,
+            HTTP_AUTHORIZATION=f"Bearer {self.source.token.key}",
+        )
+        self.assertEqual(create_response.status_code, 201)
+        group_id = create_response.json()["id"]
+        delete_response = self.client.delete(
+            reverse(
+                "authentik_sources_scim:v2-groups",
+                kwargs={"source_slug": self.source.slug, "group_id": group_id},
+            ),
+            HTTP_AUTHORIZATION=f"Bearer {self.source.token.key}",
+        )
+        self.assertEqual(delete_response.status_code, 204)
+        self.assertTrue(Group.objects.filter(name=name).exists())
+        self.assertFalse(
+            SCIMSourceGroup.objects.filter(source=self.source, group__group_uuid=group_id).exists()
+        )

authentik/sources/scim/views/v2/groups.py

@@ -16,6 +16,7 @@
 from authentik.core.models import Group, User
 from authentik.providers.scim.clients.schema import SCIM_GROUP_SCHEMA, PatchOp, PatchOperation
 from authentik.providers.scim.clients.schema import Group as SCIMGroupModel
+from authentik.sources.scim.managed import filter_group_members, resolve_group
 from authentik.sources.scim.models import SCIMSourceGroup
 from authentik.sources.scim.patch.processor import SCIMPatchProcessor
 from authentik.sources.scim.views.v2.base import SCIMObjectView
@@ -94,9 +95,7 @@ def update_group(self, connection: SCIMSourceGroup | None, data: QueryDict, appl
         if not properties.get("name"):
             raise ValidationError("Invalid group")
 
-        group = connection.group if connection else Group()
-        if _group := Group.objects.filter(name=properties.get("name")).first():
-            group = _group
+        group = resolve_group(self.source, connection, properties.get("name"))
 
         group.update_attributes(properties)
 
@@ -110,7 +109,7 @@ def update_group(self, connection: SCIMSourceGroup | None, data: QueryDict, appl
                     continue
                 query |= Q(uuid=member.value)
             if query:
-                group.users.set(User.objects.filter(query))
+                group.users.set(filter_group_members(self.source, User.objects.filter(query)))
         data["members"] = self._convert_members(group)
         if not connection:
             connection, _ = SCIMSourceGroup.objects.update_or_create(
@@ -178,15 +177,19 @@ def patch(self, request: Request, group_id: str, **kwargs) -> Response:
                     for member in operation.value:
                         query |= Q(uuid=member["value"])
                     if query:
-                        connection.group.users.add(*User.objects.filter(query))
+                        connection.group.users.add(
+                            *filter_group_members(self.source, User.objects.filter(query))
+                        )
                 elif operation.op == PatchOp.remove:
                     if not isinstance(operation.value, list):
                         operation.value = [operation.value]
                     query = Q()
                     for member in operation.value:
                         query |= Q(uuid=member["value"])
                     if query:
-                        connection.group.users.remove(*User.objects.filter(query))
+                        connection.group.users.remove(
+                            *filter_group_members(self.source, User.objects.filter(query))
+                        )
         patcher = SCIMPatchProcessor()
         patched_data = patcher.apply_patches(
             connection.attributes, request.data.get("Operations", [])
@@ -204,6 +207,9 @@ def delete(self, request: Request, group_id: str, **kwargs) -> Response:
         ).first()
         if not connection:
             raise SCIMNotFoundError("Group not found.")
+        if self.source.managed_objects_only:
+            connection.delete()
+            return Response(status=204)
         connection.group.delete()
         connection.delete()
         return Response(status=204)

authentik/sources/scim/views/v2/users.py

@@ -14,6 +14,7 @@
 from authentik.core.models import User
 from authentik.providers.scim.clients.schema import SCIM_USER_SCHEMA, Email
 from authentik.providers.scim.clients.schema import User as SCIMUserModel
+from authentik.sources.scim.managed import filter_group_members, resolve_group, resolve_user
 from authentik.sources.scim.models import SCIMSourceUser
 from authentik.sources.scim.patch.processor import SCIMPatchProcessor
 from authentik.sources.scim.views.v2.base import SCIMObjectView
@@ -97,9 +98,7 @@ def update_user(self, connection: SCIMSourceUser | None, data: QueryDict):
         if not properties.get("username"):
             raise ValidationError("Invalid user")
 
-        user = connection.user if connection else User()
-        if _user := User.objects.filter(username=properties.get("username")).first():
-            user = _user
+        user = resolve_user(self.source, connection, properties.get("username"))
         user.update_attributes(properties)
 
         if not connection:
@@ -158,6 +157,9 @@ def delete(self, request: Request, user_id: str, **kwargs) -> Response:
         connection = SCIMSourceUser.objects.filter(source=self.source, user__uuid=user_id).first()
         if not connection:
             raise SCIMNotFoundError("User not found.")
+        if self.source.managed_objects_only:
+            connection.delete()
+            return Response(status=204)
         connection.user.delete()
         connection.delete()
         return Response(status=204)