242 lines
12 KiB
Python
242 lines
12 KiB
Python
|
"""v6 资源访问边界：组织、所有者、可见性和授权统一判定。"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
from fastapi import HTTPException, Request
|
|||
|
|
from sqlmodel import Session, select
|
|||
|
|
|
|||
|
|
from .models import Agent, AgentGrant, AgentSkill, AgentSkillGrant, AppUser, Canvas, CanvasGrant, KnowledgeBase, KnowledgeBaseGrant, Memory, MemoryGrant, ResourceRevision
|
|||
|
|
|
|||
|
|
|
|||
|
|
def request_user(request: Request | None) -> AppUser | None:
|
|||
|
|
return getattr(getattr(request, "state", None), "user", None) if request else None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def actor_fields(request: Request | None) -> tuple[int, int]:
|
|||
|
|
user = request_user(request)
|
|||
|
|
return (int(user.organization_id or 1), int(user.id or 0)) if user else (1, 0)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def can_access(resource: Any, user: AppUser | None, operation: str = "read",
|
|||
|
|
grant_permission: str | None = None, db: Session | None = None) -> bool:
|
|||
|
|
"""返回资源是否允许当前用户执行 read/use/manage/export。"""
|
|||
|
|
if user is None:
|
|||
|
|
return True # 单元测试和内部任务调用没有 HTTP 身份上下文
|
|||
|
|
organization_id = int(getattr(resource, "organization_id", 1) or 1)
|
|||
|
|
if organization_id not in {0, int(user.organization_id or 1)}:
|
|||
|
|
return False
|
|||
|
|
# 管理员是租户管理员，不是跨租户平台超级管理员；全局资源（organization_id=0）仍可使用。
|
|||
|
|
if user.role == "admin":
|
|||
|
|
return organization_id in {0, int(user.organization_id or 1)}
|
|||
|
|
owner_id = int(getattr(resource, "owner_user_id", 0) or 0)
|
|||
|
|
if owner_id == int(user.id or 0):
|
|||
|
|
# 普通用户始终不能导出已发布的全局资源；管理端导出接口由管理员权限保护。
|
|||
|
|
if operation == "export" and str(getattr(resource, "visibility", "private") or "private") == "global":
|
|||
|
|
return False
|
|||
|
|
return True
|
|||
|
|
visibility = str(getattr(resource, "visibility", "private") or "private")
|
|||
|
|
if visibility == "global" and operation in {"read", "use"}:
|
|||
|
|
return True
|
|||
|
|
# 全局资源对非所有者仅 read/use（不能改/删/导出）；需要管理权走 shared+manage 授权。
|
|||
|
|
if visibility == "global":
|
|||
|
|
return False
|
|||
|
|
if not grant_permission or visibility != "shared":
|
|||
|
|
return False
|
|||
|
|
# 授权是分级的：use 包含 read，manage 包含全部操作，export 仅允许导出和读取。
|
|||
|
|
allowed_by_operation = {
|
|||
|
|
"read": {"read", "use", "export", "manage"},
|
|||
|
|
"use": {"use", "manage"},
|
|||
|
|
"export": {"export", "manage"},
|
|||
|
|
"manage": {"manage"},
|
|||
|
|
}
|
|||
|
|
return grant_permission in allowed_by_operation.get(operation, {"manage"})
|
|||
|
|
|
|||
|
|
|
|||
|
|
def visibility_for_create(request: Request | None, requested: object = "private") -> str:
|
|||
|
|
value = str(requested or "private").strip().lower()
|
|||
|
|
if value not in {"private", "shared", "global"}:
|
|||
|
|
raise HTTPException(422, "可见性必须是 private、shared 或 global")
|
|||
|
|
# Creation/upload is always private. Visibility changes belong exclusively to
|
|||
|
|
# the administrator's Access > Resource access workflow (publish/grants).
|
|||
|
|
# Never trust a client-supplied visibility, including an administrator client.
|
|||
|
|
return "private"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def require_kb(session: Session, kb_id: int, request: Request | None,
|
|||
|
|
operation: str = "read") -> KnowledgeBase:
|
|||
|
|
kb = session.get(KnowledgeBase, kb_id)
|
|||
|
|
if not kb:
|
|||
|
|
raise HTTPException(404, "知识库不存在")
|
|||
|
|
user = request_user(request)
|
|||
|
|
if user and user.role != "admin" and session.exec(select(ResourceRevision).where(
|
|||
|
|
ResourceRevision.resource_type == "knowledge_base", ResourceRevision.resource_id == kb_id,
|
|||
|
|
ResourceRevision.owner_user_id == user.id, ResourceRevision.operation == "delete",
|
|||
|
|
ResourceRevision.status.in_(["pending", "approved"])) ).first():
|
|||
|
|
raise HTTPException(404, "知识库不存在或当前用户已隐藏该资源")
|
|||
|
|
grant = None
|
|||
|
|
if user and kb.visibility == "shared":
|
|||
|
|
grant = session.exec(select(KnowledgeBaseGrant).where(
|
|||
|
|
KnowledgeBaseGrant.kb_id == kb_id,
|
|||
|
|
KnowledgeBaseGrant.user_id == user.id,
|
|||
|
|
KnowledgeBaseGrant.enabled == True,
|
|||
|
|
)).first() # noqa: E712
|
|||
|
|
if (not can_access(kb, user, operation, grant.permission if grant else None)
|
|||
|
|
and not canvas_allows_resource(session, kb, user, operation)):
|
|||
|
|
raise HTTPException(404, "知识库不存在或当前用户没有访问权限")
|
|||
|
|
return kb
|
|||
|
|
|
|||
|
|
|
|||
|
|
def require_memory(session: Session, memory_id: int, request: Request | None,
|
|||
|
|
operation: str = "read") -> Memory:
|
|||
|
|
memory = session.get(Memory, memory_id)
|
|||
|
|
if not memory:
|
|||
|
|
raise HTTPException(404, "记忆不存在")
|
|||
|
|
user = request_user(request)
|
|||
|
|
if user and user.role != "admin" and session.exec(select(ResourceRevision).where(
|
|||
|
|
ResourceRevision.resource_type == "memory", ResourceRevision.resource_id == memory_id,
|
|||
|
|
ResourceRevision.owner_user_id == user.id, ResourceRevision.operation == "delete",
|
|||
|
|
ResourceRevision.status.in_(["pending", "approved"])) ).first():
|
|||
|
|
raise HTTPException(404, "记忆不存在或当前用户已隐藏该资源")
|
|||
|
|
grant = None
|
|||
|
|
if user and memory.visibility == "shared":
|
|||
|
|
grant = session.exec(select(MemoryGrant).where(
|
|||
|
|
MemoryGrant.memory_id == memory_id,
|
|||
|
|
MemoryGrant.user_id == user.id,
|
|||
|
|
MemoryGrant.enabled == True,
|
|||
|
|
)).first() # noqa: E712
|
|||
|
|
if (not can_access(memory, user, operation, grant.permission if grant else None)
|
|||
|
|
and not canvas_allows_resource(session, memory, user, operation)):
|
|||
|
|
raise HTTPException(404, "记忆不存在或当前用户没有访问权限")
|
|||
|
|
return memory
|
|||
|
|
|
|||
|
|
def skill_grant(session: Session, skill_id: int, user: AppUser | None) -> AgentSkillGrant | None:
|
|||
|
|
if not user:
|
|||
|
|
return None
|
|||
|
|
return session.exec(select(AgentSkillGrant).where(
|
|||
|
|
AgentSkillGrant.skill_id == skill_id,
|
|||
|
|
AgentSkillGrant.user_id == user.id,
|
|||
|
|
AgentSkillGrant.enabled == True,
|
|||
|
|
)).first() # noqa: E712
|
|||
|
|
|
|||
|
|
def agent_grant(session: Session, agent_id: int, user: AppUser | None) -> AgentGrant | None:
|
|||
|
|
if not user:
|
|||
|
|
return None
|
|||
|
|
return session.exec(select(AgentGrant).where(
|
|||
|
|
AgentGrant.agent_id == agent_id,
|
|||
|
|
AgentGrant.user_id == user.id,
|
|||
|
|
AgentGrant.enabled == True,
|
|||
|
|
)).first() # noqa: E712
|
|||
|
|
|
|||
|
|
def can_access_agent(session: Session, agent: Agent, user: AppUser | None,
|
|||
|
|
operation: str = "read") -> bool:
|
|||
|
|
grant = agent_grant(session, int(agent.id or 0), user) if getattr(agent, "visibility", "private") == "shared" else None
|
|||
|
|
return can_access(agent, user, operation, grant.permission if grant else None)
|
|||
|
|
|
|||
|
|
def require_skill(session: Session, skill_id: int, request: Request | None,
|
|||
|
|
operation: str = "read") -> AgentSkill:
|
|||
|
|
skill = session.get(AgentSkill, skill_id)
|
|||
|
|
if not skill:
|
|||
|
|
raise HTTPException(404, "技能不存在")
|
|||
|
|
user = request_user(request)
|
|||
|
|
grant = skill_grant(session, skill_id, user) if skill.visibility == "shared" else None
|
|||
|
|
if not can_access(skill, user, operation, grant.permission if grant else None):
|
|||
|
|
raise HTTPException(404, "技能不存在或当前用户没有访问权限")
|
|||
|
|
return skill
|
|||
|
|
|
|||
|
|
def canvas_grant(session: Session, canvas_id: int, user: AppUser | None) -> CanvasGrant | None:
|
|||
|
|
if not user:
|
|||
|
|
return None
|
|||
|
|
return session.exec(select(CanvasGrant).where(
|
|||
|
|
CanvasGrant.canvas_id == canvas_id, CanvasGrant.user_id == user.id,
|
|||
|
|
CanvasGrant.enabled == True,
|
|||
|
|
)).first() # noqa: E712
|
|||
|
|
|
|||
|
|
def can_use_canvas(session: Session, canvas: Canvas, user: AppUser | None, operation: str = "read") -> bool:
|
|||
|
|
"""画布授权决定整条 Agent 流程及其关联知识/记忆的可见范围。"""
|
|||
|
|
grant = canvas_grant(session, int(canvas.id or 0), user)
|
|||
|
|
return can_access(canvas, user, operation, grant.permission if grant else None)
|
|||
|
|
|
|||
|
|
def canvas_allows_resource(session: Session, resource: Any, user: AppUser | None, operation: str) -> bool:
|
|||
|
|
"""画布授权会连带开放节点 Agent 所挂接的知识库和确认记忆。"""
|
|||
|
|
if not user or user.role == "admin":
|
|||
|
|
return False
|
|||
|
|
# 画布授权只代表查看/运行依赖，不把依赖资源的编辑、删除或导出权一起放大。
|
|||
|
|
if operation not in {"read", "use"}:
|
|||
|
|
return False
|
|||
|
|
if int(getattr(resource, "organization_id", 1)) not in {0, int(user.organization_id or 1)}:
|
|||
|
|
return False
|
|||
|
|
canvas_operation = "use" if operation == "use" else "read"
|
|||
|
|
for canvas in session.exec(select(Canvas)).all():
|
|||
|
|
if not can_use_canvas(session, canvas, user, canvas_operation):
|
|||
|
|
continue
|
|||
|
|
for node in canvas.nodes or []:
|
|||
|
|
agent = session.get(Agent, node.get("agent_id"))
|
|||
|
|
if not agent:
|
|||
|
|
continue
|
|||
|
|
if isinstance(resource, KnowledgeBase) and resource.id in (agent.kb_ids or []):
|
|||
|
|
return True
|
|||
|
|
if isinstance(resource, Memory) and memory_belongs_to_agent_scope(resource, agent):
|
|||
|
|
return True
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def memory_belongs_to_agent_scope(memory: Memory, agent: Agent) -> bool:
|
|||
|
|
"""画布携带配置者沉淀的 Agent 经验，不携带其他使用者的私有偏好。"""
|
|||
|
|
return (memory.agent_id == agent.id
|
|||
|
|
and int(memory.organization_id) in {0, int(agent.organization_id or 1)}
|
|||
|
|
and int(memory.owner_user_id or 0) == int(agent.owner_user_id or 0))
|
|||
|
|
|
|||
|
|
def require_canvas(session: Session, canvas_id: int, request: Request | None,
|
|||
|
|
operation: str = "read") -> Canvas:
|
|||
|
|
canvas = session.get(Canvas, canvas_id)
|
|||
|
|
if not canvas or not can_use_canvas(session, canvas, request_user(request), operation):
|
|||
|
|
raise HTTPException(404, "画布不存在或当前用户没有访问权限")
|
|||
|
|
return canvas
|
|||
|
|
|
|||
|
|
|
|||
|
|
def visible_kb_ids(session: Session, request: Request | None, operation: str = "read") -> set[int]:
|
|||
|
|
user = request_user(request)
|
|||
|
|
if user is None:
|
|||
|
|
return {int(item.id) for item in session.exec(select(KnowledgeBase)).all() if item.id}
|
|||
|
|
result: set[int] = set()
|
|||
|
|
for kb in session.exec(select(KnowledgeBase).where(
|
|||
|
|
KnowledgeBase.organization_id.in_([0, int(user.organization_id or 1)])
|
|||
|
|
)).all():
|
|||
|
|
if user.role != "admin" and session.exec(select(ResourceRevision).where(
|
|||
|
|
ResourceRevision.resource_type == "knowledge_base", ResourceRevision.resource_id == kb.id,
|
|||
|
|
ResourceRevision.owner_user_id == user.id, ResourceRevision.operation == "delete",
|
|||
|
|
ResourceRevision.status.in_(["pending", "approved"])) ).first():
|
|||
|
|
continue
|
|||
|
|
grant = session.exec(select(KnowledgeBaseGrant).where(
|
|||
|
|
KnowledgeBaseGrant.kb_id == kb.id, KnowledgeBaseGrant.user_id == user.id,
|
|||
|
|
KnowledgeBaseGrant.enabled == True,
|
|||
|
|
)).first() if kb.visibility == "shared" else None # noqa: E712
|
|||
|
|
if (can_access(kb, user, operation, grant.permission if grant else None)
|
|||
|
|
or canvas_allows_resource(session, kb, user, operation)) and kb.id:
|
|||
|
|
result.add(kb.id)
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
|
|||
|
|
def visible_memory_ids(session: Session, request: Request | None, operation: str = "read") -> set[int]:
|
|||
|
|
user = request_user(request)
|
|||
|
|
if user is None:
|
|||
|
|
return {int(item.id) for item in session.exec(select(Memory)).all() if item.id}
|
|||
|
|
result: set[int] = set()
|
|||
|
|
for memory in session.exec(select(Memory).where(
|
|||
|
|
Memory.organization_id.in_([0, int(user.organization_id or 1)])
|
|||
|
|
)).all():
|
|||
|
|
if user.role != "admin" and session.exec(select(ResourceRevision).where(
|
|||
|
|
ResourceRevision.resource_type == "memory", ResourceRevision.resource_id == memory.id,
|
|||
|
|
ResourceRevision.owner_user_id == user.id, ResourceRevision.operation == "delete",
|
|||
|
|
ResourceRevision.status.in_(["pending", "approved"])) ).first():
|
|||
|
|
continue
|
|||
|
|
grant = session.exec(select(MemoryGrant).where(
|
|||
|
|
MemoryGrant.memory_id == memory.id, MemoryGrant.user_id == user.id,
|
|||
|
|
MemoryGrant.enabled == True,
|
|||
|
|
)).first() if memory.visibility == "shared" else None # noqa: E712
|
|||
|
|
if (can_access(memory, user, operation, grant.permission if grant else None)
|
|||
|
|
or canvas_allows_resource(session, memory, user, operation)) and memory.id:
|
|||
|
|
result.add(memory.id)
|
|||
|
|
return result
|