f902ccbecb
Servico que roda dentro da rede da Comercial Scardua e faz a ponte entre a API publica da FLS (na VPS) e o Oracle privado (10.16.x), inalcancavel pela internet. Fluxo: Holmes -> API na VPS (trata e cifra) -> ESTE conector (abre e valida) -> Oracle. Recebimento: - app/v1/compras.py: POST /v1/compras/dados - app/services/cripto.py: abre o envelope AES-256-GCM + RSA-OAEP-SHA256 com a chave privada. O GCM autentica: corpo adulterado levanta InvalidTag em vez de devolver lixo - app/schemas.py: modulo folha (so pydantic) com a config e o contrato PayloadCompras, que espelha o da API. Fora de sincronia devolve 422 de proposito, pra falhar explicito em vez de gravar dado torto Seguranca: - app/seguranca.py: header X-Token com compare_digest (nao vaza por tempo de resposta) + allowlist de IP da VPS - .gitignore barra configs.json.*, *.bak-*, *.pem e *.key Config: - app/config.py e so o carregamento do configs.json - o bloco "vps" guarda chave privada, token e ips permitidos Estado: ainda NAO persiste. Recebe, valida e descarta (persistido: false). O INSERT no Oracle e a proxima fase, e tem que ser MERGE por id_processo porque o Holmes reentrega webhook. Docs em docs/ — arquitetura, a decisao do conector (opcao A, com as alternativas descartadas), deploy e problemas conhecidos. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
"""
|
|
Quem pode falar com o conector.
|
|
|
|
Este servico escreve no banco de producao do cliente — e alvo de alto valor.
|
|
As duas travas aqui sao o minimo: segredo compartilhado e origem conhecida.
|
|
Nao e seguranca por obscuridade (URL secreta nao conta).
|
|
"""
|
|
|
|
import logging
|
|
import secrets
|
|
|
|
from app.config import settings
|
|
from fastapi import Header, HTTPException, Request, status
|
|
|
|
|
|
def _origem(request: Request) -> str:
|
|
"""
|
|
IP de origem.
|
|
|
|
Se um dia entrar proxy na frente, o IP do socket vira o do proxy — ai o
|
|
uvicorn precisa de --proxy-headers e isto passa a ler X-Forwarded-For.
|
|
"""
|
|
return request.client.host if request.client else ""
|
|
|
|
|
|
async def autorizar(request: Request, x_token: str = Header(default="")) -> None:
|
|
"""Dependencia de rota: barra quem nao for a VPS."""
|
|
permitidos = settings.vps.ips_permitidos
|
|
if permitidos:
|
|
ip = _origem(request)
|
|
if ip not in permitidos:
|
|
logging.warning(f"Origem recusada: {ip}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="origem nao autorizada"
|
|
)
|
|
|
|
# compare_digest pra nao vazar o segredo por tempo de resposta.
|
|
if not secrets.compare_digest(x_token, settings.vps.token):
|
|
logging.warning(f"Token invalido vindo de {_origem(request)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="token invalido"
|
|
)
|