26bd096a06
Major changes across backend, frontend, infrastructure: - Provider system with model selection (Google, DeepL, OpenAI, Ollama, Google Cloud) - Admin panel: user management, pricing, settings - Glossary system with CSV import/export - Subscription and tier quota management - Security hardening (rate limiting, API key auth, path traversal fixes) - Docker compose for dev, prod, and IONOS deployment - Alembic migrations for new tables - Frontend: dashboard, pricing page, landing page, i18n (en/fr) - Test suite and verification scripts Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
113 lines
3.1 KiB
Python
113 lines
3.1 KiB
Python
"""
|
|
Shared authentication dependencies for routes.
|
|
Story 3.9: Extracted from api_key_routes.py and glossary_routes.py
|
|
Story 6.4: Bind user_id to structlog context when user is resolved.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
|
|
from services.auth_service import verify_token, get_user_by_id
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
security = HTTPBearer(auto_error=False)
|
|
|
|
|
|
class ProUser:
|
|
"""Wrapper for authenticated user with tier info."""
|
|
|
|
def __init__(self, user):
|
|
self._user = user
|
|
self.id = user.id
|
|
self.email = getattr(user, "email", None)
|
|
self._tier = None
|
|
|
|
@property
|
|
def tier(self) -> str:
|
|
if self._tier is None:
|
|
user_tier = getattr(self._user, "tier", None)
|
|
if user_tier:
|
|
self._tier = user_tier
|
|
else:
|
|
plan_value = getattr(self._user, "plan", None)
|
|
if plan_value and hasattr(plan_value, "value"):
|
|
if plan_value.value in ("pro", "business", "enterprise"):
|
|
self._tier = "pro"
|
|
else:
|
|
self._tier = "free"
|
|
else:
|
|
self._tier = "free"
|
|
return self._tier
|
|
|
|
|
|
def require_auth(
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
|
|
):
|
|
"""Dependency that requires a valid JWT token."""
|
|
if not credentials:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail={
|
|
"error": "UNAUTHORIZED",
|
|
"message": "Authentification requise",
|
|
},
|
|
)
|
|
|
|
payload = verify_token(credentials.credentials)
|
|
if not payload:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail={
|
|
"error": "UNAUTHORIZED",
|
|
"message": "Token invalide ou expiré",
|
|
},
|
|
)
|
|
|
|
sub = payload.get("sub")
|
|
if not sub or not isinstance(sub, str):
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail={
|
|
"error": "UNAUTHORIZED",
|
|
"message": "Token invalide",
|
|
},
|
|
)
|
|
|
|
user = get_user_by_id(sub)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail={
|
|
"error": "UNAUTHORIZED",
|
|
"message": "Utilisateur non trouvé",
|
|
},
|
|
)
|
|
|
|
try:
|
|
from core.logging import bind_request_context
|
|
bind_request_context(user_id=str(getattr(user, "id", user)))
|
|
except Exception:
|
|
pass
|
|
return user
|
|
|
|
|
|
def require_pro_user(user=Depends(require_auth)) -> ProUser:
|
|
"""Dependency that requires a valid Pro user JWT token."""
|
|
pro_user = ProUser(user)
|
|
|
|
if pro_user.tier != "pro":
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail={
|
|
"error": "PRO_FEATURE_REQUIRED",
|
|
"message": "Cette fonctionnalité nécessite un abonnement Pro.",
|
|
"details": {"current_tier": pro_user.tier},
|
|
},
|
|
)
|
|
|
|
return pro_user
|