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