Implementation

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());

// Basic Auth authentication middleware
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 for idempotency control
const processedTransactions = new Set<string>();

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

  // Respond quickly (webhook requires response within 10s)
  res.status(200).json({ acknowledged: true });

  // Check idempotency
  if (processedTransactions.has(payload.transactionId)) {
    console.log(`Transaction ${payload.transactionId} already processed`);
    return;
  }

  // Mark as processed
  processedTransactions.add(payload.transactionId);

  // Process asynchronously
  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(`Error processing ${payload.event}:`, error);
    processedTransactions.delete(payload.transactionId);
  }
});

async function handleCashIn(payload: PixWebhookPayload) {
  console.log(`[CashIn] Received: R$ ${payload.finalAmount}`);
}

async function handleCashOut(payload: PixWebhookPayload) {
  console.log(`[CashOut] Sent: R$ ${payload.originalAmount}`);
}

async function handleCashInReversal(payload: PixWebhookPayload) {
  console.log(`[CashInReversal] Refunded: R$ ${payload.originalAmount}`);
}

async function handleCashOutReversal(payload: PixWebhookPayload) {
  console.log(`[CashOutReversal] Returned: R$ ${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)
    if payload.transaction_id in processed_transactions:
        return jsonify({'acknowledged': True}), 200
    processed_transactions.add(payload.transaction_id)
    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)));
}

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);

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

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

if (isProcessed($payload['transactionId'])) {
    exit;
}

markProcessed($payload['transactionId']);

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;
}

🔐 HMAC Signing Key

Allows your system to cryptographically validate that webhooks came
from Brasil Bitcoin and were not modified. Optional — without it, webhooks
keep being delivered normally.

┌──────────────────────────────────────────────────────┐
│ 🔒 You don't have a key configured yet               │
│    Webhooks are delivered without a signature.       │
│    Generate a key to start validating the origin     │
│    of webhooks on your system.                       │
│                                                      │
│           [🔐 Generate my key]                        │
└──────────────────────────────────────────────────────┘

🔐 HMAC Signing Key

Current key:   •••••••• a14c78
Last rotated:  05/19/2026 14:32

[🔄 Rotate]  [📖 How to validate in my system ↗]

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

const app = express();

// Option A: dedicated route with broad matcher — captures any Content-Type
// (including `application/json; charset=utf-8` from 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');
  // CRITICAL: reject requests WITHOUT the header, not only when signature is wrong
  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'));
  // ...process payload
  res.status(200).json({ acknowledged: true });
});

// Option B: preserve the rawBody in parallel to automatic parsing.
// Useful when you already have `express.json()` global and cannot replace it.
app.use(express.json({
  verify: (req, _res, buf) => {
    (req as any).rawBody = buf; // Buffer with the raw body
  },
}));

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 is already parsed here
  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()  # UTF-8 bytes, before 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()
    # ...process 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 in 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);
// ...process payload
http_response_code(200);
echo json_encode(['acknowledged' => true]);