Validación de input y prevención de SQL injection

La entrada del usuario es el vector de ataque más explotado en aplicaciones web, y Python no te protege de él por defecto. Que tu aplicación use un ORM moderno o valide tipos con Pydantic no significa que estés a salvo — la mayoría de las vulnerabilidades aparecen en los bordes donde el input externo toca la lógica interna.

El problema fundamental es que los datos externos son strings sin semántica. Tu aplicación les asigna significado: “esto es un nombre de usuario”, “esto es una ruta de archivo”, “esto es un filtro de búsqueda”. Si no validas ese significado antes de actuar sobre él, el atacante controla cómo tu código interpreta sus datos.

Hay tres capas donde esto explota: validación de estructura y tipos, construcción de queries SQL, y manejo de rutas del sistema de archivos.

Pydantic como primer filtro

Pydantic actúa como contrato entre el mundo exterior y tus funciones internas. Cuando recibes JSON de una API o parámetros de un formulario, Pydantic verifica que la estructura y los tipos sean los esperados antes de que el dato entre al sistema. No es una solución de seguridad completa — no detecta SQL injection ni path traversal — pero sí elimina una clase entera de errores donde input malformado llega a lugares donde no debería.

Lo que debes combinar: Pydantic valida estructura, y tú validas semántica (valores permitidos, rangos, rutas seguras) después.

SQL injection: el ORM no te salva siempre

SQLAlchemy es seguro cuando usas su API de alto nivel (.filter(), .where() con columnas tipadas). El problema aparece cuando necesitas SQL crudo y usas text() con f-strings. En ese momento has cortocircuitado todas las protecciones del ORM y estás concatenando strings directamente — exactamente lo que hace vulnerable al código legacy en cualquier lenguaje.

La regla es simple: nunca interpoles variables en strings SQL. Siempre usa parámetros vinculados (bindparams en SQLAlchemy, o el segundo argumento de cursor.execute). El driver envía la query y los parámetros por separado; la base de datos los trata como datos, no como código.

El problema de path traversal funciona igual: si construyes una ruta con os.path.join(base_dir, user_input) sin verificar que el resultado sigue dentro del directorio permitido, un atacante puede usar ../../etc/passwd para salir del directorio base.

import os
from pathlib import Path
from typing import Annotated

import pydantic
from pydantic import BaseModel, field_validator, StringConstraints
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

# ── Modelos Pydantic ──────────────────────────────────────────────────────────

ShortStr = Annotated[str, StringConstraints(min_length=1, max_length=100, strip_whitespace=True)]


class UserSearchRequest(BaseModel):
    username: ShortStr
    # Pydantic fuerza que 'limit' sea int; un string "DROP TABLE" falla aquí
    limit: Annotated[int, pydantic.Field(ge=1, le=100)] = 10

    @field_validator("username")
    @classmethod
    def no_special_chars(cls, v: str) -> str:
        # Validación semántica: solo alfanumérico y guiones bajos
        if not v.replace("_", "").isalnum():
            raise ValueError("username solo puede contener letras, números y _")
        return v


class FileRequest(BaseModel):
    filename: ShortStr

    @field_validator("filename")
    @classmethod
    def no_path_separators(cls, v: str) -> str:
        # Rechaza cualquier intento de path traversal antes de tocar el filesystem
        if "/" in v or "\\" in v or ".." in v:
            raise ValueError("filename no puede contener separadores de ruta")
        return v


# ── Capa de acceso a datos ────────────────────────────────────────────────────

engine = create_engine("sqlite:///app.db", echo=False)


def search_users_safe(request: UserSearchRequest) -> list[dict]:
    """Usa parámetros vinculados: la DB recibe query y datos por separado."""
    with Session(engine) as session:
        # :username es un placeholder; bindparams() lo enlaza al valor real.
        # SQLite/PostgreSQL/MySQL nunca interpretan el valor como SQL.
        stmt = text(
            "SELECT id, username, email FROM users "
            "WHERE username = :username "
            "LIMIT :limit"
        ).bindparams(username=request.username, limit=request.limit)

        rows = session.execute(stmt).mappings().all()
        return [dict(row) for row in rows]


def search_users_VULNERABLE(username: str) -> list[dict]:
    """NUNCA hagas esto. Dejado aquí solo para mostrar el patrón peligroso."""
    with Session(engine) as session:
        # ❌ Un username como "' OR '1'='1" devuelve todos los registros
        stmt = text(f"SELECT id, username FROM users WHERE username = '{username}'")
        return list(session.execute(stmt).mappings().all())


# ── Acceso seguro al filesystem ───────────────────────────────────────────────

UPLOAD_DIR = Path("/var/app/uploads").resolve()


def read_user_file(request: FileRequest) -> bytes:
    """
    Construye la ruta y verifica que siga dentro del directorio permitido.
    Path.resolve() expande '..' y symlinks antes de comparar.
    """
    target = (UPLOAD_DIR / request.filename).resolve()

    # Esta verificación es la defensa real contra path traversal.
    # Pydantic rechazó '../' explícito, pero resolve() captura casos
    # donde un symlink apunta fuera del directorio permitido.
    if not target.is_relative_to(UPLOAD_DIR):
        raise PermissionError(
            f"Acceso denegado: {target} está fuera de {UPLOAD_DIR}"
        )

    return target.read_bytes()


