Endpoint POST /api/scripts/{id}/execute · Ejecución de scripts SSH

Descripción

POST /api/scripts/{script_id}/execute — Endpoint que ejecuta un template de script sobre un dispositivo de red específico. Implementa lógica híbrida Cloud/On-Premise:

Si la IP es privada (RFC1918, loopback, link-local, CGNAT, unspecified) → retorna 422 USE_LOCAL_AGENT para que el Agente local lo ejecute en la LAN.
Si es pública → abre sesión SSH desde el servidor hacia el dispositivo y corre los comandos.

Criticidad: ALTA. Ejecuta comandos sobre equipos de producción; todo acceso debe ser autorizado y auditado.

Firma de la función

@scripts_router.post("/{script_id}/execute", response={200: dict, 400: ErrorSchema, 403: ErrorSchema, 422: dict})
def execute_script(request, script_id: int, payload: ScriptExecuteInputSchema):
    """
    Ejecuta un script template sobre un dispositivo de red.
    Híbrida Cloud/On-Premise: local-agent si privada, SSH-server si pública.
    """

Payload

class ScriptExecuteInputSchema(Schema):
    device_id: int          # Device sobre el que ejecutar
    username: str           # Credencial SSH (body)
    password: str           # Credencial SSH (body)

Respuestas

Código	Body	Significado
200	`{"status": "success", "output": str, "executed_on": "LOCAL_AGENT" \| "CLOUD"}`	Ejecución exitosa
400	`{"message": str}`	Validación fallida, org missing, device no encontrado, comando peligroso
403	`{"message": "Unauthorized..."}`	Usuario sin perm `network:edit`
422	`{"error": "USE_LOCAL_AGENT", "ip_address": str, "script_content": str, "vendor": str}`	IP privada → enrutar a Agente local

Autorización y contexto

Gate

require_perm(request, "network", "edit")  # Roles: admin, operator

✅ Admin: ejecuta siempre
✅ Operator: ejecuta siempre
❌ Readonly: bloqueado con 403

Scoping de entidades

org = get_current_org(request)
if not org:
    return 400, {"message": "No Organization Context"}

script = ScriptTemplate.objects.get(id=script_id, organization=org)
device = Device.objects.get(id=device_id, rack__organization=org)

ScriptTemplate se carga por org: solo templates de la org actual o globales (inherited).
Device se carga por org via rack: solo dispositivos de la org actual.
RLS multi-tenant: garantizado por get_current_org() + organization=org.

Flujo de ejecución

Fase 1: Validación

Autorización: require_perm(network, edit) → 403 si falla
Contexto de org: get_current_org() → 400 si None
Cargar script: ScriptTemplate.get(id, org) → 404 si no existe
Cargar device: Device.get(id, rack__org) → 404 si no existe
Parsear config: parse_device_config(device) → dict con management_config
Extraer IP descifrada: get_device_ip(config, decrypt=True) → IP o None si descifrado falla
Validar IP: chequear privada/pública + SSRF guard

Fase 2: Decisión de ruta

ip_is_private = is_private_ip(ip)  # RFC1918 + loopback + link-local + CGNAT + unspecified

if ip_is_private:
    return 422, {
        "error": "USE_LOCAL_AGENT",
        "ip_address": ip,
        "script_content": script.script_content,
        "vendor": vendor
    }

Cubre estos rangos como “privada” (SSRF guard):

RFC1918: 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16
Loopback: 127.0.0.0/8, ::1/128
Link-local/metadata: 169.254.0.0/16, fe80::/10
CGNAT (VPN NetBird del equipo): 100.64.0.0/10
Unspecified: 0.0.0.0, ::/128
Multicast: 224.0.0.0/4, ff00::/8

Fase 3: Ejecución (solo si pública)

# B9: Parsear comandos (multilínea → lista individual)
commands = []
if script.script_content:
    try:
        parsed = json.loads(script.script_content)
        commands = parsed if isinstance(parsed, list) else [str(parsed)]
    except json.JSONDecodeError:
        commands = [ln.strip() for ln in script.script_content.splitlines() if ln.strip()]

vendor = get_device_vendor(config)
port = get_device_port(config)

# A2: Validación de comandos peligrosos
cmds_valid, cmd_error = validate_commands(commands, vendor)
if not cmds_valid:
    return 400, {"message": cmd_error}

# A3: Audit log
log_network_action(request, "execute_script", f"Ran script {script.id} on device {device.id} (vendor={vendor})")

# A5: Ejecutar en modo lectura (send_show_commands), NO config
result = await ScrapliManager.execute_script(
    host=ip, username=payload.username, password=payload.password,
    commands=commands, config=False  # ← Modo show, no config
)

Fase 4: Respuesta

return 200, {
    "status": "success",
    "output": result.output,
    "executed_on": "CLOUD"
}

Seguridad y riesgos mitigados

A1: Autorización unificada (ALTA) ✓

Antes: if request.user.role not in ["admin", "operator"] (debil, fail-open con AttributeError).
Ahora: require_perm(request, "network", "edit") (robusto, valida contra roles + permisos).
Efecto: readonly bloqueado, inconsistencia de patterns eliminada.

