Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions team/apps.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,3 +4,6 @@
class TeamConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "team"

def ready(self):
import team.signals # noqa F401
Empty file added team/management/__init__.py
Empty file.
Empty file.
53 changes: 53 additions & 0 deletions team/management/commands/create_user_groups.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
from django.contrib.auth.models import Group, Permission
from django.contrib.contenttypes.models import ContentType
from django.core.management.base import BaseCommand

from team.models import (
GROUP_NAMES,
COLLECTION_TEAM_ADMIN,
JOURNAL_TEAM_ADMIN,
CollectionTeamMember,
Company,
CompanyTeamMember,
JournalCompanyContract,
JournalTeamMember,
)


class Command(BaseCommand):
help = "Create default user groups and assign permissions for team management"

def handle(self, *args, **options):
for name in GROUP_NAMES:
Group.objects.get_or_create(name=name)
self.stdout.write(f"Group '{name}' ensured.")

self._assign_permissions()
self.stdout.write(self.style.SUCCESS("User groups created/updated successfully."))

def _assign_permissions(self):
# COLLECTION_TEAM_ADMIN: can manage all team members and Company CRUD
collection_admin_group, _ = Group.objects.get_or_create(name=COLLECTION_TEAM_ADMIN)
collection_admin_permissions = self._get_model_permissions(
[CollectionTeamMember, Company, JournalTeamMember, CompanyTeamMember, JournalCompanyContract]
)
collection_admin_group.permissions.set(collection_admin_permissions)

# JOURNAL_TEAM_ADMIN: can manage journal team members and Company Contracts CRUD
journal_admin_group, _ = Group.objects.get_or_create(name=JOURNAL_TEAM_ADMIN)
journal_admin_permissions = self._get_model_permissions(
[JournalTeamMember, JournalCompanyContract]
)
journal_admin_group.permissions.set(journal_admin_permissions)

# COMPANY_TEAM_ADMIN: can manage company team members
company_admin_group, _ = Group.objects.get_or_create(name=COMPANY_TEAM_ADMIN)
company_admin_permissions = self._get_model_permissions([CompanyTeamMember])
company_admin_group.permissions.set(company_admin_permissions)
Comment on lines +28 to +46

Copilot AI Feb 20, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The management command assigns Django model permissions to the groups (view, add, change, delete), but the ViewSets don't use a permission_helper_class to enforce these permissions for CRUD operations. The get_queryset methods only filter which records users can see, but don't prevent users from creating/editing/deleting records they shouldn't have access to. Consider implementing a custom permission_helper_class for each ViewSet to enforce the permission rules described in the PR (e.g., COLLECTION_TEAM_ADMIN can manage Company CRUD, JOURNAL_TEAM_ADMIN can manage JournalCompanyContract CRUD). See upload/permission_helper.py and upload/wagtail_hooks.py for examples of this pattern used in the codebase.

Copilot uses AI. Check for mistakes.

def _get_model_permissions(self, models):
permissions = []
for model in models:
ct = ContentType.objects.get_for_model(model)
permissions.extend(Permission.objects.filter(content_type=ct))
return permissions
116 changes: 116 additions & 0 deletions team/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,31 @@

ALLOWED_COLLECTIONS = ["dom", "spa", "scl", "pan"]

# Django group names for team-based access control.
# COLLECTION_TEAM_ADMIN: collection managers — can CRUD Company, JournalTeamMember,
# CompanyTeamMember, and members of their own collections.
COLLECTION_TEAM_ADMIN = "COLLECTION_TEAM_ADMIN"
# COLLECTION_TEAM_MEMBER: regular collection members — read-only access to own record.
COLLECTION_TEAM_MEMBER = "COLLECTION_TEAM_MEMBER"
# JOURNAL_TEAM_ADMIN: journal managers — can CRUD JournalTeamMember and JournalCompanyContract
# for their managed journals.
JOURNAL_TEAM_ADMIN = "JOURNAL_TEAM_ADMIN"
# JOURNAL_TEAM_MEMBER: regular journal members — read-only access to own record.
JOURNAL_TEAM_MEMBER = "JOURNAL_TEAM_MEMBER"
# COMPANY_TEAM_ADMIN: company managers — can CRUD CompanyTeamMember for their companies.
COMPANY_TEAM_ADMIN = "COMPANY_TEAM_ADMIN"
# COMPANY_MEMBER: regular company members — read-only access to own record.
COMPANY_MEMBER = "COMPANY_MEMBER"

