Implementação

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') ?: 'firebanking';
$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;
}