驗證回撥請求

回撥地址必須能被公網訪問，任何人都可以向它傳送偽造的請求。IM 服務發出的每個請求（事件回撥、同步回撥、URL 驗證和測試傳送）都帶有簽名，接收端必須先校驗簽名、檢查時間戳，透過後再處理，否則請求可能是偽造的或被截獲後重放的。

使用 Go、Python 或 Node.js 時，服務端 SDK 已經實現了下面的校驗、防重放和去重，見 Go、Python 和 Node.js 的“接收回撥”。

一定要用原始請求體計算簽名

簽名覆蓋完整的原始請求體。請讀取 HTTP 請求的原始位元組來計算，不要先把 JSON 解析成物件、再重新序列化：欄位的順序、空格、中文和 <、> 等字元的編碼方式都可能改變，簽名隨之不符。

簽名金鑰

每個回撥地址有一個簽名金鑰，形如 cbs_ 加 43 個字元，共 47 個字元，由系統生成，你不能自己設定：

• 新建地址和輪換金鑰時，控制檯顯示一次金鑰的完整內容，只顯示這一次，請立即儲存到接收端的配置或金鑰管理系統中；之後控制檯只顯示它的前 8 個字元（如 cbs_oSi8），用於辨認是哪一個；
• 金鑰不會自動過期，需要更換時輪換，見輪換金鑰；
• 每個地址的金鑰不同。多個地址共用一個接收端時，按請求頭 X-IM-Endpoint-Id（或請求體中的 endpoint_id）選擇對應的金鑰。

簽名演算法

签名串 = "v1:" + X-IM-Timestamp + ":" + X-IM-Nonce + ":" + 原始请求体
签名   = 小写十六进制( HMAC-SHA256( 密钥, 签名串 ) )
X-IM-Signature: v1=<签名>

• 金鑰：完整的金鑰字串（包括 cbs_ 字首）的 ASCII 位元組，不需要做 Base64 解碼；
• 簽名串：v1:、請求頭 X-IM-Timestamp 的原值、:、請求頭 X-IM-Nonce 的原值、:，後面緊跟原始請求體的位元組；
• 簽名：HMAC-SHA256 的結果編碼為 64 個小寫十六進位制字元，請求頭中寫作 v1= 加簽名。輪換金鑰的寬限期內同時用新舊兩個金鑰簽名，請求頭中有兩個簽名，用逗號分隔，新金鑰的在前，如 v1=3be2...9c07,v1=787b...6b9a。

接收端按以下步驟校驗，任何一步不透過都返回 401，不要處理請求：

1. 讀取原始請求體；
2. 檢查 X-IM-Timestamp 與本機當前時間相差不超過 5 分鐘（請保持伺服器時鐘同步）；
3. 用你儲存的金鑰按上面的演算法計算簽名，與 X-IM-Signature 中每一個 v1= 後面的簽名逐個比較，有一個相同即透過。請使用常數時間的比較函式（如 Python 的 hmac.compare_digest、Go 的 hmac.Equal、Node.js 的 crypto.timingSafeEqual），防止透過響應時間猜出簽名；
4. 記住 10 分鐘內見過的 X-IM-Nonce，拒絕重複的隨機數，防止請求在 5 分鐘內被原樣重放；
5. 事件回撥再按事件的 id 去重：同一個事件的重試是新的請求，時間戳、隨機數和簽名都不同，但事件 ID 相同。

每次傳送（包括重試、重新投遞、URL 驗證和測試傳送）都使用當時的時間和新的隨機數重新簽名，幾天前失敗的事件重新投遞時也能透過時間戳的檢查。簽名使用傳送時的當前金鑰：輪換之後，已經排隊的事件也改用新金鑰簽名。

計算示例

下面是一個真實的 URL 驗證請求，簽名所用的金鑰已經作廢，可以用來核對你的實現：

專案	值
金鑰	cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE
X-IM-Timestamp	1791141598
X-IM-Nonce	c90a695d9294c5f757dd88350e0e3187
請求體（213 位元組）	{"kind":"verify","request_id":"100310966629040128","app":"6195295143#doc-callback","endpoint_id":"100310966557736960","sent_at":"2026-10-04T19:19:58.132Z","challenge":"J9Kk_hKp_H74ZScznGSzU_-kL8b4r1MTACmzTFsNVP4"}
X-IM-Signature	v1=2e5cebe9c6c67af85d62c4cc0250579223acc3e89549eabf036d3c523cb08610

用 OpenSSL 計算（請求體儲存在 body.json 中）：

printf 'v1:1791141598:c90a695d9294c5f757dd88350e0e3187:%s' "$(cat body.json)" \
  | openssl dgst -sha256 -hmac 'cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE'

SHA2-256(stdin)= 2e5cebe9c6c67af85d62c4cc0250579223acc3e89549eabf036d3c523cb08610

