¿Cómo implementar un sistema RBAC en una aplicación FastAPI para gestionar permisos de usuario?

El control de acceso basado en roles (RBAC, por sus siglas en inglés) es una técnica fundamental en la gestión de permisos dentro de aplicaciones que requieren diferentes niveles de acceso a sus usuarios. Este modelo ayuda a asignar permisos de manera eficiente según el rol del usuario, lo que facilita tanto la administración como la seguridad. En este artículo se explica cómo implementar un sistema RBAC en una aplicación FastAPI, detallando los pasos necesarios para asignar roles, restringir el acceso a ciertos endpoints y gestionar usuarios con diferentes privilegios.

Definición de Roles

En el archivo models.py, primero se debe definir la clase Role que contendrá los diferentes roles posibles en el sistema. En este caso, utilizamos dos roles básicos: basic y premium. Esta clase se convierte en un Enum que asegura que los roles sean controlados de manera rígida, evitando valores no deseados. A continuación, se debe añadir un campo role al modelo de usuario:

python
Copy code
from enum import Enum
class Role(str, Enum):
    basic = "basic"
    premium = "premium"
class User(Base):
    __tablename__ = "users"
    # campos existentes
    role: Mapped[Role] = mapped_column(
        default=Role.basic
    )

Aquí, el campo role de la clase User almacena el rol del usuario, el cual se puede asignar durante el registro. De forma predeterminada, todos los usuarios tendrán el rol basic, pero este puede modificarse si es necesario.

Registro de Usuarios con Diferentes Roles

Una vez que hemos definido los roles, es necesario permitir la asignación de roles durante el registro de usuarios. Para ello, se modifica la función add_user en el archivo operations.py, permitiendo la inclusión del parámetro role que puede ser basic o premium. El rol por defecto será basic, pero si se requiere crear un usuario con privilegios especiales, se puede asignar el rol premium:

python
Copy code
from models import Role

def add_user(
    session: Session,
    username: str,
    password: str,
    email: str,
    role: Role = Role.basic,
) -> User | None:
    hashed_password = pwd_context.hash(password)
    db_user = User(
        username=username,
        email=email,
        hashed_password=hashed_password,
        role=role,
    )
    # resto de la función

Aquí, el parámetro role se define como opcional y se asigna un valor por defecto. Esto permite que, al agregar un usuario, se pueda elegir si el rol será el estándar o uno más avanzado.

Endpoints para Diferentes Roles

Para implementar el acceso restringido a ciertos endpoints dependiendo del rol del usuario, se puede crear un nuevo archivo premium_access.py que definirá un endpoint exclusivo para usuarios premium. Este endpoint será similar al de un usuario básico, pero con el añadido de verificar que el usuario sea premium antes de permitirle el acceso:

python
Copy code
@router.post(

    "/register/premium-user",
    status_code=status.HTTP_201_CREATED,
    response_model=ResponseCreateUser,
    responses={...}  # Documentar respuestas
)
def register_premium_user(
    user: UserCreateBody,
    session: Session = Depends(get_session),
):
    user = add_user(
        session=session,
        *user.model_dump(),
        role=Role.premium,
    )
    if not user:
        raise HTTPException(
            status.HTTP_409_CONFLICT,
            "El nombre de usuario o el correo electrónico ya existe",
        )
    user_response = UserCreate(
        username=user.username,
        email=user.email,
    )
    return {
        "message": "Usuario creado",
        "user": user_response,
    }

Este endpoint permite registrar usuarios premium, asegurando que sus credenciales se gestionen adecuadamente.

Protegiendo Endpoints con Roles Específicos

El siguiente paso es definir el comportamiento del sistema de RBAC en función de los roles. Para ello, se implementan dos funciones auxiliares en el módulo rbac.py: get_current_user para obtener al usuario actual y get_premium_user para obtener solo a los usuarios premium. Estas funciones se utilizan como dependencias en los endpoints para gestionar el acceso:

python
Copy code
def get_current_user(

    token: str = Depends(oauth2_scheme),
    session: Session = Depends(get_session),
) -> UserCreateRequestWithRole:
    user = decode_access_token(token, session)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Usuario no autorizado",
        )
    return UserCreateRequestWithRole(
        username=user.username,
        email=user.email,
        role=user.role,
    )
def get_premium_user(
    current_user: Annotated[get_current_user, Depends()],
):
    if current_user.role != Role.premium:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Usuario no autorizado",
        )
    return current_user

