ForbiddenStarsApp/backend/tests/test_csrf.py

"""CSRF double-submit: токен восстанавливается, если сессия пережила cookie csrf_token,
а сама проверка мутаций остаётся такой же строгой."""
from __future__ import annotations

import asyncio

from fastapi.testclient import TestClient

from tests.conftest import csrf_headers, login


def _set_cookie(resp, name: str) -> str | None:
    """Заголовок Set-Cookie для cookie name (или None, если ответ её не ставит)."""
    for header in resp.headers.get_list("set-cookie"):
        if header.startswith(f"{name}="):
            return header
    return None


def _max_age(set_cookie: str) -> int:
    for part in set_cookie.split(";"):
        key, _, value = part.strip().partition("=")
        if key.lower() == "max-age":
            return int(value)
    raise AssertionError(f"нет Max-Age: {set_cookie}")


def test_missing_token_reissued_on_safe_request(client: TestClient):
    """Сессия жива, csrf_token истёк → первый же GET отдаёт новый токен, мутации проходят."""
    login(client, "Игрок")
    client.cookies.delete("csrf_token")

    me = client.get("/api/users/me")
    assert me.status_code == 200, me.text
    assert _set_cookie(me, "csrf_token") is not None
    assert client.cookies.get("csrf_token")

    r = client.patch(
        "/api/users/me/profile", json={"favorite_faction_id": None}, headers=csrf_headers(client)
    )
    assert r.status_code == 200, r.text


def test_rejected_mutation_reissues_token(client: TestClient):
    """Отказ CSRF без cookie сам выдаёт токен: иначе не выйти и не перезайти."""
    login(client, "Игрок")
    client.cookies.delete("csrf_token")

    r = client.post("/api/auth/logout")
    assert r.status_code == 403
    assert r.json()["error"]["code"] == "CSRF_FAILED"
    assert _set_cookie(r, "csrf_token") is not None

    r2 = client.post("/api/auth/logout", headers=csrf_headers(client))
    assert r2.status_code == 200, r2.text


def test_admin_login_does_not_shorten_token(client: TestClient, make_admin):
    """Вход в админку перезаписывает общий csrf_token — срок не короче сессии игрока."""
    r_user = client.post("/api/auth/dev/login", json={"nickname": "Игрок"})
    session_age = _max_age(_set_cookie(r_user, "fs_session"))

    make_admin("boss", "secret123")
    r_admin = client.post(
        "/api/admin/auth/login",
        json={"username": "boss", "password": "secret123"},
        headers=csrf_headers(client),
    )
    assert r_admin.status_code == 200, r_admin.text
    assert _max_age(_set_cookie(r_admin, "csrf_token")) >= session_age


def test_check_is_not_weakened(client: TestClient):
    """Cookie есть, заголовка нет или он чужой — 403, и токен при этом не перевыдаётся."""
    login(client, "Игрок")
    token = client.cookies.get("csrf_token")
    body = {"favorite_faction_id": None}

    no_header = client.patch("/api/users/me/profile", json=body)
    assert no_header.status_code == 403
    assert _set_cookie(no_header, "csrf_token") is None

    wrong = client.patch("/api/users/me/profile", json=body, headers={"X-CSRF-Token": "forged"})
    assert wrong.status_code == 403
    assert _set_cookie(wrong, "csrf_token") is None

    assert client.cookies.get("csrf_token") == token


def test_anonymous_gets_no_token(client: TestClient):
    r = client.get("/api/auth/config")
    assert r.status_code == 200
    assert _set_cookie(r, "csrf_token") is None


def _run_middleware(app_messages: list[dict], cookie: bytes) -> list[dict]:
    """Прогоняет CSRFMiddleware над фейковым приложением и возвращает отправленное."""
    from app.main import CSRFMiddleware

    async def fake_app(scope, receive, send):  # noqa: ANN001
        for message in app_messages:
            await send(message)

    sent: list[dict] = []

    async def send(message):  # noqa: ANN001
        sent.append(message)

    async def receive():  # pragma: no cover — фейковому приложению тело запроса не нужно
        return {"type": "http.request", "body": b"", "more_body": False}

    scope = {
        "type": "http",
        "method": "GET",
        "path": "/api/events",
        "raw_path": b"/api/events",
        "root_path": "",
        "scheme": "http",
        "server": ("testserver", 80),
        "query_string": b"",
        "headers": [(b"cookie", cookie)],
    }
    asyncio.run(CSRFMiddleware(fake_app)(scope, receive, send))
    return sent


def test_reissue_keeps_stream_unbuffered():
    """Перевыдача трогает только стартовое сообщение: чанки SSE идут по одному, без склейки."""
    start = {"type": "http.response.start", "status": 200, "headers": [(b"content-type", b"text/event-stream")]}
    chunks = [
        {"type": "http.response.body", "body": b": connected\n\n", "more_body": True},
        {"type": "http.response.body", "body": b": ping\n\n", "more_body": True},
        {"type": "http.response.body", "body": b"", "more_body": False},
    ]
    sent = _run_middleware([start, *chunks], cookie=b"fs_session=abc")

    assert sent[1:] == chunks
    cookies = [v for k, v in sent[0]["headers"] if k == b"set-cookie"]
    assert len(cookies) == 1 and cookies[0].startswith(b"csrf_token=")


def test_reissue_does_not_duplicate_app_cookie():
    """Если приложение само ставит csrf_token (вход), второй Set-Cookie не дописывается."""
    own = (b"set-cookie", b"csrf_token=from-app; Path=/")
    start = {"type": "http.response.start", "status": 200, "headers": [own]}
    body = {"type": "http.response.body", "body": b"{}", "more_body": False}
    sent = _run_middleware([start, body], cookie=b"fs_session=abc")

    cookies = [v for k, v in sent[0]["headers"] if k == b"set-cookie"]
    assert cookies == [own[1]]