Push delivery for hiring events.

curl -X POST https://api.placement.solutions/v1/webhooks \
  -H "Authorization: Bearer pk_live_REPLACE_ME" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{
    "url": "https://example.com/hooks/placement",
    "events": ["lateral_move.detected", "partner_departure.detected"],
    "description": "Production CI feed"
  }'

import requests, uuid

resp = requests.post(
    "https://api.placement.solutions/v1/webhooks",
    headers={
        "Authorization": "Bearer pk_live_REPLACE_ME",
        "Idempotency-Key": str(uuid.uuid4()),
    },
    json={
        "url": "https://example.com/hooks/placement",
        "events": ["lateral_move.detected", "partner_departure.detected"],
        "description": "Production CI feed",
    },
    timeout=10,
)
endpoint = resp.json()
# SAVE endpoint["secret"] now. It will not be shown again.
secret = endpoint["secret"]

import { randomUUID } from "node:crypto";

const resp = await fetch("https://api.placement.solutions/v1/webhooks", {
  method: "POST",
  headers: {
    "Authorization": "Bearer pk_live_REPLACE_ME",
    "Content-Type": "application/json",
    "Idempotency-Key": randomUUID(),
  },
  body: JSON.stringify({
    url: "https://example.com/hooks/placement",
    events: ["lateral_move.detected", "partner_departure.detected"],
    description: "Production CI feed",
  }),
});
const endpoint = await resp.json();
// Save endpoint.secret now; we cannot show it again.

Placement-Signature: t=1746542555,v1=8a3f1b9c0e7d4256af912e3c6d7f8b1a4c5e9d0f7e2a3b4c5d6e7f8a9b0c1d2e

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

WEBHOOK_SECRET = os.environ["PLACEMENT_WEBHOOK_SECRET"]
TOLERANCE = 300  # 5 minutes

app = Flask(__name__)

def verify(raw_body: bytes, header: str, secret: str) -> None:
    parts = dict(p.split("=", 1) for p in header.split(","))
    ts = int(parts["t"])
    sig = parts["v1"]
    if abs(time.time() - ts) > TOLERANCE:
        raise ValueError("replay")
    expected = hmac.new(
        secret.encode(),
        f"{ts}.".encode() + raw_body,
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(expected, sig):
        raise ValueError("bad signature")

@app.post("/hooks/placement")
def handle():
    try:
        verify(request.get_data(), request.headers["Placement-Signature"], WEBHOOK_SECRET)
    except Exception:
        abort(400)
    event = request.get_json()
    # enqueue event["id"]; respond fast
    return "", 200

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

const WEBHOOK_SECRET = process.env.PLACEMENT_WEBHOOK_SECRET;
const TOLERANCE = 300;
const app = express();

// IMPORTANT: capture the raw body BEFORE JSON parsing.
app.use("/hooks/placement", express.raw({ type: "application/json" }));

function verify(rawBody, header, secret) {
  const parts = Object.fromEntries(header.split(",").map(p => p.split("=")));
  const ts = Number(parts.t);
  const sig = parts.v1;
  if (Math.abs(Date.now() / 1000 - ts) > TOLERANCE) throw new Error("replay");
  const expected = createHmac("sha256", secret)
    .update(Buffer.concat([Buffer.from(`${ts}.`), rawBody]))
    .digest("hex");
  if (!timingSafeEqual(Buffer.from(expected), Buffer.from(sig))) {
    throw new Error("bad signature");
  }
}

app.post("/hooks/placement", (req, res) => {
  try { verify(req.body, req.headers["placement-signature"], WEBHOOK_SECRET); }
  catch { return res.status(400).end(); }
  const event = JSON.parse(req.body.toString("utf8"));
  // enqueue event.id; ack fast
  res.status(200).end();
});

require "sinatra"
require "openssl"

WEBHOOK_SECRET = ENV.fetch("PLACEMENT_WEBHOOK_SECRET")
TOLERANCE = 300

def verify!(raw_body, header, secret)
  parts = header.split(",").map { |p| p.split("=", 2) }.to_h
  ts  = parts["t"].to_i
  sig = parts["v1"]
  raise "replay" if (Time.now.to_i - ts).abs > TOLERANCE
  expected = OpenSSL::HMAC.hexdigest("SHA256", secret, "#{ts}.#{raw_body}")
  raise "bad signature" unless Rack::Utils.secure_compare(expected, sig)
end

post "/hooks/placement" do
  raw = request.body.read
  begin
    verify!(raw, request.env["HTTP_PLACEMENT_SIGNATURE"], WEBHOOK_SECRET)
  rescue
    halt 400
  end
  event = JSON.parse(raw)
  # enqueue event["id"]
  200
end

curl -X POST https://api.placement.solutions/v1/webhooks/we_01J9X5AC2P3K8Q4N7M6Y3R2T1V/rotate-secret \
  -H "Authorization: Bearer rk_live_REPLACE_ME"

def verify_either(raw, header, current, previous=None):
    for secret in (current, previous):
        if not secret:
            continue
        try:
            verify(raw, header, secret)
            return  # success
        except ValueError:
            continue
    raise ValueError("signature did not match either secret")

curl -X POST https://api.placement.solutions/v1/webhooks/we_01J.../deliveries/dl_01K.../replay \
  -H "Authorization: Bearer pk_live_REPLACE_ME"

def handle(req):
    verify(req.body, req.headers["Placement-Signature"], SECRET)
    event_id = json.loads(req.body)["id"]
    queue.enqueue("placement_event", event_id, req.body)
    return 200  # return immediately, work happens elsewhere