From c82982a6052a244e2a45bf929db3c64c7702fa73 Mon Sep 17 00:00:00 2001 From: plx Date: Thu, 30 Jul 2026 22:38:48 +0000 Subject: [PATCH] Harden admin authentication behind trusted proxy --- .env.example | 6 +- app/admin_auth.py | 72 ++++++++++++ app/admin_dashboard.py | 24 +--- app/admin_ui/pages/opportunities.py | 9 +- app/api/internal.py | 20 +--- app/config.py | 22 +++- tests/test_admin_auth_modes.py | 164 ++++++++++++++++++++++++++++ 7 files changed, 273 insertions(+), 44 deletions(-) create mode 100644 app/admin_auth.py create mode 100644 tests/test_admin_auth_modes.py diff --git a/.env.example b/.env.example index 918f5fc..ba1e6f8 100644 --- a/.env.example +++ b/.env.example @@ -67,7 +67,11 @@ PACKLINK_SENDER_EMAIL= PACKLINK_FALLBACK_PHONE= PACKLINK_FALLBACK_EMAIL= -# Admin interno: configure atrás de proxy/auth se exposto fora da rede local +# Admin interno: proxy (header autenticado), token, ou local (apenas loopback) +CLIENTFLOW_ADMIN_AUTH_MODE=proxy +# Em proxy mode, o reverse proxy deve substituir (não apenas encaminhar) +# X-ClientFlow-Admin-User pelo utilizador autenticado. +# Obrigatório apenas quando CLIENTFLOW_ADMIN_AUTH_MODE=token. CLIENTFLOW_ADMIN_TOKEN= # Jasmin API validada em testes reais diff --git a/app/admin_auth.py b/app/admin_auth.py new file mode 100644 index 0000000..b5c181d --- /dev/null +++ b/app/admin_auth.py @@ -0,0 +1,72 @@ +"""Explicit authentication policy shared by the admin UI and internal API.""" + +from __future__ import annotations + +from hmac import compare_digest +from ipaddress import ip_address +from urllib.parse import urlsplit + +from fastapi import HTTPException, Request + +from app.config import settings + + +PROXY_ADMIN_USER_HEADER = "X-ClientFlow-Admin-User" +TOKEN_HEADER = "X-ClientFlow-Admin-Token" +TOKEN_COOKIE = "clientflow_admin_token" + + +def _is_loopback_request(request: Request) -> bool: + """Use the transport peer, never a caller-controlled forwarded header.""" + host = request.client.host if request.client else "" + try: + return ip_address(host).is_loopback + except ValueError: + return False + + +def require_admin_auth(request: Request, *, area: str) -> None: + mode = settings.clientflow_admin_auth_mode + detail_prefix = "internal api" if area == "internal_api" else "admin" + + if mode == "proxy": + admin_user = (request.headers.get(PROXY_ADMIN_USER_HEADER) or "").strip() + if not admin_user: + raise HTTPException(status_code=401, detail=f"{detail_prefix} proxy auth required") + request.state.clientflow_admin_user = admin_user + return + + if mode == "token": + expected = (settings.clientflow_admin_token or "").strip() + if not expected: + raise HTTPException(status_code=503, detail=f"{detail_prefix} auth not configured") + received = ( + request.headers.get(TOKEN_HEADER) + or request.cookies.get(TOKEN_COOKIE) + or "" + ).strip() + if not received or not compare_digest(received, expected): + raise HTTPException(status_code=401, detail=f"{detail_prefix} auth required") + return + + if mode == "local": + if not _is_loopback_request(request): + raise HTTPException(status_code=401, detail=f"{detail_prefix} local access required") + return + + # Settings validates the value, but fail closed if it is mutated at runtime. + raise HTTPException(status_code=503, detail=f"{detail_prefix} auth mode invalid") + + +def safe_local_redirect(referer: str | None, *, fallback: str) -> str: + """Return only an absolute-path local redirect, preserving its query.""" + value = (referer or "").strip() + if not value: + return fallback + parsed = urlsplit(value) + if parsed.scheme or parsed.netloc or not parsed.path.startswith("/") or parsed.path.startswith("//"): + return fallback + target = parsed.path + if parsed.query: + target += f"?{parsed.query}" + return target diff --git a/app/admin_dashboard.py b/app/admin_dashboard.py index 7663661..cd40e13 100644 --- a/app/admin_dashboard.py +++ b/app/admin_dashboard.py @@ -4,7 +4,6 @@ from datetime import datetime, timezone import html import json from uuid import UUID -from hmac import compare_digest from typing import Optional from sqlalchemy import text from fastapi import APIRouter, Request, Depends, HTTPException @@ -12,7 +11,8 @@ from fastapi.responses import HTMLResponse, RedirectResponse, PlainTextResponse, from starlette.concurrency import run_in_threadpool from app.admin_queries import list_action_runs, list_business_events from app.integration_outbox_service import get_outbox_item, list_outbox, set_outbox_status -from app.config import is_production_like_env, settings +from app.admin_auth import require_admin_auth +from app.config import settings from app.preparation_service import prepare_task as run_task_preparation from app.preparation_view_model import build_preparation_view_model from app.workflow_guard import OperationActionBlocked, get_workflow_action_plan @@ -63,24 +63,8 @@ from app.admin_ui.layout import layout from app.admin_ui.styles import ADMIN_UI_V451_CSS # Route handlers moved to app.admin_ui.pages.* in v4.7.2. ADMIN_UI_CSS moved to app.admin_ui.styles. Já existe documento atual. A associação direta fica bloqueada def require_admin_access(request: Request) -> None: - """Proteção opcional da UI admin. - Se CLIENTFLOW_ADMIN_TOKEN estiver vazio, mantém compatibilidade local. - Em produção deve ser definido e enviado em X-ClientFlow-Admin-Token, - cookie clientflow_admin_token, ou query param admin_token atrás de HTTPS/proxy. - """ - expected = (settings.clientflow_admin_token or "").strip() - if not expected: - if is_production_like_env(): - raise HTTPException(status_code=503, detail="admin auth not configured") - return - received = ( - request.headers.get("X-ClientFlow-Admin-Token") - or request.cookies.get("clientflow_admin_token") - or (request.query_params.get("admin_token") if not is_production_like_env() else None) - or "" - ).strip() - if not received or not compare_digest(received, expected): - raise HTTPException(status_code=401, detail="admin auth required") + """Apply UI auth (including X-ClientFlow-Admin-Token in token mode).""" + require_admin_auth(request, area="admin_ui") router = APIRouter(prefix="", tags=["admin"], dependencies=[Depends(require_admin_access)]) def esc(value) -> str: return html.escape(str(value or "")) diff --git a/app/admin_ui/pages/opportunities.py b/app/admin_ui/pages/opportunities.py index 9dd4364..a251b91 100644 --- a/app/admin_ui/pages/opportunities.py +++ b/app/admin_ui/pages/opportunities.py @@ -3109,13 +3109,17 @@ async def fiscal_suggestion_accept_action(suggestion_id: str, request: Request): @router.post("/fiscal-suggestions/{suggestion_id}/reject") async def fiscal_suggestion_reject_action(suggestion_id: str, request: Request): + from app.admin_auth import safe_local_redirect try: from app.fiscal_enrichment_service import reject_fiscal_suggestion reject_fiscal_suggestion(suggestion_id, actor="operator_ui") except Exception as exc: return PlainTextResponse(f"Erro ao rejeitar sugestão fiscal: {exc}", status_code=500) - referer = request.headers.get("referer") or "/opportunities" - return RedirectResponse(referer, status_code=303) + redirect_target = safe_local_redirect( + request.headers.get("referer"), + fallback="/opportunities", + ) + return RedirectResponse(redirect_target, status_code=303) @@ -3422,4 +3426,3 @@ async def opportunity_odoo_link_candidate_action(opportunity_id: str, item_id: s return PlainTextResponse(f"Erro ao associar venda Odoo: {exc}", status_code=500) return RedirectResponse(url=f"/opportunities/{opportunity_id}?notice=Venda%20Odoo%20associada", status_code=303) - diff --git a/app/api/internal.py b/app/api/internal.py index b605d31..e41cb27 100644 --- a/app/api/internal.py +++ b/app/api/internal.py @@ -5,28 +5,16 @@ partials and external monitoring. They do not replace the current admin pages. """ from __future__ import annotations -from hmac import compare_digest - from fastapi import APIRouter, Body, Depends, HTTPException, Request from sqlalchemy import text -from app.config import is_production_like_env, settings +from app.admin_auth import require_admin_auth +from app.config import settings from app.db import engine def require_internal_access(request: Request) -> None: - expected = (settings.clientflow_admin_token or "").strip() - if not expected: - if is_production_like_env(): - raise HTTPException(status_code=503, detail="internal api auth not configured") - return - received = ( - request.headers.get("X-ClientFlow-Admin-Token") - or request.cookies.get("clientflow_admin_token") - or (request.query_params.get("admin_token") if not is_production_like_env() else None) - or "" - ).strip() - if not received or not compare_digest(received, expected): - raise HTTPException(status_code=401, detail="internal api auth required") + """Apply API auth (including X-ClientFlow-Admin-Token in token mode).""" + require_admin_auth(request, area="internal_api") router = APIRouter(prefix="/api/internal", tags=["internal"], dependencies=[Depends(require_internal_access)]) diff --git a/app/config.py b/app/config.py index fdfdf55..2702166 100644 --- a/app/config.py +++ b/app/config.py @@ -1,3 +1,5 @@ +from typing import Literal + from pydantic_settings import BaseSettings, SettingsConfigDict @@ -21,6 +23,7 @@ class Settings(BaseSettings): clientflow_persist: bool = True clientflow_webhook_secret: str = "" + clientflow_admin_auth_mode: Literal["proxy", "token", "local"] = "proxy" clientflow_admin_token: str = "" # UI warning for cases where the original opportunity contact and the @@ -134,10 +137,21 @@ def is_production_like_env() -> bool: return str(settings.env or "").strip().lower() in {"prod", "production", "staging"} -if is_production_like_env() and not str(settings.clientflow_admin_token or "").strip(): - raise RuntimeError( - "CLIENTFLOW_ADMIN_TOKEN é obrigatório em prod/production/staging." - ) +def validate_admin_auth_settings() -> None: + if settings.clientflow_admin_auth_mode == "local" and is_production_like_env(): + raise RuntimeError( + "CLIENTFLOW_ADMIN_AUTH_MODE=local não é permitido em prod/production/staging." + ) + if ( + settings.clientflow_admin_auth_mode == "token" + and not str(settings.clientflow_admin_token or "").strip() + ): + raise RuntimeError( + "CLIENTFLOW_ADMIN_TOKEN é obrigatório quando CLIENTFLOW_ADMIN_AUTH_MODE=token." + ) + + +validate_admin_auth_settings() if ( is_production_like_env() diff --git a/tests/test_admin_auth_modes.py b/tests/test_admin_auth_modes.py new file mode 100644 index 0000000..39627b9 --- /dev/null +++ b/tests/test_admin_auth_modes.py @@ -0,0 +1,164 @@ +from pathlib import Path + +import pytest +from fastapi import HTTPException +from starlette.requests import Request + +from app.admin_auth import require_admin_auth, safe_local_redirect +from app.config import settings, validate_admin_auth_settings + + +ROOT = Path(__file__).resolve().parents[1] + + +def request_from( + host: str = "127.0.0.1", + *, + headers: list[tuple[bytes, bytes]] | None = None, + query_string: bytes = b"", +) -> Request: + return Request( + { + "type": "http", + "method": "GET", + "scheme": "http", + "path": "/", + "raw_path": b"/", + "query_string": query_string, + "headers": headers or [], + "client": (host, 12345), + "server": ("127.0.0.1", 8020), + } + ) + + +def assert_denied(request: Request, *, area: str, status_code: int = 401) -> None: + with pytest.raises(HTTPException) as exc_info: + require_admin_auth(request, area=area) + assert exc_info.value.status_code == status_code + + +@pytest.mark.parametrize("area", ["admin_ui", "internal_api"]) +def test_proxy_mode_accepts_authenticated_user_header(monkeypatch, area): + monkeypatch.setattr(settings, "clientflow_admin_auth_mode", "proxy") + request = request_from( + headers=[(b"x-clientflow-admin-user", b"alice")], + ) + + require_admin_auth(request, area=area) + + assert request.state.clientflow_admin_user == "alice" + + +@pytest.mark.parametrize("area", ["admin_ui", "internal_api"]) +def test_proxy_mode_rejects_missing_authenticated_user_header(monkeypatch, area): + monkeypatch.setattr(settings, "clientflow_admin_auth_mode", "proxy") + + assert_denied(request_from(), area=area) + + +@pytest.mark.parametrize("area", ["admin_ui", "internal_api"]) +def test_token_mode_accepts_valid_header_and_cookie_and_rejects_invalid_token( + monkeypatch, + area, +): + monkeypatch.setattr(settings, "clientflow_admin_auth_mode", "token") + monkeypatch.setattr(settings, "clientflow_admin_token", "correct-secret") + + require_admin_auth( + request_from(headers=[(b"x-clientflow-admin-token", b"correct-secret")]), + area=area, + ) + require_admin_auth( + request_from(headers=[(b"cookie", b"clientflow_admin_token=correct-secret")]), + area=area, + ) + assert_denied( + request_from(headers=[(b"x-clientflow-admin-token", b"wrong-secret")]), + area=area, + ) + + +def test_token_mode_requires_configured_token(monkeypatch): + monkeypatch.setattr(settings, "clientflow_admin_auth_mode", "token") + monkeypatch.setattr(settings, "clientflow_admin_token", "") + + assert_denied(request_from(), area="internal_api", status_code=503) + + +@pytest.mark.parametrize("area", ["admin_ui", "internal_api"]) +def test_local_mode_accepts_loopback_and_rejects_non_loopback(monkeypatch, area): + monkeypatch.setattr(settings, "clientflow_admin_auth_mode", "local") + + require_admin_auth(request_from("127.0.0.1"), area=area) + require_admin_auth(request_from("::1"), area=area) + assert_denied( + request_from( + "192.0.2.10", + headers=[(b"x-forwarded-for", b"127.0.0.1")], + ), + area=area, + ) + + +@pytest.mark.parametrize("env", ["dev", "test"]) +def test_environment_name_never_bypasses_explicit_auth_mode(monkeypatch, env): + monkeypatch.setattr(settings, "env", env) + monkeypatch.setattr(settings, "clientflow_admin_auth_mode", "proxy") + + assert_denied(request_from(), area="admin_ui") + + +@pytest.mark.parametrize("env", ["production", "staging"]) +def test_local_mode_is_rejected_at_startup_in_production_like_env(monkeypatch, env): + monkeypatch.setattr(settings, "env", env) + monkeypatch.setattr(settings, "clientflow_admin_auth_mode", "local") + + with pytest.raises(RuntimeError, match="AUTH_MODE=local não é permitido"): + validate_admin_auth_settings() + + +def test_local_mode_is_allowed_at_startup_in_dev(monkeypatch): + monkeypatch.setattr(settings, "env", "dev") + monkeypatch.setattr(settings, "clientflow_admin_auth_mode", "local") + + validate_admin_auth_settings() + + +@pytest.mark.parametrize("area", ["admin_ui", "internal_api"]) +def test_query_string_admin_token_is_rejected(monkeypatch, area): + monkeypatch.setattr(settings, "clientflow_admin_auth_mode", "token") + monkeypatch.setattr(settings, "clientflow_admin_token", "correct-secret") + + assert_denied( + request_from(query_string=b"admin_token=correct-secret"), + area=area, + ) + + +def test_external_referer_is_rejected_as_redirect_target(): + assert ( + safe_local_redirect( + "https://attacker.example/steal?next=/admin", + fallback="/opportunities", + ) + == "/opportunities" + ) + assert ( + safe_local_redirect( + "/opportunities?notice=done#ignored", + fallback="/opportunities", + ) + == "/opportunities?notice=done" + ) + + +def test_health_and_chatwoot_webhook_are_not_subject_to_admin_auth(): + main_source = (ROOT / "app" / "main.py").read_text() + webhook_source = (ROOT / "app" / "webhooks_chatwoot.py").read_text() + + assert '@app.get("/health")' in main_source + assert 'APIRouter(prefix="/webhooks"' in webhook_source + assert '@router.post("/chatwoot")' in webhook_source + assert "require_admin_access" not in webhook_source + assert "require_internal_access" not in webhook_source