📊 Revisión de Código Python — PR #47
DataFlow Analytics SL · feat/export-reports → main · Autor: Marcos García · 18 Jun 2026
✗ BLOQUEADO
DataFlow Analytics SL
4 archivos .py
+312 / -48
ruff · mypy · bandit · black
3
Critical
5
High
4
Medium
2
Low
14
Total
Salida de herramientas de análisis estático
$ ruff check . && mypy . && bandit -r .
api/routes/reports.py:34: E501 line too long (112 > 88)
api/routes/reports.py:67: S608 possible SQL injection via string-based query construction
db/queries.py:22: S608 possible SQL injection via string-based query construction
db/queries.py:89: S603 subprocess call with shell=True
utils/file_export.py:45: B006 do not use mutable data structures for argument defaults
services/metrics_cache.py:78: ANN201 missing return type annotation
services/metrics_cache.py:113: ANN001 missing type annotation for arg 'ttl'
api/routes/reports.py:88: G004 logging statement uses f-string
Found 8 ruff errors, 0 fixed.

db/queries.py:22: error [arg-type] Argument 1 to "execute" has incompatible type "str"; expected "LiteralString"
services/metrics_cache.py:45: error [return-value] Incompatible return value type (got "Optional[bytes]", expected "dict[str, Any]")
api/routes/reports.py:101: note: By default the bodies of untyped functions are not checked
Found 2 errors in 2 files (checked 4 source files)

[HIGH] CWE-89 db/queries.py:22 sql_injection test_id: B608
[HIGH] CWE-78 db/queries.py:89 subprocess_popen_with_shell_equals_true test_id: B603
[MEDIUM] CWE-703 utils/file_export.py:78 broad_except test_id: B110
Total: 3 issue(s) [3 high(s), 1 medium(s)]
CRITICAL — Seguridad (3)
Inyección SQL por concatenación de f-string
Critical
db/queries.pylínea 22 · CWE-89 · bandit B608

El parámetro user_id recibido de la petición HTTP se interpola directamente en la query SQL usando un f-string. Un atacante puede inyectar SQL arbitrario para leer, modificar o eliminar cualquier dato de la base de datos.

- query = f"SELECT * FROM orders WHERE user_id = '{user_id}' AND store_id = '{store_id}'"
- result = await conn.execute(query)
+ query = "SELECT * FROM orders WHERE user_id = $1 AND store_id = $2"
+ result = await conn.execute(query, user_id, store_id)

Exposición total de la base de datos. Datos PII de todos los clientes accesibles. Posible borrado masivo con '; DROP TABLE orders; --.

Inyección de comandos con shell=True
Critical
db/queries.pylínea 89 · CWE-78 · bandit B603

Se usa subprocess.run(shell=True) con una cadena construida a partir de input del usuario (nombre de archivo de exportación). Permite ejecutar comandos arbitrarios en el servidor.

- subprocess.run(f"psql -c 'COPY ({query}) TO {export_path}'", shell=True)
+ export_path_safe = Path(export_path).resolve()
+ if not export_path_safe.is_relative_to(EXPORT_DIR):
+ raise PermissionError("Path traversal detectado")
+ subprocess.run(["psql", "-c", f"COPY ({query}) TO STDOUT", "-o", str(export_path_safe)], check=True)

RCE (Remote Code Execution). Un atacante con acceso al endpoint puede ejecutar comandos con los permisos del proceso servidor.

Path Traversal en exportación de archivos
Critical
utils/file_export.pylínea 45–52 · CWE-22

El nombre de archivo proporcionado por el usuario se usa sin validación para construir rutas del sistema de ficheros. Un atacante puede usar ../../etc/passwd para leer o sobreescribir archivos fuera del directorio permitido.

- output_path = os.path.join(EXPORT_DIR, filename)
- with open(output_path, 'w') as f:
+ safe_name = Path(filename).name # Elimina cualquier componente de ruta
+ output_path = (EXPORT_DIR / safe_name).resolve()
+ if not output_path.is_relative_to(EXPORT_DIR.resolve()):
+ raise ValueError(f"Nombre de archivo inválido: {filename}")
+ with open(output_path, 'w') as f:
HIGH — Calidad y Tipado (5)
Argumento mutable como valor por defecto
High
utils/file_export.pylínea 78 · bandit B006 / PEP 8

Usar [] como valor por defecto en un parámetro de función crea un único objeto compartido entre todas las llamadas. Las columnas añadidas en una llamada persistirán en llamadas futuras, causando bugs difíciles de reproducir en producción.

- def export_csv(data: list, columns: list = [], delimiter: str = ','):
+ def export_csv(data: list[dict], columns: list[str] | None = None, delimiter: str = ','):
+ if columns is None:
+ columns = list(data[0].keys()) if data else []
Tipo de retorno incompatible en cache.get()
High
services/metrics_cache.pylínea 45 · mypy error [return-value]

El método get_metrics() declara retornar dict[str, Any] pero Redis devuelve Optional[bytes]. El código nunca deserializa el valor de Redis, retornando bytes crudos al caller que espera un dict. Causará AttributeError en runtime.

- def get_metrics(self, key: str) -> dict[str, Any]:
- return self.redis.get(key)
+ def get_metrics(self, key: str) -> dict[str, Any] | None:
+ raw = self.redis.get(key)
+ if raw is None:
+ return None
+ return json.loads(raw.decode('utf-8'))
N+1 queries en loop de métricas por tienda
High
api/routes/reports.pylínea 112–128