示例程式碼

以下三個接收端示例都實現了完整的校驗（簽名、時間戳、隨機數），並回應 URL 驗證、同步回撥和事件回撥，可以直接執行，再按業務修改 handle_hook、handle_event 等處理部分。它們都已用測試伺服器發來的真實請求驗證過：URL 驗證透過，測試傳送和同步回撥測試成功；用錯誤的金鑰時返回 401，控制檯顯示驗證失敗。

示例在記憶體中記住隨機數，只適合單個例項。部署多個例項時，請把隨機數儲存在 Redis 等共享儲存中，如 SET im:nonce:<nonce> 1 NX EX 600 返回成功才處理；事件 ID 的去重同樣應儲存在共享儲存或資料庫的唯一鍵中。

示例監聽 /im/callback，請把回撥地址配置為 https://your-server.example.com/im/callback，由你的 HTTPS 閘道器或負載均衡轉發到這個埠。

Python

Python 3.8 以上，使用 Flask：

# 回调接收端示例：Python 3.8+，Flask。
# 运行：pip install flask && CALLBACK_SECRET=cbs_... python callback_server.py
import hashlib
import hmac
import os
import threading
import time

from flask import Flask, abort, jsonify, request

SECRET = os.environ["CALLBACK_SECRET"]  # 回调地址的签名密钥，cbs_ 开头
TOLERANCE = 300  # 时间戳与本机时间最多相差 5 分钟
NONCE_TTL = 600  # 随机数记住 10 分钟

app = Flask(__name__)
_nonces = {}
_lock = threading.Lock()


def verify_signature(secret, headers, body, now=None):
    """校验时间戳和签名。headers 不区分大小写，body 是原始请求体（bytes）。"""
    now = time.time() if now is None else now
    timestamp = headers.get("X-IM-Timestamp", "")
    nonce = headers.get("X-IM-Nonce", "")
    if not timestamp.isdigit() or abs(now - int(timestamp)) > TOLERANCE:
        return False
    message = b"v1:" + timestamp.encode() + b":" + nonce.encode() + b":" + body
    expected = hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()
    for part in headers.get("X-IM-Signature", "").split(","):
        part = part.strip()
        if part.startswith("v1=") and hmac.compare_digest(part[3:], expected):
            return True
    return False


def first_seen(nonce, now=None):
    """随机数第一次出现时返回 True。部署多个实例时改用 Redis：SET im:nonce:<nonce> 1 NX EX 600。"""
    now = time.time() if now is None else now
    with _lock:
        for key in [key for key, at in _nonces.items() if now - at > NONCE_TTL]:
            del _nonces[key]
        if nonce in _nonces:
            return False
        _nonces[nonce] = now
        return True


@app.post("/im/callback")
def callback():
    body = request.get_data()  # 用原始字节校验，不要先解析再重新序列化
    if not verify_signature(SECRET, request.headers, body):
        abort(401)
    if not first_seen(request.headers.get("X-IM-Nonce", "")):
        abort(409)
    payload = request.get_json()
    if payload["kind"] == "verify":
        return jsonify(challenge=payload["challenge"])
    if payload["kind"] == "hook":
        return jsonify(handle_hook(payload["hook"], payload["data"]))
    for event in payload["events"]:
        handle_event(event)
    return "", 204


def handle_hook(hook, data):
    # 按业务判断，例如 {"action": "reject", "reason": "not_same_org", "message": "只能给同事发消息"}
    return {"action": "allow"}


def handle_event(event):
    # 先按 event["id"] 去重；耗时的处理放进自己的队列异步执行，尽快返回 2xx
    print(event["id"], event["type"], event["data"])


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "8080")))

使用其他框架時，同樣要取原始請求體：Django 用 request.body，FastAPI 用 await request.body()。

Go

Go 1.22 以上，只用標準庫（go.mod 中的 go 版本不低於 1.22）：

// 回调接收端示例：Go 1.22+，只用标准库。
// 运行：CALLBACK_SECRET=cbs_... go run main.go
package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	tolerance = 5 * time.Minute  // 时间戳与本机时间最多相差 5 分钟
	nonceTTL  = 10 * time.Minute // 随机数记住 10 分钟
)

// verifySignature 校验时间戳和签名，body 是原始请求体。
func verifySignature(secret string, header http.Header, body []byte, now time.Time) bool {
	timestamp := header.Get("X-IM-Timestamp")
	seconds, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil {
		return false
	}
	if diff := now.Sub(time.Unix(seconds, 0)); diff > tolerance || diff < -tolerance {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte("v1:" + timestamp + ":" + header.Get("X-IM-Nonce") + ":"))
	mac.Write(body)
	expected := hex.EncodeToString(mac.Sum(nil))
	for _, part := range strings.Split(header.Get("X-IM-Signature"), ",") {
		signature, ok := strings.CutPrefix(strings.TrimSpace(part), "v1=")
		if ok && hmac.Equal([]byte(signature), []byte(expected)) {
			return true
		}
	}
	return false
}

