Xavier Rodriguez
6782aa47d1
- Introduced InterfaceInfo class to encapsulate interface details. - Modified CiscoMacTableParser to gather and store interface information. - Updated ZabbixLLDFormatter to include interface status and descriptions in the output. - Adjusted test_new_format.py to validate new interface information integration. - Enhanced run.sh for improved container management.
639 lines
25 KiB
Python
Executable File
639 lines
25 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Cisco MAC Address Monitor - Compatible con Python 3.6+
|
|
======================================================
|
|
|
|
Script simplificado para monitorizar direcciones MAC en un dispositivo Cisco específico.
|
|
Recibe parámetros por línea de comandos y genera salida JSON para Zabbix LLD.
|
|
Compatible con Python 3.6+ (sin dataclasses).
|
|
|
|
Compatible con modelos C2960, C3560, C3560E, C2960X, C3560CX y Catalyst series.
|
|
|
|
Uso:
|
|
python3 cisco_mac_monitor_py36.py --host <IP> --username <USER> --password <PASS>
|
|
|
|
Autor: Sistema de Monitorización de Red
|
|
Fecha: Octubre 2024
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import argparse
|
|
import paramiko
|
|
import re
|
|
import sys
|
|
from datetime import datetime
|
|
from typing import List, Dict, Any
|
|
|
|
|
|
class MacEntry:
|
|
"""Entrada de tabla MAC - Compatible con Python 3.6"""
|
|
|
|
def __init__(self, vlan: str, mac_address: str, type: str, port: str):
|
|
self.vlan = vlan
|
|
self.mac_address = mac_address
|
|
self.type = type
|
|
self.port = port
|
|
|
|
|
|
class InterfaceInfo:
|
|
"""Información de interfaz - Compatible con Python 3.6"""
|
|
|
|
def __init__(self, name: str, status: str = "unknown", description: str = ""):
|
|
self.name = name
|
|
self.status = status # up, down, disabled, unknown
|
|
self.description = description
|
|
|
|
|
|
class CiscoSSHConnector:
|
|
"""Conector SSH para dispositivos Cisco"""
|
|
|
|
def __init__(self, host: str, username: str, password: str, port: int = 22, timeout: int = 30):
|
|
self.host = host
|
|
self.username = username
|
|
self.password = password
|
|
self.port = port
|
|
self.timeout = timeout
|
|
self.ssh_client = None
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
def connect(self) -> bool:
|
|
"""Establece conexión SSH con el dispositivo"""
|
|
try:
|
|
# Configuración especial para dispositivos Cisco antiguos
|
|
# Estos dispositivos requieren algoritmos legacy para conectividad
|
|
|
|
# Crear transport manualmente para configurar algoritmos
|
|
transport = paramiko.Transport((self.host, self.port))
|
|
transport.set_keepalive(30)
|
|
|
|
# Obtener opciones de seguridad y agregar algoritmos antiguos
|
|
security_opts = transport.get_security_options()
|
|
|
|
# Añadir diffie-hellman-group1-sha1 para dispositivos antiguos
|
|
if 'diffie-hellman-group1-sha1' not in security_opts.kex:
|
|
security_opts.kex = ['diffie-hellman-group1-sha1'] + list(security_opts.kex)
|
|
|
|
# Asegurar que ssh-rsa esté disponible
|
|
if 'ssh-rsa' not in security_opts.key_types:
|
|
security_opts.key_types = ['ssh-rsa'] + list(security_opts.key_types)
|
|
|
|
self.logger.info(f"Conectando a {self.host} con algoritmos compatibles")
|
|
|
|
# Iniciar conexión transport
|
|
transport.start_client(timeout=self.timeout)
|
|
|
|
# Autenticar
|
|
transport.auth_password(self.username, self.password)
|
|
|
|
if not transport.is_authenticated():
|
|
raise Exception("Falló la autenticación SSH")
|
|
|
|
# Crear cliente SSH y asociar transport
|
|
self.ssh_client = paramiko.SSHClient()
|
|
self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
self.ssh_client._transport = transport
|
|
|
|
self.logger.info(f"Conexión exitosa a {self.host}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error conectando a {self.host}: {str(e)}")
|
|
return False
|
|
|
|
def execute_command(self, command: str) -> str:
|
|
"""Ejecuta comando en el dispositivo"""
|
|
if not self.ssh_client:
|
|
raise Exception("No hay conexión SSH activa")
|
|
|
|
try:
|
|
self.logger.debug(f"Ejecutando comando: {command}")
|
|
stdin, stdout, stderr = self.ssh_client.exec_command(command)
|
|
|
|
output = stdout.read().decode('utf-8')
|
|
error = stderr.read().decode('utf-8')
|
|
|
|
self.logger.debug(f"Comando '{command}' - Output length: {len(output)}")
|
|
self.logger.debug(f"Comando '{command}' - First 200 chars: {output[:200]}")
|
|
|
|
if error:
|
|
self.logger.warning(f"Error en comando '{command}': {error}")
|
|
|
|
if not output.strip():
|
|
self.logger.warning(f"Comando '{command}' devolvió salida vacía")
|
|
|
|
return output
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error ejecutando comando '{command}': {str(e)}")
|
|
raise
|
|
|
|
def disconnect(self):
|
|
"""Cierra la conexión SSH"""
|
|
if self.ssh_client:
|
|
self.ssh_client.close()
|
|
self.logger.info(f"Desconectado de {self.host}")
|
|
|
|
|
|
class CiscoMacTableParser:
|
|
"""Parser para tabla MAC de dispositivos Cisco"""
|
|
|
|
def __init__(self):
|
|
self.logger = logging.getLogger(__name__)
|
|
self.interface_info = {} # Dict[str, InterfaceInfo]
|
|
|
|
def get_mac_table(self, ssh_connector: CiscoSSHConnector) -> List[MacEntry]:
|
|
"""Obtiene la tabla MAC del dispositivo"""
|
|
try:
|
|
# Primero obtenemos la información completa de interfaces (nombre, estado, descripción)
|
|
self._get_interface_info(ssh_connector)
|
|
self.logger.info(f"Detectadas {len(self.interface_info)} interfaces")
|
|
|
|
# CRUCIAL: Reconectar para dispositivos Cisco antiguos
|
|
# Algunos switches Cisco antiguos corrompen la sesión SSH después de comandos largos
|
|
self.logger.info("Reconectando para comando MAC table...")
|
|
ssh_connector.disconnect()
|
|
if not ssh_connector.connect():
|
|
self.logger.error("No se pudo reconectar para comando MAC table")
|
|
return []
|
|
|
|
# Luego obtenemos la tabla MAC
|
|
command = "show mac address-table"
|
|
output = ssh_connector.execute_command(command)
|
|
|
|
mac_entries = self._parse_mac_table_output(output)
|
|
self.logger.info(f"Obtenidas {len(mac_entries)} entradas de tabla MAC")
|
|
|
|
return mac_entries
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
self.logger.error(f"Error obteniendo tabla MAC: {str(e)}")
|
|
self.logger.error(f"Traceback: {traceback.format_exc()}")
|
|
return []
|
|
|
|
def _get_interface_info(self, ssh_connector: CiscoSSHConnector):
|
|
"""Obtiene información completa de todas las interfaces del dispositivo"""
|
|
try:
|
|
# Primero intentamos con show interface status (incluye estado)
|
|
command = "show interface status"
|
|
status_output = ssh_connector.execute_command(command)
|
|
|
|
# Si no funciona, probar comando alternativo
|
|
if "Invalid input" in status_output or not status_output.strip():
|
|
command = "show ip interface brief"
|
|
status_output = ssh_connector.execute_command(command)
|
|
|
|
# Parsear información de estado
|
|
self._parse_interface_status(status_output)
|
|
|
|
# Si no obtuvimos interfaces, usar un comando más básico
|
|
if not self.interface_info:
|
|
command = "show interface | include line protocol"
|
|
output = ssh_connector.execute_command(command)
|
|
self._parse_interface_line_protocol_to_info(output)
|
|
|
|
# Ahora obtenemos las descripciones
|
|
if self.interface_info:
|
|
self._get_interface_descriptions(ssh_connector)
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error obteniendo información de interfaces: {str(e)}")
|
|
|
|
def _parse_interface_status(self, output: str):
|
|
"""Parsea la salida de show interface status o show ip interface brief"""
|
|
lines = output.split('\n')
|
|
for line in lines:
|
|
line = line.strip()
|
|
if not line or line.startswith('-') or 'Port' in line or 'Interface' in line:
|
|
continue
|
|
|
|
# Patrón para show interface status
|
|
# Format: Port Name Status Vlan Duplex Speed Type
|
|
parts = line.split()
|
|
if len(parts) >= 3:
|
|
interface = parts[0]
|
|
|
|
# Filtrar solo interfaces físicas relevantes
|
|
if self._is_physical_interface(interface):
|
|
# El estado generalmente está en la posición después del nombre/descripción
|
|
# Formato típico: Port Name Status Vlan...
|
|
# O: Port Status Vlan... (sin nombre)
|
|
status_raw = parts[1] if len(parts) > 1 else "unknown"
|
|
|
|
# Si parts[1] parece ser un estado válido, úsalo
|
|
# De lo contrario, asume que parts[1] es el nombre y parts[2] es el estado
|
|
if status_raw.lower() in ['connected', 'notconnect', 'disabled', 'up', 'down', 'admin']:
|
|
status = self._normalize_status(status_raw)
|
|
elif len(parts) > 2:
|
|
status = self._normalize_status(parts[2])
|
|
else:
|
|
status = "unknown"
|
|
|
|
self.interface_info[interface] = InterfaceInfo(
|
|
name=interface,
|
|
status=status,
|
|
description=""
|
|
)
|
|
|
|
def _parse_interface_line_protocol_to_info(self, output: str):
|
|
"""Parsea la salida de show interface | include line protocol"""
|
|
lines = output.split('\n')
|
|
for line in lines:
|
|
# Format: FastEthernet0/1 is up, line protocol is up
|
|
if ' is ' in line and 'line protocol' in line:
|
|
parts = line.split(' is ')
|
|
interface = parts[0].strip()
|
|
if self._is_physical_interface(interface):
|
|
# Extraer el estado
|
|
status = "unknown"
|
|
if len(parts) > 1:
|
|
status_part = parts[1].split(',')[0].strip()
|
|
status = self._normalize_status(status_part)
|
|
|
|
self.interface_info[interface] = InterfaceInfo(
|
|
name=interface,
|
|
status=status,
|
|
description=""
|
|
)
|
|
|
|
def _get_interface_descriptions(self, ssh_connector: CiscoSSHConnector):
|
|
"""Obtiene las descripciones de las interfaces"""
|
|
try:
|
|
command = "show interface description"
|
|
output = ssh_connector.execute_command(command)
|
|
|
|
# Si el comando no es válido, intentar alternativa
|
|
if "Invalid input" in output or not output.strip():
|
|
# Intentar obtener descripciones una por una (más lento pero funciona en IOS antiguos)
|
|
self.logger.debug("show interface description no disponible, usando método alternativo")
|
|
return
|
|
|
|
self._parse_interface_descriptions(output)
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error obteniendo descripciones de interfaces: {str(e)}")
|
|
|
|
def _parse_interface_descriptions(self, output: str):
|
|
"""Parsea la salida de show interface description"""
|
|
lines = output.split('\n')
|
|
for line in lines:
|
|
line_stripped = line.strip()
|
|
if not line_stripped or line_stripped.startswith('-') or 'Interface' in line_stripped:
|
|
continue
|
|
|
|
# Format: Interface Status Protocol Description
|
|
# Fa0/1 up up User Port
|
|
# Usamos split con máximo 3 splits para preservar la descripción completa
|
|
parts = line.split(None, 3) # Split en whitespace, máximo 4 partes
|
|
|
|
if len(parts) >= 1:
|
|
interface = parts[0]
|
|
description = ""
|
|
|
|
if len(parts) >= 4:
|
|
description = parts[3].strip()
|
|
|
|
# Si esta interfaz ya existe en nuestro diccionario, actualizar descripción
|
|
if interface in self.interface_info:
|
|
self.interface_info[interface].description = description
|
|
|
|
def _normalize_status(self, status_raw: str) -> str:
|
|
"""Normaliza el estado de la interfaz a up/down/disabled/unknown"""
|
|
status_lower = status_raw.lower()
|
|
|
|
if status_lower in ['up', 'connected']:
|
|
return "up"
|
|
elif status_lower in ['down', 'notconnect', 'notconnected']:
|
|
return "down"
|
|
elif status_lower in ['disabled', 'admin', 'administratively']:
|
|
return "disabled"
|
|
else:
|
|
return "unknown"
|
|
|
|
def _is_physical_interface(self, interface: str) -> bool:
|
|
"""Verifica si una interfaz es física y relevante para monitorización"""
|
|
interface_lower = interface.lower()
|
|
|
|
# Interfaces físicas típicas de Cisco
|
|
physical_patterns = [
|
|
r'^(fa|fastethernet)\d+/\d+', # FastEthernet0/1
|
|
r'^(gi|gigabitethernet)\d+/\d+', # GigabitEthernet0/1
|
|
r'^(te|tengigabitethernet)\d+/\d+', # TenGigabitEthernet0/1
|
|
r'^(eth|ethernet)\d+/\d+', # Ethernet0/1
|
|
r'^(fe|fe\d+)', # Fe0/1
|
|
]
|
|
|
|
# Excluir interfaces no físicas
|
|
exclude_patterns = [
|
|
r'vlan\d+',
|
|
r'loopback\d+',
|
|
r'tunnel\d+',
|
|
r'null\d+',
|
|
r'mgmt\d+',
|
|
r'management\d+'
|
|
]
|
|
|
|
# Verificar exclusiones
|
|
for exclude_pattern in exclude_patterns:
|
|
if re.match(exclude_pattern, interface_lower):
|
|
return False
|
|
|
|
# Verificar inclusiones
|
|
for physical_pattern in physical_patterns:
|
|
if re.match(physical_pattern, interface_lower):
|
|
return True
|
|
|
|
return False
|
|
|
|
def _parse_mac_table_output(self, output: str) -> List[MacEntry]:
|
|
"""Parsea la salida del comando show mac address-table"""
|
|
mac_entries = []
|
|
|
|
# Patrones para diferentes formatos de salida de IOS
|
|
patterns = [
|
|
# IOS 12.x y 15.x estándar
|
|
r'^\s*(\d+)\s+([0-9a-fA-F]{4}\.[0-9a-fA-F]{4}\.[0-9a-fA-F]{4})\s+(\w+)\s+(.+)$',
|
|
# Formato alternativo con guiones
|
|
r'^\s*(\d+)\s+([0-9a-fA-F]{2}-[0-9a-fA-F]{2}-[0-9a-fA-F]{2}-[0-9a-fA-F]{2}-[0-9a-fA-F]{2}-[0-9a-fA-F]{2})\s+(\w+)\s+(.+)$',
|
|
# Formato con dos puntos
|
|
r'^\s*(\d+)\s+([0-9a-fA-F]{2}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2})\s+(\w+)\s+(.+)$'
|
|
]
|
|
|
|
lines = output.split('\n')
|
|
|
|
for line in lines:
|
|
line = line.strip()
|
|
if not line or line.startswith('-') or 'VLAN' in line.upper() or 'MAC Address' in line:
|
|
continue
|
|
|
|
for pattern in patterns:
|
|
match = re.match(pattern, line)
|
|
if match:
|
|
vlan = match.group(1)
|
|
mac_address = self._normalize_mac_address(match.group(2))
|
|
mac_type = match.group(3)
|
|
port = match.group(4).strip()
|
|
|
|
# Filtrar MACs del CPU y otras no relevantes
|
|
if not self._is_valid_port(port):
|
|
continue
|
|
|
|
mac_entries.append(MacEntry(
|
|
vlan=vlan,
|
|
mac_address=mac_address,
|
|
type=mac_type,
|
|
port=port
|
|
))
|
|
break
|
|
|
|
return mac_entries
|
|
|
|
def _normalize_mac_address(self, mac: str) -> str:
|
|
"""Normaliza formato de dirección MAC"""
|
|
# Remover todos los separadores y convertir a minúsculas
|
|
clean_mac = re.sub(r'[.:-]', '', mac.lower())
|
|
|
|
# Formatear como XX:XX:XX:XX:XX:XX
|
|
if len(clean_mac) == 12:
|
|
return ':'.join([clean_mac[i:i+2] for i in range(0, 12, 2)])
|
|
|
|
return mac
|
|
|
|
def _is_valid_port(self, port: str) -> bool:
|
|
"""Verifica si el puerto es válido para monitorización"""
|
|
# Excluir puertos del CPU, uplinks management, etc.
|
|
exclude_patterns = [
|
|
r'cpu',
|
|
r'switch',
|
|
r'router',
|
|
r'null',
|
|
r'drop',
|
|
r'punt'
|
|
]
|
|
|
|
port_lower = port.lower()
|
|
for pattern in exclude_patterns:
|
|
if re.search(pattern, port_lower):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
class ZabbixLLDFormatter:
|
|
"""Formateador de datos para salida JSON estructurada por puertos"""
|
|
|
|
def __init__(self, device_hostname: str):
|
|
self.device_hostname = device_hostname
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
def format_for_zabbix(self, mac_entries: List[MacEntry], interface_info: Dict[str, Any] = None) -> Dict[str, Any]:
|
|
"""Formatea datos en formato JSON estructurado por puertos"""
|
|
discovery_data = []
|
|
# Usar timestamp Unix para compatibilidad con Zabbix
|
|
unix_timestamp = int(datetime.now().timestamp())
|
|
iso_timestamp = datetime.now().isoformat() # Para logs y metadatos
|
|
|
|
# Agrupar MACs por puerto
|
|
ports_data = {}
|
|
|
|
# Procesar MACs existentes
|
|
for entry in mac_entries:
|
|
port = entry.port
|
|
if port not in ports_data:
|
|
ports_data[port] = {
|
|
"port_name": port,
|
|
"device_hostname": self.device_hostname,
|
|
"mac_addresses": [],
|
|
"mac_count": 0,
|
|
"vlans": set(),
|
|
"status": "unknown",
|
|
"description": ""
|
|
}
|
|
|
|
ports_data[port]["mac_addresses"].append({
|
|
"mac": entry.mac_address,
|
|
"vlan": entry.vlan,
|
|
"type": entry.type
|
|
})
|
|
ports_data[port]["vlans"].add(entry.vlan)
|
|
|
|
# Añadir todas las interfaces conocidas (con o sin MACs)
|
|
if interface_info:
|
|
for interface_name, iface_info in interface_info.items():
|
|
if interface_name not in ports_data:
|
|
ports_data[interface_name] = {
|
|
"port_name": interface_name,
|
|
"device_hostname": self.device_hostname,
|
|
"mac_addresses": [],
|
|
"mac_count": 0,
|
|
"vlans": set(),
|
|
"status": iface_info.status,
|
|
"description": iface_info.description
|
|
}
|
|
else:
|
|
# Actualizar estado y descripción para puertos que ya tienen MACs
|
|
ports_data[interface_name]["status"] = iface_info.status
|
|
ports_data[interface_name]["description"] = iface_info.description
|
|
|
|
# Crear entradas en el nuevo formato
|
|
for port, data in ports_data.items():
|
|
data["mac_count"] = len(data["mac_addresses"])
|
|
data["vlans"] = list(data["vlans"]) if data["vlans"] else []
|
|
|
|
# Formatear MAC addresses en el nuevo formato: "MAC VLAN\nMAC VLAN\n..."
|
|
# MACs en mayúsculas y ordenadas por dirección MAC y luego por VLAN
|
|
mac_list = []
|
|
for mac_data in data["mac_addresses"]:
|
|
mac_list.append(f"{mac_data['mac'].upper()} {mac_data['vlan']}")
|
|
|
|
# Ordenar primero por MAC address y luego por VLAN
|
|
mac_list.sort(key=lambda x: (x.split()[0], int(x.split()[1])))
|
|
mac_addresses_text = "\n".join(mac_list)
|
|
|
|
discovery_item = {
|
|
"port": port,
|
|
"data": {
|
|
"count": data["mac_count"],
|
|
"vlans": ",".join(data["vlans"]) if data["vlans"] else "",
|
|
"addresses": mac_addresses_text,
|
|
"status": data.get("status", "unknown"),
|
|
"description": data.get("description", ""),
|
|
"timestamp": unix_timestamp
|
|
}
|
|
}
|
|
|
|
discovery_data.append(discovery_item)
|
|
|
|
# Ordenar por nombre de puerto para consistencia
|
|
discovery_data.sort(key=lambda x: x["port"])
|
|
|
|
result = {
|
|
"data": discovery_data,
|
|
"timestamp": iso_timestamp,
|
|
"unix_timestamp": unix_timestamp,
|
|
"device_hostname": self.device_hostname,
|
|
"total_ports": len(discovery_data),
|
|
"total_macs": len(mac_entries),
|
|
"ports_with_macs": len([p for p in ports_data.values() if p["mac_count"] > 0]),
|
|
"ports_without_macs": len([p for p in ports_data.values() if p["mac_count"] == 0])
|
|
}
|
|
|
|
self.logger.info(f"Formateados datos para dispositivo {self.device_hostname}: {len(discovery_data)} puertos ({result['ports_with_macs']} con MACs, {result['ports_without_macs']} sin MACs), {len(mac_entries)} MACs totales")
|
|
|
|
return result
|
|
|
|
|
|
def setup_logging(debug: bool = False) -> logging.Logger:
|
|
"""Configura el sistema de logging"""
|
|
level = logging.DEBUG if debug else logging.INFO
|
|
|
|
logging.basicConfig(
|
|
level=level,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.StreamHandler(sys.stderr)
|
|
]
|
|
)
|
|
|
|
return logging.getLogger(__name__)
|
|
|
|
|
|
def main():
|
|
"""Función principal"""
|
|
parser = argparse.ArgumentParser(
|
|
description='Monitor de direcciones MAC para dispositivo Cisco individual (Python 3.6+)',
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Ejemplos de uso:
|
|
%(prog)s --host 192.168.1.10 --username admin --password cisco123
|
|
%(prog)s -H 192.168.1.10 -u admin -p cisco123 --port 2222
|
|
%(prog)s -H 192.168.1.10 -u admin -p cisco123 --debug
|
|
%(prog)s -H 192.168.1.10 -u admin -p cisco123 --hostname SW-CORE-001
|
|
|
|
Salida:
|
|
El script genera JSON en formato Zabbix LLD por stdout.
|
|
Los logs van por stderr.
|
|
"""
|
|
)
|
|
|
|
# Argumentos requeridos
|
|
parser.add_argument('--host', '-H', required=True,
|
|
help='Dirección IP o hostname del dispositivo Cisco')
|
|
parser.add_argument('--username', '-u', required=True,
|
|
help='Usuario para conexión SSH')
|
|
parser.add_argument('--password', '-p', required=True,
|
|
help='Contraseña para conexión SSH')
|
|
|
|
# Argumentos opcionales
|
|
parser.add_argument('--port', '-P', type=int, default=22,
|
|
help='Puerto SSH (default: 22)')
|
|
parser.add_argument('--timeout', '-t', type=int, default=30,
|
|
help='Timeout de conexión en segundos (default: 30)')
|
|
parser.add_argument('--hostname', '-n',
|
|
help='Nombre del dispositivo para identificación (default: usa --host)')
|
|
parser.add_argument('--debug', '-d', action='store_true',
|
|
help='Habilitar modo debug')
|
|
parser.add_argument('--quiet', '-q', action='store_true',
|
|
help='Modo silencioso (solo JSON por stdout)')
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Configurar logging
|
|
if not args.quiet:
|
|
logger = setup_logging(args.debug)
|
|
else:
|
|
# En modo silencioso, solo errores críticos
|
|
logging.basicConfig(level=logging.CRITICAL)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Determinar hostname del dispositivo
|
|
device_hostname = args.hostname or args.host
|
|
|
|
# Inicializar componentes
|
|
ssh_connector = CiscoSSHConnector(
|
|
host=args.host,
|
|
username=args.username,
|
|
password=args.password,
|
|
port=args.port,
|
|
timeout=args.timeout
|
|
)
|
|
|
|
mac_parser = CiscoMacTableParser()
|
|
zabbix_formatter = ZabbixLLDFormatter(device_hostname)
|
|
|
|
try:
|
|
# Conectar al dispositivo
|
|
if not ssh_connector.connect():
|
|
logger.error("No se pudo establecer conexión SSH")
|
|
sys.exit(1)
|
|
|
|
# Obtener tabla MAC
|
|
mac_entries = mac_parser.get_mac_table(ssh_connector)
|
|
|
|
if not mac_entries:
|
|
logger.warning("No se obtuvieron entradas de tabla MAC")
|
|
|
|
# Formatear para Zabbix incluyendo información de interfaces
|
|
zabbix_data = zabbix_formatter.format_for_zabbix(
|
|
mac_entries,
|
|
mac_parser.interface_info
|
|
)
|
|
|
|
# Generar salida JSON
|
|
json_output = json.dumps(zabbix_data, indent=2, ensure_ascii=False)
|
|
print(json_output)
|
|
|
|
if not args.quiet:
|
|
logger.info("Proceso completado exitosamente")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error durante la ejecución: {str(e)}")
|
|
sys.exit(1)
|
|
|
|
finally:
|
|
ssh_connector.disconnect()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main() |