Segurança de webhooks

X-Manu-Signature: t=1714680000,v1=4f3c8b9e1d2a...

import crypto from "node:crypto";

const SECRET = process.env.MANU_WEBHOOK_SECRET!;

export function verifyManuSignature(
  rawBody: string,
  header: string,
): boolean {
  // Header tem o formato: "t=1714680000,v1=abcdef..."
  const parts = Object.fromEntries(
    header.split(",").map((p) => p.split("=") as [string, string]),
  );
  const timestamp = parts.t;
  const signature = parts.v1;
  if (!timestamp || !signature) return false;

  // Janela de 5 min pra mitigar replay attacks.
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - Number(timestamp)) > 300) return false;

  const expected = crypto
    .createHmac("sha256", SECRET)
    .update(`${timestamp}.${rawBody}`)
    .digest("hex");

  return crypto.timingSafeEqual(
    Buffer.from(signature, "hex"),
    Buffer.from(expected, "hex"),
  );
}

// Express:
app.post(
  "/webhook",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const sig = req.header("X-Manu-Signature");
    if (!sig || !verifyManuSignature(req.body.toString("utf8"), sig)) {
      return res.status(401).json({ error: "invalid signature" });
    }
    const event = JSON.parse(req.body.toString("utf8"));
    // ... process event
    res.status(200).json({ ok: true });
  },
);

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

SECRET = os.environ["MANU_WEBHOOK_SECRET"].encode()
app = Flask(__name__)

def verify(raw_body: bytes, header: str) -> bool:
    parts = dict(p.split("=", 1) for p in header.split(","))
    timestamp = parts.get("t")
    signature = parts.get("v1")
    if not timestamp or not signature:
        return False
    if abs(time.time() - int(timestamp)) > 300:
        return False
    payload = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(SECRET, payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(signature, expected)

@app.post("/webhook")
def webhook():
    sig = request.headers.get("X-Manu-Signature", "")
    if not verify(request.get_data(), sig):
        abort(401)
    event = request.get_json()
    # ... process event
    return {"ok": True}, 200

<?php
$secret = getenv('MANU_WEBHOOK_SECRET');
$rawBody = file_get_contents('php://input');
$header = $_SERVER['HTTP_X_MANU_SIGNATURE'] ?? '';

$parts = [];
foreach (explode(',', $header) as $p) {
    [$k, $v] = explode('=', $p, 2) + [null, null];
    if ($k && $v) $parts[$k] = $v;
}
$timestamp = $parts['t'] ?? null;
$signature = $parts['v1'] ?? null;

if (!$timestamp || !$signature
    || abs(time() - intval($timestamp)) > 300) {
    http_response_code(401);
    exit('invalid signature');
}

$expected = hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
if (!hash_equals($expected, $signature)) {
    http_response_code(401);
    exit('invalid signature');
}

// ... process $event = json_decode($rawBody, true);
http_response_code(200);
echo json_encode(['ok' => true]);