Webhooks

import crypto from 'node:crypto';
import express from 'express';

const SECRET = process.env.BLOGENIA_WEBHOOK_SECRET; // whsec_…
const app = express();

function isValid(rawBody, headers) {
  const timestamp = headers['blogenia-timestamp'];
  const received = headers['blogenia-signature'] || '';
  const age = Math.abs(Date.now() / 1000 - Number(timestamp));
  if (!Number.isFinite(age) || age > 300) return false;
  const expected =
    'v1=' + crypto.createHmac('sha256', SECRET).update(timestamp + '.' + rawBody).digest('hex');
  return (
    received.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(received), Buffer.from(expected))
  );
}

// express.raw conserva el cuerpo en bruto.
app.post('/webhooks/blogenia', express.raw({ type: 'application/json' }), (req, res) => {
  const rawBody = req.body.toString('utf8');
  if (!isValid(rawBody, req.headers)) return res.sendStatus(400);

  const event = JSON.parse(rawBody);
  if (event.event === 'article.published') {
    console.log('Publicado:', event.data.article.title);
  }
  res.sendStatus(200);
});

app.listen(3000);

import hashlib
import hmac
import json
import os
import time

from flask import Flask, abort, request

SECRET = os.environ["BLOGENIA_WEBHOOK_SECRET"]  # whsec_…
app = Flask(__name__)


def is_valid(raw_body: bytes, headers) -> bool:
    timestamp = headers.get("Blogenia-Timestamp", "")
    received = headers.get("Blogenia-Signature", "")
    try:
        age = abs(time.time() - int(timestamp))
    except ValueError:
        return False
    if age > 300:
        return False
    signed = timestamp.encode() + b"." + raw_body
    expected = "v1=" + hmac.new(SECRET.encode(), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(received, expected)


@app.post("/webhooks/blogenia")
def blogenia():
    raw_body = request.get_data()  # cuerpo en bruto
    if not is_valid(raw_body, request.headers):
        abort(400)
    event = json.loads(raw_body)
    if event["event"] == "article.published":
        print("Publicado:", event["data"]["article"]["title"])
    return "", 200

<?php
$secret = getenv('BLOGENIA_WEBHOOK_SECRET'); // whsec_…

$rawBody = file_get_contents('php://input'); // cuerpo en bruto
$timestamp = $_SERVER['HTTP_BLOGENIA_TIMESTAMP'] ?? '';
$received = $_SERVER['HTTP_BLOGENIA_SIGNATURE'] ?? '';

if (!ctype_digit($timestamp) || abs(time() - (int) $timestamp) > 300) {
    http_response_code(400);
    exit;
}

$expected = 'v1=' . hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
if (!hash_equals($expected, $received)) {
    http_response_code(400);
    exit;
}

$event = json_decode($rawBody, true);
if ($event['event'] === 'article.published') {
    error_log('Publicado: ' . $event['data']['article']['title']);
}
http_response_code(200);