Autenticación mediante Tokens en FastAPI empleando SQLAlchemy

▶

En la mayoría de los escenarios vamos a querer al menos proteger partes de la API; ya sea que emplees un motor de plantillas como Jinja o no, casi siempre va a ser necesario proteger los métodos de la API; para ello, la manera más básica es emplear un sistema de autenticación, ya sea en base a sesión o tokens de autenticación; con estos mecanismos, se le otorga derechos al usuario para que pueda acceder a los métodos privados de la API y podamos conocer en todo momento (realizada esta implementación) cuando el usuario manipula los datos y en qué nivel.

En este apartado veremos algunos métodos de autenticación que se integran directamente a la documentación automática y otros esquemas personalizados que constan de usuario y contraseña integrándose con la base de datos y la generación de tokens de autenticación o accesos.

Autenticación básica

▶

El esquema que se puede considerar más básico de autenticación, viene siendo en donde las credenciales del usuario se manejan en base a un token que se establece en la cabecera de la petición, específicamente en la cabecera llamada Authorization para luego ser consultadas y evaluadas por la aplicación:

api.py

from fastapi.security import APIKeyHeader
from fastapi import Depends, FastAPI, HTTPException, status

API_KEY_TOKEN = "SECRET_PASSWORD"

api_key_header = APIKeyHeader(name="Token")
@app.get("/protected-route")
async def protected_route(token: str = Depends(api_key_header)):
    if token != API_KEY_TOKEN:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
    return {"hello": "FASTAPI"}

Pruebas en la Documentación Interactiva (Swagger UI)

En el ejemplo anterior, se establece un token genérico mediante API_KEY_TOKEN y se verifica si el token suministrado en la cabecera de la petición es igual al del token; se utiliza la clase APIKeyHeader (la cual, devuelve un error de tipo 403 cuando el token no es suministrado) para recuperar un valor de un encabezado (en el ejemplo anterior la variable llamada token). Finalmente, se establece como una dependencia más de la petición.

Si vamos a la documentación, en el método creado anteriormente, veremos un candado, que indica que está protegido:

Si realizamos la petición sin autenticarnos, veremos un error 403 (el establecido anteriormente):

Code Details
403
Undocumented
Error: Forbidden

Response body
Download
{
  "detail": "Not authenticated"
}

Si damos un click sobre el candado y establecemos el token de manera correcta (un token válido):

La petición será procesada de manera exitosa con un código de estado de tipo 200.

El problema con este enfoque es que solamente podemos proteger una ruta; podemos crear una función que realice la verificación anterior:

from fastapi.security import APIKeyHeader
from fastapi import Depends, FastAPI, HTTPException, status

API_KEY_TOKEN = "SECRET_PASSWORD"

async def authenticate(token: str = Depends(APIKeyHeader(name="Token"))):
    if token != API_KEY_TOKEN:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
    return token
Y se inyecta en cada una de los métodos de la API en los cuales requerimos que el usuario esté autenticado; por ejemplo:
@app.get("/page/")
def page(db: Session = Depends(get_database_session), dependencies=Depends(authenticate)):
    print(getAll(db))
    #create_user(db)
    print(dependencies)
    return {"page": 1}

Resumen

Al recargar la documentación, notarás algo interesante: aparece un icono de un candado junto a la ruta. Swagger detecta automáticamente que este método requiere autenticación.

• Petición sin token: Si intentamos ejecutar la petición sin estar "logueados", FastAPI devolverá un error 403 Forbidden.
• Token inválido: Si hacemos clic en el candado e ingresamos un valor incorrecto (por ejemplo, "hola"), el sistema mantendrá el bloqueo y seguirá devolviendo el error 403.
• Token correcto: Al ingresar exactamente el valor SecretPassword, la ventana se cerrará indicando que estamos autorizados. Al ejecutar de nuevo, recibiremos una respuesta 200 OK.

Uso

Este es un ejemplo mínimo, pero muy potente. Imagina que vendes un servicio: el usuario realiza una compra, tú generas un token único, lo guardas en la base de datos y se lo entregas. En lugar de comparar contra un texto fijo, simplemente validarías el token contra tu base de datos.

Gracias a la potencia de las dependencias de FastAPI, podemos proteger cualquier método de nuestra API de forma extremadamente limpia. Más adelante, utilizaremos este esquema para implementar un sistema real con manejo de usuarios, contraseñas hash y métodos protegidos avanzados.

Estos ejemplos ofrecen un esquema de autenticación muy básico en donde el token recibido se verifica con una constante; no se manejan usuarios autenticados y no hay manera de saber que usuario está realizando la petición.

️ Autenticación por base de datos

▶

La autenticación anterior puede servir para definir esquemas simples, pero, en la mayoría de los casos es necesaria manejar esquemas más completos, en los cuales tendremos múltiples usuarios con la combinación ya conocida de usuario/contraseña para poder ingresar a la aplicación, para estos casos, podemos emplear el siguiente esquema.

