feat(auth): v2.1.0 - MFA TOTP avec QR code, recovery codes
CI / lint (push) Failing after 16s
CI / test (push) Skipped
CI / build (push) Skipped
CI / e2e (push) Skipped
CI / security (push) Successful in 21s
Desktop Build / build-windows (push) Canceled after 0s
Desktop Build / build-linux (push) Canceled after 0s
CI / lint (push) Failing after 16s
CI / test (push) Skipped
CI / build (push) Skipped
CI / e2e (push) Skipped
CI / security (push) Successful in 21s
Desktop Build / build-windows (push) Canceled after 0s
Desktop Build / build-linux (push) Canceled after 0s
- backend/auth/mfa.py: TOTP (pyotp), recovery codes SHA-256 - Login flow: mfa_required → totp/verify → token (ou recovery) - 6 nouveaux endpoints /api/auth/mfa/* - Frontend: QR code setup, 6-digit auto-submit, recovery codes - Settings: section Sécurité avec enable/disable MFA - CSS: mfa-challenge, setup-card, recovery-list, badges - i18n: 36 nouvelles clés EN/FR - pyotp ajouté aux dépendances - 30 tests (TOTP, recovery, API endpoints, login flow) - 439 tests passent au total
This commit is contained in:
@@ -0,0 +1,65 @@
|
||||
# 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
|
||||
|
||||
from .password import hash_password, verify_password
|
||||
|
||||
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
|
||||
+215
-10
@@ -21,6 +21,14 @@ from .jwt_handler import (
|
||||
revoke_token,
|
||||
)
|
||||
from .middleware import is_auth_enabled, require_admin, require_auth
|
||||
from .mfa import (
|
||||
generate_qr_uri,
|
||||
generate_recovery_codes,
|
||||
generate_secret,
|
||||
hash_recovery_code,
|
||||
verify_recovery_code,
|
||||
verify_totp,
|
||||
)
|
||||
from .password import hash_password, verify_password
|
||||
from .user_store import (
|
||||
create_user,
|
||||
@@ -136,16 +144,31 @@ async def login(body: LoginRequest, response: Response, request: Request):
|
||||
detail += f" ({remaining} tentative(s) restante(s))"
|
||||
raise HTTPException(401, detail)
|
||||
|
||||
# Success — clear rate limits and generate tokens
|
||||
record_login_success(body.username)
|
||||
# Success — clear rate limits
|
||||
rl_record_success(client_ip)
|
||||
|
||||
# If MFA is enabled, don't issue token yet — require TOTP verification
|
||||
if user.get("mfa_enabled") and user.get("mfa_secret"):
|
||||
logger.info(f"User '{body.username}' login deferred — MFA required")
|
||||
return {
|
||||
"mfa_required": True,
|
||||
"mfa_method": "totp",
|
||||
"username": body.username,
|
||||
"remember_me": body.remember_me,
|
||||
}
|
||||
|
||||
return _issue_tokens(user, body.username, body.remember_me, response)
|
||||
|
||||
|
||||
def _issue_tokens(user: dict, username: str, remember_me: bool, response: Response) -> dict:
|
||||
"""Issue JWT tokens after successful authentication (password or MFA verified)."""
|
||||
record_login_success(username)
|
||||
|
||||
access_token = create_access_token(user)
|
||||
refresh_token, refresh_jti = create_refresh_token(body.username)
|
||||
refresh_token, refresh_jti = create_refresh_token(username)
|
||||
|
||||
# Set refresh token as HttpOnly cookie (path-restricted to /api/auth/refresh)
|
||||
max_age = 2592000 if body.remember_me else 604800 # 30d or 7d
|
||||
import os
|
||||
max_age = 2592000 if remember_me else 604800 # 30d or 7d
|
||||
secure = os.environ.get("OBSIGATE_SECURE_COOKIES", "false").lower() == "true"
|
||||
response.set_cookie(
|
||||
key="refresh_token",
|
||||
@@ -156,10 +179,7 @@ async def login(body: LoginRequest, response: Response, request: Request):
|
||||
secure=secure,
|
||||
path="/api/auth/refresh",
|
||||
)
|
||||
|
||||
logger.info(f"User '{body.username}' logged in")
|
||||
|
||||
# Set access token as cookie for same-origin requests (e.g. popout window)
|
||||
logger.info(f"User '{username}' logged in")
|
||||
response.set_cookie(
|
||||
key="access_token",
|
||||
value=access_token,
|
||||
@@ -169,7 +189,6 @@ async def login(body: LoginRequest, response: Response, request: Request):
|
||||
secure=secure,
|
||||
path="/",
|
||||
)
|
||||
|
||||
return {
|
||||
"access_token": access_token,
|
||||
"token_type": "bearer", # nosec B105 — OAuth2 token_type, pas un mot de passe
|
||||
@@ -304,6 +323,192 @@ async def change_password(
|
||||
return {"message": "Mot de passe mis à jour"}
|
||||
|
||||
|
||||
# ── MFA endpoints ────────────────────────────────────────────────────
|
||||
|
||||
class MfaVerifyRequest(BaseModel):
|
||||
username: str
|
||||
code: str
|
||||
remember_me: bool = False
|
||||
|
||||
|
||||
class MfaRecoveryRequest(BaseModel):
|
||||
username: str
|
||||
recovery_code: str
|
||||
|
||||
|
||||
class MfaDisableRequest(BaseModel):
|
||||
password: str
|
||||
code: str
|
||||
|
||||
|
||||
class MfaEnableRequest(BaseModel):
|
||||
code: str
|
||||
|
||||
|
||||
@router.post("/mfa/totp/setup")
|
||||
async def mfa_totp_setup(current_user=Depends(require_auth)):
|
||||
"""Generate a TOTP secret and QR URI for MFA setup.
|
||||
|
||||
Returns the secret and otpauth URI — client displays QR code.
|
||||
Does NOT enable MFA yet; call /mfa/totp/enable after first successful verify.
|
||||
"""
|
||||
from .user_store import update_user
|
||||
secret = generate_secret()
|
||||
qr_uri = generate_qr_uri(secret, current_user["username"])
|
||||
# Store secret temporarily (not yet enabled)
|
||||
update_user(current_user["username"], {
|
||||
"mfa_secret_pending": secret,
|
||||
})
|
||||
return {
|
||||
"secret": secret,
|
||||
"qr_uri": qr_uri,
|
||||
"otpauth_uri": qr_uri,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/mfa/totp/enable")
|
||||
async def mfa_totp_enable(
|
||||
req: MfaEnableRequest,
|
||||
current_user=Depends(require_auth),
|
||||
):
|
||||
"""Enable MFA after verifying the first TOTP code.
|
||||
|
||||
On success: generates recovery codes, enables MFA, returns recovery codes.
|
||||
"""
|
||||
from .user_store import get_user, update_user
|
||||
|
||||
user = get_user(current_user["username"])
|
||||
secret = user.get("mfa_secret_pending")
|
||||
if not secret:
|
||||
raise HTTPException(400, "Aucune configuration MFA en cours. Commencez par /mfa/totp/setup")
|
||||
|
||||
if not verify_totp(secret, req.code):
|
||||
raise HTTPException(400, "Code TOTP invalide")
|
||||
|
||||
# Generate recovery codes
|
||||
recovery_codes = generate_recovery_codes()
|
||||
hashed_codes = [hash_recovery_code(c) for c in recovery_codes]
|
||||
|
||||
# Enable MFA
|
||||
update_user(current_user["username"], {
|
||||
"mfa_enabled": True,
|
||||
"mfa_secret": secret,
|
||||
"mfa_method": "totp",
|
||||
"mfa_recovery_codes": hashed_codes,
|
||||
"mfa_secret_pending": None, # clear pending
|
||||
})
|
||||
|
||||
logger.info(f"MFA enabled for user '{current_user['username']}'")
|
||||
return {
|
||||
"mfa_enabled": True,
|
||||
"recovery_codes": recovery_codes, # shown once, client must display/save
|
||||
}
|
||||
|
||||
|
||||
@router.post("/mfa/totp/disable")
|
||||
async def mfa_totp_disable(
|
||||
req: MfaDisableRequest,
|
||||
current_user=Depends(require_auth),
|
||||
):
|
||||
"""Disable MFA. Requires current password + valid TOTP code."""
|
||||
from .user_store import get_user, update_user
|
||||
|
||||
user = get_user(current_user["username"])
|
||||
if not user.get("mfa_enabled"):
|
||||
raise HTTPException(400, "MFA non activé")
|
||||
|
||||
if not verify_password(req.password, user["password_hash"]):
|
||||
raise HTTPException(400, "Mot de passe incorrect")
|
||||
|
||||
if not verify_totp(user["mfa_secret"], req.code):
|
||||
raise HTTPException(400, "Code TOTP invalide")
|
||||
|
||||
update_user(current_user["username"], {
|
||||
"mfa_enabled": False,
|
||||
"mfa_secret": None,
|
||||
"mfa_method": None,
|
||||
"mfa_recovery_codes": [],
|
||||
})
|
||||
|
||||
logger.info(f"MFA disabled for user '{current_user['username']}'")
|
||||
return {"mfa_enabled": False}
|
||||
|
||||
|
||||
@router.get("/mfa/status")
|
||||
async def mfa_status(current_user=Depends(require_auth)):
|
||||
"""Return current user's MFA status."""
|
||||
from .user_store import get_user
|
||||
user = get_user(current_user["username"])
|
||||
return {
|
||||
"mfa_enabled": user.get("mfa_enabled", False),
|
||||
"mfa_method": user.get("mfa_method"),
|
||||
}
|
||||
|
||||
|
||||
@router.post("/mfa/totp/verify")
|
||||
async def mfa_totp_verify(body: MfaVerifyRequest, response: Response, request: Request):
|
||||
"""Verify TOTP code during login (second factor).
|
||||
|
||||
Called after login returns mfa_required=true.
|
||||
On success: issues JWT tokens.
|
||||
"""
|
||||
from .user_store import get_user
|
||||
|
||||
user = get_user(body.username)
|
||||
if not user:
|
||||
# Timing-safe: simulate work
|
||||
hash_password("dummy_timing_protection")
|
||||
raise HTTPException(401, "Identifiants invalides")
|
||||
|
||||
if not user.get("mfa_enabled") or not user.get("mfa_secret"):
|
||||
raise HTTPException(400, "MFA non activé pour cet utilisateur")
|
||||
|
||||
if not verify_totp(user["mfa_secret"], body.code):
|
||||
raise HTTPException(401, "Code TOTP invalide")
|
||||
|
||||
# Clear IP rate limit on success
|
||||
client_ip = request.client.host if request.client else "unknown"
|
||||
rl_record_success(client_ip)
|
||||
|
||||
return _issue_tokens(user, body.username, body.remember_me, response)
|
||||
|
||||
|
||||
@router.post("/mfa/recovery")
|
||||
async def mfa_recovery_login(body: MfaRecoveryRequest, response: Response, request: Request):
|
||||
"""Login with a recovery code (when TOTP device is unavailable).
|
||||
|
||||
Each recovery code is single-use.
|
||||
"""
|
||||
from .user_store import get_user, update_user
|
||||
|
||||
user = get_user(body.username)
|
||||
if not user:
|
||||
hash_password("dummy_timing_protection")
|
||||
raise HTTPException(401, "Identifiants invalides")
|
||||
|
||||
if not user.get("mfa_enabled"):
|
||||
raise HTTPException(400, "MFA non activé pour cet utilisateur")
|
||||
|
||||
hashed_codes = user.get("mfa_recovery_codes", [])
|
||||
if not hashed_codes:
|
||||
raise HTTPException(400, "Aucun code de récupération disponible")
|
||||
|
||||
idx = verify_recovery_code(body.recovery_code, hashed_codes)
|
||||
if idx is None:
|
||||
raise HTTPException(401, "Code de récupération invalide")
|
||||
|
||||
# Remove used recovery code (single-use)
|
||||
hashed_codes.pop(idx)
|
||||
update_user(body.username, {"mfa_recovery_codes": hashed_codes})
|
||||
|
||||
# Clear IP rate limit
|
||||
client_ip = request.client.host if request.client else "unknown"
|
||||
rl_record_success(client_ip)
|
||||
|
||||
logger.info(f"User '{body.username}' logged in via recovery code")
|
||||
return _issue_tokens(user, body.username, False, response)
|
||||
|
||||
|
||||
# ── Admin endpoints ───────────────────────────────────────────────────
|
||||
|
||||
@router.get("/admin/users")
|
||||
|
||||
Reference in New Issue
Block a user