HMAC signature verification

X-ZKP-Signature: t=<unix seconds>,v1=<hex hmac-sha256>

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

const app = express();
// Important: raw body, not the parsed JSON
app.use(express.raw({ type: "application/json" }));

function verify(rawBody, signatureHeader) {
// parse t=...,v1=...
const parts = Object.fromEntries(signatureHeader.split(",").map((p) => p.split("=")));
const timestamp = parts.t;
const signature = (parts.v1 || "").toLowerCase();

// timestamp must be an integer within the 5-minute window
if (!/^d{1,12}$/.test(timestamp)) return false;
if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false;

// signature must be hex, exactly 64 chars for SHA-256
if (!/^[0-9a-f]{64}$/.test(signature)) return false;

const expected = createHmac("sha256", process.env.ZKP_WEBHOOK_SECRET)
  .update(timestamp + "." + rawBody)
  .digest("hex");

try {
  // mismatched length would throw - the format check above guards it
  return timingSafeEqual(Buffer.from(expected, "hex"), Buffer.from(signature, "hex"));
} catch {
  return false;
}
}

app.post("/webhooks/zkp", (req, res) => {
if (!verify(req.body, req.headers["x-zkp-signature"])) {
  return res.status(400).send("invalid signature");
}

const event = JSON.parse(req.body);
// deduplicate: the same event.id may be delivered more than once
if (seen(event.id)) return res.sendStatus(200);

console.log("verified:", event.type, event.data.invoice_id);
res.sendStatus(200);
});

<?php
$rawBody   = file_get_contents('php://input');
$signature = $_SERVER['HTTP_X_ZKP_SIGNATURE'] ?? '';

// 1. Structure: t=<int>,v1=<64 hex>
if (!preg_match('/^t=(d{1,12}),v1=([0-9a-f]{64})$/', $signature, $m)) {
  http_response_code(400); exit('malformed signature header');
}
$timestamp = $m[1]; $sig = $m[2];

// 2. 5-minute replay window
if (abs(time() - (int)$timestamp) > 300) { http_response_code(400); exit('stale timestamp'); }

// 3. Recompute HMAC over timestamp.rawBody, compare in constant time
$expected = hash_hmac('sha256', $timestamp . '.' . $rawBody, getenv('ZKP_WEBHOOK_SECRET'));
if (!hash_equals($expected, $sig)) { http_response_code(400); exit('bad signature'); }

$event = json_decode($rawBody, true);
// deduplicate: the same event id may be delivered more than once
if (alreadySeen($event['id'] ?? '')) { http_response_code(200); exit; }

error_log('verified: ' . $event['type']);
http_response_code(200);

import hmac
import hashlib
import re
import time
from flask import Flask, request, abort

app = Flask(__name__)
WEBHOOK_SECRET = b"whsec_..."

@app.post("/webhooks/zkp")
def webhook():
  header = request.headers.get("X-ZKP-Signature", "")
  m = re.fullmatch(r"t=(d{1,12}),v1=([0-9a-f]{64})", header)
  if not m:
      abort(400, "malformed signature header")
  timestamp, sig = m.group(1), m.group(2)

  expected = hmac.new(
      WEBHOOK_SECRET,
      f"{timestamp}.".encode() + request.get_data(),
      hashlib.sha256,
  ).hexdigest()
  if not hmac.compare_digest(expected, sig):
      abort(400, "bad signature")

  event = request.get_json()
  # deduplicate: the same event.id may be delivered more than once
  if already_seen(event.get("id")):
      return "", 200

  print("verified:", event["type"])
  return "", 200

secret:     whsec_zkp_test_vector_2026
timestamp:  1788788073
raw body:   {"id":"evt_test_001","type":"payment.confirmed","invoice_id":"inv_test_001"}
signed:     1788788073.{"id":"evt_test_001","type":"payment.confirmed","invoice_id":"inv_test_001"}
signature:  v1=ade537fa13aec79a6d1648bd7f197872066c161676c389243ab5c6b13fea7f52