from __future__ import annotations import re import secrets import string from collections.abc import Iterable from dataclasses import dataclass from sqlalchemy import func from sqlalchemy.orm import Session from govoplan_core.core.optional import reraise_unless_missing_package from govoplan_core.db.models import ( Account, ApiKey, Group, GroupRoleAssignment, Role, SystemRoleAssignment, Tenant, User, UserGroupMembership, UserRoleAssignment, ) from govoplan_core.security.passwords import hash_password from govoplan_core.security.permissions import ( DEFAULT_SYSTEM_ROLES, DEFAULT_TENANT_ROLES, normalize_email, scopes_grant, validate_system_permissions, validate_tenant_permissions, ) _SLUG_RE = re.compile(r"[^a-z0-9]+") _TEMP_PASSWORD_ALPHABET = string.ascii_letters + string.digits + "-_!@#" class AdminConflictError(ValueError): pass class AdminValidationError(ValueError): pass @dataclass(slots=True) class MembershipCreationResult: user: User account: Account temporary_password: str | None = None account_created: bool = False def slugify(value: str) -> str: slug = _SLUG_RE.sub("-", value.strip().casefold()).strip("-") if not slug: raise AdminValidationError("A slug must contain at least one letter or number.") return slug[:100] def generate_temporary_password(length: int = 20) -> str: return "".join(secrets.choice(_TEMP_PASSWORD_ALPHABET) for _ in range(length)) def ensure_default_roles(session: Session, tenant: Tenant | None = None) -> dict[str, Role]: definitions = DEFAULT_TENANT_ROLES if tenant is not None else DEFAULT_SYSTEM_ROLES roles: dict[str, Role] = {} for slug, definition in definitions.items(): query = session.query(Role).filter(Role.slug == slug) query = query.filter(Role.tenant_id == tenant.id) if tenant is not None else query.filter(Role.tenant_id.is_(None)) role = query.one_or_none() if role is None: role = Role( tenant_id=tenant.id if tenant is not None else None, slug=slug, name=str(definition["name"]), description=str(definition.get("description") or "") or None, permissions=list(definition["permissions"]), is_builtin=bool(definition.get("is_builtin", True)), is_assignable=bool(definition.get("is_assignable", True)), ) session.add(role) session.flush() else: # Tenant built-ins and explicitly managed system roles remain # code-defined. Seeded, non-protected system roles are only created # here and can subsequently be administered in the System roles UI. managed = tenant is not None or bool(definition.get("managed", False)) if managed: role.name = str(definition["name"]) role.description = str(definition.get("description") or "") or None role.permissions = list(definition["permissions"]) role.is_builtin = bool(definition.get("is_builtin", True)) role.is_assignable = bool(definition.get("is_assignable", True)) session.add(role) roles[slug] = role return roles def get_or_create_account( session: Session, *, email: str, display_name: str | None = None, password: str | None = None, password_reset_required: bool = False, ) -> tuple[Account, bool, str | None]: normalized = normalize_email(email) if not normalized or "@" not in normalized: raise AdminValidationError("Enter a valid email address.") account = session.query(Account).filter(Account.normalized_email == normalized).one_or_none() if account is not None: if not account.is_active: raise AdminConflictError( "The global account is disabled. A system administrator must reactivate it before adding a tenant membership." ) return account, False, None temporary_password = password or generate_temporary_password() account = Account( email=email.strip(), normalized_email=normalized, display_name=display_name.strip() if display_name else None, is_active=True, auth_provider="local", password_hash=hash_password(temporary_password), password_reset_required=password_reset_required or password is None, ) session.add(account) session.flush() return account, True, temporary_password def create_membership( session: Session, *, tenant: Tenant, email: str, display_name: str | None = None, password: str | None = None, password_reset_required: bool = False, is_active: bool = True, ) -> MembershipCreationResult: account, account_created, temporary_password = get_or_create_account( session, email=email, display_name=display_name, password=password, password_reset_required=password_reset_required, ) existing = ( session.query(User) .filter(User.tenant_id == tenant.id, User.account_id == account.id) .one_or_none() ) if existing is not None: raise AdminConflictError("This account already belongs to the tenant.") user = User( tenant_id=tenant.id, account_id=account.id, email=account.email, display_name=display_name.strip() if display_name else account.display_name, is_active=is_active, is_tenant_admin=False, auth_provider=account.auth_provider, password_hash=account.password_hash, ) session.add(user) session.flush() return MembershipCreationResult( user=user, account=account, temporary_password=temporary_password if account_created else None, account_created=account_created, ) def set_user_groups(session: Session, *, user: User, group_ids: Iterable[str]) -> None: ids = sorted(set(group_ids)) groups = ( session.query(Group) .filter(Group.tenant_id == user.tenant_id, Group.id.in_(ids)) .all() if ids else [] ) if len(groups) != len(ids): raise AdminValidationError("One or more selected groups do not belong to the tenant.") session.query(UserGroupMembership).filter( UserGroupMembership.tenant_id == user.tenant_id, UserGroupMembership.user_id == user.id, ).delete(synchronize_session=False) for group in groups: session.add(UserGroupMembership(tenant_id=user.tenant_id, user_id=user.id, group_id=group.id)) session.flush() def set_user_roles(session: Session, *, user: User, role_ids: Iterable[str]) -> None: ids = sorted(set(role_ids)) roles = ( session.query(Role) .filter(Role.tenant_id == user.tenant_id, Role.id.in_(ids), Role.is_assignable.is_(True)) .all() if ids else [] ) if len(roles) != len(ids): raise AdminValidationError("One or more selected roles are invalid or not assignable.") session.query(UserRoleAssignment).filter( UserRoleAssignment.tenant_id == user.tenant_id, UserRoleAssignment.user_id == user.id, ).delete(synchronize_session=False) for role in roles: session.add(UserRoleAssignment(tenant_id=user.tenant_id, user_id=user.id, role_id=role.id)) session.flush() def set_group_members(session: Session, *, group: Group, user_ids: Iterable[str]) -> None: ids = sorted(set(user_ids)) users = ( session.query(User) .filter(User.tenant_id == group.tenant_id, User.id.in_(ids)) .all() if ids else [] ) if len(users) != len(ids): raise AdminValidationError("One or more selected users do not belong to the tenant.") session.query(UserGroupMembership).filter( UserGroupMembership.tenant_id == group.tenant_id, UserGroupMembership.group_id == group.id, ).delete(synchronize_session=False) for user in users: session.add(UserGroupMembership(tenant_id=group.tenant_id, user_id=user.id, group_id=group.id)) session.flush() def set_group_roles(session: Session, *, group: Group, role_ids: Iterable[str]) -> None: ids = sorted(set(role_ids)) roles = ( session.query(Role) .filter(Role.tenant_id == group.tenant_id, Role.id.in_(ids), Role.is_assignable.is_(True)) .all() if ids else [] ) if len(roles) != len(ids): raise AdminValidationError("One or more selected roles are invalid or not assignable.") session.query(GroupRoleAssignment).filter( GroupRoleAssignment.tenant_id == group.tenant_id, GroupRoleAssignment.group_id == group.id, ).delete(synchronize_session=False) for role in roles: session.add(GroupRoleAssignment(tenant_id=group.tenant_id, group_id=group.id, role_id=role.id)) session.flush() def create_custom_role( session: Session, *, tenant: Tenant, slug: str, name: str, description: str | None, permissions: Iterable[str], ) -> Role: normalized_slug = slugify(slug) if session.query(Role).filter(Role.tenant_id == tenant.id, Role.slug == normalized_slug).count(): raise AdminConflictError("A role with this slug already exists in the tenant.") try: validated = validate_tenant_permissions(permissions) except ValueError as exc: raise AdminValidationError(str(exc)) from exc role = Role( tenant_id=tenant.id, slug=normalized_slug, name=name.strip(), description=description.strip() if description else None, permissions=validated, is_builtin=False, is_assignable=True, ) session.add(role) session.flush() return role def update_custom_role( session: Session, *, role: Role, name: str, description: str | None, permissions: Iterable[str], is_assignable: bool, ) -> None: if role.is_builtin: raise AdminConflictError("Built-in roles are managed by the application and cannot be edited.") try: validated = validate_tenant_permissions(permissions) except ValueError as exc: raise AdminValidationError(str(exc)) from exc role.name = name.strip() role.description = description.strip() if description else None role.permissions = validated role.is_assignable = is_assignable session.add(role) session.flush() def delete_custom_role(session: Session, role: Role) -> None: if role.is_builtin: raise AdminConflictError("Built-in roles cannot be deleted.") assigned = session.query(UserRoleAssignment).filter(UserRoleAssignment.role_id == role.id).count() assigned += session.query(GroupRoleAssignment).filter(GroupRoleAssignment.role_id == role.id).count() if assigned: raise AdminConflictError("Remove all user and group assignments before deleting this role.") session.delete(role) session.flush() def create_system_role( session: Session, *, slug: str, name: str, description: str | None, permissions: Iterable[str], ) -> Role: normalized_slug = slugify(slug) if session.query(Role).filter(Role.tenant_id.is_(None), Role.slug == normalized_slug).count(): raise AdminConflictError("A system role with this slug already exists.") try: validated = validate_system_permissions(permissions) except ValueError as exc: raise AdminValidationError(str(exc)) from exc if "system:*" in validated: raise AdminValidationError("Only the protected System owner role may contain system:*.") role = Role( tenant_id=None, slug=normalized_slug, name=name.strip(), description=description.strip() if description else None, permissions=validated, is_builtin=False, is_assignable=True, ) session.add(role) session.flush() return role def update_system_role( session: Session, *, role: Role, name: str, description: str | None, permissions: Iterable[str], is_assignable: bool, ) -> None: if role.tenant_id is not None: raise AdminValidationError("This is not a system role.") if role.slug == "system_owner": raise AdminConflictError("The protected System owner role cannot be edited.") try: validated = validate_system_permissions(permissions) except ValueError as exc: raise AdminValidationError(str(exc)) from exc if "system:*" in validated: raise AdminValidationError("Only the protected System owner role may contain system:*.") role.name = name.strip() role.description = description.strip() if description else None role.permissions = validated role.is_assignable = is_assignable role.is_builtin = False session.add(role) session.flush() def delete_system_role(session: Session, role: Role) -> None: if role.tenant_id is not None: raise AdminValidationError("This is not a system role.") if role.slug == "system_owner": raise AdminConflictError("The protected System owner role cannot be deleted.") assigned = session.query(SystemRoleAssignment).filter(SystemRoleAssignment.role_id == role.id).count() if assigned: raise AdminConflictError("Remove all account assignments before deleting this system role.") session.delete(role) session.flush() def assert_can_delegate_system_permissions(actor_scopes: Iterable[str], permissions: Iterable[str]) -> None: """Prevent a system administrator from defining stronger roles than they hold.""" from govoplan_core.security.permissions import delegateable_system_scopes requested = set(validate_system_permissions(permissions)) if "system:*" in requested: if not scopes_grant(actor_scopes, "system:*"): raise AdminValidationError("Only a System owner may delegate full system access.") return allowed = delegateable_system_scopes(actor_scopes) missing = sorted(scope for scope in requested if scope not in allowed) if missing: raise AdminValidationError("You cannot delegate system permissions you do not currently hold: " + ", ".join(missing)) def assert_can_delegate_system_roles( session: Session, *, actor_scopes: Iterable[str], role_ids: Iterable[str], ) -> None: ids = sorted(set(role_ids)) if not ids: return roles = session.query(Role).filter(Role.tenant_id.is_(None), Role.id.in_(ids)).all() if len(roles) != len(ids): raise AdminValidationError("One or more selected system roles are invalid.") for role in roles: assert_can_delegate_system_permissions(actor_scopes, role.permissions or []) def set_system_roles(session: Session, *, account: Account, role_ids: Iterable[str]) -> None: ids = sorted(set(role_ids)) roles = ( session.query(Role) .filter(Role.tenant_id.is_(None), Role.id.in_(ids), Role.is_assignable.is_(True)) .all() if ids else [] ) if len(roles) != len(ids): raise AdminValidationError("One or more system roles are invalid or not assignable.") for role in roles: try: validate_system_permissions(role.permissions or []) except ValueError as exc: raise AdminValidationError(str(exc)) from exc session.query(SystemRoleAssignment).filter(SystemRoleAssignment.account_id == account.id).delete(synchronize_session=False) for role in roles: session.add(SystemRoleAssignment(account_id=account.id, role_id=role.id)) session.flush() assert_system_owner_exists(session) def account_has_system_scope(session: Session, account: Account, required: str) -> bool: roles = ( session.query(Role) .join(SystemRoleAssignment, SystemRoleAssignment.role_id == Role.id) .filter(SystemRoleAssignment.account_id == account.id, Role.tenant_id.is_(None)) .all() ) return any(scopes_grant(role.permissions or [], required) for role in roles) def tenant_owner_user_ids(session: Session, tenant_id: str) -> set[str]: """Return active tenant memberships that currently satisfy the operational-owner invariant.""" users = ( session.query(User) .join(Account, Account.id == User.account_id) .filter( User.tenant_id == tenant_id, User.is_active.is_(True), Account.is_active.is_(True), ) .all() ) owner_ids: set[str] = set() user_ids = [user.id for user in users] if not user_ids: return owner_ids roles_by_user_id: dict[str, list[Role]] = {user_id: [] for user_id in user_ids} direct_role_rows = ( session.query(UserRoleAssignment.user_id, Role) .join(Role, UserRoleAssignment.role_id == Role.id) .filter( UserRoleAssignment.tenant_id == tenant_id, UserRoleAssignment.user_id.in_(user_ids), Role.tenant_id == tenant_id, ) .all() ) for user_id, role in direct_role_rows: roles_by_user_id.setdefault(user_id, []).append(role) group_role_rows = ( session.query(UserGroupMembership.user_id, Role) .join(GroupRoleAssignment, UserGroupMembership.group_id == GroupRoleAssignment.group_id) .join(Role, GroupRoleAssignment.role_id == Role.id) .join(Group, Group.id == UserGroupMembership.group_id) .filter( UserGroupMembership.tenant_id == tenant_id, UserGroupMembership.user_id.in_(user_ids), Group.is_active.is_(True), Role.tenant_id == tenant_id, ) .all() ) for user_id, role in group_role_rows: roles_by_user_id.setdefault(user_id, []).append(role) for user in users: effective = [ scope for role in roles_by_user_id.get(user.id, []) for scope in (role.permissions or []) ] if scopes_grant(effective, "admin:roles:write") and scopes_grant(effective, "campaign:send"): owner_ids.add(user.id) return owner_ids def tenant_has_owner(session: Session, tenant_id: str) -> bool: return bool(tenant_owner_user_ids(session, tenant_id)) def assert_tenant_owner_exists(session: Session, tenant_id: str) -> None: session.flush() tenant = session.get(Tenant, tenant_id) if tenant is not None and tenant.is_active and not tenant_has_owner(session, tenant_id): raise AdminConflictError("An active tenant must retain at least one active owner or administrator with full operational access.") def assert_system_owner_exists(session: Session) -> None: """Keep one active assignment of the protected System owner role. Other roles may hold broad system permissions, but they are intentionally not substitutes for the protected break-glass owner identity. """ session.flush() owner_role = ( session.query(Role) .filter(Role.tenant_id.is_(None), Role.slug == "system_owner") .one_or_none() ) if owner_role is None: raise AdminConflictError("The protected System owner role is missing.") count = ( session.query(func.count(SystemRoleAssignment.id)) .join(Account, Account.id == SystemRoleAssignment.account_id) .filter(SystemRoleAssignment.role_id == owner_role.id, Account.is_active.is_(True)) .scalar() ) if not count: raise AdminConflictError("At least one active account must retain the protected System owner role.") def role_assignment_counts(session: Session, role_id: str) -> tuple[int, int]: return ( session.query(UserRoleAssignment).filter(UserRoleAssignment.role_id == role_id).count(), session.query(GroupRoleAssignment).filter(GroupRoleAssignment.role_id == role_id).count(), ) def _optional_model(module_name: str, model_name: str): try: module = __import__(module_name, fromlist=[model_name]) except ModuleNotFoundError as exc: reraise_unless_missing_package(exc, module_name.split(".", 1)[0]) return None return getattr(module, model_name) def tenant_counts(session: Session, tenant_id: str) -> dict[str, int]: Campaign = _optional_model("govoplan_campaign.backend.db.models", "Campaign") FileAsset = _optional_model("govoplan_files.backend.db.models", "FileAsset") return { "users": session.query(User).filter(User.tenant_id == tenant_id).count(), "active_users": session.query(User).filter(User.tenant_id == tenant_id, User.is_active.is_(True)).count(), "groups": session.query(Group).filter(Group.tenant_id == tenant_id).count(), "campaigns": session.query(Campaign).filter(Campaign.tenant_id == tenant_id).count() if Campaign is not None else 0, "files": session.query(FileAsset).filter(FileAsset.tenant_id == tenant_id).count() if FileAsset is not None else 0, "api_keys": session.query(ApiKey).filter(ApiKey.tenant_id == tenant_id, ApiKey.revoked_at.is_(None)).count(), } def assert_can_delegate_tenant_permissions(actor_scopes: Iterable[str], permissions: Iterable[str]) -> None: """Prevent administrators from defining or assigning roles beyond their own effective tenant powers.""" from govoplan_core.security.permissions import delegateable_tenant_scopes requested = set(validate_tenant_permissions(permissions)) if "tenant:*" in requested: # Only a tenant owner-equivalent can delegate the wildcard. if not scopes_grant(actor_scopes, "tenant:*"): raise AdminValidationError("You cannot delegate full tenant access unless you already have it.") return allowed = delegateable_tenant_scopes(actor_scopes) missing = sorted(scope for scope in requested if scope not in allowed) if missing: raise AdminValidationError("You cannot delegate permissions you do not currently hold: " + ", ".join(missing)) def assert_can_delegate_roles(session: Session, *, actor_scopes: Iterable[str], role_ids: Iterable[str], tenant_id: str) -> None: ids = sorted(set(role_ids)) if not ids: return roles = session.query(Role).filter(Role.tenant_id == tenant_id, Role.id.in_(ids)).all() if len(roles) != len(ids): raise AdminValidationError("One or more selected roles do not belong to the tenant.") for role in roles: assert_can_delegate_tenant_permissions(actor_scopes, role.permissions or [])