PayloadFlow — Análisis de Grafo de Código para Auditoría de Seguridad

247

Funciones analizadas

64 módulos · 4 paquetes

Nodos taintados

7.3% del grafo total

Fronteras de privilegio

2 sin validación previa

Radio de explosión máx.

execute_raw_query()

1 Grafo de Llamadas — Rutas de Ataque

🕸️ Call Graph — Subgrafo entrypoint_reachable × tainted engine.preanalysis() completo · 247 nodos · 389 aristas

Entrypoint HTTP

Nodo taintado

Frontera de privilegio

Sink / alto blast radius

Seguro (autenticado)

Arista normal

- - - 

Arista incierta (async/dynamic)

2 Propagación de Taint — Rutas Críticas

💉 Ruta de Taint #1 — SQL Injection CRÍTICO

🌐 POST /webhooks/{id}/trigger trust=untrusted

↓

☣️ trigger_webhook_delivery(request) tainted · complexity=12

↓ request.body propagado sin sanitizar

☣️ build_filter_query(payload) tainted · complexity=8

↓ f-string SQL interpolation

💀 execute_raw_query(f"...{filter}...") SINK SQL · blast=31

# src/core/webhook_engine.py:184 — VULNERABLE
def build_filter_query(payload: dict) -> str:
    filter_val = payload.get("filter")  # UNTRUSTED
    return execute_raw_query(
        f"SELECT * FROM events WHERE tag='{filter_val}'"
    )  # ⚠ SQLi vía f-string directo

🔐 Ruta de Taint #2 — Privilege Escalation CRÍTICO

🌐 POST /teams/invite trust=untrusted

↓

☣️ process_invite(body) tainted · complexity=5

↓ role param sin validación de enum

🔐 check_team_permission(user, role) privilege_boundary

↓ bypass: role="admin" acepta string libre

👑 promote_to_admin(user_id) SINK PRIV · blast=11

# src/api/teams.py:67 — VULNERABLE
def process_invite(body: dict):
    role = body.get("role", "member")  # string libre
    if check_team_permission(current_user, role):
        promote_to_admin(body["invited_user_id"])
# ⚠ role debería ser Enum[member|owner] validado

3 Radio de Explosión — Top 8 Funciones

💥 Blast Radius estimado por engine.preanalysis() Nodos downstream afectados si la función es comprometida

execute_raw_query() SINK SQL 31 nodos / 89% del módulo db

export_data_csv() ALTO 18 nodos

update_user_role() PRIV 11 nodos

trigger_webhook_delivery() 9 nodos

process_billing_event() 7 nodos

bulk_export_query() 7 nodos

get_user_data() 5 nodos

session.commit() BAJO 2 nodos

4 Fronteras de Privilegio Detectadas

🛡️ Subgrafo privilege_boundary — 4 cruces detectados

Caller (trust level)	Callee (trust level)	Función	Validación previa	Riesgo
untrusted	admin	promote_to_admin()	❌ Ninguna — role es string libre	CRÍTICO
untrusted	privileged	update_user_role()	❌ check_team_permission() bypassable	CRÍTICO
authenticated	privileged	bulk_export_query()	⚠ Requiere role=admin pero sin 2FA	ALTO
stripe_signed	privileged	process_billing_event()	✅ HMAC signature verificado	BAJO

5 Hotspots de Complejidad Ciclomática (threshold=10)

🌡️ Funciones con complejidad ≥ 10 — prioritarios para revisión manual engine.complexity_hotspots(threshold=10)

Función	Módulo	Complejidad	Taint	Blast	Prioridad
export_data_csv()	core/export.py:312	14	Parcial	18	P1
process_billing_event()	core/billing.py:88	11	No	7	P3
trigger_webhook_delivery()	core/webhook_engine.py:201	12	Sí	9	P1
check_team_permission()	api/teams.py:134	9	Sí	11	P2
validate_payload()	core/validator.py:45	7	No	3	P4

6 Hallazgos de Seguridad — Priorizados

🚨 Findings — subgrafo tainted × privilege_boundary

