Callback signatures

import crypto from 'crypto'

export function verifyCallbackSignature(
  req: {
    headers: Record<string, string | undefined>
    rawBody: string
    url: string
  },
  signingSecret: string
): boolean {
  const timestamp = req.headers['x-mutationengine-timestamp']
  const nonce = req.headers['x-mutationengine-nonce']
  const signatureHeader = req.headers['x-mutationengine-signature']

  if (!timestamp || !nonce || !signatureHeader) return false
  if (!signatureHeader.startsWith('v2=')) return false

  const receivedSignature = signatureHeader.slice(3)

  const url = new URL(req.url)
  const pathAndQuery = `${url.pathname}${url.search}`

  const bodyHashHex = crypto
    .createHash('sha256')
    .update(req.rawBody, 'utf8')
    .digest('hex')

  const canonical = [
    timestamp,
    nonce,
    pathAndQuery,
    bodyHashHex,
    ''
  ].join('\n')

  const expectedSignature = crypto
    .createHmac('sha256', signingSecret)
    .update(canonical, 'utf8')
    .digest('base64')

  return crypto.timingSafeEqual(
    Buffer.from(expectedSignature),
    Buffer.from(receivedSignature)
  )
}

<?php

/**
 * Verify a Mutation Engine callback signature.
 *
 * @param array  $headers       Associative array of HTTP headers (lowercased keys recommended)
 * @param string $rawBody       Raw request body (exact bytes received)
 * @param string $requestUrl    Full request URL (as received by your server)
 * @param string $signingSecret Region-specific callback signing secret
 *
 * @return bool True if the signature is valid
 */
function verifyCallbackSignature(
    array $headers,
    string $rawBody,
    string $requestUrl,
    string $signingSecret
): bool {
    $timestamp = $headers['x-mutationengine-timestamp'] ?? null;
    $nonce     = $headers['x-mutationengine-nonce'] ?? null;
    $signature = $headers['x-mutationengine-signature'] ?? null;

    if (!$timestamp || !$nonce || !$signature) {
        return false;
    }

    if (!str_starts_with($signature, 'v2=')) {
        return false;
    }

    $receivedSignature = substr($signature, 3);

    $url = parse_url($requestUrl);
    $path = $url['path'] ?? '';
    $query = isset($url['query']) ? '?' . $url['query'] : '';
    $pathAndQuery = $path . $query;

    $bodyHashHex = hash('sha256', $rawBody);

    $canonical =
        $timestamp . "\n" .
        $nonce . "\n" .
        $pathAndQuery . "\n" .
        $bodyHashHex . "\n";

    $expectedSignature = base64_encode(
        hash_hmac('sha256', $canonical, $signingSecret, true)
    );

    // Constant-time comparison
    return hash_equals($expectedSignature, $receivedSignature);
}

import base64
import hashlib
import hmac
from urllib.parse import urlparse


def verify_callback_signature(
    headers: dict,
    raw_body: bytes,
    request_url: str,
    signing_secret: str,
) -> bool:
    """
    Verify a Mutation Engine callback signature.

    :param headers: Request headers (lowercased keys recommended)
    :param raw_body: Raw request body (exact bytes received)
    :param request_url: Full request URL
    :param signing_secret: Region-specific callback signing secret
    :return: True if the signature is valid
    """
    timestamp = headers.get("x-mutationengine-timestamp")
    nonce = headers.get("x-mutationengine-nonce")
    signature = headers.get("x-mutationengine-signature")

    if not timestamp or not nonce or not signature:
        return False

    if not signature.startswith("v2="):
        return False

    received_signature = signature[3:]

    parsed = urlparse(request_url)
    path_and_query = parsed.path
    if parsed.query:
        path_and_query += "?" + parsed.query

    body_hash_hex = hashlib.sha256(raw_body).hexdigest()

    canonical = (
        f"{timestamp}\n"
        f"{nonce}\n"
        f"{path_and_query}\n"
        f"{body_hash_hex}\n"
    )

    expected_signature = base64.b64encode(
        hmac.new(
            signing_secret.encode("utf-8"),
            canonical.encode("utf-8"),
            hashlib.sha256,
        ).digest()
    ).decode("ascii")

    return hmac.compare_digest(expected_signature, received_signature)