"""Вход по логину (нику) и паролю: регистрация, вход, защита от перебора.""" from __future__ import annotations from fastapi.testclient import TestClient from sqlmodel import Session, select from app.models import User from tests.conftest import csrf_headers PASSWORD = "correct-horse" def _register(client: TestClient, nickname: str = "Игрок", password: str = PASSWORD): return client.post( "/api/auth/register", json={"nickname": nickname, "password": password}, headers=csrf_headers(client), ) def _login(client: TestClient, nickname: str = "Игрок", password: str = PASSWORD): return client.post( "/api/auth/login", json={"nickname": nickname, "password": password}, headers=csrf_headers(client), ) # ─── Регистрация ───────────────────────────────────────────────────────────── def test_register_opens_session(client: TestClient, engine): r = _register(client) assert r.status_code == 200, r.text me = r.json() assert me["nickname"] == "Игрок" assert me["auth_provider"] == "local" assert me["has_password"] is True assert client.cookies.get("fs_session") assert client.get("/api/users/me").json()["id"] == me["id"] with Session(engine) as s: user = s.get(User, me["id"]) assert user.password_hash and PASSWORD not in user.password_hash def test_register_taken_nickname(client: TestClient): assert _register(client).status_code == 200 client.cookies.clear() r = _register(client, password="another-pass") assert r.status_code == 409 assert r.json()["error"]["code"] == "NICKNAME_TAKEN" def test_register_rejects_bad_passwords(client: TestClient): for bad in ["short", " ", "я" * 37]: # короткий, пробелы, 74 байта UTF-8 r = _register(client, password=bad) assert r.status_code == 422, (bad, r.text) assert _register(client, password="x" * 129).status_code == 422 # предел схемы def test_register_rejects_bad_nickname(client: TestClient): assert _register(client, nickname="x").status_code == 422 # ─── Вход ──────────────────────────────────────────────────────────────────── def test_login_after_logout(client: TestClient): uid = _register(client).json()["id"] assert client.post("/api/auth/logout", headers=csrf_headers(client)).status_code == 200 client.cookies.clear() r = _login(client) assert r.status_code == 200, r.text assert r.json()["id"] == uid def test_wrong_password_and_unknown_login_look_the_same(client: TestClient): _register(client) client.cookies.clear() wrong = _login(client, password="wrong-password") unknown = _login(client, nickname="Никто") assert wrong.status_code == unknown.status_code == 401 assert wrong.json() == unknown.json() assert wrong.json()["error"]["code"] == "INVALID_CREDENTIALS" assert "fs_session" not in client.cookies def test_admin_credentials_do_not_open_player_session(client: TestClient, make_admin): make_admin("boss", "secret123") r = _login(client, nickname="boss", password="secret123") assert r.status_code == 401 assert "fs_session" not in client.cookies def test_player_password_does_not_open_admin_session(client: TestClient): _register(client) client.cookies.clear() r = client.post("/api/admin/auth/login", json={"username": "Игрок", "password": PASSWORD}) assert r.status_code == 401 def test_account_without_password_cannot_log_in(client: TestClient, engine): with Session(engine) as s: s.add(User(nickname="Телеграмщик", role="player", auth_provider="telegram")) s.commit() r = _login(client, nickname="Телеграмщик", password="anything-at-all") assert r.status_code == 401 def test_disabled_account_cannot_log_in(client: TestClient, engine): uid = _register(client).json()["id"] client.cookies.clear() with Session(engine) as s: user = s.get(User, uid) user.is_active = False s.add(user) s.commit() r = _login(client) assert r.status_code == 403 assert r.json()["error"]["code"] == "ACCOUNT_DISABLED" def test_login_is_audited_without_secrets(client: TestClient, engine): from app.models import AuditLog uid = _register(client).json()["id"] with Session(engine) as s: logs = s.exec(select(AuditLog).where(AuditLog.entity_id == uid)).all() assert {(log.action, (log.payload or {}).get("provider")) for log in logs} >= { ("create", "local"), ("login", "local"), } assert all(PASSWORD not in str(log.payload) for log in logs) # ─── Защита от перебора ────────────────────────────────────────────────────── def test_throttle_blocks_after_five_failures(client: TestClient, monkeypatch): import app.core.ratelimit as ratelimit now = [1000.0] monkeypatch.setattr(ratelimit.time, "monotonic", lambda: now[0]) _register(client) client.cookies.clear() for _ in range(5): assert _login(client, password="wrong-password").status_code == 401 blocked = _login(client) # даже верный пароль не проверяется assert blocked.status_code == 429 err = blocked.json()["error"] assert err["code"] == "TOO_MANY_ATTEMPTS" assert 0 < err["details"]["retry_after"] <= 15 * 60 + 1 assert "fs_session" not in client.cookies now[0] += 15 * 60 # окно истекло assert _login(client).status_code == 200 def test_success_resets_pair_counter(client: TestClient): _register(client) client.cookies.clear() for _ in range(4): assert _login(client, password="wrong-password").status_code == 401 assert _login(client).status_code == 200 client.cookies.clear() for _ in range(4): assert _login(client, password="wrong-password").status_code == 401 assert _login(client).status_code == 200 def test_throttle_per_ip_across_logins(client: TestClient): """С одного адреса нельзя перебирать пароли по многим логинам: 20 неудач — блок.""" for i in range(20): assert _login(client, nickname=f"Логин{i}", password="wrong-password").status_code == 401 assert _login(client, nickname="Ещё один", password="wrong-password").status_code == 429