La función get_current_user decodifica el token de acceso y devuelve los detalles del usuario, mientras que get_premium_user verifica si el usuario tiene el rol adecuado antes de permitirle el acceso a un recurso premium.

Creación de Endpoints Accesibles por Roles

Finalmente, se crean dos endpoints en el archivo rbac.py, uno accesible para todos los usuarios y otro solo para los premium. El endpoint /welcome/all-users es accesible para cualquier usuario, mientras que /welcome/premium-user está restringido solo a los usuarios premium:

python
Copy code
router = APIRouter()

@router.get("/welcome/all-users")
def all_users_can_access(
    user: Annotated[get_current_user, Depends()]
):
    return {f"Hola {user.username}, bienvenido a tu espacio"}
@router.get("/welcome/premium-user")
def only_premium_users_can_access(
    user: UserCreateResponseWithRole = Depends(get_premium_user),
):
    return {f"Hola {user.username}, bienvenido a tu espacio premium"}

Integración con la Aplicación

Una vez definidos los endpoints, es necesario incluir el enrutador de rbac.py en el archivo principal de la aplicación, main.py:

python
Copy code
import security
import premium_access
import rbac
app.include_router(premium_access.router)
app.include_router(rbac.router)

Con esto, la aplicación ya está configurada para gestionar el acceso a los endpoints según el rol del usuario. Al acceder a http://localhost:8000/docs, se puede probar el funcionamiento de los endpoints creados.

Consideraciones Finales

Es fundamental que, además de los roles básicos, se piense en la flexibilidad del sistema de permisos. Por ejemplo, la implementación de roles adicionales, la asignación de permisos más finos a nivel de función o el uso de scopes en los tokens pueden ampliar la seguridad y el control. El uso de OAuth2 para gestionar roles y permisos, y la integración de autenticación de terceros, como GitHub, puede facilitar el acceso a la aplicación y mejorar la experiencia del usuario.

¿Cómo integrar la autenticación de terceros en tu aplicación con GitHub?

La autenticación de terceros es un mecanismo ampliamente utilizado para permitir a los usuarios iniciar sesión en tu aplicación utilizando sus credenciales de servicios externos, como GitHub. En este capítulo, exploraremos cómo implementar la autenticación de GitHub en una aplicación utilizando OAuth2, un protocolo estándar de autorización. A través de este proceso, no solo autenticarás usuarios, sino que también podrás acceder a información relevante de su cuenta de GitHub, como su nombre de usuario y correo electrónico.

Creación de una nueva aplicación OAuth en GitHub

• Nombre de la aplicación: Por ejemplo, "SaasFastAPIapp".

• URL de la página de inicio: La dirección de tu aplicación, que en este caso será http://localhost:8000/ (la URL que utilizarás en tu servidor local).

• URL de redirección de autorización: Esta es la URL de tu aplicación que se utilizará para recibir el token de acceso después de la autenticación, como http://localhost:8000/github/auth/token.

Después de completar estos campos, haz clic en "Registrar aplicación". Al hacerlo, GitHub te proporcionará un Client ID y un Client Secret. Guarda ambos valores, ya que los necesitarás para la implementación en tu aplicación.

Implementación de la autenticación en tu aplicación

Con la información obtenida de GitHub, ya podemos comenzar a integrar la autenticación de GitHub en tu aplicación. Lo primero que haremos es crear un módulo llamado third_party_login.py que contendrá todas las variables y funciones necesarias para manejar la autenticación.

En este módulo, definimos las variables clave que GitHub utiliza en su proceso de autenticación:

python
Copy code
GITHUB_CLIENT_ID = "tu_github_client_id"

GITHUB_CLIENT_SECRET = "tu_github_client_secret"
GITHUB_REDIRECT_URI = "http://localhost:8000/github/auth/token"
GITHUB_AUTHORIZATION_URL = "https://github.com/login/oauth/authorize"

Importante: Asegúrate de no dejar estos valores en el código base en producción. Utiliza variables de entorno o un sistema de gestión de configuraciones más seguro.

Luego, creamos una función llamada resolve_github_token que será responsable de obtener el token de GitHub y devolver información sobre el usuario autenticado. Esta función se apoyará en la biblioteca httpx para hacer la solicitud a la API de GitHub.

python
Copy code
import httpx

