Build a webhook receiver

# signature_of BODY: lowercase hex HMAC-SHA256 of the exact bytes, keyed with the webhook's
# signature value. verify_signature BODY HEADER: succeeds when HEADER is that signature.
signature_of() {
  printf '%s' "$1" | openssl dgst -sha256 -hmac "$WEBHOOK_SECRET" | sed 's/^.*= //'
}
verify_signature() {
  [ "$(signature_of "$1")" = "$(printf '%s' "$2" | tr -d '[:space:]' | tr 'A-F' 'a-f')" ]
}

// WebHook-Signature is the lowercase hex HMAC-SHA256 of the raw request body, keyed with the
// webhook's signature value. Check the bytes as received, before parsing, in constant time.
function verifySignature(rawBody: Buffer, header: string | undefined, secret: string): boolean {
  if (!header) return false;
  const expected = Buffer.from(createHmac("sha256", secret).update(rawBody).digest("hex"));
  const received = Buffer.from(header.trim().toLowerCase());
  return expected.length === received.length && timingSafeEqual(expected, received);
}

# WebHook-Signature is the lowercase hex HMAC-SHA256 of the raw request body, keyed with the
# webhook's signature value. Check the bytes as received, before parsing, in constant time.
def verify_signature(raw_body: bytes, header: Optional[str], secret: str) -> bool:
    if not header:
        return False
    expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header.strip().lower())

COEVERA_BASE="https://{region}.api.coevera.com/api/v100/rest/spaces/{space_id}"
SPACE_ID="{space_id}"
RECEIVER_URL="${RECEIVER_URL:-http://localhost:8787/coevera-webhooks}"

# signature_of BODY: lowercase hex HMAC-SHA256 of the exact bytes, keyed with the webhook's
# signature value. verify_signature BODY HEADER: succeeds when HEADER is that signature.
signature_of() {
  printf '%s' "$1" | openssl dgst -sha256 -hmac "$WEBHOOK_SECRET" | sed 's/^.*= //'
}
verify_signature() {
  [ "$(signature_of "$1")" = "$(printf '%s' "$2" | tr -d '[:space:]' | tr 'A-F' 'a-f')" ]
}

# The test delivery carries a real record, as the REST API returns it.
record=$(curl -sS --fail-with-body -G "$COEVERA_BASE/entities/Opportunities" \
  -d first=1 \
  -d load-only=id,name,owner_id,step_id,revision \
  -u "$COEVERA_API_KEY:$COEVERA_API_SECRET" | jq -ce '.data[0]')

# One line of JSON, like a delivery; event_time is UTC without an offset.
body=$(jq -cn --argjson entity "$record" --arg url "$RECEIVER_URL" --arg space "$SPACE_ID" \
  --arg time "$(date -u +%Y-%m-%dT%H:%M:%S)" \
  '[{webhook: {url: $url, events: ["Opportunity.Update"]}, event: "Opportunity.Update",
     event_time: $time, team_space_id: $space, entity: $entity, related_entity: null}]')
signature=$(signature_of "$body")
verify_signature "$body" "$signature"
curl -sS --fail-with-body "$RECEIVER_URL" \
  -H "Content-Type: application/json" \
  -H "WebHook-Signature: $signature" \
  -H "WebHook-Signature-Version: v1" \
  -H "Webhook-Delivery-Retry: 0" \
  --data-binary "$body"
echo

import { Buffer } from "node:buffer";
import { createHmac, randomUUID, timingSafeEqual } from "node:crypto";
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
import type { AddressInfo } from "node:net";
import { argv, env } from "node:process";

const base = `https://{region}.api.coevera.com/api/v100/rest/spaces/{space_id}`;
const spaceId = "{space_id}";
const auth = Buffer.from(`${env.COEVERA_API_KEY}:${env.COEVERA_API_SECRET}`).toString("base64");
const port = Number(env.PORT ?? 8787);

