Webhooks

curl -X POST https://app.autoposting.ai/api-proxy/webhooks \
  -H "Authorization: Bearer sk_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Production Pipeline",
    "url": "https://your-server.com/webhooks/autoposting",
    "events": ["post.published", "post.failed", "clip.render.completed"]
  }'

const res = await fetch("https://app.autoposting.ai/api-proxy/webhooks", {
  method: "POST",
  headers: {
    Authorization: "Bearer sk_your_api_key",
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    name: "Production Pipeline",
    url: "https://your-server.com/webhooks/autoposting",
    events: ["post.published", "post.failed", "clip.render.completed"],
  }),
});
const { data } = await res.json();
// data.webhook.secret — store immediately, shown only once

import { createHmac, timingSafeEqual } from "crypto";

function verifyWebhook(payload: string, timestamp: string, signature: string, secret: string) {
  const expected = "sha256=" + createHmac("sha256", secret)
    .update(`${timestamp}.${payload}`)
    .digest("hex");
  return timingSafeEqual(Buffer.from(signature), Buffer.from(expected));
}

app.post("/webhooks/autoposting", express.raw({ type: "application/json" }), (req, res) => {
  const valid = verifyWebhook(
    req.body.toString(),
    req.headers["x-autoposting-timestamp"] as string,
    req.headers["x-autoposting-signature"] as string,
    process.env.WEBHOOK_SECRET!
  );
  if (!valid) return res.status(401).send("Invalid signature");
  const event = JSON.parse(req.body.toString());
  res.status(200).send("OK");
});

import hashlib, hmac, os
from flask import Flask, request, abort

def verify_webhook(payload: bytes, timestamp: str, signature: str, secret: str) -> bool:
    expected = "sha256=" + hmac.new(
        secret.encode(), f"{timestamp}.{payload.decode()}".encode(), hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)

@app.route("/webhooks/autoposting", methods=["POST"])
def webhook():
    if not verify_webhook(request.data,
        request.headers.get("X-Autoposting-Timestamp", ""),
        request.headers.get("X-Autoposting-Signature", ""),
        os.environ["WEBHOOK_SECRET"]):
        abort(401)
    return "OK", 200