Webhooks

{
  "id": "task:cgt-2026…:succeeded",
  "type": "task.completed",
  "created_at": "2026-09-02T00:10:00Z",
  "data": {
    "id": "cgt-2026…", "client_request_id": "9d72…", "model": "seedance-2.0", "status": "succeeded",
    "usage": {"completion_tokens": 246840, "total_tokens": 246840},
    "billing": {"charged_micro_usd": 1819440, "frozen_micro_usd": 2500000},
    "result_url": "https://…/video.mp4", "storage_status": "pending"
  }
}

import hmac, hashlib, time

def verify(secret: str, header: str, body: bytes, tolerance: int = 300) -> bool:
    parts = dict(p.split("=", 1) for p in header.split(","))
    ts, sig = parts["t"], parts["v1"]
    if abs(time.time() - int(ts)) > tolerance:
        return False
    expected = hmac.new(secret.encode(), f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, sig)

import crypto from "node:crypto";

export function verify(secret, header, rawBody, tolerance = 300) {
  const parts = Object.fromEntries(header.split(",").map((p) => p.split("=")));
  if (Math.abs(Date.now() / 1000 - Number(parts.t)) > tolerance) return false;
  const expected = crypto.createHmac("sha256", secret).update(`${parts.t}.`).update(rawBody).digest("hex");
  return crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(parts.v1));
}

import crypto from "node:crypto";
import express from "express";

const secret = process.env.LINK42_CALLBACK_SECRET; // whcb_…

// X-Signature: t=<unix>,v1=<hex>[,v1=<hex>] — during the 24 hours after a
// rotation it carries one v1 per secret; any one matching is enough.
function verify(header, rawBody, tolerance = 300) {
  const pairs = (header ?? "").split(",").map((part) => part.trim().split("="));
  const t = pairs.find(([name]) => name === "t")?.[1];
  const signatures = pairs.filter(([name, value]) => name === "v1" && value).map(([, value]) => value);
  if (!t || signatures.length === 0 || Math.abs(Date.now() / 1000 - Number(t)) > tolerance) return false;
  const expected = Buffer.from(crypto.createHmac("sha256", secret).update(`${t}.`).update(rawBody).digest("hex"));
  return signatures.some((signature) => signature.length === expected.length && crypto.timingSafeEqual(expected, Buffer.from(signature)));
}

const app = express();
// Keep the raw body: the signature covers the exact bytes that were sent.
app.post("/link42/callback", express.raw({ type: "application/json" }), (req, res) => {
  if (!verify(req.get("X-Signature"), req.body)) return res.status(401).end();
  const event = JSON.parse(req.body.toString("utf8"));
  console.log(event.type, event.data?.status);
  res.sendStatus(200);
});

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

SECRET = os.environ["LINK42_CALLBACK_SECRET"]  # whcb_…
app = Flask(__name__)


def verify(header: str, body: bytes, tolerance: int = 300) -> bool:
    # X-Signature: t=<unix>,v1=<hex>[,v1=<hex>] - during the 24 hours after a
    # rotation it carries one v1 per secret; any one matching is enough.
    pairs = [part.strip().split("=", 1) for part in header.split(",") if "=" in part]
    timestamps = [value for name, value in pairs if name == "t"]
    signatures = [value for name, value in pairs if name == "v1" and value]
    try:
        if not timestamps or not signatures or abs(time.time() - int(timestamps[0])) > tolerance:
            return False
    except ValueError:
        return False
    expected = hmac.new(SECRET.encode(), f"{timestamps[0]}.".encode() + body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, signature) for signature in signatures)


@app.post("/link42/callback")
def callback():
    # request.get_data() is the raw body the signature covers.
    if not verify(request.headers.get("X-Signature", ""), request.get_data()):
        abort(401)
    event = request.get_json()
    print(event["type"], event["data"].get("status"))
    return "", 200