los-vecinos-de-cisco/no/cisco_mac_monitor.py

#!/usr/bin/env python3
"""
Cisco MAC Address Monitor - Single Device Version
=================================================

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 modelos C2960, C3560, C3560E, C2960X, C3560CX y Catalyst series.

Uso:
    python3 cisco_single_mac_monitor.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
from dataclasses import dataclass


@dataclass
class MacEntry:
    """Entrada de tabla MAC"""
    vlan: str
    mac_address: str
    type: str
    port: str


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:
            self.ssh_client = paramiko.SSHClient()
            self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            
            self.logger.info(f"Conectando a {self.host}")
            
            self.ssh_client.connect(
                hostname=self.host,
                port=self.port,
                username=self.username,
                password=self.password,
                timeout=self.timeout,
                look_for_keys=False,
                allow_agent=False
            )
            
            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')
            
            if error:
                self.logger.warning(f"Error en comando '{command}': {error}")
            
            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__)
    
    def get_mac_table(self, ssh_connector: CiscoSSHConnector) -> List[MacEntry]:
        """Obtiene la tabla MAC del dispositivo"""
        try:
            # Comando para obtener tabla MAC (compatible con IOS 12.x y 15.x)
            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:
            self.logger.error(f"Error obteniendo tabla MAC: {str(e)}")
            return []
    
    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 Zabbix Low Level Discovery"""
    
    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]) -> Dict[str, Any]:
        """Formatea datos para Zabbix LLD"""
        discovery_data = []
        timestamp = datetime.now().isoformat()
        
        # Agrupar MACs por puerto
        ports_data = {}
        
        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()
                }
            
            ports_data[port]["mac_addresses"].append({
                "mac": entry.mac_address,
                "vlan": entry.vlan,
                "type": entry.type
            })
            ports_data[port]["vlans"].add(entry.vlan)
        
        # Crear entradas para LLD
        for port, data in ports_data.items():
            data["mac_count"] = len(data["mac_addresses"])
            data["vlans"] = list(data["vlans"])
            
            discovery_item = {
                "{#DEVICE_HOSTNAME}": self.device_hostname,
                "{#PORT_NAME}": port,
                "{#MAC_COUNT}": data["mac_count"],
                "{#VLANS}": ",".join(data["vlans"]),
                "{#MAC_ADDRESSES}": json.dumps(data["mac_addresses"]),
                "{#LAST_UPDATE}": timestamp
            }
            
            discovery_data.append(discovery_item)
        
        result = {
            "data": discovery_data,
            "timestamp": timestamp,
            "device_hostname": self.device_hostname,
            "total_ports": len(discovery_data),
            "total_macs": len(mac_entries)
        }
        
        self.logger.info(f"Formateados datos para dispositivo {self.device_hostname}: {len(discovery_data)} puertos, {len(mac_entries)} MACs")
        
        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',
        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
        zabbix_data = zabbix_formatter.format_for_zabbix(mac_entries)
        
        # 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()