← 中文资源

简体中文版本

令牌内省

令牌内省让资源服务器向授权服务器查询不透明令牌是否有效、授予了什么,从而集中处理吊销与策略。

阅读英文原文

它解决的问题

JWT 等自包含令牌可以本地验证,却无法撤回。不透明令牌不携带声明,资源服务器必须询问授权服务器它代表什么。内省就是这个问题,并标准化了不同厂商的响应结构。

工作机制

资源服务器向 POST /introspect 发送 token=...,并用客户端凭证认证,通常经 TLS。有效时响应包含 active,以及 scope、client_id、username、token_type、exp、iat、aud、iss 等字段。无效或未知令牌只返回 {"active": false},避免泄露令牌是否存在。

资源服务器仍须执行授权:有效令牌若受众不符或缺少 scope,也不能访问。结果可以缓存很短时间,且不超过 exp;缓存键应使用令牌哈希,日志中绝不记录原始值。

何时使用与取舍

内省适合不透明访问令牌、引用令牌,以及吊销必须在数秒内生效的部署;也适合 scope 或租户绑定动态变化的情况。高流量、短寿命且发行方可信的 JWT,用本地验证可以省掉每次请求的网络跳转。

每次内省都增加延迟,并让授权服务器成为硬依赖。它一旦故障,API 也随之故障,除非有缓存或降级策略。集中内省还聚集负载,并形成敏感的令牌使用日志,应尽量缩短保留期。

示例与常见误解

网关收到不透明令牌,内省得到 active: true、scope: "orders:read"、aud: "orders-api",且 60 秒后过期。它缓存 30 秒,并拒绝同一令牌发起的 orders:write 请求,即使令牌仍然有效。

常见误解是内省返回身份和权限本身。它返回元数据,具体允许什么仍由资源服务器决定。相关概念:JSON Web Token、令牌吊销、OAuth 作用域。

代码示例

export function parseContentDisposition(header) {
  if (typeof header !== "string" || header.trim() === "") {
    throw new TypeError("header must be a non-empty string");
  }
  const segments = splitParameters(header);
  const type = segments.shift().trim().toLowerCase();
  if (type !== "inline" && type !== "attachment") {
    throw new Error(`unsupported disposition type "${type}"`);
  }
  const parameters = new Map();
  for (const segment of segments) {
    const equals = segment.indexOf("=");
    if (equals === -1) continue;
    const name = segment.slice(0, equals).trim().toLowerCase();
    if (name === "") continue;
    parameters.set(name, unquote(segment.slice(equals + 1).trim()));
  }
  return { type, filename: resolveFilename(parameters), parameters };
}

function splitParameters(header) {
  const segments = [];
  let current = "";
  let quoted = false;
  for (const character of header) {
    if (character === '"') quoted = !quoted;
    if (character === ";" && !quoted) {
      segments.push(current);
      current = "";
      continue;
    }
    current += character;
  }
  segments.push(current);
  return segments;
}

function unquote(value) {
  if (value.length >= 2 && value.startsWith('"') && value.endsWith('"')) {
    return value.slice(1, -1).replace(/\\(.)/g, "$1");
  }
  return value;
}

function resolveFilename(parameters) {
  const extended = parameters.get("filename*");
  if (extended !== undefined) {
    const match = /^([\w-]+)'([\w-]*)'(.+)$/.exec(extended);
    if (match !== null) {
      try {
        return sanitize(decodeURIComponent(match[3]));
      } catch {
        return sanitize(extended);
      }
    }
  }
  const plain = parameters.get("filename");
  return plain === undefined ? null : sanitize(plain);
}

function sanitize(filename) {
  return filename.split(/[\\/]/).pop() || null;
}

console.log(parseContentDisposition('attachment; filename="report.pdf"'));
console.log(parseContentDisposition("attachment; filename*=UTF-8''%E4%B8%AD%E6%96%87.pdf"));
console.log(parseContentDisposition('inline; filename="../../etc/passwd"'));
import hmac
import secrets

ALPHABET = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'


def api_key(prefix: str = 'sk') -> str:
    body = ''.join(secrets.choice(ALPHABET) for _ in range(40))
    return f'{prefix}_{body}'


def url_token(nbytes: int = 32) -> str:
    return secrets.token_urlsafe(nbytes)


def otp_code(digits: int = 6) -> str:
    # 朴素做法 random.randint 可预测,必须用 secrets
    floor = 10 ** (digits - 1)
    return str(floor + secrets.randbelow(9 * floor))


def constant_time_match(candidate: str, expected: str) -> bool:
    if len(candidate) != len(expected):
        return False
    return hmac.compare_digest(
        candidate.encode(), expected.encode()
    )


def scoped_token(scope: str, nbytes: int = 16) -> dict:
    return {
        'token': url_token(nbytes),
        'scope': scope,
        'hint': f'scope={scope}',
    }


key = api_key()
code = otp_code()
print('key length:', len(key), 'otp:', code)
print('compare ok:', constant_time_match(code, code))
print('compare tampered:', constant_time_match(code, '123456'))