from django.db.models import Q from django.utils import timezone from .models import Grant, Revoke class AuthorizationService: """Singleton service for permission resolution.""" def _get_system_user_scopes(self, user) -> set[str] | None: from .auth.api_keys import SystemUser if isinstance(user, SystemUser): return set(user.scopes) return None def has_permission(self, user, codename: str) -> bool: if not user or not user.is_authenticated: return False if user.is_superuser: return True system_scopes = self._get_system_user_scopes(user) if system_scopes is not None: return codename in system_scopes if Revoke.objects.filter(user=user, codename=codename).exists(): return False if ( Grant.objects.filter(user=user, codename=codename) .filter(Q(expires_at__isnull=True) | Q(expires_at__gt=timezone.now())) .exists() ): return True user_roles = self._get_role_permission_lists(user) for perm_list in user_roles: if codename in (perm_list or []): return True return False def get_effective_permissions(self, user) -> set[str]: if not user or not user.is_authenticated: return set() if user.is_superuser: return {"*"} system_scopes = self._get_system_user_scopes(user) if system_scopes is not None: return system_scopes revoked = set(Revoke.objects.filter(user=user).values_list("codename", flat=True)) granted = set( Grant.objects.filter(user=user) .filter(Q(expires_at__isnull=True) | Q(expires_at__gt=timezone.now())) .values_list("codename", flat=True) ) role_perms = set() for perm_list in self._get_role_permission_lists(user): role_perms.update(perm_list or []) return (granted | role_perms) - revoked @staticmethod def _get_role_permission_lists(user) -> list[list[str]]: from infrasynth.tenancy.context import get_current_tenant from infrasynth.tenancy.models import TenantMembership from .models import Role, RoleAssignment role_perms: list[list[str]] = [] # Global role assignments (Role.users) apply in every tenant. roles = getattr(user, "roles", None) if roles is not None: role_perms.extend(list(roles.values_list("permissions", flat=True))) tenant = get_current_tenant() if tenant is not None: # Tenant-local role assignments (many roles per user per tenant). role_perms.extend( list(RoleAssignment.objects.filter(user=user).values_list("role__permissions", flat=True)) ) # Roles assigned via membership in the current tenant (TENANCY.md ยง6). slugs = TenantMembership.objects.filter(user=user, tenant=tenant, is_active=True).values_list( "role", flat=True ) for slug in set(slugs): role = Role.objects.filter(slug=slug).first() if role is not None: role_perms.append(role.permissions or []) return role_perms def has_all_permissions(self, user, codenames: list[str]) -> bool: return all(self.has_permission(user, c) for c in codenames) def has_any_permission(self, user, codenames: list[str]) -> bool: return any(self.has_permission(user, c) for c in codenames)