// WebHook-Signature is the lowercase hex HMAC-SHA256 of the raw request body, keyed with the
// webhook's signature value. Check the bytes as received, before parsing, in constant time.
function verifySignature(rawBody: Buffer, header: string | undefined, secret: string): boolean {
  if (!header) return false;
  const expected = Buffer.from(createHmac("sha256", secret).update(rawBody).digest("hex"));
  const received = Buffer.from(header.trim().toLowerCase());
  return expected.length === received.length && timingSafeEqual(expected, received);
}

interface WebhookEvent {
  event: string;
  event_time: string;
  team_space_id: string;
  entity: { id: string; revision?: number; [field: string]: unknown };
  related_entity: Record<string, unknown> | null;
}

// A delivery has no id, and a retry sends the same body again for about 24 hours. Remember
// what you processed for longer than that (in a database, in production), and drop older
// events: a replayed body still carries a valid signature.
const MAX_AGE_MS = 25 * 60 * 60 * 1000;
const seen = new Map<string, number>();

function ageMs(e: WebhookEvent, now: number): number {
  // event_time is UTC. Without an offset, JavaScript would read it as local time: add the Z.
  const time = /(?:Z|[+-]\d\d:\d\d)$/.test(e.event_time) ? e.event_time : `${e.event_time}Z`;
  return now - Date.parse(time);
}

function firstTime(e: WebhookEvent, now: number): boolean {
  for (const [key, at] of seen) if (now - at > MAX_AGE_MS) seen.delete(key);
  const key = [e.event, e.entity.id, e.entity.revision, e.event_time].join("|");
  if (seen.has(key)) return false;
  seen.set(key, now);
  return true;
}

// Deliveries are not ordered, and each chunk retries on its own: an older revision can arrive
// after a newer one. Keep the newest, and make every write safe to repeat.
const records = new Map<string, { revision: number; deleted: boolean; data: Record<string, unknown> }>();

function processEvent(e: WebhookEvent): string {
  const revision = e.entity.revision ?? 0;
  const current = records.get(e.entity.id);
  if (current && current.revision >= revision) return `kept revision ${current.revision}`;
  records.set(e.entity.id, { revision, deleted: e.event.endsWith(".Delete"), data: e.entity });
  return `stored revision ${revision}`;
}

// The webhook's signature value. The self-test makes one up when WEBHOOK_SECRET is not set.
const secret = env.WEBHOOK_SECRET ?? randomUUID();

function reply(res: ServerResponse, status: number, body: object): void {
  res.writeHead(status, { "Content-Type": "application/json" });
  res.end(JSON.stringify(body));
}

function header(req: IncomingMessage, name: string): string | undefined {
  const value = req.headers[name];
  return Array.isArray(value) ? value[0] : value;
}

// Answer within 10 seconds, then process. Never answer 403 or 410 (Coevera disables the
// webhook at once on either): a bad signature gets 401.
const server = createServer(async (req, res) => {
  // Deleting the webhook sends DELETE to its URL: accept it.
  if (req.method === "DELETE") return reply(res, 200, { removed: true });
  if (req.method !== "POST") return reply(res, 405, { error: "POST only" });
  const chunks: Buffer[] = [];
  for await (const chunk of req) chunks.push(chunk as Buffer);
  const rawBody = Buffer.concat(chunks);
  if (!verifySignature(rawBody, header(req, "webhook-signature"), secret)) return reply(res, 401, { error: "bad signature" });
  let events: WebhookEvent[];
  try {
    events = JSON.parse(rawBody.toString("utf8"));
  } catch {
    return reply(res, 400, { error: "not JSON" });
  }
  if (!Array.isArray(events)) return reply(res, 400, { error: "not a list of events" });
  reply(res, 200, { received: events.length });
  // The answer is on its way. In production, hand the events to a job queue here.
  const now = Date.now();
  for (const e of events) {
    const outcome = ageMs(e, now) > MAX_AGE_MS ? "too old, dropped" : firstTime(e, now) ? processEvent(e) : "duplicate, skipped";
    console.log(`${e.event} ${e.entity.id}: ${outcome}`);
  }
});

