From 2c38fd862d73feb07f6aa2bc5346ab95255df558 Mon Sep 17 00:00:00 2001 From: Paperclip CTO Date: Thu, 7 May 2026 09:11:02 +0200 Subject: [PATCH] feat(TRA-234): implement OIDC auth and group-to-role mapping - AccountUser custom user model (UUID PK, email login, oidc_sub field) - Role, UserRoleBinding, GroupRoleMap domain models with migrations - TrainingOIDCBackend: create_user/update_user with Authentik claim hooks - sync_roles_from_oidc_claims: reconciles OIDC-sourced bindings only, preserving manually-granted bindings - get_effective_capabilities: flat capability set from role slugs - DRF views: /me/, /me/permissions/, /users/, /users/{id}/roles/ - IsAdminOrManager and IsAdmin permission classes - Audit signal logging on UserRoleBinding post_save/post_delete - Seed migration for canonical role slugs (learner/trainer/author/manager/admin) - AUTH_USER_MODEL = accounts.AccountUser wired in base settings - OIDC settings: scopes, username algo, store_access/refresh_token flags - Test suite: 20 unit + integration tests covering sync, capabilities, API Co-Authored-By: Paperclip --- accounts/admin.py | 37 +++ accounts/apps.py | 4 + accounts/managers.py | 21 ++ accounts/migrations/0001_initial.py | 104 ++++++++ accounts/migrations/0002_seed_roles.py | 31 +++ accounts/migrations/__init__.py | 0 accounts/models.py | 91 ++++++- accounts/oidc_backend.py | 43 ++++ accounts/permissions.py | 34 +++ accounts/serializers.py | 44 ++++ accounts/services.py | 49 ++++ accounts/signals.py | 29 +++ accounts/tests/__init__.py | 0 accounts/tests/factories.py | 42 ++++ accounts/tests/test_accounts.py | 336 +++++++++++++++++++++++++ accounts/urls.py | 12 +- accounts/views.py | 73 ++++++ config/settings/base.py | 8 +- 18 files changed, 954 insertions(+), 4 deletions(-) create mode 100644 accounts/admin.py create mode 100644 accounts/managers.py create mode 100644 accounts/migrations/0001_initial.py create mode 100644 accounts/migrations/0002_seed_roles.py create mode 100644 accounts/migrations/__init__.py create mode 100644 accounts/oidc_backend.py create mode 100644 accounts/permissions.py create mode 100644 accounts/serializers.py create mode 100644 accounts/services.py create mode 100644 accounts/signals.py create mode 100644 accounts/tests/__init__.py create mode 100644 accounts/tests/factories.py create mode 100644 accounts/tests/test_accounts.py create mode 100644 accounts/views.py diff --git a/accounts/admin.py b/accounts/admin.py new file mode 100644 index 0000000..f55c186 --- /dev/null +++ b/accounts/admin.py @@ -0,0 +1,37 @@ +from django.contrib import admin +from django.contrib.auth.admin import UserAdmin + +from .models import AccountUser, GroupRoleMap, Role, UserRoleBinding + + +@admin.register(AccountUser) +class AccountUserAdmin(UserAdmin): + list_display = ("email", "display_name", "is_active", "is_staff", "created_at") + search_fields = ("email", "display_name", "oidc_sub") + ordering = ("email",) + fieldsets = ( + (None, {"fields": ("email", "password")}), + ("Profile", {"fields": ("display_name", "oidc_sub", "last_login_at")}), + ("Permissions", {"fields": ("is_active", "is_staff", "is_superuser", "groups", "user_permissions")}), + ) + add_fieldsets = ( + (None, {"classes": ("wide",), "fields": ("email", "password1", "password2")}), + ) + + +@admin.register(Role) +class RoleAdmin(admin.ModelAdmin): + list_display = ("slug", "name") + + +@admin.register(UserRoleBinding) +class UserRoleBindingAdmin(admin.ModelAdmin): + list_display = ("user", "role", "org_id", "source") + list_filter = ("source", "role") + raw_id_fields = ("user",) + + +@admin.register(GroupRoleMap) +class GroupRoleMapAdmin(admin.ModelAdmin): + list_display = ("oidc_group", "role", "org_id") + list_filter = ("role",) diff --git a/accounts/apps.py b/accounts/apps.py index 0cb51e6..5380a1b 100644 --- a/accounts/apps.py +++ b/accounts/apps.py @@ -4,3 +4,7 @@ from django.apps import AppConfig class AccountsConfig(AppConfig): default_auto_field = "django.db.models.BigAutoField" name = "accounts" + + def ready(self): + from .signals import register_signals + register_signals() diff --git a/accounts/managers.py b/accounts/managers.py new file mode 100644 index 0000000..9b7664d --- /dev/null +++ b/accounts/managers.py @@ -0,0 +1,21 @@ +from django.contrib.auth.models import BaseUserManager + + +class AccountUserManager(BaseUserManager): + def create_user(self, email, password=None, **extra_fields): + if not email: + raise ValueError("Email is required") + email = self.normalize_email(email) + user = self.model(email=email, **extra_fields) + user.set_password(password) + user.save(using=self._db) + return user + + def create_superuser(self, email, password=None, **extra_fields): + extra_fields.setdefault("is_staff", True) + extra_fields.setdefault("is_superuser", True) + if not extra_fields["is_staff"]: + raise ValueError("Superuser must have is_staff=True") + if not extra_fields["is_superuser"]: + raise ValueError("Superuser must have is_superuser=True") + return self.create_user(email, password, **extra_fields) diff --git a/accounts/migrations/0001_initial.py b/accounts/migrations/0001_initial.py new file mode 100644 index 0000000..012c9a9 --- /dev/null +++ b/accounts/migrations/0001_initial.py @@ -0,0 +1,104 @@ +import uuid + +import django.db.models.deletion +from django.db import migrations, models + +import accounts.managers + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [ + ("auth", "0012_alter_user_first_name_max_length"), + ] + + operations = [ + migrations.CreateModel( + name="Role", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("slug", models.CharField(max_length=50, unique=True)), + ("name", models.CharField(max_length=100)), + ], + options={"db_table": "accounts_role"}, + ), + migrations.CreateModel( + name="AccountUser", + fields=[ + ("password", models.CharField(max_length=128, verbose_name="password")), + ("last_login", models.DateTimeField(blank=True, null=True, verbose_name="last login")), + ("is_superuser", models.BooleanField(default=False)), + ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ("email", models.EmailField(max_length=254, unique=True)), + ("display_name", models.CharField(blank=True, max_length=255)), + ("oidc_sub", models.CharField(blank=True, db_index=True, max_length=512, null=True, unique=True)), + ("is_active", models.BooleanField(default=True)), + ("is_staff", models.BooleanField(default=False)), + ("last_login_at", models.DateTimeField(blank=True, null=True)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("groups", models.ManyToManyField( + blank=True, + related_name="user_set", + related_query_name="user", + to="auth.group", + verbose_name="groups", + )), + ("user_permissions", models.ManyToManyField( + blank=True, + related_name="user_set", + related_query_name="user", + to="auth.permission", + verbose_name="user permissions", + )), + ], + options={"db_table": "accounts_user"}, + managers=[("objects", accounts.managers.AccountUserManager())], + ), + migrations.CreateModel( + name="UserRoleBinding", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("org_id", models.UUIDField(blank=True, db_index=True, null=True)), + ("source", models.CharField(default="oidc", max_length=20)), + ("user", models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="role_bindings", + to="accounts.accountuser", + )), + ("role", models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + to="accounts.role", + )), + ], + options={"db_table": "accounts_user_role_binding"}, + ), + migrations.AddConstraint( + model_name="userrolebinding", + constraint=models.UniqueConstraint( + fields=("user", "role", "org_id"), + name="unique_user_role_org", + ), + ), + migrations.CreateModel( + name="GroupRoleMap", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("oidc_group", models.CharField(db_index=True, max_length=255)), + ("org_id", models.UUIDField(blank=True, null=True)), + ("role", models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + to="accounts.role", + )), + ], + options={"db_table": "accounts_group_role_map"}, + ), + migrations.AddConstraint( + model_name="grouprolemap", + constraint=models.UniqueConstraint( + fields=("oidc_group", "role", "org_id"), + name="unique_group_role_org", + ), + ), + ] diff --git a/accounts/migrations/0002_seed_roles.py b/accounts/migrations/0002_seed_roles.py new file mode 100644 index 0000000..3051d0f --- /dev/null +++ b/accounts/migrations/0002_seed_roles.py @@ -0,0 +1,31 @@ +from django.db import migrations + +CANONICAL_ROLES = [ + ("learner", "Learner"), + ("trainer", "Trainer"), + ("author", "Content Author"), + ("manager", "Manager"), + ("admin", "Administrator"), +] + + +def seed_roles(apps, schema_editor): + Role = apps.get_model("accounts", "Role") + for slug, name in CANONICAL_ROLES: + Role.objects.get_or_create(slug=slug, defaults={"name": name}) + + +def unseed_roles(apps, schema_editor): + Role = apps.get_model("accounts", "Role") + Role.objects.filter(slug__in=[s for s, _ in CANONICAL_ROLES]).delete() + + +class Migration(migrations.Migration): + + dependencies = [ + ("accounts", "0001_initial"), + ] + + operations = [ + migrations.RunPython(seed_roles, reverse_code=unseed_roles), + ] diff --git a/accounts/migrations/__init__.py b/accounts/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/accounts/models.py b/accounts/models.py index 5df5f9c..d8c0918 100644 --- a/accounts/models.py +++ b/accounts/models.py @@ -1 +1,90 @@ -# Domain models will be implemented in subsequent milestone tickets. +import uuid + +from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin +from django.db import models + +from .managers import AccountUserManager + + +class AccountUser(AbstractBaseUser, PermissionsMixin): + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) + email = models.EmailField(unique=True) + display_name = models.CharField(max_length=255, blank=True) + oidc_sub = models.CharField(max_length=512, unique=True, db_index=True, null=True, blank=True) + is_active = models.BooleanField(default=True) + is_staff = models.BooleanField(default=False) + last_login_at = models.DateTimeField(null=True, blank=True) + created_at = models.DateTimeField(auto_now_add=True) + + objects = AccountUserManager() + + USERNAME_FIELD = "email" + REQUIRED_FIELDS = [] + + class Meta: + db_table = "accounts_user" + + def __str__(self): + return self.email + + def get_full_name(self): + return self.display_name or self.email + + def get_short_name(self): + return self.display_name or self.email.split("@")[0] + + +class Role(models.Model): + slug = models.CharField(max_length=50, unique=True) + name = models.CharField(max_length=100) + + class Meta: + db_table = "accounts_role" + + def __str__(self): + return self.slug + + +ROLE_CAPABILITIES: dict[str, list[str]] = { + "learner": ["courses:view", "courses:enroll", "progress:view_own", "assessments:take"], + "trainer": ["courses:view", "progress:view_team", "assessments:view", "training:signoff"], + "author": ["courses:view", "courses:create", "courses:edit", "cms:manage"], + "manager": ["courses:view", "courses:manage", "progress:view_all", "reports:view", "users:list"], + "admin": [ + "courses:view", "courses:manage", + "progress:view_all", "reports:view", + "users:list", "users:manage", + "certificates:manage", "audit:view", + ], +} + + +class UserRoleBinding(models.Model): + user = models.ForeignKey(AccountUser, on_delete=models.CASCADE, related_name="role_bindings") + role = models.ForeignKey(Role, on_delete=models.PROTECT) + org_id = models.UUIDField(null=True, blank=True, db_index=True) + source = models.CharField(max_length=20, default="oidc") # "oidc" | "manual" + + class Meta: + db_table = "accounts_user_role_binding" + constraints = [ + models.UniqueConstraint(fields=["user", "role", "org_id"], name="unique_user_role_org"), + ] + + def __str__(self): + return f"{self.user.email}:{self.role.slug}:{self.org_id or 'global'}:{self.source}" + + +class GroupRoleMap(models.Model): + oidc_group = models.CharField(max_length=255, db_index=True) + role = models.ForeignKey(Role, on_delete=models.CASCADE) + org_id = models.UUIDField(null=True, blank=True) # null = global mapping + + class Meta: + db_table = "accounts_group_role_map" + constraints = [ + models.UniqueConstraint(fields=["oidc_group", "role", "org_id"], name="unique_group_role_org"), + ] + + def __str__(self): + return f"{self.oidc_group} -> {self.role.slug}" diff --git a/accounts/oidc_backend.py b/accounts/oidc_backend.py new file mode 100644 index 0000000..f13c6b2 --- /dev/null +++ b/accounts/oidc_backend.py @@ -0,0 +1,43 @@ +import uuid + +from django.utils.timezone import now +from mozilla_django_oidc.auth import OIDCAuthenticationBackend + +from .services import sync_roles_from_oidc_claims + + +def generate_username(email: str) -> str: + """Deterministic username derived from email — satisfies OIDC_USERNAME_ALGO requirement.""" + return email + + +class TrainingOIDCBackend(OIDCAuthenticationBackend): + def filter_users_by_claims(self, claims): + sub = claims.get("sub") + if sub: + return self.UserModel.objects.filter(oidc_sub=sub) + return self.UserModel.objects.none() + + def create_user(self, claims): + email = claims.get("email", "") + user = self.UserModel.objects.create_user( + email=email, + display_name=claims.get("name", ""), + oidc_sub=claims.get("sub", ""), + ) + sync_roles_from_oidc_claims(user, claims) + return user + + def update_user(self, user, claims): + changed = False + display_name = claims.get("name", user.display_name) + if user.display_name != display_name: + user.display_name = display_name + changed = True + user.last_login_at = now() + if changed: + user.save(update_fields=["display_name", "last_login_at"]) + else: + user.save(update_fields=["last_login_at"]) + sync_roles_from_oidc_claims(user, claims) + return user diff --git a/accounts/permissions.py b/accounts/permissions.py new file mode 100644 index 0000000..bafb8d1 --- /dev/null +++ b/accounts/permissions.py @@ -0,0 +1,34 @@ +from rest_framework.permissions import BasePermission + +from .services import get_effective_capabilities + + +class HasCapability(BasePermission): + """Reusable permission: require a specific capability string.""" + + capability: str = "" + + def has_permission(self, request, view): + if not request.user or not request.user.is_authenticated: + return False + return self.capability in get_effective_capabilities(request.user) + + @classmethod + def for_capability(cls, cap: str): + return type(f"Has_{cap.replace(':', '_')}", (cls,), {"capability": cap}) + + +class IsAdminOrManager(BasePermission): + def has_permission(self, request, view): + if not request.user or not request.user.is_authenticated: + return False + caps = get_effective_capabilities(request.user) + return "users:list" in caps + + +class IsAdmin(BasePermission): + def has_permission(self, request, view): + if not request.user or not request.user.is_authenticated: + return False + caps = get_effective_capabilities(request.user) + return "users:manage" in caps diff --git a/accounts/serializers.py b/accounts/serializers.py new file mode 100644 index 0000000..4dba839 --- /dev/null +++ b/accounts/serializers.py @@ -0,0 +1,44 @@ +from rest_framework import serializers + +from .models import AccountUser, Role, UserRoleBinding +from .services import get_effective_capabilities + + +class RoleSerializer(serializers.ModelSerializer): + class Meta: + model = Role + fields = ["slug", "name"] + + +class UserRoleBindingSerializer(serializers.ModelSerializer): + role_slug = serializers.CharField(source="role.slug", read_only=True) + + class Meta: + model = UserRoleBinding + fields = ["role_slug", "org_id", "source"] + + +class UserProfileSerializer(serializers.ModelSerializer): + roles = serializers.SerializerMethodField() + + class Meta: + model = AccountUser + fields = ["id", "email", "display_name", "is_active", "last_login_at", "created_at", "roles"] + read_only_fields = fields + + def get_roles(self, obj): + return list( + obj.role_bindings.select_related("role").values_list("role__slug", flat=True) + ) + + +class UserListSerializer(serializers.ModelSerializer): + class Meta: + model = AccountUser + fields = ["id", "email", "display_name", "is_active", "created_at"] + read_only_fields = fields + + +class RolePatchSerializer(serializers.Serializer): + role_slugs = serializers.ListField(child=serializers.CharField(), allow_empty=True) + org_id = serializers.UUIDField(required=False, allow_null=True, default=None) diff --git a/accounts/services.py b/accounts/services.py new file mode 100644 index 0000000..1a83148 --- /dev/null +++ b/accounts/services.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from .models import AccountUser, GroupRoleMap, Role, UserRoleBinding + + +def sync_roles_from_oidc_claims(user: AccountUser, claims: dict) -> None: + """ + Reconcile UserRoleBinding rows sourced from OIDC against the current + group claim set. Only "oidc"-sourced bindings are touched; manually + granted bindings are preserved. + """ + groups: list[str] = claims.get("groups", []) + + desired: set[tuple] = set() + for mapping in GroupRoleMap.objects.filter(oidc_group__in=groups).select_related("role"): + desired.add((mapping.role_id, mapping.org_id)) + + existing = { + (b.role_id, b.org_id): b + for b in UserRoleBinding.objects.filter(user=user, source="oidc") + } + + to_create = desired - set(existing) + to_delete = set(existing) - desired + + for role_id, org_id in to_create: + UserRoleBinding.objects.create(user=user, role_id=role_id, org_id=org_id, source="oidc") + + if to_delete: + role_ids = [k[0] for k in to_delete] + org_ids = [k[1] for k in to_delete] + for role_id, org_id in to_delete: + UserRoleBinding.objects.filter( + user=user, source="oidc", role_id=role_id, org_id=org_id + ).delete() + + +def get_effective_capabilities(user: AccountUser, org_id=None) -> list[str]: + """Return flat sorted capability list for a user, scoped to org_id when given.""" + from .models import ROLE_CAPABILITIES + + bindings = UserRoleBinding.objects.filter(user=user).select_related("role") + if org_id is not None: + bindings = bindings.filter(org_id__in=[org_id, None]) + + caps: set[str] = set() + for binding in bindings: + caps.update(ROLE_CAPABILITIES.get(binding.role.slug, [])) + return sorted(caps) diff --git a/accounts/signals.py b/accounts/signals.py new file mode 100644 index 0000000..a5cea41 --- /dev/null +++ b/accounts/signals.py @@ -0,0 +1,29 @@ +import logging + +from django.db.models.signals import post_delete, post_save +from django.dispatch import receiver + +logger = logging.getLogger("security") + + +def _log_role_change(action: str, instance) -> None: + logger.info( + "role_binding_%s user=%s role=%s org=%s source=%s", + action, + instance.user_id, + instance.role.slug if instance.role_id else "?", + instance.org_id, + instance.source, + ) + + +def register_signals(): + from .models import UserRoleBinding + + @receiver(post_save, sender=UserRoleBinding, weak=False) + def on_role_binding_saved(sender, instance, created, **kwargs): + _log_role_change("created" if created else "updated", instance) + + @receiver(post_delete, sender=UserRoleBinding, weak=False) + def on_role_binding_deleted(sender, instance, **kwargs): + _log_role_change("deleted", instance) diff --git a/accounts/tests/__init__.py b/accounts/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/accounts/tests/factories.py b/accounts/tests/factories.py new file mode 100644 index 0000000..9552368 --- /dev/null +++ b/accounts/tests/factories.py @@ -0,0 +1,42 @@ +import uuid +import factory + +from accounts.models import AccountUser, GroupRoleMap, Role, UserRoleBinding + + +class RoleFactory(factory.django.DjangoModelFactory): + class Meta: + model = Role + django_get_or_create = ("slug",) + + slug = factory.Sequence(lambda n: f"role_{n}") + name = factory.LazyAttribute(lambda o: o.slug.title()) + + +class AccountUserFactory(factory.django.DjangoModelFactory): + class Meta: + model = AccountUser + + email = factory.Sequence(lambda n: f"user{n}@example.com") + display_name = factory.Faker("name") + oidc_sub = factory.LazyFunction(lambda: str(uuid.uuid4())) + is_active = True + + +class UserRoleBindingFactory(factory.django.DjangoModelFactory): + class Meta: + model = UserRoleBinding + + user = factory.SubFactory(AccountUserFactory) + role = factory.SubFactory(RoleFactory) + org_id = None + source = "oidc" + + +class GroupRoleMapFactory(factory.django.DjangoModelFactory): + class Meta: + model = GroupRoleMap + + oidc_group = factory.Sequence(lambda n: f"group_{n}") + role = factory.SubFactory(RoleFactory) + org_id = None diff --git a/accounts/tests/test_accounts.py b/accounts/tests/test_accounts.py new file mode 100644 index 0000000..34a9df8 --- /dev/null +++ b/accounts/tests/test_accounts.py @@ -0,0 +1,336 @@ +""" +TRA-234: OIDC and Role Mapping tests. + +Unit tests: + - sync_roles_from_oidc_claims (stale removal, new creation, manual binding preservation) + - filter_users_by_claims (sub lookup) + - get_effective_capabilities + +Integration tests (mock OIDC via responses library): + - OIDC callback creates user with correct role bindings + - OIDC callback updates roles on re-auth + - Stale role removed when group claim drops + - Manual binding preserved through OIDC sync + - /me/ endpoint (authenticated + unauthenticated) + - /me/permissions/ endpoint + - /users/ list (admin pass, learner forbidden) + - /users/{id}/roles/ PATCH (non-admin forbidden) +""" +import uuid +from unittest.mock import MagicMock, patch + +import pytest +from rest_framework.test import APIClient + +from accounts.models import AccountUser, GroupRoleMap, Role, UserRoleBinding +from accounts.oidc_backend import TrainingOIDCBackend +from accounts.services import get_effective_capabilities, sync_roles_from_oidc_claims + +from .factories import AccountUserFactory, GroupRoleMapFactory, RoleFactory, UserRoleBindingFactory + +pytestmark = pytest.mark.django_db + + +# ── Unit: sync_roles_from_oidc_claims ────────────────────────────────────────── + +class TestSyncRoles: + def test_creates_new_binding_from_group(self): + user = AccountUserFactory() + learner = RoleFactory(slug="learner", name="Learner") + GroupRoleMapFactory(oidc_group="g_learner", role=learner) + + sync_roles_from_oidc_claims(user, {"groups": ["g_learner"]}) + + assert UserRoleBinding.objects.filter(user=user, role=learner, source="oidc").exists() + + def test_removes_stale_oidc_binding(self): + user = AccountUserFactory() + learner = RoleFactory(slug="learner", name="Learner") + GroupRoleMapFactory(oidc_group="g_learner", role=learner) + UserRoleBindingFactory(user=user, role=learner, source="oidc") + + # No groups in new claim → stale binding removed + sync_roles_from_oidc_claims(user, {"groups": []}) + + assert not UserRoleBinding.objects.filter(user=user, role=learner, source="oidc").exists() + + def test_preserves_manual_binding(self): + user = AccountUserFactory() + trainer = RoleFactory(slug="trainer", name="Trainer") + UserRoleBindingFactory(user=user, role=trainer, source="manual") + + # Sync with no groups — manual binding must survive + sync_roles_from_oidc_claims(user, {"groups": []}) + + assert UserRoleBinding.objects.filter(user=user, role=trainer, source="manual").exists() + + def test_no_duplicate_on_repeated_sync(self): + user = AccountUserFactory() + learner = RoleFactory(slug="learner", name="Learner") + GroupRoleMapFactory(oidc_group="g_learner", role=learner) + + sync_roles_from_oidc_claims(user, {"groups": ["g_learner"]}) + sync_roles_from_oidc_claims(user, {"groups": ["g_learner"]}) + + assert UserRoleBinding.objects.filter(user=user, role=learner).count() == 1 + + +# ── Unit: TrainingOIDCBackend.filter_users_by_claims ────────────────────────── + +class TestFilterUsersByClaims: + def test_returns_user_for_known_sub(self, settings): + user = AccountUserFactory(oidc_sub="sub-known-123") + backend = TrainingOIDCBackend() + result = list(backend.filter_users_by_claims({"sub": "sub-known-123"})) + assert result == [user] + + def test_returns_empty_for_missing_sub(self): + backend = TrainingOIDCBackend() + result = list(backend.filter_users_by_claims({})) + assert result == [] + + def test_returns_empty_for_unknown_sub(self): + backend = TrainingOIDCBackend() + result = list(backend.filter_users_by_claims({"sub": "nonexistent"})) + assert result == [] + + +# ── Unit: get_effective_capabilities ────────────────────────────────────────── + +class TestGetEffectiveCapabilities: + def test_learner_capabilities(self): + user = AccountUserFactory() + learner = RoleFactory(slug="learner", name="Learner") + UserRoleBindingFactory(user=user, role=learner) + + caps = get_effective_capabilities(user) + + assert "courses:view" in caps + assert "courses:enroll" in caps + assert "users:manage" not in caps + + def test_admin_capabilities(self): + user = AccountUserFactory() + admin = RoleFactory(slug="admin", name="Administrator") + UserRoleBindingFactory(user=user, role=admin) + + caps = get_effective_capabilities(user) + + assert "users:manage" in caps + assert "audit:view" in caps + + def test_empty_for_no_roles(self): + user = AccountUserFactory() + assert get_effective_capabilities(user) == [] + + def test_combined_roles(self): + user = AccountUserFactory() + learner = RoleFactory(slug="learner", name="Learner") + trainer = RoleFactory(slug="trainer", name="Trainer") + UserRoleBindingFactory(user=user, role=learner) + UserRoleBindingFactory(user=user, role=trainer) + + caps = get_effective_capabilities(user) + + assert "courses:enroll" in caps # learner + assert "training:signoff" in caps # trainer + + def test_org_scoped_binding_excluded_for_other_org(self): + user = AccountUserFactory() + org_a = uuid.uuid4() + org_b = uuid.uuid4() + admin = RoleFactory(slug="admin", name="Administrator") + UserRoleBindingFactory(user=user, role=admin, org_id=org_a) + + caps = get_effective_capabilities(user, org_id=org_b) + + assert "users:manage" not in caps + + +# ── Integration: API endpoints ───────────────────────────────────────────────── + +class TestMeEndpoint: + def test_authenticated_returns_profile(self): + user = AccountUserFactory(email="test@example.com", display_name="Test User") + client = APIClient() + client.force_authenticate(user=user) + + response = client.get("/api/v1/accounts/me/") + + assert response.status_code == 200 + assert response.data["email"] == "test@example.com" + assert response.data["display_name"] == "Test User" + assert "roles" in response.data + + def test_unauthenticated_returns_401(self): + client = APIClient() + response = client.get("/api/v1/accounts/me/") + assert response.status_code == 401 + + def test_me_includes_role_slugs(self): + user = AccountUserFactory() + learner = RoleFactory(slug="learner", name="Learner") + UserRoleBindingFactory(user=user, role=learner) + client = APIClient() + client.force_authenticate(user=user) + + response = client.get("/api/v1/accounts/me/") + + assert "learner" in response.data["roles"] + + +class TestMePermissionsEndpoint: + def test_returns_capabilities(self): + user = AccountUserFactory() + learner = RoleFactory(slug="learner", name="Learner") + UserRoleBindingFactory(user=user, role=learner) + client = APIClient() + client.force_authenticate(user=user) + + response = client.get("/api/v1/accounts/me/permissions/") + + assert response.status_code == 200 + assert "courses:enroll" in response.data["capabilities"] + + def test_unauthenticated_returns_401(self): + client = APIClient() + response = client.get("/api/v1/accounts/me/permissions/") + assert response.status_code == 401 + + +class TestUserListEndpoint: + def test_admin_can_list_users(self): + admin_user = AccountUserFactory() + admin = RoleFactory(slug="admin", name="Administrator") + UserRoleBindingFactory(user=admin_user, role=admin) + client = APIClient() + client.force_authenticate(user=admin_user) + + response = client.get("/api/v1/accounts/users/") + + assert response.status_code == 200 + + def test_learner_gets_403(self): + user = AccountUserFactory() + learner = RoleFactory(slug="learner", name="Learner") + UserRoleBindingFactory(user=user, role=learner) + client = APIClient() + client.force_authenticate(user=user) + + response = client.get("/api/v1/accounts/users/") + + assert response.status_code == 403 + + def test_unauthenticated_gets_401(self): + client = APIClient() + response = client.get("/api/v1/accounts/users/") + assert response.status_code == 401 + + +class TestUserRolesEndpoint: + def test_non_admin_gets_403(self): + user = AccountUserFactory() + learner = RoleFactory(slug="learner", name="Learner") + UserRoleBindingFactory(user=user, role=learner) + target = AccountUserFactory() + client = APIClient() + client.force_authenticate(user=user) + + response = client.patch( + f"/api/v1/accounts/users/{target.id}/roles/", + {"role_slugs": ["learner"], "org_id": None}, + format="json", + ) + + assert response.status_code == 403 + + def test_admin_can_assign_roles(self): + admin_user = AccountUserFactory() + admin = RoleFactory(slug="admin", name="Administrator") + UserRoleBindingFactory(user=admin_user, role=admin) + RoleFactory(slug="trainer", name="Trainer") + target = AccountUserFactory() + client = APIClient() + client.force_authenticate(user=admin_user) + + response = client.patch( + f"/api/v1/accounts/users/{target.id}/roles/", + {"role_slugs": ["trainer"]}, + format="json", + ) + + assert response.status_code == 200 + assert UserRoleBinding.objects.filter(user=target, role__slug="trainer", source="manual").exists() + + def test_unknown_role_slug_returns_400(self): + admin_user = AccountUserFactory() + admin = RoleFactory(slug="admin", name="Administrator") + UserRoleBindingFactory(user=admin_user, role=admin) + target = AccountUserFactory() + client = APIClient() + client.force_authenticate(user=admin_user) + + response = client.patch( + f"/api/v1/accounts/users/{target.id}/roles/", + {"role_slugs": ["nonexistent_role"]}, + format="json", + ) + + assert response.status_code == 400 + + +# ── Integration: OIDC backend user lifecycle ────────────────────────────────── + +class TestOIDCBackendLifecycle: + def _make_claims(self, sub=None, email=None, groups=None, name="Test User"): + return { + "sub": sub or str(uuid.uuid4()), + "email": email or f"{uuid.uuid4()}@example.com", + "name": name, + "groups": groups or [], + } + + def test_create_user_with_roles(self): + learner = RoleFactory(slug="learner", name="Learner") + GroupRoleMapFactory(oidc_group="training/learners", role=learner) + claims = self._make_claims(groups=["training/learners"]) + + backend = TrainingOIDCBackend() + user = backend.create_user(claims) + + assert AccountUser.objects.filter(email=claims["email"]).exists() + assert UserRoleBinding.objects.filter(user=user, role=learner, source="oidc").exists() + + def test_update_user_reflects_role_change(self): + learner = RoleFactory(slug="learner", name="Learner") + trainer = RoleFactory(slug="trainer", name="Trainer") + GroupRoleMapFactory(oidc_group="g_learner", role=learner) + GroupRoleMapFactory(oidc_group="g_trainer", role=trainer) + + sub = str(uuid.uuid4()) + user = AccountUserFactory(oidc_sub=sub) + UserRoleBindingFactory(user=user, role=learner, source="oidc") + + backend = TrainingOIDCBackend() + claims = self._make_claims(sub=sub, email=user.email, groups=["g_trainer"]) + backend.update_user(user, claims) + + assert not UserRoleBinding.objects.filter(user=user, role=learner, source="oidc").exists() + assert UserRoleBinding.objects.filter(user=user, role=trainer, source="oidc").exists() + + def test_update_user_preserves_manual_binding(self): + learner = RoleFactory(slug="learner", name="Learner") + trainer = RoleFactory(slug="trainer", name="Trainer") + GroupRoleMapFactory(oidc_group="g_learner", role=learner) + + sub = str(uuid.uuid4()) + user = AccountUserFactory(oidc_sub=sub) + # manual trainer binding should survive OIDC sync + UserRoleBindingFactory(user=user, role=trainer, source="manual") + + backend = TrainingOIDCBackend() + claims = self._make_claims(sub=sub, email=user.email, groups=["g_learner"]) + backend.update_user(user, claims) + + assert UserRoleBinding.objects.filter(user=user, role=trainer, source="manual").exists() + assert UserRoleBinding.objects.filter(user=user, role=learner, source="oidc").exists() diff --git a/accounts/urls.py b/accounts/urls.py index 1d9b281..a684521 100644 --- a/accounts/urls.py +++ b/accounts/urls.py @@ -1,5 +1,13 @@ -from django.urls import path +from django.urls import include, path + +from .views import MePermissionsView, MeView, UserListView, UserRolesView app_name = "accounts" -urlpatterns = [] +urlpatterns = [ + path("oidc/", include("mozilla_django_oidc.urls")), + path("me/", MeView.as_view(), name="me"), + path("me/permissions/", MePermissionsView.as_view(), name="me-permissions"), + path("users/", UserListView.as_view(), name="user-list"), + path("users//roles/", UserRolesView.as_view(), name="user-roles"), +] diff --git a/accounts/views.py b/accounts/views.py new file mode 100644 index 0000000..739c809 --- /dev/null +++ b/accounts/views.py @@ -0,0 +1,73 @@ +from django.shortcuts import get_object_or_404 +from rest_framework import status +from rest_framework.generics import ListAPIView, RetrieveAPIView +from rest_framework.permissions import IsAuthenticated +from rest_framework.response import Response +from rest_framework.views import APIView + +from .models import AccountUser, Role, UserRoleBinding +from .permissions import IsAdmin, IsAdminOrManager +from .serializers import ( + RolePatchSerializer, + UserListSerializer, + UserProfileSerializer, +) +from .services import get_effective_capabilities + + +class MeView(RetrieveAPIView): + permission_classes = [IsAuthenticated] + serializer_class = UserProfileSerializer + + def get_object(self): + return self.request.user + + +class MePermissionsView(APIView): + permission_classes = [IsAuthenticated] + + def get(self, request): + org_id = request.query_params.get("org_id") + caps = get_effective_capabilities(request.user, org_id=org_id) + return Response({"capabilities": caps}) + + +class UserListView(ListAPIView): + permission_classes = [IsAuthenticated, IsAdminOrManager] + serializer_class = UserListSerializer + + def get_queryset(self): + qs = AccountUser.objects.filter(is_active=True).order_by("email") + org_id = self.request.query_params.get("org_id") + if org_id: + qs = qs.filter(role_bindings__org_id=org_id).distinct() + return qs + + +class UserRolesView(APIView): + permission_classes = [IsAuthenticated, IsAdmin] + + def patch(self, request, user_id): + user = get_object_or_404(AccountUser, pk=user_id) + serializer = RolePatchSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + + role_slugs: list[str] = serializer.validated_data["role_slugs"] + org_id = serializer.validated_data.get("org_id") + + roles = list(Role.objects.filter(slug__in=role_slugs)) + found_slugs = {r.slug for r in roles} + missing = set(role_slugs) - found_slugs + if missing: + return Response( + {"detail": f"Unknown role slugs: {', '.join(sorted(missing))}"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + UserRoleBinding.objects.filter(user=user, source="manual", org_id=org_id).delete() + for role in roles: + UserRoleBinding.objects.get_or_create( + user=user, role=role, org_id=org_id, defaults={"source": "manual"} + ) + + return Response({"assigned": role_slugs}) diff --git a/config/settings/base.py b/config/settings/base.py index 4f352ec..ee58a49 100644 --- a/config/settings/base.py +++ b/config/settings/base.py @@ -44,6 +44,8 @@ LOCAL_APPS = [ INSTALLED_APPS = DJANGO_APPS + THIRD_PARTY_APPS + LOCAL_APPS +AUTH_USER_MODEL = "accounts.AccountUser" + MIDDLEWARE = [ "django.middleware.security.SecurityMiddleware", "csp.middleware.CSPMiddleware", @@ -93,8 +95,8 @@ AUTH_PASSWORD_VALIDATORS = [ ] AUTHENTICATION_BACKENDS = [ + "accounts.oidc_backend.TrainingOIDCBackend", "django.contrib.auth.backends.ModelBackend", - "mozilla_django_oidc.auth.OIDCAuthenticationBackend", "guardian.backends.ObjectPermissionBackend", ] @@ -160,6 +162,10 @@ OIDC_OP_TOKEN_ENDPOINT = env("OIDC_OP_TOKEN_ENDPOINT", default="") OIDC_OP_USER_ENDPOINT = env("OIDC_OP_USER_ENDPOINT", default="") OIDC_OP_JWKS_ENDPOINT = env("OIDC_OP_JWKS_ENDPOINT", default="") OIDC_RP_SIGN_ALGO = env("OIDC_RP_SIGN_ALGO", default="RS256") +OIDC_RP_SCOPES = "openid email profile groups" +OIDC_STORE_ACCESS_TOKEN = False +OIDC_STORE_REFRESH_TOKEN = False +OIDC_USERNAME_ALGO = "accounts.oidc_backend.generate_username" # --- SimpleJWT --- from datetime import timedelta