Instalación de bcrypt para el manejo de Seguridad

Para este desarrollo, vamos a necesitar instalar un paquete al proyecto con el cual, podremos convertir textos planos en hash, específicamente lo utilizaremos para generar un hash del password del usuario:

$ pip install bcrypt

Modelos de Datos y Entidades

A continuación, modificaremos el modelo de usuarios para especificar la columna del password (en formato hash) y creamos una entidad relacional al usuario con la cual, manejaremos los tokens de acceso:

database\models.py

class User(Base):
    ***
    hashed_password = Column(String(255))
   
class AccessToken(Base):
    __tablename__ = 'access_tokens'
    user_id = Column(Integer, ForeignKey('users.id'), primary_key=True)
    access_token = Column(String(255))
    expiration_date = Column(DateTime(timezone=True))
    user = relationship('User',lazy="joined")

Con cada una de las columnas de la relación anterior, tenemos:

• user_id permite manejar la relación foránea con los usuarios es importante notar la clave primaria para esta columna, lo que significa que solamente puede existir un token de acceso por usuario.
• access_token mantiene el token de acceso del usuario autenticado.
• expiration_date especifica el tiempo de duración del token.

Recuerda borrar todas las tablas de la base de datos para crear la nueva relación y verse reflejado el cambio en el modelo de usuario.

▶

▶

A continuación, modificaremos el modelo de usuarios para especificar la columna del password (en formato hash) y creamos una entidad relacional al usuario con la cual, manejaremos los tokens de acceso:

database\models.py

class User(Base):
    ***
    hashed_password = Column(String(255))
    
class AccessToken(Base):
    __tablename__ = 'access_tokens'
    user_id = Column(Integer, ForeignKey('users.id'), primary_key=True)
    access_token = Column(String(255))
    expiration_date = Column(DateTime(timezone=True))
    user = relationship('User',lazy="joined")

Con cada una de las columnas de la relación anterior, tenemos:

• user_id permite manejar la relación foránea con los usuarios es importante notar la clave primaria para esta columna, lo que significa que solamente puede existir un token de acceso por usuario.
• access_token mantiene el token de acceso del usuario autenticado.
• expiration_date especifica el tiempo de duración del token.

▶

Recuerda borrar todas las tablas de la base de datos para crear la nueva relación y verse reflejado el cambio en el modelo de usuario.

Creamos un esquema de la nueva entidad y modificamos la entidad del usuario para indicar el campo de contraseña:

schemas.py

class User(BaseModel):
    name: str = Field(min_length=5)
    surname: str
    email: EmailStr
    website: str #HttpUrl
    class Config:
        from_attributes = True

class UserCreate(User):
    password: str

class UserDB(User):
    hashed_password: str 

class AccessToken(BaseModel):
    user_id: int
    access_token: str
    expiration_date: datetime
    class Config:
        from_attributes = True

▶

Para el usuario, si se va a crear un usuario, entonces la contraseña esta en texto plano, ya que, es la introducida por el usuario (UserCreate.password), pero, cuando se trata del usuario obtenido desde la base de datos, la contraseña esta en formato hash (UserDB.hashed_password).

Esquema de Base de Datos

Posteriormente, creamos UserDB. Este esquema está relacionado directamente con la persistencia en la base de datos.

• hashed_password: Aquí el campo cambia de nombre para reflejar que lo que almacenamos es el hash de la contraseña y no el texto plano, por motivos de seguridad.

Esquema para el Access Token

Ahora definiremos la clase AccessToken. Esta clase heredará de BaseModel y contendrá la configuración necesaria para gestionar las sesiones:

• user_id: De tipo entero, para vincular el token con el usuario.
• access_token: De tipo texto, que contendrá el string del token.
• expiration_date: De tipo datetime, para controlar la fecha de expiración.

Archivos de Apoyo (Helpers)

Vamos a crear el archivo para manejar las contraseñas y la generación de tokens. Para evitar saturar el futuro archivo de rutas del usuario (donde irá el login, el registro, etc.), separaremos esta lógica.

En la raíz del proyecto, crearemos una carpeta llamada authentication. He decidido usar una carpeta porque más adelante añadiremos otros archivos en esta misma ubicación. Dentro de ella, creamos el archivo password.py:

authentication\password.py

import bcrypt
import secrets

def get_password_hash(password: str) -> str:
    # Convertimos la contraseña a bytes
    pwd_bytes = password.encode('utf-8')
    # Generamos el salt y el hash
    salt = bcrypt.gensalt()
    hashed_password = bcrypt.hashpw(pwd_bytes, salt)
    # Retornamos como string para guardar en la BD
    return hashed_password.decode('utf-8')

