From 0cab237b1d71260270641092a7f255e020b2646b Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Tue, 9 Jun 2026 23:12:32 +0200 Subject: [PATCH] fix: fix_snmp SSH fallback when driver is connected via REST API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The REST API has no writable config endpoint for SNMP communities. When _action_fix_snmp() is called on a REST-connected driver, it now opens a temporary Netmiko SSH session (standard KEX, then legacy KEX fallback), runs the config-mode CLI commands, and restores the REST transport state on exit. SSH errors are surfaced verbatim instead of being silently swallowed. Also: - _save_config: guard against REST transport (was calling self._device directly → NoneType AttributeError when transport == "api") - _action_fix_snmp: extracted CLI logic into _action_fix_snmp_cli() so it is reused for both the SSH-fallback and native SSH/Telnet paths - api_client: add get_snmp_communities() + configure_snmp_community() for future use; improve GET error logging (warn vs debug on non-404) - _try_api: probe timeout capped at 15 s; SSL retry loop for self-signed certificates; improved log levels - _api_set_interface / _cli_set_interface: handle enabled/description fields that were previously ignored - probe(): broaden accepted status codes (301/302/303/403) Co-Authored-By: Claude Sonnet 4.6 --- napalm_procurve/api_client.py | 41 +++++++++- napalm_procurve/procurve.py | 149 ++++++++++++++++++++++++++-------- 2 files changed, 153 insertions(+), 37 deletions(-) diff --git a/napalm_procurve/api_client.py b/napalm_procurve/api_client.py index a982df1..878123e 100644 --- a/napalm_procurve/api_client.py +++ b/napalm_procurve/api_client.py @@ -79,7 +79,9 @@ class ProcurveApiClient: # 401 = API exists but not authenticated (most common) # 200 = API exists and accessible without auth (unusual) # 405 = wrong HTTP verb but API is there - if resp.status_code in (200, 401, 405): + # 302/301/303 = redirect to login page (older firmware) + # 403 = API exists but access forbidden + if resp.status_code in (200, 301, 302, 303, 401, 403, 405): logger.debug( "API detected at %s (HTTP %s, %s %s)", hostname, resp.status_code, proto, ver, @@ -169,9 +171,12 @@ class ProcurveApiClient: resp = self._session.get(url, timeout=self.timeout) if resp.ok: return resp.json() - logger.debug("GET %s returned HTTP %s", url, resp.status_code) + if resp.status_code == 404: + logger.debug("GET %s returned HTTP 404", url) + else: + logger.warning("GET %s returned HTTP %s: %s", url, resp.status_code, resp.text[:200]) except Exception as exc: - logger.debug("GET %s error: %s", url, exc) + logger.warning("GET %s error: %s", url, exc) return {} def post(self, endpoint: str, **kwargs: Any) -> requests.Response: @@ -481,6 +486,36 @@ class ProcurveApiClient: f"delete_vlan({vlan_id}): REST API returned HTTP {resp.status_code}" ) + def get_snmp_communities(self) -> Dict[str, Dict]: + """Return configured SNMP communities from REST API.""" + data = self.get("snmp-server/community") + result: Dict[str, Dict] = {} + for entry in data.get("snmp_community_element", []): + name = entry.get("community_name", "") + if name: + result[name] = {"access_type": entry.get("access_type", "")} + return result + + def configure_snmp_community(self, community: str = "public") -> Tuple[bool, str]: + """Ensure an SNMP read-only community exists via REST resource endpoint. + + Returns (success, message). + """ + existing = self.get_snmp_communities() + if community in existing: + return True, f"SNMP community '{community}' already exists." + + resp = self.post("snmp-server/community", json={ + "community_name": community, + "access_type": "MIB2", + }) + if resp.ok: + return True, f"SNMP community '{community}' configured via REST API." + if resp.status_code == 409: + return True, f"SNMP community '{community}' already exists." + + return False, f"REST API returned HTTP {resp.status_code}: {resp.text[:200]}" + def ping( self, destination: str, diff --git a/napalm_procurve/procurve.py b/napalm_procurve/procurve.py index cb0fef8..f94c2a2 100644 --- a/napalm_procurve/procurve.py +++ b/napalm_procurve/procurve.py @@ -203,37 +203,46 @@ class ProcurveDriver(SwitchDriver): def _try_api(self) -> bool: """Probe and connect via REST API. Returns True on success.""" - logger.debug("Trying REST API for %s", self.hostname) + 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=5, ssl_verify=self.ssl_verify + 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.debug("REST API not available on %s", self.hostname) + logger.warning("REST API not detected on %s — falling back to CLI", self.hostname) return False - client = ProcurveApiClient( - hostname=self.hostname, - username=self.username, - password=self.password, - timeout=self.timeout, - ssl_verify=self.ssl_verify, - api_version=ver, - ) - client.setup(ver, proto) - try: - client.connect() - except Exception as exc: - logger.debug("REST API connect failed: %s", exc) - 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 - self._api = client - self._transport = "api" - logger.info("Connected to %s via REST API (%s %s)", self.hostname, proto, ver) - return True + logger.warning("REST API connect failed for %s — falling back to CLI", self.hostname) + return False def _try_ssh(self, legacy: bool = False) -> bool: """Probe and connect via SSH. Returns True on success.""" @@ -351,6 +360,9 @@ class ProcurveDriver(SwitchDriver): 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(), @@ -780,6 +792,17 @@ class ProcurveDriver(SwitchDriver): 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", []): @@ -811,6 +834,17 @@ class ProcurveDriver(SwitchDriver): 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}") @@ -1067,28 +1101,75 @@ class ProcurveDriver(SwitchDriver): 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): + disabled = _SSH_DISABLED_LEGACY if legacy else _SSH_DISABLED_STANDARD + try: + conn = ConnectHandler( + 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=disabled, + **self.netmiko_optional_args, + ) + 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() - # Enable SNMP with read-only community 'public' - # ProCurve: manager = standard access level, restricted = read-only - self._send_command('snmp-server community "public" manager restricted') - lines.append("[config] SNMP community 'public' (manager restricted / read-only) configured.") - - self._exit_config_mode() self._save_config() lines.append("[config] Configuration saved.") - # Verify out = self._send_command("show snmp-server") success = "public" in out - if success: - lines.append("[ok] SNMP is active with community 'public'.") - else: - lines.append(f"[warn] Verification — community not found: {out[:200]}") - + 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)} # ------------------------------------------------------------------