Skip to content

Webhooks

Trinity Insights signs every webhook payload with a time-bound HMAC-SHA256 scheme (Stripe-style). This page documents the verification procedure - Node.js and Python implementations ready to paste - and lists the webhook endpoints exposed by the OpenAPI specification.

Signature scheme

Header sent on every outbound call:

Trinity-Signature: t=<unix-seconds>,v1=<hex-hmac-sha256>

The signed string is:

signed_payload = "<unix-seconds>" + "." + raw_request_body

The v1 signature is the hexadecimal HMAC-SHA256 of that string computed with your webhook secret (set when you create your webhook alert channel). The timestamp t is in UTC seconds.

Verification rules

  1. Read the Trinity-Signature header and extract t and v1.
  2. Reject if t is missing, if v1 is missing, or if |now - t| exceeds 5 minutes (replay protection).
  3. Read the request body as RAW BYTES (never after JSON parsing, which would lose the signature).
  4. Compute HMAC-SHA256(secret, t + '.' + raw_body) in hex and compare in constant time against v1.
  5. If valid: process the event and reply 2xx. If invalid: reply 401.

Trinity considers a webhook delivered upon the first 2xx response. On a timeout or a non-2xx response, delivery is retried a few times with a short exponential backoff.

Node.js implementation

// Node.js (Express + crypto) - verify a Trinity webhook payload.
import crypto from "node:crypto";
import type { Request, Response, NextFunction } from "express";

const TOLERANCE_SECONDS = 5 * 60; // reject signatures older than 5 minutes

export function verifyTrinityWebhook(secret: string) {
  return (req: Request, res: Response, next: NextFunction) => {
    const header = req.header("Trinity-Signature");
    if (!header) return res.status(400).send("missing Trinity-Signature header");

    // Format: "t=<unix>,v1=<hexhmac>"
    const parts = Object.fromEntries(
      header.split(",").map((kv) => {
        const [k, v] = kv.split("=");
        return [k, v];
      }),
    ) as Record<string, string>;

    const ts = Number(parts.t);
    const v1 = parts.v1;
    if (!ts || !v1) return res.status(400).send("malformed signature");

    const skew = Math.abs(Math.floor(Date.now() / 1000) - ts);
    if (skew > TOLERANCE_SECONDS) return res.status(400).send("signature expired");

    // req.rawBody must be the unmodified UTF-8 request body.
    // (use express.json({ verify: (req, _res, buf) => (req.rawBody = buf) }))
    const payload = (req as unknown as { rawBody: Buffer }).rawBody.toString("utf8");
    const signed = `${ts}.${payload}`;
    const expected = crypto
      .createHmac("sha256", secret)
      .update(signed, "utf8")
      .digest("hex");

    // Constant-time comparison.
    if (
      v1.length !== expected.length ||
      !crypto.timingSafeEqual(Buffer.from(v1, "hex"), Buffer.from(expected, "hex"))
    ) {
      return res.status(401).send("signature mismatch");
    }
    next();
  };
}

// Usage
// app.post("/hook", verifyTrinityWebhook(process.env.TRINITY_WEBHOOK_SECRET!), handler);

Python implementation

# Python (Flask or FastAPI) - verify a Trinity webhook payload.
import hmac
import hashlib
import time

TOLERANCE_SECONDS = 5 * 60  # reject signatures older than 5 minutes


def verify_trinity_webhook(secret: str, raw_body: bytes, header: str) -> bool:
    """Return True if the Trinity-Signature header is valid for raw_body.

    secret      - your webhook secret (set when you create your webhook alert channel).
    raw_body    - the UNMODIFIED request body as bytes.
    header      - the Trinity-Signature header string ('t=<unix>,v1=<hexhmac>').
    """
    parts = dict(p.split("=", 1) for p in header.split(","))
    ts = int(parts.get("t", "0"))
    v1 = parts.get("v1", "")
    if not ts or not v1:
        return False

    if abs(int(time.time()) - ts) > TOLERANCE_SECONDS:
        return False

    signed = f"{ts}.".encode("utf-8") + raw_body
    expected = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(v1, expected)


# Usage (FastAPI)
# @app.post("/hook")
# async def hook(request: Request):
#     body = await request.body()
#     sig = request.headers.get("Trinity-Signature", "")
#     if not verify_trinity_webhook(os.environ["TRINITY_WEBHOOK_SECRET"], body, sig):
#         raise HTTPException(status_code=401, detail="invalid signature")
#     payload = json.loads(body)
#     ...

Exposed webhook endpoints

MethodPathSummary
POST/api/v1/webhooks/clerkClerk Webhook
POST/api/v1/webhooks/paddlePaddle Webhook

Related resources

Institutional disclaimer

Trinity Insights is an educational and analytical tool. Webhook payloads do not constitute investment advice. Trinity Insights is not a Crypto-Asset Service Provider (CASP) registered under MiCA Regulation (EU) 2023/1114. See the full disclaimer.