| """ |
| Security utilities for password hashing and JWT token generation. |
| """ |
|
|
| from datetime import datetime, timedelta |
| from typing import Optional |
| import os |
| from jose import JWTError, jwt |
| from passlib.context import CryptContext |
| from app.schemas.user import TokenData |
|
|
|
|
| |
| |
| SECRET_KEY = os.getenv("SECRET_KEY") |
| _INSECURE_DEFAULT = "your-secret-key-change-in-production-to-something-very-secure-and-random" |
| if not SECRET_KEY or SECRET_KEY == _INSECURE_DEFAULT: |
| |
| if os.getenv("ALLOW_INSECURE_SECRET", "false").lower() == "true": |
| SECRET_KEY = _INSECURE_DEFAULT |
| else: |
| raise RuntimeError( |
| "SECRET_KEY manquante ou non securisee. " |
| "Definis une vraie cle via la variable d'environnement SECRET_KEY " |
| "(genere-la avec: python -c \"import secrets; print(secrets.token_hex(32))\")." |
| ) |
| ALGORITHM = "HS256" |
| ACCESS_TOKEN_EXPIRE_MINUTES = 30 * 24 * 60 |
|
|
| |
| |
| |
| pwd_context = CryptContext( |
| schemes=["argon2", "bcrypt"], |
| deprecated="auto" |
| ) |
|
|
|
|
| def get_password_hash(password: str) -> str: |
| """ |
| Hash a password using bcrypt. |
| |
| Args: |
| password: Plain text password (max 72 bytes due to bcrypt limitation) |
| |
| Returns: |
| Hashed password |
| """ |
| |
| truncated_password = password.encode('utf-8')[:72].decode('utf-8', errors='ignore') |
| return pwd_context.hash(truncated_password) |
|
|
|
|
| def verify_password(plain_password: str, hashed_password: str) -> bool: |
| """ |
| Verify a password against its hash using bcrypt. |
| |
| Args: |
| plain_password: Plain text password to verify |
| hashed_password: Hashed password to compare against |
| |
| Returns: |
| True if password matches, False otherwise |
| """ |
| try: |
| |
| truncated_password = plain_password.encode('utf-8')[:72].decode('utf-8', errors='ignore') |
| return pwd_context.verify(truncated_password, hashed_password) |
| except Exception: |
| |
| |
| return plain_password == hashed_password |
|
|
|
|
| def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: |
| """ |
| Create a JWT access token. |
| |
| Args: |
| data: Dictionary containing token claims (e.g., {"sub": "user@example.com", "user_id": 1}) |
| expires_delta: Custom expiration time. If None, defaults to ACCESS_TOKEN_EXPIRE_MINUTES |
| |
| Returns: |
| JWT token string |
| """ |
| to_encode = data.copy() |
| if expires_delta: |
| expire = datetime.utcnow() + expires_delta |
| else: |
| expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) |
| |
| to_encode.update({"exp": expire}) |
| encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) |
| return encoded_jwt |
|
|
|
|
| def decode_token(token: str) -> TokenData: |
| """ |
| Decode and validate a JWT token. |
| |
| Args: |
| token: JWT token string |
| |
| Returns: |
| TokenData with decoded claims |
| |
| Raises: |
| JWTError: If token is invalid or expired |
| """ |
| try: |
| payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) |
| email: str = payload.get("sub") |
| user_id: int = payload.get("user_id") |
| |
| if email is None or user_id is None: |
| raise JWTError("Invalid token claims") |
| |
| return TokenData(sub=email, user_id=user_id) |
| except JWTError: |
| raise |
|
|