From b1f93c2d6853a000cc85e0517b0d834c339105b4 Mon Sep 17 00:00:00 2001 From: Gustavo Date: Wed, 23 Sep 2026 22:36:40 -0300 Subject: [PATCH] feat: read-only internal endpoint for GTI Fiscal to mirror clients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GET /internal/fiscal/clients returns every CLIENTE (active and inactive) as {external_client_id, code, display_name, active} — CLIENTE.ID is the identity (CODIGO is editable, same reasoning as FUNCIONARIO_PAINEL.ID in SSO). Authenticated by a dedicated bearer token (FISCAL_INTERNAL_API_TOKEN, never reused from SESSION_SECRET/SSO key/admin password), rate-limited per IP, returns 503 rather than opening anonymously when the token isn't configured. GTI Gerenciador stays the single source of truth: the Fiscal Hub never receives MySQL credentials. Co-Authored-By: Claude Sonnet 5 --- .env.example | 7 +++ app/db/clientes.py | 22 ++++++++++ app/internal_fiscal.py | 66 ++++++++++++++++++++++++++++ app/main.py | 2 + tests/test_internal_fiscal.py | 81 +++++++++++++++++++++++++++++++++++ 5 files changed, 178 insertions(+) create mode 100644 app/internal_fiscal.py create mode 100644 tests/test_internal_fiscal.py diff --git a/.env.example b/.env.example index f7d1df8..2428135 100644 --- a/.env.example +++ b/.env.example @@ -19,6 +19,13 @@ DB_NAME=gtisoftw_clientes_dev DB_USER=root DB_PASSWORD=devpassword123 +# Endpoint interno read-only GET /internal/fiscal/clients (app/internal_fiscal.py): +# o GTI Fiscal espelha o cadastro de clientes daqui em vez de receber credencial +# MySQL. Token dedicado, compartilhado só com o FISCAL_INTERNAL_API_TOKEN do +# Fiscal — nunca o SESSION_SECRET, a chave privada do SSO ou senha de admin. +# Vazio = endpoint desligado (503). +FISCAL_INTERNAL_API_TOKEN= + # Opcional: integração de LEITURA com a Data Table "pagamentos_processados" # do n8n, para mostrar os envios automáticos de chave na tela /log. Gere a # chave em Configurações > API dentro do n8n (devn8n.gtigrupo.com.br). diff --git a/app/db/clientes.py b/app/db/clientes.py index 265a220..039c7a3 100644 --- a/app/db/clientes.py +++ b/app/db/clientes.py @@ -87,6 +87,28 @@ def listar_clientes( conn.close() +def listar_clientes_para_fiscal() -> list[dict]: + """Para o endpoint interno GET /internal/fiscal/clients (app/internal_fiscal.py): + o Fiscal espelha TODOS os clientes (ativos e inativos — inativo continua + aparecendo em histórico do lado de lá), nunca só uma página. Payload + mínimo: ID é a PK técnica estável (CODIGO é editável, não serve de + identidade — mesma razão de FUNCIONARIO_PAINEL.ID no SSO). Sem CNPJ/ + endereço/contato: o Fiscal não precisa desses campos hoje.""" + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "SELECT ID, CODIGO, COALESCE(NULLIF(FANTASIA, ''), RAZAO) AS NOME, ATIVO " + "FROM CLIENTE ORDER BY ID" + ) + return [ + {"external_client_id": row["ID"], "code": row["CODIGO"], "display_name": row["NOME"], "active": bool(row["ATIVO"])} + for row in cur.fetchall() + ] + finally: + conn.close() + + def proximo_codigo_sugerido() -> str: """ Sugere o próximo CODIGO sequencial (maior CODIGO numérico + 1, com zeros diff --git a/app/internal_fiscal.py b/app/internal_fiscal.py new file mode 100644 index 0000000..74e96d8 --- /dev/null +++ b/app/internal_fiscal.py @@ -0,0 +1,66 @@ +""" +Endpoint interno server-to-server, read-only, para o GTI Fiscal espelhar o +cadastro de clientes. GTI Gerenciador continua sendo a fonte única da +verdade: o Fiscal NUNCA recebe credencial MySQL direta (nem lê a tabela +CLIENTE por fora) — só chama esta rota, autenticada por um bearer token +dedicado (FISCAL_INTERNAL_API_TOKEN), nunca reaproveitado de SESSION_SECRET, +da chave privada do SSO, de senha de admin ou de token de Agent. + +Rate limit em memória por IP de origem — mesmo espírito de app/rate_limit.py +(container único, sem load balancer; ver ali o aviso sobre proxy reverso). +""" +from __future__ import annotations + +import hmac +import os +import time + +from fastapi import APIRouter, Header, HTTPException, Request + +from .db.clientes import listar_clientes_para_fiscal + +router = APIRouter() + +MAX_TENTATIVAS = 10 +JANELA_SEGUNDOS = 5 * 60 +_falhas: dict[str, list[float]] = {} + + +def _token_configurado() -> bool: + return bool(os.environ.get("FISCAL_INTERNAL_API_TOKEN")) + + +def _token_valido(fornecido: str | None) -> bool: + esperado = os.environ.get("FISCAL_INTERNAL_API_TOKEN", "") + if not esperado or not fornecido: + return False + return hmac.compare_digest(fornecido, esperado) + + +def _bloqueado(origem: str, agora: float | None = None) -> bool: + momento = agora if agora is not None else time.time() + limite = momento - JANELA_SEGUNDOS + return len([t for t in _falhas.get(origem, []) if t > limite]) >= MAX_TENTATIVAS + + +def _registrar_falha(origem: str, agora: float | None = None) -> None: + momento = agora if agora is not None else time.time() + limite = momento - JANELA_SEGUNDOS + recentes = [t for t in _falhas.get(origem, []) if t > limite] + recentes.append(momento) + _falhas[origem] = recentes + + +@router.get("/internal/fiscal/clients") +def listar_clientes_fiscal(request: Request, authorization: str | None = Header(default=None)): + origem = request.client.host if request.client else "unknown" + if _bloqueado(origem): + raise HTTPException(status_code=429, detail="muitas tentativas") + if not _token_configurado(): + # Sem token configurado = integração desligada, nunca "aberto por padrão". + raise HTTPException(status_code=503, detail="FISCAL_INTERNAL_API_TOKEN não configurado") + fornecido = authorization.removeprefix("Bearer ").strip() if authorization and authorization.startswith("Bearer ") else None + if not _token_valido(fornecido): + _registrar_falha(origem) + raise HTTPException(status_code=401, detail="token inválido ou ausente") + return {"clients": listar_clientes_para_fiscal()} diff --git a/app/main.py b/app/main.py index be28942..bdd67e6 100644 --- a/app/main.py +++ b/app/main.py @@ -12,6 +12,7 @@ from . import session from .asset_versao import ASSET_VERSAO from .auth import NaoAutenticado, UsuarioLogado, exigir_login from .csrf import CSRF_COOKIE_NAME, gerar_token +from . import internal_fiscal from .routers import auth, chave_manual, clientes, contatos, funcionarios, log, minha_conta, sso app = FastAPI(title="Painel de Clientes", docs_url=None, redoc_url=None, openapi_url=None) @@ -23,6 +24,7 @@ app.include_router(chave_manual.router) app.include_router(funcionarios.router) app.include_router(minha_conta.router) app.include_router(log.router) +app.include_router(internal_fiscal.router) app.include_router(sso.router) diff --git a/tests/test_internal_fiscal.py b/tests/test_internal_fiscal.py new file mode 100644 index 0000000..b9c2be0 --- /dev/null +++ b/tests/test_internal_fiscal.py @@ -0,0 +1,81 @@ +"""GET /internal/fiscal/clients — endpoint read-only server-to-server para o +GTI Fiscal espelhar o cadastro de clientes. Usa CODIGO na faixa de teste +"999950"-"999959" (ver tests/test_db_clientes.py) e limpa o que criou.""" +import pytest +from fastapi.testclient import TestClient + +from app.db import clientes as db_clientes +from app.db.connection import get_connection +from app.main import app + +TOKEN = "teste-token-fiscal-interno-suficientemente-longo" + +BASE = { + "CNPJ": "11222333000181", "RAZAO": "TESTE INTERNO RAZAO", "IE": "ISENTO", + "ENDERECO": "RUA TESTE, 1", "CIDADE": "FEIRA DE SANTANA", "BAIRRO": "CENTRO", "UF": "BA", + "CONTATO": "FULANO", "TELEFONE": "75999090000", "CELULAR": "75999090000", "CEP": "44000000", + "IMPLANTACAO": "0.00", "MENSALIDADE": "100.00", "JUROS_MORA": "0.00", "JUROS_MULTA": "0.00", + "DATA_CONTRATO": "2026-01-01", "EMAIL": "teste@example.com", "OBS": "teste automatizado", + "VENCIMENTO_CONTRATO": "2027-01-01", "DIA_VENCIMENTO": "10", +} + + +def _limpar(codigos): + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute(f"DELETE FROM CLIENTE WHERE CODIGO IN ({','.join(['%s'] * len(codigos))})", codigos) + finally: + conn.close() + + +@pytest.fixture +def clientes_teste(): + ativo_id = db_clientes.criar_cliente({**BASE, "CODIGO": "999950", "FANTASIA": "Ativo Fiscal Teste", "ATIVO": "1"}) + inativo_id = db_clientes.criar_cliente({**BASE, "CODIGO": "999951", "FANTASIA": "", "RAZAO": "Inativo Fiscal Teste Razao", "ATIVO": "0"}) + yield ativo_id, inativo_id + _limpar(["999950", "999951"]) + + +@pytest.fixture +def client(monkeypatch): + monkeypatch.setenv("FISCAL_INTERNAL_API_TOKEN", TOKEN) + from app import internal_fiscal + internal_fiscal._falhas.clear() + return TestClient(app) + + +def test_returns_active_and_inactive_clients_with_minimal_fields(client, clientes_teste): + ativo_id, inativo_id = clientes_teste + response = client.get("/internal/fiscal/clients", headers={"Authorization": f"Bearer {TOKEN}"}) + assert response.status_code == 200 + by_id = {row["external_client_id"]: row for row in response.json()["clients"]} + assert by_id[ativo_id] == {"external_client_id": ativo_id, "code": "999950", "display_name": "Ativo Fiscal Teste", "active": True} + assert by_id[inativo_id] == {"external_client_id": inativo_id, "code": "999951", "display_name": "Inativo Fiscal Teste Razao", "active": False} + # never leaks fields the Fiscal doesn't need + assert set(by_id[ativo_id]) == {"external_client_id", "code", "display_name", "active"} + + +def test_missing_token_rejected(client, clientes_teste): + assert client.get("/internal/fiscal/clients").status_code == 401 + + +def test_wrong_token_rejected(client, clientes_teste): + response = client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer wrong-token"}) + assert response.status_code == 401 + + +def test_endpoint_disabled_when_token_not_configured(monkeypatch): + monkeypatch.delenv("FISCAL_INTERNAL_API_TOKEN", raising=False) + from app import internal_fiscal + internal_fiscal._falhas.clear() + client = TestClient(app) + response = client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer anything"}) + assert response.status_code == 503 + + +def test_rate_limited_after_repeated_invalid_attempts(client): + from app import internal_fiscal + for _ in range(internal_fiscal.MAX_TENTATIVAS): + assert client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer wrong"}).status_code == 401 + assert client.get("/internal/fiscal/clients", headers={"Authorization": "Bearer wrong"}).status_code == 429