cobranca.py

"""
Python, só com a biblioteca padrão.

    P2D_KEY='p2d_test_...' python3 cobranca.py

Cobre: criar, idempotência, consultar, e a validação de assinatura do webhook —
que é o pedaço que você vai copiar para o seu servidor.
"""

import hashlib
import hmac
import json
import os
import time
import urllib.error
import urllib.request
import uuid

CHAVE = os.environ.get("P2D_KEY")
BASE = os.environ.get("P2D_URL", "https://api.pix2depix.com")
if not CHAVE:
    raise SystemExit("defina P2D_KEY com sua chave de API")


class ErroDaApi(Exception):
    """O `code` é o que não muda — escreva seu if em cima dele."""

    def __init__(self, status: int, corpo: dict):
        erro = corpo["error"]
        self.status = status
        self.code = erro["code"]
        self.request_id = erro["requestId"]
        super().__init__(f'[{status} {self.code}] {erro["message"]} (requestId: {self.request_id})')


def chamar(caminho: str, corpo: dict | None = None, cabecalhos: dict | None = None):
    requisicao = urllib.request.Request(
        f"{BASE}{caminho}",
        data=json.dumps(corpo).encode() if corpo is not None else None,
        headers={
            "authorization": f"Bearer {CHAVE}",
            "content-type": "application/json",
            **(cabecalhos or {}),
        },
        method="POST" if corpo is not None else "GET",
    )
    try:
        with urllib.request.urlopen(requisicao, timeout=30) as resposta:
            return json.loads(resposta.read()), resposta.status
    except urllib.error.HTTPError as erro:
        raise ErroDaApi(erro.code, json.loads(erro.read())) from None


# ── 1. Conta e taxa vigente ─────────────────────────────────────────────────
conta, _ = chamar("/v1/account")
print("ambiente:", conta["environment"], "· taxa:", conta["plan"]["feeBps"] / 100, "%")

# ── 2. Criar a cobrança ─────────────────────────────────────────────────────
pedido = f"pedido-{int(time.time())}"
corpo = {
    "amountInCents": 25000,
    "payerTaxNumber": "529.982.247-25",
    "payerName": "Maria Silva",
    "externalId": pedido,
    "metadata": {"origem": "exemplo"},
}

# Retry seguro de rede: se a resposta se perder, repetir com a mesma chave
# devolve a cobrança original em vez de criar uma segunda.
cobranca, status = chamar("/v1/charges", corpo, {"idempotency-key": str(uuid.uuid4())})
print(f'cobrança {cobranca["id"]} criada (HTTP {status})')
print("copia-e-cola:", cobranca["pix"]["qrCode"])
print("líquido:", cobranca["amount"]["netInCents"] / 100)

# ── 3. A mesma requisição de novo não duplica ───────────────────────────────
repetida, status = chamar("/v1/charges", corpo)
print(f"repetição: HTTP {status}, mesma cobrança:", repetida["id"] == cobranca["id"])

# ── 4. Sandbox: mover a cobrança sem Pix ────────────────────────────────────
if conta["environment"] == "test":
    for alvo in ("paid_pending_settlement", "settled"):
        movida, _ = chamar(f'/v1/test/charges/{cobranca["id"]}/advance', {"status": alvo})
        print("→", movida["status"], movida["settlement"].get("estimatedAt", ""))

# ── 5. A confirmação que decide se o produto sai ────────────────────────────
atual, _ = chamar(f'/v1/charges/{cobranca["id"]}')
print(
    "DePix liquidado — pode liberar o produto."
    if atual["status"] == "settled"
    else f'ainda em {atual["status"]} — não libere.'
)


# ── 6. Validação da assinatura do webhook ───────────────────────────────────
# Este é o pedaço que vai para o seu servidor.


def assinatura_valida(segredo: str, corpo_bruto: bytes, cabecalhos) -> bool:
    ts = int(cabecalhos.get("x-p2d-timestamp", 0) or 0)
    # Fora da janela de 5 minutos é entrega antiga sendo reenviada por terceiro.
    if not ts or abs(time.time() - ts) > 300:
        return False

    # O corpo **bruto**, como chegou. Reserializar o JSON muda a assinatura.
    assinatura = hmac.new(
        segredo.encode(), f"{ts}.".encode() + corpo_bruto, hashlib.sha256
    ).hexdigest()

    # compare_digest: `==` em string vaza o segredo pelo tempo de resposta.
    return hmac.compare_digest(f"v1={assinatura}", cabecalhos.get("x-p2d-signature", ""))


# Prova de que a função concorda com o algoritmo documentado.
if __name__ == "__main__":
    segredo = "whsec_exemplo"
    corpo_bruto = b'{"id":"evt_1","type":"charge.settled"}'
    ts = int(time.time())
    esperada = "v1=" + hmac.new(
        segredo.encode(), f"{ts}.".encode() + corpo_bruto, hashlib.sha256
    ).hexdigest()
    cabecalhos = {"x-p2d-timestamp": str(ts), "x-p2d-signature": esperada}

    # `if` e não `assert`: com `python3 -O` os asserts somem, e um teste que
    # some é um teste que passa sem testar.
    if not assinatura_valida(segredo, corpo_bruto, cabecalhos):
        raise SystemExit("a assinatura correta deveria validar")
    if assinatura_valida(segredo, corpo_bruto + b" ", cabecalhos):
        raise SystemExit("corpo adulterado deveria falhar")
    print("validação de assinatura: ok")