from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
from models import User, get_session
def resolve_github_token(access_token: str = Depends(OAuth2()), session: Session = Depends(get_session)) -> User:
    user_response = httpx.get(
        "https://api.github.com/user", 
        headers={"Authorization": access_token},
    ).json()
    username = user_response.get("login", "")
    user = get_user(session, username)
    if not user:
        email = user_response.get("email", "")
        user = get_user(session, email)
    if not user:
        raise HTTPException(status_code=403, detail="Token no válido")
    return user

En esta función, verificamos si el usuario existe en nuestra base de datos mediante su nombre de usuario o correo electrónico. Si no se encuentra, se lanza un error de autenticación.

Creación de los endpoints de autenticación

Una vez que tenemos la lógica de autenticación, el siguiente paso es configurar los endpoints en tu aplicación. Creamos un nuevo archivo github_login.py en el cual definimos un endpoint para obtener la URL de autorización de GitHub. Este endpoint será utilizado por el frontend para redirigir al usuario a la página de inicio de sesión de GitHub.

python
Copy code
import httpx

from fastapi import APIRouter
router = APIRouter()
@router.get("/auth/url")
def github_login():
    return {
        "auth_url": GITHUB_AUTHORIZATION_URL + f"?client_id={GITHUB_CLIENT_ID}"
    }

Configuración del servidor

Una vez definidos los endpoints de autenticación, es necesario incluir el router en el archivo principal main.py para que estos sean accesibles desde la aplicación:

python
Copy code
import github_login
app.include_router(github_login.router)

De esta forma, cuando el usuario acceda al endpoint /auth/url, se le redirigirá a la página de inicio de sesión de GitHub.

Manejo del callback de GitHub

python
Copy code
@router.get("/github/auth/token")

async def github_callback(code: str):

    token_response = httpx.post(
        "https://github.com/login/oauth/access_token",
        data={
            "client_id": GITHUB_CLIENT_ID,
            "client_secret": GITHUB_CLIENT_SECRET,
            "code": code,
            "redirect_uri": GITHUB_REDIRECT_URI,
        },
        headers={"Accept": "application/json"},
    ).json()
    access_token = token_response.get("access_token")
    if not access_token:
        raise HTTPException(status_code=401, detail="Usuario no registrado")
    return {"access_token": access_token, "token_type": "bearer"}

Este endpoint maneja el intercambio del código por el token de acceso, que luego podrá ser utilizado para hacer solicitudes autenticadas a la API de GitHub.

Validación y acceso a la información del usuario

python
Copy code
@router.get("/home")

def homepage(user: UserCreateResponse = Depends(resolve_github_token)):

    return {"message": f"¡Has iniciado sesión como {user.username}!"}

Prueba de la autenticación

Para probar que todo funciona correctamente, asegúrate de tener en tu base de datos un usuario con el mismo nombre de usuario o correo electrónico que el que usas en GitHub. Luego, con el token obtenido del proceso de autenticación, podrás hacer una solicitud GET al endpoint /home para comprobar que la autenticación se ha realizado con éxito.

Expansión a otros proveedores de autenticación

GitHub no es el único proveedor de autenticación que puedes integrar en tu aplicación. Otros servicios como Google y Twitter también utilizan OAuth2, aunque con pequeñas diferencias en su configuración. El proceso de integración es en gran parte el mismo: registrar una aplicación en el servicio de autenticación, obtener un Client ID y Client Secret, y luego configurar los endpoints adecuados en tu aplicación para manejar la autenticación.

¿Cómo implementar un sistema de Webhooks en FastAPI y proteger tu aplicación con middleware?

En el desarrollo de aplicaciones web modernas, la capacidad de integrar sistemas externos a través de eventos en tiempo real se ha vuelto fundamental. Esto se logra mediante el uso de los webhooks, que son esencialmente llamadas HTTP disparadas por eventos específicos en un sistema y que envían una carga útil a otro sistema. Esta arquitectura asíncrona y basada en eventos permite integrar fácilmente servicios de terceros, notificaciones en tiempo real y flujos de trabajo automatizados. A través de este artículo, exploraremos cómo implementar un sistema de webhooks utilizando FastAPI, y también cómo proteger tu aplicación utilizando middleware como TrustedHostMiddleware.

El uso de middleware es esencial para mantener la seguridad y la estabilidad de una aplicación. En particular, TrustedHostMiddleware es una herramienta poderosa que ayuda a controlar qué hosts pueden acceder a tu aplicación. Cuando se configura correctamente, permite bloquear solicitudes provenientes de hosts no deseados, asegurando que solo las solicitudes de hosts confiables puedan interactuar con tu API. Para activar este middleware, simplemente debes agregarlo a tu aplicación FastAPI, como se muestra a continuación:

