diff --git a/README.md b/README.md index b5c4475..0213e1c 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,8 @@ 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 HTTP Basic em toda rota, 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). + ## Rodando localmente (sem Docker) 1. Instale as dependências de desenvolvimento: diff --git a/app/csrf.py b/app/csrf.py new file mode 100644 index 0000000..f57940f --- /dev/null +++ b/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/app/db/clientes.py b/app/db/clientes.py index c291904..6ae113f 100644 --- a/app/db/clientes.py +++ b/app/db/clientes.py @@ -21,7 +21,7 @@ class ClienteComDependenciasError(Exception): """Levantada ao tentar excluir um CLIENTE que ainda tem CLIENTE_CONTATO.""" -def listar_clientes(busca: str | None = None, limite: int = 100) -> list[dict]: +def listar_clientes(busca: str | None = None, limite: int = 100, offset: int = 0) -> list[dict]: conn = get_connection() try: with conn.cursor() as cur: @@ -29,17 +29,37 @@ def listar_clientes(busca: str | None = None, limite: int = 100) -> list[dict]: sql = ( f"SELECT {_COLUNAS_SELECT} FROM CLIENTE " "WHERE CODIGO LIKE %s OR CNPJ LIKE %s OR RAZAO LIKE %s OR FANTASIA LIKE %s " - "ORDER BY RAZAO LIMIT %s" + "ORDER BY RAZAO LIMIT %s OFFSET %s" ) termo = f"%{busca}%" - cur.execute(sql, (termo, termo, termo, termo, limite)) + cur.execute(sql, (termo, termo, termo, termo, limite, offset)) else: - cur.execute(f"SELECT {_COLUNAS_SELECT} FROM CLIENTE ORDER BY RAZAO LIMIT %s", (limite,)) + cur.execute( + f"SELECT {_COLUNAS_SELECT} FROM CLIENTE ORDER BY RAZAO LIMIT %s OFFSET %s", (limite, offset) + ) return cur.fetchall() finally: conn.close() +def contar_clientes(busca: str | None = None) -> int: + conn = get_connection() + try: + with conn.cursor() as cur: + if busca: + sql = ( + "SELECT COUNT(*) AS total FROM CLIENTE " + "WHERE CODIGO LIKE %s OR CNPJ LIKE %s OR RAZAO LIKE %s OR FANTASIA LIKE %s" + ) + termo = f"%{busca}%" + cur.execute(sql, (termo, termo, termo, termo)) + else: + cur.execute("SELECT COUNT(*) AS total FROM CLIENTE") + return cur.fetchone()["total"] + finally: + conn.close() + + def obter_cliente(cliente_id: int) -> dict | None: conn = get_connection() try: diff --git a/app/main.py b/app/main.py index b44e859..b662809 100644 --- a/app/main.py +++ b/app/main.py @@ -2,11 +2,12 @@ Painel de Clientes — monta os routers de CLIENTE, CLIENTE_CONTATO e o gerador de chave manual. Ver docs/superpowers/specs/2026-08-13-painel-clientes-web-design.md. """ -from fastapi import Depends, FastAPI +from fastapi import Depends, FastAPI, Request from fastapi.responses import RedirectResponse from fastapi.staticfiles import StaticFiles from .auth import exigir_login +from .csrf import CSRF_COOKIE_NAME, gerar_token from .routers import chave_manual, clientes, contatos app = FastAPI(title="Painel de Clientes", docs_url=None, redoc_url=None, openapi_url=None) @@ -16,6 +17,27 @@ app.include_router(contatos.router) app.include_router(chave_manual.router) +@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" + ) + return response + + @app.get("/") def raiz(usuario: str = Depends(exigir_login)): return RedirectResponse(url="/clientes") diff --git a/app/routers/clientes.py b/app/routers/clientes.py index 306221f..1c7b98d 100644 --- a/app/routers/clientes.py +++ b/app/routers/clientes.py @@ -1,8 +1,9 @@ # app/routers/clientes.py """ -Rotas de gestão de CLIENTE: listar/buscar, criar, editar, ver detalhe -(contatos + status de Clientes_Chaves), excluir. +Rotas de gestão de CLIENTE: listar/buscar (paginado), criar, editar, ver +detalhe (contatos + status de Clientes_Chaves), excluir. """ +import math from typing import Optional from urllib.parse import urlencode @@ -12,6 +13,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from ..auth import exigir_login +from ..csrf import exigir_csrf, token_do_contexto from ..db import chaves as db_chaves from ..db import clientes as db_clientes from ..db import contatos as db_contatos @@ -20,6 +22,8 @@ from ..normalize import normalizar_documento, normalizar_telefone, validar_dv_cp router = APIRouter(prefix="/clientes") templates = Jinja2Templates(directory="app/templates") +TAMANHO_PAGINA = 100 + def _redirect_com_mensagem(url: str, mensagem: str, tipo: str = "success") -> RedirectResponse: separador = "&" if "?" in url else "?" @@ -48,15 +52,25 @@ def _normalizar_documento_e_telefones(dados: dict) -> None: def listar( request: Request, busca: Optional[str] = None, + pagina: int = 1, mensagem: Optional[str] = None, mensagem_tipo: Optional[str] = None, usuario: str = Depends(exigir_login), ): - clientes = db_clientes.listar_clientes(busca=busca or None) + pagina = max(pagina, 1) + total = db_clientes.contar_clientes(busca=busca or None) + total_paginas = max(math.ceil(total / TAMANHO_PAGINA), 1) + pagina = min(pagina, total_paginas) + offset = (pagina - 1) * TAMANHO_PAGINA + + clientes = db_clientes.listar_clientes(busca=busca or None, limite=TAMANHO_PAGINA, offset=offset) return templates.TemplateResponse( request, "clientes_lista.html", - {"clientes": clientes, "busca": busca, "mensagem": mensagem, "mensagem_tipo": mensagem_tipo}, + { + "clientes": clientes, "busca": busca, "mensagem": mensagem, "mensagem_tipo": mensagem_tipo, + "pagina": pagina, "total_paginas": total_paginas, "total": total, + }, ) @@ -65,11 +79,14 @@ def form_novo(request: Request, usuario: str = Depends(exigir_login)): return templates.TemplateResponse( request, "cliente_form.html", - {"cliente": None, "dados": {c: "" for c in db_clientes.CAMPOS_CLIENTE}, "campos": db_clientes.CAMPOS_CLIENTE}, + { + "cliente": None, "dados": {c: "" for c in db_clientes.CAMPOS_CLIENTE}, "campos": db_clientes.CAMPOS_CLIENTE, + "csrf_token": token_do_contexto(request), + }, ) -@router.post("/novo") +@router.post("/novo", dependencies=[Depends(exigir_csrf)]) async def criar(request: Request, usuario: str = Depends(exigir_login)): form = dict((await request.form())) dados = _ler_form_cliente(form) @@ -85,6 +102,7 @@ async def criar(request: Request, usuario: str = Depends(exigir_login)): "cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "aviso_codigo_duplicado": True, "confirmar_codigo_duplicado": confirmar_duplicado, "confirmar_dv_invalido": confirmar_dv, + "csrf_token": token_do_contexto(request), }, ) @@ -102,6 +120,7 @@ async def criar(request: Request, usuario: str = Depends(exigir_login)): "cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "aviso_dv_invalido": True, "confirmar_codigo_duplicado": confirmar_duplicado, "confirmar_dv_invalido": confirmar_dv, + "csrf_token": token_do_contexto(request), }, ) @@ -111,7 +130,10 @@ async def criar(request: Request, usuario: str = Depends(exigir_login)): return templates.TemplateResponse( request, "cliente_form.html", - {"cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "erro_gravacao": str(exc)}, + { + "cliente": None, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "erro_gravacao": str(exc), + "csrf_token": token_do_contexto(request), + }, ) return _redirect_com_mensagem(f"/clientes/{novo_id}", "Cliente criado.") @@ -136,6 +158,7 @@ def detalhe( { "cliente": cliente, "contatos": contatos, "status_chaves": status, "mensagem": mensagem, "mensagem_tipo": mensagem_tipo, + "csrf_token": token_do_contexto(request), }, ) @@ -148,11 +171,14 @@ def form_editar(request: Request, cliente_id: int, usuario: str = Depends(exigir return templates.TemplateResponse( request, "cliente_form.html", - {"cliente": cliente, "dados": cliente, "campos": db_clientes.CAMPOS_CLIENTE}, + { + "cliente": cliente, "dados": cliente, "campos": db_clientes.CAMPOS_CLIENTE, + "csrf_token": token_do_contexto(request), + }, ) -@router.post("/{cliente_id}/editar") +@router.post("/{cliente_id}/editar", dependencies=[Depends(exigir_csrf)]) async def editar(request: Request, cliente_id: int, usuario: str = Depends(exigir_login)): form = dict((await request.form())) dados = _ler_form_cliente(form) @@ -163,7 +189,10 @@ async def editar(request: Request, cliente_id: int, usuario: str = Depends(exigi return templates.TemplateResponse( request, "cliente_form.html", - {"cliente": {"ID": cliente_id}, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "aviso_codigo_duplicado": True}, + { + "cliente": {"ID": cliente_id}, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, + "aviso_codigo_duplicado": True, "csrf_token": token_do_contexto(request), + }, ) try: @@ -172,12 +201,15 @@ async def editar(request: Request, cliente_id: int, usuario: str = Depends(exigi return templates.TemplateResponse( request, "cliente_form.html", - {"cliente": {"ID": cliente_id}, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, "erro_gravacao": str(exc)}, + { + "cliente": {"ID": cliente_id}, "dados": dados, "campos": db_clientes.CAMPOS_CLIENTE, + "erro_gravacao": str(exc), "csrf_token": token_do_contexto(request), + }, ) return _redirect_com_mensagem(f"/clientes/{cliente_id}", "Cliente atualizado.") -@router.post("/{cliente_id}/excluir") +@router.post("/{cliente_id}/excluir", dependencies=[Depends(exigir_csrf)]) def excluir(cliente_id: int, usuario: str = Depends(exigir_login)): try: db_clientes.excluir_cliente(cliente_id) diff --git a/app/routers/contatos.py b/app/routers/contatos.py index 008d4e2..d16d689 100644 --- a/app/routers/contatos.py +++ b/app/routers/contatos.py @@ -9,6 +9,7 @@ from fastapi import APIRouter, Depends, Form from fastapi.responses import RedirectResponse from ..auth import exigir_login +from ..csrf import exigir_csrf from ..db import clientes as db_clientes from ..db import contatos as db_contatos from ..normalize import normalizar_telefone @@ -21,7 +22,7 @@ def _redirect_com_mensagem(cliente_id: int, mensagem: str, tipo: str = "success" return RedirectResponse(f"/clientes/{cliente_id}?{query}", status_code=303) -@router.post("/{cliente_id}/contatos") +@router.post("/{cliente_id}/contatos", dependencies=[Depends(exigir_csrf)]) def adicionar( cliente_id: int, nome: str = Form(""), @@ -43,7 +44,7 @@ def adicionar( return _redirect_com_mensagem(cliente_id, "Contato adicionado.") -@router.post("/{cliente_id}/contatos/{contato_id}/editar") +@router.post("/{cliente_id}/contatos/{contato_id}/editar", dependencies=[Depends(exigir_csrf)]) def editar( cliente_id: int, contato_id: int, @@ -63,7 +64,7 @@ def editar( return _redirect_com_mensagem(cliente_id, "Contato atualizado.") -@router.post("/{cliente_id}/contatos/{contato_id}/excluir") +@router.post("/{cliente_id}/contatos/{contato_id}/excluir", dependencies=[Depends(exigir_csrf)]) def excluir(cliente_id: int, contato_id: int, usuario: str = Depends(exigir_login)): db_contatos.excluir_contato(contato_id, cliente_id) return _redirect_com_mensagem(cliente_id, "Contato excluído.") diff --git a/app/templates/cliente_detalhe.html b/app/templates/cliente_detalhe.html index f9dc7e7..e104f6b 100644 --- a/app/templates/cliente_detalhe.html +++ b/app/templates/cliente_detalhe.html @@ -7,6 +7,7 @@
@@ -21,6 +22,7 @@