"""RBAC management endpoints: roles, permissions catalog, user role assignment.""" import logging from collections import defaultdict from django.db import models from django.db.models import Count from django.shortcuts import get_object_or_404 from rest_framework import status, viewsets from rest_framework.decorators import action from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from django.utils import timezone from .authorization import registry from .authorization.cache import invalidate_role as _invalidate_role_users from .authorization.cache import invalidate_user as _invalidate_user_access from .models import Invitation, Permission, Role, User, UserRole from .permissions import IsRbacAdmin from .rbac_serializers import ( AdminCreateUserSerializer, AdminUserSerializer, AssignRolesSerializer, InvitationSerializer, PermissionSerializer, RoleSerializer, validate_role_permission_set, ) from .tenancy import get_active_organization from django.conf import settings logger = logging.getLogger(__name__) class PermissionCatalogView(APIView): """Lists menu permissions (legacy, grouped by domain) plus the v2 capability catalog (grouped by module/resource with registry metadata). Read-only, any auth user.""" permission_classes = [IsAuthenticated] def get(self, request): perms = list(Permission.objects.all()) legacy = [ p for p in perms if not p.code or p.code.startswith("legacy.") ] v2_rows = {p.code: p for p in perms if p.code and not p.code.startswith("legacy.")} grouped: dict[str, list] = defaultdict(list) for p in legacy: grouped[p.domain].append(PermissionSerializer(p).data) capabilities: dict[str, list] = defaultdict(list) for definition in registry.all_capabilities(): row = v2_rows.get(definition.code) capabilities[definition.module].append({ "id": row.id if row else None, "code": definition.code, "module": definition.module, "resource": definition.resource, "verb": definition.verb, "label": definition.label, "description": definition.description, "risk_level": definition.risk, "baseline": definition.baseline, "prerequisites": list(definition.prerequisites), }) return Response({ "permissions": PermissionSerializer(legacy, many=True).data, "by_domain": grouped, "capabilities": capabilities, }) class RoleViewSet(viewsets.ModelViewSet): """CRUD for Role. Admin-only writes; reads for any authenticated user (frontend needs role names to render badges/assignment UI).""" queryset = ( Role.objects.exclude(slug__startswith="archived-dept-") .select_related("manager_role") .prefetch_related("permissions") .annotate(user_count_anno=Count("user_assignments", distinct=True)) ) serializer_class = RoleSerializer def get_permissions(self): if self.action in ("list", "retrieve"): return [IsAuthenticated()] return [IsRbacAdmin()] def destroy(self, request, *args, **kwargs): role = self.get_object() if role.is_system: return Response( {"detail": "System role cannot be deleted."}, status=status.HTTP_400_BAD_REQUEST, ) _invalidate_role_users(role) return super().destroy(request, *args, **kwargs) def perform_update(self, serializer): # Capability or manager changes must deterministically bust effective # access for every holder, not wait out the cache TTL. role = serializer.save() _invalidate_role_users(role) @action(detail=True, methods=["post"], url_path="permissions") def set_permissions(self, request, pk=None): """Replace role.permissions with the given list of permission IDs.""" role = self.get_object() ids = request.data.get("permission_ids", []) perms = list(Permission.objects.filter(id__in=ids)) validate_role_permission_set(perms) role.permissions.set(perms) _invalidate_role_users(role) return Response(RoleSerializer(role).data) class AdminUserViewSet(viewsets.ReadOnlyModelViewSet): """Admin user list for the Users settings page (roles attached), scoped to the caller's active organization — both the visible users and their roles/assignments. POST creates a user directly (the alternative to an invitation): membership in the active org, optional roles, and a forced password change on first login.""" serializer_class = AdminUserSerializer permission_classes = [IsRbacAdmin] def create(self, request, *args, **kwargs): organization = get_active_organization(request) serializer = AdminCreateUserSerializer(data=request.data) serializer.is_valid(raise_exception=True) data = serializer.validated_data from django.db import transaction from apps.companies.models import OrganizationMembership from apps.hr.signals import ensure_employee_stub with transaction.atomic(): user = User.objects.create_user( email=data["email"], password=data["password"], first_name=data["first_name"].strip(), last_name=data["last_name"].strip(), must_change_password=True, ) OrganizationMembership.objects.create(organization=organization, user=user) ensure_employee_stub(user) for role in Role.objects.filter(id__in=data["role_ids"]): UserRole.objects.create( user=user, role=role, organization=organization, assigned_by=request.user, ) _invalidate_user_access(user.pk, organization.id) user = self.get_queryset().get(pk=user.pk) return Response( AdminUserSerializer(user, context={"request": request}).data, status=status.HTTP_201_CREATED, ) def get_queryset(self): organization = get_active_organization(self.request) return ( User.objects.filter(organizations=organization) .prefetch_related( models.Prefetch( "user_roles", queryset=UserRole.objects.filter(organization=organization).select_related("role"), ) ) .order_by("email") ) @action(detail=True, methods=["post"], url_path="roles") def assign_roles(self, request, pk=None): """Replace user's roles WITHIN the caller's active org with the given role IDs.""" organization = get_active_organization(request) user = self.get_object() serializer = AssignRolesSerializer(data=request.data) serializer.is_valid(raise_exception=True) role_ids = serializer.validated_data["role_ids"] roles = Role.objects.filter(id__in=role_ids) UserRole.objects.filter(user=user, organization=organization).exclude(role__in=roles).delete() for role in roles: UserRole.objects.get_or_create( user=user, role=role, organization=organization, defaults={"assigned_by": request.user}, ) _invalidate_user_access(user.pk, organization.id) user = self.get_queryset().get(pk=user.pk) return Response(AdminUserSerializer(user, context={"request": request}).data) class InvitationViewSet(viewsets.ModelViewSet): """Invite a user into the caller's active org, with an optional role. List/create/revoke only — no update. Resending an invite is done by creating one again for the same (org, email); a pending row is reused (token/expiry/role refreshed) instead of erroring on the unique constraint. """ serializer_class = InvitationSerializer permission_classes = [IsRbacAdmin] http_method_names = ["get", "post", "delete"] def get_queryset(self): return Invitation.objects.filter( organization=get_active_organization(self.request) ).select_related("role", "invited_by") def create(self, request, *args, **kwargs): organization = get_active_organization(request) serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) email = serializer.validated_data["email"].strip().lower() role = serializer.validated_data.get("role") from apps.companies.models import OrganizationMembership if OrganizationMembership.objects.filter(organization=organization, user__email__iexact=email).exists(): return Response({"email": "This user is already a member of your organization."}, status=400) from .models import _invitation_token invite, _ = Invitation.objects.update_or_create( organization=organization, email=email, status="pending", defaults={ "role": role, "invited_by": request.user, "token": _invitation_token(), "expires_at": timezone.now() + timezone.timedelta(days=7), }, ) from apps.notifications.email import _send link = InvitationSerializer(invite, context={"request": request}).data["link"] try: _send( invite.email, f"You're invited to join {organization.name}", f"You've been invited to join {organization.name} on {settings.PRODUCT_NAME}. " f"Use the link below to accept.", link, ) except Exception: logger.exception("invitation email send failed for %s", invite.email) # the response always includes the link as a copy-link fallback return Response(InvitationSerializer(invite, context={"request": request}).data, status=201) def perform_destroy(self, instance): instance.status = "revoked" instance.save(update_fields=["status"])