Merge pull request 'Добавление аутентификации логин/пароль (#24)' (#53) from issue-24-login-password into dev
Reviewed-on: #53
This commit was merged in pull request #53.
This commit is contained in:
@@ -18,6 +18,7 @@ from sqlalchemy.pool import StaticPool # noqa: E402
|
||||
from sqlmodel import Session, SQLModel, create_engine, select # noqa: E402
|
||||
|
||||
import app.models # noqa: F401,E402 (регистрация моделей)
|
||||
from app.core.ratelimit import login_throttle # noqa: E402
|
||||
from app.core.security import hash_password # noqa: E402
|
||||
from app.db.session import get_session # noqa: E402
|
||||
from app.main import app # noqa: E402
|
||||
@@ -48,6 +49,7 @@ def client(engine):
|
||||
yield s
|
||||
|
||||
app.dependency_overrides[get_session] = _get_session
|
||||
login_throttle.clear() # счётчики неудачных входов глобальны для процесса
|
||||
with TestClient(app) as c:
|
||||
yield c
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
@@ -10,8 +10,9 @@ from fastapi.testclient import TestClient
|
||||
from tests.conftest import csrf_headers
|
||||
|
||||
|
||||
def test_auth_config_dev_has_both_methods(client: TestClient):
|
||||
def test_auth_config_dev_has_all_methods(client: TestClient):
|
||||
cfg = client.get("/api/auth/config").json()
|
||||
assert "password" in cfg["methods"]
|
||||
assert "telegram" in cfg["methods"]
|
||||
assert "stub" in cfg["methods"] # dev → доступен вход по нику
|
||||
|
||||
@@ -21,11 +22,11 @@ def test_enabled_methods_by_env(monkeypatch):
|
||||
from app.core.config import settings
|
||||
|
||||
monkeypatch.setattr(settings, "app_env", "development")
|
||||
assert set(enabled_methods()) == {"telegram", "stub"}
|
||||
assert set(enabled_methods()) == {"password", "telegram", "stub"}
|
||||
monkeypatch.setattr(settings, "app_env", "test")
|
||||
assert enabled_methods() == ["telegram"] # test (прод-клон) → только Telegram
|
||||
assert enabled_methods() == ["password", "telegram"] # test (прод-клон) → без stub
|
||||
monkeypatch.setattr(settings, "app_env", "production")
|
||||
assert enabled_methods() == ["telegram"] # prod → только Telegram
|
||||
assert enabled_methods() == ["password", "telegram"] # prod → без stub
|
||||
|
||||
|
||||
def test_env_flags_and_db_path(monkeypatch):
|
||||
|
||||
@@ -0,0 +1,360 @@
|
||||
"""Вход по логину (нику) и паролю: регистрация, вход, защита от перебора."""
|
||||
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
|
||||
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 _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
|
||||
|
||||
|
||||
# ─── Привязка 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
|
||||
Reference in New Issue
Block a user