Revisión FastAPI — TalentoFlow API v1.0.0

Hallazgos Críticos — Bloquean despliegue

CRITICAL Credencial de base de datos hardcodeada en el código fuente #C-01

📄 Archivo

app/routers/bookings.py:6

⚠ Problema

La variable DATABASE_URL contiene usuario, contraseña y hostname de producción directamente en el código fuente. Cualquier acceso al repositorio (colaboradores, CI logs, GitHub history) expone las credenciales completas. En caso de breach del repo, la base de datos queda comprometida de inmediato.

✅ Fix

Mover la URL a una variable de entorno y cargarla mediante pydantic-settings. Nunca commitear secrets.

# ❌ Antes (bookings.py:6)
DATABASE_URL = "postgresql://admin:SuperSecret123@prod-db.talentoflow.com:5432/talentoflow"

# ✅ Después — app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str

    class Config:
        env_file = ".env"

settings = Settings()

# ✅ Después — .env (añadir a .gitignore)
DATABASE_URL=postgresql://admin:SuperSecret123@prod-db.talentoflow.com:5432/talentoflow

CRITICAL Inyección SQL mediante interpolación de strings (3 endpoints) #C-02

📄 Archivos

app/routers/bookings.py:17, 26, 34

⚠ Problema

Los tres endpoints construyen queries SQL con f-strings interpolando directamente valores del usuario (user.id, booking.notes, booking_id). Un atacante puede ejecutar SQL arbitrario pasando payloads como notes="'; DROP TABLE bookings; --". Nivel de riesgo máximo: pérdida o corrupción total de datos de producción.

✅ Fix

Usar siempre parámetros parametrizados (%s) de psycopg2 o migrar a SQLAlchemy con ORM/Core que parametriza automáticamente.

# ❌ Antes — vulnerable a SQL injection
cur.execute(f"INSERT INTO bookings (...) VALUES ({user.id}, '{booking.notes}'...)")

# ✅ Después — parámetros psycopg2
cur.execute(
    "INSERT INTO bookings (user_id, talent_id, date, notes) VALUES (%s, %s, %s, %s) RETURNING id, user_id, talent_id, date, status",
    (user.id, booking.talent_id, booking.date, booking.notes)
)

# ✅ Cancel — validar ownership antes de actualizar
cur.execute(
    "UPDATE bookings SET status='cancelled' WHERE id=%s AND user_id=%s",
    (booking_id, user.id)
)

CRITICAL Secret JWT de desarrollo hardcodeado y excepciones JWT silenciadas #C-03

📄 Archivo

app/auth.py:5, 11

⚠ Problema

El secreto "dev-secret-key-do-not-use-in-prod" está en el código; cualquier persona con acceso al repo puede forjar tokens JWT válidos. Además, el except: sin tipo silencia todas las excepciones (incluida la de expiración ExpiredSignatureError), por lo que tokens expirados pueden seguir siendo aceptados dependiendo del comportamiento del decoder.

✅ Fix

Leer el secreto desde variable de entorno, capturar sólo las excepciones esperadas de PyJWT y verificar explícitamente la expiración.

import jwt
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
from app.config import settings

async def get_current_user(authorization: str = Header(...)):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid authorization header")
    token = authorization[7:]
    try:
        payload = jwt.decode(
            token, settings.jwt_secret,
            algorithms=["HS256"],
            options={"require": ["exp", "sub"]}
        )
    except ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return UserContext(id=payload["sub"], email=payload.get("email"))

Hallazgos Altos — Corregir antes de producción

HIGH Cliente de BD bloqueante (psycopg2 síncrono) dentro de rutas async #H-01

📄 Archivo

app/routers/bookings.py:10-12

⚠ Problema

psycopg2.connect() es completamente síncrono. Invocado dentro de handlers async def, bloquea el event loop de Uvicorn durante toda la operación de BD. Con alta concurrencia, el servidor queda efectivamente colgado. El correcto uso de FastAPI async requiere asyncpg o SQLAlchemy async.

