Validating Webhook Requests

<?php

$signatureHeader = $_SERVER['HTTP_X_SATWS_SIGNATURE'] ?? '';
$bodyPayload = file_get_contents('php://input');
$signingSecret = getenv('SYNTAGE_WEBHOOK_SIGNING_SECRET');

if ($bodyPayload === false) {
    throw new RuntimeException('Missing webhook payload.');
}

if ($signingSecret === false) {
    throw new RuntimeException('Missing webhook signing secret.');
}

$values = [];

foreach (explode(',', $signatureHeader) as $part) {
    if (!str_contains($part, '=')) {
        throw new RuntimeException('Malformed webhook signature.');
    }

    [$key, $value] = explode('=', $part, 2);
    $values[$key][] = $value;
}

$timestamp = $values['t'][0] ?? null;
$signatures = $values['s'] ?? [];

if ($timestamp === null || $signatures === []) {
    throw new RuntimeException('Missing webhook signature.');
}

$toleranceSeconds = 300;

if (abs(time() - (int) $timestamp) > $toleranceSeconds) {
    throw new RuntimeException('Webhook signature timestamp is outside the tolerance window.');
}

$expectedSignature = hash_hmac('sha256', sprintf('%s.%s', $timestamp, $bodyPayload), $signingSecret);
$isValid = false;

foreach ($signatures as $signature) {
    if (hash_equals($expectedSignature, $signature)) {
        $isValid = true;
        break;
    }
}

if (!$isValid) {
    throw new RuntimeException('Invalid webhook signature.');
}

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

function validateSyntageWebhookSignature({
  signatureHeader,
  bodyPayload,
  signingSecret,
  now = Math.floor(Date.now() / 1000),
}) {
  if (!bodyPayload) {
    throw new Error("Missing webhook payload.");
  }

  if (!signingSecret) {
    throw new Error("Missing webhook signing secret.");
  }

  const values = new Map();

  for (const part of signatureHeader.split(",")) {
    const separatorIndex = part.indexOf("=");

    if (separatorIndex === -1) {
      throw new Error("Malformed webhook signature.");
    }

    const key = part.slice(0, separatorIndex);
    const value = part.slice(separatorIndex + 1);
    values.set(key, [...(values.get(key) ?? []), value]);
  }

  const timestamp = values.get("t")?.[0];
  const signatures = values.get("s") ?? [];

  if (!timestamp || signatures.length === 0) {
    throw new Error("Missing webhook signature.");
  }

  const toleranceSeconds = 300;

  if (Math.abs(now - Number(timestamp)) > toleranceSeconds) {
    throw new Error("Webhook signature timestamp is outside the tolerance window.");
  }

  const expectedSignature = createHmac("sha256", signingSecret)
    .update(`${timestamp}.${bodyPayload}`)
    .digest("hex");
  const expectedBuffer = Buffer.from(expectedSignature, "hex");

  const isValid = signatures.some((signature) => {
    const signatureBuffer = Buffer.from(signature, "hex");

    return (
      signatureBuffer.length === expectedBuffer.length &&
      timingSafeEqual(signatureBuffer, expectedBuffer)
    );
  });

  if (!isValid) {
    throw new Error("Invalid webhook signature.");
  }
}

import hashlib
import hmac
import time


def validate_syntage_webhook_signature(
    signature_header: str,
    body_payload: bytes,
    signing_secret: str,
    now: int | None = None,
) -> None:
    if not body_payload:
        raise ValueError("Missing webhook payload.")

    if not signing_secret:
        raise ValueError("Missing webhook signing secret.")

    values: dict[str, list[str]] = {}

    for part in signature_header.split(","):
        if "=" not in part:
            raise ValueError("Malformed webhook signature.")

        key, value = part.split("=", 1)
        values.setdefault(key, []).append(value)

    timestamp = values.get("t", [None])[0]
    signatures = values.get("s", [])

    if timestamp is None or not signatures:
        raise ValueError("Missing webhook signature.")

    now = now or int(time.time())
    tolerance_seconds = 300

    if abs(now - int(timestamp)) > tolerance_seconds:
        raise ValueError("Webhook signature timestamp is outside the tolerance window.")

    expected_signature = hmac.new(
        signing_secret.encode("utf-8"),
        f"{timestamp}.".encode("utf-8") + body_payload,
        hashlib.sha256,
    ).hexdigest()

    if not any(hmac.compare_digest(expected_signature, signature) for signature in signatures):
        raise ValueError("Invalid webhook signature.")