"""权限管理服务，负责算法访问权限和用户权限管理""" from typing import Dict, List, Optional, Any from enum import Enum from datetime import datetime import logging from sqlalchemy.orm import Session from sqlalchemy import and_, or_ from app.models.models import User, Algorithm logger = logging.getLogger(__name__) class PermissionType(Enum): """权限类型枚举""" READ = "read" WRITE = "write" EXECUTE = "execute" ADMIN = "admin" class AccessLevel(Enum): """访问级别枚举""" NONE = "none" READ = "read" READ_WRITE = "read_write" EXECUTE = "execute" FULL = "full" class PermissionManager: """权限管理器，处理用户对算法的访问权限""" def __init__(self): pass def check_algorithm_access(self, db: Session, user_id: str, algorithm_id: str, permission_type: PermissionType) -> bool: """检查用户对算法的访问权限""" try: # 获取用户和算法信息 user = db.query(User).filter(User.id == user_id).first() algorithm = db.query(Algorithm).filter(Algorithm.id == algorithm_id).first() if not user or not algorithm: return False # 管理员拥有所有权限 if user.role == "admin": return True # 算法所有者拥有完全权限 # 注意：在这个模型中，我们没有直接的算法所有者字段 # 实际应用中可能需要添加算法创建者的关联 # 根据算法状态决定权限 if algorithm.status == "inactive": # 非活跃算法只有管理员可以访问 return False # 根据权限类型检查权限 if permission_type in [PermissionType.READ, PermissionType.EXECUTE]: # 读取和执行权限：活跃算法对所有认证用户开放 return algorithm.status == "active" elif permission_type == PermissionType.WRITE: # 写入权限：只有特定用户或管理员可以 return user.role in ["admin", "manager"] elif permission_type == PermissionType.ADMIN: # 管理权限：只有管理员可以 return user.role == "admin" return False except Exception as e: logger.error(f"Error checking algorithm access: {str(e)}") return False def grant_permission(self, db: Session, admin_user_id: str, user_id: str, algorithm_id: str, access_level: AccessLevel) -> bool: """授予用户对算法的权限""" try: # 验证管理员权限 admin_user = db.query(User).filter( User.id == admin_user_id, User.role.in_(["admin", "manager"]) ).first() if not admin_user: logger.warning(f"User {admin_user_id} is not authorized to grant permissions") return False # 验证目标用户和算法存在 target_user = db.query(User).filter(User.id == user_id).first() algorithm = db.query(Algorithm).filter(Algorithm.id == algorithm_id).first() if not target_user or not algorithm: logger.warning("Target user or algorithm not found") return False # 在实际实现中，这里应该创建权限记录 # 由于当前数据模型中没有专门的权限表，我们记录日志 logger.info(f"Permission granted: user {user_id} -> algorithm {algorithm_id}, " f"level: {access_level.value}, by admin: {admin_user_id}") # 在实际实现中，应该创建权限记录到数据库 # db.add(PermissionRecord(user_id=user_id, algorithm_id=algorithm_id, access_level=access_level)) # db.commit() return True except Exception as e: logger.error(f"Error granting permission: {str(e)}") return False def revoke_permission(self, db: Session, admin_user_id: str, user_id: str, algorithm_id: str) -> bool: """撤销用户对算法的权限""" try: # 验证管理员权限 admin_user = db.query(User).filter( User.id == admin_user_id, User.role.in_(["admin", "manager"]) ).first() if not admin_user: logger.warning(f"User {admin_user_id} is not authorized to revoke permissions") return False # 在实际实现中，这里应该删除权限记录 logger.info(f"Permission revoked: user {user_id} -> algorithm {algorithm_id}, " f"by admin: {admin_user_id}") return True except Exception as e: logger.error(f"Error revoking permission: {str(e)}") return False def get_user_permissions(self, db: Session, user_id: str) -> List[Dict[str, Any]]: """获取用户的所有权限""" try: user = db.query(User).filter(User.id == user_id).first() if not user: return [] permissions = [] if user.role == "admin": # 管理员拥有所有算法的完全权限 algorithms = db.query(Algorithm).all() for algorithm in algorithms: permissions.append({ "algorithm_id": algorithm.id, "algorithm_name": algorithm.name, "access_level": AccessLevel.FULL.value, "granted_at": algorithm.created_at, "granted_by": "system", "permission_type": "administrative" }) else: # 普通用户只能访问活跃算法 algorithms = db.query(Algorithm).filter(Algorithm.status == "active").all() for algorithm in algorithms: permissions.append({ "algorithm_id": algorithm.id, "algorithm_name": algorithm.name, "access_level": AccessLevel.READ.value, # 默认只读权限 "granted_at": algorithm.created_at, "granted_by": "system", "permission_type": "public" }) return permissions except Exception as e: logger.error(f"Error getting user permissions: {str(e)}") return [] def get_algorithm_permissions(self, db: Session, algorithm_id: str) -> List[Dict[str, Any]]: """获取算法的所有权限分配""" try: algorithm = db.query(Algorithm).filter(Algorithm.id == algorithm_id).first() if not algorithm: return [] permissions = [] # 如果是活跃算法，所有用户都有读取权限 if algorithm.status == "active": # 添加公共访问权限信息 permissions.append({ "user_id": "*", "user_name": "All Users", "access_level": AccessLevel.READ.value, "granted_at": algorithm.created_at, "granted_by": "system", "permission_type": "public" }) # 获取特定用户的权限（如果有专门的权限表的话） # 这里我们模拟返回一些示例数据 # 在实际实现中，应该查询权限表 return permissions except Exception as e: logger.error(f"Error getting algorithm permissions: {str(e)}") return [] def validate_user_algorithm_operation(self, db: Session, user_id: str, algorithm_id: str, operation: str) -> bool: """验证用户对算法的操作权限""" try: perm_type = { 'read': PermissionType.READ, 'execute': PermissionType.EXECUTE, 'write': PermissionType.WRITE, 'admin': PermissionType.ADMIN }.get(operation.lower(), PermissionType.READ) return self.check_algorithm_access(db, user_id, algorithm_id, perm_type) except Exception as e: logger.error(f"Error validating user operation: {str(e)}") return False class RBACManager: """基于角色的访问控制管理器""" def __init__(self): # 定义角色权限映射 self.role_permissions = { "admin": [ PermissionType.READ, PermissionType.WRITE, PermissionType.EXECUTE, PermissionType.ADMIN ], "manager": [ PermissionType.READ, PermissionType.WRITE, PermissionType.EXECUTE ], "user": [PermissionType.READ, PermissionType.EXECUTE], "guest": [PermissionType.READ] } def get_role_permissions(self, role: str) -> List[PermissionType]: """获取角色的权限列表""" return self.role_permissions.get(role, []) def user_has_permission(self, db: Session, user_id: str, permission_type: PermissionType) -> bool: """检查用户是否具有某种权限""" try: user = db.query(User).filter(User.id == user_id).first() if not user: return False user_perms = self.get_role_permissions(user.role) return permission_type in user_perms except Exception as e: logger.error(f"Error checking user permission: {str(e)}") return False # 全局权限管理器实例 permission_manager = PermissionManager() rbac_manager = RBACManager()