diff --git a/gerenciador/.env.example b/gerenciador/.env.example new file mode 100644 index 0000000..2428135 --- /dev/null +++ b/gerenciador/.env.example @@ -0,0 +1,34 @@ +# Copie para .env e preencha antes de rodar (nunca commitar o .env real). + +ADMIN_USER=admin +ADMIN_PASSWORD=SUBSTITUA_POR_UMA_SENHA_FORTE + +# Chave usada para assinar o cookie de sessão de login (app/session.py). +# Sem isso definido, uma chave aleatória é gerada em memória a cada start +# do processo — funciona, mas todo restart derruba as sessões ativas +# (força login de novo). Em produção, defina um valor fixo e comprido. +SESSION_SECRET=SUBSTITUA_POR_UM_VALOR_ALEATORIO_COMPRIDO + +# Banco de dev (mesmo container do projeto n8n / contatos-chave-web): +# DB_HOST=host.docker.internal (padrao — para rodar via docker-compose.yml, +# que executa este app dentro de um container) +# DB_HOST=127.0.0.1 (se este app rodar direto na maquina, fora do Docker) +DB_HOST=host.docker.internal +DB_PORT=3307 +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). +# Se deixar em branco, a tela /log só mostra o log de ações do painel. +N8N_API_URL= +N8N_API_KEY= diff --git a/gerenciador/.gitignore b/gerenciador/.gitignore new file mode 100644 index 0000000..bc4f72a --- /dev/null +++ b/gerenciador/.gitignore @@ -0,0 +1,4 @@ +__pycache__/ +*.pyc +.venv/ +.env diff --git a/gerenciador/Dockerfile b/gerenciador/Dockerfile new file mode 100644 index 0000000..d3d1c4f --- /dev/null +++ b/gerenciador/Dockerfile @@ -0,0 +1,14 @@ +FROM python:3.12-slim +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY app ./app +# Não roda como root dentro do container — reduz o estrago de uma +# eventual RCE (mesmo sem vulnerabilidade conhecida hoje, é uma +# mitigação de baixo custo e sem efeito colateral aqui, já que a app não +# precisa gravar nada no filesystem além de escutar a porta 8000, que não +# é privilegiada). +RUN useradd --no-create-home --shell /usr/sbin/nologin painel +USER painel +EXPOSE 8000 +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/gerenciador/README.md b/gerenciador/README.md new file mode 100644 index 0000000..3386ad8 --- /dev/null +++ b/gerenciador/README.md @@ -0,0 +1,77 @@ +# painel-clientes-web + +Painel administrativo (FastAPI + Jinja2) para gestão de `CLIENTE`, `CLIENTE_CONTATO` e geração manual de chave de licenciamento no MySQL do sistema legado. + +Proteções: login por sessão em toda rota (cookie assinado — ver seção "Login e permissões" abaixo), CSRF via cookie de duplo envio (`app/csrf.py` — todo formulário que muda estado embute o mesmo token do cookie `csrf_token`, definido pelo middleware em `app/main.py`) e a lista de clientes é paginada (100 por página), com busca e ordenação por código/razão/fantasia/CNPJ. + +`CODIGO` de cliente não aceita duplicado (bloqueio duro, sem bypass) e o formulário de novo cliente já sugere o próximo código sequencial (editável). Só `CODIGO`, `CNPJ`, `RAZAO` e `FANTASIA` são obrigatórios — os campos de contrato (implantação, mensalidade, juros, datas) podem ficar em branco e recebem default (0,00 / data de hoje). + +`CLIENTE` tem uma coluna `ATIVO` (criada no banco de dev — **precisa ser criada manualmente em produção também**, `ALTER TABLE` completo no início de `app/db/clientes.py`) que controla o filtro de status da listagem (Ativos/Inativos/Todos, padrão Ativos). Sem essa coluna, a listagem quebra com erro 500. O `DEFAULT 1` do `ALTER TABLE` é intencional: todo cliente cadastrado antes da migração nasce como Ativo. + +A rota `/log` mostra o histórico de ações do painel (tabela `LOG_PAINEL`, criada no banco de dev — **precisa ser criada manualmente em produção também**, DDL no início de `app/db/log.py`) e, se `N8N_API_URL`/`N8N_API_KEY` estiverem configuradas (`.env.example`), os últimos envios automáticos de chave via WhatsApp (lidos da Data Table `pagamentos_processados` do n8n, somente leitura). + +## Login e permissões + +O painel usa login por sessão (cookie assinado, `app/session.py`) — não é +mais HTTP Basic Auth. Defina `SESSION_SECRET` no `.env` (`.env.example` +tem o exemplo); sem isso, uma chave é gerada em memória a cada start do +processo, o que derruba todas as sessões ativas a cada restart (funciona, +só incomoda). + +Cada funcionário (`FUNCIONARIO_PAINEL`, criada no banco de dev — **precisa +ser criada manualmente em produção também**, DDL no início de +`app/db/funcionarios.py`) tem um perfil: **admin** (gerencia outros +funcionários em `/funcionarios`: cadastra, edita, redefine senha, +ativa/desativa, promove/rebaixa perfil) ou **padrão** (sem acesso a +`/funcionarios`). Se `FUNCIONARIO_PAINEL` já existir de uma instalação +anterior sem a coluna `PERFIL`, rode o `ALTER TABLE FUNCIONARIO_PAINEL ADD +COLUMN PERFIL ...` (DDL completo em `app/db/funcionarios.py`) — **precisa +ser rodado manualmente em produção também**; sem essa coluna, todo login +falha com erro 500 (a query de autenticação seleciona `PERFIL`, que não +existe na tabela antiga). O `DEFAULT 'ADMIN'` desse `ALTER TABLE` é +intencional: todo funcionário cadastrado antes da migração vira admin +automaticamente. O sistema nunca deixa ficar sem nenhum admin ativo — +rebaixar ou desativar o último admin é bloqueado. + +Qualquer usuário logado troca a própria senha em **Minha conta** +(`/minha-conta`), pedindo a senha atual antes de trocar. + +`ADMIN_USER`/`ADMIN_PASSWORD` (`.env.example`) continuam funcionando como +conta de bootstrap, sempre com perfil admin — é com ela que você entra a +primeira vez para cadastrar o primeiro funcionário, e serve de plano B se +a tabela ficar vazia. Login checa primeiro o admin do `.env`, depois +`FUNCIONARIO_PAINEL` (só contas com `ATIVO=1`). + +## Rodando localmente (sem Docker) + +1. Instale as dependências de desenvolvimento: + ``` + pip install -r requirements-dev.txt + ``` +2. Defina as variáveis de ambiente necessárias (`ADMIN_USER`, `ADMIN_PASSWORD`, `DB_HOST`, `DB_PORT`, `DB_NAME`, `DB_USER`, `DB_PASSWORD`, `SESSION_SECRET` — veja `.env.example`) diretamente no shell antes de subir a aplicação. `SESSION_SECRET` é opcional em dev — sem ela, uma chave aleatória é gerada em memória a cada start (sessões caem a cada restart); defina um valor fixo em produção. **Importante:** apesar de `python-dotenv` estar nas dependências, nada no código chama `load_dotenv()` hoje — um arquivo `.env` sozinho NÃO é lido automaticamente fora do Docker. Exporte as variáveis manualmente (ex.: `set -a; source .env; set +a` no bash, ou `$env:...` no PowerShell) ou defina-as por outro meio. +3. Rode a aplicação: + ``` + uvicorn app.main:app --reload + ``` +4. Rodando direto na máquina (fora de container), use `DB_HOST=127.0.0.1` (porta `3307`, a mesma do container de dev `licenciamento-mysql-dev`). + +## Rodando os testes + +``` +pytest -v +``` + +Os testes usam defaults definidos em `tests/conftest.py` (equivalentes ao `.env.example`) e incluem testes de integração que precisam do container MySQL de dev (`licenciamento-mysql-dev`, `127.0.0.1:3307`) rodando. + +## Rodando via Docker + +1. Copie o arquivo de exemplo de variáveis de ambiente e edite `ADMIN_PASSWORD` (e demais valores, se necessário): + ``` + cp .env.example .env + ``` +2. Suba o container: + ``` + docker compose up -d --build + ``` + O `docker-compose.yml` já usa `env_file: .env` e mapeia `host.docker.internal` para o host (via `extra_hosts`), então o valor padrão `DB_HOST=host.docker.internal` do `.env.example` funciona também em hosts Docker Linux, não só no Docker Desktop (Windows/Mac). +3. A aplicação fica disponível em `http://localhost:8010`. diff --git a/gerenciador/app/__init__.py b/gerenciador/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/gerenciador/app/asset_versao.py b/gerenciador/app/asset_versao.py new file mode 100644 index 0000000..6172853 --- /dev/null +++ b/gerenciador/app/asset_versao.py @@ -0,0 +1,29 @@ +""" +Versão dos arquivos estáticos (CSS/JS) usada como cache-busting nos links +de app/templates/base.html (?v=...). Sem isso, um navegador que já tinha +carregado /static/estilo.css antes de um deploy pode continuar servindo a +cópia antiga em cache — a URL não muda, então não há sinal pro navegador +buscar de novo — mesmo numa aba anônima aberta antes do deploy terminar. + +Calculada uma única vez, na inicialização do processo, a partir do mtime +mais recente entre os arquivos em app/static/. Cada novo deploy recria o +container com os arquivos atualizados (mtime novo), então o valor muda +sozinho a cada implantação, sem precisar editar nada aqui. +""" +import os + +_DIR_STATIC = os.path.join(os.path.dirname(__file__), "static") + + +def _calcular_versao() -> str: + arquivos = [ + os.path.join(_DIR_STATIC, nome) + for nome in os.listdir(_DIR_STATIC) + if os.path.isfile(os.path.join(_DIR_STATIC, nome)) + ] + if not arquivos: + return "0" + return str(int(max(os.path.getmtime(caminho) for caminho in arquivos))) + + +ASSET_VERSAO = _calcular_versao() diff --git a/gerenciador/app/auth.py b/gerenciador/app/auth.py new file mode 100644 index 0000000..4126bec --- /dev/null +++ b/gerenciador/app/auth.py @@ -0,0 +1,97 @@ +""" +Login por sessão. Duas fontes de credencial, checadas nessa ordem em +autenticar_credenciais() (chamada tanto por POST /login quanto, +indiretamente, por quem valida o cookie de sessão): + +1. ADMIN_USER/ADMIN_PASSWORD (variáveis de ambiente) — conta única de + bootstrap, útil para o primeiro acesso (cadastrar o primeiro + funcionário em /funcionarios) e como plano B se a tabela de + funcionários ficar vazia/inacessível. Continua exigindo as duas + variáveis preenchidas (nunca autentica credencial vazia). Sempre + perfil ADMIN, não depende do banco. +2. FUNCIONARIO_PAINEL (app/db/funcionarios.py) — conta individual por + funcionário, com senha em hash (app/senha.py). Só autentica se ATIVO=1. + Perfil vem da coluna PERFIL ("ADMIN" ou "PADRAO"). + +A credencial em si só é usada em POST /login (app/routers/auth.py), que +troca por um cookie de sessão assinado (app/session.py) — as rotas do +painel não recebem usuário/senha a cada requisição, só o cookie. +""" +import os +import secrets +from dataclasses import dataclass + +from fastapi import Depends, HTTPException, Request, status + +from . import session +from .db import funcionarios as db_funcionarios +from .senha import verificar_senha + + +@dataclass +class UsuarioLogado: + login: str + nome: str + perfil: str # "ADMIN" ou "PADRAO" + + def __str__(self) -> str: + return self.nome + + @property + def eh_admin(self) -> bool: + return self.perfil == "ADMIN" + + +class NaoAutenticado(Exception): + def __init__(self, caminho_original: str = "/"): + self.caminho_original = caminho_original + + +def autenticar_credenciais(usuario: str, senha: str) -> UsuarioLogado | None: + usuario_admin = os.environ.get("ADMIN_USER", "") + senha_admin = os.environ.get("ADMIN_PASSWORD", "") + + if usuario_admin and senha_admin: + usuario_ok = secrets.compare_digest(usuario, usuario_admin) + senha_ok = secrets.compare_digest(senha, senha_admin) + if usuario_ok and senha_ok: + return UsuarioLogado(login=usuario_admin, nome=usuario_admin, perfil="ADMIN") + + funcionario = db_funcionarios.obter_por_login(usuario) + if funcionario and funcionario["ATIVO"] and verificar_senha(senha, funcionario["SENHA_HASH"]): + return UsuarioLogado(login=funcionario["LOGIN"], nome=funcionario["NOME"], perfil=funcionario["PERFIL"]) + + return None + + +def exigir_login(request: Request) -> UsuarioLogado: + """ + Decodifica o cookie de sessão e, para logins que existem em + FUNCIONARIO_PAINEL, reconfirma ATIVO/PERFIL contra o banco a cada + requisição — o cookie assinado tem validade de até 12h e não é + revogado, então sem essa checagem desativar ou rebaixar um funcionário + não teria efeito imediato (continuaria com acesso até o cookie + expirar). A conta de bootstrap (ADMIN_USER do .env) não tem linha em + FUNCIONARIO_PAINEL e continua sempre admin via variável de ambiente, + sem consulta ao banco. + """ + token = request.cookies.get(session.SESSION_COOKIE_NAME) + dados = session.ler_token(token) + if dados is None: + raise NaoAutenticado(caminho_original=request.url.path) + + usuario_admin = os.environ.get("ADMIN_USER", "") + if usuario_admin and dados["login"] == usuario_admin: + return UsuarioLogado(login=dados["login"], nome=dados["nome"], perfil="ADMIN") + + funcionario = db_funcionarios.obter_por_login(dados["login"]) + if funcionario is None or not funcionario["ATIVO"]: + raise NaoAutenticado(caminho_original=request.url.path) + + return UsuarioLogado(login=funcionario["LOGIN"], nome=funcionario["NOME"], perfil=funcionario["PERFIL"]) + + +def exigir_admin(usuario: UsuarioLogado = Depends(exigir_login)) -> UsuarioLogado: + if not usuario.eh_admin: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Só administradores acessam esta página.") + return usuario diff --git a/gerenciador/app/cnpj_lookup.py b/gerenciador/app/cnpj_lookup.py new file mode 100644 index 0000000..b28920e --- /dev/null +++ b/gerenciador/app/cnpj_lookup.py @@ -0,0 +1,57 @@ +""" +Consulta de CNPJ na BrasilAPI (https://brasilapi.com.br/api/cnpj/v1) para +preencher automaticamente o formulário de cliente. Fonte pública/gratuita, +sem autenticação, mantida pela comunidade a partir de dados abertos da +Receita Federal — não é a API oficial da Receita (que não tem consulta +pública direta por CNPJ). + +Só faz sentido para CNPJ (empresa) — CPF é dado pessoal e não tem consulta +pública, então o chamador nunca deve passar um CPF aqui (ver +app/routers/clientes.py, que checa o tipo via normalizar_documento antes). + +Degrada de forma segura: qualquer falha (CNPJ não encontrado, rede fora do +ar, formato que a BrasilAPI ainda não reconhece — o novo CNPJ alfanumérico, +vigente desde 31/07/2026, pode não estar coberto por todo provedor ainda) +retorna None. O formulário simplesmente não preenche nada sozinho nesse caso. +""" +import httpx + +_BASE_URL = "https://brasilapi.com.br/api/cnpj/v1" + + +def consultar_cnpj(cnpj_normalizado: str) -> dict | None: + """ + cnpj_normalizado: 14 caracteres (12 alfanuméricos + 2 dígitos), já sem + máscara — ver app/normalize.py. Retorna um dict só com os campos que + existem em CLIENTE, prontos para preencher o formulário, ou None se não + encontrou/a consulta falhou. + """ + if len(cnpj_normalizado) != 14: + return None + + try: + resposta = httpx.get(f"{_BASE_URL}/{cnpj_normalizado}", timeout=8.0) + if resposta.status_code != 200: + return None + dados = resposta.json() + except (httpx.HTTPError, ValueError): + return None + + if not isinstance(dados, dict): + return None + + logradouro = dados.get("logradouro") or "" + numero = dados.get("numero") or "" + endereco = f"{logradouro}, {numero}" if numero else logradouro + + return { + "RAZAO": dados.get("razao_social") or "", + "FANTASIA": dados.get("nome_fantasia") or dados.get("razao_social") or "", + "ENDERECO": endereco[:45], # CLIENTE.ENDERECO é varchar(45) + "CIDADE": dados.get("municipio") or "", + "BAIRRO": dados.get("bairro") or "", + "UF": dados.get("uf") or "", + "CEP": (dados.get("cep") or "")[:10], + "TELEFONE": dados.get("ddd_telefone_1") or "", + "EMAIL": dados.get("email") or "", + } diff --git a/gerenciador/app/csrf.py b/gerenciador/app/csrf.py new file mode 100644 index 0000000..f57940f --- /dev/null +++ b/gerenciador/app/csrf.py @@ -0,0 +1,44 @@ +""" +Proteção CSRF via "double-submit cookie": o middleware registrado em +app/main.py garante que todo navegador que acessa o painel tenha um cookie +"csrf_token" (gerado uma vez, aleatório, HttpOnly). Todo formulário que +muda estado (POST) embute esse mesmo valor como campo oculto — os +templates leem de request.state.csrf_token, que o middleware preenche a +cada requisição. Toda rota POST que muda estado usa Depends(exigir_csrf), +que confere se o valor enviado no formulário bate com o valor do cookie. + +Por quê isso barra CSRF: um site hostil consegue forçar o navegador da +vítima a mandar um POST para o painel (o cookie vai junto automaticamente), +mas não consegue LER o valor do cookie para replicá-lo como campo oculto do +formulário forjado — só JavaScript rodando na própria origem do painel +consegue (e não roda em outra origem, por causa do same-origin policy). +""" +import secrets + +from fastapi import HTTPException, Request, status + +CSRF_COOKIE_NAME = "csrf_token" + + +def gerar_token() -> str: + return secrets.token_urlsafe(32) + + +def token_do_contexto(request: Request) -> str: + """ + Lê o token que o middleware colocou em request.state.csrf_token. + Retorna "" se o middleware não estiver montado (por exemplo, testes que + sobem só um router isolado, sem app/main.py) — nesse caso o formulário + simplesmente não terá csrf_token preenchido, e exigir_csrf abaixo vai + rejeitar o POST a menos que o teste monte o cookie/campo manualmente. + """ + return getattr(request.state, "csrf_token", "") + + +async def exigir_csrf(request: Request) -> None: + """Dependency para toda rota POST que muda estado.""" + form = await request.form() + token_formulario = str(form.get(CSRF_COOKIE_NAME, "")) + token_cookie = request.cookies.get(CSRF_COOKIE_NAME, "") + if not token_cookie or not token_formulario or not secrets.compare_digest(token_formulario, token_cookie): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Token CSRF ausente ou inválido.") diff --git a/gerenciador/app/db/__init__.py b/gerenciador/app/db/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/gerenciador/app/db/chaves.py b/gerenciador/app/db/chaves.py new file mode 100644 index 0000000..6577304 --- /dev/null +++ b/gerenciador/app/db/chaves.py @@ -0,0 +1,31 @@ +""" +Leitura somente-leitura de Clientes_Chaves (tabela legada — ver +licenciamento-n8n/docs/MAPEAMENTO_BANCO.md seção 3). O painel NUNCA escreve +aqui; só mostra status para conferência. +""" +from .connection import get_connection + +MESES = [f"M{i}" for i in range(1, 13)] + + +def status_chaves(codigo: str) -> dict | None: + conn = get_connection() + try: + with conn.cursor() as cur: + colunas = ", ".join(MESES) + cur.execute( + f"SELECT ID, Cliente, {colunas} FROM Clientes_Chaves WHERE ID = %s", + (codigo,), + ) + linha = cur.fetchone() + finally: + conn.close() + + if linha is None: + return None + + return { + "codigo": linha["ID"], + "cliente": linha["Cliente"], + "meses": {mes: (linha[mes] or None) for mes in MESES}, + } diff --git a/gerenciador/app/db/clientes.py b/gerenciador/app/db/clientes.py new file mode 100644 index 0000000..0470f0b --- /dev/null +++ b/gerenciador/app/db/clientes.py @@ -0,0 +1,267 @@ +""" +CRUD completo de CLIENTE. Única tabela cadastral (ID autoincrement é a PK +técnica; CODIGO é o identificador de negócio usado pelo resto do sistema — +ver licenciamento-n8n/docs/MAPEAMENTO_BANCO.md). + +CLIENTE.ATIVO (TINYINT(1) NOT NULL DEFAULT 1) controla se o cliente +aparece no filtro padrão da listagem ("Ativos"). Se CLIENTE já existir de +uma instalação anterior sem essa coluna, rode: + + ALTER TABLE CLIENTE + ADD COLUMN ATIVO TINYINT(1) NOT NULL DEFAULT 1 + AFTER DIA_VENCIMENTO; + +DEFAULT 1 é intencional: todo cliente cadastrado antes da migração nasce +como Ativo — decisão do projeto (quem administra desativa manualmente os +que sabe que não são mais, ver +docs/superpowers/specs/2026-08-20-status-ativo-clientes-design.md). +Cadastros/edições novos sempre informam ATIVO explicitamente (vindo do +formulário), nunca dependem desse default. +""" +import pymysql + +from .connection import get_connection + +CAMPOS_CLIENTE = [ + "CODIGO", "CNPJ", "RAZAO", "FANTASIA", "IE", "ENDERECO", "CIDADE", + "BAIRRO", "UF", "CONTATO", "TELEFONE", "CELULAR", "CEP", "IMPLANTACAO", + "MENSALIDADE", "JUROS_MORA", "JUROS_MULTA", "DATA_CONTRATO", "EMAIL", + "OBS", "VENCIMENTO_CONTRATO", "DIA_VENCIMENTO", "ATIVO", +] + +_COLUNAS_SELECT = ", ".join(["ID", *CAMPOS_CLIENTE]) + +# Whitelist de colunas aceitas para ordenar a listagem — nunca interpolar +# o valor de "ordenar_por" vindo da URL direto no SQL sem passar por aqui. +ORDENACOES_VALIDAS = { + "codigo": "CODIGO", + "razao": "RAZAO", + "fantasia": "FANTASIA", + "cnpj": "CNPJ", +} +ORDENACAO_PADRAO = "razao" + +# Idem para a direção — nunca interpolar "asc"/"desc" vindo da URL sem +# passar por aqui. +DIRECOES_VALIDAS = {"asc": "ASC", "desc": "DESC"} +DIRECAO_PADRAO = "asc" + + +class ClienteComDependenciasError(Exception): + """Levantada ao tentar excluir um CLIENTE que ainda tem CLIENTE_CONTATO.""" + + +def listar_clientes( + busca: str | None = None, + limite: int = 100, + offset: int = 0, + ordenar_por: str = ORDENACAO_PADRAO, + direcao: str = DIRECAO_PADRAO, + filtro_ativo: int | None = 1, +) -> list[dict]: + coluna = ORDENACOES_VALIDAS.get(ordenar_por, ORDENACOES_VALIDAS[ORDENACAO_PADRAO]) + direcao_sql = DIRECOES_VALIDAS.get(direcao, DIRECOES_VALIDAS[DIRECAO_PADRAO]) + # A busca é restrita à mesma coluna escolhida em "ordenar_por" — selecionar + # "Código" busca só em CODIGO, não em todos os campos de uma vez. + # filtro_ativo: 1 = só ativos, 0 = só inativos, None = sem filtro (todos). + condicoes = [] + params: list = [] + if busca: + condicoes.append(f"{coluna} LIKE %s") + params.append(f"%{busca}%") + if filtro_ativo is not None: + condicoes.append("ATIVO = %s") + params.append(filtro_ativo) + where_sql = f"WHERE {' AND '.join(condicoes)} " if condicoes else "" + + conn = get_connection() + try: + with conn.cursor() as cur: + sql = ( + f"SELECT {_COLUNAS_SELECT} FROM CLIENTE {where_sql}" + f"ORDER BY {coluna} {direcao_sql} LIMIT %s OFFSET %s" + ) + cur.execute(sql, [*params, limite, offset]) + return cur.fetchall() + finally: + 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). CNPJ (só dígitos) entra só para a busca/identificação no seletor do + Fiscal; sem endereço/contato.""" + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "SELECT ID, CODIGO, CNPJ, 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"], "cnpj": "".join(ch for ch in (row["CNPJ"] or "") if ch.isdigit()) or None, "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 + à esquerda até 6 dígitos). É só uma sugestão pré-preenchida no formulário + de novo cliente — o campo continua editável, e codigo_em_uso() ainda + confere unicidade de verdade antes de gravar. + """ + conn = get_connection() + try: + with conn.cursor() as cur: + # CODIGO >= 900000 é a faixa reservada para dados de teste deste + # painel (ver tests/test_db_clientes.py) — excluída aqui para a + # sugestão não disparar lá em cima se sobrar alguma linha de teste. + cur.execute( + "SELECT MAX(CAST(CODIGO AS UNSIGNED)) AS maior FROM CLIENTE " + "WHERE CODIGO REGEXP '^[0-9]+$' AND CAST(CODIGO AS UNSIGNED) < 900000" + ) + maior = cur.fetchone()["maior"] + finally: + conn.close() + proximo = (maior or 0) + 1 + return str(proximo).zfill(6) + + +def contar_clientes(busca: str | None = None, ordenar_por: str = ORDENACAO_PADRAO, filtro_ativo: int | None = 1) -> int: + coluna = ORDENACOES_VALIDAS.get(ordenar_por, ORDENACOES_VALIDAS[ORDENACAO_PADRAO]) + condicoes = [] + params: list = [] + if busca: + condicoes.append(f"{coluna} LIKE %s") + params.append(f"%{busca}%") + if filtro_ativo is not None: + condicoes.append("ATIVO = %s") + params.append(filtro_ativo) + where_sql = f"WHERE {' AND '.join(condicoes)}" if condicoes else "" + + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute(f"SELECT COUNT(*) AS total FROM CLIENTE {where_sql}", params) + return cur.fetchone()["total"] + finally: + conn.close() + + +def obter_cliente(cliente_id: int) -> dict | None: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute(f"SELECT {_COLUNAS_SELECT} FROM CLIENTE WHERE ID = %s", (cliente_id,)) + return cur.fetchone() + finally: + conn.close() + + +def cnpj_em_uso(documento_normalizado: str, ignorar_id: int | None = None) -> bool: + """ + True se já existe outro CLIENTE com esse mesmo CNPJ/CPF (o mesmo campo + CLIENTE.CNPJ aceita os dois — ver app/normalize.py). Compara ignorando + pontuação em vez de igualdade exata porque CLIENTE.CNPJ tem + formatação inconsistente em registros legados (alguns com máscara, + outros sem — ver licenciamento-n8n/docs/ANALISE_CLIENTE_IPREAL.md); + sem normalizar dos dois lados, um cadastro novo (sempre gravado sem + máscara) não bateria com um legado que ainda tem pontos/traços, + deixando passar o duplicado. `documento_normalizado` já chega sem + máscara (normalizar_documento); a normalização abaixo só precisa + tratar o lado do banco. + """ + conn = get_connection() + try: + with conn.cursor() as cur: + sql = ( + "SELECT ID FROM CLIENTE WHERE " + "UPPER(REPLACE(REPLACE(REPLACE(REPLACE(CNPJ, '.', ''), '/', ''), '-', ''), ' ', '')) = %s" + ) + params = [documento_normalizado] + if ignorar_id is not None: + sql += " AND ID != %s" + params.append(ignorar_id) + cur.execute(sql, params) + return cur.fetchone() is not None + finally: + conn.close() + + +def codigo_em_uso(codigo: str, ignorar_id: int | None = None) -> bool: + conn = get_connection() + try: + with conn.cursor() as cur: + sql = "SELECT ID FROM CLIENTE WHERE CODIGO = %s" + params = [codigo] + if ignorar_id is not None: + sql += " AND ID != %s" + params.append(ignorar_id) + cur.execute(sql, params) + return cur.fetchone() is not None + finally: + conn.close() + + +def criar_cliente(dados: dict) -> int: + colunas = ", ".join(CAMPOS_CLIENTE) + marcadores = ", ".join(["%s"] * len(CAMPOS_CLIENTE)) + valores = [dados[campo] for campo in CAMPOS_CLIENTE] + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute(f"INSERT INTO CLIENTE ({colunas}) VALUES ({marcadores})", valores) + return cur.lastrowid + finally: + conn.close() + + +def atualizar_cliente(cliente_id: int, dados: dict) -> None: + atribuicoes = ", ".join(f"{campo} = %s" for campo in CAMPOS_CLIENTE) + valores = [dados[campo] for campo in CAMPOS_CLIENTE] + valores.append(cliente_id) + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute(f"UPDATE CLIENTE SET {atribuicoes} WHERE ID = %s", valores) + finally: + conn.close() + + +def excluir_cliente(cliente_id: int) -> bool: + """True se a linha realmente existia (cur.rowcount > 0) — um ID + inexistente não gera erro nenhum do MySQL, então sem checar rowcount + o chamador não tem como distinguir "excluiu" de "não tinha nada pra + excluir".""" + conn = get_connection() + try: + with conn.cursor() as cur: + try: + cur.execute("DELETE FROM CLIENTE WHERE ID = %s", (cliente_id,)) + except pymysql.err.IntegrityError as exc: + raise ClienteComDependenciasError( + "Este cliente tem contatos cadastrados (CLIENTE_CONTATO) — remova-os antes de excluir." + ) from exc + return cur.rowcount > 0 + finally: + conn.close() + + +def alternar_ativo(cliente_id: int) -> bool: + """True se a linha existia (cur.rowcount > 0) — mesma lógica de + excluir_cliente/redefinir_senha pra distinguir "alternou" de "não + tinha nada pra alternar". UPDATE ... SET ATIVO = 1 - ATIVO inverte o + valor atomicamente, sem precisar de um SELECT antes.""" + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute("UPDATE CLIENTE SET ATIVO = 1 - ATIVO WHERE ID = %s", (cliente_id,)) + return cur.rowcount > 0 + finally: + conn.close() diff --git a/gerenciador/app/db/connection.py b/gerenciador/app/db/connection.py new file mode 100644 index 0000000..7614801 --- /dev/null +++ b/gerenciador/app/db/connection.py @@ -0,0 +1,22 @@ +""" +Única porta de entrada para conexão MySQL do painel. Todo módulo em +app/db/*.py chama get_connection() daqui — nunca abre pymysql.connect() +por conta própria. +""" +import os + +import pymysql +import pymysql.cursors + + +def get_connection() -> pymysql.connections.Connection: + """Abre uma conexão nova por requisição (escala MVP; sem pool).""" + return pymysql.connect( + host=os.environ["DB_HOST"], + port=int(os.environ.get("DB_PORT", "3306")), + user=os.environ["DB_USER"], + password=os.environ["DB_PASSWORD"], + database=os.environ["DB_NAME"], + cursorclass=pymysql.cursors.DictCursor, + autocommit=True, + ) diff --git a/gerenciador/app/db/contatos.py b/gerenciador/app/db/contatos.py new file mode 100644 index 0000000..58902b7 --- /dev/null +++ b/gerenciador/app/db/contatos.py @@ -0,0 +1,83 @@ +""" +CRUD de CLIENTE_CONTATO — relacionado a CLIENTE por CLIENTE_ID (FK real, +ON DELETE RESTRICT ON UPDATE CASCADE). Substitui CLIENTE_CONTATO_CHAVE +(superseded — ver licenciamento-n8n/docs/CANDIDATO_CLIENTE_CONTATO.md). +""" +from .connection import get_connection + + +def listar_contatos_por_cliente(cliente_id: int) -> list[dict]: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "SELECT ID, NOME, TELEFONE, EMAIL FROM CLIENTE_CONTATO " + "WHERE CLIENTE_ID = %s ORDER BY ID", + (cliente_id,), + ) + return list(cur.fetchall()) + finally: + conn.close() + + +def contato_duplicado(cliente_id: int, telefone: str, ignorar_id: int | None = None) -> bool: + conn = get_connection() + try: + with conn.cursor() as cur: + sql = "SELECT ID FROM CLIENTE_CONTATO WHERE CLIENTE_ID = %s AND TELEFONE = %s" + params = [cliente_id, telefone] + if ignorar_id is not None: + sql += " AND ID != %s" + params.append(ignorar_id) + cur.execute(sql, params) + return cur.fetchone() is not None + finally: + conn.close() + + +def inserir_contato(cliente_id: int, nome: str | None, telefone: str, email: str | None = None) -> int: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO CLIENTE_CONTATO (CLIENTE_ID, NOME, TELEFONE, EMAIL) VALUES (%s, %s, %s, %s)", + (cliente_id, nome, telefone, email), + ) + return cur.lastrowid + finally: + conn.close() + + +def atualizar_contato( + contato_id: int, cliente_id: int, nome: str | None, telefone: str, email: str | None = None +) -> bool: + """True se realmente existia um CLIENTE_CONTATO com esse ID nesse + cliente (cur.rowcount > 0) — um ID inexistente, ou de outro cliente, + não gera erro nenhum do MySQL (WHERE simplesmente não bate com + nada), então sem checar rowcount o chamador não tem como saber que + não alterou nada.""" + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "UPDATE CLIENTE_CONTATO SET NOME = %s, TELEFONE = %s, EMAIL = %s " + "WHERE ID = %s AND CLIENTE_ID = %s", + (nome, telefone, email, contato_id, cliente_id), + ) + return cur.rowcount > 0 + finally: + conn.close() + + +def excluir_contato(contato_id: int, cliente_id: int) -> bool: + """True se uma linha foi de fato excluída — ver atualizar_contato.""" + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "DELETE FROM CLIENTE_CONTATO WHERE ID = %s AND CLIENTE_ID = %s", + (contato_id, cliente_id), + ) + return cur.rowcount > 0 + finally: + conn.close() diff --git a/gerenciador/app/db/funcionarios.py b/gerenciador/app/db/funcionarios.py new file mode 100644 index 0000000..4b7d49b --- /dev/null +++ b/gerenciador/app/db/funcionarios.py @@ -0,0 +1,153 @@ +""" +CRUD de FUNCIONARIO_PAINEL — contas individuais de funcionário para login +no painel (substitui/complementa o ADMIN_USER/ADMIN_PASSWORD único). + +A tabela já existe no banco de dev. Em qualquer outro ambiente, crie-a +primeiro com: + + CREATE TABLE FUNCIONARIO_PAINEL ( + ID INT UNSIGNED NOT NULL AUTO_INCREMENT, + NOME VARCHAR(80) NOT NULL, + LOGIN VARCHAR(40) NOT NULL, + SENHA_HASH VARCHAR(255) NOT NULL, + PERFIL ENUM('ADMIN','PADRAO') NOT NULL DEFAULT 'ADMIN', + ATIVO TINYINT(1) NOT NULL DEFAULT 1, + CRIADO_EM DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (ID), + UNIQUE KEY UQ_FUNCIONARIO_PAINEL_LOGIN (LOGIN) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci; + +Se a tabela já existir sem a coluna PERFIL (ambientes criados antes dessa +funcionalidade), rode: + + ALTER TABLE FUNCIONARIO_PAINEL + ADD COLUMN PERFIL ENUM('ADMIN','PADRAO') NOT NULL DEFAULT 'ADMIN' + AFTER SENHA_HASH; + +O DEFAULT 'ADMIN' do ALTER TABLE é intencional: todo funcionário cadastrado +antes dessa migração vira admin automaticamente (decisão do projeto — ver +docs/superpowers/specs/2026-08-13-permissoes-login-sessao-design.md). +Cadastros novos sempre informam PERFIL explicitamente (vindo do +formulário), nunca dependem desse default. + +Nunca excluído de verdade — "excluir" aqui é alternar ATIVO para 0 +(mesma lógica de nunca apagar histórico usada no resto do projeto). +""" +from .connection import get_connection +from ..senha import gerar_hash + + +def listar_funcionarios() -> list[dict]: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "SELECT ID, NOME, LOGIN, PERFIL, ATIVO, CRIADO_EM FROM FUNCIONARIO_PAINEL ORDER BY NOME" + ) + return cur.fetchall() + finally: + conn.close() + + +def obter_funcionario(funcionario_id: int) -> dict | None: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "SELECT ID, NOME, LOGIN, PERFIL, ATIVO, CRIADO_EM FROM FUNCIONARIO_PAINEL WHERE ID = %s", + (funcionario_id,), + ) + return cur.fetchone() + finally: + conn.close() + + +def obter_por_login(login: str) -> dict | None: + """Usado pelo login (app/auth.py) — inclui SENHA_HASH, ao contrário das outras consultas.""" + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "SELECT ID, NOME, LOGIN, SENHA_HASH, PERFIL, ATIVO FROM FUNCIONARIO_PAINEL WHERE LOGIN = %s", + (login,), + ) + return cur.fetchone() + finally: + conn.close() + + +def login_em_uso(login: str, ignorar_id: int | None = None) -> bool: + conn = get_connection() + try: + with conn.cursor() as cur: + sql = "SELECT ID FROM FUNCIONARIO_PAINEL WHERE LOGIN = %s" + params = [login] + if ignorar_id is not None: + sql += " AND ID != %s" + params.append(ignorar_id) + cur.execute(sql, params) + return cur.fetchone() is not None + finally: + conn.close() + + +def criar_funcionario(nome: str, login: str, senha: str, perfil: str) -> int: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO FUNCIONARIO_PAINEL (NOME, LOGIN, SENHA_HASH, PERFIL, ATIVO) VALUES (%s, %s, %s, %s, 1)", + (nome, login, gerar_hash(senha), perfil), + ) + return cur.lastrowid + finally: + conn.close() + + +def atualizar_funcionario(funcionario_id: int, nome: str, login: str, perfil: str) -> None: + """Nome/login/perfil — troca de senha é redefinir_senha().""" + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "UPDATE FUNCIONARIO_PAINEL SET NOME = %s, LOGIN = %s, PERFIL = %s WHERE ID = %s", + (nome, login, perfil, funcionario_id), + ) + finally: + conn.close() + + +def redefinir_senha(funcionario_id: int, nova_senha: str) -> bool: + """True se o ID realmente existia (cur.rowcount > 0) — sem essa + checagem um ID inexistente fazia um UPDATE que não afeta nenhuma + linha, sem erro nenhum do MySQL, e o chamador registrava "sucesso" + no log mesmo sem ter mudado a senha de ninguém.""" + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "UPDATE FUNCIONARIO_PAINEL SET SENHA_HASH = %s WHERE ID = %s", + (gerar_hash(nova_senha), funcionario_id), + ) + return cur.rowcount > 0 + finally: + conn.close() + + +def alternar_ativo(funcionario_id: int) -> None: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute("UPDATE FUNCIONARIO_PAINEL SET ATIVO = 1 - ATIVO WHERE ID = %s", (funcionario_id,)) + finally: + conn.close() + + +def contar_admins_ativos() -> int: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute("SELECT COUNT(*) AS total FROM FUNCIONARIO_PAINEL WHERE PERFIL = 'ADMIN' AND ATIVO = 1") + return cur.fetchone()["total"] + finally: + conn.close() diff --git a/gerenciador/app/db/log.py b/gerenciador/app/db/log.py new file mode 100644 index 0000000..0045093 --- /dev/null +++ b/gerenciador/app/db/log.py @@ -0,0 +1,80 @@ +""" +Log de ações do painel (LOG_PAINEL) — uma linha por ação que muda estado +(criar/editar/excluir cliente, adicionar/editar/excluir contato), sucesso +ou erro. Escrita feita pelos routers via `registrar(...)`, nunca +sobrescrita/editada depois — é histórico, não estado. + +A tabela já existe no banco de dev. Em qualquer outro ambiente (produção), +crie-a primeiro com: + + CREATE TABLE LOG_PAINEL ( + ID INT UNSIGNED NOT NULL AUTO_INCREMENT, + CRIADO_EM DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + USUARIO VARCHAR(60) NOT NULL, + ACAO VARCHAR(40) NOT NULL, + ENTIDADE VARCHAR(20) NOT NULL, + ENTIDADE_ID VARCHAR(20) DEFAULT NULL, + SUCESSO TINYINT(1) NOT NULL, + DETALHE VARCHAR(500) DEFAULT NULL, + PRIMARY KEY (ID), + KEY IDX_LOG_PAINEL_CRIADO_EM (CRIADO_EM) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci; +""" +from .connection import get_connection + + +def registrar( + usuario: str, + acao: str, + entidade: str, + entidade_id: str | None, + sucesso: bool, + detalhe: str | None = None, +) -> None: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO LOG_PAINEL (USUARIO, ACAO, ENTIDADE, ENTIDADE_ID, SUCESSO, DETALHE) " + "VALUES (%s, %s, %s, %s, %s, %s)", + (usuario, acao, entidade, entidade_id, 1 if sucesso else 0, detalhe), + ) + finally: + conn.close() + + +def listar(limite: int = 100, offset: int = 0) -> list[dict]: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute( + "SELECT ID, CRIADO_EM, USUARIO, ACAO, ENTIDADE, ENTIDADE_ID, SUCESSO, DETALHE " + "FROM LOG_PAINEL ORDER BY CRIADO_EM DESC, ID DESC LIMIT %s OFFSET %s", + (limite, offset), + ) + return cur.fetchall() + finally: + conn.close() + + +def contar() -> int: + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute("SELECT COUNT(*) AS total FROM LOG_PAINEL") + return cur.fetchone()["total"] + finally: + conn.close() + + +def limpar() -> None: + """Apaga todo o histórico de LOG_PAINEL. Quem chama (o router) grava + uma entrada nova logo depois disso, registrando quem limpou — nunca + chamar isso sem seguir com um registrar() depois, senão o rastro da + própria limpeza se perde.""" + conn = get_connection() + try: + with conn.cursor() as cur: + cur.execute("DELETE FROM LOG_PAINEL") + finally: + conn.close() diff --git a/gerenciador/app/gerador_chave.py b/gerenciador/app/gerador_chave.py new file mode 100644 index 0000000..b7ce02f --- /dev/null +++ b/gerenciador/app/gerador_chave.py @@ -0,0 +1,56 @@ +""" +Algoritmo de geração de chave (reativação e instalação), reverso-projetado +do sistema legado (Delphi). Fica só aqui no backend — NUNCA deve voltar a +ser exposto em JavaScript de página, porque isso permite qualquer pessoa +com acesso ao painel extrair o algoritmo e gerar chaves fora do sistema. + +Verificado byte-a-byte contra 3 exemplos reais de chave de instalação +(cliente 000001, ALUGUEL): byte inicial 0xDC -> "DC45092C", +0x2E -> "2E35083D", 0x3D -> "3D25090C" (ver testes). +""" +import secrets + +CHAVE_INTERNA = ( + "YUQL23KL23DF90WI5E1JAS467NMCXXL6JAOAUWWMCL0AOMM4A4VZYW9KHJUI2347" + "EJHJKDF3424SKL K3LAKDJSL9RTIKJ" +) + + +def _cifrar(texto: str, byte_inicial: int) -> list[int]: + """Cifra por feedback de byte anterior + XOR com CHAVE_INTERNA rotativa. + Retorna a lista de bytes cifrados (sem o byte_inicial na frente).""" + anterior = byte_inicial + cifrados = [] + for i, ch in enumerate(texto): + valor = (ord(ch) + anterior) % 255 + cifrado = valor ^ ord(CHAVE_INTERNA[i % len(CHAVE_INTERNA)]) + cifrados.append(cifrado) + anterior = cifrado + return cifrados + + +def gerar_reativacao(codigo_cliente: str, mes: int, ano: int, byte_inicial: int | None = None) -> str: + """Chave de reativação — mês/cliente/ano cifrados por inteiro (22 caracteres hex).""" + cliente = str(codigo_cliente).zfill(6) + texto = str(mes).zfill(2) + cliente + str(ano)[-2:] + aleatorio = byte_inicial if byte_inicial is not None else secrets.randbelow(256) + cifrados = _cifrar(texto, aleatorio) + return "".join(f"{v:02X}" for v in [aleatorio, *cifrados]) + + +def gerar_instalacao(codigo_cliente: str, modalidade: str, byte_inicial: int | None = None) -> str: + """Chave de instalação — só a modalidade ("30"=ALUGUEL / "00"=COMPRA) é + cifrada; o valor derivado do código do cliente (cliente*19+31) entra sem + cifrar, no meio dos 6 caracteres cifrados (3 antes + 3 depois).""" + try: + cliente_num = int(str(codigo_cliente)) + except ValueError: + cliente_num = 0 + valor_cliente = cliente_num * 19 + 31 + texto_modalidade = "30" if modalidade == "ALUGUEL" else "00" + + aleatorio = byte_inicial if byte_inicial is not None else secrets.randbelow(256) + cifrados = _cifrar(texto_modalidade, aleatorio) + criptografado = "".join(f"{v:02X}" for v in [aleatorio, *cifrados]) + parte1, parte2 = criptografado[0:3], criptografado[3:6] + return f"{parte1}{valor_cliente}{parte2}" diff --git a/gerenciador/app/internal_fiscal.py b/gerenciador/app/internal_fiscal.py new file mode 100644 index 0000000..74e96d8 --- /dev/null +++ b/gerenciador/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/gerenciador/app/main.py b/gerenciador/app/main.py new file mode 100644 index 0000000..bdd67e6 --- /dev/null +++ b/gerenciador/app/main.py @@ -0,0 +1,119 @@ +""" +Painel de Clientes — monta os routers de CLIENTE, CLIENTE_CONTATO, o +gerador de chave manual, login/logoff, funcionários, minha conta e log. +Ver docs/superpowers/specs/2026-08-13-painel-clientes-web-design.md e +docs/superpowers/specs/2026-08-13-permissoes-login-sessao-design.md. +""" +from fastapi import Depends, FastAPI, Request +from fastapi.responses import RedirectResponse +from fastapi.staticfiles import StaticFiles + +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) +app.mount("/static", StaticFiles(directory="app/static"), name="static") +app.include_router(auth.router) +app.include_router(clientes.router) +app.include_router(contatos.router) +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) + + +@app.exception_handler(NaoAutenticado) +def _redirecionar_para_login(request: Request, exc: NaoAutenticado): + return RedirectResponse(url=f"/login?next={exc.caminho_original}", status_code=303) + + +@app.middleware("http") +async def csrf_cookie_middleware(request: Request, call_next): + """ + Garante que todo navegador que acessa o painel tenha um cookie + csrf_token (gerado uma vez, HttpOnly). Disponibiliza o valor em + request.state.csrf_token para os templates embutirem em formulários — + ver app/csrf.py para a validação do lado do POST. + """ + token_existente = request.cookies.get(CSRF_COOKIE_NAME) + precisa_definir_cookie = token_existente is None + request.state.csrf_token = token_existente or gerar_token() + + response = await call_next(request) + + if precisa_definir_cookie: + response.set_cookie( + CSRF_COOKIE_NAME, request.state.csrf_token, httponly=True, samesite="strict", + # Secure só quando a requisição já chegou em HTTPS — ver o + # comentário equivalente em app/routers/auth.py sobre o + # cookie de sessão (mesma razão, mesmo cuidado com proxy). + secure=request.url.scheme == "https", + ) + return response + + +@app.middleware("http") +async def security_headers_middleware(request: Request, call_next): + """ + Cabeçalhos de resposta que não dependem de nenhuma lógica de negócio, + só reduzem a superfície de ataque do navegador: impedem o painel de + ser carregado dentro de um