fix(emission): validate emitente CNPJ against branch certificate at emission

CRITICAL (F2 review, Opus+Fable): the service decoupled emitente
(free-form payload.emitente.cnpj) from the certificate (looked up by
branch_ref) with nothing re-checking the equality the auto had structurally
via Branch.cnpj. A caller passing emitente.cnpj=B with a branch whose
certificate CNPJ=A got a signed, persisted ASSINADO document with
chave/emit=B but signature=A, having burned a nNF.

Adds EmitenteCertificateCnpjMismatchError, checked in the completeness
block (before allocate_fiscal_number, so a mismatch never consumes a
número), normalizing both sides (digits only) before comparing. Router maps
it to 409 emitente_certificate_cnpj_mismatch; message never leaks the
certificate's CNPJ.

Test: tests/emission/test_emissao.py::
test_emitente_cnpj_divergente_do_certificado_e_409_e_nao_queima_numero --
uploads a cert for CNPJ A, POSTs emissão with the same branch_ref but
emitente.cnpj=B, asserts 409 + no fiscal number consumed (a follow-up
emission with the matching CNPJ gets the number that would have been
burned).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jonatanritter
2026-08-08 15:30:58 -03:00
co-authored by Claude Opus 4.8
parent d9c1c444fb
commit 3e3a1abc6f
3 changed files with 94 additions and 2 deletions
+55
View File
@@ -355,6 +355,61 @@ async def test_serie_ausente_e_409_fiscal_config_missing(db_session):
assert response.json()["detail"]["code"] == "fiscal_config_missing"
# --- FIX 1 (F2 review): emitente.cnpj vs certificado do branch_ref ----------
@pytest.mark.asyncio
async def test_emitente_cnpj_divergente_do_certificado_e_409_e_nao_queima_numero(db_session):
"""Certificado do branch_ref carrega o CNPJ A; o payload declara
`emitente.cnpj`=B (outro CNPJ válido, 14 dígitos) -- a auto teria
barrado isso estruturalmente (emitente vem de `Branch.cnpj`, o MESMO
vínculo do certificado); este serviço, com `emitente` free-form no
payload, precisa da checagem explícita ou assina B com a chave de A."""
product, key = await _product_and_key(db_session)
payload, raw = _payload_from_golden("caso_padrao_intra")
cnpj_divergente = "99887766000155"
assert cnpj_divergente != _CNPJ_EMITENTE
payload["emitente"] = {**payload["emitente"], "cnpj": cnpj_divergente}
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await _setup_certificate_and_series(
db_session, client, key, tenant_ref="t1", branch_ref="b1",
serie=raw["serie"], next_number=raw["numero"], cnpj=_CNPJ_EMITENTE,
)
mismatch_response = await client.post(
"/v1/emissoes", json=payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
assert mismatch_response.status_code == 409, mismatch_response.text
assert mismatch_response.json()["detail"]["code"] == "emitente_certificate_cnpj_mismatch"
# A mensagem NÃO deve vazar o CNPJ real do certificado.
assert _CNPJ_EMITENTE not in mismatch_response.text
result = await db_session.execute(
select(FiscalSeries).where(
FiscalSeries.product_id == product.id, FiscalSeries.tenant_ref == "t1"
)
)
series = result.scalar_one()
assert series.next_number == raw["numero"], (
"o número NÃO pode ter sido queimado por um emitente.cnpj divergente do certificado"
)
# O número que seria queimado acima segue disponível -- uma emissão
# com o CNPJ CORRETO recebe exatamente esse número.
matching_payload, _ = _payload_from_golden("caso_padrao_intra")
success_response = await client.post(
"/v1/emissoes", json=matching_payload,
headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"},
)
assert success_response.status_code == 201, success_response.text
assert success_response.json()["numero"] == raw["numero"]
# --- prova do outbox --------------------------------------------------------