// nonceStore 记住最近 10 分钟的随机数。部署多个实例时改用 Redis：SET im:nonce:<nonce> 1 NX EX 600。
type nonceStore struct {
	mu   sync.Mutex
	seen map[string]time.Time
}

// firstSeen 在随机数第一次出现时返回 true。
func (s *nonceStore) firstSeen(nonce string, now time.Time) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	for key, at := range s.seen {
		if now.Sub(at) > nonceTTL {
			delete(s.seen, key)
		}
	}
	if _, ok := s.seen[nonce]; ok {
		return false
	}
	s.seen[nonce] = now
	return true
}

type callbackRequest struct {
	Kind      string          `json:"kind"`
	Hook      string          `json:"hook"`
	Challenge string          `json:"challenge"`
	Data      json.RawMessage `json:"data"`
	Events    []struct {
		ID         string          `json:"id"`
		Type       string          `json:"type"`
		OccurredAt string          `json:"occurred_at"`
		Attempt    int             `json:"attempt"`
		Test       bool            `json:"test"`
		Data       json.RawMessage `json:"data"`
	} `json:"events"`
}

func main() {
	secret := os.Getenv("CALLBACK_SECRET") // 回调地址的签名密钥，cbs_ 开头
	nonces := &nonceStore{seen: map[string]time.Time{}}
	http.HandleFunc("POST /im/callback", func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(io.LimitReader(r.Body, 2<<20))
		if err != nil {
			http.Error(w, "bad request", http.StatusBadRequest)
			return
		}
		now := time.Now()
		if !verifySignature(secret, r.Header, body, now) {
			http.Error(w, "invalid signature", http.StatusUnauthorized)
			return
		}
		if !nonces.firstSeen(r.Header.Get("X-IM-Nonce"), now) {
			http.Error(w, "replayed", http.StatusConflict)
			return
		}
		var req callbackRequest
		if err := json.Unmarshal(body, &req); err != nil {
			http.Error(w, "bad request", http.StatusBadRequest)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		switch req.Kind {
		case "verify":
			json.NewEncoder(w).Encode(map[string]string{"challenge": req.Challenge})
		case "hook":
			// 按 req.Hook 和 req.Data 判断，例如 {"action": "reject", "reason": "not_same_org", "message": "只能给同事发消息"}
			json.NewEncoder(w).Encode(map[string]string{"action": "allow"})
		default:
			for _, event := range req.Events {
				// 先按 event.ID 去重；耗时的处理放进自己的队列异步执行，尽快返回 2xx
				log.Println(event.ID, event.Type, string(event.Data))
			}
			w.WriteHeader(http.StatusNoContent)
		}
	})
	addr := ":8080"
	if port := os.Getenv("PORT"); port != "" {
		addr = ":" + port
	}
	log.Fatal(http.ListenAndServe(addr, nil))
}

Node.js

Node.js 18 以上，只用內建模組：

// 回调接收端示例：Node.js 18+，只用内置模块。
// 运行：CALLBACK_SECRET=cbs_... node server.js
const crypto = require('node:crypto');
const http = require('node:http');

const SECRET = process.env.CALLBACK_SECRET; // 回调地址的签名密钥，cbs_ 开头
const TOLERANCE = 300; // 时间戳与本机时间最多相差 5 分钟
const NONCE_TTL = 600 * 1000; // 随机数记住 10 分钟
const nonces = new Map();

// 校验时间戳和签名。headers 的键为小写（Node.js 的 req.headers），body 是原始请求体（Buffer）。
function verifySignature(secret, headers, body, now = Date.now() / 1000) {
  const timestamp = headers['x-im-timestamp'] || '';
  const nonce = headers['x-im-nonce'] || '';
  if (!/^\d+$/.test(timestamp) || Math.abs(now - Number(timestamp)) > TOLERANCE) return false;
  const expected = Buffer.from(
    crypto.createHmac('sha256', secret).update(`v1:${timestamp}:${nonce}:`).update(body).digest('hex'),
  );
  return (headers['x-im-signature'] || '').split(',').some((part) => {
    part = part.trim();
    if (!part.startsWith('v1=')) return false;
    const signature = Buffer.from(part.slice(3));
    return signature.length === expected.length && crypto.timingSafeEqual(signature, expected);
  });
}

// 随机数第一次出现时返回 true。部署多个实例时改用 Redis：SET im:nonce:<nonce> 1 NX EX 600。
function firstSeen(nonce, now = Date.now()) {
  for (const [key, at] of nonces) if (now - at > NONCE_TTL) nonces.delete(key);
  if (nonces.has(nonce)) return false;
  nonces.set(nonce, now);
  return true;
}

