Auditoria de seguridad: lo que encontramos y como lo arreglamos

Toda plataforma de pagos tiene debilidades de seguridad. La pregunta es si las encuentras antes que tus atacantes. Despues de mas de 60 sesiones construyendo 0fee.dev a velocidad de startup, realizamos una auditoria de seguridad exhaustiva. Los hallazgos fueron aleccionadores. Algunos problemas eran criticos. Algunos eran limitaciones arquitectonicas de nuestras decisiones iniciales. Todos necesitaban correccion.

Este articulo es un relato transparente de lo que encontramos y como lo abordamos. Publicamos esto no para avergonzarnos sino porque la comunidad fintech se beneficia de postmortems honestos. Si estas construyendo una plataforma de pagos, estos son los escollos que debes evitar.

Problemas criticos

1. Discrepancia de entorno en el enrutamiento

El error mas peligroso que encontramos fue una discrepancia de entorno en la logica de enrutamiento de pagos. El motor de enrutamiento podia, bajo condiciones especificas, enviar un pago en vivo a un proveedor de prueba o viceversa.

python# ANTES: El error
async def route_payment(app_id: str, payment_data: dict):
    app = await get_app(app_id)
    providers = await get_available_providers(
        country=payment_data["country"],
        method=payment_data["method"],
    )
    # ERROR: No filtraba por app.mode (test/live)
    best_provider = select_best_provider(providers)
    return best_provider

La funcion get_available_providers devolvia todos los proveedores que coincidian con el pais y metodo, independientemente de si la app estaba en modo prueba o en vivo. Una app en modo prueba teoricamente podia enrutar a una instancia real de Stripe.

python# DESPUES: La correccion
async def route_payment(app_id: str, payment_data: dict):
    app = await get_app(app_id)
    providers = await get_available_providers(
        country=payment_data["country"],
        method=payment_data["method"],
        mode=app.mode,  # Filtrado explicito por modo
    )
    if not providers:
        raise HTTPException(400, f"No {app.mode} providers available for this payment")
    best_provider = select_best_provider(providers)
    return best_provider

Esta fue una correccion de una linea con enormes implicaciones. Sin ella, una clave API de prueba podria provocar cargos reales.

2. Falta de validacion de datos semilla

El script de datos semilla creaba usuarios de prueba con credenciales conocidas pero no validaba que esas credenciales fueran eliminadas antes del despliegue en produccion. Anadimos una verificacion al arranque:

python# main.py verificacion al arranque
@app.on_event("startup")
async def validate_no_seed_data():
    if ENVIRONMENT == "production":
        seed_users = await db.scalars(
            select(User).where(User.email.in_([
                "[email protected]",
                "[email protected]",
                "[email protected]",
            ]))
        )
        if seed_users.all():
            logger.critical(
                "SEED DATA DETECTED IN PRODUCTION. "
                "Remove test users before deployment."
            )
            raise RuntimeError("Seed data present in production database")

3. Sin validacion de credenciales al configurar proveedores

Cuando los desarrolladores configuraban proveedores de pago, el sistema aceptaba cualquier credencial sin validacion. Un desarrollador podia introducir claves de Stripe invalidas y solo descubrir el problema cuando el primer pago fallaba.

python# DESPUES: Validacion de credenciales
async def configure_provider(
    app_id: str,
    provider_name: str,
    credentials: dict,
) -> AppProvider:
    provider = get_provider_instance(provider_name)

    # Validar credenciales haciendo una llamada API de prueba
    validation = await provider.validate_credentials(credentials)
    if not validation.valid:
        raise HTTPException(400, f"Invalid credentials: {validation.error}")

    # Almacenar solo despues de que la validacion pase
    encrypted_creds = encrypt_credentials(credentials)
    ...

Cada adaptador de proveedor ahora implementa un metodo validate_credentials que hace una llamada API ligera (por ejemplo, obtener detalles de la cuenta) para verificar que las credenciales son funcionales.

Correcciones de seguridad

4. Salt de cifrado fijo

