把云厂商身份换成自己的:托管计算环境中的工作负载证明

2026-09-26 25 预计阅读时间: 1 分钟
来源: netflixtechblog.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 分钟

托管虚拟机、容器平台和函数计算通常都能向工作负载签发云身份令牌。问题在于,这些令牌的签发者、声明格式和生命周期都由云厂商决定。如果内部服务直接依赖它们,认证逻辑很快会被某一家云的 IAM 模型绑住。

更可控的做法是增加一个证明交换层:工作负载先向云平台取得短期身份凭据,再把它交给组织自己的认证服务验证;验证通过后,由后者签发内部统一使用的工作负载令牌。这不是简单地“换一张 JWT”,而是把云平台提供的运行环境证据,转换成组织可以治理的身份。

身份交换解决的不是令牌格式,而是信任边界

一条典型链路可以表示为:

托管工作负载
  -> 获取云平台短期身份令牌
  -> 调用内部 attestation/exchange 服务
  -> 服务验证签名、issuer、audience 和工作负载声明
  -> 根据映射策略生成内部 workload_id
  -> 签发短期内部令牌
  -> 工作负载访问内部 API

这里存在两个不同的信任域:

  • 云平台信任域负责证明“这段代码运行在某个受管环境中”。证据可能包含项目、账号、服务身份、实例或部署信息。
  • 组织内部信任域负责决定“这个运行实体在公司系统中是谁,以及可以做什么”。内部身份可以保持跨云一致,例如 payments/reconciler/prod。

交换服务应验证云令牌,而不是只解码它。最低限度需要检查:

  1. 签名是否来自预先配置的云端 JWKS;
  2. iss 是否是允许的签发者;
  3. aud 是否明确指向交换服务;
  4. exp、iat 等时间声明是否合理;
  5. 云端主体、项目和环境是否命中显式映射规则;
  6. 同一证明是否可能在窗口期内被重复使用。

如果只验证签名,却不验证 audience,一个原本签给其他服务的令牌也可能被拿来交换内部身份。这是此类系统中最常见、也最危险的边界错误之一。

不要把云声明原样传播到内部

内部令牌最好使用稳定、语义明确的身份,而不是复制几十个云厂商声明。例如,可以维护如下映射:

# identity-map.yaml
mappings:
  - cloud_subject: "cloud-service-account:reconciler-prod"
    internal_subject: "payments/reconciler/prod"
    allowed_audiences:
      - "ledger-api"
    scopes:
      - "ledger:read"
      - "reconciliation:write"

这层映射带来两个重要效果:

  • 云端服务账号改名、工作负载迁移到另一家云时,内部 API 不必同步修改认证逻辑;
  • 权限由内部策略决定,云身份只作为来源证明,不能自动继承过大的内部权限。

映射应默认拒绝未知主体。不要采用“任何来自某个云账号的工作负载都映射为同一个生产身份”这样的宽泛规则,否则一个低权限工作负载可能借交换服务完成横向移动。

可以这样实践:搭建一个最小令牌交换服务

下面是一个可改造的 FastAPI 示例。它假设云平台能够向工作负载提供 JWT,并公开固定的 JWKS 地址。不同云平台获取令牌的方式不同,因此示例从 Authorization: Bearer 请求头接收令牌,而不假设具体 metadata API。

安装依赖并生成内部 Ed25519 签名密钥:

python -m venv .venv
. .venv/bin/activate
pip install fastapi 'uvicorn[standard]' pyjwt cryptography

openssl genpkey -algorithm ED25519 -out internal-private.pem
openssl pkey -in internal-private.pem -pubout -out internal-public.pem

保存为 app.py:

import os
import time
import uuid

import jwt
from fastapi import FastAPI, Header, HTTPException
from jwt import PyJWKClient

CLOUD_ISSUER = os.environ["CLOUD_ISSUER"]
CLOUD_AUDIENCE = os.getenv("CLOUD_AUDIENCE", "https://attestor.internal/exchange")
CLOUD_JWKS_URL = os.environ["CLOUD_JWKS_URL"]
INTERNAL_ISSUER = os.getenv("INTERNAL_ISSUER", "https://identity.internal")
INTERNAL_AUDIENCE = os.getenv("INTERNAL_AUDIENCE", "ledger-api")

