Webhook signature verification

const crypto = require('crypto');
const express = require('express');

const app = express();

// key point: verify the signature against the raw body; do not run express.json() first
app.post('/chainshield/webhook',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    const secret = process.env.CS_WEBHOOK_SECRET;   // whsec_xxx — server side only
    const timestamp = req.header('X-CS-Timestamp') || '';
    const signature = req.header('X-CS-Signature') || '';
    const rawBody = req.body.toString('utf8');

    // 1) timestamp skew ≤ 300 seconds
    const skew = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
    if (!timestamp || Number.isNaN(skew) || skew > 300) {
      return res.status(400).json({ ok: false, message: 'timestamp outside the allowed window (±300 seconds)' });
    }

    // 2) HMAC-SHA256 signature
    const expected = 'sha256=' + crypto
      .createHmac('sha256', secret)
      .update(`${timestamp}.${rawBody}`)
      .digest('hex');

    const a = Buffer.from(expected);
    const b = Buffer.from(signature);
    if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
      return res.status(401).json({ ok: false, message: 'signature verification failed' });
    }

    // 3) dedupe: handle the same endpoint_id + event_id only once
    const event = JSON.parse(rawBody);
    // example: check those keys for duplicates before running business logic

    // 4) return 2xx only after processing completes
    res.status(200).json({ ok: true });
  });

app.listen(3000);

import hmac
import hashlib
import os
import time
from flask import Flask, request, jsonify

app = Flask(__name__)
SECRET = os.environ["CS_WEBHOOK_SECRET"].encode()   # whsec_xxx — server side only

@app.post("/chainshield/webhook")
def webhook():
    raw_body = request.get_data()                   # raw bytes; do not use request.json
    timestamp = request.headers.get("X-CS-Timestamp", "")
    signature = request.headers.get("X-CS-Signature", "")

    # 1) timestamp skew ≤ 300 seconds
    try:
        skew = abs(int(time.time()) - int(timestamp))
    except (TypeError, ValueError):
        return jsonify(ok=False, message="missing or invalid timestamp"), 400
    if skew > 300:
        return jsonify(ok=False, message="timestamp outside the allowed window (±300 seconds)"), 400

    # 2) HMAC-SHA256 signature (constant-time comparison)
    signed_payload = timestamp.encode() + b"." + raw_body
    expected = "sha256=" + hmac.new(SECRET, signed_payload, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature):
        return jsonify(ok=False, message="signature verification failed"), 401

    # 3) dedupe by endpoint_id + event_id before returning 2xx
    return jsonify(ok=True), 200

<?php
$secret = getenv('CS_WEBHOOK_SECRET');        // whsec_xxx — server side only
$rawBody = file_get_contents('php://input');   // raw request body

$timestamp = $_SERVER['HTTP_X_CS_TIMESTAMP'] ?? '';
$signature = $_SERVER['HTTP_X_CS_SIGNATURE'] ?? '';

// 1) timestamp skew ≤ 300 seconds
if ($timestamp === '' || abs(time() - (int)$timestamp) > 300) {
    http_response_code(400);
    echo json_encode(['ok' => false, 'message' => 'timestamp outside the allowed window (±300 seconds)']);
    exit;
}

// 2) HMAC-SHA256 signature
$expected = 'sha256=' . hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
if (!hash_equals($expected, $signature)) {   // hash_equals is a constant-time comparison
    http_response_code(401);
    echo json_encode(['ok' => false, 'message' => 'signature verification failed']);
    exit;
}

// 3) dedupe, then return 2xx after processing
http_response_code(200);
echo json_encode(['ok' => true]);