Task webhooks

const crypto = require("node:crypto");

// rawBody: the request body as a Buffer, before JSON parsing.
function verifyWebhook(secret, rawBody, headers, toleranceSec = 300) {
  const timestamp = headers["x-gateway-webhook-timestamp"] ?? "";
  const signature = headers["x-gateway-webhook-signature"] ?? "";
  if (!/^\d+$/.test(timestamp)) return false;
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > toleranceSec) return false;
  const expected = Buffer.from(
    "v1=" +
      crypto
        .createHmac("sha256", secret)
        .update(`${timestamp}.`)
        .update(rawBody)
        .digest("hex"),
  );
  const given = Buffer.from(signature);
  return given.length === expected.length && crypto.timingSafeEqual(expected, given);
}

import hashlib
import hmac
import time


def verify_webhook(secret, raw_body, headers, tolerance_sec=300):
    """raw_body: the request body as bytes, before JSON parsing."""
    timestamp = headers.get("X-Gateway-Webhook-Timestamp", "")
    signature = headers.get("X-Gateway-Webhook-Signature", "")
    if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > tolerance_sec:
        return False
    expected = "v1=" + hmac.new(
        secret.encode(),
        timestamp.encode() + b"." + raw_body,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

const http = require("node:http");

const secret = process.env.FREEMODEL_WEBHOOK_SECRET;

http.createServer((req, res) => {
  const chunks = [];
  req.on("data", (chunk) => chunks.push(chunk));
  req.on("end", () => {
    const rawBody = Buffer.concat(chunks);
    if (!verifyWebhook(secret, rawBody, req.headers)) {
      res.writeHead(401).end();
      return;
    }
    res.writeHead(200).end("ok"); // acknowledge first
    const event = JSON.parse(rawBody.toString("utf8"));
    const task = event.data.task;
    console.log(event.type, task.id, task.result);
    // dedupe on event.id, then save the result files to your own storage
  });
}).listen(3000);

import json
import os
from http.server import BaseHTTPRequestHandler, HTTPServer

SECRET = os.environ["FREEMODEL_WEBHOOK_SECRET"]


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get("Content-Length") or 0)
        raw_body = self.rfile.read(length)
        if not verify_webhook(SECRET, raw_body, self.headers):
            self.send_response(401)
            self.end_headers()
            return
        self.send_response(200)
        self.end_headers()  # acknowledge first
        event = json.loads(raw_body)
        task = event["data"]["task"]
        print(event["type"], task["id"], task["result"])
        # dedupe on event["id"], then save the result files to your own storage


HTTPServer(("0.0.0.0", 3000), Handler).serve_forever()