Receive webhooks

import { createHmac, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";

// The secret of the webhook, as the Developers page shows it. The key is the base64 after whsec_.
const key = Buffer.from((process.env.GPT_VIDEO_WEBHOOK_SECRET ?? "").replace(/^whsec_/, ""), "base64");
const handled = new Set();

createServer((req, res) => {
  let body = "";
  req.setEncoding("utf8");
  req.on("data", (chunk) => {
    body += chunk;
  });
  req.on("end", async () => {
    // Check the signature on the raw body, before parsing it.
    const id = String(req.headers["webhook-id"] ?? "");
    const timestamp = String(req.headers["webhook-timestamp"] ?? "");
    const expected = Buffer.from(createHmac("sha256", key).update(`${id}.${timestamp}.${body}`).digest("base64"));
    const signed = String(req.headers["webhook-signature"] ?? "")
      .split(" ")
      .some((part) => {
        const given = Buffer.from(part.slice(3));
        return part.startsWith("v1,") && given.length === expected.length && timingSafeEqual(given, expected);
      });
    const recent = Math.abs(Date.now() / 1000 - Number(timestamp)) <= 300;
    if (!signed || !recent) {
      res.writeHead(400).end();
      return;
    }

    // Answer first, then do the work. A retry of this event has the same webhook-id.
    res.writeHead(200).end();
    if (handled.has(id)) return;
    handled.add(id);

    const event = JSON.parse(body);
    if (event.type === "export.completed") {
      const answer = await fetch(`https://api.gpt-video.com/v1/exports/${event.data.exportId}`, {
        headers: { Authorization: `Bearer ${process.env.GPT_VIDEO_API_KEY}` },
      });
      const done = (await answer.json()).export;
      console.log(`Export ${done.id} is ${done.status}. Download it now from ${done.downloadUrl}`);
    }
  });
}).listen(3000);

import base64
import hashlib
import hmac
import json
import os
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

import requests

# The secret of the webhook, as the Developers page shows it. The key is the base64 after whsec_.
KEY = base64.b64decode(os.environ["GPT_VIDEO_WEBHOOK_SECRET"].removeprefix("whsec_"))
handled = set()


class Receiver(BaseHTTPRequestHandler):
    def do_POST(self):
        # Check the signature on the raw body, before parsing it.
        body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
        event_id = self.headers.get("webhook-id", "")
        timestamp = self.headers.get("webhook-timestamp", "")
        expected = base64.b64encode(
            hmac.new(KEY, f"{event_id}.{timestamp}.".encode() + body, hashlib.sha256).digest()
        ).decode()
        signed = any(
            part.startswith("v1,") and hmac.compare_digest(part[3:].encode(), expected.encode())
            for part in self.headers.get("webhook-signature", "").split(" ")
        )
        recent = timestamp.isdigit() and abs(time.time() - int(timestamp)) <= 300
        if not (signed and recent):
            self.send_response(400)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return

        # Answer first, then do the work. A retry of this event has the same webhook-id.
        self.send_response(200)
        self.send_header("Content-Length", "0")
        self.end_headers()
        if event_id in handled:
            return
        handled.add(event_id)

        event = json.loads(body)
        if event["type"] == "export.completed":
            answer = requests.get(
                f"https://api.gpt-video.com/v1/exports/{event['data']['exportId']}",
                headers={"Authorization": f"Bearer {os.environ['GPT_VIDEO_API_KEY']}"},
            )
            done = answer.json()["export"]
            print(f"Export {done['id']} is {done['status']}. Download it now from {done['downloadUrl']}")


HTTPServer(("", 3000), Receiver).serve_forever()