El endpoint de informe itera sobre cada store_id y lanza una query separada a la base de datos. Con 50 tiendas esto genera 50 queries consecutivas. A escala (clientes con cientos de tiendas) esto colapsa la base de datos.

- for store_id in store_ids:
- metrics = await db.get_store_metrics(store_id) # N queries
- results.append(metrics)
+ # Batch query — 1 sola petición a la base de datos
+ results = await db.get_bulk_store_metrics(store_ids) # WHERE store_id = ANY($1)
Excepción demasiado amplia silencia fallos reales
High
utils/file_export.pylínea 78 · bandit B110 / PEP 8

Un bloque except Exception: pass oculta cualquier error — incluyendo errores de I/O, permisos, encoding y desbordamientos de disco. El usuario recibirá un CSV vacío sin ningún mensaje de error.

- try:
- writer.writerows(data)
- except Exception:
- pass
+ try:
+ writer.writerows(data)
+ except (IOError, OSError) as exc:
+ logger.error("Error escribiendo CSV: %s", exc)
+ raise ExportError("No se pudo escribir el archivo de exportación") from exc
Función de 78 líneas sin tipado — refactorizar
High
api/routes/reports.pylínea 55–133 · >50 líneas · 7 parámetros

La función generate_report() tiene 78 líneas, 7 parámetros y no tiene type hints en ninguno de sus argumentos ni en el retorno. Viola el límite de 50 líneas y 5 parámetros. Refactorizar usando un dataclass ReportConfig.

- async def generate_report(store_id, start_date, end_date, format, columns, filters, timezone):
+ @dataclass
+ class ReportConfig:
+ store_id: str
+ start_date: datetime
+ end_date: datetime
+ format: ExportFormat # Enum, no str mágico
+ columns: list[str] | None = None
+ filters: dict[str, Any] = field(default_factory=dict)
+ timezone: str = "UTC"
+
+ async def generate_report(config: ReportConfig) -> ExportResult:
MEDIUM — Buenas prácticas (4)
f-string en logging.info() — usar % formatting
Medium
api/routes/reports.pylínea 88 · ruff G004

Los f-strings en logging se evalúan siempre, incluso si el nivel de log está desactivado. Usar %s deferrido evita la concatenación innecesaria de strings en producción.

- logger.info(f"Exportando informe para store_id={store_id}, filas={len(data)}")
+ logger.info("Exportando informe para store_id=%s, filas=%d", store_id, len(data))
Comparación con None usando == en vez de is
Medium
services/metrics_cache.pylíneas 67, 89, 102 · PEP 8 E711

3 ocurrencias de value == None. En Python, None es un singleton; la comparación correcta y eficiente es value is None. El operador == puede ser sobreescrito por __eq__ en objetos personalizados.

- if cached_value == None:
+ if cached_value is None:
Números mágicos sin constantes nombradas
Medium
services/metrics_cache.pylíneas 34, 56

TTL hardcodeados como 3600 y 86400 sin explicación. Si el negocio decide cambiar la caché de métricas, no es obvio qué valores cambiar ni por qué.

- self.redis.setex(key, 3600, value)
+ METRICS_CACHE_TTL_SECONDS: Final[int] = 3600 # 1 hora — métricas en tiempo real
+ DAILY_REPORT_CACHE_TTL: Final[int] = 86400 # 24 horas — informes históricos
+ self.redis.setex(key, METRICS_CACHE_TTL_SECONDS, value)
Docstrings ausentes en funciones públicas de exportación
Medium
utils/file_export.pytodas las funciones públicas

Ninguna de las 5 funciones públicas (export_csv, export_excel, export_pdf, sanitize_filename, compress_export) tiene docstring. En un módulo que maneja PII, el contrato de cada función debe estar documentado.

LOW — Estilo (2)
Orden de imports no sigue PEP 8 (isort)
Low
api/routes/reports.pylíneas 1–12

Los imports de stdlib, terceros y propios están mezclados sin separación. Añadir isort al pipeline de CI o configurar ruff --select I.

print() en vez de logger en utils/file_export.py
Low
utils/file_export.pylíneas 23, 67

2 print() statements usados para debug que llegaron al PR. En producción no hay forma de desactivarlos ni de capturarlos en el sistema de logging centralizado.

- print(f"DEBUG: exportando {len(data)} filas")
+ logger.debug("Exportando %d filas", len(data))
✗ PR BLOQUEADO — No apto para merge
Este PR tiene 3 vulnerabilidades de seguridad críticas que deben resolverse antes de cualquier merge a main. El código maneja datos PII de clientes y tiene acceso directo a la base de datos de producción.
Checklist para nueva revisión
☐ Parametrizar todas las queries SQL
☐ Eliminar shell=True del subprocess
☐ Validar rutas con Path.is_relative_to()
☐ Deserializar bytes de Redis a dict
☐ Crear dataclass ReportConfig
☐ Añadir query batch para N+1
☐ Reemplazar bare except por excepción específica
☐ Quitar mutable default argument []
☐ Reemplazar f-strings en logging
☐ Cambiar == None a is None (×3)
☐ Extraer constantes TTL nombradas
☐ Añadir docstrings a funciones públicas
☐ Corregir orden de imports (isort)
☐ Eliminar print() de debug
Revisión generada por Revisor Experto de Código Python · CULTIVA IA · DataFlow Analytics SL · PR #47 · 18 Jun 2026