Webhooks

<?php

function verifySmsendSignature(
    string $header,
    string $rawBody,
    array $secrets,
    int $toleranceSeconds = 300,
): bool {
    $parts = [];

    foreach (explode(',', $header) as $pair) {
        [$key, $value] = array_pad(explode('=', trim($pair), 2), 2, null);
        $parts[$key] = $value;
    }

    $timestamp = $parts['t'] ?? null;
    $signature = $parts['v1'] ?? null;

    if ($timestamp === null || $signature === null || $signature === '') {
        return false;
    }

    if (! preg_match('/^\d{1,12}$/', $timestamp)) {
        return false;
    }

    if (abs(time() - (int) $timestamp) > $toleranceSeconds) {
        return false;
    }

    // Accept either secret during a rotation window.
    foreach ($secrets as $secret) {
        $expected = hash_hmac('sha256', $timestamp.'.'.$rawBody, $secret);

        if (hash_equals($expected, $signature)) {
            return true;
        }
    }

    return false;
}

// Read the RAW body — never a re-encoded parse.
$rawBody = file_get_contents('php://input');
$header = $_SERVER['HTTP_X_SMSEND_SIGNATURE'] ?? '';

if (! verifySmsendSignature($header, $rawBody, [$currentSecret, $previousSecret])) {
    http_response_code(400);
    exit;
}

$event = json_decode($rawBody, true);

import crypto from 'node:crypto'

function verifySmsendSignature(header, rawBody, secrets, toleranceSeconds = 300) {
  const parts = Object.fromEntries(
    header.split(',').map((pair) => pair.trim().split('=', 2)),
  )

  const timestamp = parts.t
  const signature = parts.v1

  if (!timestamp || !signature) return false
  if (!/^\d{1,12}$/.test(timestamp)) return false

  const skew = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp))
  if (skew > toleranceSeconds) return false

  // Accept either secret during a rotation window.
  return secrets.some((secret) => {
    const expected = crypto
      .createHmac('sha256', secret)
      .update(`${timestamp}.${rawBody}`)
      .digest('hex')

    return (
      expected.length === signature.length &&
      crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(signature))
    )
  })
}

// Express: mount express.raw() so req.body is the exact bytes received.
app.post('/smsend', express.raw({ type: 'application/json' }), (req, res) => {
  const rawBody = req.body.toString('utf8')

  if (!verifySmsendSignature(req.get('X-SMSend-Signature') ?? '', rawBody, secrets)) {
    return res.sendStatus(400)
  }

  const event = JSON.parse(rawBody)
  res.sendStatus(200)
})

import hashlib
import hmac
import re
import time


def verify_smsend_signature(header, raw_body, secrets, tolerance_seconds=300):
    parts = dict(
        pair.strip().split("=", 1) for pair in header.split(",") if "=" in pair
    )

    timestamp = parts.get("t")
    signature = parts.get("v1")

    if not timestamp or not signature:
        return False

    if not re.fullmatch(r"\d{1,12}", timestamp):
        return False

    if abs(int(time.time()) - int(timestamp)) > tolerance_seconds:
        return False

    # Accept either secret during a rotation window.
    signed = f"{timestamp}.{raw_body}".encode()

    return any(
        hmac.compare_digest(
            hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest(),
            signature,
        )
        for secret in secrets
    )


# Flask: request.get_data() returns the exact bytes received.
@app.post("/smsend")
def smsend_webhook():
    raw_body = request.get_data(as_text=True)
    header = request.headers.get("X-SMSend-Signature", "")

    if not verify_smsend_signature(header, raw_body, secrets):
        return "", 400

    event = request.get_json()
    return "", 200

# Reproduce a signature by hand to debug a mismatch.
TIMESTAMP=1787654461
SECRET="whsec_3f8a1c9e0b7d452a6e18cf30b95d7e421ac6f08b3d92e574a10c8fb26d94e753"
BODY='{"id":"01K3F9DELIVERYULID00000000","type":"MESSAGE_TERMINAL"}'

printf '%s.%s' "$TIMESTAMP" "$BODY" \
  | openssl dgst -sha256 -hmac "$SECRET" -hex \
  | sed 's/^.*= //'