Webhooks

import hashlib
import hmac
import time

from flask import Flask, request

app = Flask(__name__)

# Saved when you submit jobs: job id -> job key (the one-time "pjk_..."
# value from the 202 response). Use your database in production.
JOB_KEYS: dict[str, str] = {}
SEEN_EVENTS: set[str] = set()


def verify_webhook(job_key: str, signature: str, raw_body: bytes,
                   tolerance_s: int = 300) -> bool:
    try:
        parts = dict(p.split("=", 1) for p in signature.split(","))
        ts, v1 = int(parts["t"]), parts["v1"]
    except (KeyError, ValueError):
        return False
    if abs(time.time() - ts) > tolerance_s:
        return False  # stale timestamp: possible replay
    sig_key = hashlib.sha256(b"pc-webhook-sig-v1|" + job_key.encode()).digest()
    expected = hmac.new(sig_key, f"{ts}.".encode() + raw_body,
                        hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)


@app.post("/privconvert-webhook")
def privconvert_webhook():
    raw = request.get_data()          # exact bytes: sign THESE, never re-parse
    event = request.get_json(silent=True) or {}
    job_key = JOB_KEYS.get(event.get("id", ""))
    signature = request.headers.get("X-PrivConvert-Signature", "")
    if not job_key or not verify_webhook(job_key, signature, raw):
        return "", 400

    if event["event_id"] in SEEN_EVENTS:
        return "", 200                # duplicate delivery: acknowledge, skip
    SEEN_EVENTS.add(event["event_id"])

    # Acknowledge fast; do real work (e.g. download the result) afterwards,
    # for example on a task queue.
    if event["event"] == "job.finished":
        queue_download(event["id"], job_key)   # your code
    return "", 200

import express from "express";
import { createHash, createHmac, timingSafeEqual } from "node:crypto";

const app = express();

// Saved when you submit jobs: job id -> job key (the one-time "pjk_..."
// value from the 202 response). Use your database in production.
const JOB_KEYS = new Map();
const SEEN_EVENTS = new Set();

function verifyWebhook(jobKey, signature, rawBody, toleranceS = 300) {
  const m = signature.match(/^t=(\d+),v1=([0-9a-f]{64})$/);
  if (!m) return false;
  const ts = parseInt(m[1], 10);
  if (Math.abs(Date.now() / 1000 - ts) > toleranceS) return false;
  const sigKey = createHash("sha256")
    .update("pc-webhook-sig-v1|" + jobKey).digest();
  const expected = createHmac("sha256", sigKey)
    .update(`${ts}.`).update(rawBody).digest("hex");
  return timingSafeEqual(Buffer.from(expected), Buffer.from(m[2]));
}

// express.raw keeps the exact bytes: sign THESE, never re-serialized JSON.
app.post("/privconvert-webhook",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const event = JSON.parse(req.body.toString());
    const jobKey = JOB_KEYS.get(event.id);
    const signature = req.get("X-PrivConvert-Signature") || "";
    if (!jobKey || !verifyWebhook(jobKey, signature, req.body)) {
      return res.sendStatus(400);
    }

    if (SEEN_EVENTS.has(event.event_id)) return res.sendStatus(200);
    SEEN_EVENTS.add(event.event_id);

    // Acknowledge fast; do real work (e.g. download the result) afterwards.
    if (event.event === "job.finished") {
      queueDownload(event.id, jobKey);   // your code
    }
    res.sendStatus(200);
  });

app.listen(3000);