# ── Punto de entrada ──────────────────────────────────────────────────────────

def handle_search(raw_username: str, raw_limit: str) -> None:
    try:
        # Pydantic convierte y valida; si raw_limit es "abc", lanza ValidationError
        req = UserSearchRequest(username=raw_username, limit=raw_limit)  # type: ignore[arg-type]
        results = search_users_safe(req)
        print(f"Resultados: {results}")
    except pydantic.ValidationError as e:
        # Nunca expongas e.errors() directamente al cliente en producción
        print(f"Input inválido: {e.error_count()} error(s)")


def handle_file_read(raw_filename: str) -> None:
    try:
        req = FileRequest(filename=raw_filename)
        content = read_user_file(req)
        print(f"Leídos {len(content)} bytes")
    except (pydantic.ValidationError, PermissionError, FileNotFoundError) as e:
        print(f"Error de acceso: {type(e).__name__}")


if __name__ == "__main__":
    handle_search("alice", "10")           # ✅ Normal
    handle_search("' OR '1'='1", "10")    # ✅ Pydantic rechaza por caracteres especiales
    handle_search("alice", "DROP TABLE")  # ✅ Pydantic rechaza: no es int

    handle_file_read("report.pdf")        # ✅ Normal
    handle_file_read("../../etc/passwd")  # ✅ Pydantic rechaza '..'
    handle_file_read("symlink_evil")      # ✅ resolve() + is_relative_to() lo captura

Desglose del código

El modelo UserSearchRequest hace dos cosas distintas que no debes confundir. ShortStr con StringConstraints es validación de formato: longitud mínima, máxima, whitespace eliminado. El field_validator no_special_chars es validación semántica: solo caracteres que tienen sentido en un username. Primero defines la forma del dato, luego su significado.

search_users_safe muestra el patrón correcto con text(). La query tiene :username como placeholder — un nombre simbólico que no tiene nada que ver con la sintaxis SQL. bindparams() conecta ese símbolo con el valor real, y SQLAlchemy le pasa al driver dos cosas separadas: el template de la query y un diccionario de parámetros. El motor de base de datos compila el plan de ejecución con el template y aplica los parámetros después, sin posibilidad de que alteren la estructura de la query.

search_users_VULNERABLE existe para que veas exactamente qué patrón evitar. Con un f-string, el string "' OR '1'='1" produce WHERE username = '' OR '1'='1', que es siempre verdadero. La función devolvería toda la tabla.

read_user_file tiene dos líneas críticas. La primera, (UPLOAD_DIR / request.filename).resolve(), expande la ruta completa incluyendo symlinks. La segunda, target.is_relative_to(UPLOAD_DIR), verifica containment. Estas dos operaciones juntas son necesarias: Pydantic rechazó ../ literal, pero un atacante que haya logrado crear un symlink dentro de tu directorio de uploads podría apuntar fuera de él — resolve() seguido de is_relative_to() captura ese caso también.

Errores que debes conocer

Error: Usar cursor.execute con format strings en lugar del segundo parámetro posicional, confiando en que el ORM “ya valida todo”.

# ❌ Wrong
cursor.execute(f"SELECT * FROM users WHERE name = '{nombre}'")

# ✅ Right
cursor.execute("SELECT * FROM users WHERE name = ?", (nombre,))

El segundo argumento de execute es el mecanismo de parámetros vinculados del driver; el placeholder ? (SQLite/MySQL) o %s (psycopg2) nunca se interpola como SQL, el driver los serializa de forma segura.


Error: Llamar a Path.resolve() sin verificar containment después, asumiendo que os.path.join con un directorio base ya es suficiente.

# ❌ Wrong
def serve_file(filename: str) -> bytes:
    path = Path("/var/uploads") / filename   # "../secret.txt" escapa igual
    return path.read_bytes()

# ✅ Right
def serve_file(filename: str) -> bytes:
    base = Path("/var/uploads").resolve()
    target = (base / filename).resolve()
    if not target.is_relative_to(base):
        raise PermissionError("Path traversal detectado")
    return target.read_bytes()

resolve() normaliza la ruta antes de comparar, así que .. y symlinks quedan expuestos; sin ese paso, is_relative_to puede dar falsos positivos.


Error: Exponer pydantic.ValidationError completo al cliente, filtrando detalles de estructura interna.

# ❌ Wrong
except pydantic.ValidationError as e:
    return {"error": e.errors()}   # Revela nombres de campos, tipos esperados, etc.

# ✅ Right
except pydantic.ValidationError as e:
    return {"error": "Datos de entrada inválidos"}   # Loguea e internamente

Los detalles del error de validación son valiosos para un atacante que está sondeando tu esquema; logéalos internamente pero devuelve siempre mensajes genéricos al exterior.

199

Dejar un comentario

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *

Scroll al inicio