Implementación de Autenticación JWT para Qwen3-14B con FastAPI

La necesidad imperiosa de seguridad en endpoints de IA

Desplegar modelos de lenguaje potentes como el Qwen3-14B a través de APIs expuestas públicamente conlleva riesgos significativos si no se implementa un control de acceso adecuado. Sin una capa de autenticación, el servicio es vulnerable al uso no autorizado, lo que puede derivar en costos operativos inesperados por el uso de GPU y potenciales filtraciones de datos sensibles. Para mitigar estos riesgos, es fundamental establecer un punto de entrada seguro que valide la identidad de cada solicitante antes de procesar la solicitud en el modelo.

Este artículo detalla cómo construir una pasarela de autenticación utilizando FastAPI y JSON Web Tokens (JWT). Esta solución actúa como un intermediario seguro: intercepta las solicitudes, valida las credenciales y reenvía el tráfico autorizado al servicio subyacente de Ollama que ejecuta el modelo Qwen3-14B.

Fundamentos de JWT para servicios de IA

A diferencia de las sesiones tradicionales que almacenan el estado en el servidor, los JWT (JSON Web Tokens) son "autocontenidos". Esto significa que el token en sí contiene toda la información necesaria para validar al usuario (el "payload") y una firma criptográfica para garantizar su integridad. Para servicios de IA de alto rendimiento, esta característica es ideal porque elimina la necesidad de consultas repetidas a una base de datos durante cada inferencia, permitiendo que la GPU se centre únicamente en la generación de texto.

Un JWT se divide en tres partes concatenadas: Header (algoritmo y tipo), Payload (datos del usuario y expiración) y Signature (firma HMAC). La implementación a continuación utiliza el algoritmo HS256 para firmar estos tokens, garantizando que cualquier manipulación del contenido sea detectada inmediatamente por el servidor.

Configuración y dependencias del proyecto

Comenzaremos creando un entorno virtual e instalando las librerías necesarias. Optareoms por pyjwt para la manipulación de tokens y passlib para el manejo seguro de contraseñas.

mkdir qwen-security-gateway && cd qwen-security-gateway
python -m venv .venv
source .venv/bin/activate  # En Windows use .venv\Scripts\activate

pip install fastapi uvicorn pyjwt passlib[bcrypt] httpx pydantic-settings

La estructura del proyecto se simplifica en dos módulos principales: uno para la lógica de seguridad y otro para la aplicación API.

Lógica de seguridad y gestión de tokens

En el archivo security.py, centralizaremos la lógica de cifrado y validación. Utilizaremos una clase AuthHandler para encapsular la generación y verificación de tokens, manteniendo el código limpio y reutilizable. Se simula un repositorio de usuarios con una contraseña hasheada utilizando bcrypt.

# security.py
import os
import jwt
import bcrypt
from datetime import datetime, timedelta
from typing import Dict, Optional
from pydantic import BaseModel

class AuthConfig:
    SECRET_KEY: str = os.getenv("JWT_SECRET", "clave-super-secreta-cambiar-en-prod")
    ALGORITHM: str = "HS256"
    TOKEN_EXPIRATION: int = 60  # minutos

class User(BaseModel):
    username: str
    disabled: Optional[bool] = False

class UserInDB(User):
    hashed_password: str

# Simulación de base de datos de usuarios
# Usuario: admin, Contraseña: secret123
_db_users: Dict[str, UserInDB] = {
    "admin": UserInDB(
        username="admin",
        hashed_password="$2b$12$8r1yY4.V.qZ8.5r3.3.3eO1.1.1.1.1.1.1.1.1.1.1.1.1", # bcrypt hash para "secret123"
        disabled=False
    )
}

class AuthHandler:
    def __init__(self):
        self.config = AuthConfig()

    def verify_password(self, plain_pwd: str, hashed_pwd: str) -> bool:
        return bcrypt.checkpw(plain_pwd.encode('utf-8'), hashed_pwd.encode('utf-8'))

    def authenticate_user(self, username: str, password: str) -> Optional[UserInDB]:
        user = _db_users.get(username)
        if not user:
            return None
        if not self.verify_password(password, user.hashed_password):
            return None
        return user

    def create_token(self, username: str) -> str:
        expire = datetime.utcnow() + timedelta(minutes=self.config.TOKEN_EXPIRATION)
        payload = {
            "sub": username,
            "exp": expire,
            "iat": datetime.utcnow()
        }
        return jwt.encode(payload, self.config.SECRET_KEY, algorithm=self.config.ALGORITHM)

    def decode_token(self, token: str) -> str:
        try:
            payload = jwt.decode(token, self.config.SECRET_KEY, algorithms=[self.config.ALGORITHM])
            username: str = payload.get("sub")
            if username is None:
                raise ValueError("Token inválido")
            return username
        except jwt.PyJWTError:
            raise ValueError("Credenciales inválidas")

