NAPALM/napalm-hpe-aruba-procurve
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(cli): read J.15 port status, and say why each transport failed

Browse Source 
A 2800-series switch on J.15.09 polled over CLI came back with no
interfaces and an empty OS version, while VLANs and ARP parsed fine.

`show interfaces brief` on that firmware has an Intrusion Alert column
between the `|` and Enabled, and puts Mode before MDI rather than after:

      Port   Type      | Alert     Enabled Status Mode       Mode ...
      3-Trk3 100/1000T | No        Yes     Down   1000FDx    MDI  ...

The regex read Alert as Enabled, then failed on Yes where it wanted
Up|Down, so no line matched. It now skips the Alert column where there is
one and takes the speed from either position. Trunk members are listed as
`<port>-Trk<n>`; the port is `<port>` and the suffix becomes its
trunk_group, so the per-port `show interfaces 3` is a command the switch
knows.

The empty OS version was the alternatives list in _send_command. It moved
to the next command on "% Invalid" or "Error", but ProCurve rejects an
unknown command with "Invalid input: system-information" -- so that line
was parsed as system information. "Invalid input" now counts as failure.

Getting there took longer than it should have, because the first symptom
was "Authentication failed: Login failed". That was Telnet's error, the
last transport tried; the REST probe and both SSH attempts had failed
before it and said nothing above debug level. In fact the switch had run
out of CLI sessions and closed SSH straight after the password. open()
now records why each transport failed, and both the auth error and the
final "Cannot connect" carry that list.

Fixtures are that switch's output, with hostname, serial and MAC
replaced.

