如何按 Usage Plan 限制单个 API endpoint 的访问权限

结论

S3 background polling 配合内存 allowlist 是可行的。授权状态会生成只读快照，并分发到各个服务实例，这是常见的 control plane 与 data plane 分离方式。请求处理期间只需访问本地内存，不受 API Gateway Management API 配额影响。

Usage Plan 适合管理 quota 和 throttling，不适合判断某个客户能否访问特定 route。可以按下面的方式拆分：

• API Gateway Usage Plan 负责配额和限流。
• 独立的 entitlement 数据记录 Standard、Premium 等功能权限。
• /premium-query 由 Lambda authorizer 或 Node.js origin 检查 entitlement。
• Standard 客户访问该 endpoint 时返回 403 Forbidden。
• 客户升级时，同时更新 Usage Plan 和 entitlement，Base URL 与 API key 保持不变。

客户数量不多且可以接受几十秒的权限传播延迟时，可以使用 S3 快照。如果权限需要立即生效，通常应将 entitlement 存入 DynamoDB，并搭配 Lambda authorizer 或应用内缓存。

Usage Plan 为什么不适合承担 route authorization

API Gateway Usage Plan 主要管理两类流量限制：

1. API key 在指定 stage 上可使用的 quota。
2. API key 访问 API 的 rate。

Method-level throttling 仍属于流量控制。如果将 /premium-query 的 rate limit 配置为 0，API Gateway 会按限流失败处理，返回 429 Too Many Requests，与实际情况不符。

客户已经通过身份识别，只是没有 Premium 权限时，应返回：

HTTP/1.1 403 Forbidden
Content-Type: application/json

{
  "code": "PREMIUM_PLAN_REQUIRED",
  "message": "This endpoint requires a Premium plan."
}

429 应用于真正的 rate limit 或 quota 问题。客户端通常会重试 429，而套餐权限不足无法通过重试解决。

独立维护 entitlement

可以将客户权限建模为独立数据，避免在请求过程中反查 Usage Plan：

API key ID -> customer ID -> plan/entitlements

例如：

{
  "apiKeyId": "abc123",
  "customerId": "cus_001",
  "plan": "premium",
  "entitlements": [
    "premium-query"
  ],
  "updatedAt": "2026-10-28T10:30:00Z"
}

订阅状态变化时，provisioning service 需要完成两项操作：

1. 更新 API Gateway API key 与 Usage Plan 的关联，用于 quota 和 throttling。
2. 更新 entitlement store，用于 endpoint authorization。

两次更新之间可能出现短暂的不一致。因此，webhook 处理需要支持幂等，并保存 subscription event ID、版本号或更新时间，防止旧事件覆盖新状态。

方案一：在 Node.js origin 中检查 S3 快照

S3 快照适合读取频繁、变更较少且允许短暂延迟的权限数据。对象中应包含版本信息，不要只存一个数组：

{
  "version": 42,
  "generatedAt": "2026-10-28T10:30:00Z",
  "premiumApiKeyIds": [
    "abc123",
    "def456"
  ]
}

Node.js 服务必须在启动时完成首次加载。随后定期使用 If-None-Match 轮询。收到新数据后，先构造新的 Set，通过校验再一次性替换旧引用。

import { GetObjectCommand, S3Client } from "@aws-sdk/client-s3";

const s3 = new S3Client({});
const bucket = process.env.ENTITLEMENT_BUCKET;
const key = process.env.ENTITLEMENT_KEY;

let premiumApiKeyIds = new Set();
let currentEtag;
let lastSuccessfulLoadAt = 0;
let initialized = false;

async function streamToJson(body) {
  const text = await body.transformToString();
  return JSON.parse(text);
}

