# -*- 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" OUI_PREFIXES = [ "3C:D9:2B", # Hewlett Packard — IEEE "3C:52:82", # Hewlett Packard — IEEE "44:31:92", # Hewlett Packard — IEEE "00:17:A4", # Hewlett Packard — IEEE ] 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": result = self._api.get_config() if result.get("running"): return result # REST endpoint not available on this firmware (e.g. HP 2530 / YA series). # Fall back to a temporary SSH connection to run show running-config. running = self._get_config_via_ssh() return {"running": running, "startup": "", "candidate": ""} 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} def _get_config_via_ssh(self) -> str: """Open a one-shot SSH session to retrieve running-config. Used as fallback when the REST API does not expose the config endpoint (e.g. HP 2530 / YA firmware with Mocana SSH). Tries legacy KEX algorithms first (required for older ProCurve firmware), then standard. Returns empty string on any failure. """ for legacy in (True, False): try: conn = ConnectHandler(**self._netmiko_kwargs(legacy=legacy)) try: return conn.send_command("show running-config") finally: conn.disconnect() except Exception as exc: logger.debug( "SSH config fallback (%s) failed for %s: %s", "legacy" if legacy else "standard", self.hostname, exc, ) logger.warning("SSH config fallback failed for %s (all transports)", self.hostname) return "" # ------------------------------------------------------------------ # 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 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 repetitions " 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//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, } }