| """Seguridad: hashing de contraseñas, emisión/validación de JWT y |
| dependencias de autenticación para FastAPI. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from datetime import timedelta |
| from typing import Annotated |
|
|
| import bcrypt |
| import jwt |
| from fastapi import Depends |
| from fastapi.security import OAuth2PasswordBearer |
|
|
| from app.core.config import settings |
| from app.database import collections |
| from app.database.mongodb import get_collection |
| from app.database.objectid import to_object_id |
| from app.exceptions import AuthenticationError, AuthorizationError |
| from app.models.common import utcnow |
| from app.models.user import User |
|
|
| |
| _BCRYPT_MAX_BYTES = 72 |
|
|
| oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_PREFIX}/auth/login") |
|
|
|
|
| |
| |
| |
|
|
|
|
| def _encode_password(password: str) -> bytes: |
| """Codifica la contraseña a bytes, truncando al límite de bcrypt.""" |
| return password.encode("utf-8")[:_BCRYPT_MAX_BYTES] |
|
|
|
|
| def hash_password(password: str) -> str: |
| """Devuelve el hash bcrypt de una contraseña en texto plano.""" |
| return bcrypt.hashpw(_encode_password(password), bcrypt.gensalt()).decode("utf-8") |
|
|
|
|
| def verify_password(password: str, hashed_password: str) -> bool: |
| """Comprueba si una contraseña coincide con su hash.""" |
| try: |
| return bcrypt.checkpw( |
| _encode_password(password), hashed_password.encode("utf-8") |
| ) |
| except (ValueError, TypeError): |
| return False |
|
|
|
|
| |
| |
| |
|
|
|
|
| def create_access_token(subject: str, expires_minutes: int | None = None) -> str: |
| """Crea un token de acceso firmado para el ``subject`` indicado.""" |
| minutes = expires_minutes or settings.ACCESS_TOKEN_EXPIRE_MINUTES |
| now = utcnow() |
| payload = { |
| "sub": subject, |
| "iat": now, |
| "exp": now + timedelta(minutes=minutes), |
| } |
| return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM) |
|
|
|
|
| def decode_access_token(token: str) -> str: |
| """Decodifica un token y devuelve su ``subject`` (id de usuario).""" |
| try: |
| payload = jwt.decode( |
| token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM] |
| ) |
| except jwt.PyJWTError as exc: |
| raise AuthenticationError("Token inválido o expirado.") from exc |
|
|
| subject = payload.get("sub") |
| if not subject: |
| raise AuthenticationError("Token inválido: falta el sujeto.") |
| return subject |
|
|
|
|
| |
| |
| |
|
|
|
|
| async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User: |
| """Resuelve el usuario autenticado a partir del token Bearer.""" |
| user_id = decode_access_token(token) |
| collection = get_collection(collections.USERS) |
| document = await collection.find_one({"_id": to_object_id(user_id)}) |
| if document is None: |
| raise AuthenticationError("El usuario del token ya no existe.") |
| return User(**document) |
|
|
|
|
| async def get_current_active_user( |
| current_user: Annotated[User, Depends(get_current_user)], |
| ) -> User: |
| """Garantiza que el usuario autenticado está activo.""" |
| if not current_user.is_active: |
| raise AuthorizationError("La cuenta está desactivada.") |
| return current_user |
|
|
|
|
| async def get_current_superuser( |
| current_user: Annotated[User, Depends(get_current_active_user)], |
| ) -> User: |
| """Garantiza que el usuario autenticado es superusuario.""" |
| if not current_user.is_superuser: |
| raise AuthorizationError("Se requieren privilegios de administrador.") |
| return current_user |
|
|