Closes #2
Closes #3
Closes #4
This commit is contained in:
Christian Manivong
2026-09-25 08:55:11 +02:00
parent e3bbac0656
commit f35b59ec2d
3 changed files with 205 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_procurve/parsers.py
+12 -9
View File
@@ -234,12 +234,13 @@ def parse_model_from_version(output: str) -> tuple[str, str]:
# 1 100/1000T | Yes Down Auto Unknown off 0
_INTF_BRIEF_RE = re.compile(
r"^\s*(\S+)\s+" # Port
r"^\s*([^\s-]+)" # Port
r"(?:-(Trk\d+))?\s+" # Trunk group of a member port ("3-Trk3")
r"(\S+)\s+\|" # Type |
r"\s+(Yes|No)\s+" # Enabled
r"\s+(?:(?:Yes|No)\s+)?" # Intrusion Alert, on firmware that has the column
r"(Yes|No)\s+" # Enabled
r"(Up|Down)\s+" # Link
r"\S+\s+" # MDI (ignored)
r"(\S+)", # Mode (speed/duplex)
r"(\S+)\s+(\S+)", # Mode and MDI, in either order depending on firmware
re.IGNORECASE,
)
@@ -255,13 +256,13 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]:
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)
)
port, trunk_group, itype, enabled, link, *mode_or_mdi = m.groups()
speed = 0.0
duplex = ""
# Mode examples: "1000FDx", "100HDx", "Unknown", "Auto"
sm = re.match(r"(\d+)(FDx|HDx)?", mode, re.I)
# Mode examples: "1000FDx", "100HDx", "Unknown"; MDI is "Auto", "MDI", "MDIX", "NA"
sm = next(
filter(None, (re.match(r"(\d+)(FDx|HDx)?", t, re.I) for t in mode_or_mdi)), None
)
if sm:
speed = float(sm.group(1))
duplex = "full" if (sm.group(2) or "").lower() == "fdx" else "half"
@@ -276,6 +277,8 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]:
"mtu": -1,
"mac_address": "",
}
if trunk_group:
interfaces[port]["trunk_group"] = trunk_group
return interfaces
napalm_procurve/procurve.py
+28 -14
View File
@@ -138,6 +138,8 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
self._device: Optional[ConnectHandler] = None
# REST API backend
self._api: Optional[ProcurveApiClient] = None
# Why each transport failed during the current open(), as "<transport>: <reason>"
self._attempts: List[str] = []
# Config management state (CLI only)
self._candidate_config: Optional[str] = None
@@ -159,38 +161,45 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
4. Telnet
"""
forced = self.force_transport
tried: List[str] = []
self._attempts = []
# --- 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)}. "
f"Tried transports: {'; '.join(self._attempts)}. "
"Check connectivity, credentials and whether SSH/Telnet/API is enabled."
)
def _auth_failure(self, transport: str, exc: Exception) -> ConnectionException:
"""An authentication error that also says why the earlier transports failed.
Without them, a Telnet "Login failed" reads as a wrong password when SSH
was merely refused and Telnet was the only transport left to answer (#2).
"""
msg = f"Authentication failed for {self.hostname} via {transport}: {exc}"
if self._attempts:
msg += f" (earlier: {'; '.join(self._attempts)})"
return ConnectionException(msg)
def close(self) -> None:
"""Close the active connection."""
if self._api:
@@ -234,10 +243,12 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
if not ver:
logger.warning("REST API not detected on %s — falling back to CLI", self.hostname)
self._attempts.append("api: not detected")
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.
error: Optional[Exception] = None
for ssl_verify in ([self.ssl_verify] if not self.ssl_verify else [True, False]):
client = ProcurveApiClient(
hostname=self.hostname,
@@ -252,6 +263,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
client.connect()
except Exception as exc:
logger.debug("REST API connect failed (ssl_verify=%s): %s", ssl_verify, exc)
error = exc
continue
self._api = client
self._transport = "api"
@@ -260,6 +272,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
return True
logger.warning("REST API connect failed for %s — falling back to CLI", self.hostname)
self._attempts.append(f"api: {error}")
return False
def _netmiko_kwargs(self, legacy: bool = False) -> dict:
@@ -285,22 +298,23 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
def _try_ssh(self, legacy: bool = False) -> bool:
"""Probe and connect via SSH. Returns True on success."""
label = "SSH-legacy" if legacy else "SSH"
transport = "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"
self._transport = transport
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
raise self._auth_failure(transport, exc) from exc
except NetmikoTimeoutException:
logger.debug("%s timeout for %s", label, self.hostname)
self._attempts.append(f"{transport}: timed out")
return False
except Exception as exc:
logger.debug("%s failed for %s: %s", label, self.hostname, exc)
self._attempts.append(f"{transport}: {exc}")
return False
def _try_telnet(self) -> bool:
@@ -321,11 +335,10 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
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
raise self._auth_failure("telnet", exc) from exc
except Exception as exc:
logger.debug("Telnet failed for %s: %s", self.hostname, exc)
self._attempts.append(f"telnet: {exc}")
return False
# ------------------------------------------------------------------
@@ -355,7 +368,8 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
last = ""
for cmd in command:
last = _do(cmd)
if "% Invalid" not in last and "Error" not in last:
# ProCurve says "Invalid input: …"; other firmware "% Invalid …"
if "Invalid input" not in last and "% Invalid" not in last and "Error" not in last:
return last
return last
return _do(command)
tests/unit/test_driver.py
+165 -1
View File
@@ -3,7 +3,9 @@
import pytest
from unittest.mock import MagicMock, patch
from napalm_procurve.procurve import ProcurveDriver, _parse_ping_output
from napalm.base.exceptions import ConnectionException
from napalm_procurve.procurve import _SSH_DISABLED_STANDARD, ProcurveDriver, _parse_ping_output
from napalm_procurve import parsers
@@ -43,6 +45,30 @@ Status and Counters - General System Information
Base MAC Addr : aabbcc-ddeeff
"""
# J.15.09 firmware (#3/#4): no model anywhere, uptime in minutes.
SHOW_SYSTEM_J15 = """\
Status and Counters - General System Information
System Name : myswitch
System Contact :
System Location :
MAC Age Time (sec) : 300
Time Zone : 120
Daylight Time Rule : None
Software revision : J.15.09.0028 Base MAC Addr : a1b2c3-d4e5f6
ROM Version : J.14.05 Serial Number : SG12345678
Allow V1 Modules : Yes
Up Time : 5 mins Memory - Total : 58,720,256
CPU Util (%) : 98 Free : 39,550,272
"""
SHOW_SYSTEM_INFORMATION_INVALID = "Invalid input: system-information"
SHOW_VERSION_2520G = """\
HP J9565A 2520G-8-PoE Switch
Software revision : R.11.27
@@ -59,6 +85,25 @@ Status and Counters - Port Status
3 100/1000T | No Down Auto Unknown off 0
"""
# Layout with the Intrusion Alert column; trunk members are "<port>-Trk<n>" (#3).
SHOW_INTERFACES_BRIEF_INTRUSION = """\
Status and Counters - Port Status
| Intrusion MDI Flow Bcast
Port Type | Alert Enabled Status Mode Mode Ctrl Limit
------ --------- + --------- ------- ------ ---------- ---- ---- -----
1 100/1000T | No Yes Up 1000FDx MDI on 0
2 100/1000T | No Yes Down 1000FDx MDI off 0
3-Trk3 100/1000T | No Yes Down 1000FDx MDI off 0
4-Trk3 100/1000T | No Yes Down 1000FDx MDI off 0
5 100/1000T | No Yes Up 1000FDx MDI on 0
6-Trk6 100/1000T | No No Down 1000FDx NA off 0
7-Trk6 100/1000T | No Yes Up 1000FDx MDI off 0
8 100/1000T | No Yes Up 1000FDx MDIX off 0
9 100/1000T | No Yes Down 1000FDx MDIX off 0
10 1000SX | No Yes Up 1000FDx NA off 0
"""
SHOW_INTERFACES_PORT = """\
Status and Counters - Port Counters for port 1
@@ -212,6 +257,13 @@ class TestParseSystemInfo:
assert info["os_version"] == "YA.16.04.0006"
assert info["serial_number"] == "SG87654321"
def test_j15(self):
info = parsers.parse_system_info(SHOW_SYSTEM_J15)
assert info["hostname"] == "myswitch"
assert info["os_version"] == "J.15.09.0028"
assert info["serial_number"] == "SG12345678"
assert info["base_mac"] == "a1:b2:c3:d4:e5:f6"
class TestParseInterfacesBrief:
def test_parses_ports(self):
@@ -234,6 +286,30 @@ class TestParseInterfacesBrief:
def test_port_3_disabled(self):
ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF)
assert ifaces["3"]["is_enabled"] is False
assert "trunk_group" not in ifaces["3"]
class TestParseInterfacesBriefIntrusionAlert:
"""The layout with an Intrusion Alert column before Enabled (#3)."""
def test_parses_every_port(self):
ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF_INTRUSION)
assert sorted(ifaces, key=int) == [str(n) for n in range(1, 11)]
def test_alert_column_is_not_read_as_enabled(self):
ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF_INTRUSION)
assert ifaces["1"]["is_enabled"] is True
assert ifaces["1"]["is_up"] is True
assert ifaces["1"]["speed"] == 1000.0
assert ifaces["2"]["is_up"] is False
assert ifaces["6"]["is_enabled"] is False
def test_trunk_suffix_becomes_trunk_group(self):
ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF_INTRUSION)
assert ifaces["3"]["trunk_group"] == "Trk3"
assert ifaces["4"]["trunk_group"] == "Trk3"
assert ifaces["7"]["trunk_group"] == "Trk6"
assert "trunk_group" not in ifaces["1"]
class TestParseArpTable:
@@ -349,6 +425,40 @@ class TestDriverGetInterfaces:
assert "1" in ifaces
assert ifaces["1"]["is_up"] is True
def test_trunk_member_detail_uses_bare_port_name(self, driver):
"""`show interfaces 3-Trk3` is not a command; the port is `3` (#3)."""
driver._send_command = MagicMock(return_value=SHOW_INTERFACES_BRIEF_INTRUSION)
driver.get_interfaces()
sent = [c.args[0] for c in driver._send_command.call_args_list]
assert "show interfaces 3" in sent
assert not any("Trk" in c for c in sent)
class TestSendCommandAlternatives:
"""A list of commands falls through to the next on a CLI error (#4)."""
def test_invalid_input_tries_next_command(self, driver):
outputs = {
"show system-information": SHOW_SYSTEM_INFORMATION_INVALID,
"show system information": SHOW_SYSTEM_J15,
}
driver._device.send_command.side_effect = lambda cmd, **kw: outputs[cmd]
out = driver._send_command(["show system-information", "show system information"])
assert out == SHOW_SYSTEM_J15.strip()
def test_facts_on_j15_firmware(self, driver):
outputs = {
"show system-information": SHOW_SYSTEM_INFORMATION_INVALID,
"show system information": SHOW_SYSTEM_J15,
"show interfaces brief": SHOW_INTERFACES_BRIEF_INTRUSION,
"show version": "Image stamp: /ws/swbuildm/J_rel/code/build\n J.15.09.0028",
}
driver._device.send_command.side_effect = lambda cmd, **kw: outputs[cmd]
facts = driver.get_facts()
assert facts["os_version"] == "J.15.09.0028"
assert facts["serial_number"] == "SG12345678"
assert len(facts["interface_list"]) == 10
class TestDriverGetArpTable:
def test_arp_from_cli(self, driver):
@@ -425,6 +535,60 @@ class TestDriverTransportDetection:
with pytest.raises(Exception):
drv.open()
def test_auth_failure_names_why_earlier_transports_failed(self):
"""A Telnet login failure reports the transport and the earlier failures.
Otherwise "Login failed" reads as a wrong password when SSH was merely
refused and Telnet was the only transport left to answer (#2).
"""
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
def connect(**kwargs):
if kwargs["device_type"] == ProcurveDriver.NETMIKO_DEVICE_TYPE_TELNET:
raise NetmikoAuthenticationException("Login failed: 192.168.0.1")
if kwargs["disabled_algorithms"] == _SSH_DISABLED_STANDARD:
raise ConnectionRefusedError("[Errno 111] Connection refused")
raise NetmikoTimeoutException("TCP connection to device failed")
with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)):
with patch("napalm_procurve.procurve.ConnectHandler", side_effect=connect):
drv = ProcurveDriver("192.168.0.1", "manager", "secret")
with pytest.raises(ConnectionException) as exc_info:
drv.open()
msg = str(exc_info.value)
assert "Authentication failed for 192.168.0.1 via telnet: Login failed: 192.168.0.1" in msg
assert "api: not detected" in msg
assert "ssh: [Errno 111] Connection refused" in msg
assert "ssh_legacy: timed out" in msg
def test_all_transports_fail_names_each_reason(self):
"""The final error gives a reason per transport, not just its name."""
with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)):
with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("no route to host")):
drv = ProcurveDriver("192.168.0.1", "manager", "secret")
with pytest.raises(ConnectionException) as exc_info:
drv.open()
msg = str(exc_info.value)
assert "api: not detected" in msg
assert "ssh: no route to host" in msg
assert "ssh_legacy: no route to host" in msg
assert "telnet: no route to host" in msg
def test_reopen_does_not_carry_earlier_attempts(self):
"""Each open() reports only its own attempts."""
with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)):
with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("first")):
drv = ProcurveDriver("192.168.0.1", "manager", "secret")
with pytest.raises(ConnectionException):
drv.open()
with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("second")):
with pytest.raises(ConnectionException) as exc_info:
drv.open()
assert "first" not in str(exc_info.value)
# ===========================================================================
# VLAN parser tests