// The test delivery carries a real record, as the REST API returns it.
async function findRecord(): Promise<Record<string, unknown>> {
  const res = await fetch(`${base}/entities/Opportunities?first=1&load-only=id,name,owner_id,step_id,revision`, {
    headers: { Authorization: `Basic ${auth}` },
  });
  const body = await res.json();
  if (!res.ok) throw new Error(`HTTP ${res.status} ${body.name}: ${body.message}`);
  if (!body.data.length) throw new Error("The self-test needs one opportunity in the space.");
  return body.data[0];
}

// A delivery signed the way Coevera signs, then what retries, replays, forgeries and a
// deleted webhook look like to the receiver.
async function selfTest(url: string): Promise<void> {
  const record = await findRecord();
  const delivery = (eventTime: Date) =>
    JSON.stringify([
      {
        webhook: { id: randomUUID(), url, events: ["Opportunity.Update"] },
        event: "Opportunity.Update",
        event_time: eventTime.toISOString().slice(0, -1),
        team_space_id: spaceId,
        entity: record,
        related_entity: null,
      },
    ]);
  const send = (body: string, signature: string, retry = 0) =>
    fetch(url, {
      method: "POST",
      headers: { "Content-Type": "application/json", "WebHook-Signature": signature, "WebHook-Signature-Version": "v1", "Webhook-Delivery-Retry": String(retry) },
      body,
    }).then((r) => r.status);
  const sign = (body: string) => createHmac("sha256", secret).update(body).digest("hex");

  const fresh = delivery(new Date());
  const old = delivery(new Date(Date.now() - 48 * 60 * 60 * 1000));
  const answers = {
    first: await send(fresh, sign(fresh)),
    retry: await send(fresh, sign(fresh), 1),
    old: await send(old, sign(old)),
    forged: await send(fresh, "0".repeat(64)),
    deleted: await fetch(url, { method: "DELETE" }).then((r) => r.status),
  };
  console.log("Answers:", JSON.stringify(answers));
  const expected = { first: 200, retry: 200, old: 200, forged: 401, deleted: 200 };
  if (JSON.stringify(answers) !== JSON.stringify(expected)) throw new Error(`Expected ${JSON.stringify(expected)}`);
  if (records.size !== 1 || seen.size !== 1) throw new Error("A retry or an old event was processed.");
}

if (argv.includes("--serve")) {
  server.listen(port, () => console.log(`Listening on http://localhost:${port}/`));
} else {
  server.listen(0, "127.0.0.1");
  await new Promise((resolve) => server.once("listening", resolve));
  const { port: testPort } = server.address() as AddressInfo;
  try {
    await selfTest(`http://127.0.0.1:${testPort}/coevera-webhooks`);
    console.log("Self-test passed.");
  } finally {
    server.close();
    server.closeAllConnections();
  }
}

import hashlib
import hmac
import json
import os
import queue
import sys
import threading
import uuid
from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Optional

import requests


BASE = "https://{region}.api.coevera.com/api/v100/rest/spaces/{space_id}"
SPACE_ID = "{space_id}"
session = requests.Session()
session.auth = (os.environ["COEVERA_API_KEY"], os.environ["COEVERA_API_SECRET"])
PORT = int(os.environ.get("PORT", "8787"))


# WebHook-Signature is the lowercase hex HMAC-SHA256 of the raw request body, keyed with the
# webhook's signature value. Check the bytes as received, before parsing, in constant time.
def verify_signature(raw_body: bytes, header: Optional[str], secret: str) -> bool:
    if not header:
        return False
    expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header.strip().lower())


