Implementação

import express from 'express';

interface PixWebhookPayload {
  event: 'CashIn' | 'CashOut' | 'CashInReversal' | 'CashOutReversal';
  status: 'PENDING' | 'CONFIRMED' | 'ERROR';
  transactionType: 'PIX';
  movementType: 'CREDIT' | 'DEBIT';
  transactionId: string;
  externalId: string | null;
  endToEndId: string;
  pixKey: string | null;
  feeAmount: number;
  originalAmount: number;
  finalAmount: number;
  processingDate: string;
  errorCode: string | null;
  errorMessage: string | null;
  counterpart?: Counterpart;
  parentTransaction?: ParentTransaction;
  metadata: Record<string, unknown>;
}

interface Counterpart {
  name: string;
  document: string;
  bank: {
    bankISPB: string | null;
    bankName: string | null;
    bankCode: string | null;
    accountBranch: string | null;
    accountNumber: string | null;
  };
}

interface ParentTransaction {
  transactionId: string;
  externalId: string;
  endToEndId: string;
  processingDate: string;
  wasTotalRefunded: boolean;
  remainingAmountForRefund: number;
  metadata: Record<string, unknown>;
  counterpart: Counterpart;
}

const app = express();
app.use(express.json());

// Middleware de autenticação Basic Auth
function validateBasicAuth(
  req: express.Request,
  res: express.Response,
  next: express.NextFunction
) {
  const authHeader = req.headers.authorization;

  if (!authHeader || !authHeader.startsWith('Basic ')) {
    return res.status(401).json({ error: 'Unauthorized' });
  }

  const base64Credentials = authHeader.split(' ')[1];
  const credentials = Buffer.from(base64Credentials, 'base64').toString('ascii');
  const [username, password] = credentials.split(':');

  if (
    username !== process.env.WEBHOOK_USER ||
    password !== process.env.WEBHOOK_PASS
  ) {
    return res.status(401).json({ error: 'Invalid credentials' });
  }

  next();
}

// Set para controle de idempotência
const processedTransactions = new Set<string>();

app.post('/webhooks/pix', validateBasicAuth, async (req, res) => {
  const payload: PixWebhookPayload = req.body;

  // Responder rapidamente (webhook exige resposta em até 10s)
  res.status(200).json({ acknowledged: true });

  // Verificar idempotência
  if (processedTransactions.has(payload.transactionId)) {
    console.log(`Transação ${payload.transactionId} já processada`);
    return;
  }

  // Marcar como processada
  processedTransactions.add(payload.transactionId);

  // Processar assincronamente
  try {
    switch (payload.event) {
      case 'CashIn':
        await handleCashIn(payload);
        break;
      case 'CashOut':
        await handleCashOut(payload);
        break;
      case 'CashInReversal':
        await handleCashInReversal(payload);
        break;
      case 'CashOutReversal':
        await handleCashOutReversal(payload);
        break;
    }
  } catch (error) {
    console.error(`Erro ao processar ${payload.event}:`, error);
    processedTransactions.delete(payload.transactionId);
  }
});

async function handleCashIn(payload: PixWebhookPayload) {
  console.log(`[CashIn] Recebido: R$ ${payload.finalAmount}`);
  // await orderService.markAsPaid(payload.externalId);
}

async function handleCashOut(payload: PixWebhookPayload) {
  console.log(`[CashOut] Enviado: R$ ${payload.originalAmount}`);
  // await transferService.markAsCompleted(payload.transactionId);
}

async function handleCashInReversal(payload: PixWebhookPayload) {
  console.log(`[CashInReversal] Estornado: R$ ${payload.originalAmount}`);
  // await refundService.markAsCompleted(payload.transactionId);
}

async function handleCashOutReversal(payload: PixWebhookPayload) {
  console.log(`[CashOutReversal] Devolvido: R$ ${payload.finalAmount}`);
  // await balanceService.credit(payload.finalAmount);
}

app.listen(3000);

from flask import Flask, request, jsonify
from functools import wraps
import base64
import os
from typing import Dict, Any, Optional
from dataclasses import dataclass

app = Flask(__name__)
processed_transactions: set = set()

@dataclass
class PixWebhookPayload:
    event: str
    status: str
    transaction_id: str
    external_id: Optional[str]
    end_to_end_id: str
    fee_amount: float
    original_amount: float
    final_amount: float
    counterpart: Optional[Dict[str, Any]]
    parent_transaction: Optional[Dict[str, Any]]

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'PixWebhookPayload':
        return cls(
            event=data.get('event'),
            status=data.get('status'),
            transaction_id=data.get('transactionId'),
            external_id=data.get('externalId'),
            end_to_end_id=data.get('endToEndId'),
            fee_amount=data.get('feeAmount', 0),
            original_amount=data.get('originalAmount', 0),
            final_amount=data.get('finalAmount', 0),
            counterpart=data.get('counterpart'),
            parent_transaction=data.get('parentTransaction'),
        )