async function refreshEntitlements() {
  try {
    const response = await s3.send(
      new GetObjectCommand({
        Bucket: bucket,
        Key: key,
        IfNoneMatch: currentEtag
      })
    );

    const snapshot = await streamToJson(response.Body);

    if (!Array.isArray(snapshot.premiumApiKeyIds)) {
      throw new Error("Invalid entitlement snapshot");
    }

    const nextSet = new Set(snapshot.premiumApiKeyIds);

    premiumApiKeyIds = nextSet;
    currentEtag = response.ETag;
    lastSuccessfulLoadAt = Date.now();
    initialized = true;
  } catch (error) {
    if (error?.$metadata?.httpStatusCode === 304) {
      lastSuccessfulLoadAt = Date.now();
      return;
    }

    console.error("Failed to refresh entitlements", error);
  }
}

export async function initializeEntitlements() {
  await refreshEntitlements();

  if (!initialized) {
    throw new Error("Unable to load initial entitlement snapshot");
  }

  setInterval(() => {
    refreshEntitlements().catch((error) => {
      console.error("Unexpected entitlement refresh error", error);
    });
  }, 60_000).unref();
}

export function requirePremium(req, res, next) {
  const apiKeyId = req.get("x-api-key-id");

  if (!apiKeyId || !premiumApiKeyIds.has(apiKeyId)) {
    return res.status(403).json({
      code: "PREMIUM_PLAN_REQUIRED",
      message: "This endpoint requires a Premium plan."
    });
  }

  return next();
}

在 Express 中，只对目标 route 使用 middleware：

app.get(
  "/premium-query",
  requirePremium,
  async (req, res) => {
    const result = await runPremiumQuery(req.query);
    res.json(result);
  }
);

服务启动时要等待初始快照加载完成：

await initializeEntitlements();

app.listen(3000, () => {
  console.log("API listening on port 3000");
});

这样能防止新实例使用空集合，误拒所有 Premium 请求。

不要信任客户端提交的 x-api-key-id

x-api-key-id 必须由 API Gateway 根据已经验证的 API key 生成。如果允许客户端自行设置，Standard 客户就可能提交其他 API key ID 绕过检查。

对于 API Gateway REST API，可以在 integration request parameter mapping 中，将 API Gateway context 里的 API key ID 映射到内部 header。具体的 CloudFormation、OpenAPI 或控制台配置语法取决于集成方式，应按实际部署类型配置。

实现时需要满足以下条件：

• API Gateway 覆盖客户端传入的同名 header。
• origin 只能通过 Internal NLB 或其他受控网络路径访问。
• 客户端不能绕过 API Gateway 直接连接 origin。
• 内部服务接收 API key ID，不接收原始 API key value。
• 日志不得记录原始 API key value。

x-api-key-id 这个名称本身不代表可信。信任来自 API Gateway 对 header 的覆盖和 origin 的网络隔离。

S3 方案需要处理的问题

权限传播延迟

轮询间隔为 60 秒时，升级或降级最多需要等待约一个轮询周期，此外还要算上各实例的刷新时间。

升级延迟主要影响使用体验，降级延迟可能带来安全风险。Premium 权限必须立即撤销时，60 秒轮询可能无法满足要求。

可以缩短轮询间隔，也可以在权限变更后通过 SNS、SQS、EventBridge 等机制通知实例刷新。通知机制仍需搭配周期性同步，以免消息丢失后，实例长期保留旧数据。

多实例状态不一致

每个 Node.js 实例独立轮询，短时间内可能出现以下情况：

• 实例 A 已加载新快照。
• 实例 B 仍在使用旧快照。
• 同一客户的连续请求得到不同结果。

业务能够接受短暂的最终一致性时，这类差异通常可以接受。权限必须严格一致时，应改用共享的低延迟数据存储，或交给 authorizer 统一判断。

刷新失败策略

一次 S3 请求失败后，不要清空现有的 Set。应继续使用 last known good snapshot，并监控该快照的年龄。

还需要设定最大陈旧时间。例如，快照超过 10 分钟仍未刷新成功时，可以采用以下策略之一：

