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>
57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
import jwt
|
|
from app.config import settings
|
|
from app.infra.database import db_instance
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
|
|
ALGORITHM = "HS256"
|
|
TOKEN_EXPIRE_MINUTES = 5
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
|
|
|
|
|
|
def criar_token(data: dict) -> str:
|
|
payload = data.copy()
|
|
payload["exp"] = datetime.now(timezone.utc) + timedelta(
|
|
minutes=TOKEN_EXPIRE_MINUTES
|
|
)
|
|
return jwt.encode(payload, settings.api.jwt_secret, algorithm=ALGORITHM)
|
|
|
|
|
|
def verificar_credenciais(client_id: str, client_secret: str) -> bool:
|
|
db_instance.create_pool()
|
|
assert db_instance.pool is not None
|
|
conn = db_instance.pool.acquire()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""
|
|
SELECT 1
|
|
FROM orvel_ti.CAD_USUARIO
|
|
WHERE PERFIL_TI = 'operador_ti'
|
|
AND ativo = 'S'
|
|
AND API = 'S'
|
|
AND login = :login
|
|
AND senha = :senha
|
|
""",
|
|
{"login": client_id, "senha": client_secret},
|
|
)
|
|
return cursor.fetchone() is not None
|
|
finally:
|
|
db_instance.pool.release(conn)
|
|
|
|
|
|
async def token_valido(token: str = Depends(oauth2_scheme)) -> dict:
|
|
try:
|
|
payload = jwt.decode(token, settings.api.jwt_secret, algorithms=[ALGORITHM])
|
|
return payload
|
|
except jwt.ExpiredSignatureError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expirado"
|
|
)
|
|
except jwt.InvalidTokenError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="Token inválido"
|
|
)
|