Webhooks

{
  "event_id": "8c1f0b6e-5a24-4d31-9b77-0e2a6c4d1f93",
  "event": "cfdi.timbrado",
  "version": "v1",
  "created_at": "2026-08-26T18:42:07.512Z",
  "data": {
    "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "uuid": "6128396F-C09B-4EC6-8699-43DA5A244971",
    "serie": "A",
    "folio": 1042,
    "rfc_emisor": "EKU9003173C9",
    "rfc_receptor": "XAXX010101000",
    "total": 1160.00,
    "estatus": "timbrado",
    "stamped_at": "2026-08-26T18:42:06Z"
  }
}

{
  "event_id": "b4d9e2a1-7c30-4f58-8a12-91cf35e7d604",
  "event": "pago.timbrado",
  "version": "v1",
  "created_at": "2026-08-26T19:05:44.108Z",
  "data": {
    "id": "77c1e0f2-3a45-4b6c-8d9e-0f1a2b3c4d5e",
    "uuid": "9B3D71A8-2E44-4C10-B0F5-7A62D9E4C118",
    "serie": "P",
    "folio": 318,
    "rfc_emisor": "EKU9003173C9",
    "rfc_receptor": "XAXX010101000",
    "total": 0,
    "estatus": "timbrado",
    "stamped_at": "2026-08-26T19:05:43Z"
  }
}

{
  "event_id": "c7a52f18-9d06-4e73-bb21-4c8f0a5e6d92",
  "event": "cfdi.cancelado",
  "version": "v1",
  "created_at": "2026-08-26T20:11:02.774Z",
  "data": {
    "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "uuid": "6128396F-C09B-4EC6-8699-43DA5A244971",
    "serie": "A",
    "folio": 1042,
    "rfc_emisor": "EKU9003173C9",
    "rfc_receptor": "XAXX010101000",
    "total": 1160.00,
    "estatus": "cancelado",
    "motivo_cancelacion": "02",
    "estatus_cancelacion": "201",
    "cancelled_at": "2026-08-26T20:11:01Z"
  }
}

{
  "event_id": "d0e13b74-6f29-4a85-9c30-2b7e5d1a8f46",
  "event": "cfdi.cancelacion_en_proceso",
  "version": "v1",
  "created_at": "2026-08-26T20:14:39.220Z",
  "data": {
    "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "uuid": "6128396F-C09B-4EC6-8699-43DA5A244971",
    "serie": "A",
    "folio": 1042,
    "rfc_emisor": "EKU9003173C9",
    "rfc_receptor": "TME960709KP0",
    "total": 1160.00,
    "estatus": "en_proceso",
    "motivo_cancelacion": "02",
    "estatus_cancelacion": "En proceso",
    "cancelled_at": "2026-08-26T20:14:38Z"
  }
}

import crypto from 'crypto';
import express from 'express';

const app = express();
const SECRET = process.env.IPSO_WEBHOOK_SECRET; // whsec_live_...
const TOLERANCIA_SEGUNDOS = 5 * 60;

function verificar(rawBody, header, secret) {
  if (!header) return false;

  // t=<epoch>,v1=<hex>
  const partes = Object.fromEntries(
    header.split(',').map((p) => p.split('=', 2))
  );
  const t = partes.t;
  const recibida = partes.v1;
  if (!t || !recibida) return false;

  // 1. Tolerancia contra replay
  const edad = Math.abs(Math.floor(Date.now() / 1000) - Number(t));
  if (!Number.isFinite(edad) || edad > TOLERANCIA_SEGUNDOS) return false;

  // 2. HMAC-SHA256 sobre "<t>.<cuerpo_crudo>"
  const esperada = crypto
    .createHmac('sha256', secret)
    .update(`${t}.${rawBody.toString('utf8')}`)
    .digest('hex');

  // 3. Comparación en tiempo constante
  const a = Buffer.from(esperada, 'utf8');
  const b = Buffer.from(recibida, 'utf8');
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

app.post(
  '/webhooks/ipsofactura',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    if (!verificar(req.body, req.get('X-Ipso-Signature'), SECRET)) {
      return res.status(400).send('firma inválida');
    }

    const evento = JSON.parse(req.body.toString('utf8'));

    // Responde YA; procesa en segundo plano.
    res.status(200).send('ok');
    encolar(evento); // dedupe por evento.event_id
  }
);

import hashlib
import hmac
import json
import os
import time

from flask import Flask, request

app = Flask(__name__)
SECRET = os.environ["IPSO_WEBHOOK_SECRET"]  # whsec_live_...
TOLERANCIA_SEGUNDOS = 5 * 60


def verificar(raw_body: bytes, header: str | None, secret: str) -> bool:
    if not header:
        return False

    # t=<epoch>,v1=<hex>
    try:
        partes = dict(p.split("=", 1) for p in header.split(","))
        t = partes["t"]
        recibida = partes["v1"]
    except (ValueError, KeyError):
        return False

    # 1. Tolerancia contra replay
    try:
        if abs(int(time.time()) - int(t)) > TOLERANCIA_SEGUNDOS:
            return False
    except ValueError:
        return False

    # 2. HMAC-SHA256 sobre "<t>.<cuerpo_crudo>"
    firmado = t.encode("utf-8") + b"." + raw_body
    esperada = hmac.new(
        secret.encode("utf-8"), firmado, hashlib.sha256
    ).hexdigest()

    # 3. Comparación en tiempo constante
    return hmac.compare_digest(esperada, recibida)


@app.post("/webhooks/ipsofactura")
def recibir():
    raw = request.get_data()  # bytes crudos, sin parsear
    if not verificar(raw, request.headers.get("X-Ipso-Signature"), SECRET):
        return "firma inválida", 400

    evento = json.loads(raw)
    encolar(evento)  # dedupe por evento["event_id"]
    return "ok", 200