用 OBO 令牌交换为多租户 Bedrock AgentCore 智能体建立纵深授权

2026-07-14 41 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:10 分钟

多租户智能体一旦开始调用企业 API,仅验证“请求来自某个已登录用户”就不够了。系统还需要回答三个问题:用户属于哪个租户、智能体是否可以代表该用户调用目标服务,以及这枚令牌是否只能交给指定 API 使用。

Amazon Bedrock AgentCore Gateway 的 on-behalf-of(OBO,代表用户)令牌交换把这些问题落到可执行的授权链路上。入口令牌负责证明用户与租户身份,Gateway interceptor 完成校验和交换,下游令牌则通过 aud、租户声明和权限范围绑定到具体服务。这样即使某一跳的令牌泄露,其可利用范围也会受到限制。

一条请求经过哪些身份变化

可以把完整链路拆成四个参与者:

  1. 用户在 Okta 完成登录,客户端获得面向 AgentCore Gateway 的访问令牌。
  2. 客户端携带该令牌调用多租户智能体。
  3. Gateway interceptor 校验签名、发行方、受众、租户和权限,并向 Okta 发起 OBO token exchange。
  4. Gateway 使用交换后的令牌调用目标 API;目标 API再次验证自己的受众和租户边界。

入口令牌可以抽象为:

{
  "iss": "https://example.okta.com/oauth2/default",
  "sub": "00u123456789",
  "aud": "agentcore-gateway",
  "tenant_id": "tenant-a",
  "scp": ["agent:invoke"],
  "exp": 1735689600
}

交换后,给订单 API 的令牌应收窄受众和权限:

{
  "iss": "https://example.okta.com/oauth2/default",
  "sub": "00u123456789",
  "aud": "orders-api",
  "tenant_id": "tenant-a",
  "scp": ["orders:read"],
  "act": {
    "sub": "agentcore-gateway"
  },
  "exp": 1735688700
}

这里最关键的变化不是令牌格式,而是授权语义:sub 仍然代表最终用户,act 可以记录执行代理,aud 从 Gateway 变成订单 API,权限也从调用智能体收窄为读取订单。实际声明名称和 Okta 授权服务器配置可能不同,应以组织中的 claim mapping 为准。

受众绑定为什么是纵深防御

只依赖 scope 容易留下横向调用空间。假设库存 API 和订单 API 都接受 data:read,一枚发给库存 API 的令牌就可能被错误地拿去调用订单 API。

受众绑定要求每个服务同时检查:

  • iss 是否来自受信任的 Okta 授权服务器;
  • JWT 签名和有效期是否合法;
  • aud 是否精确包含当前服务的资源标识;
  • tenant_id 是否与请求资源所属租户一致;
  • scope 是否覆盖当前操作;
  • 代理身份或交换链路是否符合策略。

因此,一枚 aud=orders-api 的令牌即便被送到 billing-api,也应在鉴权阶段被拒绝。租户声明提供数据边界,受众提供服务边界,scope 提供操作边界,三者不能互相替代。

在多租户场景中,还要避免让客户端自由指定租户。tenant_id 应来自经过验证的入口令牌或服务端租户目录,不能直接信任 HTTP Header、请求体或智能体生成的参数。

可以这样实践:调用 Okta Token Exchange

下面示例按照 OAuth 2.0 Token Exchange 的常见参数组织,可直接改造成部署脚本。运行前需要确认你的 Okta 授权服务器已经启用对应的 token exchange 能力,并按实际配置调整授权服务器 ID、客户端认证方式、受众和 scope。

#!/usr/bin/env bash
set -euo pipefail

: "${OKTA_DOMAIN:?Set OKTA_DOMAIN, for example example.okta.com}"
: "${OKTA_AUTH_SERVER_ID:=default}"
: "${OBO_CLIENT_ID:?Set OBO_CLIENT_ID}"
: "${OBO_CLIENT_SECRET:?Set OBO_CLIENT_SECRET}"
: "${SUBJECT_TOKEN:?Set SUBJECT_TOKEN to the inbound access token}"

curl --fail-with-body --silent --show-error \
  --request POST \
  "https://${OKTA_DOMAIN}/oauth2/${OKTA_AUTH_SERVER_ID}/v1/token" \
  --user "${OBO_CLIENT_ID}:${OBO_CLIENT_SECRET}" \
  --header "Content-Type: application/x-www-form-urlencoded" \
  --data-urlencode "grant_type=urn:ietf:params:oauth:grant-type:token-exchange" \
  --data-urlencode "subject_token=${SUBJECT_TOKEN}" \
  --data-urlencode "subject_token_type=urn:ietf:params:oauth:token-type:access_token" \
  --data-urlencode "requested_token_type=urn:ietf:params:oauth:token-type:access_token" \
  --data-urlencode "audience=orders-api" \
  --data-urlencode "scope=orders:read"