El servicio de cifrado para almacenar credenciales de proveedores usaba un salt estatico. Esto significaba que credenciales identicas cifradas para diferentes apps producirian texto cifrado identico, haciendo posible detectar cuando dos apps usaban las mismas credenciales de proveedor.

python# ANTES: Salt estatico
class EncryptionService:
    SALT = b"0fee_static_salt_v1"  # Igual para todos los cifrados

    def encrypt(self, plaintext: str) -> str:
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=self.SALT,
            iterations=100_000,
        )
        ...

python# DESPUES: Salt aleatorio por cifrado
class EncryptionService:
    def encrypt(self, plaintext: str) -> str:
        salt = os.urandom(16)  # Salt unico por cifrado
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100_000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(self.master_key))
        fernet = Fernet(key)
        encrypted = fernet.encrypt(plaintext.encode())
        # Anteponer salt al texto cifrado para descifrado
        return base64.b64encode(salt + encrypted).decode()

    def decrypt(self, ciphertext: str) -> str:
        raw = base64.b64decode(ciphertext)
        salt = raw[:16]  # Extraer salt
        encrypted = raw[16:]
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100_000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(self.master_key))
        fernet = Fernet(key)
        return fernet.decrypt(encrypted).decode()

Con un salt aleatorio por cifrado, texto plano identico produce texto cifrado diferente cada vez.

5. Enmascaramiento de claves API

Las claves API se estaban registrando y devolviendo completas en algunas respuestas de la API. Implementamos enmascaramiento consistente:

python# utils/masking.py
def mask_api_key(key: str) -> str:
    """Mostrar solo prefijo y ultimos 4 caracteres."""
    if not key or len(key) < 12:
        return "****"
    prefix = key[:8]   # ej., "sk_live_"
    suffix = key[-4:]
    return f"{prefix}...{suffix}"

def mask_for_logging(data: dict) -> dict: """Enmascarar recursivamente campos sensibles en la salida de logs.""" sensitive_keys = { "api_key", "secret_key", "password", "token", "client_secret", "webhook_secret", "private_key", } masked = {} for k, v in data.items(): if k.lower() in sensitive_keys: masked[k] = "*MASKED*" elif isinstance(v, dict): masked[k] = mask_for_logging(v) else: masked[k] = v return masked ```

Cada respuesta de API que incluye informacion de claves ahora devuelve la version enmascarada. Los logs usan mask_for_logging para evitar que las credenciales aparezcan en archivos de log.

6. Limitacion de tasa en autenticacion

Los endpoints de login y OAuth no tenian limitacion de tasa, haciendolos vulnerables a ataques de fuerza bruta:

python# middleware/rate_limit.py
from datetime import datetime, timedelta
from collections import defaultdict

class AuthRateLimiter:
    def __init__(self, max_attempts: int = 5, window: int = 300):
        self.max_attempts = max_attempts
        self.window = window  # segundos
        self.attempts: dict[str, list[datetime]] = defaultdict(list)

    def check(self, identifier: str) -> bool:
        """Devuelve True si la solicitud debe permitirse."""
        now = datetime.utcnow()
        cutoff = now - timedelta(seconds=self.window)

        # Limpiar intentos antiguos
        self.attempts[identifier] = [
            t for t in self.attempts[identifier] if t > cutoff
        ]

        if len(self.attempts[identifier]) >= self.max_attempts:
            return False

        self.attempts[identifier].append(now)
        return True

auth_limiter = AuthRateLimiter(max_attempts=5, window=300) BLANK @router.post("/auth/login") async def login(request: Request, data: LoginRequest): client_ip = request.client.host BLANK if not auth_limiter.check(client_ip): raise HTTPException( status_code=429, detail="Too many login attempts. Try again in 5 minutes.", headers={"Retry-After": "300"}, ) ... ```

Cinco intentos por direccion IP dentro de una ventana de 5 minutos. Despues de eso, el cliente recibe un 429 con una cabecera Retry-After.

7. Brechas en verificacion de webhooks

