身份验证与访问控制

from langgraph_sdk import Auth

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    # 验证凭据（例如，API 密钥、JWT 令牌）
    api_key = headers.get(b"x-api-key")
    if not api_key or not is_valid_key(api_key):
        raise Auth.exceptions.HTTPException(
            status_code=401,
            detail="无效的 API 密钥"
        )

    # 返回用户信息 - 仅 identity 和 is_authenticated 是必需的
    # 添加你授权所需的任何额外字段
    return {
        "identity": "user-123",        # 必需：唯一的用户标识符
        "is_authenticated": True,      # 可选：默认为 True
        "permissions": ["read", "write"], # 可选：用于基于权限的授权
        # 如果你想实现其他授权模式，可以添加更多自定义字段
        "role": "admin",
        "org_id": "org-456"

    }

@auth.on
async def add_owner(
    ctx: Auth.types.AuthContext,
    value: dict  # 发送到此访问方法的负载
) -> dict:  # 返回一个过滤器字典，限制对资源的访问
    """授权对线程、运行、定时任务和助手的所有访问。

    此处理程序做两件事：
        - 向资源元数据添加一个值（以便与资源一起持久化，以便以后可以过滤）
        - 返回一个过滤器（以限制对现有资源的访问）

    参数：
        ctx: 包含用户信息、权限、路径和
        value: 发送到端点的请求负载。对于创建
              操作，这包含资源参数。对于读取
              操作，这包含正在访问的资源。

    返回：
        一个过滤器字典，LangGraph 用它来限制对资源的访问。
        有关支持的运算符，请参阅[过滤器操作](#过滤器操作)。
    """
    # 创建过滤器，仅限制为此用户的资源
    filters = {"owner": ctx.user.identity}

    # 获取或创建负载中的元数据字典
    # 这是我们存储有关资源的持久信息的地方
    metadata = value.setdefault("metadata", {})

    # 将所有者添加到元数据 - 如果这是创建或更新操作，
    # 此信息将与资源一起保存
    # 这样我们以后可以在读取操作中按它过滤
    metadata.update(filters)

    # 返回过滤器以限制访问
    # 这些过滤器应用于所有操作（创建、读取、更新、搜索等）
    # 以确保用户只能访问自己的资源
    return filters

# 通用 / 全局处理程序捕获未被更具体处理程序处理的调用
@auth.on
async def reject_unhandled_requests(ctx: Auth.types.AuthContext, value: Any) -> False:
    print(f"请求到 {ctx.path} 来自 {ctx.user.identity}")
    raise Auth.exceptions.HTTPException(
        status_code=403,
        detail="禁止访问"
    )

# 匹配 "thread" 资源和所有操作 - 创建、读取、更新、删除、搜索
# 由于这比通用的 @auth.on 处理程序**更具体**，它将优先于
# 通用处理程序，用于 "threads" 资源上的所有操作
@auth.on.threads
async def on_thread(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create.value
):
    # 在正在创建的线程上设置元数据
    # 将确保资源包含一个 "owner" 字段
    # 然后，每当用户尝试访问此线程或线程内的运行时，
    # 我们可以按所有者过滤
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}


# 线程创建。这将仅匹配线程创建操作
# 由于这比通用的 @auth.on 处理程序和 @auth.on.threads 处理程序都**更具体**，
# 它将优先用于 "threads" 资源上的任何 "create" 操作
@auth.on.threads.create
async def on_thread_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create.value
):
    # 如果用户没有写权限，则拒绝
    if "write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="用户缺少所需权限。"
        )
    # 在正在创建的线程上设置元数据
    # 将确保资源包含一个 "owner" 字段
    # 然后，每当用户尝试访问此线程或线程内的运行时，
    # 我们可以按所有者过滤
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

# 读取线程。由于这也比通用的 @auth.on 处理程序和 @auth.on.threads 处理程序更具体，
# 它将优先用于 "threads" 资源上的任何 "read" 操作
@auth.on.threads.read
async def on_thread_read(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.read.value
):
    # 由于我们正在读取（而不是创建）一个线程，
    # 我们不需要设置元数据。我们只需要
    # 返回一个过滤器以确保用户只能看到自己的线程
    return {"owner": ctx.user.identity}

# 运行创建、流式传输、更新等。
# 这优先于通用的 @auth.on 处理程序和 @auth.on.threads 处理程序
@auth.on.threads.create_run
async def on_run_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.threads.create_run.value
):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    # 继承线程的访问控制
    return {"owner": ctx.user.identity}

# 助手创建
@auth.on.assistants.create
async def on_assistant_create(
    ctx: Auth.types.AuthContext,
    value: Auth.types.assistants.create.value
):
    if "assistants:create" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="用户缺少所需权限。"
        )

@auth.on
async def owner_only(ctx: Auth.types.AuthContext, value: dict):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

# 在你的身份验证处理程序中：
@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    ...
    return {
        "identity": "user-123",
        "is_authenticated": True,
        "permissions": ["threads:write", "threads:read"]  # 在身份验证中定义权限
    }

def _default(ctx: Auth.types.AuthContext, value: dict):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity
    return {"owner": ctx.user.identity}

@auth.on.threads.create
async def create_thread(ctx: Auth.types.AuthContext, value: dict):
    if "threads:write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="未授权"
        )
    return _default(ctx, value)


@auth.on.threads.read
async def rbac_create(ctx: Auth.types.AuthContext, value: dict):
    if "threads:read" not in ctx.permissions and "threads:write" not in ctx.permissions:
        raise Auth.exceptions.HTTPException(
            status_code=403,
            detail="未授权"
        )
    return _default(ctx, value)