# A delivery has no id, and a retry sends the same body again for about 24 hours. Remember
# what you processed for longer than that (in a database, in production), and drop older
# events: a replayed body still carries a valid signature.
MAX_AGE = timedelta(hours=25)
seen = {}
seen_lock = threading.Lock()


def age(event, now):
    # event_time is UTC; read it as UTC whether or not it carries an offset.
    when = datetime.fromisoformat(event["event_time"].replace("Z", "+00:00"))
    if when.tzinfo is None:
        when = when.replace(tzinfo=timezone.utc)
    return now - when


def first_time(event, now):
    key = "|".join(str(part) for part in (event["event"], event["entity"]["id"], event["entity"].get("revision"), event["event_time"]))
    with seen_lock:
        for old_key, at in list(seen.items()):
            if now - at > MAX_AGE:
                del seen[old_key]
        if key in seen:
            return False
        seen[key] = now
        return True


# Deliveries are not ordered, and each chunk retries on its own: an older revision can arrive
# after a newer one. Keep the newest, and make every write safe to repeat.
records = {}


def process_event(event):
    entity = event["entity"]
    revision = entity.get("revision") or 0
    current = records.get(entity["id"])
    if current and current["revision"] >= revision:
        return f"kept revision {current['revision']}"
    records[entity["id"]] = {"revision": revision, "deleted": event["event"].endswith(".Delete"), "data": entity}
    return f"stored revision {revision}"


# The webhook's signature value. The self-test makes one up when WEBHOOK_SECRET is not set.
SECRET = os.environ.get("WEBHOOK_SECRET") or str(uuid.uuid4())
jobs = queue.Queue()


def worker():
    while True:
        events = jobs.get()
        now = datetime.now(timezone.utc)
        for event in events:
            if age(event, now) > MAX_AGE:
                outcome = "too old, dropped"
            elif first_time(event, now):
                outcome = process_event(event)
            else:
                outcome = "duplicate, skipped"
            print(f"{event['event']} {event['entity']['id']}: {outcome}", flush=True)
        jobs.task_done()


class Receiver(BaseHTTPRequestHandler):
    # Answer within 10 seconds, then process. Never answer 403 or 410 (Coevera disables the
    # webhook at once on either): a bad signature gets 401.
    def do_POST(self):
        raw_body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        if not verify_signature(raw_body, self.headers.get("WebHook-Signature"), SECRET):
            return self.reply(401, {"error": "bad signature"})
        try:
            events = json.loads(raw_body)
        except ValueError:
            return self.reply(400, {"error": "not JSON"})
        if not isinstance(events, list):
            return self.reply(400, {"error": "not a list of events"})
        self.reply(200, {"received": len(events)})
        jobs.put(events)

    def do_DELETE(self):
        # Deleting the webhook sends DELETE to its URL: accept it.
        self.reply(200, {"removed": True})

    def reply(self, status, body):
        payload = json.dumps(body).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, format, *args):
        pass


threading.Thread(target=worker, daemon=True).start()


# The test delivery carries a real record, as the REST API returns it.
def find_record():
    response = session.get(
        f"{BASE}/entities/Opportunities",
        params={"first": 1, "load-only": "id,name,owner_id,step_id,revision"},
        timeout=30,
    )
    body = response.json()
    if not response.ok:
        sys.exit(f"HTTP {response.status_code} {body.get('name')}: {body.get('message')}")
    if not body["data"]:
        sys.exit("The self-test needs one opportunity in the space.")
    return body["data"][0]