def require_basic_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        auth_header = request.headers.get('Authorization')

        if not auth_header or not auth_header.startswith('Basic '):
            return jsonify({'error': 'Unauthorized'}), 401

        try:
            credentials = base64.b64decode(
                auth_header.split(' ')[1]
            ).decode('utf-8')
            username, password = credentials.split(':')

            if (
                username != os.environ.get('WEBHOOK_USER') or
                password != os.environ.get('WEBHOOK_PASS')
            ):
                return jsonify({'error': 'Invalid credentials'}), 401
        except Exception:
            return jsonify({'error': 'Invalid auth header'}), 401

        return f(*args, **kwargs)
    return decorated


@app.route('/webhooks/pix', methods=['POST'])
@require_basic_auth
def handle_pix_webhook():
    data = request.get_json()
    payload = PixWebhookPayload.from_dict(data)

    # Idempotência
    if payload.transaction_id in processed_transactions:
        return jsonify({'acknowledged': True}), 200

    processed_transactions.add(payload.transaction_id)

    # Processar
    if payload.event == 'CashIn':
        print(f"[CashIn] R$ {payload.final_amount:.2f}")
    elif payload.event == 'CashOut':
        print(f"[CashOut] R$ {payload.original_amount:.2f}")
    elif payload.event == 'CashInReversal':
        print(f"[CashInReversal] R$ {payload.original_amount:.2f}")
    elif payload.event == 'CashOutReversal':
        print(f"[CashOutReversal] R$ {payload.final_amount:.2f}")

    return jsonify({'acknowledged': True}), 200


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000)

<?php

$WEBHOOK_USER = getenv('WEBHOOK_USER') ?: 'brbtc';
$WEBHOOK_PASS = getenv('WEBHOOK_PASS') ?: 'secret';
$PROCESSED_FILE = '/tmp/processed_transactions.json';

function validateBasicAuth($user, $pass): bool {
    $authHeader = $_SERVER['HTTP_AUTHORIZATION'] ?? '';

    if (empty($authHeader) || !str_starts_with($authHeader, 'Basic ')) {
        return false;
    }

    $credentials = base64_decode(substr($authHeader, 6));
    list($u, $p) = explode(':', $credentials, 2);

    return $u === $user && $p === $pass;
}

function isProcessed($txId): bool {
    global $PROCESSED_FILE;
    if (!file_exists($PROCESSED_FILE)) return false;
    $processed = json_decode(file_get_contents($PROCESSED_FILE), true) ?? [];
    return in_array($txId, $processed);
}

function markProcessed($txId): void {
    global $PROCESSED_FILE;
    $processed = file_exists($PROCESSED_FILE)
        ? json_decode(file_get_contents($PROCESSED_FILE), true) ?? []
        : [];
    $processed[] = $txId;
    file_put_contents($PROCESSED_FILE, json_encode(array_slice($processed, -10000)));
}

// Validações
if ($_SERVER['REQUEST_METHOD'] !== 'POST') {
    http_response_code(405);
    exit;
}

if (!validateBasicAuth($WEBHOOK_USER, $WEBHOOK_PASS)) {
    http_response_code(401);
    echo json_encode(['error' => 'Unauthorized']);
    exit;
}

$payload = json_decode(file_get_contents('php://input'), true);

// Responder rapidamente
http_response_code(200);
header('Content-Type: application/json');
echo json_encode(['acknowledged' => true]);

if (function_exists('fastcgi_finish_request')) {
    fastcgi_finish_request();
}

// Idempotência
if (isProcessed($payload['transactionId'])) {
    exit;
}

markProcessed($payload['transactionId']);

// Processar
switch ($payload['event']) {
    case 'CashIn':
        error_log("[CashIn] R$ " . $payload['finalAmount']);
        break;
    case 'CashOut':
        error_log("[CashOut] R$ " . $payload['originalAmount']);
        break;
    case 'CashInReversal':
        error_log("[CashInReversal] R$ " . $payload['originalAmount']);
        break;
    case 'CashOutReversal':
        error_log("[CashOutReversal] R$ " . $payload['finalAmount']);
        break;
}

🔐 Chave de assinatura HMAC