def verify_password(plain_password: str, hashed_password: str) -> bool:
    password_byte_enc = plain_password.encode('utf-8')
    hashed_password_enc = hashed_password.encode('utf-8')
    return bcrypt.checkpw(password_byte_enc, hashed_password_enc)

def generate_token() -> str:
    return secrets.token_urlsafe(32)

Explicación de las funciones anteriores

• Con schemes=["bcrypt"] se especifica el algoritmo de encriptado a emplear.
• Con la función get_password_hash() permite crear un hash de la contraseña en texto plano; esta función será empleada más adelante al momento de registrar un usuario.
• Con la función verify_password() permite verificar una contraseña; la característica principal de los hash es que los mismos no pueden revertirse a un texto plano una vez convertidas, por lo tanto, existe una función para comparar si un hash corresponde con un texto plano, en este caso, el texto plano corresponde a la contraseña en texto plano; esta función será empleada más adelante para realizar el login.
• Con la función de generate_token() permite generar un token empleando la función de Python token_urlsafe() que permite generar tokens seguros de 32 bytes.

Para realizar el proceso de la autenticación del usuario y generar el token de acceso, usaremos un nuevo archivo:

▶

▶

authentication\authentication.py

from sqlalchemy.orm import Session
from datetime import datetime, timedelta

# from time
from database.models import User, AccessToken
from authentication.password import verify_password, generate_token

def authenticate(email: str, password: str, db: Session) -> User|None:
    user = db.query(User).filter(User.email == email).first()

    if user is None:
        return None
    
    if not verify_password(password, user.hashed_password):
        return None
    
    return user

def create_access_token(user:User, db: Session) -> AccessToken:
    tomorrow = datetime.now() + timedelta(days=1) # time.time + 60 * 60 * 24

    accessToken = AccessToken(user_id=user.id, expiration_date=tomorrow, access_token=generate_token())

    db.add(accessToken)
    db.commit()
    db.refresh(accessToken)

    return accessToken

Explicación de las funciones anteriores

• La función de authenticate() no tiene mucha complicación, simplemente busca el usuario por el email y verifica la contraseña en texto plano contra el hash de la contraseña guardada en la base de datos.
• La función de create_access_token() permite generar un token de acceso, para ello, se inserta un registro en la tabla de access_tokens especificando el id del usuario, la fecha (de mañana para indicar que la fecha de expiración del token) y generar el token, para ello se emplea la función de generate_token() creada en el apartado anterior.

Métodos para la API

Crearemos un nuevo archivo en donde implementamos las opciones de registrar un usuario y crear el token (login) del usuario:

▶

user.py

from fastapi import APIRouter, HTTPException, status, Depends
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from schemes import UserCreate
from authentication import password, authentication
from database import database, models

user_router = APIRouter()

@user_router.post('/token')
def create_token(form_data : OAuth2PasswordRequestForm = Depends(OAuth2PasswordRequestForm), db: Session = Depends(database.get_database_session)):
    email = form_data.username
    password = form_data.password

    user = authentication.authenticate(email,password,db)

    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    
    token = authentication.create_access_token(user, db)
    return {"access_token": token.access_token}

@user_router.post('/register', status_code=status.HTTP_201_CREATED)
def register(user: UserCreate, db:Session = Depends(database.get_database_session)): #  -> models.User

    user_exist = db.query(models.User).filter(models.User.email == user.email).first()
    if user_exist:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT,
                            detail="User email already exist")
    
    hashed_password = password.get_password_hash(user.password)
    print(hashed_password)

    userdb = models.User(name=user.name, 
                         email= user.email, 
                         surname = user.surname, 
                         website= user.website, 
                         hashed_password=hashed_password)
    
    db.add(userdb)
    db.commit()
    db.refresh(userdb)

    return userdb

Con la clase OAuth2PasswordRequestForm es inyectada como dependencia de la función para generar el token y permite declarar en un cuerpo de formulario con los siguientes campos:

1. El nombre de usuario.
2. La contraseña.
3. Un campo de scope opcional.
4. Una clave para el cliente pública y privada también opcional.

Claro está, pudieras emplear en su lugar las clases de Pydantic o las clases Body, Form o similares, algo como:

create_token(email: str = Form(), password: str = Form())

 Finalmente, son incluidas las rutas anteriores en la aplicación:

api.py

from user import user_router
***
app.include_router(user_router)
***

Prueba en Swagger

Al recargar la documentación interactiva, veremos nuestra nueva ruta /register. Al realizar una prueba con datos reales (nombre, email, sitio web y contraseña), el sistema procesará la solicitud.

Si revisamos la base de datos, verás el nuevo usuario insertado correctamente con su contraseña encriptada. El proceso ha funcionado a la perfección.

Y con esto tendremos:

Este material forma parte de mi curso y libro completo sobre FastAPI.