Los webhooks entrantes de proveedores de pago no se verificaban consistentemente. Algunos proveedores (Stripe) tienen verificacion de firma robusta. Otros se aceptaban sin validacion:

python# ANTES: Algunos proveedores no tenian verificacion de webhook
@router.post("/webhooks/{provider}")
async def handle_webhook(provider: str, request: Request):
    body = await request.body()
    # Procesar directamente sin verificacion para algunos proveedores
    await process_webhook(provider, body)

python# DESPUES: Todos los proveedores verifican webhooks
@router.post("/webhooks/{provider}")
async def handle_webhook(provider: str, request: Request):
    body = await request.body()
    headers = dict(request.headers)

    provider_instance = get_provider_instance(provider)

    # Cada proveedor debe implementar verify_webhook
    if not await provider_instance.verify_webhook(body, headers):
        logger.warning(f"Webhook verification failed for {provider}")
        raise HTTPException(400, "Webhook verification failed")

    await process_webhook(provider, body)

La clase base del proveedor ahora define verify_webhook como un metodo abstracto. Cada adaptador debe implementarlo. Los proveedores que no ofrecen verificacion de firma usan validacion alternativa (por ejemplo, verificar que la IP de origen coincida con las IPs conocidas del proveedor).

8. Riesgos de inyeccion SQL

Varios endpoints tempranos usaban formateo de cadenas en consultas SQL:

python# ANTES: Vulnerabilidad de inyeccion SQL
@router.get("/transactions")
async def list_transactions(status: str = None):
    query = f"SELECT * FROM transactions WHERE status = '{status}'"
    result = await db.execute(text(query))

python# DESPUES: Consultas parametrizadas (via ORM)
@router.get("/transactions")
async def list_transactions(status: str = None):
    query = select(Transaction)
    if status:
        query = query.where(Transaction.status == status)
    result = await db.scalars(query)

La migracion a modelos ORM de SQLAlchemy (cubierta en el articulo de SQLAdmin) elimino todas las consultas SQL sin procesar, lo que resolvio el riesgo de inyeccion de manera sistemica en lugar de consulta por consulta.

Problemas arquitectonicos

9. Limitacion de escritor unico de SQLite

SQLite permite solo un escritor a la vez. Con solicitudes API concurrentes, las operaciones de escritura se serializarian, creando un cuello de botella. Mas criticamente, el modo WAL (Write-Ahead Logging) podia servir lecturas obsoletas a nuevas conexiones.

Este no era un error a corregir sino una arquitectura a reemplazar. La migracion de SQLite a PostgreSQL (cubierta en el articulo 055) fue la solucion. PostgreSQL maneja escrituras concurrentes con MVCC (Control de Concurrencia Multi-Version), eliminando tanto el cuello de botella de escritura como el problema de lecturas obsoletas.

10. Falta de idempotencia

Las operaciones de pago deben ser idempotentes. Si un timeout de red causa un reintento, la segunda solicitud no deberia crear un segundo pago. Varios endpoints carecian de aplicacion de idempotencia:

python# DESPUES: Soporte de clave de idempotencia
@router.post("/payments")
async def create_payment(
    data: PaymentCreate,
    idempotency_key: str = Header(None, alias="Idempotency-Key"),
):
    if idempotency_key:
        existing = await get_by_idempotency_key(idempotency_key)
        if existing:
            return existing  # Devolver respuesta en cache

    payment = await process_payment(data)

    if idempotency_key:
        await store_idempotency_key(idempotency_key, payment, ttl=86400)

    return payment

Las claves de idempotencia se almacenan durante 24 horas. La misma clave siempre devuelve la misma respuesta, previniendo cargos duplicados.

11. Falta de historial de eventos

El sistema carecia de un registro de eventos completo para los cambios de estado de los pagos. Cuando un pago transitaba de pendiente a completado, solo se almacenaba el estado final. No habia registro de cuando ocurrio cada transicion o que la provoco.