# 生产环境应从数据库或策略引擎读取,并经过审计和变更审批。
SUBJECT_MAP = {
    "cloud-service-account:reconciler-prod": "payments/reconciler/prod",
}

with open("internal-private.pem", "rb") as f:
    SIGNING_KEY = f.read()

jwks_client = PyJWKClient(CLOUD_JWKS_URL)
app = FastAPI()


@app.post("/exchange")
def exchange(authorization: str = Header(default="")):
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "missing bearer token")

    cloud_token = authorization.removeprefix("Bearer ").strip()

    try:
        signing_key = jwks_client.get_signing_key_from_jwt(cloud_token)
        claims = jwt.decode(
            cloud_token,
            signing_key.key,
            algorithms=["RS256", "ES256"],
            issuer=CLOUD_ISSUER,
            audience=CLOUD_AUDIENCE,
            options={"require": ["exp", "iat", "iss", "aud", "sub"]},
        )
    except jwt.PyJWTError as exc:
        raise HTTPException(401, f"invalid cloud attestation: {exc}") from exc

    internal_subject = SUBJECT_MAP.get(claims["sub"])
    if internal_subject is None:
        raise HTTPException(403, "cloud subject is not mapped")

    now = int(time.time())
    internal_claims = {
        "iss": INTERNAL_ISSUER,
        "sub": internal_subject,
        "aud": INTERNAL_AUDIENCE,
        "iat": now,
        "nbf": now,
        "exp": now + 300,
        "jti": str(uuid.uuid4()),
        "scope": "ledger:read reconciliation:write",
        "attested_by": CLOUD_ISSUER,
    }
    token = jwt.encode(internal_claims, SIGNING_KEY, algorithm="EdDSA")
    return {"access_token": token, "token_type": "Bearer", "expires_in": 300}

配置环境变量后启动服务。请把示例值替换为实际云平台的 issuer、JWKS 地址和专用于交换服务的 audience:

export CLOUD_ISSUER='https://issuer.example-cloud.invalid'
export CLOUD_JWKS_URL='https://issuer.example-cloud.invalid/.well-known/jwks.json'
export CLOUD_AUDIENCE='https://attestor.internal/exchange'
export INTERNAL_ISSUER='https://identity.internal'
export INTERNAL_AUDIENCE='ledger-api'

uvicorn app:app --host 127.0.0.1 --port 8000

工作负载取得真实云令牌后,可以执行:

export CLOUD_TOKEN='替换为云平台签发的短期令牌'

curl --fail-with-body \
  -X POST http://127.0.0.1:8000/exchange \
  -H "Authorization: Bearer ${CLOUD_TOKEN}"

这个示例展示的是信任链骨架,而不是完整生产实现。实际系统还应缓存并安全刷新 JWKS、限制请求速率、记录映射决策、发布内部公钥,并为 kid、密钥轮换和撤销设计明确流程。

上线前重点检查这些边界

证明交换服务会成为高价值安全组件,部署时应重点处理以下问题:

  • 令牌寿命:内部令牌通常只需存活数分钟,不应把短期云证明换成长达数小时的凭据。
  • 重放防护:对高风险操作,可缓存外部令牌的 jti 或摘要,并拒绝重复交换;还可以要求一次性 nonce。
  • 权限收敛:根据内部主体、目标 audience 和环境计算 scope,不要相信调用方提交的权限列表。
  • 来源约束:除 sub 外,还可验证租户、项目、区域、部署或镜像身份,但只应依赖云平台签名保护且语义稳定的声明。
  • 密钥隔离:内部签名私钥应放入 KMS、HSM 或专用密钥服务,不应像示例一样保存在本地文件。
  • 可观测性与隐私:记录 issuer、映射结果和拒绝原因,但不要把完整 bearer token 写入日志。
  • 故障策略:JWKS 获取失败或策略服务不可用时应默认拒绝,而不是跳过验证。

最稳妥的落地顺序,是先选择一个权限较低、调用链清晰的内部 API 做试点。明确云身份到内部身份的映射,强制校验专用 audience,将令牌有效期控制在分钟级,再逐步加入重放检测、密钥轮换和策略引擎。这样,“使用云身份”不会演变成“被云身份模型永久绑定”,而会成为组织自有零信任体系中的一条可验证输入。


相关推荐