Put the interpretation everywhere your service already goes.

POST /your/endpoint
content-type: application/json
Acts2-Event-Id: evt_01J9Z6Q4T8
Acts2-Signature: t=1760436000,v1=5f2b…c9e1

{
  "id": "evt_01J9Z6Q4T8",
  "type": "service.ended",
  "created": 1760436000,
  "org_id": "7c1e…",
  "data": {
    "service_id": "b3f0…",
    "started_at": "2026-10-12T02:00:11Z",
    "ended_at": "2026-10-12T03:31:40Z",
    "billable_seconds": 3180,
    "peak_listeners": 128,
    "end_reason": "host"
  }
}

import crypto from "node:crypto";

// rawBody: the exact request body string (verify before JSON.parse)
export function verifyActs2(rawBody, signatureHeader, secret, toleranceSec = 300) {
  const parts = Object.fromEntries(
    String(signatureHeader).split(",").map((kv) => kv.trim().split("=", 2)),
  );
  const t = Number(parts.t);
  if (!Number.isFinite(t) || Math.abs(Date.now() / 1000 - t) > toleranceSec) return false;
  const expected = crypto.createHmac("sha256", secret).update(`${t}.${rawBody}`).digest("hex");
  const a = Buffer.from(expected, "utf8");
  const b = Buffer.from(String(parts.v1 ?? ""), "utf8");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

// Express: app.post("/acts2", express.raw({ type: "application/json" }), (req, res) => {
//   if (!verifyActs2(req.body.toString("utf8"), req.get("Acts2-Signature"), process.env.ACTS2_WEBHOOK_SECRET))
//     return res.sendStatus(400);
//   const event = JSON.parse(req.body);   // dedupe on event.id (Acts2-Event-Id)
//   res.sendStatus(204);
// });

import hashlib, hmac, time

def verify_acts2(raw_body: bytes, signature_header: str, secret: str, tolerance: int = 300) -> bool:
    try:
        parts = dict(kv.strip().split("=", 1) for kv in signature_header.split(","))
        t = int(parts["t"])
    except (KeyError, ValueError):
        return False
    if abs(time.time() - t) > tolerance:
        return False
    signed = f"{t}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts.get("v1", ""))

# Flask:
# @app.post("/acts2")
# def acts2():
#     if not verify_acts2(request.get_data(), request.headers.get("Acts2-Signature", ""), SECRET):
#         abort(400)
#     event = request.get_json()   # dedupe on event["id"]
#     return "", 204