Permite que seu sistema valide criptograficamente que os webhooks
vieram da Brasil Bitcoin e não foram alterados. Opcional — sem ela, os
webhooks continuam sendo entregues normalmente.

┌──────────────────────────────────────────────────────┐
│ 🔒 Você ainda não tem uma chave configurada          │
│    Webhooks são entregues sem assinatura.            │
│    Gere uma chave para começar a validar a origem    │
│    dos webhooks no seu sistema.                      │
│                                                      │
│           [🔐 Gerar minha chave]                      │
└──────────────────────────────────────────────────────┘

🔐 Chave de assinatura HMAC

Chave atual:    •••••••• a14c78
Última rotação: 19/05/2026 14:32

[🔄 Rotacionar]  [📖 Como validar no meu sistema ↗]

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

const app = express();

// Opção A: rota dedicada com matcher amplo — captura qualquer Content-Type
// (inclusive `application/json; charset=utf-8` que vem do Brasil Bitcoin).
app.use('/webhooks', express.raw({ type: '*/*' }));

function isValidSignature(rawBody: Buffer, received: string, secret: string): boolean {
  const expected = crypto.createHmac('sha256', secret).update(rawBody).digest('hex');
  const a = Buffer.from(expected, 'hex');
  const b = Buffer.from(received, 'hex');
  if (a.length !== b.length) return false;
  return crypto.timingSafeEqual(a, b);
}

app.post('/webhooks/pix', (req, res) => {
  const signature = req.header('X-Avista-Signature');
  // CRÍTICO: rejeite requisições SEM o header, não só quando a assinatura está errada
  if (!signature || !isValidSignature(req.body, signature, process.env.BRBTC_WEBHOOK_SECRET!)) {
    return res.status(401).json({ error: 'invalid signature' });
  }

  const payload = JSON.parse(req.body.toString('utf8'));
  // ...processar payload
  res.status(200).json({ acknowledged: true });
});

// Opção B: preserve o rawBody em paralelo ao parsing automático.
// Útil quando você já tem `express.json()` global e não pode trocar.
app.use(express.json({
  verify: (req, _res, buf) => {
    (req as any).rawBody = buf; // Buffer com o corpo bruto
  },
}));

app.post('/webhooks/pix', (req, res) => {
  const signature = req.header('X-Avista-Signature');
  const rawBody = (req as any).rawBody as Buffer | undefined;
  if (!signature || !rawBody || !isValidSignature(rawBody, signature, secret)) {
    return res.status(401).json({ error: 'invalid signature' });
  }
  // req.body já vem parseado aqui
  res.status(200).json({ acknowledged: true });
});

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

app = Flask(__name__)
SECRET = os.environ['BRBTC_WEBHOOK_SECRET']

@app.route('/webhooks/pix', methods=['POST'])
def webhook():
    raw_body = request.get_data()  # bytes UTF-8, antes de JSON parse
    received = request.headers.get('X-Avista-Signature', '')
    if not received:
        abort(401)
    expected = hmac.new(SECRET.encode('utf-8'), raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, received):
        abort(401)

    payload = request.get_json()
    # ...processar payload
    return {'acknowledged': True}, 200

func validSignature(rawBody []byte, received, secret string) bool {
    if received == "" {
        return false
    }
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write(rawBody)
    expected := hex.EncodeToString(mac.Sum(nil))
    return hmac.Equal([]byte(expected), []byte(received))
}

http.HandleFunc("/webhooks/pix", func(w http.ResponseWriter, r *http.Request) {
    body, _ := io.ReadAll(r.Body)
    if !validSignature(body, r.Header.Get("X-Avista-Signature"), os.Getenv("BRBTC_WEBHOOK_SECRET")) {
        http.Error(w, "invalid signature", http.StatusUnauthorized)
        return
    }
    // json.Unmarshal(body, &payload) ...
    w.WriteHeader(http.StatusOK)
})

<?php
function validSignature(string $rawBody, string $received, string $secret): bool {
    if ($received === '') return false;
    $expected = hash_hmac('sha256', $rawBody, $secret);
    return hash_equals($expected, $received);  // timing-safe em PHP
}

$rawBody = file_get_contents('php://input');
$received = $_SERVER['HTTP_X_BRBTC_SIGNATURE'] ?? '';
if (!validSignature($rawBody, $received, getenv('BRBTC_WEBHOOK_SECRET'))) {
    http_response_code(401);
    echo json_encode(['error' => 'invalid signature']);
    exit;
}

$payload = json_decode($rawBody, true);
// ...processar payload
http_response_code(200);
echo json_encode(['acknowledged' => true]);