14 KiB
14 KiB
Guide de Développement : Créer son propre Parseur Switch
Ce guide explique comment étendre l'outil d'analyse réseau en ajoutant un nouveau parseur pour un modèle de switch non supporté nativement (ex: Cisco Meraki, Nexus, etc.).
Note : Tout au long de ce guide, remplacez
MODELE(en majuscule ou minuscule) par le nom de la marque ou du modèle du switch ciblé.
Étape 1 : Création du script de parsing
Dans le dossier Parseurs_logs_Switch/src/scripts/ (ou script/ selon votre arborescence), créez un nouveau fichier Python nommé json_MODELE.py.
Il est fortement recommandé de s'inspirer des parseurs existants comme json_hpe.py.
Voici le squelette de code à utiliser (exemple basé sur des logs au format TXT) :
#!/usr/bin/env python3
"""
Parser pour configuration switch MODELE vers un format json normalisé OpenConfig YANG
Extrait les différentes sections de logs et les convertit en objets Python, puis exporte le tout au format JSON OpenConfig.
"""
import json
import re
from scripts.export_modele import ParserMixin
from scripts.objets.data import MacAddress, Arp, Route, FIB, SlotInfo, VersionInfo, ComponentInfo, SlotManuInfo, ManuInfo, Power, SlotFanInfo, Fan, Sensor, EnvironmentInfo, SlotEnvInfo, SlotBootInfo, IrfMember, IrfLink, IrfTopology, IrfConfiguration, CpuUsage, CpuProcessInfo, ProcessInfo, Memory, TransceiverDiag, Transceiver, TransceiverInterface, Log, Multicast, Qos, Aaa, StpGlobal, StpInfo, StpPort, InterfaceCounter, NetworkInterface, LinkAggregation, ConfigVlan, ConfigIrfPort, ConfigInterface, ConfigAcl, ConfigUser, ConfigSnmp, DeviceConfiguration
from pathlib import Path
class ModeleParser(ParserMixin):
"""Parser pour fichier XML MODELE"""
def __init__(self, file_path: str):
"""Initialise le parser avec le chemin du fichier de logs MODELE"""
self.file = file_path
self.root = None
self.raw_data = None
self.config = None
self.config = {
'mac_addresses': [],
'arp_table': [],
'routing_table': [],
'fib_table': [],
'version': {},
'manuinfo': {},
'power': [],
'fans': [],
'environment': [],
'boot_loader': [],
'irf': [],
'irf_topology': [],
'irf_configuration': [],
'cpu_process': {},
'process': {},
'memory': [],
'transceiver': [],
'transceiver_interface': [],
'logbuffer': [],
'multicast': [],
'qos': [],
'aaa': [],
'stp': {},
'interfaces': [],
'link_aggregation': [],
'current_configuration': {},
'saved_configuration': {}
}
self.master = None
def load_file(self):
"""Charge le fichier texte brut avec gestion d'encodage"""
encodings_to_try = ['utf-8', 'latin-1', 'cp1252']
for encoding in encodings_to_try:
try:
with open(self.file, 'r', encoding=encoding) as f:
self.raw_data = f.readlines()
print(f"✓ Fichier chargé ({encoding}): {self.file}")
return
except UnicodeDecodeError:
continue
raise ValueError("Impossible de décoder le fichier avec les encodages standards")
def _parse_mac_address(self):
"""Parse display mac-address"""
mac_addresses = []
# TODO: Parcourir les logs et instancier des objets Mac Address
self.config['mac_addresses'] = mac_addresses
def _parse_arp_table(self):
"""Parse display arp all"""
arp_table = []
# TODO: Parcourir les logs et instancier des objets ARP
self.config['arp_table'] = arp_table
def _parse_routing_table(self):
"""Parse display ip routing-table"""
routes = []
# TODO: Parcourir les logs et instancier des objets Routing table
self.config['routing_table'] = routes
def _parse_fib_table(self):
"""Parse display fib"""
fib_entries = []
# TODO: Parcourir les logs et instancier des objets FIB
self.config['fib_table'] = fib_entries
def _parse_version(self):
"""Parse display version"""
version_info = VersionInfo()
# TODO: Parcourir les logs et instancier des objets Version
self.config["version"] = version_info
def _parse_device_manuinfo(self):
"""Parse display device manuinfo"""
manuinfo = ManuInfo()
# TODO: Parcourir les logs et instancier des objets ManuInfo
self.config["manuinfo"] = manuinfo
def _parse_power(self):
"""Parse display power"""
power_data = []
# TODO: Parcourir les logs et instancier des objets Power
self.config["power"] = power_data
def _parse_fan(self):
"""Parse display fan"""
fan_data = []
# TODO: Parcourir les logs et instancier des objets Fan
self.config["fans"] = fan_data
def _parse_environment(self):
"""Parse display environment"""
env_data = EnvironmentInfo()
# TODO: Parcourir les logs et instancier des objets Environment
self.config["environment"] = env_data
def _bootloader_information(self):
"""Parse boot-loader information"""
boot_data = []
# TODO: Parcourir les logs et instancier des objets Boot loader
self.config["boot_loader"] = boot_data
def _parse_irf(self):
"""Parse display irf"""
irf_data = []
# TODO: Parcourir les logs et instancier des objets IRF
self.config["irf"] = irf_data
def _parse_irf_topology(self):
"""Parse display irf topology"""
topo = []
# TODO: Parcourir les logs et instancier des objets IRF Topology
self.config["irf_topology"] = topo
def _parse_irf_configuration(self):
"""Parse display irf configuration"""
config = []
# TODO: Parcourir les logs et instancier des objets IRF Configuration
self.config["irf_configuration"] = config
def _parse_cpu_process(self):
"""Parse display process cpu slot [master]"""
cpu_data = CpuProcessInfo(slot=self.master)
# TODO: Parcourir les logs et instancier des objets CPU Process
self.config["cpu_process"] = cpu_data
def _parse_process(self):
"""Parse display process slot [master]"""
process_data = ProcessInfo(slot=self.master)
# TODO: Parcourir les logs et instancier des objets Process
self.config["process"] = process_data
def _parse_memory(self):
"""Parse display memory"""
memory_data = []
# TODO: Parcourir les logs et instancier des objets Memory
self.config["memory"] = memory_data
def _parse_transceiver_diagnosis(self):
"""Parse display transceiver diagnosis interface"""
data = []
# TODO: Parcourir les logs et instancier des objets Transceiver diagnosis
self.config["transceiver"] = data
def _parse_transceiver_interface(self):
"""Parse display transceiver interface"""
data = []
# TODO: Parcourir les logs et instancier des objets Transceiver interface
self.config["transceiver_interface"] = data
def _parse_logbuffer(self):
"""Parse display logbuffer"""
raw_table = []
# TODO: Parcourir les logs et instancier des objets Log
self.config["logbuffer"] = {
"raw_table": raw_table
}
def _parse_multicast(self):
"""Parse display multicast forwarding-table"""
multicast = []
# TODO: Parcourir les logs et instancier des objets Multicast
self.config["multicast"] = multicast
def _parse_qos(self):
"""Parse qos policy user-defined"""
qos = []
# TODO: Parcourir les logs et instancier des objets QOS
self.config["qos"] = qos
def _parse_aaa(self):
"""Parse dot1x"""
aaa = []
# TODO: Parcourir les logs et instancier des objets AAA
self.config["aaa"] = aaa
def _parse_stp(self):
"""Parse display stp"""
stp = StpInfo()
# TODO: Parcourir les logs et instancier des objets STP
self.config["stp"] = stp
def _parse_interface(self):
"""Parse display interface"""
INTERFACE_NAME_RE = re.compile(
r'^('
r'Bridge-Aggregation\d+'
r'|HundredGigE[\d/]+'
r'|Ten-GigabitEthernet[\d/]+'
r'|GigabitEthernet[\d/]+'
r'|M-GigabitEthernet[\d/]+'
r'|FortyGigE[\d/]+'
r'|XGE[\d/]+'
r'|Vlan-interface\d+'
r'|LoopBack\d+'
r'|InLoopBack\d+'
r'|NULL\d+'
r'|Register-Tunnel\d+'
r'|Tunnel\d+'
r')$',
re.IGNORECASE
)
interfaces = []
# TODO: Parcourir les logs et instancier des objets Interface
self.config["interfaces"] = interfaces
def _parse_link_aggregation(self):
"""Parse display link-aggregation verbose"""
aggs = []
# TODO: Parcourir les logs et instancier des objets Link Aggregation
self.config["link_aggregation"] = aggs
def _parse_current_configuration(self):
"""Parse display current-configuration"""
self.config["current_configuration"] = self._parse_configuration_section(
"display current-configuration"
)
def _parse_saved_configuration(self):
"""Parse display saved-configuration"""
self.config["saved_configuration"] = self._parse_configuration_section(
"display saved-configuration"
)
def _parse_configuration_section(self, section_marker: str) -> DeviceConfiguration:
"""Parse une section de configuration (current ou saved)"""
config = DeviceConfiguration()
# TODO: Parcourir les logs et instancier des objets Configuration
return config
def parse_all(self):
"""Parse tous les éléments de configuration"""
print("Début du parsing de la configuration MODELE...")
self.load_file()
self._parse_mac_address()
self._parse_arp_table()
self._parse_routing_table()
self._parse_fib_table()
self._parse_version()
self._parse_device_manuinfo()
self._parse_power()
self._parse_fan()
self._parse_environment()
self._bootloader_information()
self._parse_irf()
self._parse_irf_topology()
self._parse_irf_configuration()
self._parse_cpu_process()
self._parse_process()
self._parse_memory()
self._parse_transceiver_diagnosis()
self._parse_transceiver_interface()
self._parse_logbuffer()
self._parse_multicast() # Pas implémenté
self._parse_qos() # Pas implémenté
self._parse_aaa() # Pas implémenté
self._parse_stp()
self._parse_interface()
self._parse_link_aggregation()
self._parse_current_configuration()
self._parse_saved_configuration()
# display poe interface (a voir si existe dans d'autres logs)
print("✓ Parsing terminé avec succès!")
def export_to_json(self, output_file1: str, output_file2: str):
"""Exporte la configuration au format JSON OpenConfig"""
fw_name = self.config['current_configuration'].hostname
openconfig_data_c = self.to_openconfig_yang("modele", fw_name, "current_configuration")
openconfig_data_s = self.to_openconfig_yang("modele", fw_name, "saved_configuration")
with open(output_file1, "w", encoding="utf-8") as f:
json.dump(openconfig_data_c, f, indent=2, ensure_ascii=False)
with open(output_file2, "w", encoding="utf-8") as f:
json.dump(openconfig_data_s, f, indent=2, ensure_ascii=False)
print(f"✓ Configuration exportée vers: {output_file1} {output_file2}")
def generate_json_modele(input_file: str, output_file1: str, output_file2: str):
"""Génère le fichier JSON OpenConfig à partir du fichier XML MODELE"""
file = Path(input_file)
if not file.exists():
print(f"✗ Erreur: Le fichier '{file}' n'existe pas")
return
try:
parser = ModeleParser(str(file))
parser.parse_all()
parser.print_summary()
parser.export_to_json(output_file1, output_file2)
print(f"\n✓ Conversion terminée! Vérifiez le fichier: {output_file1} {output_file2}")
except Exception as e:
print(f"✗ Erreur: {e}")
return
Étape 2 : Intégration dans le moteur principal (main.py)
Pour que le script global prenne en charge le nouveau modèle, ouvrez le fichier Parseurs_logs_Switch/src/main.py :
- Importez votre nouvelle fonction en haut du fichier :
import scripts.json_modele as json_modele
- Ajoutez la condition correspondante dans le bloc de sélection du type de switch :
elif switch_type == "modele":
json_modele.generate_json_modele(input_data, output_file1_json, output_file2_json)
- Mettez à jour les instructions textuelles d'usage affichées dans la console si le paramètre saisi est incorrect afin d'inclure
"modele"dans la liste des choix valides.
Étape 3 : Adaptation de l'Interface Graphique (GUI)
Afin que le nouveau modèle apparaisse dans l'interface graphique unifiée :
- Ouvrez le fichier de gestion de la GUI (
gui_switch.pyracine ou script GUI dédié). - Repérez le composant de sélection (Menu déroulant / ComboBox ou liste de boutons radio) listant les switches.
- Ajoutez
MODELEà la liste des choix graphiques.
Important : faire attention si le script prend en compte un dossier ou un fichier.