"""Управление пользователями: один сквозной сценарий (создание → права → отключение → пароли → удаление).""" import time import uuid import httpx import pyotp from tests.conftest import BASE def _client(username: str, password: str): """Клиент с токеном пользователя; None — вход не удался.""" c = httpx.Client(base_url=BASE, timeout=30) r = c.post("/auth/login", json={"username": username, "password": password}) if r.status_code != 200: c.close() return None c.headers["Authorization"] = "Bearer " + r.json()["access_token"] return c def test_users_management(client, org): name, uid, other = f"qa-{uuid.uuid4().hex[:8]}", None, None try: created = client.post("/users", json={"username": name, "password": "start-pass-123", "role": "viewer", "organization_id": org["id"]}) # изменение 033 assert created.status_code == 201, created.text uid = created.json()["id"] assert created.json()["role"] == "viewer" and created.json()["is_active"] is True assert created.json()["organization_id"] == org["id"] # изменение 033 assert client.post("/users", json={"username": name.upper(), "password": "start-pass-123", "organization_id": org["id"]}).status_code == 409 # логин занят без учёта регистра assert client.post("/users", json={"username": "system", "password": "start-pass-123"}).status_code == 422 # служебный логин журнала assert client.post("/users", json={"username": "ab", "password": "start-pass-123"}).status_code == 422 # короткий логин assert client.post("/users", json={"username": "short-pw", "password": "123"}).status_code == 422 # короткий пароль assert client.post("/users", json={"username": f"qa-{uuid.uuid4().hex[:8]}", "password": "start-pass-123", "role": "admin"}).status_code == 422 # админ без организации (изменение 033, находка №2) other = _client(name, "start-pass-123") assert other is not None assert other.get("/auth/me").json()["role"] == "viewer" assert other.get("/users").status_code == 403 # список пользователей только для суперадминистратора (изменение 032) assert other.post("/organizations", json={"name": "qa", "inn": "1234567890"}).status_code == 403 # роль «просмотр» — только чтение # последний вход фиксируется и виден суперадминистратору (изменение 035) found = client.get("/users", params={"q": name}).json()["items"][0] assert isinstance(found["last_login_at"], str) and found["last_login_at"] assert isinstance(found["last_login_ip"], str) and found["last_login_ip"] # отключение действует немедленно, включая ранее выданный токен assert client.patch(f"/users/{uid}", json={"is_active": False}).json()["is_active"] is False assert other.get("/auth/me").status_code == 401 assert _client(name, "start-pass-123") is None # возврат доступа и сброс пароля администратором client.patch(f"/users/{uid}", json={"is_active": True, "password": "reset-pass-123"}) assert _client(name, "start-pass-123") is None other = _client(name, "reset-pass-123") assert other is not None # смена своего пароля: нужен текущий, новый не должен совпадать с ним assert other.post("/users/me/password", json={"current_password": "wrong", "new_password": "third-pass-123"}).status_code == 403 assert other.post("/users/me/password", json={"current_password": "reset-pass-123", "new_password": "reset-pass-123"}).status_code == 422 changed = other.post("/users/me/password", json={"current_password": "reset-pass-123", "new_password": "third-pass-123"}) assert changed.status_code == 200 assert _client(name, "third-pass-123") is not None assert other.get("/auth/me").status_code == 401 # старый токен отозван сменой пароля (изменение 023) other.headers["Authorization"] = "Bearer " + changed.json()["access_token"] assert other.get("/auth/me").status_code == 200 # токен из ответа смены пароля рабочий # свою учётную запись удалить или отключить нельзя me = client.get("/auth/me").json() assert client.delete(f"/users/{me['id']}").status_code == 409 assert client.patch(f"/users/{me['id']}", json={"is_active": False}).status_code == 409 assert client.patch(f"/users/{uid}", json={"role": "admin"}).json()["role"] == "admin" # реконсиляция «роль — организация» по итоговому состоянию (изменение 033, находка №3) promoted = client.patch(f"/users/{uid}", json={"role": "superadmin"}) assert promoted.status_code == 200 and promoted.json()["organization_id"] is None # повышение снимает организацию само assert client.patch(f"/users/{uid}", json={"role": "viewer"}).status_code == 422 # понижение без организации demoted = client.patch(f"/users/{uid}", json={"role": "viewer", "organization_id": org["id"]}) assert demoted.status_code == 200 and demoted.json()["role"] == "viewer" assert client.delete(f"/users/{uid}").status_code == 204 uid = None assert client.get("/audit", params={"event_type": "user.created"}).json()["total"] >= 1 assert client.get("/audit", params={"event_type": "user.password_reset"}).json()["total"] >= 1 finally: if other is not None: other.close() if uid is not None: client.delete(f"/users/{uid}") def test_totp_flow(client, org): """Двухфакторная аутентификация TOTP (изменение 037): включение, вход в два шага, коды восстановления, сброс. Лимит перебора — 5 неудач за 10 минут на пару логин+IP (изменение 026): сценарий укладывается в 2 неудачи (повтор TOTP-кода, повтор кода восстановления).""" name, uid, other = f"qa-totp-{uuid.uuid4().hex[:8]}", None, None try: created = client.post("/users", json={"username": name, "password": "start-pass-123", "role": "viewer", "organization_id": org["id"]}) assert created.status_code == 201, created.text uid = created.json()["id"] other = _client(name, "start-pass-123") assert other is not None assert other.get("/auth/me").json()["totp_enabled"] is False setup = other.post("/users/me/2fa/setup", json={"password": "start-pass-123"}) assert setup.status_code == 200, setup.text secret, qr = setup.json()["secret"], setup.json()["qr"] assert qr.startswith("data:image/svg+xml;base64,") gen = pyotp.TOTP(secret) enabled = other.post("/users/me/2fa/enable", json={"code": gen.now()}) assert enabled.status_code == 200, enabled.text codes = enabled.json()["recovery_codes"] assert len(codes) == 10 assert other.get("/auth/me").json()["totp_enabled"] is True # вход паролем теперь останавливается на первом факторе anon = httpx.Client(base_url=BASE, timeout=30) step1 = anon.post("/auth/login", json={"username": name, "password": "start-pass-123"}) assert step1.status_code == 200 body = step1.json() assert body["mfa_required"] is True and body["access_token"] is None and body["mfa_token"] mfa_token = body["mfa_token"] # mfa_token не работает как токен доступа probe = httpx.Client(base_url=BASE, timeout=30, headers={"Authorization": "Bearer " + mfa_token}) assert probe.get("/auth/me").status_code == 401 # код, использованный в enable(), уже погашен (totp_last_step) — дожидаемся следующего шага TOTP (30 с), # чтобы не столкнуться с этим же шагом при первом входе time.sleep(30 - (time.time() % 30) + 2) code = gen.now() ok = anon.post("/auth/login/2fa", json={"mfa_token": mfa_token, "code": code}) assert ok.status_code == 200, ok.text assert ok.json()["access_token"] # повторное использование того же кода отклоняется (mfa_token остаётся действителен 5 минут — используем его же) replay = anon.post("/auth/login/2fa", json={"mfa_token": mfa_token, "code": code}) assert replay.status_code == 401 # вход кодом восстановления rc = codes[0] rec = anon.post("/auth/login/2fa", json={"mfa_token": mfa_token, "code": rc}) assert rec.status_code == 200, rec.text assert rec.json()["access_token"] # повтор того же кода восстановления отклоняется rec2 = anon.post("/auth/login/2fa", json={"mfa_token": mfa_token, "code": rc}) assert rec2.status_code == 401 anon.close() probe.close() # сброс суперадминистратором — вход снова только паролем reset = client.post(f"/users/{uid}/2fa/reset") assert reset.status_code == 200, reset.text other.close() other = _client(name, "start-pass-123") assert other is not None assert other.get("/auth/me").json()["totp_enabled"] is False finally: if other is not None: other.close() if uid is not None: client.delete(f"/users/{uid}")