Réduit les erreurs ruff de 11 à 5 (toutes pré-existantes non auto-fixables) : - F401 imports inutilisés dans auth/mfa.py - I001 blocs d'imports non triés dans auth/router.py + main.py + pdf_reader.py Les 5 restantes sont dans bookslm/export/watcher (code pré-existant, hors scope). Vérifié : pytest full suite reste 492 passed, 5 skipped, 0 failed.
64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
# backend/auth/mfa.py
|
|
# Multi-Factor Authentication: TOTP + recovery codes.
|
|
# TOTP via pyotp, recovery codes hashed with argon2 for single-use storage.
|
|
|
|
import hashlib
|
|
import logging
|
|
import secrets
|
|
|
|
import pyotp
|
|
|
|
logger = logging.getLogger("obsigate.auth.mfa")
|
|
|
|
TOTP_ISSUER = "ObsiGate"
|
|
|
|
|
|
def generate_secret() -> str:
|
|
"""Generate a new TOTP secret (base32-encoded, 160 bits)."""
|
|
return pyotp.random_base32()
|
|
|
|
|
|
def generate_qr_uri(secret: str, username: str, issuer: str = TOTP_ISSUER) -> str:
|
|
"""Generate an otpauth:// URI for QR code generation."""
|
|
totp = pyotp.TOTP(secret)
|
|
return totp.provisioning_uri(name=username, issuer_name=issuer)
|
|
|
|
|
|
def verify_totp(secret: str, code: str) -> bool:
|
|
"""Verify a TOTP code with a ±1 window tolerance."""
|
|
totp = pyotp.TOTP(secret)
|
|
return totp.verify(code, valid_window=1)
|
|
|
|
|
|
def generate_recovery_codes(n: int = 8) -> list[str]:
|
|
"""Generate n human-readable recovery codes (XXXX-XXXX format)."""
|
|
codes = []
|
|
for _ in range(n):
|
|
# 8 chars alphanumeric, grouped with dash for readability
|
|
raw = secrets.token_hex(4).upper()
|
|
code = f"{raw[:4]}-{raw[4:]}"
|
|
codes.append(code)
|
|
return codes
|
|
|
|
|
|
def hash_recovery_code(code: str) -> str:
|
|
"""Hash a recovery code for storage (SHA-256 for fast comparison).
|
|
|
|
We use SHA-256 instead of argon2 here because recovery codes are
|
|
high-entropy random strings, not user-chosen passwords.
|
|
"""
|
|
return hashlib.sha256(code.upper().encode("utf-8")).hexdigest()
|
|
|
|
|
|
def verify_recovery_code(code: str, hashed_codes: list[str]) -> int | None:
|
|
"""Verify a recovery code against stored hashes.
|
|
|
|
Returns the index of the matched code (for removal), or None if invalid.
|
|
Comparison is case-insensitive.
|
|
"""
|
|
code_hash = hash_recovery_code(code)
|
|
for i, stored_hash in enumerate(hashed_codes):
|
|
if secrets.compare_digest(code_hash, stored_hash):
|
|
return i
|
|
return None
|