GROUP_NAMES = [
COLLECTION_TEAM_ADMIN,
COLLECTION_TEAM_MEMBER,
JOURNAL_TEAM_ADMIN,
JOURNAL_TEAM_MEMBER,
COMPANY_TEAM_ADMIN,
COMPANY_MEMBER,
]


class TeamRole(models.TextChoices):
"""Role types for team members."""
Expand Down Expand Up @@ -206,6 +231,20 @@ def members(user, is_active_member=None):
def has_upload_permission(cls, user):
return cls.objects.filter(user=user, collection__acron__in=ALLOWED_COLLECTIONS).exists()

@classmethod
def get_queryset_for_user(cls, user, qs):
"""Return the queryset of CollectionTeamMember records visible to the user.

- Managers see all members of their own collection(s).
- Regular members see only their own record.
"""
managed_collection_ids = cls.objects.filter(
user=user, role=TeamRole.MANAGER, is_active_member=True
).values_list("collection", flat=True)
if managed_collection_ids:
return qs.filter(collection__in=managed_collection_ids)
return qs.filter(user=user)


class Company(VisualIdentityMixin, CommonControlField):
"""
Expand Down Expand Up @@ -265,6 +304,23 @@ def get_members(cls, company_id):
is_active_member=True
)

@classmethod
def get_queryset_for_user(cls, user, qs):
"""Return the queryset of Company records visible to the user.

- COLLECTION_TEAM_ADMIN (collection managers) can see all companies.
- Company members see only the companies they belong to.
"""
is_collection_manager = CollectionTeamMember.objects.filter(
user=user, role=TeamRole.MANAGER, is_active_member=True
).exists()
if is_collection_manager:
return qs
company_ids = CompanyTeamMember.objects.filter(
user=user, is_active_member=True
).values_list("company", flat=True)
return qs.filter(id__in=company_ids)


class JournalTeamMember(TeamMember):
"""
Expand Down Expand Up @@ -339,6 +395,26 @@ def get_user_journals(cls, user, role=None, is_active=True):
filters["is_active_member"] = is_active
return cls.objects.filter(**filters).select_related("journal")

@classmethod
def get_queryset_for_user(cls, user, qs):
"""Return the queryset of JournalTeamMember records visible to the user.

- COLLECTION_TEAM_ADMIN sees all journal team members.
- JOURNAL_TEAM_ADMIN sees members of their managed journals.
- Regular members see only their own record.
"""
is_collection_manager = CollectionTeamMember.objects.filter(
user=user, role=TeamRole.MANAGER, is_active_member=True
).exists()
if is_collection_manager:
return qs
managed_journal_ids = cls.objects.filter(
user=user, role=TeamRole.MANAGER, is_active_member=True
).values_list("journal", flat=True)
if managed_journal_ids:
return qs.filter(journal__in=managed_journal_ids)
return qs.filter(user=user)


class CompanyTeamMember(TeamMember):
"""
Expand Down Expand Up @@ -413,6 +489,26 @@ def get_user_companies(cls, user, role=None, is_active=True):
filters["is_active_member"] = is_active
return cls.objects.filter(**filters).select_related("company")

@classmethod
def get_queryset_for_user(cls, user, qs):
"""Return the queryset of CompanyTeamMember records visible to the user.

