From f35b59ec2d4a7bf8b8eeb1db3c57c0fc71081d6e Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 25 Sep 2026 08:55:11 +0200 Subject: [PATCH] fix(cli): read J.15 port status, and say why each transport failed 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 `-Trk`; the port is `` 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 --- napalm_procurve/parsers.py | 21 +++-- napalm_procurve/procurve.py | 42 ++++++--- tests/unit/test_driver.py | 166 +++++++++++++++++++++++++++++++++++- 3 files changed, 205 insertions(+), 24 deletions(-) diff --git a/napalm_procurve/parsers.py b/napalm_procurve/parsers.py index ac9b835..c520294 100644 --- a/napalm_procurve/parsers.py +++ b/napalm_procurve/parsers.py @@ -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 diff --git a/napalm_procurve/procurve.py b/napalm_procurve/procurve.py index 313c1f5..c8057ae 100644 --- a/napalm_procurve/procurve.py +++ b/napalm_procurve/procurve.py @@ -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 ": " + 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) diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 950fb09..02b29c7 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -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 "-Trk" (#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