auth_handler = AuthHandler()

Implementación del Gateway API con FastAPI

El archivo main.py define la aplicación FastAPI. Crearemos dos endpoints principales: /login para emitir el token y /v1/completions que actúa como proxy seguro hacia el modelo Qwen3. Este último endpoint utiliza una dependencia para inyectar la lógica de validación antes de procesar la petición.

# main.py
from fastapi import FastAPI, HTTPException, Depends, status, Header
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import httpx
from pydantic import BaseModel
from security import auth_handler, UserInDB

app = FastAPI(title="Pasarela Segura Qwen3")
security = HTTPBearer()

# Configuración del modelo Ollama subyacente
OLLAMA_HOST = "http://localhost:11434"
MODEL_NAME = "qwen3:14b"

# Modelos de solicitud
class LoginRequest(BaseModel):
    username: str
    password: str

class ChatRequest(BaseModel):
    prompt: str

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        username = auth_handler.decode_token(token)
    except ValueError:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Token no válido o expirado"
        )
    # En un caso real, buscaríamos el usuario en la DB nuevamente
    return username

@app.post("/login")
async def login(form_data: LoginRequest):
    user = auth_handler.authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Credenciales incorrectas"
        )
    token = auth_handler.create_token(user.username)
    return {"access_token": token, "token_type": "bearer", "expires_in": 3600}

@app.post("/v1/completions")
async def chat_completion(request: ChatRequest, current_user: str = Depends(get_current_user)):
    """
    Endpoint protegido que reenvía la consulta a Qwen3-14B.
    Requiere un Bearer Token válido en el Header Authorization.
    """
    print(f"Petición recibida de usuario autenticado: {current_user}")
    
    # Preparar el payload para la API de Ollama
    ollama_payload = {
        "model": MODEL_NAME,
        "prompt": request.prompt,
        "stream": False
    }
    
    async with httpx.AsyncClient(timeout=60.0) as client:
        try:
            response = await client.post(f"{OLLAMA_HOST}/api/generate", json=ollama_payload)
            response.raise_for_status()
            data = response.json()
            return {
                "model": MODEL_NAME,
                "user": current_user,
                "response": data.get("response", "")
            }
        except httpx.RequestError as exc:
            raise HTTPException(status_code=503, detail="Error de conexión con el servicio de IA")
        except httpx.HTTPStatusError as exc:
            raise HTTPException(status_code=exc.response.status_code, detail="Error en el modelo")

@app.get("/status")
async def health_check():
    return {"status": "ok", "service": "qwen-gateway"}

Validación del sistema

Una vez iniciado el servidor con uvicorn main:app --reload, procedemos a verificar el flujo completo.

1. Autenticación y obtención del token:

curl -X POST "http://localhost:8000/login" \
-H "Content-Type: application/json" \
-d '{"username": "admin", "password": "secret123"}'

La respuesta incluirá un access_token largo. Copie este valor.

2. Acceso al modelo protegido:

curl -X POST "http://localhost:8000/v1/completions" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer <SU_TOKEN_AQUI>" \
-d '{"prompt": "Explícame qué es un JWT en una frase."}'

Si el token es correcto, el servidor reenviará la petición a Qwen3 y devolverá la respuesta generada por la IA junto con el nombre del usuario autorizado. Si intentamos acceder sin el token o con uno caducado, recibiremos un error 403 Forbidden.

Endurecimiento para entornos de producción

El código presentado sirve como base funcional, pero para un despliegue en producción es necesario implementar medidas adicionales:

  • Gestión de secretos: Nunca almacene las claves JWT en el código fuente. Utilice variables de entorno o servicios de secretos como AWS Secrets Manager o Vault.
  • Cifrado en tránsito: Asegúrese de que el Gateway se sirva exclusivamente a través de HTTPS para evitar la interceptación del Bearer Token.
  • Rotación de tokens: Implemente mecanismos de Refresh Token para mantener la sesión activa sin requerir que el usuario introduzca sus credenciales frecuentemente, pero permitiendo la revocación inmediata de accesos.
  • Limitación de tasa (Rate Limiting): Dado que la inferencia de IA es costosa, integre middlewares como slowapi para limitar las solicitudes por usuario y prevenir abusos.

Etiquetas: Qwen3-14B FastAPI JWT ollama Python

Publicado el 9-14 01:22

Etiquetas populares

©2026 Friki Work