Webhooks

curl -X POST https://api.postkit.eu/v1/webhooks \
  -H "Authorization: Bearer pk_live_abc123..." \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://example.com/webhooks/postkit",
    "events": ["email.delivered", "email.bounced", "email.complained"]
  }'

const response = await fetch('https://api.postkit.eu/v1/webhooks', {
  method: 'POST',
  headers: {
    'Authorization': 'Bearer pk_live_abc123...',
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    url: 'https://example.com/webhooks/postkit',
    events: ['email.delivered', 'email.bounced', 'email.complained'],
  }),
});
const webhook = await response.json();

import requests

response = requests.post(
    'https://api.postkit.eu/v1/webhooks',
    headers={'Authorization': 'Bearer pk_live_abc123...'},
    json={
        'url': 'https://example.com/webhooks/postkit',
        'events': ['email.delivered', 'email.bounced', 'email.complained'],
    },
)
webhook = response.json()

# Verify a webhook signature with openssl
WEBHOOK_ID="whl_abc123def456"
WEBHOOK_TIMESTAMP="1711900000"
WEBHOOK_SECRET="whsec_dGVzdHNlY3JldGtleXRlc3Rz"

# 1. Reconstruct signed content
SIGNED_CONTENT="${WEBHOOK_ID}.${WEBHOOK_TIMESTAMP}.$(cat /tmp/body.json)"

# 2. Decode secret (strip whsec_ prefix, base64 decode)
SECRET_BYTES=$(echo -n "${WEBHOOK_SECRET#whsec_}" | base64 -d)

# 3. Compute HMAC-SHA256 and base64 encode
EXPECTED="v1,$(echo -n "$SIGNED_CONTENT" | openssl dgst -sha256 -hmac "$SECRET_BYTES" -binary | base64)"

echo "Expected: $EXPECTED"
# Compare with webhook-signature header

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

function verifyWebhook(
  body: string,
  headers: Record<string, string>,
  secret: string
): boolean {
  const msgId = headers['webhook-id'];
  const timestamp = headers['webhook-timestamp'];
  const signatures = headers['webhook-signature'];

  // Reject old timestamps (5 min tolerance)
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - parseInt(timestamp)) > 300) {
    throw new Error('Timestamp too old');
  }

  // Decode the signing secret
  const secretBytes = Buffer.from(
    secret.replace('whsec_', ''),
    'base64'
  );

  // Reconstruct signed content
  const signedContent = `${msgId}.${timestamp}.${body}`;

  // Compute expected signature
  const hmac = createHmac('sha256', secretBytes);
  hmac.update(signedContent);
  const expected = `v1,${hmac.digest('base64')}`;

  // Check against all provided signatures (key rotation support)
  return signatures.split(' ').some((sig) => {
    try {
      return timingSafeEqual(
        Buffer.from(expected),
        Buffer.from(sig)
      );
    } catch {
      return false;
    }
  });
}

import base64
import hashlib
import hmac
import time

def verify_webhook(body: str, headers: dict, secret: str) -> bool:
    msg_id = headers["webhook-id"]
    timestamp = headers["webhook-timestamp"]
    signatures = headers["webhook-signature"]

    # Reject old timestamps (5 min tolerance)
    if abs(time.time() - int(timestamp)) > 300:
        raise ValueError("Timestamp too old")

    # Decode the signing secret
    secret_bytes = base64.b64decode(secret.removeprefix("whsec_"))

    # Reconstruct signed content
    signed_content = f"{msg_id}.{timestamp}.{body}"

    # Compute expected signature
    mac = hmac.new(secret_bytes, signed_content.encode(), hashlib.sha256)
    expected = f"v1,{base64.b64encode(mac.digest()).decode()}"

    # Check against all provided signatures (key rotation support)
    return any(
        hmac.compare_digest(expected, sig)
        for sig in signatures.split(" ")
    )