Refactor Cisco MAC Monitor: Implement HTTPS service and restructure code
- Removed old requirements.txt and test_new_format.py files. - Added new src/cisco_mac_http_service.py to implement an HTTPS service for MAC monitoring. - Introduced src/cisco_mac_monitor.py for the main MAC monitoring functionality. - Created new requirements.txt for dependencies. - Added example usage documentation in Docs/example. - Updated test_new_format.py to validate new JSON output format.
This commit is contained in:
Xavier Rodriguez
Executable
+440
@@ -0,0 +1,440 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Cisco MAC Monitor HTTPS Service
|
||||
===============================
|
||||
|
||||
Servicio HTTPS que expone la funcionalidad de monitorización de MACs vía REST API.
|
||||
Mantiene todos los logs informativos pero devuelve solo JSON limpio en las respuestas.
|
||||
Usa POST para credenciales seguras y soporte SSL/TLS.
|
||||
|
||||
Endpoints:
|
||||
POST /mac-monitor - Body JSON con credenciales
|
||||
GET /health - Health check
|
||||
GET /metrics - Métricas del servicio
|
||||
|
||||
Autor: Sistema de Monitorización de Red
|
||||
Fecha: Noviembre 2024
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
import ssl
|
||||
import os
|
||||
from datetime import datetime
|
||||
from typing import Dict, Any
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||
import threading
|
||||
import signal
|
||||
import time
|
||||
|
||||
# Importar las clases existentes del cisco_mac_monitor
|
||||
from cisco_mac_monitor import CiscoSSHConnector, CiscoMacTableParser, ZabbixLLDFormatter
|
||||
|
||||
|
||||
class CiscoMacHTTPHandler(BaseHTTPRequestHandler):
|
||||
"""Handler HTTP para el servicio de monitorización MAC"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
self.logger = logging.getLogger(__name__)
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def do_GET(self):
|
||||
"""Maneja peticiones GET - Solo endpoints públicos"""
|
||||
try:
|
||||
if self.path == '/health' or self.path == '/':
|
||||
self._handle_health()
|
||||
elif self.path == '/metrics':
|
||||
self._handle_metrics()
|
||||
elif self.path.startswith('/mac-monitor'):
|
||||
self._send_error(405, "Método no permitido. Use POST con JSON body para /mac-monitor")
|
||||
else:
|
||||
self._send_error(404, "Endpoint no encontrado")
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error procesando petición GET: {str(e)}")
|
||||
self._send_error(500, f"Error interno: {str(e)}")
|
||||
|
||||
def do_POST(self):
|
||||
"""Maneja peticiones POST - Endpoint seguro de monitorización"""
|
||||
try:
|
||||
if self.path == '/mac-monitor':
|
||||
self._handle_mac_monitor_post()
|
||||
else:
|
||||
self._send_error(404, "Endpoint no encontrado")
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error procesando petición POST: {str(e)}")
|
||||
self._send_error(500, f"Error interno: {str(e)}")
|
||||
|
||||
def _handle_mac_monitor_post(self):
|
||||
"""Maneja el endpoint POST de monitorización MAC con JSON body"""
|
||||
try:
|
||||
# Leer Content-Length
|
||||
content_length = int(self.headers.get('Content-Length', 0))
|
||||
if content_length == 0:
|
||||
self._send_error(400, "Body JSON requerido")
|
||||
return
|
||||
|
||||
# Leer body JSON
|
||||
post_data = self.rfile.read(content_length)
|
||||
|
||||
try:
|
||||
json_data = json.loads(post_data.decode('utf-8'))
|
||||
except json.JSONDecodeError as e:
|
||||
self._send_error(400, f"JSON inválido: {str(e)}")
|
||||
return
|
||||
|
||||
# Validar parámetros requeridos
|
||||
required_fields = ['host', 'username', 'password']
|
||||
missing_fields = []
|
||||
|
||||
for field in required_fields:
|
||||
if field not in json_data or not json_data[field]:
|
||||
missing_fields.append(field)
|
||||
|
||||
if missing_fields:
|
||||
self._send_error(400, f"Campos faltantes en JSON: {', '.join(missing_fields)}")
|
||||
return
|
||||
|
||||
# Extraer parámetros
|
||||
host = json_data['host']
|
||||
username = json_data['username']
|
||||
password = json_data['password']
|
||||
hostname = json_data.get('hostname', host)
|
||||
port = int(json_data.get('port', 22))
|
||||
timeout = int(json_data.get('timeout', 30))
|
||||
|
||||
self.logger.info(f"Iniciando monitorización para {host}")
|
||||
|
||||
# Ejecutar monitorización
|
||||
result = self._execute_mac_monitoring(host, username, password, hostname, port, timeout)
|
||||
|
||||
if result:
|
||||
self._send_json_response(result)
|
||||
self.logger.info(f"Monitorización completada para {host}")
|
||||
else:
|
||||
self._send_error(500, "Error obteniendo datos del dispositivo")
|
||||
|
||||
except ValueError as e:
|
||||
self._send_error(400, f"Parámetro inválido: {str(e)}")
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error en monitorización POST: {str(e)}")
|
||||
self._send_error(500, f"Error de monitorización: {str(e)}")
|
||||
|
||||
def _execute_mac_monitoring(self, host: str, username: str, password: str,
|
||||
hostname: str, port: int, timeout: int) -> Dict[str, Any]:
|
||||
"""Ejecuta el proceso de monitorización MAC"""
|
||||
# Inicializar componentes
|
||||
ssh_connector = CiscoSSHConnector(
|
||||
host=host,
|
||||
username=username,
|
||||
password=password,
|
||||
port=port,
|
||||
timeout=timeout
|
||||
)
|
||||
|
||||
mac_parser = CiscoMacTableParser()
|
||||
zabbix_formatter = ZabbixLLDFormatter(hostname)
|
||||
|
||||
try:
|
||||
# Conectar al dispositivo
|
||||
if not ssh_connector.connect():
|
||||
self.logger.error(f"No se pudo establecer conexión SSH con {host}")
|
||||
return None
|
||||
|
||||
# Obtener tabla MAC
|
||||
mac_entries = mac_parser.get_mac_table(ssh_connector)
|
||||
|
||||
if not mac_entries:
|
||||
self.logger.warning(f"No se obtuvieron entradas de tabla MAC de {host}")
|
||||
|
||||
# Formatear para Zabbix incluyendo información de interfaces
|
||||
zabbix_data = zabbix_formatter.format_for_zabbix(
|
||||
mac_entries,
|
||||
mac_parser.interface_info
|
||||
)
|
||||
|
||||
return zabbix_data
|
||||
|
||||
finally:
|
||||
ssh_connector.disconnect()
|
||||
|
||||
def _handle_health(self):
|
||||
"""Endpoint de health check"""
|
||||
health_data = {
|
||||
"status": "healthy",
|
||||
"service": "cisco-mac-monitor-https",
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"version": "2.1-https",
|
||||
"ssl_enabled": hasattr(self.server, 'socket') and isinstance(self.server.socket, ssl.SSLSocket)
|
||||
}
|
||||
self._send_json_response(health_data)
|
||||
|
||||
def _handle_metrics(self):
|
||||
"""Endpoint básico de métricas"""
|
||||
metrics_data = {
|
||||
"service": "cisco-mac-monitor-https",
|
||||
"uptime_seconds": time.time() - start_time,
|
||||
"requests_total": getattr(self.server, 'request_count', 0),
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"ssl_enabled": hasattr(self.server, 'socket') and isinstance(self.server.socket, ssl.SSLSocket)
|
||||
}
|
||||
self._send_json_response(metrics_data)
|
||||
|
||||
def _send_json_response(self, data: Dict[str, Any]):
|
||||
"""Envía respuesta JSON"""
|
||||
json_data = json.dumps(data, indent=2, ensure_ascii=False)
|
||||
|
||||
self.send_response(200)
|
||||
self.send_header('Content-Type', 'application/json; charset=utf-8')
|
||||
self.send_header('Content-Length', len(json_data.encode('utf-8')))
|
||||
self.send_header('Access-Control-Allow-Origin', '*')
|
||||
self.end_headers()
|
||||
|
||||
self.wfile.write(json_data.encode('utf-8'))
|
||||
|
||||
def _send_error(self, code: int, message: str):
|
||||
"""Envía respuesta de error"""
|
||||
error_data = {
|
||||
"error": message,
|
||||
"code": code,
|
||||
"timestamp": datetime.now().isoformat()
|
||||
}
|
||||
|
||||
json_data = json.dumps(error_data, indent=2, ensure_ascii=False)
|
||||
|
||||
self.send_response(code)
|
||||
self.send_header('Content-Type', 'application/json; charset=utf-8')
|
||||
self.send_header('Content-Length', len(json_data.encode('utf-8')))
|
||||
self.end_headers()
|
||||
|
||||
self.wfile.write(json_data.encode('utf-8'))
|
||||
|
||||
self.logger.warning(f"Error {code}: {message}")
|
||||
|
||||
def log_message(self, format, *args):
|
||||
"""Sobrescribir para evitar logs HTTP automáticos"""
|
||||
# Los logs HTTP automáticos se manejan mediante nuestro logger
|
||||
pass
|
||||
|
||||
|
||||
class CiscoMacHTTPSServer:
|
||||
"""Servidor HTTPS para el servicio de monitorización MAC"""
|
||||
|
||||
def __init__(self, host='0.0.0.0', port=8443, ssl_cert=None, ssl_key=None):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.ssl_cert = ssl_cert
|
||||
self.ssl_key = ssl_key
|
||||
self.httpd = None
|
||||
self.server_thread = None
|
||||
self.logger = logging.getLogger(__name__)
|
||||
self.running = False
|
||||
|
||||
def _generate_self_signed_cert(self):
|
||||
"""Genera certificado autofirmado si no se proporciona"""
|
||||
cert_file = '/tmp/cisco_mac_server.crt'
|
||||
key_file = '/tmp/cisco_mac_server.key'
|
||||
|
||||
if os.path.exists(cert_file) and os.path.exists(key_file):
|
||||
self.logger.info("Usando certificado autofirmado existente")
|
||||
return cert_file, key_file
|
||||
|
||||
self.logger.info("Generando certificado autofirmado...")
|
||||
|
||||
try:
|
||||
import subprocess
|
||||
|
||||
# Generar key privada
|
||||
result = subprocess.run([
|
||||
'openssl', 'genrsa', '-out', key_file, '2048'
|
||||
], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
|
||||
if result.returncode != 0:
|
||||
self.logger.error(f"Error generando key: {result.stderr.decode()}")
|
||||
return None, None
|
||||
|
||||
# Generar certificado autofirmado
|
||||
result = subprocess.run([
|
||||
'openssl', 'req', '-new', '-x509', '-key', key_file,
|
||||
'-out', cert_file, '-days', '365', '-subj',
|
||||
'/C=ES/ST=Madrid/L=Madrid/O=CiscoMonitor/CN=cisco-mac-monitor'
|
||||
], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
|
||||
if result.returncode != 0:
|
||||
self.logger.error(f"Error generando certificado: {result.stderr.decode()}")
|
||||
return None, None
|
||||
|
||||
# Cambiar permisos
|
||||
os.chmod(key_file, 0o600)
|
||||
os.chmod(cert_file, 0o644)
|
||||
|
||||
self.logger.info(f"Certificado autofirmado generado: {cert_file}")
|
||||
return cert_file, key_file
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
self.logger.error(f"Error generando certificado: {e}")
|
||||
return None, None
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error en subprocess: {e}")
|
||||
return None, None
|
||||
except FileNotFoundError:
|
||||
self.logger.warning("OpenSSL no disponible, funcionando en modo HTTP")
|
||||
return None, None
|
||||
|
||||
def start(self):
|
||||
"""Inicia el servidor HTTPS"""
|
||||
try:
|
||||
self.httpd = HTTPServer((self.host, self.port), CiscoMacHTTPHandler)
|
||||
self.httpd.request_count = 0
|
||||
|
||||
# Configurar SSL si está disponible
|
||||
ssl_enabled = False
|
||||
if self.ssl_cert and self.ssl_key:
|
||||
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
|
||||
ssl_context.load_cert_chain(self.ssl_cert, self.ssl_key)
|
||||
self.httpd.socket = ssl_context.wrap_socket(self.httpd.socket, server_side=True)
|
||||
ssl_enabled = True
|
||||
self.logger.info(f"SSL habilitado con certificados personalizados")
|
||||
else:
|
||||
# Intentar generar certificado autofirmado
|
||||
cert_file, key_file = self._generate_self_signed_cert()
|
||||
if cert_file and key_file:
|
||||
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
|
||||
ssl_context.load_cert_chain(cert_file, key_file)
|
||||
self.httpd.socket = ssl_context.wrap_socket(self.httpd.socket, server_side=True)
|
||||
ssl_enabled = True
|
||||
self.logger.info(f"SSL habilitado con certificado autofirmado")
|
||||
|
||||
protocol = "https" if ssl_enabled else "http"
|
||||
self.logger.info(f"Iniciando servidor {protocol.upper()} en {self.host}:{self.port}")
|
||||
|
||||
# Configurar shutdown graceful
|
||||
signal.signal(signal.SIGTERM, self._signal_handler)
|
||||
signal.signal(signal.SIGINT, self._signal_handler)
|
||||
|
||||
self.running = True
|
||||
self.logger.info(f"Servidor disponible en {protocol}://{self.host}:{self.port}")
|
||||
self.logger.info("Endpoints disponibles:")
|
||||
self.logger.info(f" POST /mac-monitor - Body JSON con credenciales")
|
||||
self.logger.info(f" GET /health")
|
||||
self.logger.info(f" GET /metrics")
|
||||
|
||||
if ssl_enabled:
|
||||
self.logger.info("⚠️ Usando certificado autofirmado - ignorar warnings SSL")
|
||||
|
||||
# Iniciar servidor
|
||||
self.httpd.serve_forever()
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error iniciando servidor: {str(e)}")
|
||||
raise
|
||||
|
||||
def stop(self):
|
||||
"""Detiene el servidor HTTP"""
|
||||
if self.httpd and self.running:
|
||||
self.logger.info("Deteniendo servidor HTTP...")
|
||||
self.running = False
|
||||
self.httpd.shutdown()
|
||||
self.httpd.server_close()
|
||||
self.logger.info("Servidor HTTP detenido")
|
||||
|
||||
def _signal_handler(self, signum, frame):
|
||||
"""Maneja señales de sistema para shutdown graceful"""
|
||||
self.logger.info(f"Recibida señal {signum}, deteniendo servidor...")
|
||||
self.stop()
|
||||
|
||||
|
||||
def setup_logging(debug: bool = False) -> logging.Logger:
|
||||
"""Configura el sistema de logging"""
|
||||
level = logging.DEBUG if debug else logging.INFO
|
||||
|
||||
# Formato de logging
|
||||
formatter = logging.Formatter(
|
||||
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
|
||||
# Handler para stderr
|
||||
handler = logging.StreamHandler(sys.stderr)
|
||||
handler.setFormatter(formatter)
|
||||
|
||||
# Configurar logger principal
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(level)
|
||||
logger.addHandler(handler)
|
||||
|
||||
# Reducir verbosidad de paramiko si no está en debug
|
||||
if not debug:
|
||||
logging.getLogger('paramiko').setLevel(logging.WARNING)
|
||||
|
||||
return logging.getLogger(__name__)
|
||||
|
||||
|
||||
def main():
|
||||
"""Función principal"""
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description='Cisco MAC Monitor HTTPS Service',
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
Ejemplos de uso:
|
||||
%(prog)s --host 0.0.0.0 --port 8443
|
||||
%(prog)s --host 127.0.0.1 --port 8443 --debug
|
||||
%(prog)s --ssl-cert /path/to/cert.pem --ssl-key /path/to/key.pem
|
||||
|
||||
Endpoints:
|
||||
POST /mac-monitor - Body JSON: {"host": "192.168.1.10", "username": "admin", "password": "cisco123"}
|
||||
GET /health
|
||||
GET /metrics
|
||||
|
||||
Ejemplo de petición:
|
||||
curl -k -X POST "https://localhost:8443/mac-monitor" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"host":"192.168.1.10","username":"admin","password":"cisco123"}'
|
||||
"""
|
||||
)
|
||||
|
||||
parser.add_argument('--host', default='0.0.0.0',
|
||||
help='IP de escucha del servidor (default: 0.0.0.0)')
|
||||
parser.add_argument('--port', '-p', type=int, default=8443,
|
||||
help='Puerto de escucha del servidor (default: 8443)')
|
||||
parser.add_argument('--ssl-cert',
|
||||
help='Archivo de certificado SSL (opcional)')
|
||||
parser.add_argument('--ssl-key',
|
||||
help='Archivo de clave privada SSL (opcional)')
|
||||
parser.add_argument('--debug', '-d', action='store_true',
|
||||
help='Habilitar modo debug')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Configurar logging
|
||||
logger = setup_logging(args.debug)
|
||||
|
||||
# Variable global para métricas
|
||||
global start_time
|
||||
start_time = time.time()
|
||||
|
||||
# Crear y iniciar servidor
|
||||
server = CiscoMacHTTPSServer(
|
||||
host=args.host,
|
||||
port=args.port,
|
||||
ssl_cert=args.ssl_cert,
|
||||
ssl_key=args.ssl_key
|
||||
)
|
||||
|
||||
try:
|
||||
server.start()
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Interrupción por teclado recibida")
|
||||
except Exception as e:
|
||||
logger.error(f"Error fatal: {str(e)}")
|
||||
sys.exit(1)
|
||||
finally:
|
||||
server.stop()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Executable
+639
@@ -0,0 +1,639 @@
|
||||
#!/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()
|
||||
@@ -0,0 +1,11 @@
|
||||
# Dependencias Python para Cisco MAC Monitor
|
||||
# ==========================================
|
||||
|
||||
# SSH y conexiones de red
|
||||
paramiko>=2.12.0,<4.0
|
||||
|
||||
# Criptografía para SSH legacy
|
||||
cryptography>=3.4.8,<42.0
|
||||
|
||||
# Opcional: Para testing y debugging
|
||||
PyYAML>=5.0,<7.0
|
||||
Executable
+52
@@ -0,0 +1,52 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script para verificar el nuevo formato JSON
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from cisco_mac_monitor import MacEntry, InterfaceInfo, ZabbixLLDFormatter
|
||||
import json
|
||||
|
||||
# Crear datos de prueba
|
||||
test_mac_entries = [
|
||||
MacEntry(vlan="600", mac_address="f0:8e:db:2e:4d:86", type="DYNAMIC", port="Gi1/0/1"),
|
||||
MacEntry(vlan="601", mac_address="f0:8e:db:2e:4d:86", type="DYNAMIC", port="Gi1/0/1"),
|
||||
MacEntry(vlan="151", mac_address="00:50:56:a6:91:69", type="DYNAMIC", port="Gi1/0/10"),
|
||||
]
|
||||
|
||||
# Crear información de interfaces de prueba (incluye estado y descripción)
|
||||
test_interface_info = {
|
||||
"Gi1/0/1": InterfaceInfo(name="Gi1/0/1", status="up", description="User Port - Office"),
|
||||
"Gi1/0/10": InterfaceInfo(name="Gi1/0/10", status="up", description="Server Link"),
|
||||
"Gi1/0/11": InterfaceInfo(name="Gi1/0/11", status="down", description="Unused Port"),
|
||||
}
|
||||
|
||||
# Crear formatter y formatear datos
|
||||
formatter = ZabbixLLDFormatter("10.150.250.2")
|
||||
result = formatter.format_for_zabbix(test_mac_entries, test_interface_info)
|
||||
|
||||
# Mostrar resultado
|
||||
print(json.dumps(result, indent=2, ensure_ascii=False))
|
||||
|
||||
# Verificar estructura esperada
|
||||
print("\n" + "="*50)
|
||||
print("VERIFICACIÓN DE ESTRUCTURA:")
|
||||
print("="*50)
|
||||
|
||||
data = result["data"]
|
||||
print(f"✓ Total de puertos: {len(data)}")
|
||||
|
||||
for item in data:
|
||||
port = item["port"]
|
||||
port_data = item["data"]
|
||||
print(f"✓ Puerto {port}:")
|
||||
print(f" - Count: {port_data['count']}")
|
||||
print(f" - VLANs: '{port_data['vlans']}'")
|
||||
print(f" - Addresses: '{port_data['addresses']}'")
|
||||
print(f" - Status: '{port_data['status']}'")
|
||||
print(f" - Description: '{port_data['description']}'")
|
||||
print(f" - Timestamp: {port_data['timestamp']}")
|
||||
print()
|
||||
Reference in New Issue
Block a user