mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-08-20 17:53:20 +00:00
Enhancement: sync OIDC groups to superuser and staff roles (#13060)
Co-authored-by: shamoon <4887959+shamoon@users.noreply.github.com> Co-authored-by: SoleroTG <github-29h@solero.quietmail.eu> Co-authored-by: stumpylog <797416+stumpylog@users.noreply.github.com>
This commit is contained in:
co-authored by
shamoon
SoleroTG
stumpylog
parent
fd3c525f03
commit
f1c8a72f26
@@ -344,6 +344,12 @@ SOCIAL_ACCOUNT_SYNC_GROUPS_CLAIM: Final[str] = os.getenv(
|
||||
"PAPERLESS_SOCIAL_ACCOUNT_SYNC_GROUPS_CLAIM",
|
||||
"groups",
|
||||
)
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP: Final[str | None] = os.getenv(
|
||||
"PAPERLESS_SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP",
|
||||
)
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP: Final[str | None] = os.getenv(
|
||||
"PAPERLESS_SOCIAL_ACCOUNT_SYNC_STAFF_GROUP",
|
||||
)
|
||||
|
||||
HEADLESS_TOKEN_STRATEGY = "paperless.adapter.DrfTokenStrategy"
|
||||
|
||||
|
||||
@@ -38,6 +38,18 @@ def handle_social_account_updated(sender, request, sociallogin, **kwargs):
|
||||
"""
|
||||
from django.contrib.auth.models import Group
|
||||
|
||||
if not sociallogin.user.is_active:
|
||||
# allauth looks up and updates the social account, firing this
|
||||
# signal, before checking if the user is allowed to actually log
|
||||
# in. Syncing groups/roles here would arm a deactivated account
|
||||
# with permissions it never exercised, which would silently take
|
||||
# effect if the account is later reactivated for an unrelated
|
||||
# reason.
|
||||
logger.debug(
|
||||
f"Skipping social account sync for inactive user `{sociallogin.user}`",
|
||||
)
|
||||
return
|
||||
|
||||
extra_data = sociallogin.account.extra_data or {}
|
||||
social_account_groups = extra_data.get(
|
||||
settings.SOCIAL_ACCOUNT_SYNC_GROUPS_CLAIM,
|
||||
@@ -61,3 +73,31 @@ def handle_social_account_updated(sender, request, sociallogin, **kwargs):
|
||||
f"Syncing groups for user `{sociallogin.user}`: {social_account_groups}",
|
||||
)
|
||||
sociallogin.user.groups.set(groups, clear=True)
|
||||
|
||||
modified_fields = []
|
||||
if settings.SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP:
|
||||
is_superuser = (
|
||||
settings.SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP in social_account_groups
|
||||
)
|
||||
if sociallogin.user.is_superuser != is_superuser:
|
||||
sociallogin.user.is_superuser = is_superuser
|
||||
modified_fields.append("is_superuser")
|
||||
|
||||
if settings.SOCIAL_ACCOUNT_SYNC_STAFF_GROUP:
|
||||
is_staff = (
|
||||
settings.SOCIAL_ACCOUNT_SYNC_STAFF_GROUP in social_account_groups
|
||||
) or sociallogin.user.is_superuser
|
||||
if sociallogin.user.is_staff != is_staff:
|
||||
sociallogin.user.is_staff = is_staff
|
||||
modified_fields.append("is_staff")
|
||||
elif settings.SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP:
|
||||
is_staff = sociallogin.user.is_superuser or sociallogin.user.is_staff
|
||||
if sociallogin.user.is_staff != is_staff:
|
||||
sociallogin.user.is_staff = is_staff
|
||||
modified_fields.append("is_staff")
|
||||
|
||||
if modified_fields:
|
||||
logger.debug(
|
||||
f"Syncing roles for user `{sociallogin.user}`: superuser={sociallogin.user.is_superuser}, staff={sociallogin.user.is_staff}",
|
||||
)
|
||||
sociallogin.user.save(update_fields=modified_fields)
|
||||
|
||||
@@ -163,6 +163,47 @@ class TestSyncSocialLoginGroups(TestCase):
|
||||
)
|
||||
self.assertEqual(list(user.groups.all()), [])
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_GROUPS=True,
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP="admin-group",
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP="staff-group",
|
||||
)
|
||||
def test_no_sync_for_inactive_user(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Enabled group, superuser, and staff syncing
|
||||
- A deactivated user with a matching social login
|
||||
WHEN:
|
||||
- The social login is updated via signal
|
||||
THEN:
|
||||
- Groups and roles are left untouched, since the login itself
|
||||
would be rejected for a deactivated user anyway
|
||||
"""
|
||||
Group.objects.create(name="admin-group")
|
||||
user = User.objects.create_user(
|
||||
username="inactive_user",
|
||||
is_active=False,
|
||||
is_superuser=False,
|
||||
is_staff=False,
|
||||
)
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(
|
||||
extra_data={
|
||||
"groups": ["admin-group", "staff-group"],
|
||||
},
|
||||
),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertEqual(list(user.groups.all()), [])
|
||||
self.assertFalse(user.is_superuser)
|
||||
self.assertFalse(user.is_staff)
|
||||
|
||||
@override_settings(SOCIAL_ACCOUNT_SYNC_GROUPS=True)
|
||||
def test_no_groups(self) -> None:
|
||||
"""
|
||||
@@ -254,6 +295,344 @@ class TestSyncSocialLoginGroups(TestCase):
|
||||
|
||||
self.assertEqual(list(user.groups.all()), [group])
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP="admin-group",
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP=None,
|
||||
)
|
||||
def test_sync_superuser_enabled(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Configured superuser group sync, and user with that group
|
||||
WHEN:
|
||||
- Social login updated via signal
|
||||
THEN:
|
||||
- User becomes superuser and staff
|
||||
"""
|
||||
user = User.objects.create_user(
|
||||
username="testuser_s_e",
|
||||
is_superuser=False,
|
||||
is_staff=False,
|
||||
)
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(
|
||||
extra_data={
|
||||
"groups": ["admin-group"],
|
||||
},
|
||||
),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertTrue(user.is_superuser)
|
||||
self.assertTrue(user.is_staff)
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP="admin-group",
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP=None,
|
||||
)
|
||||
def test_sync_superuser_disabled(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Configured superuser group sync, and user without that group
|
||||
WHEN:
|
||||
- Social login updated via signal
|
||||
THEN:
|
||||
- User loses superuser status but preserves staff status if they had it
|
||||
"""
|
||||
user = User.objects.create_user(
|
||||
username="testuser_s_d",
|
||||
is_superuser=True,
|
||||
is_staff=True,
|
||||
)
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(
|
||||
extra_data={
|
||||
"groups": ["other-group"],
|
||||
},
|
||||
),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertFalse(user.is_superuser)
|
||||
self.assertTrue(user.is_staff)
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP=None,
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP="staff-group",
|
||||
)
|
||||
def test_sync_staff_enabled(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Configured staff group sync, and user with that group
|
||||
WHEN:
|
||||
- Social login updated via signal
|
||||
THEN:
|
||||
- User becomes staff
|
||||
"""
|
||||
user = User.objects.create_user(
|
||||
username="testuser_st_e",
|
||||
is_superuser=False,
|
||||
is_staff=False,
|
||||
)
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(
|
||||
extra_data={
|
||||
"groups": ["staff-group"],
|
||||
},
|
||||
),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertTrue(user.is_staff)
|
||||
self.assertFalse(user.is_superuser)
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP=None,
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP="staff-group",
|
||||
)
|
||||
def test_sync_staff_disabled(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Configured staff group sync, and user without that group
|
||||
WHEN:
|
||||
- Social login updated via signal
|
||||
THEN:
|
||||
- User loses staff status
|
||||
"""
|
||||
user = User.objects.create_user(
|
||||
username="testuser_st_d",
|
||||
is_superuser=False,
|
||||
is_staff=True,
|
||||
)
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(
|
||||
extra_data={
|
||||
"groups": ["other-group"],
|
||||
},
|
||||
),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertFalse(user.is_staff)
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP="admin-group",
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP="staff-group",
|
||||
)
|
||||
def test_sync_both_groups(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Configured both superuser and staff group sync
|
||||
WHEN:
|
||||
- Social login updated via signal
|
||||
THEN:
|
||||
- Roles are correctly assigned/revoked according to groups
|
||||
"""
|
||||
# Case 1: has both
|
||||
user = User.objects.create_user(
|
||||
username="testuser_b_1",
|
||||
is_superuser=False,
|
||||
is_staff=False,
|
||||
)
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(extra_data={"groups": ["admin-group", "staff-group"]}),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertTrue(user.is_superuser)
|
||||
self.assertTrue(user.is_staff)
|
||||
|
||||
# Case 2: has only staff
|
||||
user2 = User.objects.create_user(
|
||||
username="testuser_b_2",
|
||||
is_superuser=True,
|
||||
is_staff=True,
|
||||
)
|
||||
sociallogin2 = Mock(
|
||||
user=user2,
|
||||
account=Mock(extra_data={"groups": ["staff-group"]}),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin2,
|
||||
)
|
||||
user2.refresh_from_db()
|
||||
self.assertFalse(user2.is_superuser)
|
||||
self.assertTrue(user2.is_staff)
|
||||
|
||||
# Case 3: has neither
|
||||
user3 = User.objects.create_user(
|
||||
username="testuser_b_3",
|
||||
is_superuser=True,
|
||||
is_staff=True,
|
||||
)
|
||||
sociallogin3 = Mock(
|
||||
user=user3,
|
||||
account=Mock(extra_data={"groups": ["other-group"]}),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin3,
|
||||
)
|
||||
user3.refresh_from_db()
|
||||
self.assertFalse(user3.is_superuser)
|
||||
self.assertFalse(user3.is_staff)
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP=None,
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP=None,
|
||||
)
|
||||
def test_no_sync_when_not_configured(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- No sync settings configured
|
||||
WHEN:
|
||||
- Social login updated via signal
|
||||
THEN:
|
||||
- Existing roles are not modified
|
||||
"""
|
||||
user = User.objects.create_user(
|
||||
username="testuser_n_s",
|
||||
is_superuser=True,
|
||||
is_staff=True,
|
||||
)
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(extra_data={"groups": ["admin-group", "staff-group"]}),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertTrue(user.is_superuser)
|
||||
self.assertTrue(user.is_staff)
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP="admin-group",
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP=None,
|
||||
)
|
||||
def test_sync_superuser_demotes_local_user_without_group(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Configured superuser group sync
|
||||
- User with a usable (local) password, but without the group
|
||||
WHEN:
|
||||
- Social login updated via signal
|
||||
THEN:
|
||||
- User's superuser status is demoted, matching the group claim exactly
|
||||
"""
|
||||
user = User.objects.create_user(
|
||||
username="local_admin",
|
||||
password="password123",
|
||||
is_superuser=True,
|
||||
is_staff=True,
|
||||
)
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(extra_data={"groups": ["other-group"]}),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertFalse(user.is_superuser)
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP="admin-group",
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP=None,
|
||||
)
|
||||
def test_sync_superuser_demotes_last_admin(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Configured superuser group sync
|
||||
- User without the group, and no other active superuser exists
|
||||
WHEN:
|
||||
- Social login updated via signal
|
||||
THEN:
|
||||
- User's superuser status is demoted, even though they are the last admin
|
||||
"""
|
||||
user = User.objects.create_user(
|
||||
username="last_admin",
|
||||
is_superuser=True,
|
||||
is_staff=True,
|
||||
)
|
||||
user.set_unusable_password()
|
||||
user.save()
|
||||
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(extra_data={"groups": ["other-group"]}),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertFalse(user.is_superuser)
|
||||
|
||||
@override_settings(
|
||||
SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP=None,
|
||||
SOCIAL_ACCOUNT_SYNC_STAFF_GROUP="staff-group",
|
||||
)
|
||||
def test_sync_staff_demotes_local_user_without_group(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Configured staff group sync
|
||||
- User with a usable (local) password, but without the group
|
||||
WHEN:
|
||||
- Social login updated via signal
|
||||
THEN:
|
||||
- User's staff status is demoted, matching the group claim exactly
|
||||
"""
|
||||
user = User.objects.create_user(
|
||||
username="local_staff",
|
||||
password="password123",
|
||||
is_superuser=False,
|
||||
is_staff=True,
|
||||
)
|
||||
sociallogin = Mock(
|
||||
user=user,
|
||||
account=Mock(extra_data={"groups": ["other-group"]}),
|
||||
)
|
||||
handle_social_account_updated(
|
||||
sender=None,
|
||||
request=HttpRequest(),
|
||||
sociallogin=sociallogin,
|
||||
)
|
||||
user.refresh_from_db()
|
||||
self.assertFalse(user.is_staff)
|
||||
|
||||
|
||||
class TestUserGroupDeletionCleanup(TestCase):
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user