NAPALM/napalm-linux
1
0 Fork 0
Code Issues 12 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

chore: adopt the NAPALM driver standard #38

Merged
christianmanivong merged 6 commits from chore/driver-standard into main 2026-10-23 12:55:55 +00:00
Conversation 0 Commits 6 Files Changed 14
+748 -294
14 changed files with 748 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitea/workflows/ci.yml
+17
View File
@@ -0,0 +1,17 @@
name: CI
on:
push:
branches: [main, master]
tags: ["v*"]
pull_request:
jobs:
driver:
uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
with:
standard-ref: main
# Until napalm-device-types 4.22.0 (AptProxyMixin, #53) is on PyPI.
pre-install: "git+https://git.netork.io/NAPALM/napalm-device-types.git"
secrets:
PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
.github/workflows/ci.yml
-48
View File
@@ -1,48 +0,0 @@
name: CI
on:
push:
branches: ["**"]
pull_request:
branches: ["**"]
jobs:
test:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.10", "3.11", "3.12"]
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: pip
- name: Install package with dev extras
run: |
python -m pip install --upgrade pip
# napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
# pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
python -m pip install -e ".[dev]"
- name: Run unit tests
run: |
python -m pytest -q --tb=short
- name: Build wheel and sdist
run: |
python -m pip install build
python -m build
- name: Upload dist artifacts
# v4 refuses to run on Gitea ("not currently supported on GHES").
uses: actions/upload-artifact@v3
with:
name: dist-${{ matrix.python-version }}
path: dist/*
CHANGELOG.md
+15
View File
@@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [0.15.0] - 2026-10-23
The first release on PyPI. It collects everything since 0.4.0, which was the last version recorded here.
### Changed
- The driver follows the NAPALM driver standard
(https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
a build check, and a release tag publishes to PyPI. The package now ships `py.typed`
and a LICENSE file. **Python 3.9 is no longer supported**; napalm-device-types never
ran on it.
- Exceptions the driver deliberately ignores (the PTY width at login, closing a dead
session, unreadable route lines, the snmpd.conf community and port) are now logged
at debug level.
### Added
- apt's proxy through napalm-device-types' `AptProxyMixin` (NetOrk/netork#1397). `get_package_proxy()` is what `apt-config dump` says, not `00proxy` alone: a proxy in `90curtin-aptproxy` or `/etc/apt/apt.conf` wins over it, and was missed. New: `get_package_proxy_report()` (every line that sets it, per-host overrides, the https sources) and `check_package_proxy(url, expect)` (a fetch with `apt-helper`, past any proxy). `configure_package_proxy(url)` now also comments out other proxy lines, reads the result back and fetches one `InRelease` through the proxy; it fails when apt still fetches elsewhere. Reads run on the exec channel. Requires napalm-device-types 4.22.0; version 0.14.0.
- `get_host_firewall()` (napalm-device-types' `HostFirewallMixin`, NetOrk/netork#1344): the host's own netfilter rules — `nft -j list ruleset`, with the tables iptables manages taken from `iptables-save` — as chains and rules in one shape. Read-only, as root, on an exec channel (the nft JSON is one long line). Requires napalm-device-types 4.21.0; version 0.13.0.
LICENSE
+191
View File
@@ -0,0 +1,191 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship made available under
the License, as indicated by a copyright notice that is included in
or attached to the work (an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other transformations
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean, as submitted to the Licensor for inclusion
in the Work by the copyright owner or by an individual or Legal Entity
authorized to submit on behalf of the copyright owner. For the purposes
of this definition, "submitted" means any form of electronic, verbal,
or written communication sent to the Licensor or its representatives,
including but not limited to communication on electronic mailing lists,
source code control systems, and issue tracking systems that are managed
by, or on behalf of, the Licensor for the purpose of discussing and
improving the Work, but excluding communication that is conspicuously
marked or designated in writing by the copyright owner as "Not a
Contribution."
"Contributor" shall mean Licensor and any Legal Entity on behalf of
whom a Contribution has been received by the Licensor and incorporated
within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by the combined Contribution(s) and the Work
to which such Contribution(s) was submitted. If You institute patent
litigation against any entity (including a cross-claim or counterclaim
in a lawsuit) alleging that the Work or any Contribution embodied
within the Work constitutes direct or contributory patent infringement,
then any patent licenses granted to You under this License for that
Work shall terminate as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or Derivative
Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, You must include a readable copy of the
attribution notices contained within such NOTICE file, in
at least one of the following places: within a NOTICE text file
distributed as part of the Derivative Works; within the Source
form or documentation, if provided along with the Derivative
Works; or, within a display generated by the Derivative Works,
if and wherever such third-party notices normally appear. The
contents of the NOTICE file are for informational purposes only
and do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own license statement for Your modifications and
may provide additional grant of rights to use, copy, modify, merge,
publish, distribute, sublicense, and/or sell copies of the Work,
and to permit persons to whom the Work is furnished to do so,
subject to the following conditions:
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any conditions of TITLE,
MERCHANTIBILITY, or FITNESS FOR A PARTICULAR PURPOSE.
See the License for the specific language governing permissions and
limitations under the License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or exemplary damages of any character arising as a
result of this License or out of the use or inability to use the
Work (even if such Contributor has been advised of the possibility
of such damages).
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may offer only
conditions consistent with this License, subject to the following
additional conditions:
(a) You may offer additional warranty, indemnity, or other liability
terms and conditions; and
(b) You may offer support for the Work in return for a fee.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format in question. It may also be
possible to make it available with a creative commons license.
Copyright 2026 Christian Manivong
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
README.md
+1 -1
View File
@@ -24,7 +24,7 @@ pip install napalm napalm-linux
Or from source:
```bash
git clone https://github.com/chrismanivong/napalm-linux
git clone https://git.netork.io/NAPALM/napalm-linux
pip install -e napalm-linux/
```
napalm_linux/linux.py
+256 -161
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0
"""NAPALM driver for generic Linux systems.
@@ -21,20 +20,15 @@ from __future__ import annotations
import ipaddress
import logging
import re
import socket
from shlex import quote as _shlex_quote
from typing import Any, Dict, List, Optional
from typing import Any
from netmiko import ConnectHandler
from netmiko.exceptions import (
NetmikoAuthenticationException,
NetmikoTimeoutException,
)
from napalm.base.exceptions import ConnectionException, ConnectionClosedException
from napalm.base.exceptions import ConnectionClosedException, ConnectionException
from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types import (
APT_UPGRADABLE_COMMAND,
DNF_SECURITY_COMMAND,
AptProxyMixin,
ByteStream,
CommandResult,
ContainerEngineMixin,
@@ -46,22 +40,20 @@ from napalm_device_types import (
ListeningSocketsMixin,
OpenSshServerMixin,
OSDriver,
AptProxyMixin,
ServiceClustersMixin,
ServicePackagesMixin,
SnmpAgentMixin,
SystemIdentityMixin,
SystemdServicesMixin,
SystemdUnavailable,
SystemIdentityMixin,
VirtualIpsMixin,
parse_apt_upgradable,
open_stream_on_transport,
parse_apt_upgradable,
parse_dnf_security,
run_fix_steps,
run_on_transport,
strip_terminal_codes,
)
from napalm_device_types.ssh_host_keys import HostKeyCheck
from napalm_device_types.models import (
ApplyUpdatesResultDict,
CronJobDict,
@@ -70,11 +62,17 @@ from napalm_device_types.models import (
PackageDict,
ProcessDict,
ServiceDict,
SNMPConfigDict,
SnmpAgentSettingsDict,
SNMPConfigDict,
UpdateDict,
UserDict,
)
from napalm_device_types.ssh_host_keys import HostKeyCheck
from netmiko import ConnectHandler
from netmiko.exceptions import (
NetmikoAuthenticationException,
NetmikoTimeoutException,
)
logger = logging.getLogger("napalm_linux")
@@ -173,25 +171,35 @@ _NO_STDIN = "exec </dev/null; "
#: Where ``_install_file_as_root`` stages content: a fresh file from ``mktemp``
#: under ``umask 077``, never a fixed name another user could create first.
_STAGE_TEMPLATE = "/tmp/netork-stage.XXXXXX"
#: The path is on the device, not a local temp file.
_STAGE_TEMPLATE = "/tmp/netork-stage.XXXXXX" # nosec B108
_STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")]
_STAGE = (
f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && "
'{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }'
)
#: What to do when sudo wants a password netOrk does not have.
_SUDO_PASSWORD_HINT = (
#: What to do when sudo wants a password netOrk does not have (a message, not a password).
_SUDO_PASSWORD_HINT = ( # nosec B105
"sudo requires a password on this host and no sudo password was given: pass one, "
"or allow this user passwordless sudo (NOPASSWD)."
)
# DMI field values that carry no useful information (OEM defaults, blanks)
_BAD_DMI: frozenset[str] = frozenset({
"", "none", "n/a", "not specified", "not applicable",
"to be filled by o.e.m.", "default string", "unknown",
"no asset tag", "not present",
})
_BAD_DMI: frozenset[str] = frozenset(
{
"",
"none",
"n/a",
"not specified",
"not applicable",
"to be filled by o.e.m.",
"default string",
"unknown",
"no asset tag",
"not present",
}
)
# systemd-detect-virt output → human-readable vendor name
_VIRT_VENDOR_MAP: dict[str, str] = {
@@ -210,16 +218,28 @@ _VIRT_VENDOR_MAP: dict[str, str] = {
}
# Container technologies reported by systemd-detect-virt
_CONTAINER_VIRT: frozenset[str] = frozenset({
"docker", "podman", "lxc", "lxc-libvirt", "systemd-nspawn",
})
_CONTAINER_VIRT: frozenset[str] = frozenset(
{
"docker",
"podman",
"lxc",
"lxc-libvirt",
"systemd-nspawn",
}
)
# DMI sys_vendor strings that indicate a VM when detect-virt is unavailable
_VM_DMI_VENDORS: frozenset[str] = frozenset({
"qemu", "vmware, inc.", "microsoft corporation",
"innotek gmbh", "xen", "bochs",
_VM_DMI_VENDORS: frozenset[str] = frozenset(
{
"qemu",
"vmware, inc.",
"microsoft corporation",
"innotek gmbh",
"xen",
"bochs",
"parallels software international inc.",
})
}
)
# Known ARM board model prefixes → canonical vendor name
_ARM_VENDOR_PREFIXES: list[tuple[str, str]] = [
@@ -255,7 +275,10 @@ def _arm_vendor_from_model(model: str) -> str:
#: Where net-snmp reads its configuration, and how each package manager installs it.
_SNMPD_CONF = "/etc/snmp/snmpd.conf"
_SNMP_INSTALL = {
"apt": "apt-get update -y >/dev/null 2>&1; DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1",
"apt": (
"apt-get update -y >/dev/null 2>&1; "
"DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1"
),
"dnf": "dnf install -y net-snmp net-snmp-utils 2>&1",
"yum": "yum install -y net-snmp net-snmp-utils 2>&1",
"apk": "apk add --no-cache net-snmp net-snmp-tools 2>&1",
@@ -394,13 +417,17 @@ class LinuxDriver(
try:
self._device.send_command(_HISTORY_OFF, expect_string=_PROMPT_RE)
except Exception:
logger.debug("Could not switch the shell history off on %s", self.hostname, exc_info=True)
logger.debug(
"Could not switch the shell history off on %s", self.hostname, exc_info=True
)
# Prevent PTY from wrapping long output lines (e.g. docker JSON).
try:
self._device.send_command("stty cols 10000 2>/dev/null || true", expect_string=r"[#$>]\s*$")
self._device.send_command(
"stty cols 10000 2>/dev/null || true", expect_string=r"[#$>]\s*$"
)
except Exception:
pass
logger.debug("%s: ignored", self.hostname, exc_info=True)
self._pkg_manager = self._forced_pkg_manager or self._detect_pkg_manager()
logger.debug("Connected to %s, pkg_manager=%s", self.hostname, self._pkg_manager)
@@ -411,7 +438,7 @@ class LinuxDriver(
try:
self._device.disconnect()
except Exception:
pass
logger.debug("%s: ignored", self.hostname, exc_info=True)
self._device = None
self._pkg_manager = None
@@ -419,7 +446,7 @@ class LinuxDriver(
if self._device:
try:
return {"is_alive": self._device.remote_conn.transport.is_active()}
except (AttributeError, socket.error, EOFError):
except (OSError, AttributeError, EOFError):
return {"is_alive": False}
return {"is_alive": False}
@@ -471,7 +498,11 @@ class LinuxDriver(
staged = self.run_command(_STAGE, timeout=30, stdin=content.encode())
lines = staged.stdout.strip().splitlines()
tmp = lines[-1] if lines else ""
if staged.exit_code != 0 or not tmp.startswith(_STAGE_PREFIX) or "/" in tmp[len(_STAGE_PREFIX):]:
if (
staged.exit_code != 0
or not tmp.startswith(_STAGE_PREFIX)
or "/" in tmp[len(_STAGE_PREFIX) :]
):
reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}"
return CommandResult(reason, "", staged.exit_code or 1)
tmp_q, part_q, path_q = (_shlex_quote(p) for p in (tmp, f"{path}.netork-tmp", path))
@@ -639,13 +670,15 @@ class LinuxDriver(
"u=$(cat /sys/class/dmi/id/product_uuid 2>/dev/null); "
"d=$(systemd-detect-virt 2>/dev/null); d=${d:-none}; "
"dt=$(tr -d '\\0' </sys/firmware/devicetree/base/model 2>/dev/null); "
"cs=$(grep '^Serial' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 | xargs 2>/dev/null); "
"cm=$(grep '^Model' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 | xargs 2>/dev/null); "
"cs=$(grep '^Serial' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 "
"| xargs 2>/dev/null); "
"cm=$(grep '^Model' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 "
"| xargs 2>/dev/null); "
# DMIBEGIN sentinel: _send() strips leading blank lines (ARM has no DMI
# files, so fields 0-4 are empty). The sentinel anchors the output so
# splitlines()[start+N] always maps to the correct field index.
"printf 'DMIBEGIN\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n' "
"\"$v\" \"$n\" \"$r\" \"$s\" \"$u\" \"$d\" \"$dt\" \"$cs\" \"$cm\""
'"$v" "$n" "$r" "$s" "$u" "$d" "$dt" "$cs" "$cm"'
)
try:
raw_lines = self._send(dmi_cmd).splitlines()
@@ -672,10 +705,7 @@ class LinuxDriver(
cpuinfo_mdl = _clean(8)
is_container = detect_virt in _CONTAINER_VIRT
is_vm = (
detect_virt not in ("none", "")
or sys_vendor.lower() in _VM_DMI_VENDORS
)
is_vm = detect_virt not in ("none", "") or sys_vendor.lower() in _VM_DMI_VENDORS
if is_container:
return {
@@ -741,7 +771,9 @@ class LinuxDriver(
platform = self._collect_platform_info()
iface_out = self._send("ip -o link show | awk -F': ' '{print $2}' | cut -d@ -f1")
interface_list = [i.strip() for i in iface_out.splitlines() if i.strip() and i.strip() != "lo"]
interface_list = [
i.strip() for i in iface_out.splitlines() if i.strip() and i.strip() != "lo"
]
# Currently-booted kernel release, distinct from an installed-but-not-yet-
# booted newer kernel (used for kernel CVE relevance).
@@ -767,7 +799,7 @@ class LinuxDriver(
except (IndexError, ValueError):
return 0
def get_lldp_neighbors(self) -> Dict[str, List[dict[str, Any]]]:
def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]:
"""Return LLDP neighbors if lldpd is installed and currently running.
Uses ``lldpctl -f keyvalue``. Returns an empty dict when lldpd is
@@ -787,8 +819,8 @@ class LinuxDriver(
return {}
output = self._send("lldpctl -f keyvalue 2>/dev/null || true")
neighbors: Dict[str, List[dict[str, Any]]] = {}
entries: Dict[str, Dict[str, str]] = {}
neighbors: dict[str, list[dict[str, Any]]] = {}
entries: dict[str, dict[str, str]] = {}
for line in output.splitlines():
line = line.strip()
@@ -897,14 +929,16 @@ class LinuxDriver(
net.version == 6 and net.prefixlen >= 128
):
continue
networks.append({
networks.append(
{
"network": str(net),
"interface": iface_name,
"gateway": str(iface_obj.ip),
"family": family,
"prefix_length": net.prefixlen,
"vlan_id": None,
})
}
)
except ValueError:
pass
@@ -915,13 +949,13 @@ class LinuxDriver(
destination: str = "",
protocol: str = "",
longer: bool = False,
) -> Dict[str, List[dict[str, Any]]]:
) -> dict[str, list[dict[str, Any]]]:
"""Return the routing table via ``ip route show``.
OSPF routes (from FRR/Quagga) are included via ``ip route show proto ospf``
if any are present. The result is keyed by network prefix.
"""
routes: Dict[str, List[dict[str, Any]]] = {}
routes: dict[str, list[dict[str, Any]]] = {}
proto_map = {
"kernel": "connected",
@@ -935,7 +969,9 @@ class LinuxDriver(
"zebra": "zebra",
}
def _make_entry(proto: str, nexthop: str, iface: str, metric: int, network: str) -> dict[str, Any]:
def _make_entry(
proto: str, nexthop: str, iface: str, metric: int, network: str
) -> dict[str, Any]:
family = "ipv6" if (":" in network or (nexthop and ":" in nexthop)) else "ipv4"
return {
"protocol": proto,
@@ -970,7 +1006,11 @@ class LinuxDriver(
if not dest_m:
continue
raw_dest = dest_m.group(1)
network = "0.0.0.0/0" if raw_dest == "default" else ("::/0" if raw_dest == "default6" else raw_dest)
network = (
"0.0.0.0/0"
if raw_dest == "default"
else ("::/0" if raw_dest == "default6" else raw_dest)
)
if "/" not in network:
network += "/32"
@@ -998,9 +1038,14 @@ class LinuxDriver(
# FRR/Zebra enrichment via vtysh — properly attributes OSPF/BGP/RIP protocols.
# FRR installs routes into the kernel as "proto zebra"; vtysh gives the real source.
_frr_code: Dict[str, str] = {
"O": "ospf", "B": "bgp", "R": "rip", "I": "isis",
"S": "static", "K": "connected", "C": "connected",
_frr_code: dict[str, str] = {
"O": "ospf",
"B": "bgp",
"R": "rip",
"I": "isis",
"S": "static",
"K": "connected",
"C": "connected",
}
if self._send("command -v vtysh 2>/dev/null").strip():
try:
@@ -1034,27 +1079,27 @@ class LinuxDriver(
_make_entry(mapped_proto, nexthop, iface, metric, prefix)
)
except Exception:
pass
logger.debug("%s: ignored", self.hostname, exc_info=True)
return routes
def get_arp_table(self, vrf: str = "") -> List[dict[str, Any]]:
def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
"""Return the ARP/neighbour table."""
entries = []
neigh_out = self._send("ip -4 neigh show")
for line in neigh_out.splitlines():
# 192.168.1.1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE
m = re.match(
r"^([\d.]+)\s+dev\s+(\S+)\s+lladdr\s+([\da-f:]+)\s+(\S+)", line
)
m = re.match(r"^([\d.]+)\s+dev\s+(\S+)\s+lladdr\s+([\da-f:]+)\s+(\S+)", line)
if not m:
continue
entries.append({
entries.append(
{
"interface": m.group(2),
"mac": m.group(3),
"ip": m.group(1),
"age": 0.0,
})
}
)
return entries
def get_config(
@@ -1131,9 +1176,7 @@ class LinuxDriver(
sent, received = int(m.group(1)), int(m.group(2))
# Parse rtt line: "rtt min/avg/max/mdev = 0.123/0.456/0.789/0.100 ms"
rtt_m = re.search(
r"rtt .* = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+) ms", output
)
rtt_m = re.search(r"rtt .* = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+) ms", output)
results = []
for line in output.splitlines():
@@ -1141,10 +1184,12 @@ class LinuxDriver(
r"bytes from ([\d.]+).*icmp_seq=\d+ ttl=(\d+) time=([\d.]+) ms", line
)
if icmp_m:
results.append({
results.append(
{
"ip_address": icmp_m.group(1),
"rtt": float(icmp_m.group(3)),
})
}
)
return {
"success": {
@@ -1231,9 +1276,7 @@ class LinuxDriver(
return self._get_packages_apk()
if self._pkg_manager == "pacman":
return self._get_packages_pacman()
raise NotImplementedError(
f"Package manager '{self._pkg_manager}' is not supported"
)
raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")
def _get_packages_apt(self) -> list[PackageDict]:
out = self._send(
@@ -1249,7 +1292,9 @@ class LinuxDriver(
continue
name = parts[0].strip()
version = parts[1].strip()
size = int(parts[2].strip()) * 1024 if len(parts) > 2 and parts[2].strip().isdigit() else 0
size = (
int(parts[2].strip()) * 1024 if len(parts) > 2 and parts[2].strip().isdigit() else 0
)
# Debian source package (e.g. openssh-server → openssh) for OSV matching.
source_package = parts[3].strip() if len(parts) > 3 and parts[3].strip() else name
# And its version, which is a different number from this package's.
@@ -1267,7 +1312,8 @@ class LinuxDriver(
# dpkg that does not know the field at all.
source_version = parts[4].strip() if len(parts) > 4 and parts[4].strip() else version
description = parts[5].strip() if len(parts) > 5 else ""
packages.append({
packages.append(
{
"name": name,
"version": version,
"installed": True,
@@ -1276,26 +1322,32 @@ class LinuxDriver(
"source": "apt",
"source_package": source_package,
"source_version": source_version,
})
}
)
return packages
def _get_packages_rpm(self) -> list[PackageDict]:
out = self._send(
"rpm -qa --queryformat '%{NAME}\\t%{VERSION}-%{RELEASE}\\t%{SIZE}\\t%{SUMMARY}\\n' 2>/dev/null"
"rpm -qa --queryformat "
"'%{NAME}\\t%{VERSION}-%{RELEASE}\\t%{SIZE}\\t%{SUMMARY}\\n' 2>/dev/null"
)
packages: list[PackageDict] = []
for line in out.splitlines():
parts = line.split("\t", 3)
if len(parts) < 2:
continue
packages.append({
packages.append(
{
"name": parts[0].strip(),
"version": parts[1].strip(),
"installed": True,
"description": parts[3].strip() if len(parts) > 3 else "",
"size": int(parts[2].strip()) if len(parts) > 2 and parts[2].strip().isdigit() else 0,
"size": int(parts[2].strip())
if len(parts) > 2 and parts[2].strip().isdigit()
else 0,
"source": self._pkg_manager or "rpm",
})
}
)
return packages
def _get_packages_apk(self) -> list[PackageDict]:
@@ -1306,14 +1358,16 @@ class LinuxDriver(
m = re.match(r"^(\S+)-(\d[\S]*)\s*(.*)", line)
if not m:
continue
packages.append({
packages.append(
{
"name": m.group(1),
"version": m.group(2),
"installed": True,
"description": m.group(3).strip(),
"size": 0,
"source": "apk",
})
}
)
return packages
def _get_packages_pacman(self) -> list[PackageDict]:
@@ -1323,27 +1377,32 @@ class LinuxDriver(
parts = line.split(None, 1)
if len(parts) < 2:
continue
packages.append({
packages.append(
{
"name": parts[0],
"version": parts[1],
"installed": True,
"description": "",
"size": 0,
"source": "pacman",
})
}
)
return packages
def search_packages(self, query: str) -> List[dict[str, Any]]:
def search_packages(self, query: str) -> list[dict[str, Any]]:
"""Search available (installable) packages matching *query*."""
from shlex import quote as _q
safe_q = _q(query)
installed = {p["name"] for p in self.get_packages()}
packages: List[dict[str, Any]] = []
packages: list[dict[str, Any]] = []
if self._pkg_manager == "apt":
out = self._send(f"apt-cache search {safe_q} 2>/dev/null")
versions: Dict[str, str] = {}
ver_out = self._send(f"apt-cache show {safe_q} 2>/dev/null | grep -E '^(Package|Version):' || true")
versions: dict[str, str] = {}
ver_out = self._send(
f"apt-cache show {safe_q} 2>/dev/null | grep -E '^(Package|Version):' || true"
)
cur_pkg = ""
for line in ver_out.splitlines():
if line.startswith("Package:"):
@@ -1355,14 +1414,16 @@ class LinuxDriver(
continue
name, _, description = line.partition(" - ")
name = name.strip()
packages.append({
packages.append(
{
"name": name,
"version": versions.get(name, ""),
"installed": name in installed,
"description": description.strip(),
"size": 0,
"source": "apt",
})
}
)
elif self._pkg_manager in ("dnf", "yum"):
cmd = "dnf" if self._pkg_manager == "dnf" else "yum"
@@ -1373,14 +1434,16 @@ class LinuxDriver(
pkg_ver, _, description = line.partition(" : ")
name = pkg_ver.split(".")[0].strip()
version = ""
packages.append({
packages.append(
{
"name": name,
"version": version,
"installed": name in installed,
"description": description.strip(),
"size": 0,
"source": self._pkg_manager or "rpm",
})
}
)
elif self._pkg_manager == "apk":
out = self._send(f"apk search {safe_q} 2>/dev/null")
@@ -1393,14 +1456,16 @@ class LinuxDriver(
name, version, description = m.group(1), m.group(2), (m.group(3) or "")
else:
name, version, description = line, "", ""
packages.append({
packages.append(
{
"name": name,
"version": version,
"installed": name in installed,
"description": description,
"size": 0,
"source": "apk",
})
}
)
elif self._pkg_manager == "pacman":
out = self._send(f"pacman -Ss {safe_q} 2>/dev/null || true")
@@ -1415,14 +1480,16 @@ class LinuxDriver(
name = name_parts[0]
version = parts[1] if len(parts) > 1 else ""
description = lines[i + 1].strip() if i + 1 < len(lines) else ""
packages.append({
packages.append(
{
"name": name,
"version": version,
"installed": name in installed,
"description": description,
"size": 0,
"source": "pacman",
})
}
)
i += 2
continue
i += 1
@@ -1432,10 +1499,13 @@ class LinuxDriver(
def install_package(self, name: str) -> dict[str, Any]:
"""Install a package by name. Returns ``{"success": bool, "output": str}``."""
from shlex import quote as _q
safe = _q(name)
pm = self._pkg_manager
if pm == "apt":
raw = self._sudo(f"DEBIAN_FRONTEND=noninteractive apt-get install -y {safe} 2>&1 || true")
raw = self._sudo(
f"DEBIAN_FRONTEND=noninteractive apt-get install -y {safe} 2>&1 || true"
)
elif pm in ("dnf", "yum"):
raw = self._sudo(f"{pm} install -y {safe} 2>&1 || true")
elif pm == "apk":
@@ -1445,7 +1515,17 @@ class LinuxDriver(
else:
return {"success": False, "output": f"Unsupported package manager: {pm}"}
low = raw.lower()
success = not any(kw in low for kw in ("error:", "failed", "no packages", "not found", "unable to locate", "no match"))
success = not any(
kw in low
for kw in (
"error:",
"failed",
"no packages",
"not found",
"unable to locate",
"no match",
)
)
return {"success": success, "output": raw.strip()}
#: Words in a package manager's output that mean it did not do the job.
@@ -1477,6 +1557,7 @@ class LinuxDriver(
eventually break something apt would have refused to.
"""
from shlex import quote as _q
safe = _q(name)
pm = self._pkg_manager
if pm == "apt":
@@ -1538,9 +1619,7 @@ class LinuxDriver(
return self._get_updates_apk()
if self._pkg_manager == "pacman":
return self._get_updates_pacman()
raise NotImplementedError(
f"Package manager '{self._pkg_manager}' is not supported"
)
raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")
def _get_updates_apt(self) -> list[UpdateDict]:
# apt list --upgradable does not need root; avoid sudo so it works even
@@ -1550,23 +1629,31 @@ class LinuxDriver(
def _get_updates_rpm(self) -> list[UpdateDict]:
"""dnf/yum check-update: exit 100 means updates, 0 none, anything else failed."""
cmd = "dnf check-update --quiet 2>/dev/null" if self._pkg_manager == "dnf" else "yum check-update -q 2>/dev/null"
cmd = (
"dnf check-update --quiet 2>/dev/null"
if self._pkg_manager == "dnf"
else "yum check-update -q 2>/dev/null"
)
output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
if status not in (0, 100):
raise RuntimeError(f"{self._pkg_manager} check-update failed (exit {status}): {output[-200:]}")
raise RuntimeError(
f"{self._pkg_manager} check-update failed (exit {status}): {output[-200:]}"
)
security = self._rpm_security_names()
updates: list[UpdateDict] = []
for line in output.splitlines():
parts = line.split()
if len(parts) >= 2 and not line.startswith(" ") and "." in parts[0]:
name = parts[0].rsplit(".", 1)[0]
updates.append({
updates.append(
{
"name": name,
"current_version": "",
"new_version": parts[1],
"origin": parts[2] if len(parts) >= 3 else None,
"security": None if security is None else name in security,
})
}
)
return updates
def _rpm_security_names(self) -> set[str] | None:
@@ -1604,11 +1691,13 @@ class LinuxDriver(
# openssh-9.3_p2-r3 < 9.3_p2-r4
m = re.match(r"^(\S+)-(\S+)\s+<\s+(\S+)", line)
if m:
updates.append({
updates.append(
{
"name": m.group(1),
"current_version": m.group(2),
"new_version": m.group(3),
})
}
)
return updates
def _get_updates_pacman(self) -> list[UpdateDict]:
@@ -1618,11 +1707,13 @@ class LinuxDriver(
# openssh 9.3p2-1 -> 9.4p1-1
m = re.match(r"^(\S+)\s+(\S+)\s+->\s+(\S+)", line)
if m:
updates.append({
updates.append(
{
"name": m.group(1),
"current_version": m.group(2),
"new_version": m.group(3),
})
}
)
return updates
# ------------------------------------------------------------------
@@ -1630,9 +1721,9 @@ class LinuxDriver(
# ------------------------------------------------------------------
# Allowlist for package names – same pattern used by napalm-proxmox
_PKG_NAME_RE = re.compile(r'^[a-zA-Z0-9_\-\+\.]+$')
_PKG_NAME_RE = re.compile(r"^[a-zA-Z0-9_\-\+\.]+$")
def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
def apply_updates(self, packages: list[str]) -> ApplyUpdatesResultDict:
"""Upgrade *packages* (or all pending updates when the list is empty).
Package names are validated against ``^[a-zA-Z0-9_\\-\\+\\.]+$`` before
@@ -1650,30 +1741,27 @@ class LinuxDriver(
return self._apply_updates_apk(packages)
if self._pkg_manager == "pacman":
return self._apply_updates_pacman(packages)
raise NotImplementedError(
f"Package manager '{self._pkg_manager}' is not supported"
)
raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")
def _apply_updates_apt(self, packages: List[str]) -> ApplyUpdatesResultDict:
def _apply_updates_apt(self, packages: list[str]) -> ApplyUpdatesResultDict:
pkg_args = " ".join(packages) if packages else "--with-new-pkgs"
cmd = (
"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y "
f"{pkg_args} 2>&1"
if packages else
"DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1"
f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1"
if packages
else "DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1"
)
try:
output = self._sudo(cmd, read_timeout=600)
success = not re.search(r'^E:', output, re.MULTILINE)
success = not re.search(r"^E:", output, re.MULTILINE)
result: ApplyUpdatesResultDict = {"success": success, "output": output}
if not success:
m = re.search(r'^E:.*', output, re.MULTILINE)
m = re.search(r"^E:.*", output, re.MULTILINE)
result["error"] = m.group(0) if m else "apt-get exited with errors"
return result
except Exception as exc:
return {"success": False, "output": "", "error": str(exc)}
def _apply_updates_rpm(self, packages: List[str]) -> ApplyUpdatesResultDict:
def _apply_updates_rpm(self, packages: list[str]) -> ApplyUpdatesResultDict:
bin_ = self._pkg_manager # "dnf" or "yum"
if packages:
pkg_args = " ".join(packages)
@@ -1684,16 +1772,16 @@ class LinuxDriver(
output = self._sudo(cmd, read_timeout=600)
# dnf/yum signal failure via "Error:" lines or non-zero exit;
# since we can't check the exit code directly, look for error markers.
success = not re.search(r'^Error:', output, re.MULTILINE | re.IGNORECASE)
success = not re.search(r"^Error:", output, re.MULTILINE | re.IGNORECASE)
result: ApplyUpdatesResultDict = {"success": success, "output": output}
if not success:
m = re.search(r'^Error:.*', output, re.MULTILINE | re.IGNORECASE)
m = re.search(r"^Error:.*", output, re.MULTILINE | re.IGNORECASE)
result["error"] = m.group(0) if m else f"{bin_} exited with errors"
return result
except Exception as exc:
return {"success": False, "output": "", "error": str(exc)}
def _apply_updates_apk(self, packages: List[str]) -> ApplyUpdatesResultDict:
def _apply_updates_apk(self, packages: list[str]) -> ApplyUpdatesResultDict:
if packages:
pkg_args = " ".join(packages)
cmd = f"apk upgrade {pkg_args} 2>&1"
@@ -1709,7 +1797,7 @@ class LinuxDriver(
except Exception as exc:
return {"success": False, "output": "", "error": str(exc)}
def _apply_updates_pacman(self, packages: List[str]) -> ApplyUpdatesResultDict:
def _apply_updates_pacman(self, packages: list[str]) -> ApplyUpdatesResultDict:
if packages:
pkg_args = " ".join(packages)
cmd = f"pacman --noconfirm -S {pkg_args} 2>&1"
@@ -1753,12 +1841,14 @@ class LinuxDriver(
m = re.match(r"^\s*\[\s*([+\-?])\s*\]\s+(\S+)", line)
if not m:
continue
services.append({
services.append(
{
"name": m.group(2),
"running": m.group(1) == "+",
"enabled": False,
"pid": 0,
})
}
)
return services
# ------------------------------------------------------------------
@@ -1770,20 +1860,8 @@ class LinuxDriver(
passwd_out = self._send("getent passwd 2>/dev/null || cat /etc/passwd")
groups_out = self._send("getent group 2>/dev/null || cat /etc/group")
# Build uid→[group] map from /etc/group
uid_to_groups: Dict[int, List[str]] = {}
for line in groups_out.splitlines():
parts = line.split(":")
if len(parts) < 4:
continue
gname = parts[0]
members = [m.strip() for m in parts[3].split(",") if m.strip()]
for member in members:
# We'll convert username→uid below; collect by username first
uid_to_groups.setdefault(-1, []) # placeholder
# Simpler: collect username→groups, then join with passwd
username_to_groups: Dict[str, List[str]] = {}
# username → supplementary groups, joined with passwd below
username_to_groups: dict[str, list[str]] = {}
for line in groups_out.splitlines():
parts = line.split(":")
if len(parts) < 4:
@@ -1803,14 +1881,16 @@ class LinuxDriver(
uid, gid = int(uid_s), int(gid_s)
except ValueError:
continue
users.append({
users.append(
{
"username": username,
"uid": uid,
"gid": gid,
"home": home,
"shell": shell,
"groups": username_to_groups.get(username, []),
})
}
)
return users
# ------------------------------------------------------------------
@@ -1846,7 +1926,8 @@ class LinuxDriver(
command = parts[10]
except (ValueError, IndexError):
continue
processes.append({
processes.append(
{
"pid": pid,
"ppid": ppid,
"user": user,
@@ -1858,7 +1939,8 @@ class LinuxDriver(
"state": state,
"started": started,
"command": command,
})
}
)
return processes
# ------------------------------------------------------------------
@@ -1897,9 +1979,7 @@ class LinuxDriver(
return jobs
@staticmethod
def _parse_cron_line(
line: str, source_user: str, has_user_field: bool
) -> Optional[CronJobDict]:
def _parse_cron_line(line: str, source_user: str, has_user_field: bool) -> CronJobDict | None:
"""Parse a single crontab line; returns ``None`` for comments/blanks."""
stripped = line.strip()
# Remove trailing comment
@@ -1936,12 +2016,11 @@ class LinuxDriver(
# ── Device actions ────────────────────────────────────────────────────────
def get_snmp_config(self) -> Optional[SNMPConfigDict]:
def get_snmp_config(self) -> SNMPConfigDict | None:
"""Return SNMP agent config if snmpd is installed and running."""
try:
running = (
self._send("systemctl is-active snmpd 2>/dev/null || true").strip()
== "active"
self._send("systemctl is-active snmpd 2>/dev/null || true").strip() == "active"
)
if not running:
return None
@@ -1968,7 +2047,7 @@ class LinuxDriver(
community = parts[3]
break
except Exception:
pass
logger.debug("%s: ignored", self.hostname, exc_info=True)
# Detect port override
try:
@@ -1976,11 +2055,11 @@ class LinuxDriver(
"grep -E '^agentAddress' /etc/snmp/snmpd.conf 2>/dev/null | head -1"
).strip()
if port_line:
m = re.search(r':(\d+)', port_line)
m = re.search(r":(\d+)", port_line)
if m:
port = int(m.group(1))
except Exception:
pass
logger.debug("%s: ignored", self.hostname, exc_info=True)
return SNMPConfigDict(running=True, community=community, port=port, version="2c")
except Exception as exc:
@@ -2067,7 +2146,9 @@ class LinuxDriver(
if not _USER_NAME.match(user):
raise ValueError(f"not a user name: {user!r}")
return self._root_outcome(
f"usermod -aG docker {user}", 30, f"{user} is in the docker group from the next login"
f"usermod -aG docker {user}",
30,
f"{user} is in the docker group from the next login",
)
return run_fix_steps([("grant", grant)])
@@ -2092,15 +2173,19 @@ class LinuxDriver(
raw = self._sudo("wg show all dump 2>&1 || true")
_perm_errors = (
"operation not permitted", "permission denied",
"a password is required", "a terminal is required",
"command not found", "not found", "no such file",
"operation not permitted",
"permission denied",
"a password is required",
"a terminal is required",
"command not found",
"not found",
"no such file",
)
has_peer_data = raw and not any(e in raw.lower() for e in _perm_errors)
if has_peer_data:
# Build listen-port map from interface lines (5 tab-separated fields)
listen_ports: Dict[str, str] = {}
listen_ports: dict[str, str] = {}
for line in raw.splitlines():
parts = line.split("\t")
if len(parts) == 5:
@@ -2112,8 +2197,17 @@ class LinuxDriver(
parts = line.split("\t")
if len(parts) != 9:
continue
(iface, pubkey, _psk, endpoint, allowed_ips,
latest_hs, rx_bytes, tx_bytes, _keepalive) = parts
(
iface,
pubkey,
_psk,
endpoint,
allowed_ips,
latest_hs,
rx_bytes,
tx_bytes,
_keepalive,
) = parts
iface = iface.strip()
pubkey = pubkey.strip()
@@ -2172,13 +2266,14 @@ class LinuxDriver(
try:
import json as _json
iface_list = _json.loads(ip_raw)
except Exception:
return tunnels
# /proc/net/dev for total RX/TX bytes per interface
proc_dev = self._send("cat /proc/net/dev 2>/dev/null || true")
proc_bytes: Dict[str, tuple] = {}
proc_bytes: dict[str, tuple] = {}
for line in proc_dev.splitlines()[2:]:
line = line.strip()
if ":" not in line:
napalm_linux/py.typed
View File
Whitespace-only changes.
pyproject.toml
+47 -13
View File
@@ -1,14 +1,14 @@
[build-system]
requires = ["setuptools>=68", "wheel"]
requires = ["setuptools>=77"]
build-backend = "setuptools.build_meta"
[project]
name = "napalm-linux"
version = "0.14.0"
version = "0.15.0"
description = "NAPALM driver for generic Linux systems via SSH"
readme = "README.md"
requires-python = ">=3.9"
license = { text = "Apache-2.0" }
requires-python = ">=3.10"
license = "Apache-2.0"
authors = [
{ name = "Christian Manivong", email = "christian@manivong.de" },
]
@@ -24,13 +24,12 @@ classifiers = [
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"Intended Audience :: System Administrators",
"License :: OSI Approved :: Apache Software License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: System :: Networking",
"Topic :: System :: Systems Administration",
"Typing :: Typed",
@@ -44,23 +43,58 @@ dependencies = [
[project.optional-dependencies]
dev = [
"pytest",
"pytest-cov",
"black",
"ruff",
"mypy",
"pytest>=8",
"pytest-cov>=5",
"ruff>=0.6",
"mypy>=1.11",
]
[project.entry-points."napalm.drivers"]
linux = "napalm_linux:LinuxDriver"
[project.urls]
Repository = "https://github.com/chrismanivong/napalm-linux"
Repository = "https://git.netork.io/NAPALM/napalm-linux"
Issues = "https://git.netork.io/NAPALM/napalm-linux/issues"
Changelog = "https://git.netork.io/NAPALM/napalm-linux/src/branch/main/CHANGELOG.md"
[tool.setuptools.packages.find]
where = ["."]
include = ["napalm_linux*"]
[tool.setuptools.package-data]
napalm_linux = ["py.typed"]
[tool.ruff]
line-length = 100
target-version = "py39"
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
# napalm itself is untyped, so every driver subclasses Any.
disallow_subclassing_any = false
[[tool.mypy.overrides]]
module = ["napalm.*", "netmiko.*", "paramiko.*"]
ignore_missing_imports = true
[[tool.mypy.overrides]]
# Not yet under strict mypy (quality baseline); may only shrink.
module = ["napalm_linux.linux"]
ignore_errors = true
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.coverage.run]
source = ["napalm_linux"]
[tool.coverage.report]
fail_under = 57
tests/quality_baseline.json
+28
View File
@@ -0,0 +1,28 @@
{
"coverage_floor": 57.0,
"function_complexity": {
"napalm_linux/linux.py::LinuxDriver._collect_platform_info": 24,
"napalm_linux/linux.py::LinuxDriver._parse_cron_line": 12,
"napalm_linux/linux.py::LinuxDriver.get_networks": 13,
"napalm_linux/linux.py::LinuxDriver.get_route_to": 23,
"napalm_linux/linux.py::LinuxDriver.get_snmp_config": 13,
"napalm_linux/linux.py::LinuxDriver.get_vpn_tunnels": 28,
"napalm_linux/linux.py::LinuxDriver.search_packages": 26
},
"function_lines": {
"napalm_linux/linux.py::LinuxDriver._collect_platform_info": 106,
"napalm_linux/linux.py::LinuxDriver.get_networks": 52,
"napalm_linux/linux.py::LinuxDriver.get_route_to": 138,
"napalm_linux/linux.py::LinuxDriver.get_vpn_tunnels": 152,
"napalm_linux/linux.py::LinuxDriver.ping": 52,
"napalm_linux/linux.py::LinuxDriver.search_packages": 106,
"napalm_linux/linux.py::LinuxDriver.uninstall_package": 51
},
"mypy_relaxed": [
"napalm_linux.linux"
],
"python_file_lines": {
"napalm_linux/linux.py": 2307
},
"untyped_public": []
}
tests/test_command_channel.py
+3 -1
View File
@@ -70,7 +70,9 @@ class FakeTransport:
def _driver(*, root=False, sudo_password=None):
driver = LinuxDriver("h", "u", "p", optional_args={"sudo_password": sudo_password})
transport = FakeTransport()
driver._device = SimpleNamespace(remote_conn_pre=SimpleNamespace(get_transport=lambda: transport))
driver._device = SimpleNamespace(
remote_conn_pre=SimpleNamespace(get_transport=lambda: transport)
)
driver._root = root
return driver, transport
tests/test_fix_templates.py
+4 -2
View File
@@ -248,7 +248,7 @@ def test_every_annotation_resolves():
import inspect
import typing
for name, member in inspect.getmembers(LinuxDriver, inspect.isfunction):
for _name, member in inspect.getmembers(LinuxDriver, inspect.isfunction):
if member.__module__ == "napalm_linux.linux":
typing.get_type_hints(member)
@@ -263,7 +263,9 @@ class TestManagementPeer:
return d
def test_the_client_side_of_ssh_connection(self):
assert self._driver("10.7.224.20 51234 10.0.0.5 22\n").get_management_peer() == "10.7.224.20"
assert (
self._driver("10.7.224.20 51234 10.0.0.5 22\n").get_management_peer() == "10.7.224.20"
)
def test_an_ipv6_client(self):
assert self._driver("fd00::20 51234 fd00::5 22\n").get_management_peer() == "fd00::20"
tests/test_linux.py
+181 -65
View File
@@ -1,9 +1,10 @@
"""Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed)."""
import re
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import MagicMock, patch
from napalm_linux.linux import (
_YUM_SECURITY_COMMAND,
LinuxDriver,
@@ -11,7 +12,6 @@ from napalm_linux.linux import (
_split_status,
)
# ---------------------------------------------------------------------------
# Fixture – driver without a real connection
# ---------------------------------------------------------------------------
@@ -327,11 +327,7 @@ PASSWD_OUT = (
"daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin\n"
)
GROUP_OUT = (
"sudo:x:27:admin\n"
"docker:x:999:admin\n"
"adm:x:4:admin\n"
)
GROUP_OUT = "sudo:x:27:admin\ndocker:x:999:admin\nadm:x:4:admin\n"
def test_get_os_users(driver):
@@ -477,8 +473,7 @@ DNF_UPGRADE_SUCCESS = (
)
DNF_UPGRADE_ERROR = (
"Last metadata expiration check: 0:01:23 ago.\n"
"Error: No match for argument: nonexistent-pkg\n"
"Last metadata expiration check: 0:01:23 ago.\nError: No match for argument: nonexistent-pkg\n"
)
@@ -532,6 +527,7 @@ def test_type_label_is_linux():
def test_type_label_overrides_os_driver():
from napalm_device_types import OSDriver
assert OSDriver.TYPE_LABEL == "OS"
assert LinuxDriver.TYPE_LABEL != OSDriver.TYPE_LABEL
@@ -555,18 +551,31 @@ def _dmi_output(
# Mirrors the real shell output: DMIBEGIN sentinel followed by 9 fields.
# The sentinel prevents _send()'s .strip() from eating leading blank lines
# (which occur on ARM where all DMI files are absent).
return "\n".join([
return "\n".join(
[
"DMIBEGIN",
sys_vendor, product_name, product_version, product_serial,
product_uuid, detect_virt, dt_model, cpuinfo_serial, cpuinfo_model,
])
sys_vendor,
product_name,
product_version,
product_serial,
product_uuid,
detect_virt,
dt_model,
cpuinfo_serial,
cpuinfo_model,
]
)
class TestCollectPlatformInfo:
def test_baremetal_dell(self, driver):
raw = _dmi_output(
"Dell Inc.", "PowerEdge R720", "Not Specified", "ABC123",
"8a2e3f00-dead-beef-0000-123456789abc", "none",
"Dell Inc.",
"PowerEdge R720",
"Not Specified",
"ABC123",
"8a2e3f00-dead-beef-0000-123456789abc",
"none",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -577,8 +586,12 @@ class TestCollectPlatformInfo:
def test_baremetal_lenovo_product_version_preferred(self, driver):
raw = _dmi_output(
"LENOVO", "10M8000VUS", "ThinkCentre M910x", "MP1234",
"8a2e3f00-dead-beef-0000-123456789abc", "none",
"LENOVO",
"10M8000VUS",
"ThinkCentre M910x",
"MP1234",
"8a2e3f00-dead-beef-0000-123456789abc",
"none",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -589,8 +602,12 @@ class TestCollectPlatformInfo:
def test_vm_kvm(self, driver):
raw = _dmi_output(
"QEMU", "Standard PC (i440FX + PIIX, 1996)", "pc-i440fx-9.1", "",
"4c4c4544-0000-2010-8020-b4c04f534a31", "kvm",
"QEMU",
"Standard PC (i440FX + PIIX, 1996)",
"pc-i440fx-9.1",
"",
"4c4c4544-0000-2010-8020-b4c04f534a31",
"kvm",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -601,8 +618,12 @@ class TestCollectPlatformInfo:
def test_vm_vmware(self, driver):
raw = _dmi_output(
"VMware, Inc.", "VMware Virtual Platform", "None", "VMware-42 12 34 56",
"4244560c-dead-beef-0000-abcdef123456", "vmware",
"VMware, Inc.",
"VMware Virtual Platform",
"None",
"VMware-42 12 34 56",
"4244560c-dead-beef-0000-abcdef123456",
"vmware",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -613,8 +634,12 @@ class TestCollectPlatformInfo:
def test_vm_hyperv(self, driver):
raw = _dmi_output(
"Microsoft Corporation", "Virtual Machine", "Hyper-V UEFI Release v4.1", "",
"7C5B4B1F-1234-5678-ABCD-000000000001", "microsoft",
"Microsoft Corporation",
"Virtual Machine",
"Hyper-V UEFI Release v4.1",
"",
"7C5B4B1F-1234-5678-ABCD-000000000001",
"microsoft",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -625,8 +650,12 @@ class TestCollectPlatformInfo:
def test_junk_dmi_values_filtered(self, driver):
raw = _dmi_output(
"To Be Filled By O.E.M.", "To Be Filled By O.E.M.", "Not Specified",
"To Be Filled By O.E.M.", "", "none",
"To Be Filled By O.E.M.",
"To Be Filled By O.E.M.",
"Not Specified",
"To Be Filled By O.E.M.",
"",
"none",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -637,8 +666,12 @@ class TestCollectPlatformInfo:
def test_vm_kvm_fallback_via_dmi_when_detect_virt_unavailable(self, driver):
# systemd-detect-virt returns "none" (not installed), sys_vendor reveals QEMU
raw = _dmi_output(
"QEMU", "Standard PC (i440FX + PIIX, 1996)", "", "",
"4c4c4544-0000-2010-8020-b4c04f534a31", "none",
"QEMU",
"Standard PC (i440FX + PIIX, 1996)",
"",
"",
"4c4c4544-0000-2010-8020-b4c04f534a31",
"none",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -648,8 +681,12 @@ class TestCollectPlatformInfo:
def test_container_docker(self, driver):
raw = _dmi_output(
"QEMU", "Standard PC (i440FX + PIIX, 1996)", "", "",
"4c4c4544-0000-2010-8020-b4c04f534a31", "docker",
"QEMU",
"Standard PC (i440FX + PIIX, 1996)",
"",
"",
"4c4c4544-0000-2010-8020-b4c04f534a31",
"docker",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -674,7 +711,12 @@ class TestCollectPlatformInfo:
def test_arm_device_tree_fallback(self, driver):
raw = _dmi_output(
"", "", "", "", "", "none",
"",
"",
"",
"",
"",
"none",
dt_model="Raspberry Pi 4 Model B Rev 1.4",
cpuinfo_serial="100000002a6d96dc",
cpuinfo_model="Raspberry Pi 4 Model B Rev 1.4",
@@ -688,7 +730,12 @@ class TestCollectPlatformInfo:
def test_arm_cpuinfo_model_when_no_dt(self, driver):
raw = _dmi_output(
"", "", "", "", "", "none",
"",
"",
"",
"",
"",
"none",
dt_model="",
cpuinfo_serial="0000000012345678",
cpuinfo_model="Raspberry Pi 3 Model B Plus Rev 1.3",
@@ -713,7 +760,12 @@ class TestCollectPlatformInfo:
# d="none\nnone" which shifted subsequent lines off by one.
# With the ${d:-none} fix this no longer happens.
raw = _dmi_output(
"", "", "", "", "", "none",
"",
"",
"",
"",
"",
"none",
dt_model="",
cpuinfo_serial="0000000012345678",
cpuinfo_model="Raspberry Pi 3 Model B Rev 1.2",
@@ -726,8 +778,12 @@ class TestCollectPlatformInfo:
def test_arm_fallback_ignored_when_dmi_present(self, driver):
raw = _dmi_output(
"Dell Inc.", "PowerEdge R720", "Not Specified", "XYZ999",
"8a2e3f00-dead-beef-0000-123456789abc", "none",
"Dell Inc.",
"PowerEdge R720",
"Not Specified",
"XYZ999",
"8a2e3f00-dead-beef-0000-123456789abc",
"none",
dt_model="some-dt-model",
cpuinfo_serial="deadbeef",
cpuinfo_model="some cpuinfo model",
@@ -744,7 +800,9 @@ class TestCollectPlatformInfo:
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("model,expected", [
@pytest.mark.parametrize(
"model,expected",
[
("Raspberry Pi 4 Model B Rev 1.4", "Raspberry Pi Foundation"),
("Raspberry Pi 3 Model B Rev 1.2", "Raspberry Pi Foundation"),
("ODROID-N2L", "Hardkernel"),
@@ -757,7 +815,8 @@ class TestCollectPlatformInfo:
("BeagleBone Black", "BeagleBoard.org"),
("Unknown Board 1.0", "Unknown Board"),
("SomeSingleWordBoard", "SomeSingleWordBoard"),
])
],
)
def test_arm_vendor_from_model(model, expected):
assert _arm_vendor_from_model(model) == expected
@@ -768,11 +827,27 @@ def test_arm_vendor_from_model(model, expected):
def test_get_facts_baremetal_vendor_model_serial(driver):
platform = {"vendor": "Dell Inc.", "model": "PowerEdge R720", "serial": "ABC123", "is_vm": False}
with patch.object(driver, "_collect_platform_info", return_value=platform), \
patch.object(driver, "_parse_uptime", return_value=86400), \
patch.object(driver, "_send", side_effect=["myhost", "myhost.example.com", "Debian GNU/Linux 12", "eth0\neth1",
"6.1.0-18-amd64"]):
platform = {
"vendor": "Dell Inc.",
"model": "PowerEdge R720",
"serial": "ABC123",
"is_vm": False,
}
with (
patch.object(driver, "_collect_platform_info", return_value=platform),
patch.object(driver, "_parse_uptime", return_value=86400),
patch.object(
driver,
"_send",
side_effect=[
"myhost",
"myhost.example.com",
"Debian GNU/Linux 12",
"eth0\neth1",
"6.1.0-18-amd64",
],
),
):
facts = driver.get_facts()
assert facts["vendor"] == "Dell Inc."
assert facts["model"] == "PowerEdge R720"
@@ -785,13 +860,20 @@ def test_get_facts_baremetal_vendor_model_serial(driver):
def test_get_facts_vm_kvm(driver):
platform = {
"vendor": "KVM", "model": "Virtual Machine",
"serial": "4c4c4544-0000-2010-8020-b4c04f534a31", "is_vm": True,
"vendor": "KVM",
"model": "Virtual Machine",
"serial": "4c4c4544-0000-2010-8020-b4c04f534a31",
"is_vm": True,
}
with patch.object(driver, "_collect_platform_info", return_value=platform), \
patch.object(driver, "_parse_uptime", return_value=3600), \
patch.object(driver, "_send", side_effect=["vmhost", "vmhost.local", "Ubuntu 22.04 LTS", "eth0",
"5.15.0-91-generic"]):
with (
patch.object(driver, "_collect_platform_info", return_value=platform),
patch.object(driver, "_parse_uptime", return_value=3600),
patch.object(
driver,
"_send",
side_effect=["vmhost", "vmhost.local", "Ubuntu 22.04 LTS", "eth0", "5.15.0-91-generic"],
),
):
facts = driver.get_facts()
assert facts["vendor"] == "KVM"
assert facts["model"] == "Virtual Machine"
@@ -800,9 +882,15 @@ def test_get_facts_vm_kvm(driver):
def test_get_facts_fallback_vendor_when_dmi_empty(driver):
platform = {"vendor": "", "model": "", "serial": "", "is_vm": False}
with patch.object(driver, "_collect_platform_info", return_value=platform), \
patch.object(driver, "_parse_uptime", return_value=0), \
patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0", "6.6.7-0-lts"]):
with (
patch.object(driver, "_collect_platform_info", return_value=platform),
patch.object(driver, "_parse_uptime", return_value=0),
patch.object(
driver,
"_send",
side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0", "6.6.7-0-lts"],
),
):
facts = driver.get_facts()
assert facts["vendor"] == "Linux" # fallback to VENDOR class attribute
@@ -813,7 +901,12 @@ class TestNoDockerOfItsOwn:
only the fix that lets the login user use Docker stays."""
def test_the_cli_methods_are_gone(self):
for name in ("get_docker_info", "get_docker_outdated", "reconstruct_docker_run", "_docker_bin"):
for name in (
"get_docker_info",
"get_docker_outdated",
"reconstruct_docker_run",
"_docker_bin",
):
assert not hasattr(LinuxDriver, name), name
def test_the_engine_access_and_the_permission_fix_stay(self):
@@ -1019,7 +1112,9 @@ class TestReadToTheEnd:
def test_the_echoed_command_does_not_count_as_the_marker(self, driver):
"""Its literal ``$?`` is no number."""
channel = _Channel(["sudo true; echo __NETORK_RC=$?\n", "__NETORK_RC=0\nchris@vault-01:~$ "])
channel = _Channel(
["sudo true; echo __NETORK_RC=$?\n", "__NETORK_RC=0\nchris@vault-01:~$ "]
)
driver._device = channel
assert _split_status(driver._send(_TRUE_WITH_STATUS))[1] == 0
@@ -1043,7 +1138,7 @@ class TestReadToTheEnd:
)
def test_every_status_marker_is_waited_for(self, driver, command):
marker = re.search(r"echo (__[A-Z_]+=)", command).group(1)
channel = _Channel([f"x <a@b>\n", f"{marker}0\nchris@host:~$ "])
channel = _Channel(["x <a@b>\n", f"{marker}0\nchris@host:~$ "])
driver._device = channel
output = driver._send(command)
@@ -1167,7 +1262,11 @@ def _kernel_wire(report: str) -> str:
import base64
import gzip
return "KFACTS_BEGIN\n" + base64.encodebytes(gzip.compress(report.encode())).decode() + "KFACTS_END"
return (
"KFACTS_BEGIN\n"
+ base64.encodebytes(gzip.compress(report.encode())).decode()
+ "KFACTS_END"
)
def test_get_kernel_facts_carries_the_shared_command_across(driver):
@@ -1369,12 +1468,16 @@ class TestAvailableUpdates:
driver._pkg_manager = "dnf"
driver._device.send_command.side_effect = [
"0", # id -u
("openssl-libs.x86_64 1:3.1.4-2.fc40 updates\n"
"vim-enhanced.x86_64 2:9.1.083-1.fc40 updates\n__NETORK_RC=100"),
(
"openssl-libs.x86_64 1:3.1.4-2.fc40 updates\n"
"vim-enhanced.x86_64 2:9.1.083-1.fc40 updates\n__NETORK_RC=100"
),
# DNF_SECURITY_COMMAND pipes through cat: the outer status is cat's, dnf's own
# is in the __DNF_RC marker.
("FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"),
(
"FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"
),
]
updates = {u["name"]: u for u in driver.get_available_updates()}
@@ -1402,10 +1505,14 @@ class TestAvailableUpdates:
driver._pkg_manager = "yum"
driver._device.send_command.side_effect = [
"0", # id -u
("openssl-libs.x86_64 1:1.0.2k-26.el7_9 updates\n"
"vim-enhanced.x86_64 2:7.4.629-8.el7_9 updates\n__NETORK_RC=100"),
("RHSA-2023:3556 Important/Sec. openssl-libs-1:1.0.2k-26.el7_9.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"),
(
"openssl-libs.x86_64 1:1.0.2k-26.el7_9 updates\n"
"vim-enhanced.x86_64 2:7.4.629-8.el7_9 updates\n__NETORK_RC=100"
),
(
"RHSA-2023:3556 Important/Sec. openssl-libs-1:1.0.2k-26.el7_9.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"
),
]
updates = {u["name"]: u for u in driver.get_available_updates()}
@@ -1449,7 +1556,10 @@ class TestAvailableUpdates:
def test_dnf_that_failed_raises(self, driver):
driver._pkg_manager = "dnf"
driver._device.send_command.side_effect = ["0", "Error: Failed to download metadata\n__NETORK_RC=1"]
driver._device.send_command.side_effect = [
"0",
"Error: Failed to download metadata\n__NETORK_RC=1",
]
with pytest.raises(RuntimeError):
driver.get_available_updates()
@@ -1461,7 +1571,10 @@ class TestRefreshAvailableUpdates:
def test_apt_refreshes_its_index_as_root(self, driver):
driver._pkg_manager = "apt"
driver._device.send_command.side_effect = ["0", "Hit:1 http://archive.ubuntu.com noble InRelease\n__NETORK_RC=0"]
driver._device.send_command.side_effect = [
"0",
"Hit:1 http://archive.ubuntu.com noble InRelease\n__NETORK_RC=0",
]
result = driver.refresh_available_updates()
@@ -1470,7 +1583,10 @@ class TestRefreshAvailableUpdates:
def test_without_a_sudo_password_it_never_waits_for_one(self, driver):
driver._pkg_manager = "apt"
driver._device.send_command.side_effect = ["1000", "sudo: a password is required\n__NETORK_RC=1"]
driver._device.send_command.side_effect = [
"1000",
"sudo: a password is required\n__NETORK_RC=1",
]
result = driver.refresh_available_updates()
tests/test_ssh_host_key.py
+4 -2
View File
@@ -27,7 +27,9 @@ OTHER_KEY = paramiko.ECDSAKey.generate()
def _device() -> MagicMock:
device = MagicMock()
device.send_command.return_value = "/usr/bin/apt"
device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY
device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = (
SERVER_KEY
)
return device
@@ -76,7 +78,7 @@ def test_a_changed_key_is_refused():
try:
raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
except paramiko.SSHException as exc: # what netmiko does with it
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}")
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") from exc
driver = _driver(seen, SshHostKey.of(OTHER_KEY))
with patch("napalm_linux.linux.ConnectHandler", side_effect=refuse):
tests/test_sudo_off_the_command_line.py
+1 -1
View File
@@ -30,7 +30,7 @@ class Channel:
"""An exec channel answering with *output* and *status*; it refuses a line
that holds the password, which is what ``ps`` would show."""
def __init__(self, host: "Transport"):
def __init__(self, host: Transport):
self.host = host
self.command = ""
self.stdin = b""