A2: Validación de comandos (ALTA) ✓

Antes: Sin filtro; comando injected o peligroso se pasaba directo al equipo.
Ahora: validate_commands(commands, vendor) bloquea reload, shutdown, format, erase, separadores ; |, etc.
Efecto: 400 con mensaje claro ante comando peligroso.

A3: Audit log de ejecución (ALTA) ✓

Antes: Sin rastro; “quién ejecutó qué sobre qué equipo” era invisible.
Ahora: log_network_action(request, "execute_script", f"Ran script {script.id} on device {device.id}").
Efecto: Cada ejecución queda registrada en audit log.

A4: Timeouts SSH (ALTA) ✓

Antes: timeout_transport=3600, timeout_ops=3600 (1 hora); una sesión colgada fijaba un worker ASGI 1 hora.
Ahora: 120 segundos.
Efecto: DoS de sesión colgada limitado a 2 minutos por request.

A5: Modo CONFIG accidental (ALTA) ✓

Antes: ScrapliManager.execute_script() llamaba a send_configs() (modo CONFIG) para vendors conocidos, aunque el script fuera “show”.
Ahora: send_configs() solo si config=True explícito; por defecto send_show_commands() (modo lectura).
Efecto: Un script “show” nunca cambia la config del equipo.

M7/B10: SSRF servidor→infra interna (MEDIA) ✓

Antes: is_private_ip() solo chequeaba ipaddress.is_private() (incompleto; no cubría metadata/CGNAT/link-local en Python < 3.10).
Ahora: Chequea RFC1918 + loopback + link-local + CGNAT 100.64/10 + unspecified/multicast de forma explícita.
Efecto: El servidor nunca abre SSH a 127.0.0.1, 169.254.169.254, 100.64.0.1 ni ::1; siempre va por Local Agent.

M9: Fallback a ciphertext (MEDIA) ✓

Antes: get_device_ip() si decrypt() fallaba, devolvía el token gAAAAA... como si fuera la IP (Regla 15: 200 con fallo).
Ahora: Devuelve None, que se valida después y retorna 400.
Efecto: Un descifrado fallido (clave rotada) no enruta silenciosamente a la IP encriptada.

B3: Error messages (BAJA) ✓

Antes: {"message": str(e)} exponía detalles internos (JSONDecodeError, ValueError de is_private_ip(), etc.).
Ahora: Mensajes genéricos ("Could not execute script").
Efecto: No filtra info interna a logs/cliente.

B13: Logs SSH sin contenido (BAJA) ✓

Antes: write_to_channel() / read_channel() llamaban a logger.debug(repr(data)) → volcaban passwords/configs.
Ahora: logger.debug(f"Wrote {len(data)} bytes") (solo tamaño).
Efecto: Logs no contienen comandos tecleados ni respuestas sensibles.

Interacciones con otros servicios

`ScrapliManager.execute_script()`

Llamado si IP es pública. Abre sesión SSH, autentica y corre comandos.

Input: host, username, password, commands (ya validados aquí).
Output: result con output (stdout del equipo).
Auth: Recibe credenciales del request body (payload.username/password).

`device_config.get_device_ip(config, decrypt=True)`

Extrae la IP de management del device descifrada.

Input: config (management_config parsado).
Output: IP string o None si descifrado falla.
Guard: Si None, retorna 400.

`device_config.is_private_ip(ip)`

Decide si la IP es privada (ruta local) o pública (SSH servidor).

Input: IP string.
Output: bool.
Guard: RFC1918 + loopback + link-local + CGNAT + unspecified/multicast.

`monitoring.services.command_validation.validate_commands(commands, vendor)`

Valida comandos contra whitelist/blocklist.

Input: list de strings (comandos), vendor string.
Output: (bool, str) = (is_valid, error_message).
Guard: Bloquea reload, shutdown, format, separadores ; |.

Deuda y mejoras futuras

Item	Prioridad	Nota
SSH host-key TOFU (auth_strict_key=True)	M3	Diferido a PR hermano con sa1 A9
Sesiones SSH sin cap/TTL	M8	Tracking issue abierto; _sessions es dict de clase sin límite
N+1 en carga de Device.management_config	M6	Optimización futura
Event loop nuevo por request en handler sync	Refactor	Considerar async handler o job asincrono

Tests

12 tests nuevos en tests/api/test_network_sa2.py:

✓ test_execute_script_readonly_denied — Rol readonly → 403
✓ test_execute_script_operator_allowed — Rol operator → 200 o 422
✓ test_execute_script_admin_allowed — Rol admin → 200 o 422
✓ test_dangerous_command_blocked — Comando reload → 400
✓ test_multilne_parse_to_agent — Script multilínea → 422 USE_LOCAL_AGENT
✓ test_audit_log_execute — Log queda registrado
✓ test_is_private_ip_cgnat — CGNAT 100.64/10 → LOCAL_AGENT

Suite API: 361 passed ✓

Véase también

[[entity—network—service—scrapli-manager]]
[[entity—network—service—device-config]]
[[concept—saas—multi-tenancy]]
[[concept—infra—ssrf]]
[[feature—network—scripts-ssh-audit-s113]]