from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.core.models.permission import AdminRole, AdminRolePermission, AdminUserPermissionOverride, AdminUserRole from app.core.models.user import User from app.core.permissions.catalog import SYSTEM_ROLE_DEFS, catalog_by_module, default_role_permissions, permission_keys @dataclass class EffectivePermissions: user_id: int tenant_id: int is_owner: bool permissions: set[str] = field(default_factory=set) data_scopes: dict[str, str] = field(default_factory=dict) roles: list[dict] = field(default_factory=list) overrides: list[dict] = field(default_factory=list) def can(self, permission_key: str) -> bool: return self.is_owner or permission_key in self.permissions def data_scope(self, permission_key: str) -> str: return "all" if self.is_owner else self.data_scopes.get(permission_key, "none") def to_dict(self) -> dict: keys = permission_keys() if self.is_owner else self.permissions return { "user_id": self.user_id, "tenant_id": self.tenant_id, "is_owner": self.is_owner, "permissions": sorted(keys), "data_scopes": {key: self.data_scope(key) for key in sorted(keys)}, "roles": self.roles, "overrides": self.overrides, "catalog": catalog_by_module(), } def _legacy_permissions_for_role(role: str) -> set[str]: if role == "owner": return permission_keys() if role in SYSTEM_ROLE_DEFS: return set(default_role_permissions(role)) return set() class PermissionService: def __init__(self, db: AsyncSession): self.db = db async def get_effective_permissions(self, user: User) -> EffectivePermissions: if user.role == "owner": keys = permission_keys() return EffectivePermissions( user_id=user.id, tenant_id=user.tenant_id, is_owner=True, permissions=keys, data_scopes={key: "all" for key in keys}, roles=[{"code": "owner", "name": "拥有者", "is_system": True}], ) effective = EffectivePermissions(user_id=user.id, tenant_id=user.tenant_id, is_owner=False) result = await self.db.execute( select(AdminUserRole) .options(selectinload(AdminUserRole.role).selectinload(AdminRole.permissions)) .where(AdminUserRole.tenant_id == user.tenant_id, AdminUserRole.user_id == user.id) ) assignments = result.scalars().all() for assignment in assignments: role = assignment.role if not role or not role.is_active: continue effective.roles.append( { "id": role.id, "code": role.code, "name": role.name, "description": role.description, "is_system": bool(role.is_system), } ) for role_permission in role.permissions: if role_permission.effect == "allow": effective.permissions.add(role_permission.permission_key) effective.data_scopes[role_permission.permission_key] = role_permission.data_scope or "all" if not assignments: for permission_key in _legacy_permissions_for_role(user.role): effective.permissions.add(permission_key) effective.data_scopes[permission_key] = "all" if user.role in SYSTEM_ROLE_DEFS: role_def = SYSTEM_ROLE_DEFS[user.role] effective.roles.append( { "code": user.role, "name": role_def["name"], "description": role_def["description"], "is_system": True, "legacy": True, } ) now = datetime.utcnow() overrides_result = await self.db.execute( select(AdminUserPermissionOverride).where( AdminUserPermissionOverride.tenant_id == user.tenant_id, AdminUserPermissionOverride.user_id == user.id, ) ) overrides = [o for o in overrides_result.scalars().all() if o.expires_at is None or o.expires_at > now] for override in overrides: if override.effect == "allow": effective.permissions.add(override.permission_key) effective.data_scopes[override.permission_key] = override.data_scope or "all" for override in overrides: effective.overrides.append( { "permission_key": override.permission_key, "effect": override.effect, "data_scope": override.data_scope, "reason": override.reason, "expires_at": override.expires_at.isoformat() if override.expires_at else None, } ) if override.effect == "deny": effective.permissions.discard(override.permission_key) effective.data_scopes[override.permission_key] = "none" known = permission_keys() effective.permissions = {key for key in effective.permissions if key in known} effective.data_scopes = {key: scope for key, scope in effective.data_scopes.items() if key in known} return effective async def replace_role_permissions(self, tenant_id: int, role_id: int, permission_items: list[dict]) -> None: await self.db.execute( delete(AdminRolePermission).where( AdminRolePermission.tenant_id == tenant_id, AdminRolePermission.role_id == role_id, ) ) known = permission_keys() for item in permission_items: key = item.get("permission_key") if key in known: self.db.add( AdminRolePermission( tenant_id=tenant_id, role_id=role_id, permission_key=key, effect="allow", data_scope=item.get("data_scope") or "all", ) ) async def replace_user_roles(self, tenant_id: int, user_id: int, role_ids: list[int]) -> None: await self.db.execute(delete(AdminUserRole).where(AdminUserRole.tenant_id == tenant_id, AdminUserRole.user_id == user_id)) if not role_ids: return valid_result = await self.db.execute( select(AdminRole.id).where(AdminRole.tenant_id == tenant_id, AdminRole.id.in_(role_ids), AdminRole.is_active == 1) ) for role_id in valid_result.scalars().all(): self.db.add(AdminUserRole(tenant_id=tenant_id, user_id=user_id, role_id=role_id)) async def replace_user_overrides(self, tenant_id: int, user_id: int, overrides: list[dict], created_by: int | None) -> None: await self.db.execute( delete(AdminUserPermissionOverride).where( AdminUserPermissionOverride.tenant_id == tenant_id, AdminUserPermissionOverride.user_id == user_id, ) ) known = permission_keys() for item in overrides: key = item.get("permission_key") effect = item.get("effect") if key not in known or effect not in ("allow", "deny"): continue expires_at = item.get("expires_at") if isinstance(expires_at, str) and expires_at: expires_at = datetime.fromisoformat(expires_at.replace("Z", "+00:00")).replace(tzinfo=None) else: expires_at = None self.db.add( AdminUserPermissionOverride( tenant_id=tenant_id, user_id=user_id, permission_key=key, effect=effect, data_scope=item.get("data_scope") or "all", reason=item.get("reason"), expires_at=expires_at, created_by=created_by, ) )