- COLLECTION_TEAM_ADMIN sees all company team members.
- COMPANY_TEAM_ADMIN sees members of their managed companies.
- Regular members see only their own record.
"""
is_collection_manager = CollectionTeamMember.objects.filter(
user=user, role=TeamRole.MANAGER, is_active_member=True
).exists()
if is_collection_manager:
return qs
managed_company_ids = cls.objects.filter(
user=user, role=TeamRole.MANAGER, is_active_member=True
).values_list("company", flat=True)
if managed_company_ids:
return qs.filter(company__in=managed_company_ids)
return qs.filter(user=user)


class JournalCompanyContract(CommonControlField):
"""
Expand Down Expand Up @@ -478,3 +574,23 @@ def get_company_journals(cls, company, is_active=True):
def can_manage_contract(cls, user, journal):
"""Check if a user can manage contracts for a journal (must be a journal manager)."""
return JournalTeamMember.user_is_manager(user, journal)

@classmethod
def get_queryset_for_user(cls, user, qs):
"""Return the queryset of JournalCompanyContract records visible to the user.

- COLLECTION_TEAM_ADMIN sees all contracts.
- JOURNAL_TEAM_ADMIN sees contracts for their managed journals.
- All others see no contracts.
"""
is_collection_manager = CollectionTeamMember.objects.filter(
user=user, role=TeamRole.MANAGER, is_active_member=True
).exists()
if is_collection_manager:
return qs
managed_journal_ids = JournalTeamMember.objects.filter(
user=user, role=TeamRole.MANAGER, is_active_member=True
).values_list("journal", flat=True)
if managed_journal_ids:
return qs.filter(journal__in=managed_journal_ids)
return qs.none()
74 changes: 74 additions & 0 deletions team/signals.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
from django.contrib.auth.models import Group
from django.db.models.signals import post_delete, post_save

from .models import (
COLLECTION_TEAM_ADMIN,
COLLECTION_TEAM_MEMBER,
COMPANY_MEMBER,
COMPANY_TEAM_ADMIN,
JOURNAL_TEAM_ADMIN,
JOURNAL_TEAM_MEMBER,
CollectionTeamMember,
CompanyTeamMember,
JournalTeamMember,
TeamRole,
)


def _roles_for_user(model_class, user):
"""Return the set of active roles the user holds in a team model."""
return set(
model_class.objects.filter(user=user, is_active_member=True)
.values_list("role", flat=True)
)


def update_user_groups(user):
"""
Synchronise a user's auth.Group memberships to reflect their current
active team-member roles. Called after any team member is saved or deleted.
"""
if user is None:
return

collection_roles = _roles_for_user(CollectionTeamMember, user)
journal_roles = _roles_for_user(JournalTeamMember, user)
company_roles = _roles_for_user(CompanyTeamMember, user)

_sync_group(user, COLLECTION_TEAM_ADMIN, TeamRole.MANAGER in collection_roles)
_sync_group(user, COLLECTION_TEAM_MEMBER, TeamRole.MEMBER in collection_roles)
_sync_group(user, JOURNAL_TEAM_ADMIN, TeamRole.MANAGER in journal_roles)
_sync_group(user, JOURNAL_TEAM_MEMBER, TeamRole.MEMBER in journal_roles)
_sync_group(user, COMPANY_TEAM_ADMIN, TeamRole.MANAGER in company_roles)
_sync_group(user, COMPANY_MEMBER, TeamRole.MEMBER in company_roles)


def _sync_group(user, group_name, should_belong):
"""Add or remove a user from a group, creating the group if needed."""
group, _ = Group.objects.get_or_create(name=group_name)
if should_belong:
user.groups.add(group)
else:
user.groups.remove(group)


def _make_signal_handler(description):
def handler(sender, instance, **kwargs):
update_user_groups(instance.user)
handler.__name__ = description
return handler


_TEAM_MODELS = [CollectionTeamMember, JournalTeamMember, CompanyTeamMember]

for _model in _TEAM_MODELS:
post_save.connect(
_make_signal_handler(f"sync_{_model.__name__.lower()}_groups_on_save"),
sender=_model,
weak=False,
)
post_delete.connect(
_make_signal_handler(f"sync_{_model.__name__.lower()}_groups_on_delete"),
sender=_model,
weak=False,
)
Loading