Webhook Subscriptions

{
    "eventType": "PAYROLL_PROCESSED",
    "webhookEventId": "11111111-1111-1111-1111-111111111111",
    "webhookSubscriptionId": "00000000-0000-0000-0000-000000000000"
}

const { createHmac, timingSafeEqual } = require('node:crypto');

function verifyWebhook(secret, headers, rawBody) {
    const id = headers['webhook-id'];
    const timestamp = headers['webhook-timestamp'];
    const signatureHeader = headers['webhook-signature'];
    if (!id || !/^\d+$/.test(timestamp ?? '') || !signatureHeader) {
        return false;
    }
    if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) {
        return false;
    }

    const key = Buffer.from(secret.slice('whsec_'.length), 'base64');
    const expected = createHmac('sha256', key).update(`${id}.${timestamp}.${rawBody}`).digest();
    return signatureHeader.split(' ').some((entry) => {
        const [version, signature] = entry.split(',');
        const received = Buffer.from(signature ?? '', 'base64');
        return version === 'v1' && received.length === expected.length && timingSafeEqual(received, expected);
    });
}

import base64
import hashlib
import hmac
import time


def verify_webhook(secret: str, headers: dict, raw_body: bytes) -> bool:
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp", "")
    signature_header = headers.get("webhook-signature")
    if not msg_id or not timestamp.isdigit() or not signature_header:
        return False
    if abs(time.time() - int(timestamp)) > 300:
        return False

    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode()
    for entry in signature_header.split(" "):
        version, _, signature = entry.partition(",")
        if version == "v1" and hmac.compare_digest(signature, expected):
            return True
    return False