#	Hallazgo	Ubicación	CWE	Vector	Severidad
F-01	SQL Injection via f-string en build_filter_query() Datos de request.body llegan interpolados directamente en SQL raw	webhook_engine.py:184	CWE-89	HTTP POST /webhooks	CRÍTICO 9.8
F-02	Privilege Escalation — role param sin enum validation Cualquier usuario puede enviar role="admin" y ser promovido	teams.py:67	CWE-269	HTTP POST /teams/invite	CRÍTICO 9.1
F-03	IDOR en GET /users/{id}/profile — sin ownership check get_user_data() no verifica que el id pertenezca al usuario autenticado	api/users.py:31	CWE-639	HTTP GET /users/{id}	ALTO 7.5
F-04	bulk_export_query() — raw SQL sin parámetros bind Aunque protegido por role=admin, usa concatenación en date_range	db/export_repo.py:219	CWE-89	GET /admin/export	ALTO 6.5
F-05	Falta 2FA en rutas admin de exportación masiva bulk_export puede exfiltrar toda la DB sin step-up auth	api/admin.py:98	CWE-308	GET /admin/export	MEDIO 5.3

7 Recomendaciones — Plan de Remediación

🔴 [F-01] Migrar a queries parametrizadas ORM

Reemplazar execute_raw_query(f"...") con session.query(Event).filter(Event.tag == filter_val) o usar text() con parámetros bind. Eliminar el método raw completamente del API público.

🔴 [F-02] Validar role con Enum tipado

Introducir class TeamRole(str, Enum): member="member"; owner="owner" y usar Pydantic para validar el body. La función check_team_permission() debe rechazar strings no reconocidos antes de cualquier write.

🟠 [F-03] Agregar ownership check en get_user_data()

Verificar if user_id != current_user.id and not current_user.is_admin: raise 403 antes de la query. Añadir test de regresión IDOR con usuario B accediendo a recurso de usuario A.

🟠 [F-04] Parametrizar bulk_export_query()

Usar session.execute(text("... WHERE date >= :start"), {"start": date_start}). El blast_radius=18 hace que este sink sea de alta prioridad aunque esté detrás de auth admin.

🟡 [F-05] Step-up auth (2FA) para /admin/export

Requerir re-autenticación TOTP para cualquier operación de exportación masiva. Implementar @require_step_up_auth(minutes=15) decorator en las rutas admin críticas.

🔵 Ejecutar genotoxic skill post-remediación

Tras los fixes, correr mutation testing sobre validate_payload() y el nuevo EnumValidator para confirmar cobertura de los controles. Usar engine.preanalysis() actualizado como input al genotoxic skill.

8 Código de Análisis Ejecutado

⚙️ Script de auditoría — trailmark API Python · uv run · ~2.3s de análisis

from trailmark.parse import detect_languages, supported_languages
from trailmark.query.api import QueryEngine
from trailmark.models import AnnotationKind

TARGET = "/srv/payloadflow/src"

# 1. Detectar lenguajes y construir grafo completo
langs = detect_languages(TARGET)  # → ['python', 'typescript']
engine = QueryEngine.from_directory(TARGET, language="python")
# 247 nodos · 389 aristas en 2.3s

# 2. Pre-análisis (blast radius, taint, privilege, entrypoints)
result = engine.preanalysis()
# Subgrafos generados: tainted(18), high_blast_radius(6),
#   privilege_boundary(4), entrypoint_reachable(247)

# 3. Identificar sinks taintados
tainted = engine.subgraph("tainted")
for node_id in tainted:
    blast = engine.annotations_of(node_id, kind=AnnotationKind.BLAST_RADIUS)
    callers = engine.callers_of(node_id)
    paths = engine.paths_between("trigger_webhook", node_id)

# 4. Cruce taint × privilege_boundary → candidatos críticos
priv = engine.subgraph("privilege_boundary")
critical = tainted & priv  # → {promote_to_admin, check_team_permission}

# 5. Anotar hallazgos
engine.annotate("build_filter_query", AnnotationKind.FINDING,
             "SQLi via f-string — CWE-89 CRÍTICO", source="audit")
engine.annotate("promote_to_admin", AnnotationKind.FINDING,
             "Privilege escalation — CWE-269 CRÍTICO", source="audit")

# 6. Hotspots de complejidad
hotspots = engine.complexity_hotspots(threshold=10)
# → export_data_csv(14), trigger_webhook(12), process_billing(11)

# 7. Exportar para report
summary = engine.summary()
graph_json = engine.to_json()

Análisis de Grafo de Código — PayloadFlow SaaS