Webhooks

import { createHmac, timingSafeEqual } from "crypto";

// Express setup (capture raw body before parsing):
// app.use(express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }));

/**
* Verifies an HMAC-SHA256 webhook signature.
*
* @param {import('http').IncomingMessage & { rawBody?: Buffer }} req
* @param {string} secret - The shared webhook secret.
* @returns {boolean}
*/
function verifyWebhook(req, secret) {
const signature = req.headers["x-webhook-signature"];
const rawBody = req.rawBody; // raw bytes, NOT req.body

if (typeof signature !== "string" || !Buffer.isBuffer(rawBody)) {
  return false;
}

const expected = createHmac("sha256", secret).update(rawBody).digest("hex");

const expectedBuf = Buffer.from(expected, "hex");
const signatureBuf = Buffer.from(signature, "hex");

// timingSafeEqual throws on length mismatch, so guard first.
// Length itself isn't secret, so this comparison need not be constant-time.
if (expectedBuf.length !== signatureBuf.length) {
  return false;
}
if (!timingSafeEqual(expectedBuf, signatureBuf)) {
  return false;
}

// Parse ONLY after the signature is verified.
let payload;
try {
  payload = JSON.parse(rawBody);
} catch {
  return false;
}

// Reject replays: reject both stale and future-dated timestamps.
const webhookTime = new Date(payload.timestamp).getTime();
const skew = Date.now() - webhookTime;
if (Number.isNaN(webhookTime) || Math.abs(skew) > 5 * 60 * 1000) {
  return false;
}

return true;
}

export { verifyWebhook };

import hashlib
import hmac
import json
import time
from typing import Optional


def verify_webhook(raw_body: bytes, signature: str, secret: str,
                  max_skew_seconds: int = 300) -> Optional[dict]:
  """
  Verifies an HMAC-SHA256 webhook signature and returns the parsed
  payload if valid, or None if verification fails.

  raw_body: the exact raw request bytes (NOT a parsed/re-serialized object)
  signature: the hex-encoded signature from the request header
  secret: the shared webhook secret
  max_skew_seconds: allowed clock skew in either direction (default 5 min)
  """
  if not isinstance(raw_body, (bytes, bytearray)) or not isinstance(signature, str):
      return None

  expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()

  try:
      expected_bytes = bytes.fromhex(expected)
      signature_bytes = bytes.fromhex(signature)
  except ValueError:
      # signature wasn't valid hex
      return None

  if len(expected_bytes) != len(signature_bytes):
      return None

  if not hmac.compare_digest(expected_bytes, signature_bytes):
      return None

  # Parse ONLY after the signature is verified
  try:
      payload = json.loads(raw_body)
  except (json.JSONDecodeError, UnicodeDecodeError):
      return None

  if not isinstance(payload, dict):
      return None

  webhook_time = payload.get("timestamp")
  if not isinstance(webhook_time, (int, float)):
      return None

  # Reject both stale and future-dated timestamps
  skew = time.time() - webhook_time
  if abs(skew) > max_skew_seconds:
      return None

  return payload


# Example usage:
# raw_body = request.get_data()  # Flask; use request.body for Django
# signature = request.headers.get("X-Webhook-Signature", "")
# payload = verify_webhook(raw_body, signature, secret)
# if payload is None:
#     abort(401)

<?php

/**
* Verifies an HMAC-SHA256 webhook signature and returns the parsed
* payload if valid, or null if verification fails.
*
* @param string $rawBody          The exact raw request body (file_get_contents('php://input'))
* @param string|null $signature   The hex-encoded signature from the request header
* @param string $secret           The shared webhook secret
* @param int $maxSkewSeconds      Allowed clock skew in either direction (default 5 minutes)
* @return array|null
*/
function verifyWebhook(string $rawBody, ?string $signature, string $secret, int $maxSkewSeconds = 300): ?array
{
  if ($signature === null || $signature === '') {
      return null;
  }

  $expected = hash_hmac('sha256', $rawBody, $secret);

  $expectedBytes = hex2bin($expected);
  $signatureBytes = @hex2bin($signature); // '@' suppresses warning on invalid hex

  if ($expectedBytes === false || $signatureBytes === false) {
      return null;
  }

  if (strlen($expectedBytes) !== strlen($signatureBytes)) {
      return null;
  }

  if (!hash_equals($expectedBytes, $signatureBytes)) {
      return null;
  }

  // Parse ONLY after the signature is verified
  $payload = json_decode($rawBody, true);
  if (json_last_error() !== JSON_ERROR_NONE || !is_array($payload)) {
      return null;
  }

  if (!isset($payload['timestamp'])) {
      return null;
  }

  // Adjust to (int) $payload['timestamp'] if your timestamp is a Unix integer
  // rather than an ISO 8601 / parseable date string.
  $webhookTime = strtotime((string) $payload['timestamp']);
  if ($webhookTime === false) {
      return null;
  }

  // Reject both stale and future-dated timestamps
  $skew = time() - $webhookTime;
  if (abs($skew) > $maxSkewSeconds) {
      return null;
  }

  return $payload;
}

// Example usage:
// $rawBody = file_get_contents('php://input');
// $signature = $_SERVER['HTTP_X_WEBHOOK_SIGNATURE'] ?? null;
// $payload = verifyWebhook($rawBody, $signature, $secret);
// if ($payload === null) {
//     http_response_code(401);
//     exit;
// }