function reply(res, status, value) {
  res.writeHead(status, { 'Content-Type': 'application/json' });
  res.end(value === undefined ? '' : JSON.stringify(value));
}

const server = http.createServer((req, res) => {
  if (req.method !== 'POST' || req.url.split('?')[0] !== '/im/callback') return reply(res, 404);
  const chunks = [];
  req.on('data', (chunk) => chunks.push(chunk));
  req.on('end', () => {
    const body = Buffer.concat(chunks); // 用原始字节校验，不要先解析再重新序列化
    if (!verifySignature(SECRET, req.headers, body)) return reply(res, 401, { error: 'invalid signature' });
    if (!firstSeen(req.headers['x-im-nonce'])) return reply(res, 409, { error: 'replayed' });
    const payload = JSON.parse(body);
    if (payload.kind === 'verify') return reply(res, 200, { challenge: payload.challenge });
    // 按 payload.hook 和 payload.data 判断，例如 { action: 'reject', reason: 'not_same_org', message: '只能给同事发消息' }
    if (payload.kind === 'hook') return reply(res, 200, { action: 'allow' });
    for (const event of payload.events) {
      // 先按 event.id 去重；耗时的处理放进自己的队列异步执行，尽快返回 2xx
      console.log(event.id, event.type, JSON.stringify(event.data));
    }
    reply(res, 204);
  });
});

if (require.main === module) server.listen(Number(process.env.PORT) || 8080);
module.exports = { verifySignature };

使用 Express 時，不要讓 express.json() 先解析請求體，改為用 express.raw({ type: 'application/json' }) 取得 Buffer，校驗透過後再 JSON.parse；或者在 express.json({ verify: (req, res, buf) => { req.rawBody = buf } }) 中儲存原始位元組。

除錯簽名

• 先用測試傳送調通：在控制檯對地址點選“測試”，選擇一種事件傳送，結果中有本次發出的完整請求體、對方返回的狀態碼和響應的開頭部分。測試傳送不要求地址已經透過 URL 驗證，可以在驗證之前先調通簽名。
• 對照計算示例：用上面的計算示例檢查你的實現，結果一致後再排查其他原因。

校驗總是失敗時，常見的原因：

原因	處理
框架已經把 JSON 解析成物件，用重新序列化的結果計算	改為讀取原始請求體，見上文各語言的說明
金鑰複製時多了空格、換行，或用了其他地址的金鑰	重新核對金鑰，注意每個地址的金鑰不同
輪換金鑰後寬限期已結束，接收端仍在用舊金鑰	換成新金鑰；忘記儲存新金鑰時再輪換一次
伺服器時鐘不準，時間戳的檢查不透過	開啟 NTP 時間同步
只比較了 X-IM-Signature 的第一個簽名	寬限期內有兩個簽名，逐個比較，有一個相同即透過
閘道器或代理改動了請求體（如重新編碼、壓縮）	讓閘道器原樣轉發請求體

輪換金鑰

懷疑金鑰洩露，或按安全規範定期更換時，owner 和 admin 可以在控制檯的地址詳情中輪換金鑰（需要重新驗證身份），見回撥配置：

1. 點選“輪換”，填寫舊金鑰的寬限期（1 到 168 小時，預設 24 小時），儲存新金鑰。新金鑰立即生效；
2. 寬限期內每個請求同時帶新舊兩個簽名，接收端用新金鑰或舊金鑰都能校驗透過，可以從容地把接收端的配置換成新金鑰；
3. 全部接收端都換成新金鑰後，可以點選“提前結束寬限期”，之後只用新金鑰簽名；否則寬限期結束後舊金鑰自動作廢。

寬限期還沒有結束時不能再次輪換，返回 409 slot_unavailable，先結束寬限期再輪換。懷疑金鑰洩露時，輪換並把寬限期設為 1 小時，接收端換好新金鑰後立即結束寬限期。

其他安全建議

• 只接受 HTTPS：回撥地址只能是 HTTPS，請為接收端使用有效的證書。響應不簽名，響應的真實性由 HTTPS 保證。
• 限制來源 IP：控制檯的“回撥”頁列出了回撥請求的出口 IP 時，在防火牆上只放行這些 IP，作為簽名之外的又一道防線。
• 以請求體為準：請求頭中的 X-IM-Kind、X-IM-Request-Id、X-IM-Endpoint-Id 只是請求體中同名欄位的副本，只用於分流；需要判斷時請以校驗過簽名的請求體為準。
• URL 中的令牌：可以在 URL 的查詢引數中加上接收端自己的令牌（如 ?token=...），控制檯中顯示時會被遮掩。它只能作為附加的檢查，不能代替簽名。
• 保護金鑰：金鑰只儲存在接收端，不要寫進程式碼倉庫和日誌。