Webhooks

from pleate_backend.webhooks import dispatch, register_event

register_event("order.shipped", "An order left the warehouse.")    # e.g. in AppConfig.ready()

dispatch("order.shipped", {"id": str(order.pk), "tracking": order.tracking}, owner=order.user)
dispatch("catalog.updated", {"count": 12})                         # no owner: endpoints of staff users

POST /hooks/plate HTTP/1.1
Content-Type: application/json
User-Agent: pleate-webhooks/1
Pleate-Event: order.shipped
Pleate-Delivery: 6720e1a2b3c4d5e6f7a8b9c0
Pleate-Signature: t=1792065600,v1=5f8c0e7a1b2c…

{"id": "evt_3k9…", "type": "order.shipped", "created_at": "2026-10-26T11:00:00Z", "data": {"id": "…", "tracking": "…"}}

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

export async function POST(request: Request) {
  const raw = await request.text();
  if (!verify(process.env.PLATE_WEBHOOK_SECRET!, raw, request.headers.get("pleate-signature") ?? "")) {
    return new Response("bad signature", { status: 400 });
  }
  const event = JSON.parse(raw);
  // ...handle event.type, deduplicate on event.id
  return new Response(null, { status: 204 });
}

function verify(secret: string, raw: string, header: string, tolerance = 300) {
  const parts = Object.groupBy(header.split(","), (p) => p.split("=")[0]);
  const t = Number(parts.t?.[0]?.slice(2));
  if (!t || Math.abs(Date.now() / 1000 - t) > tolerance) return false;
  const expected = Buffer.from(createHmac("sha256", secret).update(`${t}.${raw}`).digest("hex"));
  return (parts.v1 ?? []).some((p) => {
    const candidate = Buffer.from(p.slice(3));
    return candidate.length === expected.length && timingSafeEqual(candidate, expected);
  });
}

import hashlib, hmac, time

def verify(secret: str, raw: bytes, header: str, tolerance: int = 300) -> bool:
    parts = [p.split("=", 1) for p in header.split(",") if "=" in p]
    timestamps = [v for k, v in parts if k == "t"]
    signatures = [v for k, v in parts if k == "v1"]
    if not timestamps or not signatures or abs(time.time() - int(timestamps[0])) > tolerance:
        return False
    message = timestamps[0].encode() + b"." + raw
    expected = hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, s) for s in signatures)

from pleate_backend.webhooks.inbound import claim_inbound, mark_processed, release_inbound, verify_signature

@csrf_exempt
def plate_hook(request):
    if not verify_signature(SECRET, request.body, request.headers.get("Pleate-Signature", "")):
        return HttpResponse(status=400)
    event = json.loads(request.body)
    if not claim_inbound("plate", event["id"], event):
        return HttpResponse(status=200)          # already handled
    try:
        handle(event)
    except Exception:
        release_inbound("plate", event["id"])    # let the sender's retry be handled
        raise
    mark_processed("plate", event["id"])
    return HttpResponse(status=200)