dcstephenson/lldp-mapper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update core scanner, parser, SSH client, and UI

Browse Source
This commit is contained in:
D Stephenson
2026-05-05 20:07:24 +00:00
parent 40d4679a59
commit d32ca80a22
6 changed files with 313 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ssh_client.py
+120 -61
View File
@@ -1,41 +1,33 @@
# ssh_client.py - Netmiko SSH connections to FS switches
# ssh_client.py - SSH connections to FS, HP/Aruba ProCurve, and Dell switches
import re
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from netmiko import ConnectHandler, NetmikoTimeoutException, NetmikoAuthenticationException
from config import SSH_USERNAME, SSH_PASSWORD, SSH_PORT, SSH_TIMEOUT, DEVICE_TYPE
from parser import parse_lldp_neighbors, parse_mgmt_ip_from_interfaces
from parser import (parse_lldp_neighbors, parse_mgmt_ip_from_interfaces,
parse_aruba_procurve_local, parse_aruba_procurve_neighbors,
normalize_mac)
logger = logging.getLogger(__name__)
def connect_and_query(ip):
"""
SSH into a switch, run LLDP + interface commands.
Returns dict with switch info and neighbors, or error info.
"""
device = {
"device_type": DEVICE_TYPE,
"host": ip,
"username": SSH_USERNAME,
"password": SSH_PASSWORD,
"port": SSH_PORT,
"timeout": SSH_TIMEOUT,
"device_type": DEVICE_TYPE,
"host": ip,
"username": SSH_USERNAME,
"password": SSH_PASSWORD,
"port": SSH_PORT,
"timeout": SSH_TIMEOUT,
"global_delay_factor": 2,
"fast_cli": False,
"conn_timeout": SSH_TIMEOUT,
# "ssh_config_file": None,
# "disabled_algorithms": {},
# "transport": "paramiko",
"fast_cli": False,
"conn_timeout": SSH_TIMEOUT,
}
# FS switches use legacy ssh-rsa keys — must be explicitly allowed
import paramiko
_orig_connect = ConnectHandler.__init__
try:
logger.info(f"Connecting to {ip}...")
# Patch paramiko transport to allow legacy key algorithms
# Allow legacy ssh-rsa keys used by FS switches
import paramiko.transport as _pt
_orig_preferred_keys = _pt.Transport._preferred_keys
_pt.Transport._preferred_keys = (
@@ -45,40 +37,30 @@ def connect_and_query(ip):
)
with ConnectHandler(**device) as conn:
# Restore after connect
_pt.Transport._preferred_keys = _orig_preferred_keys
# Get hostname from prompt
hostname = conn.find_prompt().replace('#', '').replace('>', '').strip()
version_output = conn.send_command("show version", read_timeout=30)
vendor = _detect_vendor(version_output)
# Get LLDP neighbors
lldp_output = conn.send_command("show lldp neighbors", read_timeout=30)
if vendor == 'aruba_procurve':
chassis_id, mgmt_ip, model, firmware, neighbors = _query_aruba(conn, version_output, ip)
else:
chassis_id, mgmt_ip, model, firmware, neighbors = _query_fs(conn, version_output, ip)
# Get management IP from interface brief
intf_output = conn.send_command("show ip interface brief", read_timeout=30)
mgmt_ip = parse_mgmt_ip_from_interfaces(intf_output) or ip
logger.info(f" {hostname} ({ip}) [{vendor}]: {len(neighbors)} neighbors")
# Get chassis ID from LLDP local info
local_info_output = conn.send_command("show lldp local-information", read_timeout=30)
chassis_id = _extract_local_chassis(local_info_output) or ip
# Parse neighbors
neighbors = parse_lldp_neighbors(lldp_output, chassis_id, hostname, mgmt_ip)
# Get system description
sys_desc = _extract_system_desc(local_info_output)
logger.info(f" {hostname} ({ip}): {len(neighbors)} neighbors found")
return {
"success": True,
"ip": ip,
"hostname": hostname,
"chassis_id": chassis_id,
"mgmt_ip": mgmt_ip,
"description": sys_desc,
"neighbors": neighbors,
}
return {
"success": True,
"ip": ip,
"hostname": hostname,
"chassis_id": chassis_id,
"mgmt_ip": mgmt_ip,
"description": model,
"firmware": firmware,
"vendor": vendor,
"neighbors": neighbors,
}
except NetmikoAuthenticationException:
logger.error(f"Auth failed for {ip}")
@@ -91,27 +73,104 @@ def connect_and_query(ip):
return {"success": False, "ip": ip, "error": str(e)}
def _query_fs(conn, version_output, ip):
"""Run FS/IES switch-specific commands and return (chassis_id, mgmt_ip, model, firmware, neighbors)."""
intf_output = conn.send_command("show ip interface brief", read_timeout=30)
local_info_output = conn.send_command("show lldp local-information", read_timeout=30)
lldp_output = conn.send_command("show lldp neighbors", read_timeout=30)
mgmt_ip = parse_mgmt_ip_from_interfaces(intf_output) or ip
chassis_id = (_extract_local_chassis(local_info_output)
or _extract_mac_from_version(version_output)
or ip)
chassis_id = normalize_mac(chassis_id) if '-' in chassis_id or ':' in chassis_id else chassis_id
model = _fs_model(version_output) or _extract_system_desc(local_info_output)
firmware = _fs_firmware(version_output)
neighbors = parse_lldp_neighbors(lldp_output, chassis_id, '', mgmt_ip)
return chassis_id, mgmt_ip, model, firmware, neighbors
def _query_aruba(conn, version_output, ip):
"""Run HP/Aruba ProCurve-specific commands and return (chassis_id, mgmt_ip, model, firmware, neighbors)."""
local_output = conn.send_command("show lldp info local-device", read_timeout=30)
remote_output = conn.send_command("show lldp info remote-device", read_timeout=30)
local_info = parse_aruba_procurve_local(local_output)
chassis_id = local_info['chassis_id'] or _extract_mac_from_version(version_output) or ip
mgmt_ip = local_info['mgmt_ip'] or ip
model = _aruba_model(local_info.get('system_desc', ''))
firmware = _aruba_firmware(version_output)
neighbors = parse_aruba_procurve_neighbors(remote_output)
return chassis_id, mgmt_ip, model, firmware, neighbors
# ── Vendor detection ──────────────────────────────────────────────────────────
def _detect_vendor(version_output):
if 'Dell EMC Networking OS10' in version_output or 'Dell OS10' in version_output:
return 'dell_os10'
if '/ws/swbuildm/' in version_output:
return 'aruba_procurve'
return 'fs'
# ── FS/IES extractors ─────────────────────────────────────────────────────────
def _extract_local_chassis(output):
import re
m = re.search(r'Chassis ID\s*:\s*([0-9A-Fa-f\-:]+)', output)
m = re.search(r'Chassis ID\s*:\s*(.+)', output)
return m.group(1).strip() if m else None
def _extract_mac_from_version(output):
m = re.search(r'MAC Address\s*:\s*([0-9A-Fa-f]{2}(?:[:\-][0-9A-Fa-f]{2}){5})', output, re.IGNORECASE)
return m.group(1).strip() if m else None
def _fs_model(output):
m = re.search(r'(?:Model Name|Product)\s*:\s*([\w\-]+)', output, re.IGNORECASE)
return f"FS {m.group(1)}" if m else ''
def _fs_firmware(output):
fw_m = re.search(r'Software Version\s*:\s*(\S+)', output, re.IGNORECASE)
date_m = re.search(r'Build Date\s*:\s*(\d{2})-(\d{2})-(\d{4})T', output, re.IGNORECASE)
if not fw_m:
return ''
version = fw_m.group(1)
if date_m:
day, month, year = date_m.group(1), date_m.group(2), date_m.group(3)
return f"{version} ({year}-{month}-{day})"
return version
def _extract_system_desc(output):
import re
m = re.search(r'System Description\s*:\s*(.+)', output)
m = re.search(r'System Description\s*:\s*(.*?)(?=\n\S|\Z)', output, re.DOTALL | re.IGNORECASE)
return ' '.join(m.group(1).split()) if m else ''
# ── HP/Aruba extractors ───────────────────────────────────────────────────────
def _aruba_model(system_desc):
"""Extract model from LLDP system description, e.g. 'Aruba JL258A 2930F-8G-PoE+-2SFP+ Switch'."""
m = re.match(r'((?:Aruba|HP)\s+\S+\s+[\w\-\+]+)', system_desc, re.IGNORECASE)
return m.group(1) if m else system_desc.split(',')[0] if system_desc else ''
def _aruba_firmware(version_output):
"""Extract firmware version from image stamp, e.g. 'WC.16.10.0012'."""
m = re.search(r'^\s+([A-Z]+\.\d+\.\d+\.\d+)\s*$', version_output, re.MULTILINE)
return m.group(1).strip() if m else ''
# ── Scan orchestration ────────────────────────────────────────────────────────
def scan_all_switches(ip_list, progress_callback=None, max_workers=10):
"""
Scan all switches in parallel.
progress_callback(done, total, current_ip, result) called after each switch.
Returns list of results.
"""
results = []
total = len(ip_list)
done = 0
done = 0
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_ip = {executor.submit(connect_and_query, ip): ip for ip in ip_list}
@@ -122,7 +181,7 @@ def scan_all_switches(ip_list, progress_callback=None, max_workers=10):
result = future.result()
except Exception as e:
result = {"success": False, "ip": ip, "error": str(e)}
import time; time.sleep(2) # 2s delay between switches to avoid RADIUS lockout
import time; time.sleep(2) # avoid RADIUS lockout
results.append(result)
done += 1