不要把客户端密钥写入 Agent prompt、工具参数或日志。生产环境应从 AWS Secrets Manager 等密钥存储中读取,并限制只有 interceptor 的执行身份可以访问。

为了在联调时查看每一跳的声明变化,可以使用下面的 Python 工具。它只解码 payload,不验证签名,因此只能用于诊断,不能替代服务端 JWT 校验。

#!/usr/bin/env python3
import base64
import json
import sys


def decode_segment(segment: str) -> dict:
    padding = "=" * (-len(segment) % 4)
    raw = base64.urlsafe_b64decode(segment + padding)
    return json.loads(raw)


if len(sys.argv) != 2:
    raise SystemExit("usage: python inspect_jwt.py <jwt>")

parts = sys.argv[1].split(".")
if len(parts) != 3:
    raise SystemExit("not a compact JWT")

print(json.dumps(decode_segment(parts[1]), indent=2, ensure_ascii=False))

将其保存为 inspect_jwt.py 后,可以分别检查入口令牌和交换结果:

python inspect_jwt.py "$SUBJECT_TOKEN"
python inspect_jwt.py "$(jq -r '.access_token' token-response.json)"

检查时重点比较 isssubaudtenant_idscpactiatexp。诊断输出中不要记录完整 JWT,因为访问令牌本身就是凭据。

Interceptor 应承担什么责任

AgentCore Gateway interceptor 位于信任边界上,不应只是把入口 Authorization Header 原样转发给下游。可以按下面的伪代码组织实现;其中框架适配层需要替换为实际 AgentCore Gateway interceptor 接口:

from dataclasses import dataclass


@dataclass(frozen=True)
class TargetPolicy:
    audience: str
    scopes: tuple[str, ...]


TARGETS = {
    "get_orders": TargetPolicy("orders-api", ("orders:read",)),
    "get_invoice": TargetPolicy("billing-api", ("invoices:read",)),
}


def authorize_and_exchange(request, jwt_verifier, token_client):
    inbound = request.bearer_token
    claims = jwt_verifier.verify(
        inbound,
        issuer="https://example.okta.com/oauth2/default",
        audience="agentcore-gateway",
    )

    tenant_id = claims.get("tenant_id")
    if not tenant_id:
        raise PermissionError("missing tenant claim")
    if "agent:invoke" not in claims.get("scp", []):
        raise PermissionError("missing agent invocation scope")

    policy = TARGETS.get(request.tool_name)
    if policy is None:
        raise PermissionError("tool is not mapped to a target policy")

    exchanged = token_client.exchange(
        subject_token=inbound,
        audience=policy.audience,
        scopes=policy.scopes,
    )

    return {
        "Authorization": f"Bearer {exchanged.access_token}",
        # 仅在下游把它当作辅助信息时发送;授权仍以已签名 JWT 为准。
        "X-Tenant-Id": tenant_id,
    }

这段示例表达了几项必须保留的约束:工具到受众和 scope 的映射由服务端维护;未知工具默认拒绝;租户来自已验证声明;每个目标服务获得不同受众的令牌。不要让大模型直接决定 audience 或请求任意 scope,否则 prompt injection 可能升级成权限提升。

交换后的令牌也不应无条件缓存。若要降低 Okta 请求压力,缓存键至少应包含用户、租户、受众、scope 集合和入口令牌授权上下文,缓存寿命不得超过交换令牌或入口令牌的剩余有效期。用户被禁用或权限撤销后,长时间缓存还会扩大撤销延迟。

上线前的检查清单

  • 为 Gateway 和每个下游 API 分配不同的 audience。
  • 在每一跳同时验证签名、issaudexp 和必要 scope。
  • 从已验证 JWT 或租户目录解析租户,不信任模型输出和客户端自报租户。
  • 在服务端固定“工具到 audience/scope”的白名单映射。
  • 使用短生命周期令牌,并让缓存过期时间服从最短有效期。
  • 日志记录交换结果、用户 ID、租户 ID 和目标受众,但不记录完整令牌或客户端密钥。
  • 测试跨租户访问、错误 audience、过期令牌、未知工具和 scope 提权请求。
  • 让下游 API 执行最终授权;Gateway 校验不能取代资源服务自己的检查。

OBO token exchange 的价值不只是“替智能体换一枚令牌”。它把用户、代理、租户、目标服务和具体操作绑定到一条可审计的授权链上。对多租户系统而言,真正可靠的实现标准是:任何令牌离开预期的租户、服务或操作范围后,都无法继续工作。


相关推荐