• fail closed：拒绝 /premium-query。安全性更高，但会影响 Premium 客户。
• fail open：继续使用旧快照。可用性更高，但已降级或停用的客户可能继续访问。

商业权限通常会选择 fail closed，具体仍取决于该 endpoint 的敏感程度。

快照规模

覆盖和解析完整 JSON 的成本会随客户数量线性增长。几千或几万个 API key ID 通常容易处理，但不能由此认定该方案适用于任意规模。需要结合对象大小、实例数量、轮询频率和内存占用进行压测。

规模增大后，可以考虑：

• 将数据分成多个快照。
• 使用 DynamoDB 按 key 查询。
• 使用 ElastiCache 保存 entitlement。
• 将权限判断放入 Lambda authorizer。

其他方案

Lambda authorizer 加 DynamoDB

可以只为 GET /premium-query 配置 Lambda authorizer。authorizer 根据请求中的 API key 或内部客户标识查询 DynamoDB entitlement，然后返回 Allow 或 Deny。

该方案具有以下特点：

• Base URL 保持不变。
• 无需调用 API Gateway Control Plane。
• DynamoDB 的 data plane 适合处理在线请求。
• 拒绝访问时可以返回 403 Forbidden。
• authorizer cache 可以减少重复查询。

需要自定义错误 JSON 时，可以配置 API Gateway 的 ACCESS_DENIED Gateway Response。Gateway Response 更适合固定的错误结构；如果错误内容需要携带复杂的客户信息，origin-side guard 会更灵活。

Origin 直接查询 DynamoDB

如果不想引入 Lambda authorizer，可以由 Node.js origin 查询 DynamoDB：

import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { GetCommand, DynamoDBDocumentClient } from "@aws-sdk/lib-dynamodb";

const documentClient = DynamoDBDocumentClient.from(
  new DynamoDBClient({})
);

async function hasPremiumAccess(apiKeyId) {
  const response = await documentClient.send(
    new GetCommand({
      TableName: process.env.ENTITLEMENT_TABLE,
      Key: { apiKeyId },
      ProjectionExpression: "entitlements"
    })
  );

  return response.Item?.entitlements?.includes("premium-query") === true;
}

进程内可以增加短 TTL cache，减少读请求成本。与 S3 全量快照相比，这种方式没有全量同步和多实例轮询的问题，但每次 cache miss 都会增加一次 data-plane 查询延迟。

使用 JWT claims

如果以后可以调整认证模型，可以给客户签发带 entitlement claim 的短期 JWT：

{
  "sub": "customer-001",
  "plan": "premium",
  "entitlements": [
    "premium-query"
  ],
  "exp": 1793185200
}

Gateway 或 origin 验证签名后即可判断权限，无需逐请求查询数据库。不过，权限变更通常要等 token 过期后才会生效，除非再加入 token revocation 机制。

这种方式更适合已经使用 OAuth 2.0、OIDC 或 Cognito 的系统。如果只是为一个 Premium endpoint 引入完整的 token 系统，增加的复杂度可能超过收益。

如何选择

目前只有少量受限 endpoints、订阅变化不频繁，并且可以接受约一分钟的权限传播延迟时，可以保留 S3 快照方案。它成本较低，实现简单，请求路径也比较稳定。

升级和降级需要快速生效时，建议使用 DynamoDB entitlement：

• 希望在 API Gateway 层拒绝请求，可以选择 Lambda authorizer。
• 需要返回更灵活的业务错误，可以选择 origin middleware。
• 请求量较大时，可在 authorizer 或 origin 中加入有界 TTL cache。

无论选择哪种实现，都应将 Usage Plan 与 entitlement 分开。Usage Plan 管理客户的调用额度，entitlement 管理客户可调用的功能。这样既能保留统一的 Base URL，也能让 403 与 429 准确表达各自的错误类型。

备注：内容仅供参考。