1036 lines
37 KiB
Python
1036 lines
37 KiB
Python
# backend/auth/router.py
|
|
# All /api/auth/* endpoints: login, logout, refresh, me, change-password,
|
|
# and admin user CRUD.
|
|
|
|
import base64
|
|
import binascii
|
|
import logging
|
|
import os
|
|
import re
|
|
|
|
from fastapi import APIRouter, Body, Depends, HTTPException, Request, Response
|
|
from pydantic import BaseModel, validator
|
|
|
|
from backend.ratelimit import is_account_rate_limited, is_rate_limited
|
|
from backend.ratelimit import record_account_failure as rl_record_account_failure
|
|
from backend.ratelimit import record_account_success as rl_record_account_success
|
|
from backend.ratelimit import record_failure as rl_record_failure
|
|
from backend.ratelimit import record_success as rl_record_success
|
|
from backend.services.net import get_client_ip, is_trusted_proxy
|
|
|
|
from .jwt_handler import (
|
|
ACCESS_TOKEN_EXPIRE_SECONDS,
|
|
API_TOKEN_EXPIRY_CHOICES,
|
|
create_access_token,
|
|
create_api_token,
|
|
create_refresh_token,
|
|
decode_token,
|
|
delete_api_token,
|
|
is_token_revoked,
|
|
list_api_tokens,
|
|
revoke_token,
|
|
)
|
|
from .mfa import (
|
|
generate_qr_uri,
|
|
generate_recovery_codes,
|
|
generate_secret,
|
|
hash_recovery_code,
|
|
verify_recovery_code,
|
|
verify_totp,
|
|
)
|
|
from .middleware import is_auth_enabled, require_admin, require_auth
|
|
from .password import hash_password, validate_password_strength, verify_password
|
|
from .user_store import (
|
|
create_user,
|
|
delete_user,
|
|
get_all_users,
|
|
get_user,
|
|
has_users,
|
|
is_locked,
|
|
record_login_failure,
|
|
record_login_success,
|
|
update_user,
|
|
)
|
|
|
|
logger = logging.getLogger("obsigate.auth.router")
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
|
|
def is_secure_cookies(request: Request | None = None) -> bool:
|
|
"""True when auth cookies must carry the ``Secure`` flag (#87 T3/T8).
|
|
|
|
``OBSIGATE_SECURE_COOKIES=true|false|auto`` (défaut : ``auto``) :
|
|
``true``/``false`` forcent le comportement ; ``auto`` met ``Secure``
|
|
si la requête arrive en https (production derrière TLS) et l'omet
|
|
sinon (dev local en http — les navigateurs jettent les cookies
|
|
``Secure`` sur http, ce qui casserait silencieusement les logins
|
|
localhost). Derrière un reverse proxy qui termine TLS, le schéma perçu
|
|
est http : avec ``OBSIGATE_TRUST_PROXY=true``, ``X-Forwarded-Proto``
|
|
est honoré (même garde que ``get_client_ip``, BUG-030).
|
|
"""
|
|
forced = os.environ.get("OBSIGATE_SECURE_COOKIES", "auto").lower()
|
|
if forced in ("1", "true", "yes", "on"):
|
|
return True
|
|
if forced in ("0", "false", "no", "off"):
|
|
return False
|
|
if request is None:
|
|
return False
|
|
if request.url.scheme == "https":
|
|
return True
|
|
if is_trusted_proxy():
|
|
proto = request.headers.get("x-forwarded-proto", "").split(",")[0].strip().lower()
|
|
if proto == "https":
|
|
return True
|
|
return False
|
|
|
|
|
|
# ── Pydantic request models ──────────────────────────────────────────
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
remember_me: bool = False # True → refresh token 30d instead of 7d
|
|
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
current_password: str
|
|
new_password: str
|
|
|
|
@validator("new_password")
|
|
def password_strength(cls, v):
|
|
return validate_password_strength(v)
|
|
|
|
|
|
class CreateUserRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
display_name: str | None = None
|
|
role: str = "user"
|
|
vaults: list[str] = []
|
|
|
|
@validator("password")
|
|
def password_valid(cls, v):
|
|
return validate_password_strength(v)
|
|
|
|
@validator("username")
|
|
def username_valid(cls, v):
|
|
if not re.match(r"^[a-zA-Z0-9_-]{2,32}$", v):
|
|
raise ValueError("2-32 caractères alphanumériques, _ ou -")
|
|
return v.lower()
|
|
|
|
@validator("role")
|
|
def role_valid(cls, v):
|
|
if v not in ("admin", "user"):
|
|
raise ValueError("Rôle invalide")
|
|
return v
|
|
|
|
|
|
class UpdateUserRequest(BaseModel):
|
|
display_name: str | None = None
|
|
vaults: list[str] | None = None
|
|
active: bool | None = None
|
|
password: str | None = None
|
|
role: str | None = None
|
|
|
|
@validator("password")
|
|
def password_valid(cls, v):
|
|
if v is None:
|
|
return v
|
|
return validate_password_strength(v)
|
|
|
|
|
|
# ── Profile avatar (#113) ───────────────────────────────────────────
|
|
|
|
#: Avatar data-URL pattern — PNG/JPEG/WebP only (no SVG: XSS surface).
|
|
_AVATAR_DATA_URL_RE = re.compile(
|
|
r"^data:image/(?:png|jpeg|webp);base64,[A-Za-z0-9+/]+={0,2}$"
|
|
)
|
|
#: ~300 KB of base64 payload (a 256px JPEG is ~15 KB; generous headroom).
|
|
_AVATAR_MAX_CHARS = 400_000
|
|
|
|
|
|
def _validate_avatar(data_url: str) -> str | None:
|
|
"""Validate an avatar data-URL for storage on the user profile.
|
|
|
|
Returns the normalized data-URL, or ``None`` when clearing the avatar
|
|
(empty string). Raises ``HTTPException(400)`` on anything else.
|
|
"""
|
|
if data_url == "":
|
|
return None
|
|
if len(data_url) > _AVATAR_MAX_CHARS:
|
|
raise HTTPException(400, "Avatar image too large")
|
|
if not _AVATAR_DATA_URL_RE.match(data_url):
|
|
raise HTTPException(400, "Avatar must be a PNG, JPEG or WebP data URL")
|
|
try:
|
|
raw = base64.b64decode(data_url.split(",", 1)[1], validate=True)
|
|
except (ValueError, binascii.Error) as exc: # pragma: no cover — regex guards
|
|
raise HTTPException(400, "Avatar payload is not valid base64") from exc
|
|
# Confirm the decoded bytes really are a supported image (magic numbers).
|
|
is_png = raw.startswith(b"\x89PNG\r\n\x1a\n")
|
|
is_jpeg = raw.startswith(b"\xff\xd8\xff")
|
|
is_webp = (
|
|
len(raw) >= 12 and raw[:4] == b"RIFF" and raw[8:12] == b"WEBP"
|
|
)
|
|
if not (is_png or is_jpeg or is_webp):
|
|
raise HTTPException(400, "Avatar payload is not a PNG, JPEG or WebP image")
|
|
return data_url
|
|
|
|
|
|
# ── Public endpoints ──────────────────────────────────────────────────
|
|
|
|
@router.get("/status")
|
|
async def auth_status():
|
|
"""Public endpoint: returns whether auth is enabled.
|
|
|
|
The frontend uses this to decide whether to show the login screen.
|
|
Also returns whether any users exist (for first-run detection).
|
|
"""
|
|
return {
|
|
"auth_enabled": is_auth_enabled(),
|
|
"has_users": has_users(),
|
|
}
|
|
|
|
|
|
@router.post("/login")
|
|
async def login(body: LoginRequest, response: Response, request: Request):
|
|
"""Authenticate a user. Returns access token and sets refresh cookie.
|
|
|
|
Implements timing-safe responses to prevent user enumeration: a failed
|
|
login with an unknown user takes the same time as one with a known user
|
|
(dummy hash is computed). BUG-039: unknown, inactive, locked and
|
|
per-account rate-limited accounts all answer the same ``401`` so the HTTP
|
|
status can never reveal whether an account exists.
|
|
"""
|
|
client_ip = get_client_ip(request)
|
|
|
|
# IP-based rate limiting (10 failures / 15 min per IP). It is not
|
|
# account-specific, so a 429 here cannot be used to enumerate accounts.
|
|
if is_rate_limited(client_ip):
|
|
raise HTTPException(429, "Trop de tentatives depuis cette adresse IP (15min)")
|
|
|
|
user = get_user(body.username)
|
|
|
|
# BUG-039: uniform 401 + equivalent timing for every account-state outcome.
|
|
if not user or not user.get("active"):
|
|
# Timing-safe: simulate hash computation to prevent user enumeration
|
|
hash_password("dummy_timing_protection")
|
|
raise HTTPException(401, "Identifiants invalides")
|
|
|
|
# BUG-031: per-account budget still applies when the attacker rotates IPs.
|
|
# Kept indistinguishable from a wrong password (BUG-039).
|
|
if is_account_rate_limited(body.username) or is_locked(body.username):
|
|
hash_password("dummy_timing_protection")
|
|
raise HTTPException(401, "Identifiants invalides")
|
|
|
|
if not verify_password(body.password, user["password_hash"]):
|
|
attempts = record_login_failure(body.username)
|
|
rl_attempts, rl_remaining = rl_record_failure(client_ip)
|
|
rl_record_account_failure(body.username)
|
|
remaining = max(0, 5 - attempts)
|
|
detail = "Identifiants invalides"
|
|
if 0 < remaining <= 2:
|
|
detail += f" ({remaining} tentative(s) restante(s))"
|
|
raise HTTPException(401, detail)
|
|
|
|
# Success — clear rate limits
|
|
rl_record_success(client_ip)
|
|
|
|
# If MFA is enabled, don't issue token yet — require second factor
|
|
if user.get("mfa_enabled"):
|
|
method = "totp" if user.get("mfa_secret") else _preferred_mfa_method(user)
|
|
logger.info(f"User '{body.username}' login deferred — MFA required ({method})")
|
|
return {
|
|
"mfa_required": True,
|
|
"mfa_method": method,
|
|
"username": body.username,
|
|
"remember_me": body.remember_me,
|
|
}
|
|
|
|
return _issue_tokens(user, body.username, body.remember_me, response, request)
|
|
|
|
|
|
def _issue_tokens(user: dict, username: str, remember_me: bool, response: Response,
|
|
request: Request | None = None) -> dict:
|
|
"""Issue JWT tokens after successful authentication (password or MFA verified)."""
|
|
record_login_success(username)
|
|
rl_record_account_success(username)
|
|
|
|
access_token = create_access_token(user)
|
|
refresh_token, refresh_jti = create_refresh_token(username, remember=remember_me)
|
|
|
|
max_age = 2592000 if remember_me else 604800 # 30d or 7d
|
|
secure = is_secure_cookies(request)
|
|
response.set_cookie(
|
|
key="refresh_token",
|
|
value=refresh_token,
|
|
max_age=max_age,
|
|
httponly=True,
|
|
samesite="strict",
|
|
secure=secure,
|
|
path="/api/auth/refresh",
|
|
)
|
|
logger.info(f"User '{username}' logged in")
|
|
response.set_cookie(
|
|
key="access_token",
|
|
value=access_token,
|
|
max_age=ACCESS_TOKEN_EXPIRE_SECONDS,
|
|
httponly=True,
|
|
samesite="lax",
|
|
secure=secure,
|
|
path="/",
|
|
)
|
|
return {
|
|
"access_token": access_token,
|
|
# OAuth2 token_type, pas un mot de passe (B105) :
|
|
"token_type": "bearer", # nosec B105
|
|
"expires_in": ACCESS_TOKEN_EXPIRE_SECONDS,
|
|
"user": {
|
|
"username": user["username"],
|
|
"display_name": user["display_name"],
|
|
"role": user["role"],
|
|
"vaults": user["vaults"],
|
|
"avatar": user.get("avatar"),
|
|
},
|
|
}
|
|
|
|
|
|
@router.post("/refresh")
|
|
async def refresh_token_endpoint(request: Request, response: Response):
|
|
"""Renew access token via refresh token cookie.
|
|
|
|
Called automatically by the frontend when the access token expires.
|
|
The refresh token is rotated on every use (BUG-027) and rejected if it
|
|
predates the user's last password change (BUG-028).
|
|
"""
|
|
refresh_tok = request.cookies.get("refresh_token")
|
|
if not refresh_tok:
|
|
raise HTTPException(401, "Refresh token manquant")
|
|
|
|
payload = decode_token(refresh_tok)
|
|
if not payload or payload.get("type") != "refresh":
|
|
raise HTTPException(401, "Refresh token invalide")
|
|
|
|
if is_token_revoked(payload["jti"]):
|
|
raise HTTPException(401, "Session expirée, veuillez vous reconnecter")
|
|
|
|
user = get_user(payload["sub"])
|
|
if not user or not user.get("active"):
|
|
raise HTTPException(401, "Utilisateur introuvable ou inactif")
|
|
|
|
# BUG-028: reject refresh tokens issued before the last password change.
|
|
pca = user.get("password_changed_at")
|
|
iat = payload.get("iat")
|
|
if pca is not None and iat is not None:
|
|
try:
|
|
stale = int(iat) < int(float(pca))
|
|
except (TypeError, ValueError):
|
|
stale = True
|
|
if stale:
|
|
raise HTTPException(401, "Session expirée, veuillez vous reconnecter")
|
|
|
|
secure = is_secure_cookies(request)
|
|
remember_me = bool(payload.get("remember", False))
|
|
|
|
# BUG-027: rotate the refresh token — the old one is now single-use.
|
|
revoke_token(payload["jti"])
|
|
new_refresh_token, _new_jti = create_refresh_token(user["username"], remember=remember_me)
|
|
max_age = 2592000 if remember_me else 604800
|
|
response.set_cookie(
|
|
key="refresh_token",
|
|
value=new_refresh_token,
|
|
max_age=max_age,
|
|
httponly=True,
|
|
samesite="strict",
|
|
secure=secure,
|
|
path="/api/auth/refresh",
|
|
)
|
|
|
|
new_access_token = create_access_token(user)
|
|
|
|
response.set_cookie(
|
|
key="access_token",
|
|
value=new_access_token,
|
|
max_age=ACCESS_TOKEN_EXPIRE_SECONDS,
|
|
httponly=True,
|
|
samesite="lax",
|
|
secure=secure,
|
|
path="/",
|
|
)
|
|
|
|
return {
|
|
"access_token": new_access_token,
|
|
# OAuth2 token_type, pas un mot de passe (B105) :
|
|
"token_type": "bearer", # nosec B105
|
|
"expires_in": ACCESS_TOKEN_EXPIRE_SECONDS,
|
|
}
|
|
|
|
|
|
@router.post("/logout")
|
|
async def logout(
|
|
request: Request,
|
|
response: Response,
|
|
):
|
|
"""Logout: revoke refresh and access tokens, then delete cookies."""
|
|
refresh_tok = request.cookies.get("refresh_token")
|
|
if refresh_tok:
|
|
payload = decode_token(refresh_tok)
|
|
if payload:
|
|
try:
|
|
revoke_token(payload["jti"])
|
|
except Exception:
|
|
pass # token already revoked
|
|
|
|
# BUG-027: revoke the access token too, otherwise it stays valid until expiry.
|
|
access_tok = None
|
|
auth_header = request.headers.get("authorization", "")
|
|
if auth_header.lower().startswith("bearer "):
|
|
access_tok = auth_header[7:].strip()
|
|
if not access_tok:
|
|
access_tok = request.cookies.get("access_token")
|
|
if access_tok:
|
|
access_payload = decode_token(access_tok)
|
|
if access_payload and access_payload.get("type") == "access":
|
|
try:
|
|
revoke_token(access_payload["jti"])
|
|
except Exception:
|
|
pass
|
|
|
|
response.delete_cookie("refresh_token", path="/api/auth/refresh")
|
|
response.delete_cookie("access_token", path="/")
|
|
response.delete_cookie("access_token", path="/api") # just in case
|
|
return {"message": "Deconnecte avec succes"}
|
|
|
|
|
|
@router.get("/me")
|
|
async def get_me(current_user=Depends(require_auth)):
|
|
"""Return current authenticated user info."""
|
|
return {
|
|
"username": current_user["username"],
|
|
"display_name": current_user["display_name"],
|
|
"role": current_user["role"],
|
|
"vaults": current_user["vaults"],
|
|
"language": current_user.get("language", "fr"),
|
|
"last_login": current_user.get("last_login"),
|
|
"avatar": current_user.get("avatar"),
|
|
}
|
|
|
|
|
|
class UpdateMeRequest(BaseModel):
|
|
"""Fields the user can update on their own profile."""
|
|
display_name: str | None = None
|
|
language: str | None = None
|
|
#: Image data-URL (PNG/JPEG/WebP), or ``""`` to remove the avatar (#113).
|
|
avatar: str | None = None
|
|
|
|
|
|
@router.patch("/me")
|
|
async def patch_me(req: UpdateMeRequest, current_user=Depends(require_auth)):
|
|
"""Update current user's profile fields (display_name, language, avatar)."""
|
|
from .user_store import update_user
|
|
updates: dict[str, object] = {}
|
|
if req.display_name is not None:
|
|
updates["display_name"] = req.display_name
|
|
if req.language is not None:
|
|
if req.language not in ("fr", "en"):
|
|
raise HTTPException(400, "language must be 'fr' or 'en'")
|
|
updates["language"] = req.language
|
|
if req.avatar is not None:
|
|
updates["avatar"] = _validate_avatar(req.avatar)
|
|
if not updates:
|
|
raise HTTPException(400, "No fields to update")
|
|
updated = update_user(current_user["username"], updates)
|
|
return {
|
|
"username": updated["username"],
|
|
"display_name": updated["display_name"],
|
|
"role": updated["role"],
|
|
"vaults": updated["vaults"],
|
|
"language": updated.get("language", "fr"),
|
|
"last_login": updated.get("last_login"),
|
|
"avatar": updated.get("avatar"),
|
|
}
|
|
|
|
|
|
@router.post("/change-password")
|
|
async def change_password(
|
|
req: ChangePasswordRequest,
|
|
response: Response,
|
|
request: Request,
|
|
current_user=Depends(require_auth),
|
|
):
|
|
"""Change own password.
|
|
|
|
BUG-028: changing the password invalidates all previously issued tokens;
|
|
a fresh pair is issued to keep the current session alive.
|
|
"""
|
|
user = get_user(current_user["username"])
|
|
assert user is not None, f"User {current_user['username']} not found"
|
|
if not verify_password(req.current_password, user["password_hash"]):
|
|
raise HTTPException(400, "Mot de passe actuel incorrect")
|
|
update_user(current_user["username"], {"password": req.new_password})
|
|
updated = get_user(current_user["username"])
|
|
result: dict = {"message": "Mot de passe mis à jour"}
|
|
if updated is not None:
|
|
result.update(_issue_tokens(updated, updated["username"], False, response, request))
|
|
return result
|
|
|
|
|
|
# ── MFA endpoints ────────────────────────────────────────────────────
|
|
|
|
def _enforce_mfa_rate_limit(request: Request, username: str) -> str:
|
|
"""Reject MFA attempts from a rate-limited IP or on a locked account.
|
|
|
|
BUG-023: the second-factor endpoints were previously unprotected, making
|
|
the 6-digit TOTP brute-forceable. Returns the resolved client IP.
|
|
"""
|
|
client_ip = get_client_ip(request)
|
|
if is_rate_limited(client_ip):
|
|
raise HTTPException(429, "Trop de tentatives depuis cette adresse IP (15min)")
|
|
if is_account_rate_limited(username):
|
|
raise HTTPException(429, "Trop de tentatives sur ce compte (15min)")
|
|
if is_locked(username):
|
|
raise HTTPException(429, "Compte temporairement verrouillé (15min)")
|
|
return client_ip
|
|
|
|
|
|
def _record_mfa_failure(client_ip: str, username: str) -> None:
|
|
"""Record a failed MFA attempt for the IP, the account and the lockout."""
|
|
record_login_failure(username)
|
|
rl_record_failure(client_ip)
|
|
rl_record_account_failure(username)
|
|
|
|
|
|
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, the otpauth URI and a ready-to-display QR code
|
|
(`qr_data_url`, SVG `data:` URI — no third-party service, CSP-safe).
|
|
|
|
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,
|
|
})
|
|
# BUG-068: the QR code is generated locally (segno, stdlib-free SVG data
|
|
# URI). The previous client-side https://api.qrserver.com image was blocked
|
|
# by the CSP (img-src 'self' data: blob:) and leaked the otpauth URI —
|
|
# including the TOTP secret — to a third party.
|
|
qr_data_url: str | None = None
|
|
try:
|
|
import segno
|
|
qr_data_url = segno.make(qr_uri).svg_data_uri(scale=5)
|
|
except Exception:
|
|
qr_data_url = None
|
|
return {
|
|
"secret": secret,
|
|
"qr_uri": qr_uri,
|
|
"otpauth_uri": qr_uri,
|
|
"qr_data_url": qr_data_url,
|
|
}
|
|
|
|
|
|
@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"])
|
|
if user is None:
|
|
raise HTTPException(404, "Utilisateur introuvable")
|
|
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 user is None:
|
|
raise HTTPException(404, "Utilisateur introuvable")
|
|
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}
|
|
|
|
|
|
# ── WebAuthn endpoints (ROADMAP #64) ─────────────────────────────────
|
|
|
|
def _preferred_mfa_method(user: dict) -> str:
|
|
"""Which second factor to offer at login: webauthn when keys exist, else totp."""
|
|
if user.get("webauthn_credentials"):
|
|
return "webauthn"
|
|
return "totp"
|
|
|
|
|
|
class WebauthnRegisterRequest(BaseModel):
|
|
credential: dict
|
|
label: str = ""
|
|
|
|
|
|
class WebauthnVerifyRequest(BaseModel):
|
|
username: str
|
|
credential: dict
|
|
remember_me: bool = False
|
|
|
|
|
|
class WebauthnRemoveRequest(BaseModel):
|
|
credential_id: str
|
|
password: str
|
|
|
|
|
|
@router.post("/mfa/webauthn/register/options")
|
|
async def mfa_webauthn_register_options(request: Request,
|
|
current_user=Depends(require_auth)):
|
|
"""Start WebAuthn key enrolment — returns publicKey creation options for the browser."""
|
|
from .webauthn_mfa import begin_registration, resolve_relying_party
|
|
|
|
# BUG-070: rp_id/origins derive from the request (exact host incl. port)
|
|
# unless explicitly configured — the old localhost defaults rejected
|
|
# every real access URL ("Unexpected client data origin").
|
|
rp, _ = resolve_relying_party(request)
|
|
options = begin_registration(current_user["username"],
|
|
current_user.get("display_name", ""),
|
|
rp_id_override=rp)
|
|
return {"options": options}
|
|
|
|
|
|
@router.post("/mfa/webauthn/register")
|
|
async def mfa_webauthn_register(
|
|
req: WebauthnRegisterRequest,
|
|
request: Request,
|
|
current_user=Depends(require_auth),
|
|
):
|
|
"""Verify the created credential, store it, and enable MFA if not already on.
|
|
|
|
Returns recovery codes when MFA is newly enabled (they were never issued).
|
|
"""
|
|
from datetime import datetime, timezone
|
|
|
|
from .user_store import get_user, update_user
|
|
from .webauthn_mfa import complete_registration, resolve_relying_party
|
|
|
|
user = get_user(current_user["username"])
|
|
if user is None:
|
|
raise HTTPException(404, "Utilisateur introuvable")
|
|
rp, origins = resolve_relying_party(request)
|
|
try:
|
|
record = complete_registration(current_user["username"], req.credential,
|
|
label=req.label,
|
|
rp_id_override=rp,
|
|
origins_override=origins)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e))
|
|
except Exception as e:
|
|
logger.warning(f"WebAuthn registration failed for {current_user['username']}: {e}")
|
|
raise HTTPException(400, "Validation du credential WebAuthn échouée")
|
|
|
|
record["registered_at"] = datetime.now(timezone.utc).isoformat()
|
|
creds = list(user.get("webauthn_credentials", []))
|
|
creds = [c for c in creds if c.get("credential_id") != record["credential_id"]]
|
|
creds.append(record)
|
|
|
|
updates: dict = {"webauthn_credentials": creds}
|
|
issued_recovery: list[str] = []
|
|
if not user.get("mfa_enabled"):
|
|
issued_recovery = generate_recovery_codes()
|
|
updates.update({
|
|
"mfa_enabled": True,
|
|
"mfa_method": "webauthn",
|
|
"mfa_recovery_codes": [hash_recovery_code(c) for c in issued_recovery],
|
|
})
|
|
update_user(current_user["username"], updates)
|
|
|
|
logger.info(f"WebAuthn credential registered for user '{current_user['username']}' "
|
|
f"({record['label']})")
|
|
return {
|
|
"ok": True,
|
|
"credentials": user_credentials_response(creds),
|
|
"mfa_enabled": True,
|
|
"recovery_codes": issued_recovery,
|
|
}
|
|
|
|
|
|
def user_credentials_response(creds: list[dict]) -> list[dict]:
|
|
from .webauthn_mfa import credentials_for_api
|
|
return credentials_for_api(creds)
|
|
|
|
|
|
@router.get("/mfa/webauthn/credentials")
|
|
async def mfa_webauthn_list(current_user=Depends(require_auth)):
|
|
from .user_store import get_user
|
|
user = get_user(current_user["username"])
|
|
if user is None:
|
|
raise HTTPException(404, "Utilisateur introuvable")
|
|
return {"credentials": user_credentials_response(user.get("webauthn_credentials", []))}
|
|
|
|
|
|
@router.post("/mfa/webauthn/credentials/remove")
|
|
async def mfa_webauthn_remove(
|
|
req: WebauthnRemoveRequest,
|
|
current_user=Depends(require_auth),
|
|
):
|
|
"""Remove a WebAuthn key. Requires password. Disables MFA if no second factor remains."""
|
|
from .user_store import get_user, update_user
|
|
from .webauthn_mfa import clear_pending
|
|
|
|
user = get_user(current_user["username"])
|
|
if user is None:
|
|
raise HTTPException(404, "Utilisateur introuvable")
|
|
if not verify_password(req.password, user["password_hash"]):
|
|
raise HTTPException(400, "Mot de passe incorrect")
|
|
|
|
creds = [c for c in user.get("webauthn_credentials", [])
|
|
if c.get("credential_id") != req.credential_id]
|
|
if len(creds) == len(user.get("webauthn_credentials", [])):
|
|
raise HTTPException(404, "Credential inconnu")
|
|
|
|
updates: dict = {"webauthn_credentials": creds}
|
|
if not creds and not user.get("mfa_secret"):
|
|
updates.update({"mfa_enabled": False, "mfa_method": None, "mfa_recovery_codes": []})
|
|
elif not creds and user.get("mfa_secret"):
|
|
updates["mfa_method"] = "totp"
|
|
update_user(current_user["username"], updates)
|
|
clear_pending(current_user["username"])
|
|
return {"ok": True, "credentials": user_credentials_response(creds),
|
|
"mfa_enabled": bool(updates.get("mfa_enabled", user.get("mfa_enabled"))) and bool(creds or user.get("mfa_secret"))}
|
|
|
|
|
|
@router.post("/mfa/webauthn/options")
|
|
async def mfa_webauthn_login_options(request: Request, body: dict = Body(...)):
|
|
"""Unauthenticated: begin the login assertion for a user with registered keys.
|
|
|
|
Enumeration-safe: always 200 — returns null options (caller falls back to
|
|
TOTP/recovery UI) when the user has no WebAuthn key or MFA is off.
|
|
"""
|
|
username = str(body.get("username", ""))
|
|
user = get_user(username)
|
|
creds = (user or {}).get("webauthn_credentials", [])
|
|
if not user or not user.get("mfa_enabled") or not creds:
|
|
return {"mfa_method": "totp", "options": None}
|
|
|
|
from .webauthn_mfa import begin_authentication, resolve_relying_party
|
|
rp, _ = resolve_relying_party(request)
|
|
options = begin_authentication(username, creds, rp_id_override=rp)
|
|
if options is None:
|
|
return {"mfa_method": "totp", "options": None}
|
|
return {"mfa_method": "webauthn", "options": options}
|
|
|
|
|
|
@router.post("/mfa/webauthn/verify")
|
|
async def mfa_webauthn_verify(
|
|
body: WebauthnVerifyRequest,
|
|
response: Response,
|
|
request: Request,
|
|
):
|
|
"""Unauthenticated: verify the WebAuthn assertion and issue JWT tokens."""
|
|
from .user_store import get_user, update_user
|
|
from .webauthn_mfa import complete_authentication, resolve_relying_party
|
|
|
|
client_ip = _enforce_mfa_rate_limit(request, body.username)
|
|
|
|
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")
|
|
|
|
rp, origins = resolve_relying_party(request)
|
|
creds = user.get("webauthn_credentials", [])
|
|
try:
|
|
credential_id = body.credential.get("id", "")
|
|
stored = next((c for c in creds if c.get("credential_id") == credential_id), None)
|
|
if stored is None:
|
|
raise ValueError("Credential non enregistré")
|
|
new_count = complete_authentication(body.username, body.credential, stored,
|
|
rp_id_override=rp,
|
|
origins_override=origins)
|
|
except ValueError as e:
|
|
_record_mfa_failure(client_ip, body.username)
|
|
raise HTTPException(401, str(e))
|
|
except Exception as e:
|
|
_record_mfa_failure(client_ip, body.username)
|
|
logger.warning(f"WebAuthn verification failed for {body.username}: {e}")
|
|
raise HTTPException(401, "Vérification WebAuthn échouée")
|
|
|
|
updated = [dict(c) for c in creds]
|
|
for c in updated:
|
|
if c.get("credential_id") == body.credential.get("id"):
|
|
c["sign_count"] = new_count
|
|
update_user(body.username, {"webauthn_credentials": updated})
|
|
|
|
rl_record_success(client_ip)
|
|
logger.info(f"User '{body.username}' logged in via WebAuthn")
|
|
return _issue_tokens(user, body.username, body.remember_me, response, request)
|
|
|
|
|
|
@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"])
|
|
if user is None:
|
|
# BUG-081 : auth désactivée (OBSIGATE_AUTH_ENABLED=false) → le
|
|
# pseudo-user "anonymous" n'a aucune entrée en store : pas de MFA,
|
|
# et surtout pas de 500 (`AttributeError` sur `user.get`).
|
|
return {
|
|
"mfa_enabled": False,
|
|
"mfa_method": None,
|
|
"totp_enabled": False,
|
|
"webauthn_credentials": 0,
|
|
}
|
|
return {
|
|
"mfa_enabled": user.get("mfa_enabled", False),
|
|
"mfa_method": user.get("mfa_method"),
|
|
"totp_enabled": bool(user.get("mfa_secret")),
|
|
"webauthn_credentials": len(user.get("webauthn_credentials", [])),
|
|
}
|
|
|
|
|
|
@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
|
|
|
|
client_ip = _enforce_mfa_rate_limit(request, body.username)
|
|
|
|
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):
|
|
_record_mfa_failure(client_ip, body.username)
|
|
raise HTTPException(401, "Code TOTP invalide")
|
|
|
|
# Clear IP rate limit on success
|
|
rl_record_success(client_ip)
|
|
|
|
return _issue_tokens(user, body.username, body.remember_me, response, request)
|
|
|
|
|
|
@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
|
|
|
|
client_ip = _enforce_mfa_rate_limit(request, body.username)
|
|
|
|
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:
|
|
_record_mfa_failure(client_ip, body.username)
|
|
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
|
|
rl_record_success(client_ip)
|
|
|
|
logger.info(f"User '{body.username}' logged in via recovery code")
|
|
return _issue_tokens(user, body.username, False, response, request)
|
|
|
|
|
|
# ── Admin endpoints ───────────────────────────────────────────────────
|
|
|
|
@router.get("/admin/users")
|
|
async def list_users(admin=Depends(require_admin)):
|
|
"""List all users (admin only). Password hashes are never included."""
|
|
return get_all_users()
|
|
|
|
|
|
@router.post("/admin/users")
|
|
async def create_user_endpoint(
|
|
req: CreateUserRequest,
|
|
admin=Depends(require_admin),
|
|
):
|
|
"""Create a new user (admin only)."""
|
|
try:
|
|
user = create_user(
|
|
req.username, req.password, req.role, req.vaults, req.display_name
|
|
)
|
|
return user
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e))
|
|
|
|
|
|
@router.patch("/admin/users/{username}")
|
|
async def update_user_endpoint(
|
|
username: str,
|
|
req: UpdateUserRequest,
|
|
admin=Depends(require_admin),
|
|
):
|
|
"""Update a user (admin only)."""
|
|
updates = req.dict(exclude_none=True)
|
|
try:
|
|
return update_user(username, updates)
|
|
except ValueError as e:
|
|
raise HTTPException(404, str(e))
|
|
|
|
|
|
@router.delete("/admin/users/{username}")
|
|
async def delete_user_endpoint(
|
|
username: str,
|
|
admin=Depends(require_admin),
|
|
):
|
|
"""Delete a user (admin only). Cannot delete own account."""
|
|
if username == admin["username"]:
|
|
raise HTTPException(400, "Impossible de supprimer son propre compte")
|
|
try:
|
|
delete_user(username)
|
|
return {"message": f"Utilisateur '{username}' supprimé"}
|
|
except ValueError as e:
|
|
raise HTTPException(404, str(e))
|
|
|
|
|
|
# ── API / MCP tokens (feature #107) ──────────────────────────────────
|
|
# One long-lived token authenticates BOTH the REST API and the MCP
|
|
# endpoint (/mcp): the MCP server resolves the caller through the same
|
|
# get_current_user() dependency, so the same Bearer JWT works everywhere.
|
|
|
|
class CreateApiTokenRequest(BaseModel):
|
|
name: str
|
|
expiry: str # 1d | 30d | 180d | 365d | never
|
|
|
|
|
|
@router.get("/tokens")
|
|
async def list_user_tokens(current_user=Depends(require_auth)):
|
|
"""List the caller's API/MCP tokens (metadata only — the secret is never stored)."""
|
|
return {
|
|
"tokens": list_api_tokens(current_user["username"]),
|
|
"expiry_choices": list(API_TOKEN_EXPIRY_CHOICES.keys()),
|
|
}
|
|
|
|
|
|
@router.post("/tokens")
|
|
async def create_user_token(
|
|
req: CreateApiTokenRequest,
|
|
request: Request,
|
|
current_user=Depends(require_auth),
|
|
):
|
|
"""Create a long-lived API/MCP token. The raw JWT is returned ONCE."""
|
|
try:
|
|
record, token = create_api_token(current_user, req.name.strip(), req.expiry)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e))
|
|
from backend.audit import log_config_change
|
|
log_config_change(current_user["username"],
|
|
{"action": "api_token_create", "name": record["name"],
|
|
"expiry": record["expiry_key"]}, ip=get_client_ip(request))
|
|
return {"token": token, **record}
|
|
|
|
|
|
@router.delete("/tokens/{jti}")
|
|
async def delete_user_token(
|
|
jti: str,
|
|
request: Request,
|
|
current_user=Depends(require_auth),
|
|
):
|
|
"""Revoke + delete an API/MCP token (immediate effect on API and MCP)."""
|
|
try:
|
|
record = delete_api_token(jti, current_user["username"])
|
|
except KeyError:
|
|
raise HTTPException(404, "Token introuvable")
|
|
from backend.audit import log_config_change
|
|
log_config_change(current_user["username"],
|
|
{"action": "api_token_revoke", "name": record["name"]},
|
|
ip=get_client_ip(request))
|
|
return {"message": f"Token '{record['name']}' révoqué"}
|