887 lines
29 KiB
Python
887 lines
29 KiB
Python
# -*- coding: utf-8 -*-
|
||
# Licensed under the Apache License, Version 2.0
|
||
|
||
"""CLI output parsers for HPE/Aruba ProCurve switches.
|
||
|
||
All functions are pure (no side effects) and operate on pre-collected
|
||
command output strings. The same parsers are shared by the SSH and Telnet
|
||
backends.
|
||
"""
|
||
|
||
import re
|
||
from typing import Dict, List, Optional
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# MAC address helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def procurve_mac_to_std(mac: str) -> str:
|
||
"""Convert ProCurve MAC format ``a1b2c3-d4e5f6`` to ``a1:b2:c3:d4:e5:f6``."""
|
||
clean = mac.replace("-", "").replace(":", "").replace(".", "").lower()
|
||
if len(clean) != 12:
|
||
return mac
|
||
return ":".join(clean[i : i + 2] for i in range(0, 12, 2))
|
||
|
||
|
||
def _kv(output: str, key: str) -> str:
|
||
"""Extract value from ``key : value`` style ProCurve output (case-insensitive)."""
|
||
for line in output.splitlines():
|
||
if re.search(re.escape(key), line, re.IGNORECASE):
|
||
parts = line.split(":", 1)
|
||
if len(parts) == 2:
|
||
return parts[1].strip()
|
||
return ""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show system-information
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_system_info(output: str) -> Dict:
|
||
"""Parse ``show system-information`` output.
|
||
|
||
Returns a dict with keys: hostname, os_version, serial_number, model,
|
||
base_mac, uptime_seconds.
|
||
|
||
Handles both older ProCurve style (two-column layout) and newer Aruba style.
|
||
|
||
Example older output::
|
||
|
||
System Name : myswitch
|
||
Software revision : R.11.27 Base MAC Addr : a1b2c3-d4e5f6
|
||
Serial Number : SGXXXXXXXXXX Up Time : 5 days 3:14
|
||
|
||
Example newer output::
|
||
|
||
System Name : ArubaSW
|
||
Firmware revision: YA.16.04.0006
|
||
Serial Number : SGXXXXXXXXXX
|
||
"""
|
||
# Use a regex to extract key:value pairs from each line.
|
||
# ProCurve CLI may pack two k/v pairs on the same line with many spaces
|
||
# between them, e.g.:
|
||
# " Software revision : R.11.27 Base MAC Addr : a1b2c3-d4e5f6"
|
||
# We capture all k:v pairs on every line.
|
||
_KV_RE = re.compile(
|
||
r"([A-Za-z][A-Za-z0-9 ()/#-]{1,40}?)\s*:\s*(\S.*?)(?=\s{3,}[A-Za-z]|$)"
|
||
)
|
||
kv_map: Dict[str, str] = {}
|
||
for raw_line in output.splitlines():
|
||
for m in _KV_RE.finditer(raw_line):
|
||
k = m.group(1).strip()
|
||
v = m.group(2).strip()
|
||
if k and v:
|
||
kv_map[k] = v
|
||
|
||
def get(key: str) -> str:
|
||
for k, v in kv_map.items():
|
||
if key.lower() in k.lower():
|
||
return v
|
||
return ""
|
||
|
||
hostname = get("System Name")
|
||
os_version = (
|
||
get("Software revision")
|
||
or get("Firmware revision")
|
||
or get("Software Revision")
|
||
or get("Firmware Version")
|
||
)
|
||
serial = get("Serial Number") or get("Serial Num")
|
||
base_mac = procurve_mac_to_std(get("Base MAC Addr") or get("MAC Address") or "")
|
||
model = get("System Model") or get("product_model") or ""
|
||
|
||
# Parse uptime — ProCurve formats: "5 days 3:14" or "0 days 0:05"
|
||
# Also: "5 day(s), 3:14:00"
|
||
uptime_seconds = _parse_procurve_uptime(get("Up Time") or get("Uptime") or "")
|
||
|
||
return {
|
||
"hostname": hostname,
|
||
"os_version": os_version,
|
||
"serial_number": serial,
|
||
"base_mac": base_mac,
|
||
"model": model,
|
||
"uptime_seconds": uptime_seconds,
|
||
}
|
||
|
||
|
||
def _parse_procurve_uptime(uptime_str: str) -> float:
|
||
"""Convert ProCurve uptime string to total seconds.
|
||
|
||
Handles formats::
|
||
|
||
5 days 3:14 → 5*86400 + 3*3600 + 14*60
|
||
0 days 0:05:03 → 303
|
||
5 day(s), 3:14:00 → same
|
||
1 hr 5 min 3 sec → some devices
|
||
"""
|
||
total = 0.0
|
||
# Format: "N days H:MM" or "N days H:MM:SS"
|
||
m = re.search(r"(\d+)\s+days?\s+(\d+):(\d+)(?::(\d+))?", uptime_str, re.I)
|
||
if m:
|
||
total = (
|
||
int(m.group(1)) * 86400
|
||
+ int(m.group(2)) * 3600
|
||
+ int(m.group(3)) * 60
|
||
+ int(m.group(4) or 0)
|
||
)
|
||
return float(total)
|
||
|
||
# Format: "Xd Xh Xm Xs" or "X day(s), X hour(s), ..."
|
||
days = re.search(r"(\d+)\s+day", uptime_str, re.I)
|
||
hours = re.search(r"(\d+)\s+h(?:r|our)", uptime_str, re.I)
|
||
minutes = re.search(r"(\d+)\s+min", uptime_str, re.I)
|
||
seconds = re.search(r"(\d+)\s+sec", uptime_str, re.I)
|
||
if days or hours or minutes or seconds:
|
||
total = (
|
||
int(days.group(1) if days else 0) * 86400
|
||
+ int(hours.group(1) if hours else 0) * 3600
|
||
+ int(minutes.group(1) if minutes else 0) * 60
|
||
+ int(seconds.group(1) if seconds else 0)
|
||
)
|
||
return float(total)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show version
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_version(output: str) -> str:
|
||
"""Extract firmware version from ``show version`` or ``show system-information``.
|
||
|
||
ProCurve ``show version`` example::
|
||
|
||
HP J9565A 2520G-8-PoE Switch
|
||
Software revision : R.11.27
|
||
|
||
Aruba image stamp format::
|
||
|
||
Image stamp: /ws/.../WC.16.11.0004
|
||
WC.16.11.0004
|
||
"""
|
||
# "Software revision : R.11.27" style
|
||
m = re.search(r"(?:software|firmware)\s+revision\s*:\s*(\S+)", output, re.I)
|
||
if m:
|
||
return m.group(1)
|
||
# Bare revision on its own line (image stamp format)
|
||
for line in output.splitlines():
|
||
line = line.strip()
|
||
if re.match(r"^[A-Z]{2}\.\d+\.\d+", line):
|
||
return line
|
||
return ""
|
||
|
||
|
||
def parse_model_from_version(output: str) -> str:
|
||
"""Extract model from ``show version`` first line.
|
||
|
||
Example: ``HP J9565A 2520G-8-PoE Switch`` → ``2520G-8-PoE``
|
||
"""
|
||
first = output.strip().splitlines()[0] if output.strip() else ""
|
||
m = re.search(r"HP\s+\S+\s+(\S+)\s+Switch", first, re.I)
|
||
if m:
|
||
return m.group(1)
|
||
# Aruba format: "Aruba 2530-8-PoE+ Switch"
|
||
m = re.search(r"(?:Aruba|HP)\s+(\d\S+)\s+(?:Switch|)", first, re.I)
|
||
if m:
|
||
return m.group(1)
|
||
return ""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show interfaces brief / show port status
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# ProCurve "show interfaces brief" tabular header line reference:
|
||
# Port Type | Enabled Link MDI Mode FlwCtrl BCast
|
||
# ------ ------------ + ------- ------ ----- ----------- -------- -----
|
||
# 1 100/1000T | Yes Down Auto Unknown off 0
|
||
|
||
_INTF_BRIEF_RE = re.compile(
|
||
r"^\s*(\S+)\s+" # Port
|
||
r"(\S+)\s+\|" # Type |
|
||
r"\s+(Yes|No)\s+" # Enabled
|
||
r"(Up|Down)\s+" # Link
|
||
r"\S+\s+" # MDI (ignored)
|
||
r"(\S+)", # Mode (speed/duplex)
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def parse_interfaces_brief(output: str) -> Dict[str, Dict]:
|
||
"""Parse ``show interfaces brief`` tabular output.
|
||
|
||
Returns dict keyed by port name with sub-keys:
|
||
is_up, is_enabled, speed, duplex, description.
|
||
"""
|
||
interfaces: Dict[str, Dict] = {}
|
||
for line in output.splitlines():
|
||
m = _INTF_BRIEF_RE.match(line)
|
||
if not m:
|
||
continue
|
||
port, itype, enabled, link, mode = (
|
||
m.group(1), m.group(2), m.group(3), m.group(4), m.group(5)
|
||
)
|
||
speed = 0.0
|
||
duplex = ""
|
||
# Mode examples: "1000FDx", "100HDx", "Unknown", "Auto"
|
||
sm = re.match(r"(\d+)(FDx|HDx)?", mode, re.I)
|
||
if sm:
|
||
speed = float(sm.group(1))
|
||
duplex = "full" if (sm.group(2) or "").lower() == "fdx" else "half"
|
||
|
||
interfaces[port] = {
|
||
"is_up": link.lower() == "up",
|
||
"is_enabled": enabled.lower() == "yes",
|
||
"speed": speed,
|
||
"duplex": duplex,
|
||
"description": "",
|
||
"last_flapped": -1.0,
|
||
"mtu": -1,
|
||
"mac_address": "",
|
||
}
|
||
return interfaces
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show interfaces (per-port detail for MAC / description)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_interface_detail(output: str, port: str) -> Dict:
|
||
"""Parse ``show interfaces <port>`` for MAC address and description."""
|
||
mac = procurve_mac_to_std(_kv(output, "MAC address"))
|
||
name = _kv(output, "Name") or _kv(output, "Description")
|
||
return {"mac_address": mac, "description": name}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show ip / show ip interface
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_ip_interfaces(output: str) -> Dict[str, Dict]:
|
||
"""Parse ``show ip`` or ``show ip interface brief`` output.
|
||
|
||
Returns NAPALM-compatible dict::
|
||
|
||
{"VLAN1": {"ipv4": {"192.168.0.1": {"prefix_length": 24}}}}
|
||
|
||
ProCurve ``show ip`` multi-VLAN format::
|
||
|
||
IP Configuration for VLAN 1:
|
||
IP Config : Manual
|
||
IP Address : 192.168.0.1
|
||
Subnet Mask : 255.255.255.0
|
||
|
||
IP Configuration for VLAN 99:
|
||
...
|
||
"""
|
||
result: Dict[str, Dict] = {}
|
||
current_vlan = ""
|
||
current_ip = ""
|
||
|
||
for line in output.splitlines():
|
||
# Detect VLAN section header: "IP Configuration for VLAN X" or "... VLAN 1:"
|
||
vm = re.search(r"IP\s+Config(?:uration)?\s+for\s+(?:VLAN\s+)?(\S+)", line, re.I)
|
||
if vm:
|
||
raw_id = vm.group(1).rstrip(":")
|
||
# Strip a leading "VLAN" prefix if the regex captured it (e.g. "VLAN1"→"1")
|
||
num = re.sub(r"^VLAN", "", raw_id, flags=re.I)
|
||
current_vlan = f"VLAN{num}"
|
||
result.setdefault(current_vlan, {"ipv4": {}, "ipv6": {}})
|
||
continue
|
||
|
||
# Single VLAN output (no header): detect from "IP Address" line
|
||
if re.search(r"IP\s+Address\s*:", line, re.I):
|
||
ip = line.split(":", 1)[-1].strip()
|
||
if ip and ip not in ("0.0.0.0", ""):
|
||
current_ip = ip
|
||
if not current_vlan:
|
||
current_vlan = "VLAN1"
|
||
result.setdefault(current_vlan, {"ipv4": {}, "ipv6": {}})
|
||
continue
|
||
|
||
if re.search(r"Subnet\s+Mask\s*:", line, re.I) and current_ip:
|
||
mask = line.split(":", 1)[-1].strip()
|
||
try:
|
||
import netaddr
|
||
prefix_len = netaddr.IPNetwork(f"{current_ip}/{mask}").prefixlen
|
||
except Exception:
|
||
prefix_len = 24
|
||
if current_vlan:
|
||
result[current_vlan]["ipv4"][current_ip] = {"prefix_length": prefix_len}
|
||
current_ip = ""
|
||
continue
|
||
|
||
# Clean up empty entries
|
||
return {k: v for k, v in result.items() if v["ipv4"] or v["ipv6"]}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show arp
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_ARP_RE = re.compile(
|
||
r"(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})" # IP
|
||
r"\s+"
|
||
r"([0-9a-fA-F]{6}-[0-9a-fA-F]{6}|" # MAC ProCurve hex-dash format
|
||
r"[0-9a-fA-F:]{17})" # or standard colon format
|
||
r"\s+(\S+)" # Type
|
||
r"(?:\s+(\S+))?", # Port (optional)
|
||
)
|
||
|
||
|
||
def parse_arp_table(output: str) -> List[Dict]:
|
||
"""Parse ``show arp`` output into NAPALM ARP table format."""
|
||
table = []
|
||
for line in output.splitlines():
|
||
m = _ARP_RE.search(line)
|
||
if not m:
|
||
continue
|
||
ip, mac, entry_type, port = (
|
||
m.group(1), m.group(2), m.group(3), m.group(4) or ""
|
||
)
|
||
table.append(
|
||
{
|
||
"interface": port,
|
||
"mac": procurve_mac_to_std(mac),
|
||
"ip": ip,
|
||
"age": -1.0,
|
||
}
|
||
)
|
||
return table
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show mac-address
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_MAC_RE = re.compile(
|
||
r"([0-9a-fA-F]{6}-[0-9a-fA-F]{6}|[0-9a-fA-F:]{17})" # MAC
|
||
r"\s+(\S+)" # Port
|
||
r"\s+(\d+)" # VLAN
|
||
r"\s+(\w+)", # Type
|
||
)
|
||
|
||
|
||
def parse_mac_table(output: str) -> List[Dict]:
|
||
"""Parse ``show mac-address`` into NAPALM MAC table format."""
|
||
table = []
|
||
for line in output.splitlines():
|
||
m = _MAC_RE.search(line)
|
||
if not m:
|
||
continue
|
||
mac, port, vlan, entry_type = m.group(1), m.group(2), m.group(3), m.group(4)
|
||
table.append(
|
||
{
|
||
"mac": procurve_mac_to_std(mac),
|
||
"interface": port,
|
||
"vlan": int(vlan),
|
||
"static": entry_type.lower() in ("static", "permanent"),
|
||
"active": True,
|
||
"moves": -1,
|
||
"last_move": -1.0,
|
||
}
|
||
)
|
||
return table
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show lldp info remote-device
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_lldp_neighbors(output: str) -> Dict[str, List[Dict]]:
|
||
"""Parse ``show lldp info remote-device`` into NAPALM neighbors dict.
|
||
|
||
Output format::
|
||
|
||
LocalPort | ChassisId PortId SysName ...
|
||
--------- + ------------------- ------------ ------------ ...
|
||
1 | a1 b2 c3 d4 e5 f6 Gi 0/1 otherswitch ...
|
||
"""
|
||
neighbors: Dict[str, List[Dict]] = {}
|
||
in_table = False
|
||
|
||
for line in output.splitlines():
|
||
# Skip header / separator lines
|
||
if re.match(r"\s*-+\s*\+", line) or re.match(r"\s*LocalPort", line, re.I):
|
||
in_table = True
|
||
continue
|
||
if not in_table:
|
||
continue
|
||
if not line.strip():
|
||
continue
|
||
|
||
# Format: " 1 | aa bb cc ... Gi0/1 hostname ..."
|
||
m = re.match(r"\s*(\S+)\s*\|\s*(.+)", line)
|
||
if not m:
|
||
continue
|
||
local_port = m.group(1)
|
||
rest = m.group(2).strip()
|
||
|
||
# ChassisId: may be space-separated octets (aa bb cc dd ee ff) or MAC
|
||
# Then PortId, then SysName
|
||
# Split by 2+ spaces to separate columns
|
||
cols = re.split(r"\s{2,}", rest)
|
||
if len(cols) < 3:
|
||
continue
|
||
chassis_raw = cols[0].strip()
|
||
port_id = cols[1].strip()
|
||
sys_name = cols[2].strip()
|
||
|
||
# Normalise chassis ID: space-separated octets → colon MAC
|
||
chassis = re.sub(r"\s+", "", chassis_raw)
|
||
if re.match(r"^[0-9a-fA-F]{12}$", chassis):
|
||
chassis = ":".join(chassis[i : i + 2] for i in range(0, 12, 2))
|
||
|
||
neighbors.setdefault(local_port, []).append(
|
||
{"hostname": sys_name, "port": port_id}
|
||
)
|
||
|
||
return neighbors
|
||
|
||
|
||
def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]:
|
||
"""Parse ``show lldp info remote-device detail`` output.
|
||
|
||
Returns NAPALM lldp_neighbors_detail format.
|
||
"""
|
||
neighbors: Dict[str, List[Dict]] = {}
|
||
current_port = ""
|
||
entry: Dict = {}
|
||
|
||
def _save():
|
||
if current_port and entry:
|
||
neighbors.setdefault(current_port, []).append(entry.copy())
|
||
|
||
for line in output.splitlines():
|
||
# New port section: "LLDP Remote Device Information Detail for Port N"
|
||
m = re.search(r"for\s+[Pp]ort\s+(\S+)", line)
|
||
if m:
|
||
_save()
|
||
current_port = m.group(1)
|
||
entry = {
|
||
"parent_interface": "",
|
||
"remote_chassis_id": "",
|
||
"remote_port": "",
|
||
"remote_port_description": "",
|
||
"remote_system_name": "",
|
||
"remote_system_description": "",
|
||
"remote_system_capab": [],
|
||
"remote_system_enable_capab": [],
|
||
}
|
||
continue
|
||
|
||
if not entry:
|
||
continue
|
||
|
||
low = line.lower()
|
||
if "chassis id" in low:
|
||
val = line.split(":", 1)[-1].strip()
|
||
entry["remote_chassis_id"] = procurve_mac_to_std(
|
||
re.sub(r"\s+", "", val)
|
||
) if re.search(r"[0-9a-fA-F ]{14,}", val) else val
|
||
elif "port id" in low and "descr" not in low:
|
||
entry["remote_port"] = line.split(":", 1)[-1].strip()
|
||
elif "port description" in low:
|
||
entry["remote_port_description"] = line.split(":", 1)[-1].strip()
|
||
elif "system name" in low:
|
||
entry["remote_system_name"] = line.split(":", 1)[-1].strip()
|
||
elif "system description" in low:
|
||
entry["remote_system_description"] = line.split(":", 1)[-1].strip()
|
||
elif "system capabilities" in low and "enabled" not in low:
|
||
entry["remote_system_capab"] = _parse_capabilities(
|
||
line.split(":", 1)[-1].strip()
|
||
)
|
||
elif "enabled capabilities" in low or ("system capabilities" in low and "enabled" in low):
|
||
entry["remote_system_enable_capab"] = _parse_capabilities(
|
||
line.split(":", 1)[-1].strip()
|
||
)
|
||
|
||
_save()
|
||
return neighbors
|
||
|
||
|
||
def _parse_capabilities(cap_str: str) -> List[str]:
|
||
"""Convert ProCurve capability abbreviations to NAPALM strings."""
|
||
_MAP = {
|
||
"B": "bridge",
|
||
"R": "router",
|
||
"T": "telephone",
|
||
"W": "wlan-access-point",
|
||
"H": "host",
|
||
"D": "docsis-cable-device",
|
||
"O": "other",
|
||
"S": "station",
|
||
"C": "customer-vlan-bridge",
|
||
}
|
||
result = []
|
||
for token in re.split(r"[,\s]+", cap_str):
|
||
token = token.strip()
|
||
if not token:
|
||
continue
|
||
result.append(_MAP.get(token, token.lower()))
|
||
return result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show ntp associations / show timesync
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_ntp_servers(output: str) -> Dict[str, Dict]:
|
||
"""Parse ``show ntp associations`` or ``show ntp status`` output.
|
||
|
||
Returns dict of {server_ip: {}} as per NAPALM get_ntp_servers.
|
||
"""
|
||
servers: Dict[str, Dict] = {}
|
||
for line in output.splitlines():
|
||
# " 10.0.0.1 sys.peer 2 ..." or "NTP server: 10.0.0.1"
|
||
m = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", line)
|
||
if m:
|
||
ip = m.group(1)
|
||
# Skip gateway / default-gateway entries
|
||
if "gateway" in line.lower() or "default" in line.lower():
|
||
continue
|
||
servers[ip] = {}
|
||
return servers
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show environment
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_environment(output: str) -> Dict:
|
||
"""Parse ``show environment`` output into NAPALM environment dict."""
|
||
env: Dict = {
|
||
"fans": {},
|
||
"temperature": {},
|
||
"power": {},
|
||
"cpu": [{"%usage": 0.0}],
|
||
"memory": {"available_ram": -1, "used_ram": -1},
|
||
}
|
||
|
||
# CPU utilisation from show system-information
|
||
m = re.search(r"CPU\s+Util(?:ization)?\s*[(%:]+\s*(\d+)", output, re.I)
|
||
if m:
|
||
env["cpu"][0]["%usage"] = float(m.group(1))
|
||
|
||
# Memory
|
||
total_m = re.search(r"Memory\s*[-–]\s*Total\s*[:\s]+(\d+)", output, re.I)
|
||
free_m = re.search(r"Free\s*[:\s]+(\d+)", output, re.I)
|
||
if total_m:
|
||
env["memory"]["available_ram"] = int(total_m.group(1))
|
||
if free_m:
|
||
env["memory"]["used_ram"] = (
|
||
env["memory"]["available_ram"] - int(free_m.group(1))
|
||
if env["memory"]["available_ram"] > 0
|
||
else -1
|
||
)
|
||
|
||
# Temperature sensors: "Sensor 1: 38C" or "Temperature: 38C"
|
||
for m in re.finditer(
|
||
r"(?:Sensor\s+(\d+)|Temperature(?:\s+Sensor)?)\s*[:\s]+(\d+)\s*[Cc°]",
|
||
output,
|
||
):
|
||
sensor_id = f"sensor_{m.group(1) or '1'}"
|
||
temp = float(m.group(2))
|
||
env["temperature"][sensor_id] = {
|
||
"temperature": temp,
|
||
"is_alert": False,
|
||
"is_critical": False,
|
||
}
|
||
|
||
# Fan status
|
||
for m in re.finditer(
|
||
r"Fan\s*(\d+)\s*[:\s]+(\S+)", output, re.I
|
||
):
|
||
fan_id = f"fan_{m.group(1)}"
|
||
status = m.group(2).lower()
|
||
env["fans"][fan_id] = {"status": status in ("ok", "good", "normal", "operating")}
|
||
|
||
return env
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show local-users
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_users(output: str) -> Dict[str, Dict]:
|
||
"""Parse ``show local-users`` or ``show users`` output.
|
||
|
||
Returns NAPALM users dict.
|
||
"""
|
||
users: Dict[str, Dict] = {}
|
||
for line in output.splitlines():
|
||
# " manager Operator Enabled ..." or " manager Manager ..."
|
||
m = re.match(
|
||
r"\s+(\w+)\s+(\w+(?:\s+\w+)?)\s+(Enabled|Disabled)",
|
||
line,
|
||
re.I,
|
||
)
|
||
if not m:
|
||
continue
|
||
uname, level, _state = m.group(1), m.group(2).lower(), m.group(3)
|
||
# Map privilege level
|
||
if "manager" in level or "admin" in level:
|
||
priv = 15
|
||
elif "operator" in level:
|
||
priv = 5
|
||
else:
|
||
priv = 1
|
||
users[uname] = {
|
||
"level": priv,
|
||
"password": "",
|
||
"sshkeys": [],
|
||
}
|
||
return users
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# show snmp-server / show snmp
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_snmp_info(output: str) -> Dict:
|
||
"""Parse ``show snmp-server`` output into NAPALM snmp_information dict."""
|
||
communities: Dict[str, Dict] = {}
|
||
contact = _kv(output, "System Contact") or _kv(output, "Contact")
|
||
location = _kv(output, "System Location") or _kv(output, "Location")
|
||
chassis_id = ""
|
||
|
||
for line in output.splitlines():
|
||
# " community_name Manager Restricted" (read only)
|
||
# " community_name Operator Unrestricted" (read write)
|
||
m = re.match(r"\s+(\S+)\s+(Manager|Operator)\s+(Restricted|Unrestricted)", line, re.I)
|
||
if m:
|
||
comm, _role, access = m.group(1), m.group(2), m.group(3)
|
||
communities[comm] = {
|
||
"acl": "",
|
||
"mode": "ro" if access.lower() == "restricted" else "rw",
|
||
}
|
||
|
||
return {
|
||
"chassis_id": chassis_id,
|
||
"community": communities,
|
||
"contact": contact,
|
||
"location": location,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# VLAN helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def expand_procurve_ports(port_str: str) -> List[str]:
|
||
"""Expand a ProCurve port-list string into a flat list of port names.
|
||
|
||
Handles:
|
||
- Numeric ranges: ``1-4,6,8`` → ``["1","2","3","4","6","8"]``
|
||
- Alphanumeric ranges: ``A1-A8`` → ``["A1","A2",..."A8"]``
|
||
- Trunk ports: ``Trk1,Trk2`` → ``["Trk1","Trk2"]``
|
||
- Mixed: ``1-3,Trk1,A1`` → ``["1","2","3","Trk1","A1"]``
|
||
"""
|
||
ports: List[str] = []
|
||
for token in port_str.split(","):
|
||
token = token.strip()
|
||
if not token:
|
||
continue
|
||
|
||
# Try alphanumeric range: e.g. "A1-A8" (prefix + digits)
|
||
alpha_range = re.match(r"^([A-Za-z]+)(\d+)-([A-Za-z]+)(\d+)$", token)
|
||
if alpha_range:
|
||
prefix = alpha_range.group(1)
|
||
start = int(alpha_range.group(2))
|
||
end = int(alpha_range.group(4))
|
||
ports.extend(f"{prefix}{i}" for i in range(start, end + 1))
|
||
continue
|
||
|
||
# Try pure numeric range: e.g. "1-4"
|
||
num_range = re.match(r"^(\d+)-(\d+)$", token)
|
||
if num_range:
|
||
start = int(num_range.group(1))
|
||
end = int(num_range.group(2))
|
||
ports.extend(str(i) for i in range(start, end + 1))
|
||
continue
|
||
|
||
# Single port (numeric, Trk, alphanumeric stacking port etc.)
|
||
ports.append(token)
|
||
|
||
return ports
|
||
|
||
|
||
def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]:
|
||
"""Parse VLAN blocks from ProCurve ``show running-config`` output.
|
||
|
||
Extracts ``name``, ``tagged``, and ``untagged`` port lists per VLAN.
|
||
|
||
Typical running-config VLAN block::
|
||
|
||
vlan 1
|
||
name "DEFAULT_VLAN"
|
||
untagged 1-4,6-8
|
||
ip address 192.168.0.1 255.255.255.0
|
||
exit
|
||
|
||
vlan 10
|
||
name "MGMT"
|
||
tagged 1,2,5
|
||
exit
|
||
|
||
Returns extended NAPALM-compatible dict::
|
||
|
||
{
|
||
1: {"name": "DEFAULT_VLAN",
|
||
"interfaces": ["1","2","3","4","6","7","8","5"],
|
||
"tagged": [],
|
||
"untagged": ["1","2","3","4","6","7","8"]},
|
||
10: {"name": "MGMT",
|
||
"interfaces": ["1","2","5"],
|
||
"tagged": ["1","2","5"],
|
||
"untagged": []},
|
||
}
|
||
"""
|
||
vlans: Dict[int, Dict] = {}
|
||
current_id: Optional[int] = None
|
||
|
||
for line in config_text.splitlines():
|
||
stripped = line.strip()
|
||
|
||
# Start of a vlan block: "vlan <id>"
|
||
m = re.match(r"^vlan\s+(\d+)\s*$", stripped, re.I)
|
||
if m:
|
||
current_id = int(m.group(1))
|
||
vlans[current_id] = {"name": f"VLAN{current_id}", "tagged": [], "untagged": []}
|
||
continue
|
||
|
||
if current_id is None:
|
||
continue
|
||
|
||
# End of vlan block
|
||
if stripped.lower() == "exit":
|
||
current_id = None
|
||
continue
|
||
|
||
# name "VLAN-Name" (with or without quotes)
|
||
nm = re.match(r'^name\s+"?([^"]+)"?\s*$', stripped, re.I)
|
||
if nm:
|
||
vlans[current_id]["name"] = nm.group(1).strip()
|
||
continue
|
||
|
||
# untagged <portlist>
|
||
utm = re.match(r"^untagged\s+(.+)$", stripped, re.I)
|
||
if utm:
|
||
vlans[current_id]["untagged"] = expand_procurve_ports(utm.group(1))
|
||
continue
|
||
|
||
# tagged <portlist>
|
||
tgm = re.match(r"^tagged\s+(.+)$", stripped, re.I)
|
||
if tgm:
|
||
vlans[current_id]["tagged"] = expand_procurve_ports(tgm.group(1))
|
||
continue
|
||
|
||
# Populate "interfaces" as union of tagged + untagged (NAPALM standard key)
|
||
for vdata in vlans.values():
|
||
seen: List[str] = []
|
||
for p in vdata["untagged"] + vdata["tagged"]:
|
||
if p not in seen:
|
||
seen.append(p)
|
||
vdata["interfaces"] = seen
|
||
|
||
return vlans
|
||
|
||
|
||
def parse_vlans_brief(output: str) -> Dict[int, str]:
|
||
"""Parse ``show vlans`` brief table for VLAN IDs and names only.
|
||
|
||
Used as a discovery step when running-config is unavailable.
|
||
|
||
Example output::
|
||
|
||
VLAN ID Name | Status Voice Jumbo
|
||
------- ---------------- + ---------- ----- -----
|
||
1 DEFAULT_VLAN | Port-based No No
|
||
10 MANAGEMENT | Port-based No No
|
||
|
||
Returns ``{vlan_id: name}`` dict.
|
||
"""
|
||
vlans: Dict[int, str] = {}
|
||
in_table = False
|
||
|
||
for line in output.splitlines():
|
||
if re.match(r"\s*-{4,}", line):
|
||
in_table = True
|
||
continue
|
||
if not in_table:
|
||
continue
|
||
m = re.match(r"\s*(\d+)\s+(\S+)", line)
|
||
if m:
|
||
vlans[int(m.group(1))] = m.group(2)
|
||
|
||
return vlans
|
||
|
||
|
||
def parse_vlan_ports_detail(output: str) -> Dict[str, List[str]]:
|
||
"""Parse ``show vlans <id>`` per-VLAN port detail output.
|
||
|
||
Returns ``{"tagged": [...], "untagged": [...]}`` for ONE vlan.
|
||
|
||
Example output::
|
||
|
||
Port Information Mode Unknown VLAN Status
|
||
---------------- -------- ------------ -------
|
||
1 Untagged Learn Up
|
||
2 Tagged Learn Up
|
||
Trk1 Tagged Learn Up
|
||
"""
|
||
tagged: List[str] = []
|
||
untagged: List[str] = []
|
||
in_table = False
|
||
|
||
for line in output.splitlines():
|
||
if re.match(r"\s*-{4,}", line):
|
||
in_table = True
|
||
continue
|
||
if not in_table:
|
||
continue
|
||
# Columns: Port | Mode | Unknown VLAN | Status
|
||
m = re.match(r"\s*(\S+)\s+(Tagged|Untagged)\s+", line, re.I)
|
||
if m:
|
||
port, mode = m.group(1), m.group(2).lower()
|
||
if mode == "tagged":
|
||
tagged.append(port)
|
||
else:
|
||
untagged.append(port)
|
||
|
||
return {"tagged": tagged, "untagged": untagged}
|
||
|
||
|
||
def derive_port_pvids(vlans: Dict[int, Dict]) -> Dict[str, int]:
|
||
"""Derive PVID (native/access VLAN) for each port from the ``get_vlans`` result.
|
||
|
||
On ProCurve a port is untagged in exactly ONE VLAN — that is its PVID.
|
||
A port that only appears in ``tagged`` lists has no PVID (trunk-only, PVID = 1 by default).
|
||
|
||
Returns ``{"port": pvid_int, ...}`` mapping.
|
||
|
||
Example::
|
||
|
||
# Port 1 is untagged in VLAN 1 → PVID 1
|
||
# Port 5 only tagged in VLAN 10 → PVID 1 (default)
|
||
derive_port_pvids({
|
||
1: {"untagged": ["1","2","3"], "tagged": []},
|
||
10: {"untagged": [], "tagged": ["5"]},
|
||
})
|
||
# → {"1": 1, "2": 1, "3": 1, "5": 1}
|
||
"""
|
||
pvids: Dict[str, int] = {}
|
||
|
||
# First pass: assign PVID from untagged membership
|
||
for vid, vdata in sorted(vlans.items()):
|
||
for port in vdata.get("untagged", []):
|
||
pvids[port] = vid
|
||
|
||
# Second pass: ports that only appear in tagged lists get PVID 1 (default)
|
||
for vdata in vlans.values():
|
||
for port in vdata.get("tagged", []):
|
||
if port not in pvids:
|
||
pvids[port] = 1
|
||
|
||
return pvids
|