# A delivery signed the way Coevera signs, then what retries, replays, forgeries and a
# deleted webhook look like to the receiver.
def self_test(url):
    record = find_record()

    def delivery(event_time):
        return json.dumps([{
            "webhook": {"id": str(uuid.uuid4()), "url": url, "events": ["Opportunity.Update"]},
            "event": "Opportunity.Update",
            "event_time": event_time.replace(tzinfo=None).isoformat(),
            "team_space_id": SPACE_ID,
            "entity": record,
            "related_entity": None,
        }]).encode("utf-8")

    def sign(body):
        return hmac.new(SECRET.encode("utf-8"), body, hashlib.sha256).hexdigest()

    def send(body, signature, retry=0):
        headers = {
            "Content-Type": "application/json",
            "WebHook-Signature": signature,
            "WebHook-Signature-Version": "v1",
            "Webhook-Delivery-Retry": str(retry),
        }
        status = requests.post(url, data=body, headers=headers, timeout=10).status_code
        jobs.join()  # wait until the receiver has processed it, so the checks below see the result
        return status

    now = datetime.now(timezone.utc)
    fresh = delivery(now)
    old = delivery(now - timedelta(hours=48))
    answers = {
        "first": send(fresh, sign(fresh)),
        "retry": send(fresh, sign(fresh), 1),
        "old": send(old, sign(old)),
        "forged": send(fresh, "0" * 64),
        "deleted": requests.delete(url, timeout=10).status_code,
    }
    print("Answers:", json.dumps(answers))
    expected = {"first": 200, "retry": 200, "old": 200, "forged": 401, "deleted": 200}
    if answers != expected:
        sys.exit(f"Expected {json.dumps(expected)}")
    if len(records) != 1 or len(seen) != 1:
        sys.exit("A retry or an old event was processed.")


if "--serve" in sys.argv:
    port = int(sys.argv[sys.argv.index("--port") + 1]) if "--port" in sys.argv else PORT
    server = ThreadingHTTPServer(("", port), Receiver)
    print(f"Listening on http://localhost:{server.server_port}/", flush=True)
    server.serve_forever()
else:
    server = ThreadingHTTPServer(("127.0.0.1", 0), Receiver)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    try:
        self_test(f"http://127.0.0.1:{server.server_port}/coevera-webhooks")
        print("Self-test passed.")
    finally:
        server.shutdown()

# The test delivery carries a real record, as the REST API returns it.
record=$(curl -sS --fail-with-body -G "$COEVERA_BASE/entities/Opportunities" \
  -d first=1 \
  -d load-only=id,name,owner_id,step_id,revision \
  -u "$COEVERA_API_KEY:$COEVERA_API_SECRET" | jq -ce '.data[0]')

// The test delivery carries a real record, as the REST API returns it.
async function findRecord(): Promise<Record<string, unknown>> {
  const res = await fetch(`${base}/entities/Opportunities?first=1&load-only=id,name,owner_id,step_id,revision`, {
    headers: { Authorization: `Basic ${auth}` },
  });
  const body = await res.json();
  if (!res.ok) throw new Error(`HTTP ${res.status} ${body.name}: ${body.message}`);
  if (!body.data.length) throw new Error("The self-test needs one opportunity in the space.");
  return body.data[0];
}

# The test delivery carries a real record, as the REST API returns it.
def find_record():
    response = session.get(
        f"{BASE}/entities/Opportunities",
        params={"first": 1, "load-only": "id,name,owner_id,step_id,revision"},
        timeout=30,
    )
    body = response.json()
    if not response.ok:
        sys.exit(f"HTTP {response.status_code} {body.get('name')}: {body.get('message')}")
    if not body["data"]:
        sys.exit("The self-test needs one opportunity in the space.")
    return body["data"][0]

# One line of JSON, like a delivery; event_time is UTC without an offset.
body=$(jq -cn --argjson entity "$record" --arg url "$RECEIVER_URL" --arg space "$SPACE_ID" \
  --arg time "$(date -u +%Y-%m-%dT%H:%M:%S)" \
  '[{webhook: {url: $url, events: ["Opportunity.Update"]}, event: "Opportunity.Update",
     event_time: $time, team_space_id: $space, entity: $entity, related_entity: null}]')
