""" auditoria_red_logitrans.py

Script de auditoría de red para LogiTrans SA — SOLO LECTURA. Recoge estado de interfaces, tabla de rutas y versión IOS en todos los dispositivos de las 3 sedes. Exporta CSV + log de errores.

Uso: export NETMIKO_USERNAME=operador export NETMIKO_PASSWORD=secreto export NETMIKO_ENABLE_SECRET=enable_pass python auditoria_red_logitrans.py

Dependencias: netmiko>=4.3, textfsm>=1.1 """

import csv import logging import os from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from getpass import getpass from pathlib import Path from typing import Any

from netmiko import ConnectHandler from netmiko.exceptions import ( NetmikoAuthenticationException, NetmikoTimeoutException, ReadTimeout, )

---------------------------------------------------------------------------

Logging

---------------------------------------------------------------------------

LOG_DIR = Path("logs") LOG_DIR.mkdir(exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(LOG_DIR / f"auditoria_{timestamp}.log"), logging.StreamHandler(), ], ) log = logging.getLogger(name)

---------------------------------------------------------------------------

Inventario (sin credenciales; se inyectan desde entorno en runtime)

---------------------------------------------------------------------------

INVENTARIO: list[dict[str, Any]] = [ # Madrid {"host": "192.0.2.10", "hostname": "MAD-RTR-01", "sede": "Madrid", "role": "router", "device_type": "cisco_ios"}, {"host": "192.0.2.11", "hostname": "MAD-SW-01", "sede": "Madrid", "role": "switch", "device_type": "cisco_ios"}, {"host": "192.0.2.12", "hostname": "MAD-SW-02", "sede": "Madrid", "role": "switch", "device_type": "cisco_ios"}, # Barcelona {"host": "192.0.2.20", "hostname": "BCN-RTR-01", "sede": "Barcelona","role": "router", "device_type": "cisco_ios"}, {"host": "192.0.2.21", "hostname": "BCN-SW-01", "sede": "Barcelona","role": "switch", "device_type": "cisco_ios"}, # Valencia {"host": "192.0.2.30", "hostname": "VLC-RTR-01", "sede": "Valencia", "role": "router", "device_type": "cisco_ios"}, {"host": "192.0.2.31", "hostname": "VLC-SW-01", "sede": "Valencia", "role": "switch", "device_type": "cisco_ios"}, ]

Comandos de solo lectura por tipo de dispositivo

COMMANDS_ROUTER = [ "show ip interface brief", "show version", "show ip route summary", ] COMMANDS_SWITCH = [ "show ip interface brief", "show version", "show interfaces status", ]

MAX_WORKERS = 4 # max conexiones concurrentes CONN_TIMEOUT = 10 # segundos para establecer SSH AUTH_TIMEOUT = 20 READ_TIMEOUT = 30

---------------------------------------------------------------------------

Credenciales desde entorno (nunca en código fuente)

---------------------------------------------------------------------------

def get_credentials() -> dict[str, str]: username = os.environ.get("NETMIKO_USERNAME") or input("Usuario SSH: ") password = os.environ.get("NETMIKO_PASSWORD") or getpass("Contraseña SSH: ") enable = os.environ.get("NETMIKO_ENABLE_SECRET") or getpass("Enable secret (Enter para omitir): ") return {"username": username, "password": password, "secret": enable or None}

---------------------------------------------------------------------------

Recolección de un dispositivo (read-only)

---------------------------------------------------------------------------

def auditar_dispositivo(device_info: dict[str, Any], creds: dict[str, str]) -> dict[str, Any]: hostname = device_info["hostname"] host = device_info["host"] role = device_info["role"] sede = device_info["sede"]

conn_params = {
    "device_type":          device_info["device_type"],
    "host":                 host,
    "username":             creds["username"],
    "password":             creds["password"],
    "secret":               creds.get("secret"),
    "conn_timeout":         CONN_TIMEOUT,
    "auth_timeout":         AUTH_TIMEOUT,
    "banner_timeout":       15,
    "read_timeout_override": READ_TIMEOUT,
}

commands = COMMANDS_ROUTER if role == "router" else COMMANDS_SWITCH
result: dict[str, Any] = {
    "hostname": hostname,
    "host":     host,
    "sede":     sede,
    "role":     role,
    "ok":       False,
    "error":    None,
    "data":     {},
}

try:
    log.info("Conectando a %s (%s)…", hostname, host)
    with ConnectHandler(**conn_params) as conn:
        if conn_params.get("secret") and not conn.check_enable_mode():
            conn.enable()

        for cmd in commands:
            # Intentar parseo TextFSM; si no hay plantilla, guardar raw
            parsed = conn.send_command(
                cmd,
                use_textfsm=True,
                raise_parsing_error=False,
                read_timeout=READ_TIMEOUT,
            )
            result["data"][cmd] = parsed

    result["ok"] = True
    log.info("  [OK] %s — %d comandos recogidos", hostname, len(commands))

except NetmikoAuthenticationException:
    result["error"] = "AuthError"
    log.error("  [ERR] %s — fallo de autenticación", hostname)
except NetmikoTimeoutException:
    result["error"] = "TimeoutConnection"
    log.error("  [ERR] %s — timeout al conectar", hostname)
except ReadTimeout:
    result["error"] = "TimeoutRead"
    log.error("  [ERR] %s — timeout en lectura de comando", hostname)
except Exception as exc:  # noqa: BLE001
    result["error"] = f"Unexpected:{type(exc).__name__}"
    log.error("  [ERR] %s — error inesperado: %s", hostname, exc)

return result

---------------------------------------------------------------------------

Batch concurrente (max MAX_WORKERS simultáneos)

---------------------------------------------------------------------------

def auditar_todos(inventario: list[dict], creds: dict) -> list[dict]: resultados: list[dict] = [] with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: futures = { pool.submit(auditar_dispositivo, dev, creds): dev["hostname"] for dev in inventario } for future in as_completed(futures): hostname = futures[future] try: resultados.append(future.result()) except Exception as exc: # noqa: BLE001 log.error("Future inesperado para %s: %s", hostname, exc) resultados.append({ "hostname": hostname, "ok": False, "error": str(exc), "host": "?", "sede": "?", "role": "?", "data": {}, }) return resultados

---------------------------------------------------------------------------

Análisis de interfaces caídas

---------------------------------------------------------------------------

def analizar_interfaces(resultados: list[dict]) -> list[dict]: alertas = [] for r in resultados: if not r["ok"]: continue ifaces = r["data"].get("show ip interface brief") if not isinstance(ifaces, list): continue # sin parseo estructurado for iface in ifaces: status = iface.get("status", "").lower() proto = iface.get("proto", "").lower() name = iface.get("intf", iface.get("interface", "?")) if status in ("down", "administratively down") or proto == "down": alertas.append({ "hostname": r["hostname"], "sede": r["sede"], "interface": name, "status": status, "protocol": proto, }) return alertas

---------------------------------------------------------------------------

Exportar CSV

---------------------------------------------------------------------------

def exportar_csv(resultados: list[dict], alertas: list[dict]) -> None: out_dir = Path("output") out_dir.mkdir(exist_ok=True)

# Resumen de dispositivos
summary_file = out_dir / f"auditoria_dispositivos_{timestamp}.csv"
with summary_file.open("w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["hostname","sede","role","host","ok","error","comandos"])
    writer.writeheader()
    for r in resultados:
        writer.writerow({
            "hostname":  r["hostname"],
            "sede":      r["sede"],
            "role":      r["role"],
            "host":      r["host"],
            "ok":        r["ok"],
            "error":     r["error"] or "",
            "comandos":  len(r["data"]) if r["ok"] else 0,
        })
log.info("Resumen guardado: %s", summary_file)

# Alertas de interfaces
alertas_file = out_dir / f"alertas_interfaces_{timestamp}.csv"
with alertas_file.open("w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["hostname","sede","interface","status","protocol"])
    writer.writeheader()
    writer.writerows(alertas)
log.info("Alertas guardadas: %s (%d interfaces)", alertas_file, len(alertas))

---------------------------------------------------------------------------

Resumen en consola

---------------------------------------------------------------------------

def mostrar_resumen(resultados: list[dict], alertas: list[dict]) -> None: ok_count = sum(1 for r in resultados if r["ok"]) err_count = len(resultados) - ok_count

print("\n" + "="*60)
print("  AUDITORÍA DE RED — LogiTrans SA")
print(f"  Fecha: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print("="*60)
print(f"  Dispositivos auditados : {len(resultados)}")
print(f"  Exitosos               : {ok_count}")
print(f"  Con errores            : {err_count}")
print(f"  Interfaces con alertas : {len(alertas)}")
print("-"*60)

if alertas:
    print("  ALERTAS DE INTERFACES:")
    for a in alertas:
        print(f"    [{a['sede']}] {a['hostname']} — {a['interface']} "
              f"({a['status']} / {a['protocol']})")
else:
    print("  Sin alertas de interfaces.")

errors = [r for r in resultados if not r["ok"]]
if errors:
    print("-"*60)
    print("  DISPOSITIVOS CON ERROR:")
    for r in errors:
        print(f"    {r['hostname']} ({r['host']}) — {r['error']}")

print("="*60 + "\n")

---------------------------------------------------------------------------

Entrypoint

---------------------------------------------------------------------------

def main() -> None: log.info("=== Iniciando auditoría de red LogiTrans SA ===") creds = get_credentials()

resultados = auditar_todos(INVENTARIO, creds)
alertas    = analizar_interfaces(resultados)

mostrar_resumen(resultados, alertas)
exportar_csv(resultados, alertas)

log.info("=== Auditoría finalizada ===")

if name == "main": main()