验证投递签名

POST /hooks/radar HTTP/1.1
Content-Type: application/json
X-Webhook-Id: evt_01JQZ8XK9M
X-Webhook-Timestamp: 1774000000
X-Webhook-Event: data.updated
X-Webhook-Signature: sha256=6f1c3b...

{"eventId":"evt_01JQZ8XK9M","event":"data.updated","deliveredAt":"...","repos":[...]}

import { createHash, createHmac } from "node:crypto"

const LABEL = "mcp-radar-subscription-v1"

/** 从你签发的 api key 明文算出验签密钥。 */
export function deriveSigningKey(apiKey) {
  const keyHash = createHash("sha256").update(apiKey, "utf8").digest("hex")
  return createHmac("sha256", LABEL).update(keyHash, "utf8").digest("base64url")
}

import base64
import hashlib
import hmac

LABEL = b"mcp-radar-subscription-v1"


def derive_signing_key(api_key: str) -> str:
    key_hash = hashlib.sha256(api_key.encode()).hexdigest()
    mac = hmac.new(LABEL, key_hash.encode(), hashlib.sha256).digest()
    return base64.urlsafe_b64encode(mac).decode().rstrip("=")

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

const WINDOW_SECONDS = 300

/** 原始请求体 + 请求头，全验一遍。返回 true 才能信这条请求。 */
export function verifyDelivery(rawBody, headers, signingKey) {
  const timestamp = headers["x-webhook-timestamp"]
  const received = headers["x-webhook-signature"]
  if (!timestamp || !received) return false

  const expected = `sha256=${createHmac("sha256", signingKey)
    .update(`${timestamp}.${rawBody}`, "utf8")
    .digest("hex")}`

  // 先比长度再常量时间比较：timingSafeEqual 在长度不等时会抛，而长度本身没有
  // 可泄漏的信息（定长摘要），先判一次只是绕开那个异常。
  const a = Buffer.from(expected, "utf8")
  const b = Buffer.from(received, "utf8")
  if (a.length !== b.length) return false
  if (!timingSafeEqual(a, b)) return false

  // 时间戳也要看：只验 HMAC 的接收方会永远接受一条被录下来的请求。
  const ageSeconds = Math.abs(Date.now() / 1000 - Number(timestamp))
  return ageSeconds <= WINDOW_SECONDS
}

import hashlib
import hmac
import time

WINDOW_SECONDS = 300


def verify_delivery(raw_body: bytes, headers, signing_key: str) -> bool:
    timestamp = headers.get("x-webhook-timestamp")
    received = headers.get("x-webhook-signature")
    if not timestamp or not received:
        return False

    message = timestamp.encode() + b"." + raw_body
    expected = "sha256=" + hmac.new(
        signing_key.encode(), message, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(expected, received):
        return False

    try:
        age_seconds = abs(time.time() - int(timestamp))
    except ValueError:
        return False
    return age_seconds <= WINDOW_SECONDS

const seen = new Set()

app.post(
  "/hooks/radar",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const raw = req.body.toString("utf8")
    if (!verifyDelivery(raw, req.headers, SIGNING_KEY)) {
      return res.status(401).send("bad signature")
    }

    const eventId = req.headers["x-webhook-id"]
    if (seen.has(eventId)) return res.sendStatus(200)
    seen.add(eventId)

    handle(JSON.parse(raw))
    res.sendStatus(200)
  }
)

curl -X POST "$CONSOLE_ORIGIN/api/v1/subscriptions/$SUBSCRIPTION_ID/test" \
  -H "Authorization: Bearer $CONSOLE_API_KEY"

curl "$CONSOLE_ORIGIN/api/v1/subscriptions/$SUBSCRIPTION_ID" \
  -H "Authorization: Bearer $CONSOLE_API_KEY"