signature=$(signature_of "$body")
verify_signature "$body" "$signature"
curl -sS --fail-with-body "$RECEIVER_URL" \
  -H "Content-Type: application/json" \
  -H "WebHook-Signature: $signature" \
  -H "WebHook-Signature-Version: v1" \
  -H "Webhook-Delivery-Retry: 0" \
  --data-binary "$body"
echo

// A delivery signed the way Coevera signs, then what retries, replays, forgeries and a
// deleted webhook look like to the receiver.
async function selfTest(url: string): Promise<void> {
  const record = await findRecord();
  const delivery = (eventTime: Date) =>
    JSON.stringify([
      {
        webhook: { id: randomUUID(), url, events: ["Opportunity.Update"] },
        event: "Opportunity.Update",
        event_time: eventTime.toISOString().slice(0, -1),
        team_space_id: spaceId,
        entity: record,
        related_entity: null,
      },
    ]);
  const send = (body: string, signature: string, retry = 0) =>
    fetch(url, {
      method: "POST",
      headers: { "Content-Type": "application/json", "WebHook-Signature": signature, "WebHook-Signature-Version": "v1", "Webhook-Delivery-Retry": String(retry) },
      body,
    }).then((r) => r.status);
  const sign = (body: string) => createHmac("sha256", secret).update(body).digest("hex");

  const fresh = delivery(new Date());
  const old = delivery(new Date(Date.now() - 48 * 60 * 60 * 1000));
  const answers = {
    first: await send(fresh, sign(fresh)),
    retry: await send(fresh, sign(fresh), 1),
    old: await send(old, sign(old)),
    forged: await send(fresh, "0".repeat(64)),
    deleted: await fetch(url, { method: "DELETE" }).then((r) => r.status),
  };
  console.log("Answers:", JSON.stringify(answers));
  const expected = { first: 200, retry: 200, old: 200, forged: 401, deleted: 200 };
  if (JSON.stringify(answers) !== JSON.stringify(expected)) throw new Error(`Expected ${JSON.stringify(expected)}`);
  if (records.size !== 1 || seen.size !== 1) throw new Error("A retry or an old event was processed.");
}

# A delivery signed the way Coevera signs, then what retries, replays, forgeries and a
# deleted webhook look like to the receiver.
def self_test(url):
    record = find_record()

    def delivery(event_time):
        return json.dumps([{
            "webhook": {"id": str(uuid.uuid4()), "url": url, "events": ["Opportunity.Update"]},
            "event": "Opportunity.Update",
            "event_time": event_time.replace(tzinfo=None).isoformat(),
            "team_space_id": SPACE_ID,
            "entity": record,
            "related_entity": None,
        }]).encode("utf-8")

    def sign(body):
        return hmac.new(SECRET.encode("utf-8"), body, hashlib.sha256).hexdigest()

    def send(body, signature, retry=0):
        headers = {
            "Content-Type": "application/json",
            "WebHook-Signature": signature,
            "WebHook-Signature-Version": "v1",
            "Webhook-Delivery-Retry": str(retry),
        }
        status = requests.post(url, data=body, headers=headers, timeout=10).status_code
        jobs.join()  # wait until the receiver has processed it, so the checks below see the result
        return status

    now = datetime.now(timezone.utc)
    fresh = delivery(now)
    old = delivery(now - timedelta(hours=48))
    answers = {
        "first": send(fresh, sign(fresh)),
        "retry": send(fresh, sign(fresh), 1),
        "old": send(old, sign(old)),
        "forged": send(fresh, "0" * 64),
        "deleted": requests.delete(url, timeout=10).status_code,
    }
    print("Answers:", json.dumps(answers))
    expected = {"first": 200, "retry": 200, "old": 200, "forged": 401, "deleted": 200}
    if answers != expected:
        sys.exit(f"Expected {json.dumps(expected)}")
    if len(records) != 1 or len(seen) != 1:
        sys.exit("A retry or an old event was processed.")