ayurishchevandClaude Opus 5.5
d3495fb077
Пользователь сам включает TOTP (QR, приложение-аутентификатор); 10 одноразовых кодов восстановления; superadmin сбрасывает 2FA другому пользователю. Вход в два шага: /auth/login → mfa_token, /auth/login/2fa — под теми же лимитами и advisory-lock, что пароль. Секрет шифруется Fernet-ключом TOTP_ENC_KEY, защита от повторного использования кода (totp_last_step), лимиты перебора кода при отключении. Миграция 0015, зависимости pyotp, segno, cryptography. UI: второй шаг входа, диалоги 2FA в меню логина, бейдж и сброс в «Пользователях»; submitDialog не закрывает окно, открытое следующим шагом цепочки. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
167 lines
10 KiB
Python
167 lines
10 KiB
Python
"""Управление пользователями: один сквозной сценарий (создание → права → отключение → пароли → удаление)."""
|
|
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}")
|