fix: what bandit and the remaining lint rules found

- Long docstring and command lines are wrapped; recorded device output in
  tests/ keeps its lines (E501 is off there).
- host_firewall: the negation token is the named constant _NEGATION. bandit
  read token == "!" as a hardcoded password.
- ssh_host_keys: _known_hosts_file takes the expected key as an argument
  instead of asserting it; both callers had already returned when it is None.
- ssh_server: the staged path is on the device, not a local temp file
  (nosec B108).
- tests: raise ... from exc, zip(strict=True).
This commit is contained in:
christianmanivong committed 2026-10-22 19:05:11 +02:00
1 parent 08296d0d2e
commit b7350a9b2c
12 files changed
+56 -28

No files matched your search

+2 -1
View File
@@ -190,7 +190,8 @@ class AccessPointDriver(
Keys are SSID names. Each value contains:
* ssid (string) - SSID name (repeated for convenience)
* bridge (string) - bridge interface the VAP is attached to (e.g. ``"br-lan"``, ``"br-guest"``)
* bridge (string) - bridge interface the VAP is attached to
(e.g. ``"br-lan"``, ``"br-guest"``)
* vlan_id (int) - 802.1Q VLAN ID (0 = untagged / no VLAN separation)
* tagged (bool) - whether traffic is 802.1Q-tagged on the uplink port
* client_isolation (bool) - whether clients on this SSID are isolated from each other
+11 -5
View File
@@ -166,6 +166,9 @@ def _sections(lines: list[str]) -> dict[str, tuple[list[str], int | None]]:
# ── iptables-save ────────────────────────────────────────────────────────────
#: The token that negates the option after it (``! -s 10.0.0.0/8``).
_NEGATION = "!"
#: Options that start a new part of a rule, outside any ``-m`` module.
_TOP_LEVEL = frozenset(
{
@@ -244,13 +247,13 @@ def _module_options(module: str, rt: _RuleTokens, matches: list[NetfilterMatchDi
unknown = module not in _KNOWN_MODULES
while rt.more():
token = rt.peek()
if token in _TOP_LEVEL or (token == "!" and rt.peek(1) in _TOP_LEVEL):
if token in _TOP_LEVEL or (token == _NEGATION and rt.peek(1) in _TOP_LEVEL):
break
negated = False
if token == "!":
if token == _NEGATION:
rt.take()
negated = True
text.append("!")
text.append(_NEGATION)
option = rt.take()
value = rt.take() if rt.more() and not str(rt.peek()).startswith("-") else ""
text += [option] + ([value] if value else [])
@@ -266,7 +269,10 @@ def _module_options(module: str, rt: _RuleTokens, matches: list[NetfilterMatchDi
def _module_match(module: str, option: str, value: str, negated: bool) -> Any:
"""The match one module option makes; False for one that matches nothing new; None if not read."""
"""The match one module option makes.
False for one that matches nothing new; None if not read.
"""
if module == "comment":
return False
if module in ("tcp", "udp", "sctp", "udplite"):
@@ -323,7 +329,7 @@ def _iptables_rule(line: str, chains: set[str]) -> tuple[str, NetfilterRuleDict]
}
while rt.more():
token = rt.take()
if token == "!":
if token == _NEGATION:
negated = True
continue
if token in simple:
+2 -1
View File
@@ -330,7 +330,8 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
:param name: VM name or its ``vmid``.
:param snapshot: Name of the snapshot to delete.
:raises ValueError: If the VM or snapshot does not exist.
:raises RuntimeError: If other snapshots depend on this one (must delete children first).
:raises RuntimeError: If other snapshots depend on this one
(must delete children first).
Example::
+2 -1
View File
@@ -41,7 +41,8 @@ KERNEL_FACTS_COMMAND = (
"if [ -r $m/modules.builtin ]; then echo '[builtin]'; cat $m/modules.builtin; fi; "
"if [ -r $m/modules.dep ]; then echo '[available]'; cut -d: -f1 $m/modules.dep; fi; "
"if [ -r /boot/config-$r ]; then echo '[config]'; grep '^CONFIG_' /boot/config-$r; "
"elif [ -r /proc/config.gz ]; then echo '[config]'; zcat /proc/config.gz | grep '^CONFIG_'; fi; "
"elif [ -r /proc/config.gz ]; then echo '[config]'; "
"zcat /proc/config.gz | grep '^CONFIG_'; fi; "
"} 2>/dev/null | gzip -c | base64; "
"printf '%s%s\\n' KFACTS_ END"
)
+8 -4
View File
@@ -43,15 +43,18 @@ class MacAclMixin:
"name": "CorpWiFi",
"policy": "allow",
"entries": [
{"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "CEO-Laptop"},
{"mac": "AA:BB:CC:DD:EE:02", "action": "allow", "description": "CFO-Laptop"},
{"mac": "AA:BB:CC:DD:EE:01", "action": "allow",
"description": "CEO-Laptop"},
{"mac": "AA:BB:CC:DD:EE:02", "action": "allow",
"description": "CFO-Laptop"},
],
},
"GuestNet": {
"name": "GuestNet",
"policy": "deny",
"entries": [
{"mac": "DE:AD:BE:EF:00:01", "action": "deny", "description": "blocked device"},
{"mac": "DE:AD:BE:EF:00:01", "action": "deny",
"description": "blocked device"},
],
},
}
@@ -72,7 +75,8 @@ class MacAclMixin:
Example::
driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"])
macs = ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"]
driver.push_mac_acl("CorpWiFi", "whitelist", macs)
driver.push_mac_acl("GuestNet", "off", [])
"""
...
+3 -1
View File
@@ -1493,7 +1493,9 @@ class ManagementProtocolsDict(TypedDict):
class SystemIdentityDict(TypedDict):
"""A host's SMBIOS identity, what ties it to its BMC (:mod:`napalm_device_types.system_identity`).
"""A host's SMBIOS identity, what ties it to its BMC.
See :mod:`napalm_device_types.system_identity`.
Each value is empty when unknown or a vendor placeholder."""
+5 -6
View File
@@ -212,7 +212,7 @@ class HostKeyCheck:
if self.expected is None:
client.set_missing_host_key_policy(_AcceptFirst())
return
with self._known_hosts_file() as path:
with self._known_hosts_file(self.expected) as path:
client.get_host_keys().load(path)
client.set_missing_host_key_policy(_RefuseUnknown())
@@ -243,7 +243,7 @@ class HostKeyCheck:
with self.refusing():
yield kwargs
return
with self._known_hosts_file() as path:
with self._known_hosts_file(self.expected) as path:
kwargs.update(ssh_strict=True, alt_host_keys=True, alt_key_file=path)
with self.refusing():
yield kwargs
@@ -285,13 +285,12 @@ class HostKeyCheck:
)
@contextmanager
def _known_hosts_file(self) -> Iterator[str]:
"""A private ``known_hosts`` file holding only the expected key."""
assert self.expected is not None
def _known_hosts_file(self, expected: SshHostKey) -> Iterator[str]:
"""A private ``known_hosts`` file holding only *expected*."""
fd, path = tempfile.mkstemp(prefix="known_hosts-")
try:
with os.fdopen(fd, "w") as f:
f.write(self.expected.known_hosts_line(self.host, self.port) + "\n")
f.write(expected.known_hosts_line(self.host, self.port) + "\n")
yield path
finally:
try:
+4 -2
View File
@@ -369,7 +369,8 @@ class OpenSshServerMixin:
stdin=content.encode(),
)
tmp = staged.stdout.strip()
if staged.exit_code != 0 or not tmp.startswith("/tmp/netork-ssh."):
# A path on the device, made by mktemp under umask 077: not a local temp file.
if staged.exit_code != 0 or not tmp.startswith("/tmp/netork-ssh."): # nosec B108
raise RuntimeError(
f"could not stage {path}: {(staged.stderr or staged.stdout).strip()}"
)
@@ -378,7 +379,8 @@ class OpenSshServerMixin:
# GNU stat, then BSD's; 644 for a file that does not exist yet.
f'r={quote(mode_of or path)}; m=$(stat -c %a -- "$r" 2>/dev/null'
f' || stat -f %Lp -- "$r" 2>/dev/null) || m=644;'
f' install -m "$m" -- {quote(tmp)} {quote(part)} && mv -f -- {quote(part)} {quote(path)};'
f' install -m "$m" -- {quote(tmp)} {quote(part)}'
f" && mv -f -- {quote(part)} {quote(path)};"
f" s=$?; rm -f -- {quote(tmp)} {quote(part)}; exit $s"
)
if result.exit_code != 0:
+10 -5
View File
@@ -63,7 +63,8 @@ class StorageDriver(PackageManagementMixin, DeviceTypeDriver):
* rpm (int) - rotational speed; ``0`` for SSD/NVMe
* temperature (int) - current temperature in Celsius; ``-1`` if unavailable
* health (string) - ``"healthy"``, ``"warning"``, ``"failed"``, or ``"unknown"``
* pool (string) - name of the pool this disk belongs to; empty string if unassigned/spare
* pool (string) - name of the pool this disk belongs to;
empty string if unassigned/spare
Example::
@@ -107,12 +108,14 @@ class StorageDriver(PackageManagementMixin, DeviceTypeDriver):
* type (string) - ``"zfs"``, ``"lvm"``, ``"md"``, ``"hardware-raid"``, ``"btrfs"``
* level (string) - RAID/redundancy level: ``"mirror"``, ``"raidz1"``, ``"raidz2"``,
``"raidz3"``, ``"stripe"``, ``"raid0"`` … ``"raid60"``, ``"single"``
* status (string) - ``"online"``, ``"degraded"``, ``"faulted"``, ``"offline"``, ``"unknown"``
* status (string) - ``"online"``, ``"degraded"``, ``"faulted"``,
``"offline"``, ``"unknown"``
* total (int) - usable capacity in bytes
* used (int) - used space in bytes
* available (int) - free space in bytes
* disks (list of strings) - slot identifiers of member drives
* auto_expand (bool) - whether the pool grows automatically when disks are replaced with larger ones
* auto_expand (bool) - whether the pool grows automatically when disks
are replaced with larger ones
* dedup (bool) - whether deduplication is enabled
* compression (string) - pool-level compression algorithm: ``"off"``, ``"lz4"``,
``"gzip"``, ``"zstd"``, etc.
@@ -155,7 +158,8 @@ class StorageDriver(PackageManagementMixin, DeviceTypeDriver):
* used (int) - space currently used in bytes
* available (int) - space available in bytes
* mountpoint (string) - local mount path; empty string for block volumes / LUNs
* compression (string) - active compression algorithm (``"off"``, ``"lz4"``, ``"zstd"``, etc.)
* compression (string) - active compression algorithm
(``"off"``, ``"lz4"``, ``"zstd"``, etc.)
* dedup (bool) - whether deduplication is active on this volume
* readonly (bool) - whether the volume is mounted read-only
* snapshots (int) - number of snapshots currently held
@@ -254,7 +258,8 @@ class StorageDriver(PackageManagementMixin, DeviceTypeDriver):
Each entry contains:
* name (string) - snapshot name (e.g. ``"auto-2026-05-11"`` or full ``"tank/media@auto-2026-05-11"``)
* name (string) - snapshot name (e.g. ``"auto-2026-05-11"``
or full ``"tank/media@auto-2026-05-11"``)
* volume (string) - parent volume / dataset path
* created (float) - creation timestamp (Unix epoch)
* size (int) - bytes of unique data referenced only by this snapshot
+4
View File
@@ -71,6 +71,10 @@ target-version = "py310"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]
[tool.ruff.lint.per-file-ignores]
# Recorded device output keeps its lines.
"tests/**" = ["E501"]
[tool.mypy]
python_version = "3.10"
strict = true
+1 -1
View File
@@ -252,7 +252,7 @@ class TestNetmiko:
other = paramiko.ECDSAKey.generate()
raise paramiko.BadHostKeyException("10.0.0.5", host_key, other)
except paramiko.SSHException as exc:
raise RuntimeError(f"A paramiko SSHException occurred: {exc}")
raise RuntimeError(f"A paramiko SSHException occurred: {exc}") from exc
except RuntimeError as exc:
raise ConnectionException(str(exc)) from exc
assert caught.value.observation.seen == SshHostKey.of(host_key)
+4 -1
View File
@@ -33,7 +33,10 @@ def _declared(name: str) -> ast.FunctionDef:
class TestCreateVmTakesACpuType:
def test_cpu_type_is_an_optional_keyword(self):
fn = _declared("create_vm_from_cloud_init")
kwonly = {arg.arg: default for arg, default in zip(fn.args.kwonlyargs, fn.args.kw_defaults)}
kwonly = {
arg.arg: default
for arg, default in zip(fn.args.kwonlyargs, fn.args.kw_defaults, strict=True)
}
assert "cpu_type" in kwonly
default = kwonly["cpu_type"]
assert isinstance(default, ast.Constant) and default.value is None