python# DESPUES: Historial de eventos de pago
class PaymentEvent(Base):
    __tablename__ = "payment_events"

    id = Column(String, primary_key=True)
    transaction_id = Column(String, ForeignKey("transactions.id"), nullable=False)
    event_type = Column(String, nullable=False)  # created, processing, completed, failed
    previous_status = Column(String, nullable=True)
    new_status = Column(String, nullable=False)
    provider_data = Column(JSON, nullable=True)  # Respuesta sin procesar del proveedor
    created_at = Column(DateTime, server_default=func.now())

async def transition_payment_status( transaction_id: str, new_status: str, provider_data: dict = None, ): tx = await get_transaction(transaction_id) previous = tx.status BLANK event = PaymentEvent( id=generate_id(), transaction_id=transaction_id, event_type=f"status_{new_status}", previous_status=previous, new_status=new_status, provider_data=provider_data, ) db.add(event) BLANK tx.status = new_status await db.commit() ```

12. Falta de monitoreo de salud

No habia endpoint de verificacion de salud ni monitoreo de salud del proveedor. Si Stripe se caia, el sistema seguiria enrutando pagos a Stripe y devolviendo errores.

python# DESPUES: Endpoint de verificacion de salud
@router.get("/health")
async def health_check():
    checks = {
        "database": await check_database_health(),
        "redis": await check_redis_health(),
        "providers": await check_provider_health(),
    }

    all_healthy = all(c["status"] == "healthy" for c in checks.values())

    return {
        "status": "healthy" if all_healthy else "degraded",
        "checks": checks,
        "timestamp": datetime.utcnow().isoformat(),
    }

async def check_provider_health() -> dict: """Verificar la salud de todos los proveedores activos.""" results = {} for name, provider in provider_registry.items(): try: status = await asyncio.wait_for( provider.health_check(), timeout=5.0 ) results[name] = {"status": "healthy", "latency_ms": status.latency} except (asyncio.TimeoutError, Exception) as e: results[name] = {"status": "unhealthy", "error": str(e)} BLANK healthy_count = sum(1 for r in results.values() if r["status"] == "healthy") return { "status": "healthy" if healthy_count == len(results) else "degraded", "providers": results, } ```

La hoja de ruta de remediacion en cuatro fases

Priorizamos las correcciones en cuatro fases:

La Fase 1 se completo en horas despues de la auditoria. Estos eran problemas que podian causar dano financiero. La Fase 2 se completo dentro de la primera semana. La Fase 3 se alineo con la migracion planificada a PostgreSQL. La Fase 4 se incorporo al trabajo de preparacion para produccion.

Lo que aprendimos

La velocidad y la seguridad estan en tension, y la velocidad usualmente gana al principio. Las primeras 60 sesiones priorizaron el desarrollo de funcionalidades. La seguridad no se ignoro, pero no fue el foco principal. La auditoria mostro el costo de esa compensacion. Para una plataforma de pagos, deberiamos haber realizado la auditoria despues de la sesion 30, no la 60.

El SQL sin procesar es una responsabilidad. Cada consulta SQL sin procesar es un vector potencial de inyeccion. La migracion al ORM elimino toda una clase de vulnerabilidades. Si estas construyendo un nuevo proyecto, comienza con ORM desde el primer dia.

Los salts de cifrado estaticos anulan el proposito del cifrado. Un salt estatico es solo marginalmente mejor que no tener salt. Siempre usa salts aleatorios y anteponlos al texto cifrado.

La verificacion de webhooks no es opcional para plataformas de pago. Un webhook no verificado puede ser falsificado, causando que el sistema marque un pago como completado cuando no lo fue. Cada webhook de proveedor debe ser verificado.

Publica tus hallazgos de seguridad. El instinto es ocultar las debilidades. Pero la comunidad fintech se beneficia de la transparencia. Si este articulo evita que otro equipo cometa los mismos errores, valio la pena publicarlo.

Este articulo es parte de la serie "Como construimos 0fee.dev". 0fee.dev es un orquestador de pagos que cubre mas de 53 proveedores en mas de 200 paises, construido por Juste A. GNIMAVO y Claude desde Abiyan sin ningun ingeniero humano. Sigue la serie para conocer la historia completa de construccion.