fix: what bandit and the remaining lint rules found

- driver.py: the logger is defined after the imports (E402). The lazy
  paramiko client is annotated Any, as _node_ssh already is; the quoted
  paramiko.SSHClient named a module that is never imported there (F821).
- Unused locals are gone: get_facts' hostname, get_users' groups (both
  read a value nobody used), the ARP flags column is _flags; the output is
  unchanged.
- Long lines are wrapped: the apt-get and snmpd.conf commands are split
  into adjacent string literals, the power-action error message and the
  system mixin's docstring are wrapped; the text is unchanged.
- get_vm_status re-raises with raise ... from e (B904).
- The two except Exception: pass (closing the SSH session, the is_alive
  probe) now log at debug level (B110), so a swallowed failure leaves a
  trace.
- nosec for the empty "password" field of NAPALM's get_users shape and
  the ssh_password_auth flag (B105); all three are false positives.
- tests: unused loop variables.
This commit is contained in:
Christian Manivong committed 2026-10-23 08:39:01 +02:00
1 parent 3b873d91fe
commit 3365207c98
6 files changed
+28 -19

No files matched your search

+6 -6
View File
@@ -32,8 +32,6 @@ from __future__ import annotations
import logging import logging
from typing import Any from typing import Any
logger = logging.getLogger(__name__)
from napalm.base.exceptions import ConnectionException from napalm.base.exceptions import ConnectionException
from napalm_device_types import ( from napalm_device_types import (
AptProxyMixin, AptProxyMixin,
@@ -68,6 +66,8 @@ from napalm_proxmox.vm_mixin import ProxmoxVMMixin
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin
logger = logging.getLogger(__name__)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# Type aliases # Type aliases
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
@@ -156,7 +156,8 @@ class ProxmoxDriver(
self._api: ProxmoxAPI | None = None self._api: ProxmoxAPI | None = None
self._node_name: str = "" self._node_name: str = ""
self._ssh_client: paramiko.SSHClient | None = None # paramiko.SSHClient; paramiko is imported lazily (see _node_ssh).
self._ssh_client: Any = None
# Candidate config (merge/replace) # Candidate config (merge/replace)
self._candidate_config: str = "" self._candidate_config: str = ""
@@ -257,7 +258,7 @@ class ProxmoxDriver(
try: try:
self._ssh_client.close() self._ssh_client.close()
except Exception: except Exception:
pass logger.debug("%s: closing the SSH session: ignored", self.hostname, exc_info=True)
self._ssh_client = None self._ssh_client = None
def is_alive(self) -> _JsonDict: def is_alive(self) -> _JsonDict:
@@ -274,7 +275,7 @@ class ProxmoxDriver(
self._api.version.get() self._api.version.get()
alive = True alive = True
except Exception: except Exception:
pass logger.debug("%s: liveness probe failed", self.hostname, exc_info=True)
return {"is_alive": alive} return {"is_alive": alive}
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
@@ -432,7 +433,6 @@ class ProxmoxDriver(
uptime = float(status.get("uptime", 0)) uptime = float(status.get("uptime", 0))
dns_search = dns.get("search", "") dns_search = dns.get("search", "")
hostname = self._node_name
fqdn = f"{self._node_name}.{dns_search}" if dns_search else self.hostname fqdn = f"{self._node_name}.{dns_search}" if dns_search else self.hostname
iface_list = sorted(iface["iface"] for iface in network if iface.get("iface")) iface_list = sorted(iface["iface"] for iface in network if iface.get("iface"))
+1 -1
View File
@@ -174,7 +174,7 @@ class ProxmoxInterfaceMixin:
parts = line.split() parts = line.split()
if len(parts) < 6: if len(parts) < 6:
continue continue
ip_addr, _, flags, mac, _, iface = ( ip_addr, _, _flags, mac, _, iface = (
parts[0], parts[0],
parts[1], parts[1],
parts[2], parts[2],
+14 -7
View File
@@ -12,7 +12,10 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
"""System-level NAPALM getters for Proxmox VE nodes (environment, NTP, SNMP, users, packages, services, updates, disk SMART).""" """System-level NAPALM getters for Proxmox VE nodes.
Environment, NTP, SNMP, users, packages, services, updates and disk SMART.
"""
from __future__ import annotations from __future__ import annotations
@@ -224,7 +227,6 @@ class ProxmoxSystemMixin:
if not uid: if not uid:
continue continue
# Proxmox roles: Administrator -> 15, otherwise 1 # Proxmox roles: Administrator -> 15, otherwise 1
groups = user.get("groups", "") or ""
level = 1 level = 1
try: try:
roles = self._api.access.users(uid).get() or {} # type: ignore[union-attr] roles = self._api.access.users(uid).get() or {} # type: ignore[union-attr]
@@ -234,7 +236,8 @@ class ProxmoxSystemMixin:
logger.debug("Failed to fetch roles for user %s: %s", uid, exc) logger.debug("Failed to fetch roles for user %s: %s", uid, exc)
result[uid] = { result[uid] = {
"level": level, "level": level,
"password": "", # NAPALM's get_users shape; the password is never read.
"password": "", # nosec B105
"sshkeys": [], "sshkeys": [],
} }
@@ -252,7 +255,8 @@ class ProxmoxSystemMixin:
if uname not in result and uid_int < 1000 or uid_int == 0: if uname not in result and uid_int < 1000 or uid_int == 0:
result[uname] = { result[uname] = {
"level": 15 if uid_int == 0 else 0, "level": 15 if uid_int == 0 else 0,
"password": "", # NAPALM's get_users shape; the password is never read.
"password": "", # nosec B105
"sshkeys": [], "sshkeys": [],
} }
@@ -445,7 +449,8 @@ class ProxmoxSystemMixin:
raise ValueError(f"Invalid package name: {pkg!r}") raise ValueError(f"Invalid package name: {pkg!r}")
pkg_args = " ".join(packages) pkg_args = " ".join(packages)
output = self._exec_ssh_command( output = self._exec_ssh_command(
f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1 || true" "DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y "
f"{pkg_args} 2>&1 || true"
) )
return {"success": True, "output": output} return {"success": True, "output": output}
@@ -475,7 +480,8 @@ class ProxmoxSystemMixin:
port = 161 port = 161
try: try:
conf = self._exec_ssh_command( conf = self._exec_ssh_command(
"grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf 2>/dev/null | head -5" "grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf "
"2>/dev/null | head -5"
) )
for line in conf.splitlines(): for line in conf.splitlines():
parts = line.split() parts = line.split()
@@ -809,7 +815,8 @@ class ProxmoxSystemMixin:
"timezone": None, "timezone": None,
"ntp_servers": [], "ntp_servers": [],
"ssh_port": 22, "ssh_port": 22,
"ssh_password_auth": False, # A flag, not a password.
"ssh_password_auth": False, # nosec B105
} }
# Cluster name and node list from /cluster/status # Cluster name and node list from /cluster/status
+3 -1
View File
@@ -387,7 +387,9 @@ class ProxmoxVMMixin:
if action not in allowed: if action not in allowed:
return { return {
"success": False, "success": False,
"error": f"Action '{action}' not supported for {vm_type} (allowed: {sorted(allowed)})", "error": (
f"Action '{action}' not supported for {vm_type} (allowed: {sorted(allowed)})"
),
} }
try: try:
vm_api = self._node_api().qemu(vmid) if vm_type == "vm" else self._node_api().lxc(vmid) vm_api = self._node_api().qemu(vmid) if vm_type == "vm" else self._node_api().lxc(vmid)
+1 -1
View File
@@ -939,7 +939,7 @@ class ProxmoxVMProvisionMixin:
raise raise
except Exception as e: except Exception as e:
_logger.exception(f"Error getting status for VM {vmid}: {e}") _logger.exception(f"Error getting status for VM {vmid}: {e}")
raise RuntimeError(f"Failed to get VM {vmid} status: {e}") raise RuntimeError(f"Failed to get VM {vmid} status: {e}") from e
def get_network_targets(self) -> list[NetworkTargetDict]: def get_network_targets(self) -> list[NetworkTargetDict]:
""" """
+3 -3
View File
@@ -86,7 +86,7 @@ class TestGetUsers:
def test_user_has_required_keys(self, driver): def test_user_has_required_keys(self, driver):
driver._node_api().execute.post.return_value = {"data": ""} driver._node_api().execute.post.return_value = {"data": ""}
result = driver.get_users() result = driver.get_users()
for user, data in result.items(): for _user, data in result.items():
assert "level" in data assert "level" in data
assert "password" in data assert "password" in data
assert "sshkeys" in data assert "sshkeys" in data
@@ -249,7 +249,7 @@ class TestGetRouteTo:
def test_route_fields(self, driver): def test_route_fields(self, driver):
driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT} driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
result = driver.get_route_to() result = driver.get_route_to()
for prefix, routes in result.items(): for _prefix, routes in result.items():
for route in routes: for route in routes:
assert "protocol" in route assert "protocol" in route
assert "next_hop" in route assert "next_hop" in route
@@ -258,7 +258,7 @@ class TestGetRouteTo:
def test_protocol_filter(self, driver): def test_protocol_filter(self, driver):
driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT} driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
result = driver.get_route_to(protocol="dhcp") result = driver.get_route_to(protocol="dhcp")
for prefix, routes in result.items(): for _prefix, routes in result.items():
for route in routes: for route in routes:
assert "dhcp" in route["protocol"].lower() assert "dhcp" in route["protocol"].lower()