Webhooks

import crypto from "node:crypto";
import express from "express";

const app = express();

/** Returns true when the X-SellOrchard-Signature header matches the raw body. */
function verifySignature(rawBody, header, secret, toleranceSec = 300) {
  let timestamp;
  const signatures = [];
  for (const part of (header ?? "").split(",")) {
    const [key, value] = part.trim().split("=");
    if (key === "t") timestamp = value;
    if (key === "v1") signatures.push(value);
  }
  if (!timestamp || !/^\d+$/.test(timestamp) || signatures.length === 0) return false;

  // Reject old (replayed) deliveries.
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > toleranceSec) return false;

  const expected = crypto
    .createHmac("sha256", secret)
    .update(`${timestamp}.`)
    .update(rawBody) // the exact bytes received — not re-serialised JSON
    .digest("hex");

  return signatures.some(
    (sig) => sig.length === expected.length && crypto.timingSafeEqual(Buffer.from(sig), Buffer.from(expected)),
  );
}

// express.raw keeps the body as a Buffer so the signature can be checked.
app.post("/webhooks/sellorchard", express.raw({ type: "application/json" }), (req, res) => {
  const ok = verifySignature(req.body, req.get("X-SellOrchard-Signature"), process.env.SELLORCHARD_WEBHOOK_SECRET);
  if (!ok) return res.sendStatus(400);

  const event = JSON.parse(req.body.toString("utf8"));
  // Queue the work and reply quickly; dedupe on event.id.
  console.log(event.type, event.data.object.id);
  res.sendStatus(200);
});

app.listen(3000);

<?php

/** Returns true when the X-SellOrchard-Signature header matches the raw body. */
function verify_sellorchard_signature(string $payload, string $header, string $secret, int $tolerance = 300): bool
{
    $timestamp = null;
    $signatures = [];
    foreach (explode(',', $header) as $part) {
        [$key, $value] = array_pad(explode('=', trim($part), 2), 2, '');
        if ($key === 't') $timestamp = $value;
        if ($key === 'v1') $signatures[] = $value;
    }
    if ($timestamp === null || !ctype_digit($timestamp) || $signatures === []) return false;

    // Reject old (replayed) deliveries.
    if (abs(time() - (int) $timestamp) > $tolerance) return false;

    $expected = hash_hmac('sha256', $timestamp . '.' . $payload, $secret);
    foreach ($signatures as $signature) {
        if (hash_equals($expected, $signature)) return true;
    }
    return false;
}

$payload = file_get_contents('php://input'); // the exact bytes received
$header = $_SERVER['HTTP_X_SELLORCHARD_SIGNATURE'] ?? '';

if (!verify_sellorchard_signature($payload, $header, getenv('SELLORCHARD_WEBHOOK_SECRET'))) {
    http_response_code(400);
    exit;
}

$event = json_decode($payload, true);
// Queue the work and reply quickly; dedupe on $event['id'].
error_log($event['type'] . ' ' . $event['data']['object']['id']);
http_response_code(200);

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)


def verify_signature(payload: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    """Returns True when the X-SellOrchard-Signature header matches the raw body."""
    timestamp, signatures = None, []
    for part in (header or "").split(","):
        key, _, value = part.strip().partition("=")
        if key == "t":
            timestamp = value
        elif key == "v1":
            signatures.append(value)
    if not timestamp or not timestamp.isdigit() or not signatures:
        return False

    # Reject old (replayed) deliveries.
    if abs(time.time() - int(timestamp)) > tolerance:
        return False

    expected = hmac.new(secret.encode(), timestamp.encode() + b"." + payload, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, sig) for sig in signatures)


@app.post("/webhooks/sellorchard")
def sellorchard_webhook():
    payload = request.get_data()  # the exact bytes received
    if not verify_signature(payload, request.headers.get("X-SellOrchard-Signature", ""), os.environ["SELLORCHARD_WEBHOOK_SECRET"]):
        abort(400)

    event = request.get_json()
    # Queue the work and reply quickly; dedupe on event["id"].
    print(event["type"], event["data"]["object"]["id"])
    return "", 200