✅ Fix

Migrar a asyncpg + sqlalchemy[asyncio] con AsyncSession como dependencia, o usar rutas def (no async def) si se mantiene psycopg2 síncrono — FastAPI ejecutará las rutas síncronas en un threadpool automáticamente.

HIGH Sesiones de BD creadas inline en cada handler (sin gestión de ciclo de vida) #H-02

📄 Archivo

app/routers/bookings.py:14, 23, 31

⚠ Problema

Cada handler llama a get_db() que abre una nueva conexión sin cerrarla nunca ni gestionar errores con try/finally. Las conexiones se acumulan, superando rápidamente el pool máximo de PostgreSQL (~100 por defecto), causando errores "too many connections" en producción.

✅ Fix

Convertir get_db() en un generador de FastAPI Depends con yield que cierra la conexión garantizadamente.

from contextlib import contextmanager

def get_db():
    conn = psycopg2.connect(settings.database_url)
    try:
        yield conn
    finally:
        conn.close()

# En el handler:
async def create_booking(booking: BookingCreate, db=Depends(get_db), user=...): ...

HIGH CORS con allow_origins=["*"] + allow_credentials=True (configuración prohibida) #H-03

📄 Archivo

app/main.py (CORS middleware)

⚠ Problema

La especificación CORS prohíbe esta combinación (wildcard + credenciales). Los navegadores la rechazan y muchos proxies la bloquean. Además, en APIs con auth, nunca se debe permitir cualquier origen; se abre la puerta a ataques CSRF desde dominios arbitrarios.

✅ Fix

Especificar orígenes permitidos explícitamente desde variables de entorno.

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,  # e.g. ["https://app.talentoflow.com"]
    allow_credentials=True,
    allow_methods=["GET", "POST", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

Hallazgos Medios — Sprint siguiente

MEDIUM Campos internos admin_flag e internal_notes expuestos en BookingOut #M-01

📄 Archivo

app/schemas.py:16-17

⚠ Problema

BookingOut incluye admin_flag e internal_notes, que son campos de uso interno. Al usarse como response_model, Pydantic los serializará y el cliente los recibirá, filtrando información de gestión interna al usuario final.

✅ Fix

Separar el schema en BookingInDB (interno, con todos los campos) y BookingOut (público, sin campos admin). Usar BookingOut como response_model.

MEDIUM GET /bookings sin paginación ni verificación de ownership en DELETE #M-02

📄 Archivos

app/routers/bookings.py:22, 30

⚠ Problema

(a) El listado no tiene límite ni offset: con muchas reservas devuelve todo el histórico sin límite de memoria. (b) El DELETE actualiza por booking_id sin verificar que pertenezca al user.id autenticado; cualquier usuario autenticado puede cancelar reservas ajenas conociendo el ID.

✅ Fix

Añadir parámetros limit/offset al listado (default 20/500 máx). Agregar AND user_id = %s al UPDATE de cancelación y devolver 404 si no se actualizó ninguna fila.

🧪 Tests verificados

⏭ pytest — omitido: no existe directorio tests/ en el proyecto
⏭ ruff check . — omitido: sin configuración pyproject.toml
⏭ mypy app/ — omitido: sin mypy.ini ni stubs de BD
✅ Análisis estático manual de los 5 archivos fuente proporcionados
✅ Revisión de schemas Pydantic contra response_models
✅ Validación de patrones async/sync contra Uvicorn event loop

⚠ Riesgo residual

🔍 No se pudo verificar si hay más endpoints en otros routers con los mismos patrones
🔍 Sin acceso a la configuración de Uvicorn/Gunicorn (workers, timeouts)
🔍 No se revisó la configuración del pool de conexiones en producción
🔍 Rate limiting ausente — no evaluado (depende de infra/proxy)
🔍 Logging de seguridad (failed auth, anomalías) no visible en el código revisado

Revisión: TalentoFlow API v1.0.0

🧪 Tests verificados

⚠ Riesgo residual