Verify the delivery contract before enabling delivery.

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

const timestamp = request.headers.get('PolicyWatcher-Timestamp');
const signature = request.headers.get('PolicyWatcher-Signature');
const eventId = request.headers.get('PolicyWatcher-Event-Id');
const rawBody = Buffer.from(await request.arrayBuffer());

if (!/^\d+$/.test(timestamp ?? '') || !/^v1=[a-f0-9]{64}$/.test(signature ?? '')) {
  throw new Error('Invalid webhook headers');
}

const age = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
if (age > 300) throw new Error('Stale webhook timestamp');

const expected = createHmac('sha256', process.env.POLICYWATCHER_WEBHOOK_SECRET)
  .update(`${timestamp}.`, 'utf8')
  .update(rawBody)
  .digest('hex');
const supplied = signature.slice(3);

if (!timingSafeEqual(Buffer.from(expected, 'hex'), Buffer.from(supplied, 'hex'))) {
  throw new Error('Invalid webhook signature');
}

// Check eventId in a bounded replay store before processing.

import hashlib
import hmac
import os
import re
import time

timestamp = request.headers["PolicyWatcher-Timestamp"]
signature = request.headers["PolicyWatcher-Signature"]
event_id = request.headers["PolicyWatcher-Event-Id"]
raw_body = request.get_data(cache=False, as_text=False)

if not timestamp.isdigit() or re.fullmatch(r"v1=[a-f0-9]{64}", signature) is None:
    raise ValueError("Invalid webhook headers")
if abs(int(time.time()) - int(timestamp)) > 300:
    raise ValueError("Stale webhook timestamp")

message = timestamp.encode() + b"." + raw_body
expected = hmac.new(
    os.environ["POLICYWATCHER_WEBHOOK_SECRET"].encode(),
    message,
    hashlib.sha256,
).hexdigest()

if not hmac.compare_digest(expected, signature[3:]):
    raise ValueError("Invalid webhook signature")

# Check event_id in a bounded replay store before processing.