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>
111 lines
5.2 KiB
Python
111 lines
5.2 KiB
Python
"""Двухфакторная аутентификация TOTP, по выбору пользователя (изменение 037).
|
|
|
|
Секрет хранится в БД зашифрованным (Fernet, ключ `TOTP_ENC_KEY`) — читаемый дамп БД не даёт готового секрета
|
|
для генератора кодов. Коды восстановления хранятся хэшем sha256 (не argon2: код высокоэнтропийный,
|
|
10 символов base32 — медленный хэш не даёт защиты, которая была бы нужна для короткого пользовательского пароля).
|
|
"""
|
|
import base64
|
|
import hashlib
|
|
import io
|
|
import secrets
|
|
import time
|
|
|
|
import pyotp
|
|
import segno
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.models import RecoveryCode, User
|
|
|
|
ISSUER = "IPAM Manager"
|
|
STEP_SECONDS = 30 # шаг TOTP по умолчанию (RFC 6238), используется и pyotp
|
|
RECOVERY_CODES_COUNT = 10
|
|
_RECOVERY_ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567" # base32-алфавит без спутываемых символов (0/O, 1/I и т.п. base32 не использует)
|
|
|
|
|
|
def _fernet() -> Fernet:
|
|
if not settings.totp_enc_key:
|
|
raise RuntimeError("TOTP_ENC_KEY не задан")
|
|
return Fernet(settings.totp_enc_key.encode())
|
|
|
|
|
|
def encrypt(secret: str) -> str:
|
|
return _fernet().encrypt(secret.encode()).decode()
|
|
|
|
|
|
def decrypt(token: str) -> str:
|
|
try:
|
|
return _fernet().decrypt(token.encode()).decode()
|
|
except InvalidToken as e: # смена/потеря TOTP_ENC_KEY после включения 2FA — секрет не читается
|
|
raise RuntimeError("TOTP_ENC_KEY не подходит к сохранённому секрету") from e
|
|
|
|
|
|
def new_secret() -> str:
|
|
return pyotp.random_base32()
|
|
|
|
|
|
def provisioning_uri(secret: str, username: str) -> str:
|
|
return pyotp.TOTP(secret).provisioning_uri(name=username, issuer_name=ISSUER)
|
|
|
|
|
|
def qr_svg_data_uri(uri: str) -> str:
|
|
"""QR в SVG, отрисованный на сервере (без внешних вызовов); отдаётся как data: URI — CSP уже разрешает img-src data:."""
|
|
buf = io.BytesIO()
|
|
segno.make(uri, error="m").save(buf, kind="svg", scale=4, border=2, dark="#111111", light="#ffffff", xmldecl=False)
|
|
return "data:image/svg+xml;base64," + base64.b64encode(buf.getvalue()).decode()
|
|
|
|
|
|
def check_code(secret: str, code: str, last_step: int | None) -> int | None:
|
|
"""Проверка TOTP-кода по конкретному секрету с окном ±1 шаг (изменение 037). Шаг принятого кода должен быть
|
|
строго больше `last_step` (повторное использование в том же/более раннем шаге отклоняется). Возвращает
|
|
принятый шаг (для сохранения в totp_last_step) или None, если код не подошёл."""
|
|
code = code.strip().replace(" ", "")
|
|
if not code.isdigit():
|
|
return None
|
|
totp = pyotp.TOTP(secret, interval=STEP_SECONDS)
|
|
now_step = int(time.time()) // STEP_SECONDS
|
|
for step in (now_step, now_step - 1, now_step + 1):
|
|
if last_step is not None and step <= last_step:
|
|
continue
|
|
if secrets.compare_digest(totp.generate_otp(step), code):
|
|
return step
|
|
return None
|
|
|
|
|
|
def verify(user: User, code: str) -> int | None:
|
|
"""Проверка кода по подтверждённому (включённому) секрету пользователя."""
|
|
if not user.totp_secret_enc:
|
|
return None
|
|
return check_code(decrypt(user.totp_secret_enc), code, user.totp_last_step)
|
|
|
|
|
|
def new_recovery_codes() -> list[tuple[str, str]]:
|
|
"""10 кодов восстановления вида XXXXX-XXXXX и их хэши sha256 (hex) — для сохранения в БД."""
|
|
out = []
|
|
for _ in range(RECOVERY_CODES_COUNT):
|
|
raw = "".join(secrets.choice(_RECOVERY_ALPHABET) for _ in range(10))
|
|
code = f"{raw[:5]}-{raw[5:]}"
|
|
out.append((code, hashlib.sha256(raw.encode()).hexdigest()))
|
|
return out
|
|
|
|
|
|
def _normalize_recovery(code: str) -> str:
|
|
return code.strip().upper().replace("-", "").replace(" ", "")
|
|
|
|
|
|
def use_recovery_code(db: Session, user: User, code: str) -> int | None:
|
|
"""Погашает код восстановления, если он подходит и ещё не использован. Возвращает остаток неиспользованных
|
|
кодов (может быть 0) или None, если код неверен/уже использован."""
|
|
norm = _normalize_recovery(code)
|
|
if not norm:
|
|
return None
|
|
h = hashlib.sha256(norm.encode()).hexdigest()
|
|
row = db.scalar(select(RecoveryCode).where(RecoveryCode.user_id == user.id, RecoveryCode.code_hash == h, RecoveryCode.used_at.is_(None)))
|
|
if row is None:
|
|
return None
|
|
row.used_at = func.now()
|
|
db.flush()
|
|
return db.scalar(select(func.count()).select_from(RecoveryCode).where(RecoveryCode.user_id == user.id, RecoveryCode.used_at.is_(None))) or 0
|