NotBigGhostandClaude Opus 4.8
038a788f99
Вход игрока и админа получают лимит login-user/admin-login-user, не зависящий от IP: ротация X-Forwarded-For (#58) больше не снимает защиту полностью. Успешный вход сбрасывает счётчики аккаунта. В docstring ratelimit — про сброс при рестарте и необходимость внешнего стора при нескольких воркерах. #60 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XfTsytzT6TojfmprRDKiV6
479 lines
20 KiB
Python
479 lines
20 KiB
Python
"""Вход по логину (нику) и паролю: регистрация, вход, защита от перебора."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import Session, select
|
|
|
|
from app.main import app
|
|
from app.models import User
|
|
from tests.conftest import csrf_headers
|
|
from tests.test_auth import _telegram_payload
|
|
|
|
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
|
|
|
|
|
|
def test_account_scoped_throttle_survives_ip_rotation(client: TestClient, engine, monkeypatch):
|
|
"""Перебор одного логина с РАЗНЫХ адресов (ротация X-Forwarded-For, #58) упирается в
|
|
IP-независимый лимит на аккаунт (#60): пара IP+логин и лимит по IP так не копятся."""
|
|
import app.core.ratelimit as ratelimit
|
|
from app.auth.password import _ACCOUNT_LIMIT, login_player
|
|
from app.core.errors import InvalidCredentialsError, TooManyAttemptsError
|
|
|
|
now = [3000.0]
|
|
monkeypatch.setattr(ratelimit.time, "monotonic", lambda: now[0])
|
|
_register(client, nickname="Жертва", password=PASSWORD)
|
|
|
|
with Session(engine) as s:
|
|
for i in range(_ACCOUNT_LIMIT): # каждый раз новый адрес
|
|
with pytest.raises(InvalidCredentialsError):
|
|
login_player(s, "Жертва", "wrong-password", ip=f"10.0.{i // 256}.{i % 256}")
|
|
# ещё одна попытка с совершенно нового адреса — уже блок по лимиту на аккаунт
|
|
with pytest.raises(TooManyAttemptsError):
|
|
login_player(s, "Жертва", "wrong-password", ip="203.0.113.7")
|
|
|
|
|
|
# ─── Установка и смена пароля ────────────────────────────────────────────────
|
|
|
|
def _set_password(client: TestClient, new: str, current: str | None = None):
|
|
body = {"new_password": new}
|
|
if current is not None:
|
|
body["current_password"] = current
|
|
return client.put("/api/users/me/password", json=body, headers=csrf_headers(client))
|
|
|
|
|
|
def _telegram_login(client: TestClient, monkeypatch, **fields):
|
|
from app.core.config import settings
|
|
|
|
monkeypatch.setattr(settings, "telegram_bot_token", "TEST_BOT_TOKEN")
|
|
return client.post(
|
|
"/api/auth/telegram",
|
|
json=_telegram_payload("TEST_BOT_TOKEN", **fields),
|
|
headers=csrf_headers(client),
|
|
)
|
|
|
|
|
|
def test_telegram_user_sets_password_then_logs_in(client: TestClient, monkeypatch):
|
|
"""Сценарий 1 и существующие аккаунты: без пароля → задаёт без текущего → входит по нику."""
|
|
r = _telegram_login(client, monkeypatch)
|
|
assert r.status_code == 200, r.text
|
|
me = r.json()
|
|
assert me["has_password"] is False
|
|
|
|
r2 = _set_password(client, PASSWORD)
|
|
assert r2.status_code == 200, r2.text
|
|
assert r2.json()["has_password"] is True
|
|
|
|
client.cookies.clear()
|
|
r3 = _login(client, nickname=me["nickname"])
|
|
assert r3.status_code == 200, r3.text
|
|
assert r3.json()["id"] == me["id"]
|
|
|
|
|
|
def test_change_password_requires_current(client: TestClient):
|
|
_register(client)
|
|
|
|
missing = _set_password(client, "new-password-1")
|
|
assert missing.status_code == 403
|
|
assert missing.json()["error"]["code"] == "WRONG_CURRENT_PASSWORD"
|
|
assert _set_password(client, "new-password-1", current="wrong-one").status_code == 403
|
|
|
|
assert _set_password(client, "new-password-1", current=PASSWORD).status_code == 200
|
|
client.cookies.clear()
|
|
assert _login(client).status_code == 401
|
|
assert _login(client, password="new-password-1").status_code == 200
|
|
|
|
|
|
def test_change_password_validates_new(client: TestClient):
|
|
_register(client)
|
|
r = _set_password(client, "short", current=PASSWORD)
|
|
assert r.status_code == 422
|
|
client.cookies.clear()
|
|
assert _login(client).status_code == 200 # старый пароль не тронут
|
|
|
|
|
|
def test_current_password_guessing_is_throttled(client: TestClient):
|
|
_register(client)
|
|
for _ in range(5):
|
|
assert _set_password(client, "new-password-1", current="wrong-one").status_code == 403
|
|
blocked = _set_password(client, "new-password-1", current=PASSWORD)
|
|
assert blocked.status_code == 429
|
|
|
|
|
|
def test_set_password_requires_session(client: TestClient):
|
|
assert _set_password(client, PASSWORD).status_code == 401
|
|
|
|
|
|
# ─── Отзыв токена: logout и смена пароля (#57, F2) ────────────────────────────
|
|
|
|
def _me_with_token(cookie_name: str, token: str):
|
|
"""Предъявить конкретный токен вручную (эмуляция «другого устройства»/украденной cookie)."""
|
|
return TestClient(app).get("/api/users/me", headers={"Cookie": f"{cookie_name}={token}"})
|
|
|
|
|
|
def test_logout_revokes_presented_token(client: TestClient):
|
|
_register(client)
|
|
tok = client.cookies.get("fs_session")
|
|
assert _me_with_token("fs_session", tok).status_code == 200 # пока жив
|
|
|
|
assert client.post("/api/auth/logout", headers=csrf_headers(client)).status_code == 200
|
|
# тот же токен, предъявленный после выхода, больше не принимается
|
|
assert _me_with_token("fs_session", tok).status_code == 401
|
|
|
|
|
|
def test_logout_does_not_revoke_other_devices(client: TestClient):
|
|
_register(client) # устройство A
|
|
tok_a = client.cookies.get("fs_session")
|
|
# устройство B: независимый вход тем же аккаунтом (свой jti)
|
|
b = TestClient(app)
|
|
assert b.post("/api/auth/login", json={"nickname": "Игрок", "password": PASSWORD}).status_code == 200
|
|
tok_b = b.cookies.get("fs_session")
|
|
assert tok_a and tok_b and tok_a != tok_b
|
|
|
|
assert client.post("/api/auth/logout", headers=csrf_headers(client)).status_code == 200
|
|
assert _me_with_token("fs_session", tok_a).status_code == 401 # A вышел
|
|
assert _me_with_token("fs_session", tok_b).status_code == 200 # B не тронут
|
|
|
|
|
|
def test_password_change_revokes_old_sessions_keeps_current(client: TestClient):
|
|
_register(client)
|
|
old = client.cookies.get("fs_session")
|
|
assert _set_password(client, "new-password-1", current=PASSWORD).status_code == 200
|
|
# это устройство осталось в сессии (cookie перевыдан со свежим ver)
|
|
assert client.get("/api/users/me").status_code == 200
|
|
# старый токен (другое устройство/утёкший) отозван инкрементом token_version
|
|
assert _me_with_token("fs_session", old).status_code == 401
|
|
|
|
|
|
def test_admin_password_reset_revokes_player_sessions(client: TestClient, monkeypatch, make_admin):
|
|
player = _telegram_login(client, monkeypatch).json()
|
|
stolen = client.cookies.get("fs_session") # действующая сессия игрока
|
|
assert _me_with_token("fs_session", stolen).status_code == 200
|
|
|
|
client.cookies.clear()
|
|
_admin_login(client, make_admin)
|
|
r = client.put(
|
|
f"/api/admin/users/{player['id']}/password",
|
|
json={"new_password": "from-admin-1"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
# сброс пароля админом обрывает прежние сессии игрока (в т.ч. злоумышленника)
|
|
assert _me_with_token("fs_session", stolen).status_code == 401
|
|
|
|
|
|
# ─── Привязка Telegram ───────────────────────────────────────────────────────
|
|
|
|
def _link_telegram(client: TestClient, monkeypatch, **fields):
|
|
from app.core.config import settings
|
|
|
|
monkeypatch.setattr(settings, "telegram_bot_token", "TEST_BOT_TOKEN")
|
|
return client.post(
|
|
"/api/users/me/telegram",
|
|
json=_telegram_payload("TEST_BOT_TOKEN", **fields),
|
|
headers=csrf_headers(client),
|
|
)
|
|
|
|
|
|
def test_link_telegram_then_login_via_telegram(client: TestClient, monkeypatch):
|
|
"""Сценарий 2: аккаунт по паролю → привязал Telegram → вход через него в тот же аккаунт."""
|
|
uid = _register(client).json()["id"]
|
|
|
|
r = _link_telegram(client, monkeypatch) # id=777, тег ivan_tg
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["telegram_id"] == 777
|
|
assert r.json()["nickname"] == "Игрок" # ник не меняется на тег
|
|
|
|
client.cookies.clear()
|
|
r2 = _telegram_login(client, monkeypatch)
|
|
assert r2.status_code == 200, r2.text
|
|
assert r2.json()["id"] == uid
|
|
assert r2.json()["nickname"] == "Игрок"
|
|
|
|
|
|
def test_link_telegram_taken_by_other_account(client: TestClient, monkeypatch):
|
|
assert _telegram_login(client, monkeypatch).status_code == 200 # 777 уже чей-то
|
|
client.cookies.clear()
|
|
_register(client)
|
|
|
|
r = _link_telegram(client, monkeypatch)
|
|
assert r.status_code == 409
|
|
assert r.json()["error"]["code"] == "TELEGRAM_TAKEN"
|
|
|
|
|
|
def test_link_telegram_twice(client: TestClient, monkeypatch):
|
|
_register(client)
|
|
assert _link_telegram(client, monkeypatch).status_code == 200
|
|
r = _link_telegram(client, monkeypatch, id=778)
|
|
assert r.status_code == 409
|
|
assert r.json()["error"]["code"] == "TELEGRAM_ALREADY_LINKED"
|
|
|
|
|
|
def test_link_telegram_bad_signature(client: TestClient, monkeypatch):
|
|
from app.core.config import settings
|
|
|
|
_register(client)
|
|
monkeypatch.setattr(settings, "telegram_bot_token", "TEST_BOT_TOKEN")
|
|
payload = _telegram_payload("TEST_BOT_TOKEN")
|
|
payload["hash"] = "deadbeef"
|
|
r = client.post("/api/users/me/telegram", json=payload, headers=csrf_headers(client))
|
|
assert r.status_code == 401
|
|
assert client.get("/api/users/me").json()["telegram_id"] is None
|
|
|
|
|
|
# ─── Пароль игроку из админки ────────────────────────────────────────────────
|
|
|
|
def _admin_login(client: TestClient, make_admin):
|
|
make_admin("boss", "secret123")
|
|
r = client.post(
|
|
"/api/admin/auth/login",
|
|
json={"username": "boss", "password": "secret123"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["id"]
|
|
|
|
|
|
def test_admin_sets_player_password(client: TestClient, monkeypatch, make_admin):
|
|
player = _telegram_login(client, monkeypatch).json()
|
|
client.cookies.clear()
|
|
_admin_login(client, make_admin)
|
|
|
|
r = client.put(
|
|
f"/api/admin/users/{player['id']}/password",
|
|
json={"new_password": "from-admin-1"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["has_password"] is True
|
|
|
|
client.cookies.clear()
|
|
assert _login(client, nickname=player["nickname"], password="from-admin-1").status_code == 200
|
|
|
|
|
|
def test_admin_cannot_set_admin_password(client: TestClient, make_admin):
|
|
admin_id = _admin_login(client, make_admin)
|
|
r = client.put(
|
|
f"/api/admin/users/{admin_id}/password",
|
|
json={"new_password": "from-admin-1"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
|
|
def test_player_cannot_set_passwords_via_admin(client: TestClient):
|
|
uid = _register(client).json()["id"]
|
|
r = client.put(
|
|
f"/api/admin/users/{uid}/password",
|
|
json={"new_password": "from-admin-1"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert r.status_code == 401
|
|
|
|
|
|
# ─── Защита от перебора пароля администратора (#56, F1) ───────────────────────
|
|
|
|
def test_admin_login_throttled_after_failures(client: TestClient, make_admin, monkeypatch):
|
|
import app.core.ratelimit as ratelimit
|
|
|
|
now = [2000.0]
|
|
monkeypatch.setattr(ratelimit.time, "monotonic", lambda: now[0])
|
|
make_admin("boss", "secret123")
|
|
|
|
for _ in range(5):
|
|
r = client.post("/api/admin/auth/login", json={"username": "boss", "password": "nope"})
|
|
assert r.status_code == 401
|
|
|
|
blocked = client.post("/api/admin/auth/login", json={"username": "boss", "password": "secret123"})
|
|
assert blocked.status_code == 429 # даже верный пароль не проверяется
|
|
assert blocked.json()["error"]["code"] == "TOO_MANY_ATTEMPTS"
|
|
|
|
now[0] += 15 * 60 # окно истекло
|
|
ok = client.post("/api/admin/auth/login", json={"username": "boss", "password": "secret123"})
|
|
assert ok.status_code == 200
|
|
|
|
|
|
def test_failed_admin_login_is_audited_without_password(client: TestClient, make_admin, engine):
|
|
from app.models import AuditLog
|
|
|
|
make_admin("boss", "secret123")
|
|
assert client.post(
|
|
"/api/admin/auth/login", json={"username": "boss", "password": "nope-secret-guess"}
|
|
).status_code == 401
|
|
|
|
with Session(engine) as s:
|
|
logs = s.exec(select(AuditLog).where(AuditLog.action == "login_failed")).all()
|
|
assert any(
|
|
log.entity_type == "admin" and (log.payload or {}).get("username") == "boss" for log in logs
|
|
)
|
|
assert all("nope-secret-guess" not in str(log.payload) for log in logs)
|