CSRF: перевыдавать токен, если сессия есть, а cookie нет
На проде любые изменения данных, включая выход из аккаунта, отвечали 403 CSRF_FAILED, хотя сайт оставался залогиненным. Cookie csrf_token одна на игрока и админку: вход в админку перезаписывал её со сроком 8 часов, а сессия игрока живёт 7 дней. Когда токен истекал, сервер выдавал новый только при входе, а войти и выйти мешала та же проверка. Из этого состояния было не выбраться, кроме как стереть cookie сайта руками. Теперь CSRFMiddleware перевыдаёт токен на любом ответе /api, если запрос несёт сессионную cookie без csrf_token, в том числе на самом отказе. SPA на загрузке делает GET /api/users/me, поэтому пользователю хватает перезагрузить страницу. Правится только стартовое сообщение ответа, тело идёт насквозь, и SSE-поток не буферизуется. Срок токена теперь не короче самой долгой сессии, так что вход в админку больше не укорачивает токен игрока. Проверка double-submit не ослаблена: запрос с cookie, но без заголовка или с чужим токеном по-прежнему получает 403, и cookie в этом случае не перевыдаётся. Перевыданный токен из кросс-сайтового ответа атакующему ничего не даёт: прочитать cookie может только JS того же origin. Тесты закрепляют восстановление на GET и на отказе, срок после входа в админку, прежнюю строгость проверки, отсутствие токена у анонимов и то, что middleware не склеивает чанки потока. На старом коде четыре из них падают. #50 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XfTsytzT6TojfmprRDKiV6
This commit is contained in:
@@ -0,0 +1,150 @@
|
||||
"""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]]
|
||||
Reference in New Issue
Block a user