git/los-vecinos-de-cisco
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
Reviewed-on: #1
This commit was merged in pull request #1.
This commit is contained in:
git
2026-03-11 16:44:40 +00:00
parent f12e236e82 5576cb209f
commit 00572f598d
10 changed files with 183 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -0,0 +1 @@
__pycache__
CHANGELOG.md
+14
View File
@@ -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
+3 -3
View File
@@ -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 cisco_mac_http_service.py /app/
COPY --chmod=755 --chown=monitor:monitor src/cisco_mac_monitor.py /app/
COPY --chmod=755 --chown=monitor:monitor src/cisco_mac_http_service.py /app/
# Exponer puerto HTTPS
EXPOSE 8443
example → Docs/example
View File
build.sh
+1 -1
View File
@@ -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
+4
View File
@@ -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 \
cisco_mac_http_service.py → src/cisco_mac_http_service.py
+5 -5
View File
@@ -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
all_interfaces = getattr(mac_parser, 'detected_interfaces', [])
# Formatear para Zabbix incluyendo todas las interfaces
zabbix_data = zabbix_formatter.format_for_zabbix(mac_entries, all_interfaces)
# Formatear para Zabbix incluyendo información de interfaces
zabbix_data = zabbix_formatter.format_for_zabbix(
mac_entries,
mac_parser.interface_info
)
return zabbix_data
cisco_mac_monitor.py → src/cisco_mac_monitor.py
+144 -43
View File
@@ -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
interfaces = self._get_interface_list(ssh_connector)
self.logger.info(f"Detectadas {len(interfaces)} interfaces")
# 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
@@ -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]:
"""Obtiene la lista de todas las interfaces del dispositivo"""
def _get_interface_info(self, ssh_connector: CiscoSSHConnector):
"""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
if not interfaces:
# 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)
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)}")
return []
self.logger.warning(f"Error obteniendo información de interfaces: {str(e)}")
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)
return list(set(interfaces)) # Eliminar duplicados
# 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(self, output: str) -> List[str]:
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)
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()
return list(set(interfaces))
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
if all_interfaces:
for interface in all_interfaces:
if interface not in ports_data:
ports_data[interface] = {
"port_name": interface,
# 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()
"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
zabbix_data = zabbix_formatter.format_for_zabbix(mac_entries)
# 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)
requirements.txt → src/requirements.txt
View File
test_new_format.py → src/test_new_format.py
+11 -2
View File
@@ -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()