Webhooks

Em vez de consultar a nota até um estado final, você pode receber o resultado da emissão por push. A NFE.io chama a URL do seu webhook, e o seu endpoint verifica a assinatura antes de confiar no conteúdo. Este guia cobre a verificação, que fica em nfeio.webhooks e não precisa de cliente nem de chave de API. Para cadastrar, alterar e testar webhooks, veja o cookbook Webhooks.

Como a NFE.io assina as entregas

Cada entrega traz o cabeçalho X-Hub-Signature com um HMAC-SHA1 do corpo bruto da requisição, calculado com o segredo que você cadastrou no webhook:

X-Hub-Signature: sha1=BCD17C02B9E3B40A18E745E7E04247E4AD2DD935

Junto vêm X-Hook-Id (identificador único da entrega), X-Hook-Event (o tipo, como service_invoice) e X-Hook-Attempts (quantas tentativas já falharam antes desta). O corpo das notas de serviço tem o formato {"action": "...", "payload": {...}}. A lista completa está no catálogo de eventos de NFS-e.

As duas funções

from nfeio.webhooks import construct_event, verify_signature

Função	O que faz
verify_signature(payload, signature, secret) -> bool	Confere sha1=<40 hex> contra o HMAC-SHA1 do corpo, em tempo constante. Nunca levanta exceção: entrada malformada, outro algoritmo ou segredo vazio dão False.
construct_event(payload, headers, secret) -> WebhookEvent	Verifica e só então interpreta o JSON. Levanta SignatureVerificationError se a assinatura não confere (sem ler o corpo) e InvalidParameterError se o corpo passar de 5 MiB ou não for um objeto JSON.

headers aceita qualquer mapeamento de cabeçalhos (a busca ignora maiúsculas e minúsculas) ou direto o valor de X-Hub-Signature. O WebhookEvent devolvido tem:

Atributo	Conteúdo
action	body["action"], como "issued_successfully" ou "ping".
event_type	Cabeçalho X-Hook-Event, como "service_invoice".
hook_id	Cabeçalho X-Hook-Id: a chave de deduplicação.
data	O payload. Nos eventos service_invoice, é um ServiceInvoice, com as mesmas propriedades da emissão. No ping, os dados do webhook.
body	O corpo inteiro decodificado (NfeObject).
raw	Os bytes recebidos.

Teste local

Para testar o seu endpoint sem depender de uma entrega real, assine um corpo de exemplo com o mesmo algoritmo:

import hashlib
import hmac
import json

from nfeio import ServiceInvoice
from nfeio.webhooks import construct_event, verify_signature

segredo = "um-segredo-longo-e-aleatorio"
corpo = json.dumps(
    {"action": "issued_successfully", "payload": {"id": "ID_DA_NOTA", "flowStatus": "Issued"}}
).encode()
assinatura = "sha1=" + hmac.new(segredo.encode(), corpo, hashlib.sha1).hexdigest().upper()

assert verify_signature(corpo, assinatura, segredo)
assert not verify_signature(corpo, assinatura, "outro-segredo")

evento = construct_event(
    corpo,
    {"X-Hub-Signature": assinatura, "X-Hook-Id": "entrega-1", "X-Hook-Event": "service_invoice"},
    segredo,
)
assert evento.action == "issued_successfully"
assert isinstance(evento.data, ServiceInvoice)
print(evento.hook_id, evento.data.id, evento.data.flow_status)  # entrega-1 ID_DA_NOTA Issued

Use o corpo bruto

A assinatura vale para os bytes exatos que a NFE.io enviou. Passe sempre o corpo bruto da requisição, nunca um json.dumps do corpo já interpretado: a ordem das chaves e os espaços mudam, e a verificação falha.

Framework	Corpo bruto	Cabeçalhos
Flask	request.get_data()	dict(request.headers)
FastAPI / Starlette	await request.body()	request.headers
Django	request.body	request.headers

Flask

import os

from flask import Flask, request

from nfeio import ServiceInvoice
from nfeio.errors import SignatureVerificationError
from nfeio.webhooks import WebhookEvent, construct_event

app = Flask(__name__)
SEGREDO = os.environ.get("NFEIO_WEBHOOK_SECRET", "um-segredo-longo-e-aleatorio")


def processar(evento: WebhookEvent) -> None:
    if evento.action == "issued_successfully" and isinstance(evento.data, ServiceInvoice):
        print("nota emitida:", evento.data.id, evento.data.flow_status)


@app.post("/nfeio/webhook")
def nfeio_webhook() -> tuple[str, int]:
    try:
        evento = construct_event(request.get_data(), dict(request.headers), SEGREDO)
    except SignatureVerificationError:
        return "", 400
    processar(evento)
    return "", 204

request.headers do Flask também funciona direto, mas não é um Mapping para o mypy; o dict(...) mantém a checagem de tipos limpa.

FastAPI

import os

from fastapi import FastAPI, HTTPException, Request, Response

from nfeio import ServiceInvoice
from nfeio.errors import SignatureVerificationError
from nfeio.webhooks import WebhookEvent, construct_event

app = FastAPI()
SEGREDO = os.environ.get("NFEIO_WEBHOOK_SECRET", "um-segredo-longo-e-aleatorio")


async def processar(evento: WebhookEvent) -> None:
    if evento.action == "issued_successfully" and isinstance(evento.data, ServiceInvoice):
        print("nota emitida:", evento.data.id, evento.data.flow_status)


@app.post("/nfeio/webhook", status_code=204)
async def nfeio_webhook(request: Request) -> Response:
    try:
        evento = construct_event(await request.body(), request.headers, SEGREDO)
    except SignatureVerificationError:
        raise HTTPException(status_code=400) from None
    await processar(evento)
    return Response(status_code=204)

Django

import os

from django.http import HttpRequest, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from nfeio import ServiceInvoice
from nfeio.errors import SignatureVerificationError
from nfeio.webhooks import WebhookEvent, construct_event

SEGREDO = os.environ.get("NFEIO_WEBHOOK_SECRET", "um-segredo-longo-e-aleatorio")


def processar(evento: WebhookEvent) -> None:
    if evento.action == "issued_successfully" and isinstance(evento.data, ServiceInvoice):
        print("nota emitida:", evento.data.id, evento.data.flow_status)


@csrf_exempt  # a NFE.io não envia token CSRF: a assinatura é a autenticação
@require_POST
def nfeio_webhook(request: HttpRequest) -> HttpResponse:
    try:
        evento = construct_event(request.body, request.headers, SEGREDO)
    except SignatureVerificationError:
        return HttpResponse(status=400)
    processar(evento)
    return HttpResponse(status=204)

Entregas repetidas

A entrega é at-least-once: o mesmo evento pode chegar mais de uma vez, e o protocolo não tem timestamp assinado, então não existe janela de replay a conferir. Torne o processamento idempotente, deduplicando por evento.hook_id (o X-Hook-Id) ou pelo id da nota, e responda 2xx rápido: enfileire o trabalho pesado em vez de fazê-lo dentro da requisição.

O ping carrega o segredo

A entrega de teste (client.webhooks.ping) chega com "action": "ping" e os dados do webhook, inclusive o segredo em texto. Não registre o corpo das entregas em log. O repr de WebhookEvent já omite o corpo de propósito.

Veja também

• Webhooks (cookbook): cadastrar, alterar, testar e remover.
• Conceitos de webhook e IPs de origem.
• Catálogo de eventos.