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

No public webhook endpoint is exposed by the OpenAPI specification at this stage. This section will populate automatically as soon as one is published.

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.