Merge pull request 'feature/20260309' (#1) from feature/20260309 into main

Reviewed-on: #1

.gitignore

@@ -0,0 +1 @@
+__pycache__

CHANGELOG.md

@@ -0,0 +1,14 @@
+CHANGELOG.md
+
+2026/03/11
+Fix SSH reconnection and improve interface description parsing in CiscoMacTableParser
+
+2026/02/09
+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.

Dockerfile

@@ -36,7 +36,7 @@ RUN dnf update -y && \
 RUN useradd -m -u 1000 -s /bin/bash monitor
 
 # Copiar requirements.txt primero para cache de Docker
-COPY requirements.txt /tmp/requirements.txt
+COPY src/requirements.txt /tmp/requirements.txt
 
 # Instalar dependencias Python desde requirements.txt
 RUN pip3 install --no-cache-dir --upgrade pip && \
@@ -48,8 +48,8 @@ USER monitor
 WORKDIR /app
 
 # Copiar scripts principales
-COPY --chmod=755 --chown=monitor:monitor cisco_mac_monitor.py /app/
+COPY --chmod=755 --chown=monitor:monitor src/cisco_mac_monitor.py /app/
-COPY --chmod=755 --chown=monitor:monitor cisco_mac_http_service.py /app/
+COPY --chmod=755 --chown=monitor:monitor src/cisco_mac_http_service.py /app/
 
 # Exponer puerto HTTPS
 EXPOSE 8443

build.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
 
-docker build --platform linux/amd64 -t localhost/cisco-friends:$(date +%Y%m%d%H) .
+docker build --platform linux/amd64 --no-cache -t localhost/cisco-friends:$(date +%Y%m%d%H) .

run.sh

@@ -1,5 +1,9 @@
 #!/bin/bash
 
+docker stop cisco-friends
+
+docker rm cisco-friends
+
 docker run -d --platform linux/amd64 \
   --restart unless-stopped \
   -p 127.0.0.1:8443:8443 \

src/cisco_mac_http_service.py

@@ -150,11 +150,11 @@ class CiscoMacHTTPHandler(BaseHTTPRequestHandler):
             if not mac_entries:
                 self.logger.warning(f"No se obtuvieron entradas de tabla MAC de {host}")
             
-            # Obtener la lista de interfaces detectadas
+            # Formatear para Zabbix incluyendo información de interfaces
-            all_interfaces = getattr(mac_parser, 'detected_interfaces', [])
+            zabbix_data = zabbix_formatter.format_for_zabbix(
-            
+                mac_entries, 
-            # Formatear para Zabbix incluyendo todas las interfaces
+                mac_parser.interface_info
-            zabbix_data = zabbix_formatter.format_for_zabbix(mac_entries, all_interfaces)
+            )
             
             return zabbix_data
             

src/cisco_mac_monitor.py

@@ -36,6 +36,15 @@ class MacEntry:
         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"""
     
@@ -131,13 +140,14 @@ class CiscoMacTableParser:
     
     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 lista de interfaces
+            # Primero obtenemos la información completa de interfaces (nombre, estado, descripción)
-            interfaces = self._get_interface_list(ssh_connector)
+            self._get_interface_info(ssh_connector)
-            self.logger.info(f"Detectadas {len(interfaces)} interfaces")
+            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
@@ -154,9 +164,6 @@ class CiscoMacTableParser:
             mac_entries = self._parse_mac_table_output(output)
             self.logger.info(f"Obtenidas {len(mac_entries)} entradas de tabla MAC")
             
-            # Almacenar las interfaces para usarlas en el formatter
-            self.detected_interfaces = interfaces
-            
             return mac_entries
             
         except Exception as e:
@@ -165,36 +172,36 @@ class CiscoMacTableParser:
             self.logger.error(f"Traceback: {traceback.format_exc()}")
             return []
     
-    def _get_interface_list(self, ssh_connector: CiscoSSHConnector) -> List[str]:
+    def _get_interface_info(self, ssh_connector: CiscoSSHConnector):
-        """Obtiene la lista de todas las interfaces del dispositivo"""
+        """Obtiene información completa de todas las interfaces del dispositivo"""
         try:
-            # Comando para obtener interfaces (funciona en IOS 12.x y 15.x)
+            # Primero intentamos con show interface status (incluye estado)
             command = "show interface status"
-            output = ssh_connector.execute_command(command)
+            status_output = ssh_connector.execute_command(command)
             
             # Si no funciona, probar comando alternativo
-            if "Invalid input" in output or not output.strip():
+            if "Invalid input" in status_output or not status_output.strip():
                 command = "show ip interface brief"
-                output = ssh_connector.execute_command(command)
+                status_output = ssh_connector.execute_command(command)
             
-            interfaces = self._parse_interface_list(output)
+            # Parsear información de estado
+            self._parse_interface_status(status_output)
             
-            # Si aún no obtenemos interfaces, usar un comando más básico
+            # Si no obtuvimos interfaces, usar un comando más básico
-            if not interfaces:
+            if not self.interface_info:
                 command = "show interface | include line protocol"
                 output = ssh_connector.execute_command(command)
-                interfaces = self._parse_interface_line_protocol(output)
+                self._parse_interface_line_protocol_to_info(output)
             
-            return interfaces
+            # Ahora obtenemos las descripciones
+            if self.interface_info:
+                self._get_interface_descriptions(ssh_connector)
             
         except Exception as e:
-            self.logger.warning(f"Error obteniendo interfaces: {str(e)}")
+            self.logger.warning(f"Error obteniendo información de interfaces: {str(e)}")
-            return []
     
-    def _parse_interface_list(self, output: str) -> List[str]:
+    def _parse_interface_status(self, output: str):
         """Parsea la salida de show interface status o show ip interface brief"""
-        interfaces = []
-        
         lines = output.split('\n')
         for line in lines:
             line = line.strip()
@@ -204,28 +211,109 @@ class CiscoMacTableParser:
             # Patrón para show interface status
             # Format: Port      Name               Status       Vlan       Duplex  Speed Type
             parts = line.split()
-            if len(parts) >= 2:
+            if len(parts) >= 3:
                 interface = parts[0]
                 
                 # Filtrar solo interfaces físicas relevantes
                 if self._is_physical_interface(interface):
-                    interfaces.append(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"
                     
-        return list(set(interfaces))  # Eliminar duplicados
+                    # 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"
                     
-    def _parse_interface_line_protocol(self, output: str) -> List[str]:
+                    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"""
-        interfaces = []
-        
         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:
-                interface = line.split(' is ')[0].strip()
+                parts = line.split(' is ')
+                interface = parts[0].strip()
                 if self._is_physical_interface(interface):
-                    interfaces.append(interface)
+                    # Extraer el estado
+                    status = "unknown"
+                    if len(parts) > 1:
+                        status_part = parts[1].split(',')[0].strip()
+                        status = self._normalize_status(status_part)
                     
-        return list(set(interfaces))
+                    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:
+            # Reconectar la sesión SSH para evitar errores de sesión inactiva
+            ssh_connector.disconnect()
+            if not ssh_connector.connect():
+                self.logger.warning("No se pudo reconectar para obtener descripciones de interfaces")
+                return
+            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():
+                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
+            # El formato Cisco es: Interface Status Protocol Description (Description puede tener espacios o estar vacía)
+            match = re.match(r'^(\S+)\s+(\S+)\s+(\S+)\s*(.*)$', line_stripped)
+            if match:
+                interface = match.group(1)
+                status_raw = match.group(2)
+                protocol_raw = match.group(3)
+                description = match.group(4).strip() if match.group(4) else ""
+                # Normalizar status (up/down/disabled/unknown)
+                status = self._normalize_status(status_raw)
+                # Actualizar o crear la interfaz en el diccionario
+                if interface in self.interface_info:
+                    self.interface_info[interface].status = status
+                    self.interface_info[interface].description = description
+                else:
+                    self.interface_info[interface] = InterfaceInfo(
+                        name=interface,
+                        status=status,
+                        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"""
@@ -343,7 +431,7 @@ class ZabbixLLDFormatter:
         self.device_hostname = device_hostname
         self.logger = logging.getLogger(__name__)
     
-    def format_for_zabbix(self, mac_entries: List[MacEntry], all_interfaces: List[str] = None) -> Dict[str, Any]:
+    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
@@ -362,7 +450,9 @@ class ZabbixLLDFormatter:
                     "device_hostname": self.device_hostname,
                     "mac_addresses": [],
                     "mac_count": 0,
-                    "vlans": set()
+                    "vlans": set(),
+                    "status": "unknown",
+                    "description": ""
                 }
             
             ports_data[port]["mac_addresses"].append({
@@ -372,17 +462,23 @@ class ZabbixLLDFormatter:
             })
             ports_data[port]["vlans"].add(entry.vlan)
         
-        # Añadir interfaces sin MACs si tenemos la lista de todas las interfaces
+        # Añadir todas las interfaces conocidas (con o sin MACs)
-        if all_interfaces:
+        if interface_info:
-            for interface in all_interfaces:
+            for interface_name, iface_info in interface_info.items():
-                if interface not in ports_data:
+                if interface_name not in ports_data:
-                    ports_data[interface] = {
+                    ports_data[interface_name] = {
-                        "port_name": interface,
+                        "port_name": interface_name,
                         "device_hostname": self.device_hostname,
                         "mac_addresses": [],
                         "mac_count": 0,
-                        "vlans": set()
+                        "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():
@@ -405,6 +501,8 @@ class ZabbixLLDFormatter:
                     "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
                 }
             }
@@ -520,8 +618,11 @@ Salida:
         if not mac_entries:
             logger.warning("No se obtuvieron entradas de tabla MAC")
         
-        # Formatear para Zabbix
+        # Formatear para Zabbix incluyendo información de interfaces
-        zabbix_data = zabbix_formatter.format_for_zabbix(mac_entries)
+        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)

src/test_new_format.py

@@ -7,7 +7,7 @@ import sys
 import os
 sys.path.append(os.path.dirname(os.path.abspath(__file__)))
 
-from cisco_mac_monitor import MacEntry, ZabbixLLDFormatter
+from cisco_mac_monitor import MacEntry, InterfaceInfo, ZabbixLLDFormatter
 import json
 
 # Crear datos de prueba
@@ -17,9 +17,16 @@ test_mac_entries = [
     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, ["Gi1/0/1", "Gi1/0/10", "Gi1/0/11"])
+result = formatter.format_for_zabbix(test_mac_entries, test_interface_info)
 
 # Mostrar resultado
 print(json.dumps(result, indent=2, ensure_ascii=False))
@@ -39,5 +46,7 @@ for item in data:
     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()