双向 TLS（mTLS）

两张证书，一次握手

普通 TLS 只由服务端证明身份。双向 TLS 中客户端也出示证书，服务端用受信任的 CA 验证。结果是双方都拥有密码学身份，而不是共享口令。

工作机制

客户端连接，服务端出示证书，客户端验证链和主机名；随后服务端请求客户端证书，客户端证明持有对应私钥，服务端检查证书链、有效期、吊销状态和允许的身份。TLS 1.3 中客户端证书在握手后发送，并由会话密钥加密。

平台通常把证书主题或 SPIFFE ID 映射为工作负载身份，再依据身份授权。由内部 CA 签发、自动轮换的短期证书可以限制密钥被盗后的价值。

何时使用与取舍

mTLS 适合服务网格内的东西向流量、双方都有基础设施的合作伙伴 API，以及能把密钥放进安全存储的设备群。它也适合零信任：身份随连接传递，不依赖网络位置。

运行 PKI 是实打实的工作：签发、轮换、吊销和信任包分发。浏览器和公共客户端使用客户端证书很不方便，移动设备被攻破时密钥也可能被提取。代理终止 mTLS 后，后端看到的是普通连接，因此代理必须转发已验证身份，后端也只能信任这条路径。

示例与常见误解

支付服务只接受由内部 CA 签发、且 SPIFFE ID 属于预期信任域的客户端证书。攻击者即使拿到数据库口令也无法连接，因为没有有效证书握手就会失败。

常见误解是 mTLS 能授权操作。它认证的是连接，授权仍需策略把身份映射到允许的操作。相关概念：TLS、零信任、证书固定。

代码示例

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'))

import { createHmac, timingSafeEqual } from "node:crypto";

export interface SignatureOptions {
  readonly secret: string;
  readonly toleranceSeconds?: number;
  readonly now?: number;
}

export function signWebhook(payload: string, timestamp: number, secret: string): string {
  if (typeof secret !== "string" || secret.length < 16) {
    throw new RangeError("secret must contain at least 16 characters");
  }
  if (!Number.isInteger(timestamp)) throw new TypeError("timestamp must be an integer");
  const signature = createHmac("sha256", secret).update(`${timestamp}.${payload}`).digest("hex");
  return `t=${timestamp},v1=${signature}`;
}

export function verifyWebhook(payload: string, header: string, options: SignatureOptions): boolean {
  if (typeof header !== "string") throw new TypeError("header must be a string");
  const { secret, toleranceSeconds = 300, now = Date.now() } = options;
  if (typeof secret !== "string" || secret.length < 16) {
    throw new RangeError("secret must contain at least 16 characters");
  }
  const parts = new Map<string, string>();
  for (const part of header.split(",")) {
    const equals = part.indexOf("=");
    if (equals === -1) continue;
    parts.set(part.slice(0, equals).trim(), part.slice(equals + 1).trim());
  }
  const timestampText = parts.get("t");
  const provided = parts.get("v1");
  if (timestampText === undefined || provided === undefined) return false;
  const timestamp = Number(timestampText);
  if (!Number.isInteger(timestamp)) return false;
  if (Math.abs(now / 1000 - timestamp) > toleranceSeconds) return false;
  const expected = createHmac("sha256", secret).update(`${timestamp}.${payload}`).digest("hex");
  const left = fromHex(expected);
  const right = fromHex(provided);
  if (left.length === 0 || left.length !== right.length) return false;
  return timingSafeEqual(left, right);
}

function fromHex(value: string): Uint8Array {
  if (value.length % 2 !== 0 || !/^[0-9a-f]*$/i.test(value)) return new Uint8Array(0);
  const bytes = new Uint8Array(value.length / 2);
  for (let index = 0; index < bytes.length; index += 1) {
    bytes[index] = Number.parseInt(value.slice(index * 2, index * 2 + 2), 16);
  }
  return bytes;
}

const secret = "0123456789abcdef0123456789abcdef";
const payload = JSON.stringify({ event: "invoice.paid", id: "evt_42" });
const header = signWebhook(payload, 1760000000, secret);
console.log(header);
console.log(verifyWebhook(payload, header, { secret, now: 1760000100 * 1000 }));
console.log(verifyWebhook(`${payload} `, header, { secret, now: 1760000100 * 1000 }));
console.log(verifyWebhook(payload, header, { secret, toleranceSeconds: 10, now: 1760000100 * 1000 }));