gendesign/tradein-mvp/backend/tests/test_internal_auth_secret.py

"""Defense-in-depth: shared-secret gate поверх trusted-header auth (#2213).

Backend доверяет X-Authenticated-User (его ставит Caddy). На общей docker-сети
gendesign_shared любой контейнер мог бы отправить поддельный
`X-Authenticated-User: admin` напрямую на tradein-backend:8000. Общий секрет
X-Internal-Auth-Secret закрывает дыру: если TRADEIN_INTERNAL_AUTH_SECRET задан,
каждый запрос с X-Authenticated-User обязан нести валидный секрет (constant-time),
иначе 401. Пусто = fail-open (backward-compat до провижининга).

MIRROR of rbac_guard из app/main.py — включая #2213 secret-gate. Держим копию
здесь (как test_rbac.py), чтобы не тянуть тяжёлый app.main (lifespan/DB/scheduler).
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import re
import secrets
from collections.abc import Awaitable, Callable

import pytest
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response
from fastapi.testclient import TestClient

from app.core import auth as auth_mod
from app.core import config

_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})


@pytest.fixture(autouse=True)
def _reset_auth_cache() -> None:
    auth_mod.reset_cache_for_tests()


def _build_test_app() -> FastAPI:
    """Копия rbac_guard из app/main.py (с #2213 secret-gate)."""
    app = FastAPI()

    @app.middleware("http")
    async def rbac_guard(
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        path = request.url.path
        if path in _PUBLIC_PATHS:
            return await call_next(request)

        username = request.headers.get("X-Authenticated-User")
        if not username:
            return JSONResponse(
                status_code=401,
                content={"detail": "no authenticated user (Caddy basic_auth required)"},
            )

        secret = config.settings.tradein_internal_auth_secret
        if secret:
            provided = request.headers.get("X-Internal-Auth-Secret", "")
            if not secrets.compare_digest(provided, secret):
                return JSONResponse(
                    status_code=401,
                    content={"detail": "invalid or missing internal auth secret"},
                )

        try:
            role = auth_mod.get_role(username)
        except KeyError:
            return JSONResponse(
                status_code=403,
                content={"detail": "user not in roles config"},
            )
        if _ADMIN_API_RE.match(path) and role != "admin":
            return JSONResponse(status_code=403, content={"detail": "admin only"})
        return await call_next(request)

    @app.get("/api/v1/ping")
    async def ping() -> dict:
        return {"ok": True}

    return app


@pytest.fixture
def client() -> TestClient:
    return TestClient(_build_test_app())


# ---------------------------------------------------------------------------
# (в) секрет НЕ задан → backward-compat: X-Authenticated-User достаточно.
# ---------------------------------------------------------------------------


def test_secret_unset_backward_compat(client: TestClient, monkeypatch) -> None:
    monkeypatch.setattr(config.settings, "tradein_internal_auth_secret", "")
    resp = client.get("/api/v1/ping", headers={"X-Authenticated-User": "admin"})
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


def test_secret_unset_ignores_provided_secret(client: TestClient, monkeypatch) -> None:
    # Если секрет не задан — присланный клиентом X-Internal-Auth-Secret игнорируется.
    monkeypatch.setattr(config.settings, "tradein_internal_auth_secret", "")
    resp = client.get(
        "/api/v1/ping",
        headers={"X-Authenticated-User": "admin", "X-Internal-Auth-Secret": "whatever"},
    )
    assert resp.status_code == 200


# ---------------------------------------------------------------------------
# (а) секрет задан + заголовок юзера БЕЗ секрета → 401.
# ---------------------------------------------------------------------------


def test_secret_set_missing_header_401(client: TestClient, monkeypatch) -> None:
    monkeypatch.setattr(config.settings, "tradein_internal_auth_secret", "s3cr3t-value")
    resp = client.get("/api/v1/ping", headers={"X-Authenticated-User": "admin"})
    assert resp.status_code == 401
    assert "internal auth secret" in resp.json()["detail"].lower()


def test_secret_set_wrong_secret_401(client: TestClient, monkeypatch) -> None:
    monkeypatch.setattr(config.settings, "tradein_internal_auth_secret", "s3cr3t-value")
    resp = client.get(
        "/api/v1/ping",
        headers={"X-Authenticated-User": "admin", "X-Internal-Auth-Secret": "wrong"},
    )
    assert resp.status_code == 401
    assert "internal auth secret" in resp.json()["detail"].lower()


# ---------------------------------------------------------------------------
# (б) секрет задан + оба заголовка корректны → 200 (существующее поведение).
# ---------------------------------------------------------------------------


def test_secret_set_correct_secret_200(client: TestClient, monkeypatch) -> None:
    monkeypatch.setattr(config.settings, "tradein_internal_auth_secret", "s3cr3t-value")
    resp = client.get(
        "/api/v1/ping",
        headers={"X-Authenticated-User": "admin", "X-Internal-Auth-Secret": "s3cr3t-value"},
    )
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


def test_secret_set_no_user_header_still_401(client: TestClient, monkeypatch) -> None:
    # Секрет без X-Authenticated-User — no-auth 401 (secret-gate не ослабляет базу).
    monkeypatch.setattr(config.settings, "tradein_internal_auth_secret", "s3cr3t-value")
    resp = client.get("/api/v1/ping", headers={"X-Internal-Auth-Secret": "s3cr3t-value"})
    assert resp.status_code == 401
    assert "no authenticated user" in resp.json()["detail"].lower()