OAuth2 con FastAPI

Configuración básica de OAuth2PasswordBearer

OAuth2 es un estándar de autorización que permite a las aplicaciones acceder a recursos protegidos de forma segura. En FastAPI, implementamos OAuth2 utilizando la clase OAuth2PasswordBearer, que maneja el flujo de autenticación mediante tokens de acceso.

La configuración básica requiere definir un esquema de seguridad que especifique dónde los clientes deben enviar las credenciales para obtener tokens. FastAPI proporciona herramientas integradas que simplifican significativamente esta implementación.

Instalación de dependencias

Antes de configurar OAuth2, necesitamos instalar las librerías necesarias para el manejo de tokens JWT:

pip install python-jose[cryptography] passlib[bcrypt] python-multipart

Estas dependencias nos proporcionan:

• python-jose: Para crear y verificar tokens JWT
• passlib: Para el hashing seguro de contraseñas
• python-multipart: Para procesar formularios de login

Configuración del esquema OAuth2

El primer paso es crear una instancia de OAuth2PasswordBearer que defina el endpoint donde los clientes solicitarán tokens:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from datetime import datetime, timedelta
import os

app = FastAPI()

# Configuración del esquema OAuth2
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Configuración para JWT
SECRET_KEY = os.getenv("SECRET_KEY", "tu-clave-secreta-muy-segura")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

El parámetro tokenUrl="token" indica que el endpoint de autenticación estará disponible en /token. Esta URL es donde los clientes enviarán sus credenciales para obtener un token de acceso.

Estructura de datos para usuarios

Definimos los modelos Pydantic necesarios para manejar usuarios y tokens:

from pydantic import BaseModel
from typing import Optional

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

class Token(BaseModel):
    access_token: str
    token_type: str

La clase UserInDB extiende User añadiendo el campo hashed_password, siguiendo el principio de separación entre datos públicos y privados.

Base de datos simulada

Para este ejemplo, utilizamos un diccionario en memoria que simula una base de datos de usuarios:

# Simulación de base de datos de usuarios
fake_users_db = {
    "testuser": {
        "username": "testuser",
        "full_name": "Usuario de Prueba",
        "email": "test@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",  # secret
        "disabled": False,
    }
}

La contraseña hasheada corresponde a "secret" procesada con bcrypt. En una aplicación real, estos datos estarían almacenados en una base de datos.

Funciones auxiliares para manejo de usuarios

Implementamos las funciones básicas para obtener y verificar usuarios:

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="No se pudieron validar las credenciales",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

La función get_current_user utiliza Depends(oauth2_scheme) para extraer automáticamente el token del header Authorization de las peticiones HTTP.

Protección de rutas

Con la configuración establecida, podemos proteger cualquier ruta simplemente añadiendo la dependencia:

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

@app.get("/protected")
async def protected_route(current_user: User = Depends(get_current_user)):
    return {"message": f"Hola {current_user.username}, esta es una ruta protegida"}

Cuando un cliente accede a estas rutas, FastAPI automáticamente:

• Extrae el token del header Authorization: Bearer <token>
• Valida el token usando la función get_current_user
• Inyecta el usuario como parámetro de la función

Manejo de errores de autenticación

FastAPI maneja automáticamente los errores de autenticación cuando no se proporciona un token válido:

# Si no hay token o es inválido, FastAPI retorna:
# {
#   "detail": "No se pudieron validar las credenciales"
# }
# Con código de estado 401 Unauthorized

El header WWW-Authenticate: Bearer indica al cliente que debe proporcionar un token Bearer para acceder al recurso protegido.

Esta configuración básica establece los fundamentos de seguridad necesarios para implementar un sistema de autenticación completo. El esquema OAuth2PasswordBearer se encarga de la extracción y validación de tokens, mientras que las funciones auxiliares gestionan la lógica de usuarios y permisos.

Login y obtención de tokens

El proceso de login en OAuth2 consiste en validar las credenciales del usuario y generar un token JWT que permita el acceso a recursos protegidos. Este flujo se implementa mediante un endpoint específico que recibe credenciales y devuelve un token de acceso.

Validación de credenciales

Primero necesitamos implementar las funciones de autenticación que verificarán las credenciales proporcionadas por el usuario:

from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordRequestForm

# Configuración para hashing de contraseñas
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password, hashed_password):
    """Verifica si la contraseña en texto plano coincide con el hash"""
    return pwd_context.verify(plain_password, hashed_password)

def authenticate_user(fake_db, username: str, password: str):
    """Autentica un usuario verificando sus credenciales"""
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

La función authenticate_user realiza una validación en dos pasos: primero verifica que el usuario existe y luego comprueba que la contraseña proporcionada coincide con el hash almacenado.

Generación de tokens JWT

Los tokens JWT contienen información del usuario codificada de forma segura. Implementamos la función que crea estos tokens:

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """Crea un token JWT con los datos proporcionados"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

El token incluye un tiempo de expiración (exp) que determina cuándo dejará de ser válido. Por defecto, los tokens expiran en 15 minutos, pero podemos especificar un tiempo personalizado.

Endpoint de login

El endpoint principal para el login utiliza OAuth2PasswordRequestForm para recibir las credenciales de forma estándar:

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """Endpoint para autenticar usuario y obtener token de acceso"""
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Nombre de usuario o contraseña incorrectos",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

Este endpoint recibe las credenciales mediante un formulario estándar y devuelve un objeto con el token y su tipo. El campo sub (subject) del JWT contiene el nombre de usuario.

Uso del endpoint de login

Los clientes pueden autenticarse enviando una petición POST al endpoint /token con las credenciales:

# Ejemplo de petición usando requests (para testing)
import requests

response = requests.post(
    "http://localhost:8000/token",
    data={
        "username": "testuser",
        "password": "secret"
    }
)

if response.status_code == 200:
    token_data = response.json()
    access_token = token_data["access_token"]
    print(f"Token obtenido: {access_token}")
else:
    print("Error en autenticación")

Validación de usuarios activos

Para mayor seguridad, podemos añadir una verificación adicional que compruebe si el usuario está activo:

def get_current_active_user(current_user: User = Depends(get_current_user)):
    """Obtiene el usuario actual y verifica que esté activo"""
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Usuario inactivo")
    return current_user

Esta función se puede usar como dependencia adicional en rutas que requieran usuarios activos:

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

Manejo de errores en el login

El sistema maneja varios tipos de errores durante el proceso de autenticación:

Credenciales incorrectas:

# Respuesta cuando username/password son incorrectos
{
    "detail": "Nombre de usuario o contraseña incorrectos"
}
# Status: 401 Unauthorized

Token expirado:

# Respuesta cuando se usa un token expirado
{
    "detail": "No se pudieron validar las credenciales"
}
# Status: 401 Unauthorized

Flujo completo de autenticación

El proceso completo de autenticación sigue estos pasos:

1. El cliente envía credenciales:

curl -X POST "http://localhost:8000/token" \
     -H "Content-Type: application/x-www-form-urlencoded" \
     -d "username=testuser&password=secret"

2. El servidor valida y responde:

{
    "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
    "token_type": "bearer"
}

3. El cliente usa el token en peticiones posteriores:

curl -X GET "http://localhost:8000/users/me" \
     -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

Este flujo garantiza que solo los usuarios autenticados correctamente puedan acceder a los recursos protegidos, manteniendo la seguridad mediante tokens con tiempo de vida limitado.