python
Copy code
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["localhost"],
)

Con esto, tu aplicación solo aceptará solicitudes provenientes de localhost, lo que es útil en entornos de desarrollo o cuando se quiere restringir el acceso a la API.

Para poner a prueba esta configuración, puedes hacer que el servidor esté accesible desde toda la red local. Esto se logra utilizando el host 0.0.0.0 al ejecutar uvicorn, como se muestra a continuación:

bash
Copy code
$ uvicorn main:app --host=0.0.0.0

Implementación de Webhooks en FastAPI

Una vez que tu aplicación esté protegida por un middleware adecuado, puedes comenzar a implementar un sistema de webhooks. Un webhook en nuestra API de FastAPI actuará como un sistema de monitoreo que notifica a los suscriptores de cada solicitud realizada a la API. Este sistema es ideal para aplicaciones que necesitan integrar con servicios externos o enviar notificaciones automáticas.

1. Configuración del Sistema de Registro de URLs

Para comenzar, necesitamos un sistema que registre las URLs a las que se enviarán las notificaciones de los webhooks. Para esto, podemos crear un endpoint dedicado que almacene las URLs en el estado de la aplicación, utilizando el siguiente código:

python
Copy code
from contextlib import asynccontextmanager

from fastapi import FastAPI, Body, Request
@asynccontextmanager
async def lifespan(app: FastAPI):
    yield {"webhook_urls": []}
app = FastAPI(lifespan=lifespan)
@app.post("/register-webhook-url")
async def add_webhook_url(request: Request, url: str = Body()):
    if not url.startswith("http"):
        url = f"http://{url}"
    request.state.webhook_urls.append(url)
    return {"url added": url}

Este endpoint permite registrar nuevas URLs que se almacenan en el estado de la aplicación. Si la URL proporcionada no tiene el protocolo http o https, se añade automáticamente http://.

2. Implementación de los Callbacks del Webhook

python
Copy code
from pydantic import BaseModel
class Event(BaseModel):
    host: str
    path: str
    time: str
    body: str | None = None

Luego, necesitamos una función que envíe los eventos a las URLs registradas. Utilizamos la biblioteca httpx para realizar solicitudes HTTP asíncronas, y un sistema de logging para hacer un seguimiento de las notificaciones enviadas.

python
Copy code
import logging

from httpx import AsyncClient
client = AsyncClient()
logger = logging.getLogger("uvicorn")
async def send_event_to_url(url: str, event: Event):
    logger.info(f"Enviando evento a {url}")
    try:

        await client.post(f"{url}/fastapi-webhook", json=event.dict())

    except Exception as e:
        logger.error(f"Error al enviar evento a {url}: {e}")

3. Middleware para Enviar Webhooks

El siguiente paso es crear el middleware que interceptará las solicitudes HTTP y enviará los eventos a los suscriptores. Aquí, definimos la clase WebhookSenderMiddleware, que se encarga de procesar las solicitudes entrantes, construir el evento correspondiente y enviarlo a las URLs registradas.

python
Copy code
from asyncio import create_task

from datetime import datetime
from fastapi import Request
from starlette.types import ASGIApp, Receive, Scope, Send
class WebhookSenderMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        if scope["type"] == "http":
            message = await receive()
            body = message.get("body", b"")
            request = Request(scope=scope)
            event = Event(
                host=request.client.host,
                path=request.url.path,
                time=datetime.now().isoformat(),
                body=body,
            )
            urls = request.state.webhook_urls
            for url in urls:
                await create_task(send_event_to_url(url, event))
        await self.app(scope, receive, send)

Este middleware filtra las solicitudes HTTP, construye el evento y envía el evento a cada URL registrada en paralelo. De esta manera, se asegura que todos los suscriptores reciban la notificación de manera eficiente.

Al implementar correctamente el middleware y los webhooks, no solo habilitas una comunicación en tiempo real entre tu API y servicios externos, sino que también proporcionas una base robusta para notificaciones automáticas y flujos de trabajo más complejos.

Consideraciones Adicionales

La seguridad también juega un papel importante al implementar webhooks. Asegúrate de que los endpoints de webhook estén protegidos mediante autenticación o verificación, para evitar que usuarios no autorizados envíen datos a tu sistema.