111 lines
2.8 KiB
Python
111 lines
2.8 KiB
Python
"""
|
||
权限服务
|
||
提供权限检查的辅助函数
|
||
"""
|
||
from sqlalchemy.orm import Session
|
||
from app.models.permission import WorkflowPermission, AgentPermission
|
||
from app.models.user import User
|
||
from app.models.workflow import Workflow
|
||
from app.models.agent import Agent
|
||
from typing import Optional
|
||
|
||
|
||
def check_workflow_permission(
|
||
db: Session,
|
||
user: User,
|
||
workflow: Workflow,
|
||
permission_type: str
|
||
) -> bool:
|
||
"""
|
||
检查用户对工作流的权限
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
user: 用户对象
|
||
workflow: 工作流对象
|
||
permission_type: 权限类型（read/write/execute/share）
|
||
|
||
Returns:
|
||
bool: 是否有权限
|
||
"""
|
||
# 管理员拥有所有权限
|
||
if user.role == "admin":
|
||
return True
|
||
|
||
# 工作流所有者拥有所有权限
|
||
if workflow.user_id == user.id:
|
||
return True
|
||
|
||
# 检查用户直接权限
|
||
user_permission = db.query(WorkflowPermission).filter(
|
||
WorkflowPermission.workflow_id == workflow.id,
|
||
WorkflowPermission.user_id == user.id,
|
||
WorkflowPermission.permission_type == permission_type
|
||
).first()
|
||
|
||
if user_permission:
|
||
return True
|
||
|
||
# 检查角色权限
|
||
for role in user.roles:
|
||
role_permission = db.query(WorkflowPermission).filter(
|
||
WorkflowPermission.workflow_id == workflow.id,
|
||
WorkflowPermission.role_id == role.id,
|
||
WorkflowPermission.permission_type == permission_type
|
||
).first()
|
||
|
||
if role_permission:
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def check_agent_permission(
|
||
db: Session,
|
||
user: User,
|
||
agent: Agent,
|
||
permission_type: str
|
||
) -> bool:
|
||
"""
|
||
检查用户对Agent的权限
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
user: 用户对象
|
||
agent: Agent对象
|
||
permission_type: 权限类型（read/write/execute/deploy）
|
||
|
||
Returns:
|
||
bool: 是否有权限
|
||
"""
|
||
# 管理员拥有所有权限
|
||
if user.role == "admin":
|
||
return True
|
||
|
||
# Agent所有者拥有所有权限
|
||
if agent.user_id == user.id:
|
||
return True
|
||
|
||
# 检查用户直接权限
|
||
user_permission = db.query(AgentPermission).filter(
|
||
AgentPermission.agent_id == agent.id,
|
||
AgentPermission.user_id == user.id,
|
||
AgentPermission.permission_type == permission_type
|
||
).first()
|
||
|
||
if user_permission:
|
||
return True
|
||
|
||
# 检查角色权限
|
||
for role in user.roles:
|
||
role_permission = db.query(AgentPermission).filter(
|
||
AgentPermission.agent_id == agent.id,
|
||
AgentPermission.role_id == role.id,
|
||
AgentPermission.permission_type == permission_type
|
||
).first()
|
||
|
||
if role_permission:
|
||
return True
|
||
|
||
return False
|