Christian ManivongandClaude Sonnet 4.6
247964178a
Ergänzt DRIVER_NAME, HTTP_FINGERPRINT, SNMP_FINGERPRINT, SSH_FINGERPRINT, PORT_SPECS und SNMP_OBJECT_ID_PREFIX gemäß docs/DISCOVERY_FINGERPRINTING.md. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1361 lines
51 KiB
Python
1361 lines
51 KiB
Python
# -*- coding: utf-8 -*-
|
||
# Licensed under the Apache License, Version 2.0
|
||
|
||
"""NAPALM driver for HPE/Aruba ProCurve switches.
|
||
|
||
Supports legacy ProCurve (2520G-8-PoE etc.) through to modern Aruba 2530/2540
|
||
by auto-detecting the best available transport:
|
||
|
||
1. REST API (HTTPS then HTTP, API v7 → v6 → v3) — newer Aruba 2530/2540
|
||
2. SSH — standard netmiko hp_procurve
|
||
3. SSH legacy KEX — forces older algorithms for 2520G etc.
|
||
4. Telnet — oldest devices
|
||
|
||
A specific transport can be forced with ``optional_args={"transport": "ssh"}``.
|
||
|
||
Netmiko device type: ``hp_procurve`` / ``hp_procurve_telnet``
|
||
"""
|
||
|
||
import logging
|
||
import re
|
||
import socket
|
||
from typing import Any, Dict, List, Optional, Union
|
||
|
||
import netaddr
|
||
from netmiko import ConnectHandler
|
||
from netmiko.exceptions import (
|
||
NetmikoAuthenticationException,
|
||
NetmikoTimeoutException,
|
||
)
|
||
from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SwitchDriver
|
||
from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict
|
||
from napalm.base import helpers as napalm_helpers
|
||
from napalm.base.exceptions import (
|
||
CommandErrorException,
|
||
ConnectionException,
|
||
ConnectionClosedException,
|
||
MergeConfigException,
|
||
ReplaceConfigException,
|
||
)
|
||
from napalm.base.netmiko_helpers import netmiko_args
|
||
|
||
from napalm_procurve.api_client import ProcurveApiClient
|
||
from napalm_procurve import parsers
|
||
|
||
logger = logging.getLogger("napalm_procurve")
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# KEX / cipher profiles for legacy SSH
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Standard: netmiko's HPProcurveBase already disables rsa-sha2-* pubkeys which
|
||
# is needed for any ProCurve running SSH (they use plain ssh-rsa host keys).
|
||
_SSH_DISABLED_STANDARD = {
|
||
"pubkeys": ["rsa-sha2-256", "rsa-sha2-512"],
|
||
}
|
||
|
||
# Legacy: additionally remove modern kex algorithms so the negotiation is
|
||
# forced down to diffie-hellman-group14-sha1 / diffie-hellman-group1-sha1,
|
||
# which very old switches (2520G etc.) may require.
|
||
_SSH_DISABLED_LEGACY = {
|
||
"pubkeys": ["rsa-sha2-256", "rsa-sha2-512"],
|
||
"kex": [
|
||
"curve25519-sha256@libssh.org",
|
||
"ecdh-sha2-nistp256",
|
||
"ecdh-sha2-nistp384",
|
||
"ecdh-sha2-nistp521",
|
||
"diffie-hellman-group16-sha512",
|
||
"diffie-hellman-group-exchange-sha256",
|
||
"diffie-hellman-group14-sha256",
|
||
],
|
||
}
|
||
|
||
|
||
class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
|
||
"""NAPALM driver for HPE/Aruba ProCurve switches.
|
||
|
||
Auto-detects transport on ``open()``. See module docstring for details.
|
||
"""
|
||
|
||
VENDOR = "HPE"
|
||
DRIVER_NAME = "procurve"
|
||
SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.11"
|
||
SSH_FINGERPRINT = [
|
||
FingerprintRule("hp", weight=5.0),
|
||
FingerprintRule("aruba", weight=4.0),
|
||
]
|
||
HTTP_FINGERPRINT = [
|
||
FingerprintRule("hp procurve", weight=8.0, mandatory=True),
|
||
FingerprintRule("procurve", weight=6.0),
|
||
FingerprintRule("ehttp", weight=5.0),
|
||
FingerprintRule("comware", weight=5.0),
|
||
]
|
||
NETMIKO_DEVICE_TYPE = "hp_procurve"
|
||
NETMIKO_DEVICE_TYPE_TELNET = "hp_procurve_telnet"
|
||
|
||
def __init__(
|
||
self,
|
||
hostname: str,
|
||
username: str,
|
||
password: str,
|
||
timeout: int = 60,
|
||
optional_args: Optional[Dict] = None,
|
||
) -> None:
|
||
self.hostname = hostname
|
||
self.username = username
|
||
self.password = password
|
||
self.timeout = timeout
|
||
|
||
if optional_args is None:
|
||
optional_args = {}
|
||
|
||
# --- optional_args ---
|
||
self.port: int = optional_args.get("port", 22)
|
||
self.telnet_port: int = optional_args.get("telnet_port", 23)
|
||
self.force_transport: Optional[str] = optional_args.get("transport")
|
||
self.ssl_verify: bool = optional_args.get(
|
||
"ssl_verify", optional_args.get("verify_ssl", optional_args.get("verify", False))
|
||
)
|
||
self.api_version_hint: Optional[str] = optional_args.get("api_version")
|
||
# Enable password for CLI (defaults to login password)
|
||
self._secret: str = optional_args.get("secret", password)
|
||
|
||
if optional_args.get("debugging"):
|
||
logger.setLevel(logging.DEBUG)
|
||
|
||
self.netmiko_optional_args = netmiko_args(optional_args)
|
||
|
||
# --- runtime state ---
|
||
# Active transport: "api" | "ssh" | "ssh_legacy" | "telnet"
|
||
self._transport: Optional[str] = None
|
||
# CLI backend (SSH or Telnet)
|
||
self._device: Optional[ConnectHandler] = None
|
||
# REST API backend
|
||
self._api: Optional[ProcurveApiClient] = None
|
||
|
||
# Config management state (CLI only)
|
||
self._candidate_config: Optional[str] = None
|
||
self._candidate_mode: Optional[str] = None # "merge" | "replace"
|
||
self._backup_config: Optional[str] = None
|
||
|
||
# ------------------------------------------------------------------
|
||
# Connection management
|
||
# ------------------------------------------------------------------
|
||
|
||
def open(self) -> None:
|
||
"""Open connection, auto-detecting the best available transport.
|
||
|
||
Detection order (unless overridden by ``optional_args["transport"]``):
|
||
|
||
1. REST API
|
||
2. SSH (standard)
|
||
3. SSH (legacy KEX — old ProCurve like 2520G-8-PoE)
|
||
4. Telnet
|
||
"""
|
||
forced = self.force_transport
|
||
tried: List[str] = []
|
||
|
||
# --- 1. REST API ---
|
||
if not forced or forced == "api":
|
||
if self._try_api():
|
||
return
|
||
tried.append("api")
|
||
|
||
# --- 2. SSH standard ---
|
||
if not forced or forced == "ssh":
|
||
if self._try_ssh(legacy=False):
|
||
return
|
||
tried.append("ssh")
|
||
|
||
# --- 3. SSH legacy KEX ---
|
||
if not forced or forced == "ssh_legacy":
|
||
if self._try_ssh(legacy=True):
|
||
return
|
||
tried.append("ssh_legacy")
|
||
|
||
# --- 4. Telnet ---
|
||
if not forced or forced == "telnet":
|
||
if self._try_telnet():
|
||
return
|
||
tried.append("telnet")
|
||
|
||
raise ConnectionException(
|
||
f"Cannot connect to {self.hostname}. "
|
||
f"Tried transports: {', '.join(tried)}. "
|
||
"Check connectivity, credentials and whether SSH/Telnet/API is enabled."
|
||
)
|
||
|
||
def close(self) -> None:
|
||
"""Close the active connection."""
|
||
if self._api:
|
||
self._api.disconnect()
|
||
self._api = None
|
||
if self._device:
|
||
try:
|
||
self._device.disconnect()
|
||
except Exception:
|
||
pass
|
||
self._device = None
|
||
self._transport = None
|
||
|
||
def is_alive(self) -> Dict[str, bool]:
|
||
"""Return transport-level liveness without writing to the channel."""
|
||
if self._api:
|
||
return {"is_alive": self._api.is_alive()}
|
||
if self._device:
|
||
try:
|
||
return {"is_alive": self._device.remote_conn.transport.is_active()}
|
||
except (AttributeError, socket.error, EOFError):
|
||
return {"is_alive": False}
|
||
return {"is_alive": False}
|
||
|
||
# ------------------------------------------------------------------
|
||
# Transport probes
|
||
# ------------------------------------------------------------------
|
||
|
||
def _try_api(self) -> bool:
|
||
"""Probe and connect via REST API. Returns True on success."""
|
||
probe_timeout = min(self.timeout, 15)
|
||
logger.info("Probing REST API for %s (timeout=%ds)", self.hostname, probe_timeout)
|
||
# Probe without SSL verification — no credentials are sent during probing,
|
||
# so this is safe and avoids failing on self-signed certificates.
|
||
ver, proto = ProcurveApiClient.probe(
|
||
self.hostname, timeout=probe_timeout, ssl_verify=False
|
||
)
|
||
# Allow hint override (e.g. user knows the API version)
|
||
if self.api_version_hint and proto:
|
||
ver = self.api_version_hint
|
||
|
||
if not ver:
|
||
logger.warning("REST API not detected on %s — falling back to CLI", self.hostname)
|
||
return False
|
||
|
||
# Try connecting with the requested SSL setting first; if it fails due to a
|
||
# self-signed certificate (ssl_verify=True), transparently retry unverified.
|
||
for ssl_verify in ([self.ssl_verify] if not self.ssl_verify else [True, False]):
|
||
client = ProcurveApiClient(
|
||
hostname=self.hostname,
|
||
username=self.username,
|
||
password=self.password,
|
||
timeout=self.timeout,
|
||
ssl_verify=ssl_verify,
|
||
api_version=ver,
|
||
)
|
||
client.setup(ver, proto)
|
||
try:
|
||
client.connect()
|
||
except Exception as exc:
|
||
logger.debug("REST API connect failed (ssl_verify=%s): %s", ssl_verify, exc)
|
||
continue
|
||
self._api = client
|
||
self._transport = "api"
|
||
logger.info("Connected to %s via REST API (%s %s, ssl_verify=%s)",
|
||
self.hostname, proto, ver, ssl_verify)
|
||
return True
|
||
|
||
logger.warning("REST API connect failed for %s — falling back to CLI", self.hostname)
|
||
return False
|
||
|
||
def _netmiko_kwargs(self, legacy: bool = False) -> dict:
|
||
"""Return kwargs for ConnectHandler, merging explicit fields with netmiko_optional_args.
|
||
|
||
Explicit fields (port, timeout, disabled_algorithms) take precedence and are
|
||
excluded from the spread to avoid 'multiple values for keyword argument' errors.
|
||
"""
|
||
extra = {k: v for k, v in self.netmiko_optional_args.items()
|
||
if k not in ("port", "timeout", "disabled_algorithms")}
|
||
return dict(
|
||
device_type=self.NETMIKO_DEVICE_TYPE,
|
||
host=self.hostname,
|
||
username=self.username,
|
||
password=self.password,
|
||
secret=self._secret,
|
||
port=self.port,
|
||
timeout=self.timeout,
|
||
disabled_algorithms=_SSH_DISABLED_LEGACY if legacy else _SSH_DISABLED_STANDARD,
|
||
**extra,
|
||
)
|
||
|
||
def _try_ssh(self, legacy: bool = False) -> bool:
|
||
"""Probe and connect via SSH. Returns True on success."""
|
||
label = "SSH-legacy" if legacy else "SSH"
|
||
logger.debug("Trying %s for %s", label, self.hostname)
|
||
try:
|
||
conn = ConnectHandler(**self._netmiko_kwargs(legacy))
|
||
self._device = conn
|
||
self._transport = "ssh_legacy" if legacy else "ssh"
|
||
logger.info("Connected to %s via %s", self.hostname, label)
|
||
return True
|
||
except NetmikoAuthenticationException as exc:
|
||
raise ConnectionException(
|
||
f"Authentication failed for {self.hostname}: {exc}"
|
||
) from exc
|
||
except NetmikoTimeoutException:
|
||
logger.debug("%s timeout for %s", label, self.hostname)
|
||
return False
|
||
except Exception as exc:
|
||
logger.debug("%s failed for %s: %s", label, self.hostname, exc)
|
||
return False
|
||
|
||
def _try_telnet(self) -> bool:
|
||
"""Probe and connect via Telnet. Returns True on success."""
|
||
logger.debug("Trying Telnet for %s", self.hostname)
|
||
try:
|
||
conn = ConnectHandler(
|
||
device_type=self.NETMIKO_DEVICE_TYPE_TELNET,
|
||
host=self.hostname,
|
||
username=self.username,
|
||
password=self.password,
|
||
secret=self._secret,
|
||
port=self.telnet_port,
|
||
timeout=self.timeout,
|
||
)
|
||
self._device = conn
|
||
self._transport = "telnet"
|
||
logger.info("Connected to %s via Telnet", self.hostname)
|
||
return True
|
||
except NetmikoAuthenticationException as exc:
|
||
raise ConnectionException(
|
||
f"Authentication failed for {self.hostname}: {exc}"
|
||
) from exc
|
||
except Exception as exc:
|
||
logger.debug("Telnet failed for %s: %s", self.hostname, exc)
|
||
return False
|
||
|
||
# ------------------------------------------------------------------
|
||
# CLI helpers
|
||
# ------------------------------------------------------------------
|
||
|
||
def _send_command(self, command: Union[str, List[str]]) -> str:
|
||
"""Send a CLI command and return stripped output.
|
||
|
||
When a list is supplied, commands are tried in order and the first
|
||
non-error result is returned.
|
||
"""
|
||
if self._device is None:
|
||
raise ConnectionClosedException("Device connection is not open.")
|
||
|
||
prompt_pattern = rf"{re.escape(self._device.base_prompt)}[>#]"
|
||
|
||
def _do(cmd: str) -> str:
|
||
return self._device.send_command(
|
||
cmd,
|
||
expect_string=prompt_pattern,
|
||
read_timeout=self.timeout,
|
||
).strip()
|
||
|
||
try:
|
||
if isinstance(command, list):
|
||
last = ""
|
||
for cmd in command:
|
||
last = _do(cmd)
|
||
if "% Invalid" not in last and "Error" not in last:
|
||
return last
|
||
return last
|
||
return _do(command)
|
||
except (socket.error, EOFError) as exc:
|
||
raise ConnectionClosedException(str(exc)) from exc
|
||
|
||
def _exec_prompt(self) -> str:
|
||
return rf"{re.escape(self._device.base_prompt)}[>#]"
|
||
|
||
def _conf_prompt(self) -> str:
|
||
return rf"{re.escape(self._device.base_prompt)}\(config[^)]*\)[>#]"
|
||
|
||
def _any_prompt(self) -> str:
|
||
return rf"{re.escape(self._device.base_prompt)}(?:\([^)]*\))?[>#]"
|
||
|
||
def _enter_config_mode(self) -> None:
|
||
self._device.send_command(
|
||
"configure",
|
||
expect_string=self._conf_prompt(),
|
||
read_timeout=self.timeout,
|
||
)
|
||
|
||
def _exit_config_mode(self) -> None:
|
||
self._device.send_command(
|
||
"end",
|
||
expect_string=self._exec_prompt(),
|
||
read_timeout=self.timeout,
|
||
)
|
||
|
||
def _save_config(self) -> None:
|
||
"""Save running configuration to startup (``write memory``)."""
|
||
if self._transport == "api":
|
||
self._api.post("cli", json={"cmd": "write memory"})
|
||
return
|
||
self._device.send_command(
|
||
"write memory",
|
||
expect_string=self._exec_prompt(),
|
||
read_timeout=self.timeout,
|
||
)
|
||
|
||
def _apply_config_lines(self, config_text: str) -> List[str]:
|
||
"""Push config lines to the device while in config mode.
|
||
|
||
Returns a list of error messages for rejected lines.
|
||
"""
|
||
ep = self._any_prompt()
|
||
errors: List[str] = []
|
||
for line in config_text.splitlines():
|
||
stripped = line.strip()
|
||
if not stripped or stripped.startswith(("!", "#", ";")):
|
||
continue
|
||
out = self._device.send_command(
|
||
stripped,
|
||
expect_string=ep,
|
||
read_timeout=self.timeout,
|
||
).strip()
|
||
if out and re.search(r"(error|invalid|unknown|unrecog)", out, re.I):
|
||
errors.append(f" {stripped!r}: {out}")
|
||
return errors
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_facts
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_facts(self) -> Dict:
|
||
"""Return general device information."""
|
||
if self._transport == "api":
|
||
return self._api.get_facts()
|
||
|
||
# CLI path
|
||
sysinfo = self._send_command(
|
||
["show system-information", "show system information", "show system"]
|
||
)
|
||
parsed = parsers.parse_system_info(sysinfo)
|
||
|
||
# get model (and part_number) from show version if not in system-info
|
||
if not parsed["model"]:
|
||
ver_out = self._send_command("show version")
|
||
parsed["model"], ver_pn = parsers.parse_model_from_version(ver_out)
|
||
if ver_pn and not parsed.get("part_number"):
|
||
parsed["part_number"] = ver_pn
|
||
if not parsed["os_version"]:
|
||
parsed["os_version"] = parsers.parse_version(ver_out)
|
||
|
||
iface_list = self._get_interface_list()
|
||
|
||
return {
|
||
"vendor": self.VENDOR,
|
||
"model": parsed["model"],
|
||
"part_number": parsed.get("part_number", ""),
|
||
"hostname": parsed["hostname"],
|
||
"fqdn": parsed["hostname"],
|
||
"os_version": parsed["os_version"],
|
||
"serial_number": parsed["serial_number"],
|
||
"uptime": parsed["uptime_seconds"],
|
||
"interface_list": iface_list,
|
||
}
|
||
|
||
def _get_interface_list(self) -> List[str]:
|
||
"""Return sorted list of interface names from ``show interfaces brief``."""
|
||
out = self._send_command("show interfaces brief")
|
||
ifaces = parsers.parse_interfaces_brief(out)
|
||
return sorted(
|
||
ifaces.keys(),
|
||
key=lambda s: [
|
||
int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)
|
||
],
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_interfaces
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_interfaces(self) -> Dict[str, Dict]:
|
||
"""Return a dictionary of interface details."""
|
||
if self._transport == "api":
|
||
return self._api.get_interfaces()
|
||
|
||
out = self._send_command("show interfaces brief")
|
||
ifaces = parsers.parse_interfaces_brief(out)
|
||
|
||
# Enrich with MAC address from per-port detail (best-effort)
|
||
for port in list(ifaces.keys()):
|
||
try:
|
||
detail_out = self._send_command(f"show interfaces {port}")
|
||
detail = parsers.parse_interface_detail(detail_out, port)
|
||
ifaces[port]["mac_address"] = detail.get("mac_address", "")
|
||
if detail.get("description"):
|
||
ifaces[port]["description"] = detail["description"]
|
||
except Exception:
|
||
pass
|
||
|
||
return ifaces
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_interfaces_ip
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_interfaces_ip(self) -> Dict[str, Dict]:
|
||
"""Return IP addresses assigned to interfaces."""
|
||
if self._transport == "api":
|
||
return self._api.get_interfaces_ip()
|
||
|
||
out = self._send_command("show ip")
|
||
return parsers.parse_ip_interfaces(out)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_arp_table
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_arp_table(self, vrf: str = "") -> List[Dict]:
|
||
"""Return the ARP table."""
|
||
if self._transport == "api":
|
||
return self._api.get_arp_table(vrf)
|
||
|
||
out = self._send_command("show arp")
|
||
return parsers.parse_arp_table(out)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_mac_address_table
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_mac_address_table(self) -> List[Dict]:
|
||
"""Return the MAC address table."""
|
||
if self._transport == "api":
|
||
return self._api.get_mac_address_table()
|
||
|
||
out = self._send_command(["show mac-address", "show mac address"])
|
||
return parsers.parse_mac_table(out)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_lldp_neighbors
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_lldp_neighbors(self) -> Dict[str, List[Dict]]:
|
||
"""Return LLDP neighbor information."""
|
||
if self._transport == "api":
|
||
return self._api.get_lldp_neighbors()
|
||
|
||
out = self._send_command(
|
||
["show lldp info remote-device", "show lldp neighbors"]
|
||
)
|
||
return parsers.parse_lldp_neighbors(out)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_lldp_neighbors_detail
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_lldp_neighbors_detail(self, interface: str = "") -> Dict[str, List[Dict]]:
|
||
"""Return detailed LLDP neighbor information."""
|
||
if self._transport == "api":
|
||
return self._api.get_lldp_neighbors_detail()
|
||
|
||
if interface:
|
||
cmd = f"show lldp info remote-device {interface}"
|
||
else:
|
||
cmd = "show lldp info remote-device detail"
|
||
out = self._send_command(cmd)
|
||
return parsers.parse_lldp_neighbors_detail(out)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_config
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_config(
|
||
self, retrieve: str = "all", full: bool = False, sanitized: bool = False
|
||
) -> Dict[str, str]:
|
||
"""Return device configuration."""
|
||
if self._transport == "api":
|
||
return self._api.get_config()
|
||
|
||
running = ""
|
||
startup = ""
|
||
candidate = self._candidate_config or ""
|
||
|
||
if retrieve in ("all", "running"):
|
||
running = self._send_command("show running-config")
|
||
if retrieve in ("all", "startup"):
|
||
startup = self._send_command(
|
||
["show startup-config", "show config"]
|
||
)
|
||
|
||
return {"running": running, "startup": startup, "candidate": candidate}
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_environment
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_environment(self) -> Dict:
|
||
"""Return environment information (temperature, fans, CPU, memory)."""
|
||
if self._transport == "api":
|
||
raise NotImplementedError("Environment data not available via REST API")
|
||
|
||
sysinfo = self._send_command(
|
||
["show system-information", "show system information"]
|
||
)
|
||
env_out = self._send_command(["show environment", "show system"])
|
||
combined = sysinfo + "\n" + env_out
|
||
return parsers.parse_environment(combined)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_ntp_servers
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_ntp_servers(self) -> Dict[str, Dict]:
|
||
"""Return configured NTP servers."""
|
||
if self._transport == "api":
|
||
return self._api.get_ntp_servers()
|
||
|
||
out = self._send_command(
|
||
["show ntp associations", "show ntp status", "show timesync"]
|
||
)
|
||
return parsers.parse_ntp_servers(out)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_ntp_stats
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_ntp_stats(self) -> List[Dict]:
|
||
"""Return NTP peer statistics."""
|
||
out = self._send_command(
|
||
["show ntp associations", "show ntp status"]
|
||
)
|
||
stats = []
|
||
for line in out.splitlines():
|
||
m = re.match(
|
||
r"\s*([*+]?)\s*(\d{1,3}(?:\.\d{1,3}){3})"
|
||
r"\s+(\S+)\s+(\d+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)",
|
||
line,
|
||
)
|
||
if not m:
|
||
continue
|
||
synced = m.group(1) == "*"
|
||
stats.append(
|
||
{
|
||
"remote": m.group(2),
|
||
"referenceid": m.group(3),
|
||
"synchronized": synced,
|
||
"stratum": int(m.group(4)),
|
||
"type": "-",
|
||
"when": m.group(5),
|
||
"hostpoll": int(m.group(6)) if m.group(6).isdigit() else -1,
|
||
"reachability": int(m.group(7), 8) if m.group(7).isdigit() else -1,
|
||
"delay": float(m.group(8)) if _is_float(m.group(8)) else 0.0,
|
||
"offset": float(m.group(9)) if _is_float(m.group(9)) else 0.0,
|
||
"jitter": 0.0,
|
||
}
|
||
)
|
||
return stats
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_users
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_users(self) -> Dict[str, Dict]:
|
||
"""Return local users."""
|
||
out = self._send_command(["show local-users", "show users"])
|
||
return parsers.parse_users(out)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_snmp_information
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_snmp_information(self) -> Dict:
|
||
"""Return SNMP configuration."""
|
||
out = self._send_command(["show snmp-server", "show snmp"])
|
||
return parsers.parse_snmp_info(out)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: get_vlans (extended: tagged + untagged per VLAN)
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_vlans(self) -> Dict[int, Dict]:
|
||
"""Return VLAN information with tagged/untagged port separation.
|
||
|
||
This method extends the standard NAPALM ``get_vlans()`` return with
|
||
two extra keys per VLAN:
|
||
|
||
- ``tagged`` — ports where this VLAN is carried tagged (802.1Q trunk)
|
||
- ``untagged`` — ports where this VLAN is the native/access VLAN (PVID)
|
||
|
||
The standard ``interfaces`` key remains the union of both lists,
|
||
keeping backward compatibility with NAPALM tooling.
|
||
|
||
Example return::
|
||
|
||
{
|
||
1: {
|
||
"name": "DEFAULT_VLAN",
|
||
"interfaces": ["1", "2", "3", "4", "5"],
|
||
"tagged": ["5"],
|
||
"untagged": ["1", "2", "3", "4"],
|
||
},
|
||
10: {
|
||
"name": "MGMT",
|
||
"interfaces": ["1", "2", "5"],
|
||
"tagged": ["1", "2", "5"],
|
||
"untagged": [],
|
||
},
|
||
}
|
||
|
||
Use :meth:`get_port_pvids` to get the per-port PVID mapping derived
|
||
from the ``untagged`` lists.
|
||
"""
|
||
if self._transport == "api":
|
||
return self._api.get_vlans()
|
||
|
||
# --- CLI path ---
|
||
# Primary: parse VLAN blocks from running-config (single command,
|
||
# reliable across all ProCurve firmware, contains untagged/tagged lines)
|
||
running = self._send_command("show running-config")
|
||
vlans = parsers.parse_vlans_from_running_config(running)
|
||
|
||
if vlans:
|
||
return vlans
|
||
|
||
# Fallback: show vlans (name discovery) + show vlans <id> per VLAN
|
||
return self._get_vlans_via_show_vlans()
|
||
|
||
def _get_vlans_via_show_vlans(self) -> Dict[int, Dict]:
|
||
"""Fallback VLAN discovery via ``show vlans`` + per-VLAN detail queries."""
|
||
brief_out = self._send_command("show vlans")
|
||
vlan_names = parsers.parse_vlans_brief(brief_out)
|
||
|
||
result: Dict[int, Dict] = {}
|
||
for vid, name in vlan_names.items():
|
||
detail_out = self._send_command(f"show vlans {vid}")
|
||
ports = parsers.parse_vlan_ports_detail(detail_out)
|
||
tagged = ports["tagged"]
|
||
untagged = ports["untagged"]
|
||
interfaces: List[str] = []
|
||
for p in untagged + tagged:
|
||
if p not in interfaces:
|
||
interfaces.append(p)
|
||
result[vid] = {
|
||
"name": name,
|
||
"interfaces": interfaces,
|
||
"tagged": tagged,
|
||
"untagged": untagged,
|
||
}
|
||
|
||
return result
|
||
|
||
# ------------------------------------------------------------------
|
||
# Custom: get_port_pvids (PVID per port, derived from get_vlans)
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_port_pvids(self) -> Dict[str, int]:
|
||
"""Return the PVID (native / access VLAN) for every switch port.
|
||
|
||
On ProCurve, a port is untagged in exactly **one** VLAN — that VLAN's
|
||
ID is the port's PVID. Ports that only appear in tagged lists have
|
||
PVID 1 (the default).
|
||
|
||
This method calls :meth:`get_vlans` internally and derives PVIDs from
|
||
the ``untagged`` lists.
|
||
|
||
Example return::
|
||
|
||
{
|
||
"1": 1, # port 1 is untagged in VLAN 1 → PVID 1
|
||
"2": 1,
|
||
"3": 10, # port 3 is untagged in VLAN 10 → PVID 10
|
||
"Trk1": 1, # Trk1 only tagged → PVID 1 (default)
|
||
}
|
||
"""
|
||
vlans = self.get_vlans()
|
||
return parsers.derive_port_pvids(vlans)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: ping
|
||
# ------------------------------------------------------------------
|
||
|
||
# ------------------------------------------------------------------
|
||
# SwitchDriver: write methods
|
||
# ------------------------------------------------------------------
|
||
|
||
def set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
|
||
"""Create or update a VLAN on the switch."""
|
||
if self._transport == "api":
|
||
self._api_set_vlan(vlan_id, config)
|
||
else:
|
||
self._cli_set_vlan(vlan_id, config)
|
||
|
||
def _api_set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
|
||
name = config.get("name", f"VLAN{vlan_id}")
|
||
payload = {"vlan_id": vlan_id, "name": name}
|
||
resp = self._api.post("vlans", json=payload)
|
||
if not resp.ok:
|
||
if resp.status_code in (400, 409):
|
||
# VLAN already exists — PUT requires the full VLAN object
|
||
existing = self._api.get(f"vlans/{vlan_id}")
|
||
existing.pop("uri", None)
|
||
existing["name"] = name
|
||
resp2 = self._api.put(f"vlans/{vlan_id}", json=existing)
|
||
if not resp2.ok:
|
||
raise ConnectionException(
|
||
f"set_vlan({vlan_id}): API PUT returned HTTP {resp2.status_code}"
|
||
)
|
||
else:
|
||
raise ConnectionException(
|
||
f"set_vlan({vlan_id}): API POST returned HTTP {resp.status_code}"
|
||
)
|
||
|
||
def _cli_set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
|
||
self._enter_config_mode()
|
||
try:
|
||
lines = [f"vlan {vlan_id}"]
|
||
if "name" in config:
|
||
# Quote the name to handle spaces
|
||
lines.append(f' name "{config["name"]}"')
|
||
lines.append("exit")
|
||
errors = self._apply_config_lines("\n".join(lines))
|
||
if errors:
|
||
raise CommandErrorException(
|
||
f"set_vlan({vlan_id}) errors: {errors}"
|
||
)
|
||
finally:
|
||
self._exit_config_mode()
|
||
self._save_config()
|
||
|
||
# ------------------------------------------------------------------
|
||
# LAG / trunk membership
|
||
# ------------------------------------------------------------------
|
||
|
||
def set_lag_members(self, lag_name: str, members: List[str]) -> None:
|
||
"""Set the full member-port list of a LAG/trunk group.
|
||
|
||
Diffs ``members`` against the trunk's current members (as reported
|
||
by ``get_interfaces()``) and issues ``trunk``/``no trunk`` commands
|
||
for the difference. Available on both REST API and CLI transports —
|
||
ArubaOS-Switch only exposes trunk membership via the CLI, including
|
||
through the REST API's ``/cli`` passthrough endpoint (``PUT /ports``
|
||
does not apply ``trunk_group``/``trunk_mode`` changes).
|
||
"""
|
||
current = self.get_interfaces().get(lag_name, {})
|
||
current_members = set(current.get("lag_members") or [])
|
||
desired = set(members)
|
||
to_remove = sorted(current_members - desired, key=lambda s: int(s) if s.isdigit() else 0)
|
||
to_add = sorted(desired - current_members, key=lambda s: int(s) if s.isdigit() else 0)
|
||
|
||
if not to_remove and not to_add:
|
||
return
|
||
|
||
mode = current.get("lag_mode") or "trunk"
|
||
trk_name = lag_name.lower()
|
||
|
||
commands: List[str] = []
|
||
if to_remove:
|
||
commands.append(f"no trunk {','.join(to_remove)}")
|
||
if to_add:
|
||
commands.append(f"trunk {','.join(to_add)} {trk_name} {mode}")
|
||
|
||
if self._transport == "api":
|
||
for cmd in ["configure terminal", *commands]:
|
||
resp = self._api.post("cli", json={"cmd": cmd})
|
||
if not resp.ok or resp.json().get("status") == "CCS_FAILURE":
|
||
raise CommandErrorException(
|
||
f"set_lag_members({lag_name}): '{cmd}' -> {resp.text[:200]}"
|
||
)
|
||
else:
|
||
self._enter_config_mode()
|
||
try:
|
||
errors = self._apply_config_lines("\n".join(commands))
|
||
if errors:
|
||
raise CommandErrorException(f"set_lag_members({lag_name}) errors: {errors}")
|
||
finally:
|
||
self._exit_config_mode()
|
||
self._save_config()
|
||
|
||
def set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
|
||
"""Configure a switch interface (mode, VLAN membership)."""
|
||
if self._transport == "api":
|
||
self._api_set_interface(interface, config)
|
||
else:
|
||
self._cli_set_interface(interface, config)
|
||
|
||
def _api_set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
|
||
patch: dict = {"id": interface}
|
||
if "enabled" in config:
|
||
patch["is_port_enabled"] = bool(config["enabled"])
|
||
if "description" in config:
|
||
patch["name"] = config["description"]
|
||
if len(patch) > 1: # more than just the id field
|
||
resp = self._api.put(f"ports/{interface}", json=patch)
|
||
if not resp.ok:
|
||
raise ConnectionException(
|
||
f"set_interface({interface}): ports PUT HTTP {resp.status_code} – {resp.text[:200]}"
|
||
)
|
||
mode = config.get("mode")
|
||
if mode == "trunk":
|
||
for vid in config.get("trunk_vlans", []):
|
||
payload = {
|
||
"vlan_id": vid,
|
||
"port_id": interface,
|
||
"port_mode": "POM_TAGGED_STATIC",
|
||
}
|
||
resp = self._api.post("vlans-ports", json=payload)
|
||
if not resp.ok and resp.status_code != 409:
|
||
raise ConnectionException(
|
||
f"set_interface({interface}): vlans-ports POST HTTP {resp.status_code}"
|
||
)
|
||
elif mode == "access":
|
||
if "access_vlan" in config:
|
||
payload = {
|
||
"vlan_id": config["access_vlan"],
|
||
"port_id": interface,
|
||
"port_mode": "POM_UNTAGGED",
|
||
}
|
||
resp = self._api.post("vlans-ports", json=payload)
|
||
if not resp.ok and resp.status_code != 409:
|
||
raise ConnectionException(
|
||
f"set_interface({interface}): vlans-ports POST HTTP {resp.status_code}"
|
||
)
|
||
|
||
def _cli_set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
|
||
mode = config.get("mode")
|
||
self._enter_config_mode()
|
||
try:
|
||
lines: List[str] = []
|
||
if "enabled" in config or "description" in config:
|
||
lines.append(f"interface {interface}")
|
||
if "enabled" in config:
|
||
lines.append(" enable" if config["enabled"] else " disable")
|
||
if "description" in config:
|
||
desc = config["description"]
|
||
if desc:
|
||
lines.append(f' name "{desc}"')
|
||
else:
|
||
lines.append(" no name")
|
||
lines.append("exit")
|
||
if mode == "trunk":
|
||
for vid in config.get("trunk_vlans", []):
|
||
lines.append(f"vlan {vid}")
|
||
lines.append(f" tagged {interface}")
|
||
lines.append("exit")
|
||
elif mode == "access":
|
||
if "access_vlan" in config:
|
||
lines.append(f"vlan {config['access_vlan']}")
|
||
lines.append(f" untagged {interface}")
|
||
lines.append("exit")
|
||
if lines:
|
||
errors = self._apply_config_lines("\n".join(lines))
|
||
if errors:
|
||
raise CommandErrorException(
|
||
f"set_interface({interface}) errors: {errors}"
|
||
)
|
||
finally:
|
||
self._exit_config_mode()
|
||
self._save_config()
|
||
|
||
def delete_vlan(self, vlan_id: int) -> None:
|
||
"""Remove a VLAN from the switch."""
|
||
if self._transport == "api":
|
||
self._api.delete_vlan(vlan_id)
|
||
else:
|
||
self._cli_delete_vlan(vlan_id)
|
||
|
||
def _cli_delete_vlan(self, vlan_id: int) -> None:
|
||
self._enter_config_mode()
|
||
try:
|
||
errors = self._apply_config_lines(f"no vlan {vlan_id}")
|
||
if errors:
|
||
raise CommandErrorException(
|
||
f"delete_vlan({vlan_id}) errors: {errors}"
|
||
)
|
||
finally:
|
||
self._exit_config_mode()
|
||
self._save_config()
|
||
|
||
# ------------------------------------------------------------------
|
||
# PoE
|
||
# ------------------------------------------------------------------
|
||
|
||
def get_poe_status(self) -> Dict[str, Dict]:
|
||
"""Return PoE configuration for all ports, keyed by interface name.
|
||
|
||
Only available via the REST API transport.
|
||
"""
|
||
if self._transport == "api":
|
||
return self._api.get_poe_ports()
|
||
raise NotImplementedError("PoE status is only available via the REST API")
|
||
|
||
def set_poe(self, interface: str, config: Dict) -> None:
|
||
"""Update PoE configuration for a single interface."""
|
||
if self._transport != "api":
|
||
raise NotImplementedError("PoE configuration is only available via the REST API")
|
||
|
||
payload: dict = {}
|
||
for key in (
|
||
"is_poe_enabled",
|
||
"poe_priority",
|
||
"poe_allocation_method",
|
||
"pre_standard_detect_enabled",
|
||
):
|
||
if key in config:
|
||
payload[key] = config[key]
|
||
if "allocated_power_in_watts" in config:
|
||
method = config.get("poe_allocation_method") or payload.get("poe_allocation_method")
|
||
if method == "PPAM_VALUE":
|
||
payload["allocated_power_in_watts"] = config["allocated_power_in_watts"]
|
||
|
||
if not payload:
|
||
return
|
||
|
||
resp = self._api.set_port_poe(interface, payload)
|
||
if not resp.ok:
|
||
raise ConnectionException(
|
||
f"set_poe({interface}): ports/poe PUT HTTP {resp.status_code} – {resp.text[:200]}"
|
||
)
|
||
|
||
def ping(
|
||
self,
|
||
destination: str,
|
||
source: str = "",
|
||
ttl: int = 255,
|
||
timeout: int = 2,
|
||
size: int = 100,
|
||
count: int = 5,
|
||
vrf: str = "",
|
||
) -> Dict:
|
||
"""Execute a ping from the device."""
|
||
if self._transport == "api":
|
||
return self._api.ping(
|
||
destination, source=source, ttl=ttl,
|
||
timeout=timeout, size=size, count=count, vrf=vrf,
|
||
)
|
||
|
||
# ProCurve CLI ping: "ping <ip> repetitions <n>"
|
||
cmd = f"ping {destination} repetitions {count} timeout {timeout}"
|
||
if source:
|
||
cmd += f" source {source}"
|
||
out = self._send_command(cmd)
|
||
return _parse_ping_output(out, count)
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: cli
|
||
# ------------------------------------------------------------------
|
||
|
||
def cli(self, commands: List[str], encoding: str = "text") -> Dict[str, str]:
|
||
"""Execute raw CLI commands and return output dict."""
|
||
if self._transport == "api":
|
||
# ArubaOS REST API supports CLI via POST /rest/<ver>/cli
|
||
results: Dict[str, str] = {}
|
||
for cmd in commands:
|
||
resp = self._api.post("cli", json={"cmd": cmd})
|
||
if resp.ok:
|
||
results[cmd] = resp.json().get("result_base64_encoded", "")
|
||
else:
|
||
results[cmd] = f"Error: HTTP {resp.status_code}"
|
||
return results
|
||
|
||
return {cmd: self._send_command(cmd) for cmd in commands}
|
||
|
||
# ------------------------------------------------------------------
|
||
# NAPALM: Configuration management (CLI only)
|
||
# ------------------------------------------------------------------
|
||
|
||
def _get_running_config(self) -> str:
|
||
return self._send_command("show running-config")
|
||
|
||
def load_merge_candidate(
|
||
self, filename: Optional[str] = None, config: Optional[str] = None
|
||
) -> None:
|
||
if self._transport == "api":
|
||
raise MergeConfigException(
|
||
"Merge config is not supported via REST API transport. Use SSH."
|
||
)
|
||
super().load_merge_candidate(filename, config)
|
||
|
||
def load_replace_candidate(
|
||
self, filename: Optional[str] = None, config: Optional[str] = None
|
||
) -> None:
|
||
if self._transport == "api":
|
||
raise ReplaceConfigException(
|
||
"Replace config is not supported via REST API transport. Use SSH."
|
||
)
|
||
super().load_replace_candidate(filename, config)
|
||
|
||
def commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None:
|
||
"""Apply the candidate configuration to the device."""
|
||
if self._candidate_config is None:
|
||
raise MergeConfigException("No candidate config loaded.")
|
||
|
||
# Save running config as backup before applying
|
||
self._backup_config = self._send_command("show running-config")
|
||
|
||
try:
|
||
self._enter_config_mode()
|
||
errors = self._apply_config_lines(self._candidate_config)
|
||
self._exit_config_mode()
|
||
except Exception as exc:
|
||
self._exit_config_mode()
|
||
raise MergeConfigException(f"Error applying config: {exc}") from exc
|
||
|
||
if errors:
|
||
raise MergeConfigException(
|
||
"The following lines were rejected:\n" + "\n".join(errors)
|
||
)
|
||
|
||
self._save_config()
|
||
self._candidate_config = None
|
||
self._candidate_mode = None
|
||
|
||
# ── SNMP / Health ──────────────────────────────────────────────────────────
|
||
|
||
def get_device_warnings(self) -> list:
|
||
"""Return device warnings. SNMP detection handled by poll task."""
|
||
return []
|
||
|
||
def get_snmp_config(self):
|
||
"""Return SNMP config if a community is configured on the switch."""
|
||
try:
|
||
from napalm_device_types.models import SNMPConfigDict
|
||
except ImportError:
|
||
return None
|
||
|
||
# Use a single command with short read_timeout to avoid blocking the poll.
|
||
try:
|
||
out = self._device.send_command(
|
||
"show snmp-server",
|
||
expect_string=rf"{re.escape(self._device.base_prompt)}[>#]",
|
||
read_timeout=10,
|
||
).strip() if self._device else ""
|
||
except Exception:
|
||
return None
|
||
if not out:
|
||
return None
|
||
|
||
# Parse communities from output
|
||
community = None
|
||
for line in out.splitlines():
|
||
m = re.match(r"\s+(\S+)\s+(Manager|Operator)\s+(Restricted|Unrestricted)", line, re.I)
|
||
if m:
|
||
comm = m.group(1)
|
||
access = m.group(3).lower()
|
||
if access == "restricted": # read-only — prefer this
|
||
community = comm
|
||
break
|
||
elif community is None:
|
||
community = comm # fallback to first found
|
||
|
||
if not community:
|
||
return None
|
||
|
||
return SNMPConfigDict(running=True, community=community, port=161, version="2c")
|
||
|
||
def run_device_action(self, action: str) -> Dict:
|
||
"""Execute a named action on the switch."""
|
||
if action == "fix_snmp":
|
||
return self._action_fix_snmp()
|
||
raise NotImplementedError(f"Unknown action: {action!r}")
|
||
|
||
def _detect_management_vlan(self) -> Optional[int]:
|
||
"""Find the VLAN that carries this switch's management IP."""
|
||
try:
|
||
out = self._send_command("show ip")
|
||
for line in out.splitlines():
|
||
# " VLAN8 | 172.22.8.x ..." or similar
|
||
m = re.search(r"VLAN\s*(\d+)[^|]*\|\s*" + re.escape(self.hostname), line, re.I)
|
||
if m:
|
||
return int(m.group(1))
|
||
# Alternative: "VLAN 8 172.22.8.x"
|
||
m2 = re.search(r"\bVLAN\s+(\d+)\b.*\b" + re.escape(self.hostname) + r"\b", line, re.I)
|
||
if m2:
|
||
return int(m2.group(1))
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
def _action_fix_snmp(self) -> Dict:
|
||
"""Enable SNMP with community 'public' (read-only) on the ProCurve switch.
|
||
|
||
ProCurve/Aruba syntax:
|
||
snmp-server community "public" manager restricted
|
||
|
||
When connected via REST API, opens a temporary SSH session for the
|
||
config-mode CLI commands (REST API has no writable SNMP endpoint).
|
||
"""
|
||
if self._transport == "api":
|
||
# REST API cannot run config-mode CLI. Temporarily open SSH.
|
||
old_api, old_device, old_transport = self._api, self._device, self._transport
|
||
self._api = None
|
||
self._device = None
|
||
self._transport = None
|
||
|
||
ssh_error: str = ""
|
||
for legacy in (False, True):
|
||
try:
|
||
conn = ConnectHandler(**self._netmiko_kwargs(legacy))
|
||
self._device = conn
|
||
self._transport = "ssh_legacy" if legacy else "ssh"
|
||
break
|
||
except Exception as exc:
|
||
ssh_error = str(exc)
|
||
|
||
if self._device is None:
|
||
self._api, self._device, self._transport = old_api, old_device, old_transport
|
||
return {
|
||
"success": False,
|
||
"output": f"REST transport active; SSH also failed: {ssh_error}",
|
||
}
|
||
try:
|
||
return self._action_fix_snmp_cli()
|
||
finally:
|
||
try:
|
||
self._device.disconnect()
|
||
except Exception:
|
||
pass
|
||
self._api, self._device, self._transport = old_api, old_device, old_transport
|
||
|
||
return self._action_fix_snmp_cli()
|
||
|
||
def _action_fix_snmp_cli(self) -> Dict:
|
||
"""Run the SNMP fix via CLI (SSH / Telnet, self._device must be open)."""
|
||
lines: list = []
|
||
self._enter_config_mode()
|
||
try:
|
||
errors = self._apply_config_lines('snmp-server community "public" manager restricted')
|
||
if errors:
|
||
lines.append(f"[warn] Config errors: {errors}")
|
||
return {"success": False, "output": "\n".join(lines)}
|
||
lines.append("[config] SNMP community 'public' (manager restricted / read-only) configured.")
|
||
finally:
|
||
self._exit_config_mode()
|
||
|
||
self._save_config()
|
||
lines.append("[config] Configuration saved.")
|
||
|
||
out = self._send_command("show snmp-server")
|
||
success = "public" in out
|
||
lines.append("[ok] SNMP is active with community 'public'." if success
|
||
else f"[warn] Verification — community not found: {out[:200]}")
|
||
return {"success": success, "output": "\n".join(lines)}
|
||
|
||
# ------------------------------------------------------------------
|
||
# Health metrics (SNMP)
|
||
# ------------------------------------------------------------------
|
||
|
||
@classmethod
|
||
async def get_health_metrics(cls, snmp_get, snmp_walk) -> dict:
|
||
import asyncio
|
||
from napalm_device_types._ucd_metrics import build_if_metrics, ticks_to_seconds
|
||
|
||
_HP_CPU_STAT = "1.3.6.1.4.1.11.2.14.11.5.1.9.6.1.0"
|
||
_HP_CPU_STAT_OLD = "1.3.6.1.4.1.11.2.3.7.8.6.1.0"
|
||
_HP_MEM_TOTAL = "1.3.6.1.4.1.11.2.14.11.5.1.1.2.1.1.1.5.1"
|
||
_HP_MEM_USED = "1.3.6.1.4.1.11.2.14.11.5.1.1.2.1.1.1.6.1"
|
||
_HP_UPTIME = "1.3.6.1.4.1.11.2.14.11.1.2.4.1.5.1"
|
||
_OID_IF_DESCR = "1.3.6.1.2.1.2.2.1.2"
|
||
_OID_IF_SPEED = "1.3.6.1.2.1.2.2.1.5"
|
||
_OID_IF_IN_OCT = "1.3.6.1.2.1.2.2.1.10"
|
||
_OID_IF_OUT_OCT = "1.3.6.1.2.1.2.2.1.16"
|
||
_OID_IF_IN_ERR = "1.3.6.1.2.1.2.2.1.14"
|
||
_OID_IF_OUT_ERR = "1.3.6.1.2.1.2.2.1.20"
|
||
|
||
(hp_uptime, cpu_raw, cpu_old, mem_t, mem_u,
|
||
descr, speed, in_oct, out_oct, in_err, out_err) = await asyncio.gather(
|
||
snmp_get(_HP_UPTIME),
|
||
snmp_get(_HP_CPU_STAT),
|
||
snmp_get(_HP_CPU_STAT_OLD),
|
||
snmp_get(_HP_MEM_TOTAL),
|
||
snmp_get(_HP_MEM_USED),
|
||
snmp_walk(_OID_IF_DESCR),
|
||
snmp_walk(_OID_IF_SPEED),
|
||
snmp_walk(_OID_IF_IN_OCT),
|
||
snmp_walk(_OID_IF_OUT_OCT),
|
||
snmp_walk(_OID_IF_IN_ERR),
|
||
snmp_walk(_OID_IF_OUT_ERR),
|
||
)
|
||
|
||
metrics: dict = {}
|
||
|
||
secs = ticks_to_seconds(hp_uptime)
|
||
if secs is not None:
|
||
metrics["uptime_seconds"] = secs
|
||
|
||
cpu = cpu_raw or cpu_old
|
||
if cpu is not None:
|
||
try:
|
||
metrics["cpu_percent"] = float(cpu)
|
||
except ValueError:
|
||
pass
|
||
|
||
if mem_t and mem_u:
|
||
try:
|
||
total = int(mem_t)
|
||
used = int(mem_u)
|
||
metrics["memory_total_bytes"] = total
|
||
metrics["memory_used_bytes"] = used
|
||
metrics["memory_percent"] = round(used / total * 100, 1) if total else 0.0
|
||
except ValueError:
|
||
pass
|
||
|
||
build_if_metrics(metrics, descr, speed, in_oct, out_oct, in_err, out_err)
|
||
return metrics
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _is_float(s: str) -> bool:
|
||
try:
|
||
float(s)
|
||
return True
|
||
except (ValueError, TypeError):
|
||
return False
|
||
|
||
|
||
def _parse_ping_output(output: str, count: int) -> Dict:
|
||
"""Parse ProCurve CLI ping output into NAPALM ping result dict.
|
||
|
||
ProCurve ping output::
|
||
|
||
PING 10.0.0.1: 100 data bytes
|
||
!!!!
|
||
----10.0.0.1 PING Statistics----
|
||
5 packets transmitted, 5 packets received, 0% packet loss
|
||
round-trip (ms) min/avg/max = 0/0/1
|
||
"""
|
||
# Packet statistics line
|
||
m = re.search(
|
||
r"(\d+)\s+packets\s+transmitted,\s+(\d+)\s+packets\s+received",
|
||
output, re.I,
|
||
)
|
||
if not m:
|
||
# Check for total failure
|
||
if "unreachable" in output.lower() or "no route" in output.lower():
|
||
return {
|
||
"success": {
|
||
"probes_sent": count,
|
||
"packet_loss": count,
|
||
"rtt_min": 0.0,
|
||
"rtt_max": 0.0,
|
||
"rtt_avg": 0.0,
|
||
"rtt_stddev": 0.0,
|
||
"results": [],
|
||
}
|
||
}
|
||
return {"error": "Could not parse ping output"}
|
||
|
||
sent = int(m.group(1))
|
||
received = int(m.group(2))
|
||
loss = sent - received
|
||
|
||
# RTT line: "round-trip (ms) min/avg/max = 0/0/1"
|
||
rtt_min = rtt_avg = rtt_max = 0.0
|
||
rtt_m = re.search(r"min/avg/max\s*=\s*([\d.]+)/([\d.]+)/([\d.]+)", output, re.I)
|
||
if rtt_m:
|
||
rtt_min = float(rtt_m.group(1))
|
||
rtt_avg = float(rtt_m.group(2))
|
||
rtt_max = float(rtt_m.group(3))
|
||
|
||
results = [{"ip_address": "", "rtt": rtt_avg}] * received
|
||
|
||
return {
|
||
"success": {
|
||
"probes_sent": sent,
|
||
"packet_loss": loss,
|
||
"rtt_min": rtt_min,
|
||
"rtt_max": rtt_max,
|
||
"rtt_avg": rtt_avg,
|
||
"rtt_stddev": 0.0,
|
||
"results": results,
|
||
}
|
||
}
|
||
|