Compare commits

...
Author SHA1 Message Date
christianmanivong 23c352aa8a feat: OpenSshServerMixin takes flagged algorithms out of what sshd offers
CI / test (3.10) (push) Successful in 23s
CI / test (3.11) (push) Successful in 22s
CI / test (3.12) (push) Successful in 23s
CI / test (3.10) (pull_request) Successful in 23s
CI / test (3.11) (pull_request) Successful in 22s
CI / test (3.12) (pull_request) Successful in 23s
ssh-audit names algorithms an SSH server should stop offering. Taking them
out is the same on every OpenSSH host, so it lives here once: a driver
mixes in OpenSshServerMixin and supplies run_command.

harden_ssh_server reads the lists sshd really uses (sshd -T), removes the
requested algorithms and writes the rest to
/etc/ssh/sshd_config.d/00-netork-ssh-hardening.conf -- or, where
sshd_config has no Include of that directory, rewrites sshd_config after
a copy. It never removes the last algorithm of a list, nor, given the
managing client's lists, the last one that client speaks. sshd -t must
accept the result and sshd -T must show it; after the reload the caller's
verify opens a new session, and if that fails the still-open one puts the
previous state back and reloads again.

Bumps to 4.5.0. NetOrk/netork#1027
2026-10-11 23:12:28 +02:00
christianmanivong a710a85de6 Merge pull request 'feat: report how full /, /boot and /var are in the host status' (#28) from fix/volume-almost-full-862 into main
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 29s
CI / test (3.12) (push) Successful in 31s
2026-10-11 08:33:03 +00:00
christianmanivong c947250179 feat: report how full /, /boot and /var are in the host status
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 23s
CI / test (3.12) (push) Successful in 23s
CI / test (3.10) (pull_request) Successful in 23s
CI / test (3.11) (pull_request) Successful in 26s
CI / test (3.12) (pull_request) Successful in 25s
A Proxmox host's root filesystem stood at 94 % with about 400 MB free for a
week; a kernel upgrade failed on it with "No space left on device" and left
dpkg halfway, and nothing had said the disk was full (NetOrk/netork#862).

HOST_STATUS_COMMAND now also runs POSIX `df -Pk / /boot /var`, without root.
get_host_status() returns `filesystems`: each filesystem once (a directory
that is not a mount of its own is the filesystem it lives on), with its
mountpoint, total, used and available bytes, or None when df said nothing.
`available` is what df says an ordinary user can still write, which on ext4
is less than total minus used. How full is too full is the caller's decision.

Version 4.4.0.
2026-10-11 09:06:00 +02:00
christianmanivong ff6dbe0206 Merge pull request 'fix: tell unattended-upgrade from Ubuntu's idle shutdown helper' (#27) from fix/unattended-upgrade-shutdown into main
CI / test (3.10) (push) Successful in 24s
CI / test (3.11) (push) Successful in 26s
CI / test (3.12) (push) Successful in 26s
2026-10-08 15:33:26 +00:00
christianmanivong 7aa731ad34 fix: tell unattended-upgrade from Ubuntu's idle shutdown helper
CI / test (3.10) (push) Successful in 24s
CI / test (3.11) (push) Successful in 23s
CI / test (3.12) (push) Successful in 24s
CI / test (3.10) (pull_request) Successful in 24s
CI / test (3.11) (pull_request) Successful in 22s
CI / test (3.12) (pull_request) Successful in 25s
On every Ubuntu host with unattended-upgrades, get_host_status() read the
package manager as busy, so an interrupted dpkg was never reported there.
Ubuntu keeps unattended-upgrade-shutdown --wait-for-signal running all the
time; its process name is cut to 15 characters, unattended-upgr, the same
as a real run's.

BUSY_CHECK now matches dpkg, apt, apt-get and aptitude by name and
unattended-upgrade by its command line. The tests run it through sh with a
stand-in ps, and check that the command does not match itself: over an exec
channel it is the command line of the shell that runs it.

Closes #26
2026-10-08 17:30:44 +02:00
christianmanivong 72679ff4f8 Merge pull request 'feat: report a dpkg left halfway and when the host's own automatic updates last succeeded' (#25) from feat/dpkg-interrupted into main
CI / test (3.10) (push) Successful in 24s
CI / test (3.11) (push) Successful in 23s
CI / test (3.12) (push) Successful in 24s
2026-10-08 14:12:35 +00:00
christianmanivong 7b44f689bc feat: report a dpkg left halfway and when the host's own automatic updates last succeeded
CI / test (3.10) (push) Successful in 1m59s
CI / test (3.11) (push) Successful in 30s
CI / test (3.12) (push) Successful in 34s
CI / test (3.10) (pull_request) Successful in 39s
CI / test (3.11) (pull_request) Successful in 23s
CI / test (3.12) (pull_request) Successful in 27s
An unattended upgrade cut off while it built an initrd left a kernel package
half-configured on an Ubuntu host. For three months apt refused every upgrade,
while apt.systemd.daily exited quietly and the timers reported success
(NetOrk/netork#738).

HOST_STATUS_COMMAND now also reads, without root:

- dpkg's state the way apt checks it: numbered files left in
  /var/lib/dpkg/updates, or a package half-installed, unpacked,
  half-configured, waiting for triggers or flagged for reinstallation.
  While dpkg, apt, aptitude or unattended-upgrade runs, packages are
  halfway on purpose, so the answer is then None.
- /var/lib/apt/periodic/upgrade-stamp, which apt.systemd.daily touches only
  after unattended-upgrade succeeded, and the interval of
  APT::Periodic::Unattended-Upgrade in apt's units. update-success-stamp is
  not read: every apt-get update touches it, whoever runs it.

HostStatusDict gains package_manager_interrupted, interrupted_packages,
auto_updates_last_success and auto_updates_interval, all NotRequired.
2026-10-08 16:07:55 +02:00
christianmanivong ba2f18ea7d Merge pull request 'feat: read the virtual IPs vip-manager and keepalived declare on a host' (#24) from feature/virtual-ips into main
CI / test (3.10) (push) Successful in 25s
CI / test (3.11) (push) Successful in 26s
CI / test (3.12) (push) Successful in 26s
2026-10-08 10:41:02 +00:00
christianmanivong 37c09d0fa0 feat: read the virtual IPs vip-manager and keepalived declare on a host
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 24s
CI / test (3.12) (push) Successful in 26s
CI / test (3.10) (pull_request) Successful in 35s
CI / test (3.11) (pull_request) Successful in 36s
CI / test (3.12) (pull_request) Successful in 39s
A virtual IP sits on whichever member holds it right now; a standby's
address list shows no trace of it. What every member has is the
configuration that declares it, and reading that is the same on every
Linux host -- so VirtualIpsMixin.get_virtual_ips() lives here and a driver
only carries the command across (NetOrk/netork#829).

- vip-manager: every vip-manager.service / vip-manager@*.service unit; the
  address from a flag (1.x's -ip=${VIP_IP} resolved through the unit's
  environment), the environment, or the --config YAML (5.x).
- keepalived: the virtual_ipaddress(_excluded) entries of every
  vrrp_instance, with its interface and virtual_router_id; include is
  followed four levels deep, and a pattern that is no plain path glob is
  never handed to the root shell.

The same files hold the etcd password and auth_pass, so an awk program
filters on the host and prints allowlisted fields only. None per
mechanism is "did not look", [] a host without it. The command is about
4 kB and goes on an exec channel.

Version 4.2.0.
2026-10-08 12:37:28 +02:00
christianmanivong 8111c4cde3 Merge pull request 'feat: report a FreeBSD host as needing a reboot when its installed kernel is not the running one' (#23) from feat/freebsd-host-status into main
CI / test (3.10) (push) Successful in 23s
CI / test (3.11) (push) Successful in 22s
CI / test (3.12) (push) Successful in 24s
2026-10-08 10:11:15 +00:00
christianmanivong e1ce72cc5c feat: report a FreeBSD host as needing a reboot when its installed kernel is not the running one
CI / test (3.10) (push) Successful in 25s
CI / test (3.11) (push) Successful in 22s
CI / test (3.12) (push) Successful in 23s
CI / test (3.10) (pull_request) Successful in 24s
CI / test (3.11) (pull_request) Successful in 22s
CI / test (3.12) (pull_request) Successful in 24s
HOST_STATUS_COMMAND also prints freebsd-version -k and -r where
freebsd-version exists. When the two differ, a kernel update was installed
and is not running yet, so the host needs a reboot. When they match, it does
not.
2026-10-08 12:08:21 +02:00
christianmanivong 7ba84acad3 Merge pull request 'refactor!: drop the Docker model' (#21) from chore/drop-legacy-docker into main
CI / test (3.10) (push) Successful in 23s
CI / test (3.11) (push) Successful in 22s
CI / test (3.12) (push) Successful in 24s
2026-10-08 06:52:25 +00:00
christianmanivong 8eb7dcd871 refactor!: drop the Docker model
CI / test (3.10) (push) Successful in 25s
CI / test (3.11) (push) Successful in 22s
CI / test (3.12) (push) Successful in 24s
CI / test (3.10) (pull_request) Successful in 24s
CI / test (3.11) (pull_request) Successful in 22s
CI / test (3.12) (pull_request) Successful in 23s
DockerInfoDict, DockerContainerDict, DockerImageDict, DockerVolumeDict,
DockerNetworkDict and the OSDriver declaration of get_docker_info() go.
They described napalm-linux' old CLI reading, which napalm-linux 0.3.0
dropped. There is no container model here: netOrk reads and handles
containers over the Engine API (NetOrk/netork#765, phase 9); this
package only gives access to the engine.

Version 4.0.0.
2026-10-08 08:48:37 +02:00
christianmanivong 95d9031c8d Merge pull request 'feat!: declare the guest agent per guest OS, and share what provisioning needs' (#20) from feat/guest-os-provisioning into main
CI / test (3.10) (push) Successful in 23s
CI / test (3.11) (push) Successful in 21s
CI / test (3.12) (push) Successful in 23s
2026-10-08 05:57:09 +00:00
christianmanivong 885c7e1f53 feat!: declare the guest agent per guest OS, and share what provisioning needs
CI / test (3.10) (push) Successful in 39s
CI / test (3.11) (push) Successful in 36s
CI / test (3.12) (push) Successful in 39s
CI / test (3.10) (pull_request) Successful in 32s
CI / test (3.11) (pull_request) Successful in 23s
CI / test (3.12) (pull_request) Successful in 24s
GUEST_AGENT_PACKAGES / GUEST_AGENT_RUNCMD described one agent per
hypervisor. A hypervisor that provisions FreeBSD and OpenBSD guests needs
a different agent per guest, and netOrk has to know which guests a driver
can provision at all (NetOrk/netork#794). Both now come from one
class-level mapping netOrk reads off the driver class:

    GUEST_AGENTS = {guest_os: (packages, runcmd)}

Its keys are the guests the driver provisions, and
create_vm_from_cloud_init(guest_os=...) refuses any other. The base keeps
Linux with qemu-guest-agent.

New module `provisioning`, generic for every hypervisor:
- QEMU_GUEST_AGENTS: the QEMU agent for Linux, FreeBSD and OpenBSD, with
  package names and service commands verified on FreeBSD 15.1 and
  OpenBSD 7.9 cloud images (NetOrk/netork#793).
- network_config(): cloud-init's network-config v2 with MAC matching,
  moved here from napalm-vmware. FreeBSD's nuageinit reads no other
  version.
- split_compression(): how a packed image (.xz, .gz, .bz2, .zst) is
  unpacked.

BREAKING CHANGE: GUEST_AGENT_PACKAGES and GUEST_AGENT_RUNCMD are gone;
drivers declare GUEST_AGENTS instead.
2026-10-08 07:53:21 +02:00
christianmanivong eb80d5cb0d Merge pull request 'feat(container-engine): run the engine CLI privileged on request' (#19) from feat/privileged-cli into main
CI / test (3.10) (push) Successful in 23s
CI / test (3.11) (push) Successful in 22s
CI / test (3.12) (push) Successful in 23s
2026-10-07 20:11:13 +00:00
christianmanivong e50e497939 feat(container-engine): run the engine CLI privileged on request
CI / test (3.10) (push) Successful in 23s
CI / test (3.11) (push) Successful in 23s
CI / test (3.12) (push) Successful in 24s
CI / test (3.10) (pull_request) Successful in 23s
CI / test (3.11) (pull_request) Successful in 22s
CI / test (3.12) (pull_request) Successful in 23s
run_cli() and stream_cli() take privileged=, passed to the driver's
run_command()/open_stream(): an engine that refuses the login user can
be retried as root, the way the driver gains root for any command.
netOrk uses it to retry a refused container start/stop/restart with
sudo (NetOrk/netork#773).
2026-10-07 22:08:37 +02:00
christianmanivong 3aed0b48d7 Merge pull request 'feat: a public command channel and container engine access' (#18) from feat/container-engine-channel into main
CI / test (3.10) (push) Successful in 22s
CI / test (3.11) (push) Successful in 21s
CI / test (3.12) (push) Successful in 23s
2026-10-07 16:01:51 +00:00
christianmanivong 735b683028 feat: a public command channel and container engine access
CI / test (3.10) (push) Successful in 25s
CI / test (3.11) (push) Successful in 22s
CI / test (3.12) (push) Successful in 23s
CI / test (3.10) (pull_request) Successful in 24s
CI / test (3.11) (pull_request) Successful in 22s
CI / test (3.12) (pull_request) Successful in 22s
netOrk reached a host's shell through napalm-linux's private `_send`: an
interactive PTY with stdout and stderr merged and no exit code. For Docker
it went further and opened its own paramiko connections around the driver.

This adds the access layer only, no container model (NetOrk/netork#765):

- `channel.py`: `CommandChannelMixin` declares `run_command()` (stdout,
  stderr, exit code) and `open_stream()` (a `ByteStream` to a running
  command). Declared under TYPE_CHECKING, so hasattr stays truthful.
  `run_on_transport()` and `open_stream_on_transport()` implement both on
  a paramiko exec channel for any SSH driver; `ParamikoExecStream` drains
  stderr on every read so the shared window never stalls.
- `container_engine.py`: `ContainerEngineMixin` with `container_engines()`
  and `open_container_engine()`. The returned `ContainerEngineConnection`
  opens the engine API over `<binary> system dial-stdio` and runs the CLI
  with caller-chosen arguments. The driver decides the binary through
  `_container_engine_binary()`; callers never see the path.
- README: why the container model lives in netOrk and not here.

Additive; the Docker*Dict declarations stay until netOrk no longer reads
them. Version 2.6.0.

Refs #17
2026-10-07 17:58:57 +02:00
christianmanivong 08ec32e93a Merge pull request 'ci: run the tests and build the package on every push and pull request' (#16) from ci/workflow into main
CI / test (3.10) (push) Successful in 23s
CI / test (3.11) (push) Successful in 22s
CI / test (3.12) (push) Successful in 24s
2026-10-07 06:22:15 +00:00
christianmanivong bd43bd75fa ci: run the tests and build the package on every push and pull request
CI / test (3.10) (push) Successful in 29s
CI / test (3.11) (push) Successful in 28s
CI / test (3.12) (push) Successful in 27s
CI / test (3.10) (pull_request) Successful in 24s
CI / test (3.11) (pull_request) Successful in 25s
CI / test (3.12) (pull_request) Successful in 26s
The same job as napalm-fritzbox and napalm-opnsense: Python 3.10, 3.11 and
3.12, pytest, then wheel and sdist. napalm-device-types comes from
git.netork.io first, because PyPI has an unrelated package of that name.

This is the package the other drivers install from git first; here it
installs itself.
2026-10-07 07:52:44 +02:00
christianmanivong 1112191aec Merge pull request 'feat: read the sockets of a host without ss from netstat, and procd's services' (#15) from feat/listening-netstat into main 2026-10-07 05:21:31 +00:00
christianmanivong b8b89acee1 feat: read the sockets of a host without ss from netstat, and procd's services
OpenWrt has no ss: the listening-socket command fell through to [no-ss]
and the reading raised ListeningSocketsUnavailable. Without ss it now runs
netstat -lntup (busybox and net-tools both), and the cgroups are read for
its PIDs alike.

- netstat names a socket's process as PID/Program; a UDP line has no
  state column, "-" is a socket without a process, and net-tools prints
  tcp6/udp6 and program names with a space ("sshd: /usr/sbin").
- On OpenWrt the cgroup is /services/<name>/<instance>, so the unit is
  the procd service -- a jailed one too, whose PID is not the one procd
  reports (dnsmasq under ujail).
- A host with neither tool still raises ListeningSocketsUnavailable.

Checked against a real OpenWrt 25.12.2 access point, and that a Linux and
a Proxmox host still read the same sockets with the longer command.

2.5.0. For netOrk#673.
2026-10-07 07:20:27 +02:00
christianmanivong f833e23422 Merge pull request 'feat: read what listens on which address, and which service it is, once for every driver' (#7) from feat/listening-sockets into main 2026-10-06 16:19:22 +00:00
christianmanivong b97ec654a0 feat: read what listens on which address, and which service it is, once for every driver
Whether a service is reachable from outside its host is decided by the
address it listens on: 0.0.0.0:5432 is, 127.0.0.1:5432 is not. Reading
that is the same on every Linux host, so the command and its parse live
here once and a driver only carries the command across
(ListeningSocketsMixin, _run_listening_sockets_command).

One framed round trip: ss -lntup for every listening TCP and bound UDP
socket, then /proc/<pid>/cgroup for each process holding one, which names
the systemd service (v2, nested slices, v1's name=systemd hierarchy) or
the container (docker-<id>.scope, /docker/<id>) it runs in.

- Root: only root sees every process. The script goes as one sh -c
  argument, so a sudo -n prefix covers all of it; when that brings no
  report back the reading runs again unprivileged and says it is not
  attributed.
- No -H: iproute2 before 4.10 fails on it, which would read as nothing
  listening. The header is skipped instead.
- A host without ss raises ListeningSocketsUnavailable; a report cut short
  or a failing ss raises ValueError.
- The reading is raw: docker-proxy shows up as docker.service, loopback as
  loopback. What counts as reachable is the consumer's call.

2.4.0. For netOrk#658.
2026-10-06 18:18:56 +02:00
christianmanivong c2d8d4a0d2 Merge pull request 'feat: say where a pending update comes from, whether it is a security fix, and whether the host needs a reboot' (#6) from feat/update-origin-host-status into main 2026-10-05 22:19:25 +00:00
23 changed files with 4026 additions and 179 deletions

No files matched your search

+45
View File
@@ -0,0 +1,45 @@
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
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/*
+69 -2
View File
@@ -81,11 +81,47 @@ too, and `hasattr(driver, "get_kernel_facts")` has to stay truthful.
`HostStatusMixin` (`get_host_status`) is mixed in the same way: whether a Linux host needs
a reboot to finish an update (`/var/run/reboot-required`, `needs-restarting -r`, or a newer
kernel of the running flavour installed) and whether it patches itself (unattended-upgrades,
dnf-automatic). `package_updates` holds the shared apt and dnf parsers: apt's suites become
kernel of the running flavour installed), whether it patches itself (unattended-upgrades,
dnf-automatic) and when that last succeeded (apt's `upgrade-stamp`), and whether dpkg was
left halfway (a journal in `/var/lib/dpkg/updates` or a half-configured package; unknown
while a package manager is running), and how full `/`, `/boot` and `/var` are (`df -Pk`,
each filesystem once, in bytes; how full is too full is the caller's decision). `package_updates` holds the shared apt and dnf parsers: apt's suites become
an update's `origin`, a `-security` suite makes it a security update, and dnf's security
advisories do the same.
`ListeningSocketsMixin` (`get_listening_sockets`) is mixed in the same way: every listening
TCP and bound UDP socket from `ss -lntup`, with the systemd service or container behind it
from `/proc/<pid>/cgroup`, in one round trip. A driver supplies
`_run_listening_sockets_command(command, privileged=)`; the command arrives as one `sh -c`
argument, so a `sudo -n` prefix covers all of it. Without root `ss` names only the login
user's processes, and the reading says so (`attributed: false`) instead of failing. A host
without `ss` is read with `netstat -lntup` (OpenWrt's busybox, old net-tools); on OpenWrt the
cgroup names the procd service (`/services/<name>/<instance>`). A host with neither raises
`ListeningSocketsUnavailable`.
`VirtualIpsMixin` (`get_virtual_ips`) reads the addresses a host's HA configuration lets
float between machines: vip-manager (5.x's `--config` YAML, 1.x's `-ip=${VIP_IP}` resolved
through the unit's environment) and keepalived (`virtual_ipaddress` of every
`vrrp_instance`, `include` followed four levels deep). It reports what is *declared*, not
what the host holds right now -- that is in its interface addresses. The same files hold
the etcd password and `auth_pass`, so an awk program filters **on the host** and prints
allowlisted fields only; a secret never reaches the caller. Root first, then without it.
A mechanism it could not read completely is `None` ("did not look"), a host without it
`[]`. The command is about 4 kB, so a driver sends it on an exec channel
(`_run_virtual_ips_command(command, privileged=)`), not typed into a shell.
`OpenSshServerMixin` (`harden_ssh_server`, `get_ssh_server_algorithms`) takes algorithms
out of what an OpenSSH server offers -- the ones ssh-audit flags. It reads the lists sshd
really uses (`sshd -T`), takes out what was asked and writes the rest to
`/etc/ssh/sshd_config.d/00-netork-ssh-hardening.conf`; only where sshd_config has no such
`Include` is sshd_config itself rewritten, after a copy. It never removes the last algorithm
of a class, nor -- given the managing client's lists -- the last one that client speaks.
`sshd -t` must accept the result and `sshd -T` must show it; after the reload the caller's
`verify` opens a new session, and if that fails the still-open one puts the old state back.
A driver mixes it in and supplies `run_command`; a platform that regenerates sshd_config
itself (an appliance with its own database) must not, or opts out with
`harden_ssh_server = None`.
**Update readers raise when they cannot read.** `get_available_updates` returns an empty
list only when nothing is pending; netOrk keeps "pending since" per package, and an empty
list for "don't know" would reset it.
@@ -99,6 +135,27 @@ trip (`list-unit-files` plus one `systemctl show` over every loaded unit) instea
`is-enabled` and a `show` per unit. A host without systemd raises `SystemdUnavailable`,
a `NotImplementedError`, so a driver can fall back to another init system.
### Access channels: why there is no container model here
`CommandChannelMixin` declares a driver's public channel to its host:
- `run_command(command, *, privileged=False, timeout=60, stdin=None)` returns a `CommandResult(stdout, stderr, exit_code)`;
- `open_stream(command, *, privileged=False)` returns a `ByteStream` to the command's stdin and stdout.
The mixin sits in `DeviceTypeDriver` and only declares, so `hasattr(driver, "run_command")` is true exactly where a driver implements it. For SSH drivers, `channel.run_on_transport` and `channel.open_stream_on_transport` are the implementation over a paramiko exec channel: no PTY, stderr kept apart, a real exit code. How a command gains root stays with the driver.
**`ContainerEngineMixin` is deliberately different from `SystemdServicesMixin`.** The systemd mixin owns its command and its parse; this one owns neither.
- `container_engines()` says which engines the host offers (`[{"engine": "docker", "api": "docker-engine"}]`).
- `open_container_engine(engine)` returns a `ContainerEngineConnection`:
- `open_api()` is a stream to the engine's API (`docker system dial-stdio`);
- `run_cli(args)` and `stream_cli(args)` run the engine's CLI with arguments the caller chooses. With `privileged=True` the driver runs it the way `run_command` gains root, for an engine that refuses the login user.
- The driver decides only *how* the engine is reached. Its hook `_container_engine_binary(engine)` returns, for QNAP, the Container Station path, and the caller never sees it.
**What runs on the engine, and what to do with it, is netOrk's** (NetOrk/netork#765): the container model, the Engine API requests, compose, updates. Above the connection everything is specific to the service, so this package abstracts the connection and nothing more.
A driver mixes `ContainerEngineMixin` in itself; `OSDriver` does not carry it. A Windows host is an OS driver too, and has no `dial-stdio` to offer over WinRM.
**Never half-close early.** `ByteStream.write` does not close anything; only `close_write` does. `dial-stdio` hands the daemon a half-close as "client gone", and the daemon then answers an unfinished request with HTTP 499.
A function class may use the **template form** — public method concrete, the
device-specific part a `_hook` declared under `if TYPE_CHECKING` — *when the base
genuinely does work* on the result: normalising, sorting, validating, or orchestrating
@@ -281,6 +338,16 @@ class ProxmoxDriver(HypervisorDriver):
...
```
A hypervisor that provisions VMs from cloud images declares, per guest OS,
the agent cloud-init installs so the hypervisor can read the new VM's IP:
`GUEST_AGENTS = {guest_os: (packages, runcmd)}`. Its keys are the guests the
driver can provision, and `create_vm_from_cloud_init(guest_os=...)` refuses
any other. A QEMU-based driver takes `provisioning.QEMU_GUEST_AGENTS`
(Linux, FreeBSD, OpenBSD). `provisioning.network_config()` writes the
network-config v2 every cloud-init flavour reads (FreeBSD's nuageinit reads
no other), and `provisioning.split_compression()` says how a packed image is
unpacked; both are the same for every hypervisor.
### OS / Linux
```python
+56
View File
@@ -34,7 +34,9 @@ Role bases -- what a device *is*:
Function classes -- what a device *can do*. Shared behaviour lives here once
instead of being restated on every role that happens to need it:
* :class:`~napalm_device_types.channel.CommandChannelMixin`
* :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin`
* :class:`~napalm_device_types.container_engine.ContainerEngineMixin`
* :class:`~napalm_device_types.dhcp.DhcpServerMixin`
* :class:`~napalm_device_types.firewall_rules.FirewallRuleMixin`
* :class:`~napalm_device_types.health_metrics.HealthMetricsMixin`
@@ -42,6 +44,7 @@ instead of being restated on every role that happens to need it:
* :class:`~napalm_device_types.host_reboot.HostRebootMixin`
* :class:`~napalm_device_types.interface_filter.InterfaceFilterMixin`
* :class:`~napalm_device_types.kernel.KernelFactsMixin`
* :class:`~napalm_device_types.listening.ListeningSocketsMixin`
* :class:`~napalm_device_types.mac_acl.MacAclMixin`
* :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
* :class:`~napalm_device_types.packages.PackageManagementMixin`
@@ -49,6 +52,7 @@ instead of being restated on every role that happens to need it:
* :class:`~napalm_device_types.services.ServiceControlMixin`
* :class:`~napalm_device_types.systemd.SystemdServicesMixin`
* :class:`~napalm_device_types.updates.UpdateMixin`
* :class:`~napalm_device_types.virtual_ips.VirtualIpsMixin`
Introspection -- :func:`~napalm_device_types.roles.roles_of`,
:func:`~napalm_device_types.roles.role_keys_of` and
@@ -57,7 +61,20 @@ Introspection -- :func:`~napalm_device_types.roles.roles_of`,
from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec
from napalm_device_types.access_point import AccessPointDriver
from napalm_device_types.channel import (
ByteStream,
CommandChannelMixin,
CommandResult,
ParamikoExecStream,
open_stream_on_transport,
run_on_transport,
)
from napalm_device_types.config_lifecycle import ConfigLifecycleMixin
from napalm_device_types.container_engine import (
ContainerEngineConnection,
ContainerEngineMixin,
ContainerEngineUnavailable,
)
from napalm_device_types.dhcp import DhcpServerMixin, normalize_cidr, normalize_mac
from napalm_device_types.firewall import FirewallDriver
from napalm_device_types.hypervisor import HypervisorDriver
@@ -68,6 +85,12 @@ from napalm_device_types.host_reboot import HostRebootMixin
from napalm_device_types.interface_filter import InterfaceFilterMixin
from napalm_device_types.kernel import KERNEL_FACTS_COMMAND, KernelFactsMixin, parse_kernel_facts
from napalm_device_types.lag import add_lag_interfaces
from napalm_device_types.listening import (
LISTENING_SOCKETS_COMMAND,
ListeningSocketsMixin,
ListeningSocketsUnavailable,
parse_listening_sockets,
)
from napalm_device_types.mac_acl import MacAclMixin
from napalm_device_types.media import MediaDriver
from napalm_device_types.nat_vpn import NatVpnMixin
@@ -91,12 +114,33 @@ from napalm_device_types.systemd import (
parse_systemd_services,
)
from napalm_device_types.updates import UpdateMixin
from napalm_device_types.virtual_ips import (
VIRTUAL_IPS_COMMAND,
VirtualIpsMixin,
parse_virtual_ips,
)
from napalm_device_types.residential_gateway import ResidentialGatewayDriver
from napalm_device_types.ssh_server import (
SSH_ALGORITHM_CLASSES,
OpenSshServerMixin,
SshHardeningPlan,
SshHardeningResult,
plan_ssh_hardening,
)
from napalm_device_types.storage import StorageDriver
from napalm_device_types.switch import SwitchDriver
__all__ = [
"AccessPointDriver",
"ByteStream",
"CommandChannelMixin",
"CommandResult",
"ContainerEngineConnection",
"ContainerEngineMixin",
"ContainerEngineUnavailable",
"ParamikoExecStream",
"open_stream_on_transport",
"run_on_transport",
"ConfigLifecycleMixin",
"DeviceTypeDriver",
"DhcpServerMixin",
@@ -123,6 +167,10 @@ __all__ = [
"KernelFactsMixin",
"KERNEL_FACTS_COMMAND",
"parse_kernel_facts",
"LISTENING_SOCKETS_COMMAND",
"ListeningSocketsMixin",
"ListeningSocketsUnavailable",
"parse_listening_sockets",
"PhoneDriver",
"PingSweepMixin",
"PortSpec",
@@ -135,6 +183,14 @@ __all__ = [
"SystemdUnavailable",
"parse_systemd_services",
"UpdateMixin",
"VIRTUAL_IPS_COMMAND",
"VirtualIpsMixin",
"OpenSshServerMixin",
"SSH_ALGORITHM_CLASSES",
"SshHardeningPlan",
"SshHardeningResult",
"plan_ssh_hardening",
"parse_virtual_ips",
"add_lag_interfaces",
"driver_supports_ping",
"normalize_cidr",
+2 -1
View File
@@ -12,6 +12,7 @@ from typing import NamedTuple
from napalm.base import NetworkDriver
from napalm_device_types.channel import CommandChannelMixin
from napalm_device_types.host_reboot import HostRebootMixin
from napalm_device_types.ping_sweep import PingSweepMixin
@@ -48,7 +49,7 @@ class PortSpec(NamedTuple):
mandatory: bool = False
class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, NetworkDriver):
class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, CommandChannelMixin, NetworkDriver):
"""Common base for all netOrk device-type drivers.
Sits between napalm.base.NetworkDriver and the type-specific abstract
+188
View File
@@ -0,0 +1,188 @@
# -*- coding: utf-8 -*-
"""The command channel: a public way to run a command on a host, or open a stream.
A driver already reaches its host, over SSH, a REST API or WinRM. What it did
not offer was a *public* way for the layer above to use that reach: netOrk sent
shell strings through napalm-linux's private ``_send``, an interactive PTY that
merges stdout and stderr and has no exit code.
:class:`CommandChannelMixin` declares two methods, and a driver that can
implements them:
* :meth:`run_command` runs a command and returns stdout, stderr and the exit code.
* :meth:`open_stream` starts a command and returns a :class:`ByteStream` to its
stdin and stdout. That is how netOrk speaks the Docker Engine API, over
``docker system dial-stdio`` (see :mod:`napalm_device_types.container_engine`).
Like the role bases, the mixin only declares (under ``TYPE_CHECKING``), so
``hasattr(driver, "run_command")`` stays a truthful answer.
:func:`run_on_transport` and :func:`open_stream_on_transport` are the SSH
implementation over a paramiko ``Transport``, for any SSH driver: an exec
channel, so no PTY, separate stderr and a real exit status. How a command gains
root (``sudo -S`` with the password on stdin, ``sudo -n``, or nothing as root)
stays with the driver.
"""
from __future__ import annotations
import socket
import time
from typing import Any, NamedTuple, Optional, Protocol, TYPE_CHECKING
#: How much of a stream's stderr is kept. Enough to tell "permission denied"
#: from "no such command"; a stream that writes megabytes there keeps its tail.
STDERR_TAIL = 64 * 1024
_READ = 32 * 1024
_POLL = 0.01
class CommandResult(NamedTuple):
"""What a command printed, and how it ended."""
stdout: str
stderr: str
exit_code: int
class ByteStream(Protocol):
"""A running command's stdin and stdout, as bytes.
``read`` returns ``b""`` at end of stream and raises :class:`TimeoutError`
when nothing arrives in time. ``write`` never closes anything:
``dial-stdio`` hands the daemon a half-close as "client gone", and the
daemon then answers an unfinished request with HTTP 499 (netork#771). Only
``close_write`` closes the writing side.
"""
def read(self, max_bytes: int, timeout: Optional[float] = None) -> bytes: ...
def write(self, data: bytes) -> None: ...
def close_write(self) -> None: ...
def close(self) -> None: ...
@property
def exit_status(self) -> Optional[int]: ...
@property
def stderr(self) -> str: ...
class CommandChannelMixin:
"""Declares the channel; a driver that can reach a shell on its host implements it."""
if TYPE_CHECKING: # pragma: no cover - declared for type checkers only
def run_command(
self,
command: str,
*,
privileged: bool = False,
timeout: float = 60,
stdin: Optional[bytes] = None,
) -> CommandResult:
"""Run *command* with the host's shell; *stdin*, if given, is written first."""
...
def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
"""Start *command* and return a stream to its stdin and stdout."""
...
class ParamikoExecStream:
"""A :class:`ByteStream` over a paramiko exec channel.
stderr is drained on every read: stdout and stderr share one window, and an
unread stderr would stall the stream.
"""
def __init__(self, channel: Any, *, stderr_limit: int = STDERR_TAIL) -> None:
self._channel = channel
self._stderr = b""
self._limit = stderr_limit
def _drain_stderr(self) -> None:
while self._channel.recv_stderr_ready():
self._stderr = (self._stderr + self._channel.recv_stderr(_READ))[-self._limit :]
def read(self, max_bytes: int, timeout: Optional[float] = None) -> bytes:
self._channel.settimeout(timeout)
try:
return self._channel.recv(max_bytes)
except socket.timeout:
raise TimeoutError(f"no data within {timeout}s") from None
finally:
self._drain_stderr()
def write(self, data: bytes) -> None:
self._channel.sendall(data)
def close_write(self) -> None:
self._channel.shutdown_write()
def close(self) -> None:
self._channel.close()
@property
def exit_status(self) -> Optional[int]:
if not self._channel.exit_status_ready():
return None
return self._channel.recv_exit_status()
@property
def stderr(self) -> str:
self._drain_stderr()
return self._stderr.decode("utf-8", "replace")
def _collect(channel: Any, command: str, timeout: float) -> tuple:
out, err = b"", b""
deadline = time.monotonic() + timeout
while True:
if channel.recv_ready():
out += channel.recv(_READ)
elif channel.recv_stderr_ready():
err += channel.recv_stderr(_READ)
elif channel.exit_status_ready():
return out, err
elif time.monotonic() > deadline:
raise TimeoutError(f"{command!r} did not finish within {timeout}s")
else:
time.sleep(_POLL)
def run_on_transport(
transport: Any, command: str, *, stdin: Optional[bytes] = None, timeout: float = 60
) -> CommandResult:
"""Run *command* on an exec channel of the paramiko *transport*."""
channel = transport.open_session()
try:
channel.settimeout(timeout)
channel.exec_command(command)
if stdin is not None:
channel.sendall(stdin)
channel.shutdown_write()
out, err = _collect(channel, command, timeout)
return CommandResult(
out.decode("utf-8", "replace"), err.decode("utf-8", "replace"), channel.recv_exit_status()
)
finally:
channel.close()
def open_stream_on_transport(
transport: Any, command: str, *, stdin_prefix: Optional[bytes] = None
) -> ParamikoExecStream:
"""Start *command* on an exec channel and return its stream.
*stdin_prefix* is written before anything else, which is how a sudo password
reaches ``sudo -S`` ahead of the stream's own bytes.
"""
channel = transport.open_session()
channel.exec_command(command)
if stdin_prefix:
channel.sendall(stdin_prefix)
return ParamikoExecStream(channel)
+141
View File
@@ -0,0 +1,141 @@
# -*- coding: utf-8 -*-
"""Access to a host's container engine, and nothing more.
This package abstracts the connection, the driver builds it, and netOrk does
the talking (NetOrk/netork#765). So there is no container model here and no
parsing:
* :meth:`ContainerEngineMixin.container_engines` says which engines the host
offers, and which API each speaks.
* :meth:`ContainerEngineMixin.open_container_engine` returns a
:class:`ContainerEngineConnection`. It opens a stream to the engine's API
(``docker system dial-stdio``), and it runs the engine's CLI with arguments
the caller chooses. The CLI is there for what the API cannot do: compose,
pulls that need the host user's registry login, and private registry digests.
The driver decides *how* the engine is reached. The hook is
:meth:`ContainerEngineMixin._container_engine_binary`. QNAP returns its
Container Station path, so callers never see where the binary lives. The
command runs through the driver's own :class:`~napalm_device_types.channel.CommandChannelMixin`,
whose privilege handling applies unchanged; Docker normally needs none, because
the login user is in the ``docker`` group.
Mixed in by a driver, not by a role base: a Windows host is an
:class:`~napalm_device_types.os.OSDriver` too, and without a shell channel it
has no ``dial-stdio`` to offer. ``hasattr(driver, "open_container_engine")``
stays truthful.
"""
from __future__ import annotations
import shlex
from typing import Any, Dict, List, Optional, Sequence, TYPE_CHECKING
from napalm_device_types.channel import ByteStream, CommandResult
from napalm_device_types.models import ContainerEngineDict
#: Engines this package knows how to reach, and the API each speaks.
ENGINES: Dict[str, str] = {"docker": "docker-engine"}
_PROBE_TIMEOUT = 15
_CLI_TIMEOUT = 120
class ContainerEngineUnavailable(NotImplementedError):
"""The host offers no such engine, or this package does not know how to reach it."""
class ContainerEngineConnection:
"""The way to one container engine on one host.
Built by :meth:`ContainerEngineMixin.open_container_engine`. Callers pass
arguments; binary, quoting and the channel are the driver's.
"""
def __init__(self, driver: Any, engine: str, binary: str) -> None:
self._driver = driver
self._binary = binary
self.engine = engine
def _command(self, args: Sequence[str]) -> str:
return shlex.join([self._binary, *args])
def open_api(self) -> ByteStream:
"""A stream to the engine's API: HTTP/1.1 over ``system dial-stdio``."""
return self._driver.open_stream(self._command(["system", "dial-stdio"]))
def run_cli(
self,
args: Sequence[str],
*,
stdin: Optional[bytes] = None,
timeout: float = _CLI_TIMEOUT,
privileged: bool = False,
) -> CommandResult:
"""Run the engine's CLI with *args* and wait for it.
*privileged* runs it as root, the way the driver gains root for any
command: for a call the engine refused to the login user.
"""
return self._driver.run_command(
self._command(args), privileged=privileged, timeout=timeout, stdin=stdin
)
def stream_cli(
self, args: Sequence[str], *, merge_stderr: bool = False, privileged: bool = False
) -> ByteStream:
"""Start the engine's CLI with *args* and stream its output.
*merge_stderr* folds stderr into the stream, for tools that report
progress there (``compose up``); *privileged* as for :meth:`run_cli`.
"""
command = self._command(args)
return self._driver.open_stream(
f"{command} 2>&1" if merge_stderr else command, privileged=privileged
)
class ContainerEngineMixin:
"""Adds container engine access to a driver that implements the channel."""
if TYPE_CHECKING: # pragma: no cover - declared for type checkers only
def run_command(
self,
command: str,
*,
privileged: bool = False,
timeout: float = 60,
stdin: Optional[bytes] = None,
) -> CommandResult: ...
def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream: ...
def _container_engine_binary(self, engine: str) -> str:
"""The host command that is *engine*'s CLI. Override where it is not on PATH."""
return engine
def container_engines(self) -> List[ContainerEngineDict]:
"""
Returns the container engines the host offers:
* engine (string) - e.g. "docker"
* api (string) - what its API is, e.g. "docker-engine"
An engine is listed when its CLI exists on the host. Whether the login
user may use it is decided by the first API call, which says
"permission denied" in the stream's stderr.
"""
found: List[ContainerEngineDict] = []
for engine, api in ENGINES.items():
probe = shlex.join(["command", "-v", self._container_engine_binary(engine)])
result = self.run_command(probe, timeout=_PROBE_TIMEOUT)
if result.exit_code == 0 and result.stdout.strip():
found.append({"engine": engine, "api": api})
return found
def open_container_engine(self, engine: str) -> ContainerEngineConnection:
"""The connection to *engine*; :class:`ContainerEngineUnavailable` if unknown."""
if engine not in ENGINES:
raise ContainerEngineUnavailable(f"no way to reach container engine {engine!r}")
return ContainerEngineConnection(self, engine, self._container_engine_binary(engine))
+154 -6
View File
@@ -16,13 +16,40 @@ carries the command across.
- A kernel newer than the running one is installed, of the same flavour. A
Raspberry Pi carries ``rpi-v8`` and ``rpi-2712`` builds side by side, and only
the running one's counts.
- On FreeBSD, ``freebsd-version -k`` (the installed kernel) differs from
``freebsd-version -r`` (the running one): a patch replaced the kernel.
It is ``None`` when none of these could be read, for example in a container
without a ``/lib/modules`` of its own.
**Auto updates** is apt's ``APT::Periodic::Unattended-Upgrade`` (set, not "0",
and ``apt-daily-upgrade.timer`` not disabled) or an enabled dnf-automatic timer.
It is ``None`` on a host with neither apt nor dnf-automatic.
It is ``None`` on a host with neither apt nor dnf-automatic. When they last ran
successfully is ``/var/lib/apt/periodic/upgrade-stamp``, which
``apt.systemd.daily`` touches only after unattended-upgrade succeeded.
``update-success-stamp`` would say nothing: every ``apt-get update`` touches it,
whoever runs it. How often they are meant to run is the same setting, in apt's
units (a bare number is days).
**An interrupted package manager** is dpkg left halfway, the way apt checks it:
numbered files left in ``/var/lib/dpkg/updates``, or a package in a state only
a run cut off leaves behind (half-installed, unpacked, half-configured, waiting
for triggers, or flagged for reinstallation). apt then refuses to install
anything until ``dpkg --configure -a`` has run, while ``apt.systemd.daily``
exits quietly every day (NetOrk/netork#738). While dpkg, apt, aptitude or
unattended-upgrade is running, packages are halfway on purpose -- also between
two of apt's dpkg calls -- so the answer is then ``None``, as on a host without
dpkg. unattended-upgrade is told from Ubuntu's idle shutdown helper by its
command line (:data:`BUSY_CHECK`).
**Filesystems** are how full ``/``, ``/boot`` and ``/var`` are, from POSIX
``df -Pk``: where an upgrade unpacks a kernel, builds its initrd and keeps the
package files. A kernel upgrade that ran out of room there left dpkg halfway on
a Proxmox host whose root had been 94 % full for a week (NetOrk/netork#862). A
directory that is not a mount of its own is reported as the filesystem it lives
on, so each filesystem appears once. ``available`` is what df says an ordinary
user can still write, which on ext4 is less than total minus used. How full is
too full is not decided here.
"""
from __future__ import annotations
@@ -30,7 +57,11 @@ from __future__ import annotations
import re
from typing import Dict, List, Optional, Tuple, TYPE_CHECKING
from napalm_device_types.models import HostStatusDict
from napalm_device_types.models import (
FilesystemUsageDict,
HostStatusDict,
InterruptedPackageDict,
)
from napalm_device_types.terminal import strip_terminal_codes
_BEGIN = "HSTAT_BEGIN"
@@ -38,6 +69,20 @@ _END = "HSTAT_END"
_REBOOT_FILE = "/var/run/reboot-required"
_APT_TIMER = "apt-daily-upgrade.timer"
_DNF_TIMERS = ("dnf-automatic.timer", "dnf-automatic-install.timer")
_UPGRADE_STAMP = "/var/lib/apt/periodic/upgrade-stamp"
_DPKG_JOURNAL = "/var/lib/dpkg/updates"
#: Where an upgrade writes: the kernel's modules, its initrd, the package files.
_FILESYSTEMS = "/ /boot /var"
#: Prints ``[dpkg-busy]`` while a process that leaves packages halfway is working:
#: dpkg, apt, apt-get or aptitude by name, unattended-upgrade by its command line.
#: Its name is cut to 15 characters, ``unattended-upgr``, the same as that of
#: ``unattended-upgrade-shutdown --wait-for-signal``, which Ubuntu keeps running all
#: the time and which installs nothing.
BUSY_CHECK = (
"{ ps -e -o comm= 2>/dev/null | grep -qxE 'dpkg|apt|apt-get|aptitude' || "
"ps -e -o args= 2>/dev/null | grep -qE '(^|[ /])unattended-upgrade( |$)'; } "
"&& echo '[dpkg-busy]'; "
)
#: One line, POSIX ``sh``, read-only, no privileges. The frame markers are
#: printed in two halves so that an echoing transport does not show them early.
@@ -50,8 +95,18 @@ HOST_STATUS_COMMAND = (
"if command -v needs-restarting >/dev/null 2>&1; then echo '[needs-restarting]'; "
"needs-restarting -r >/dev/null 2>&1; echo $?; fi; "
"echo '[kernel]'; uname -r; echo '[modules]'; ls -1 /lib/modules 2>/dev/null; "
"if command -v freebsd-version >/dev/null 2>&1; then "
"echo '[freebsd-kernel]'; freebsd-version -k; "
"echo '[freebsd-running]'; freebsd-version -r; fi; "
"if command -v apt-config >/dev/null 2>&1; then echo '[apt-config]'; "
"apt-config dump 2>/dev/null | grep '^APT::Periodic::Unattended-Upgrade '; fi; "
"apt-config dump 2>/dev/null | grep '^APT::Periodic::Unattended-Upgrade '; "
f"echo '[upgrade-stamp]'; stat -c %Y {_UPGRADE_STAMP} 2>/dev/null; fi; "
"if command -v dpkg-query >/dev/null 2>&1; then "
f"{BUSY_CHECK}"
f"echo '[dpkg-journal]'; ls -1 {_DPKG_JOURNAL} 2>/dev/null | head -n 20; "
"echo '[dpkg-audit]'; dpkg-query -W -f='${db:Status-Abbrev} ${Package}\\n' 2>/dev/null "
"| awk 'substr($0, 2, 1) ~ /[HUFWt]/ || substr($0, 3, 1) == \"R\"' | head -n 50; fi; "
f"echo '[filesystems]'; df -Pk {_FILESYSTEMS}; "
f"echo '[timers]'; for u in {_APT_TIMER} {' '.join(_DNF_TIMERS)}; do "
'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled "$u" 2>/dev/null)"; done; '
"printf '%s%s\\n' HSTAT_ END; } 2>/dev/null | cat"
@@ -59,6 +114,20 @@ HOST_STATUS_COMMAND = (
_PERIODIC = re.compile(r'^APT::Periodic::Unattended-Upgrade\s+"([^"]*)"')
_OFF_STATES = frozenset({"disabled", "masked"})
_INTERVAL = re.compile(r"(\d+)([smhd]?)")
_UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "": 86400}
_JOURNAL_ENTRY = re.compile(r"\d+")
#: One line of ``df -P``: device, 1024-blocks, used, available, capacity, mountpoint.
#: The device and the mountpoint may hold blanks; the four numbers between them do not.
_DF_LINE = re.compile(r"^.+?\s+(\d+)\s+(\d+)\s+(\d+)\s+(?:\d+%|-)\s+(/.*)$")
#: dpkg's second status letter -> its word, for the states a cut-off run leaves.
_HALFWAY = {
"H": "half-installed",
"U": "unpacked",
"F": "half-configured",
"W": "triggers-awaited",
"t": "triggers-pending",
}
def _sections(output: str) -> Dict[str, List[str]]:
@@ -100,12 +169,26 @@ def kernel_reboot_pending(running: str, installed: List[str]) -> Optional[str]:
return None
def _freebsd_reboot(sections: Dict[str, List[str]]) -> Optional[Tuple[bool, Optional[str]]]:
"""FreeBSD's answer from freebsd-version, or None on any other system."""
installed = (sections.get("freebsd-kernel") or [""])[0]
running = (sections.get("freebsd-running") or [""])[0]
if not (installed and running):
return None
if installed != running:
return True, f"kernel {installed} installed, {running} running"
return False, None
def _reboot(sections: Dict[str, List[str]]) -> Tuple[Optional[bool], Optional[str]]:
if "reboot-required" in sections:
return True, f"{_REBOOT_FILE} is present"
needs = sections.get("needs-restarting")
if needs and needs[0] == "1":
return True, "needs-restarting -r reports a reboot"
freebsd = _freebsd_reboot(sections)
if freebsd is not None:
return freebsd
running = (sections.get("kernel") or [""])[0]
modules = sections.get("modules") or []
newer = kernel_reboot_pending(running, modules) if running and modules else None
@@ -124,17 +207,70 @@ def _timer_states(sections: Dict[str, List[str]]) -> Dict[str, str]:
return states
def _periodic(sections: Dict[str, List[str]]) -> Optional[str]:
"""The value of ``APT::Periodic::Unattended-Upgrade``, or None when it is not set."""
lines = sections.get("apt-config") or []
match = next(filter(None, (_PERIODIC.match(line) for line in lines)), None)
return match.group(1) if match else None
def _auto_updates(sections: Dict[str, List[str]]) -> Optional[bool]:
timers = _timer_states(sections)
if any(timers.get(t) == "enabled" for t in _DNF_TIMERS):
return True
if "apt-config" not in sections:
return None
match = next(filter(None, (_PERIODIC.match(line) for line in sections["apt-config"])), None)
switched_on = match is not None and match.group(1) not in ("", "0")
switched_on = _periodic(sections) not in (None, "", "0")
return switched_on and timers.get(_APT_TIMER) not in _OFF_STATES
def _interval(sections: Dict[str, List[str]]) -> Optional[int]:
"""apt's interval in seconds, the way ``apt.systemd.daily`` reads it; 0 for "always"."""
value = _periodic(sections)
if value == "always":
return 0
match = _INTERVAL.fullmatch(value or "")
if not match or int(match.group(1)) == 0:
return None
return int(match.group(1)) * _UNIT_SECONDS[match.group(2)]
def _last_success(sections: Dict[str, List[str]]) -> Optional[int]:
stamp = (sections.get("upgrade-stamp") or [""])[0]
return int(stamp) if stamp.isdigit() else None
def _halfway(line: str) -> Optional[InterruptedPackageDict]:
status, _, name = line.partition(" ")
name = name.strip()
state = "reinstall-required" if status[2:3] == "R" else _HALFWAY.get(status[1:2])
return {"name": name, "state": state} if name and state else None
def _dpkg(sections: Dict[str, List[str]]) -> Tuple[Optional[bool], List[InterruptedPackageDict]]:
if "dpkg-audit" not in sections or "dpkg-busy" in sections:
return None, []
packages = [p for p in map(_halfway, sections["dpkg-audit"]) if p is not None]
journal = any(_JOURNAL_ENTRY.fullmatch(f) for f in sections.get("dpkg-journal") or [])
return bool(packages) or journal, packages
def _filesystems(sections: Dict[str, List[str]]) -> Optional[List[FilesystemUsageDict]]:
"""Each filesystem df named, once and in its order; None when it named none."""
found: Dict[str, FilesystemUsageDict] = {}
for line in sections.get("filesystems") or []:
match = _DF_LINE.match(line)
if not match or int(match.group(1)) == 0:
continue
total, used, available = (int(match.group(i)) * 1024 for i in (1, 2, 3))
mountpoint = match.group(4)
found.setdefault(
mountpoint,
{"mountpoint": mountpoint, "total": total, "used": used, "available": available},
)
return list(found.values()) or None
def parse_host_status(output: str) -> HostStatusDict:
"""Parse what :data:`HOST_STATUS_COMMAND` printed.
@@ -142,10 +278,16 @@ def parse_host_status(output: str) -> HostStatusDict:
"""
sections = _sections(output)
required, reason = _reboot(sections)
interrupted, packages = _dpkg(sections)
return {
"reboot_required": required,
"reboot_reason": reason,
"auto_updates": _auto_updates(sections),
"auto_updates_last_success": _last_success(sections),
"auto_updates_interval": _interval(sections),
"package_manager_interrupted": interrupted,
"interrupted_packages": packages,
"filesystems": _filesystems(sections),
}
@@ -166,11 +308,17 @@ class HostStatusMixin:
def get_host_status(self) -> HostStatusDict:
"""
Returns whether the host needs a reboot and whether it patches itself.
Returns whether the host needs a reboot, whether it patches itself, and
whether its package manager was left halfway.
* reboot_required (bool or None)
* reboot_reason (string or None)
* auto_updates (bool or None)
* auto_updates_last_success (Unix time or None)
* auto_updates_interval (seconds or None)
* package_manager_interrupted (bool or None)
* interrupted_packages (list of {name, state})
* filesystems (list of {mountpoint, total, used, available} in bytes, or None)
:raises ValueError: if the host's output carried no intact report.
"""
+14 -5
View File
@@ -14,6 +14,7 @@ from typing import Any, Dict, List, TYPE_CHECKING
from napalm_device_types.base import DeviceTypeDriver
from napalm_device_types.packages import PackageManagementMixin
from napalm_device_types.health_metrics import HealthMetricsMixin
from napalm_device_types.provisioning import QEMU_GUEST_AGENTS, GuestAgents
from napalm_device_types.models import (
NICConfigDict,
NetworkTargetDict,
@@ -44,11 +45,11 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
ROLE: str = "hypervisor"
TYPE_LABEL: str = "Hypervisor"
#: What cloud-init installs and starts on a VM provisioned through this
#: driver, so the hypervisor can read the guest's IP address back.
GUEST_AGENT_PACKAGES: tuple[str, ...] = ("qemu-guest-agent",)
GUEST_AGENT_RUNCMD: tuple[str, ...] = ("systemctl enable --now qemu-guest-agent",)
#: What cloud-init installs and runs on a VM provisioned through this
#: driver, per guest OS, so the hypervisor can read the guest's IP
#: address back. The guest operating systems listed here are the ones
#: the driver can provision; netOrk reads this off the class.
GUEST_AGENTS: GuestAgents = {"linux": QEMU_GUEST_AGENTS["linux"]}
# ------------------------------------------------------------------
@@ -464,6 +465,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
disk_resize_gb: int | None = None,
storage: str | None = None,
cpu_type: str | None = None,
guest_os: str = "linux",
download_timeout: int = 300,
timeout: int = 180,
) -> VMProvisionResultDict:
@@ -512,6 +514,13 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
value other than None with ValueError; one that has it raises
ValueError for a name it does not list or that is not
``available`` on this node, before creating anything.
guest_os (string) - the operating system in the image, a key of
``GUEST_AGENTS`` (``"linux"``, ``"freebsd"``, ``"openbsd"``). The
driver sets up the VM's hardware for it (OS type, how the guest
agent is attached) and, where the guest needs it, writes the
network-config itself (``provisioning.network_config``). A
guest_os the driver does not list raises ValueError before
anything is created. Default ``"linux"``.
download_timeout (int) - maximum seconds to wait for the image download
(skipped entirely if already cached on the hypervisor). Default 300.
timeout (int) - maximum seconds to wait for the remaining provisioning
+276
View File
@@ -0,0 +1,276 @@
# -*- coding: utf-8 -*-
"""Listening sockets: what listens on which address, and which service it is.
Whether a service is reachable from outside its host is decided by what it
listens on -- ``0.0.0.0:5432`` is, ``127.0.0.1:5432`` is not -- and that is read
the same way on every Linux host. So the command and its parse live here once,
and a driver only carries the command across.
**One round trip.** ``ss -lntup`` lists every listening TCP and bound UDP
socket with the processes holding it; for each of those processes,
``/proc/<pid>/cgroup`` says which systemd service or container it runs in. The
report is framed, and a report whose end is missing raises: a list cut short
must never read as sockets that closed.
**Root, and without it.** Only root sees every process behind a socket.
The whole script therefore goes to the host as one ``sh -c`` argument -- a
driver that prefixes ``sudo -n`` would otherwise run only its first command as
root. When that call brings no report back (no sudo, a wrong password), the
command runs again without privilege: the sockets are still worth having, and
the reading says it is not ``attributed``.
**No ``-H``.** iproute2 before 4.10 has no option to leave out the header and
fails on it, which would read as nothing listening. The parse skips the header
instead.
**Without ``ss``, ``netstat``.** OpenWrt's busybox and old net-tools hosts have
no ``ss``; ``netstat -lntup`` lists the same sockets with ``PID/Program``, and
the cgroups are read for its PIDs alike. On OpenWrt the cgroup names the procd
service (``/services/<name>/<instance>``), a jailed one too, whose PID is not the
one procd reports. A host with neither raises :class:`ListeningSocketsUnavailable`.
"""
from __future__ import annotations
import re
from shlex import quote
from typing import Dict, List, Optional, Tuple, TYPE_CHECKING
from napalm_device_types.models import ListeningSocketDict, ListeningSocketsDict
from napalm_device_types.terminal import strip_terminal_codes
_BEGIN = "SOCK_BEGIN"
_END = "SOCK_END"
_NO_SS = "no-ss"
_RC_RE = re.compile(r"^__SS_RC=(\d+)$")
#: One line, POSIX ``sh``, read-only. The frame markers are printed in two
#: halves so that a transport which echoes the command does not show them early.
#: ``ss`` is in ``/usr/sbin`` on some systems, outside a login user's ``PATH``.
#: ``netstat`` names a socket's process as ``PID/Program``, ``ss`` as ``pid=PID``.
LISTENING_SOCKETS_COMMAND = (
"PATH=$PATH:/usr/sbin:/sbin; "
"printf '%s%s\\n' SOCK_ BEGIN; "
"if command -v ss >/dev/null 2>&1; then "
"t=ss; s=$(ss -lntup 2>&1); r=$?; "
"pids=$(printf '%s\\n' \"$s\" | grep -o 'pid=[0-9]*' | cut -d= -f2); "
"elif command -v netstat >/dev/null 2>&1; then "
"t=netstat; s=$(netstat -lntup 2>&1); r=$?; "
"pids=$(printf '%s\\n' \"$s\" | grep -o ' [0-9][0-9]*/' | tr -d ' /'); "
"else t=; fi; "
"if [ -n \"$t\" ]; then "
"echo \"[$t]\"; printf '%s\\n' \"$s\"; echo \"__SS_RC=$r\"; "
"echo '[cgroups]'; "
"for p in $(printf '%s\\n' \"$pids\" | sort -u); do "
"sed \"s|^|$p |\" /proc/$p/cgroup 2>/dev/null; done; "
"else echo '[no-ss]'; fi; "
"printf '%s%s\\n' SOCK_ END"
)
_PROTOCOLS = frozenset({"tcp", "udp"})
#: netstat names the IPv6 sockets of net-tools ``tcp6``/``udp6``; busybox does not.
_NETSTAT_PROTOCOLS = {"tcp": "tcp", "tcp6": "tcp", "udp": "udp", "udp6": "udp"}
#: ``1604/dropbear``; net-tools prints ``700/sshd: /usr/sbin``.
_PROGRAM_RE = re.compile(r"^(\d+)/(\S*)")
#: ``users:(("nginx",pid=901,fd=6),("nginx",pid=900,fd=6))``
_USER_RE = re.compile(r'\("((?:[^"\\]|\\.)*)",pid=(\d+),fd=\d+\)')
#: The service a cgroup path runs in: its deepest ``*.service`` component.
_SERVICE_RE = re.compile(r"/([^/]+)\.service(?=/|$)")
#: OpenWrt's procd: ``/services/<name>/<instance>``.
_PROCD_RE = re.compile(r"^/services/([^/]+)(?:/|$)")
#: A container's cgroup: ``docker-<id>.scope`` (systemd driver), ``/docker/<id>`` (cgroupfs).
_CONTAINER_RE = re.compile(
r"(?:docker|libpod)-([0-9a-f]{64})\.scope|/(?:docker|libpod)/([0-9a-f]{64})(?=/|$)"
)
class ListeningSocketsUnavailable(NotImplementedError):
"""The host has neither ``ss`` nor ``netstat``; nothing to read, nothing to retry."""
def _frame(output: str) -> List[str]:
lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
try:
start = lines.index(_BEGIN)
end = lines.index(_END, start)
except ValueError:
raise ValueError("no intact listening socket report in the output") from None
return lines[start + 1 : end]
def _sections(lines: List[str]) -> Dict[str, List[str]]:
sections: Dict[str, List[str]] = {}
current: List[str] = []
for line in lines:
if line.startswith("[") and line.endswith("]") and " " not in line:
current = sections.setdefault(line[1:-1], [])
else:
current.append(line)
return sections
def _split_local(local: str) -> Optional[Tuple[str, Optional[str], int]]:
"""``[fe80::1%eth0]:546`` -> ``("fe80::1", "eth0", 546)``; None if no port."""
host, sep, port = local.rpartition(":")
if not sep or not port.isdigit():
return None
if host.startswith("[") and host.endswith("]"):
host = host[1:-1]
address, _, zone = host.partition("%")
return address or "*", zone or None, int(port)
def _cgroup_paths(lines: List[str]) -> Dict[int, str]:
"""Each process's cgroup path: the unified hierarchy, or systemd's under v1."""
paths: Dict[int, str] = {}
for line in lines:
pid, _, entry = line.partition(" ")
parts = entry.split(":", 2)
if not pid.isdigit() or len(parts) != 3:
continue
hierarchy, controllers, path = parts
if (hierarchy == "0" and controllers == "") or controllers == "name=systemd":
paths[int(pid)] = path
return paths
def _unit(path: Optional[str]) -> Optional[str]:
services = _SERVICE_RE.findall(path or "")
if services:
return str(services[-1])
procd = _PROCD_RE.match(path or "")
return str(procd.group(1)) if procd else None
def _container(path: Optional[str]) -> Optional[str]:
match = _CONTAINER_RE.search(path or "")
return (match.group(1) or match.group(2)) if match else None
def _entry(
proto: str,
local: Tuple[str, Optional[str], int],
process: Optional[str],
pid: Optional[int],
paths: Dict[int, str],
) -> ListeningSocketDict:
address, interface, port = local
path = paths.get(pid) if pid is not None else None
return {
"proto": proto,
"address": address,
"port": port,
"interface": interface,
"process": process,
"pid": pid,
"unit": _unit(path),
"container_id": _container(path),
}
def _ss_socket(line: str, paths: Dict[int, str]) -> Optional[ListeningSocketDict]:
parts = line.split()
if len(parts) < 5 or parts[0] not in _PROTOCOLS:
return None
local = _split_local(parts[4])
if local is None:
return None
users = _USER_RE.findall(line)
process, pid = (users[0][0], int(users[0][1])) if users else (None, None)
return _entry(parts[0], local, process, pid, paths)
def _netstat_socket(line: str, paths: Dict[int, str]) -> Optional[ListeningSocketDict]:
"""``Proto Recv-Q Send-Q Local Foreign [State] PID/Program`` -- a UDP line
has no state, and ``-`` is a socket without a process."""
parts = line.split()
proto = _NETSTAT_PROTOCOLS.get(parts[0]) if parts else None
if proto is None or len(parts) < 6:
return None
local = _split_local(parts[3])
if local is None:
return None
process, pid = None, None
for token in parts[5:7]:
program = _PROGRAM_RE.match(token)
if program:
pid, process = int(program.group(1)), program.group(2).rstrip(":") or None
break
return _entry(proto, local, process, pid, paths)
_PARSERS = {"ss": _ss_socket, "netstat": _netstat_socket}
def parse_listening_sockets(output: str) -> List[ListeningSocketDict]:
"""Parse what :data:`LISTENING_SOCKETS_COMMAND` printed, sorted by protocol,
port and address.
:raises ListeningSocketsUnavailable: when the host has neither ``ss`` nor ``netstat``.
:raises ValueError: when the output carries no intact report, or the tool failed.
"""
sections = _sections(_frame(output))
if _NO_SS in sections:
raise ListeningSocketsUnavailable("the host has neither ss nor netstat")
tool = "netstat" if "netstat" in sections else "ss"
lines = sections.get(tool, [])
statuses = [m.group(1) for m in map(_RC_RE.match, lines) if m]
if not statuses or statuses[-1] != "0":
detail = " ".join(line for line in lines if not _RC_RE.match(line))[:200]
raise ValueError(f"{tool} did not list the sockets: {detail or 'no exit status'}")
paths = _cgroup_paths(sections.get("cgroups", []))
parse = _PARSERS[tool]
sockets = [s for s in (parse(line, paths) for line in lines) if s is not None]
return sorted(sockets, key=lambda s: (s["proto"], s["port"], s["address"], s["interface"] or ""))
class ListeningSocketsMixin:
"""Adds :meth:`get_listening_sockets` to a driver that can run a command on a Linux host.
With ``ss``, or ``netstat`` where there is none (OpenWrt's busybox).
The template form (README, "Function classes"): the command and its parse
are the same everywhere, so they are concrete here, and a driver supplies
only :meth:`_run_listening_sockets_command` -- how a command reaches its
host, and how it gains root there. Mixed in by the drivers that can, not by
:class:`~napalm_device_types.os.OSDriver`: a Windows host is an OS driver
too, and ``hasattr(driver, "get_listening_sockets")`` has to stay truthful.
"""
if TYPE_CHECKING: # pragma: no cover - declared for type checkers only
def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
"""Run *command* on the host and return what it printed; as root
when *privileged*. The command is a single ``sh -c`` invocation."""
...
def get_listening_sockets(self) -> ListeningSocketsDict:
"""
Returns every listening TCP and bound UDP socket, with the process,
systemd service and container behind it.
* attributed (bool) - read as root, so every process is named
* sockets (list) - see :class:`~napalm_device_types.models.ListeningSocketDict`
Example::
{
"attributed": True,
"sockets": [
{"proto": "tcp", "address": "0.0.0.0", "port": 5432,
"interface": None, "process": "postgres", "pid": 812,
"unit": "postgresql@16-main", "container_id": None},
],
}
:raises ListeningSocketsUnavailable: if the host has neither ``ss`` nor ``netstat``.
:raises ValueError: if neither reading carried an intact report.
"""
command = f"sh -c {quote(LISTENING_SOCKETS_COMMAND)}"
try:
output = self._run_listening_sockets_command(command, privileged=True)
return {"attributed": True, "sockets": parse_listening_sockets(output)}
except ValueError:
pass
output = self._run_listening_sockets_command(command, privileged=False)
return {"attributed": False, "sockets": parse_listening_sockets(output)}
+101 -54
View File
@@ -75,6 +75,27 @@ class UpdateDict(TypedDict):
security: NotRequired[Optional[bool]]
class InterruptedPackageDict(TypedDict):
"""A package an interrupted package manager run left halfway."""
name: str
#: dpkg's word for it: "half-installed", "unpacked", "half-configured",
#: "triggers-awaited", "triggers-pending", or "reinstall-required".
state: str
class FilesystemUsageDict(TypedDict):
"""How full one filesystem is, as ``df -Pk`` says, in bytes."""
#: Where it is mounted, e.g. "/" or "/boot".
mountpoint: str
total: int
used: int
#: What an ordinary user can still write: less than total minus used on a
#: filesystem that keeps blocks for root (ext4 keeps 5 %).
available: int
class HostStatusDict(TypedDict):
"""What a host says about its own patch state (``HostStatusMixin.get_host_status``)."""
@@ -84,6 +105,19 @@ class HostStatusDict(TypedDict):
reboot_reason: Optional[str]
#: True when the host installs updates on its own (unattended-upgrades, dnf-automatic).
auto_updates: Optional[bool]
#: True when a package manager run was cut off and has to be finished
#: (``dpkg --configure -a``) before anything else installs; None when it
#: cannot tell, or a package manager is running right now.
package_manager_interrupted: NotRequired[Optional[bool]]
#: The packages that run left halfway; empty when only its journal says so.
interrupted_packages: NotRequired[List[InterruptedPackageDict]]
#: When the host's own automatic updates last ran successfully (Unix time).
auto_updates_last_success: NotRequired[Optional[int]]
#: How often they are meant to run, in seconds; 0 for every time the timer fires.
auto_updates_interval: NotRequired[Optional[int]]
#: How full ``/``, ``/boot`` and ``/var`` are, each filesystem once; None when
#: it could not be read.
filesystems: NotRequired[Optional[List[FilesystemUsageDict]]]
# ---------------------------------------------------------------------------
@@ -335,6 +369,66 @@ class KernelFactsDict(TypedDict):
config: Optional[Dict[str, str]] # build configuration, set options only; quotes stripped
class ListeningSocketDict(TypedDict):
"""A TCP socket that listens, or a UDP socket that is bound, on the host.
One entry per socket as ``ss`` lists it. ``address`` is printed the way ss
prints it, without brackets or zone: ``0.0.0.0`` and ``::`` are every
address of their family, ``*`` every address of both. Whether that is
reachable from outside the host is the consumer's call.
"""
proto: str # "tcp" or "udp"
address: str # "0.0.0.0", "::", "*", "127.0.0.1", "::ffff:127.0.0.1", ...
port: int
interface: Optional[str] # the %zone a socket is bound to ("lo", "eth0"), if any
process: Optional[str] # the first process holding it; None without one or without root
pid: Optional[int]
unit: Optional[str] # the process's systemd service, without ".service"
container_id: Optional[str] # the full container ID, for a container on the host network
class ListeningSocketsDict(TypedDict):
"""What ``ListeningSocketsMixin.get_listening_sockets`` read.
``attributed`` is False when the reading ran without root: ``ss`` then names
only the login user's own processes, so a socket without a process means
"not told", not "the kernel's".
"""
attributed: bool
sockets: List[ListeningSocketDict]
class VirtualIpDict(TypedDict):
"""An address the host's HA configuration lets float between machines.
Declared, not observed: whether the host holds the address right now is in
its interface addresses, not here. Only these fields leave the host -- never
a password, an etcd endpoint or a notify script.
"""
mechanism: str # "vip-manager" | "keepalived"
instance: str # the vrrp_instance name, or the vip-manager unit without ".service"
address: str # normalised: "10.7.224.10", "fd00::10"
prefix_length: Optional[int] # None where the configuration gives none
interface: Optional[str]
group_key: Optional[str] # vip-manager's trigger-key, or "vrid:<n>" for keepalived
running: Optional[bool] # the unit is active
class VirtualIpsDict(TypedDict):
"""What ``VirtualIpsMixin.get_virtual_ips`` read, per mechanism.
A list is what was looked at -- ``[]`` a host without that mechanism. None is
"did not look": a configuration it could not read, an include it could not
follow, a host without systemd. A consumer keeps what it knew for a None.
"""
vip_manager: Optional[List[VirtualIpDict]]
keepalived: Optional[List[VirtualIpDict]]
class PortForwardDict(TypedDict):
"""A port the WAN side can reach, forwarded to a host inside.
@@ -782,62 +876,15 @@ class ApplyUpdatesResultDict(TypedDict):
error: NotRequired[str] # error message if success is False
class DockerContainerDict(TypedDict):
"""A Docker container entry as returned by ``docker ps -a``."""
class ContainerEngineDict(TypedDict):
"""One container engine a host offers (``container_engines()``).
id: str # short container ID
name: str # container name(s)
image: str # image reference
image_version: str # OCI org.opencontainers.image.version label; empty if absent
command: str # entrypoint / command
created: str # creation timestamp string
status: str # human-readable status (e.g. "Up 3 hours")
ports: str # port mapping string
state: str # "running", "exited", "paused", etc.
Only how to reach it: which engine, and which API it speaks. What runs on
it is read and modelled above the driver, in netOrk (netork#765).
"""
class DockerImageDict(TypedDict):
"""A local Docker image entry as returned by ``docker images``."""
id: str # short image ID
repository: str # image repository
tag: str # image tag
size: str # human-readable size string (e.g. "187MB")
created: str # creation timestamp string
version: str # OCI org.opencontainers.image.version label; empty if absent
class DockerVolumeDict(TypedDict):
"""A Docker volume entry as returned by ``docker volume ls``."""
name: str # volume name
driver: str # volume driver (e.g. "local")
mountpoint: str # host filesystem path
scope: str # "local" or "global"
class DockerNetworkDict(TypedDict):
"""A Docker network entry as returned by ``docker network ls``."""
id: str # short network ID
name: str # network name
driver: str # network driver (e.g. "bridge", "host", "overlay")
scope: str # network scope
ipv6: str # "true" if IPv6 is enabled, "false" otherwise
internal: str # "true" if the network is internal, "false" otherwise
class DockerInfoDict(TypedDict):
"""Return value of ``get_docker_info()``."""
available: bool # False if Docker is absent/inaccessible
permission_denied: NotRequired[bool] # True when socket access is denied
version: NotRequired[str] # Docker Engine version string
containers: NotRequired[List[DockerContainerDict]]
images: NotRequired[List[DockerImageDict]]
volumes: NotRequired[List[DockerVolumeDict]]
networks: NotRequired[List[DockerNetworkDict]]
outdated_images: NotRequired[List[str]] # image names with a newer remote digest
engine: str # "docker"
api: str # "docker-engine": the Docker Engine API over ``system dial-stdio``
class DeviceActionResultDict(TypedDict):
-92
View File
@@ -19,7 +19,6 @@ from napalm_device_types.health_metrics import HealthMetricsMixin
from napalm_device_types.models import (
CronJobDict,
DeviceActionResultDict,
DockerInfoDict,
ProcessDict,
SNMPConfigDict,
UserDict,
@@ -193,97 +192,6 @@ class OSDriver(UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthM
"""
...
# ------------------------------------------------------------------
# Docker
# ------------------------------------------------------------------
def get_docker_info(self) -> DockerInfoDict:
"""
Returns information about the local Docker environment.
If Docker is not installed or the current user lacks access to the
Docker socket, returns ``{"available": False}``. When the user has
no socket permission, ``permission_denied`` is additionally set to
``True``.
When Docker is available the dict contains:
* available (bool) - always ``True``
* version (string) - Docker Engine version string
* containers (list) - all containers (running and stopped), each with:
* id (string) - short container ID
* name (string) - container name(s)
* image (string) - image reference
* image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
* command (string) - entrypoint / command string
* created (string) - creation timestamp string
* status (string) - human-readable status (e.g. ``"Up 3 hours"``)
* ports (string) - port mapping string
* state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
* images (list) - local images, each with:
* id (string) - short image ID
* repository (string) - image repository
* tag (string) - image tag
* size (string) - human-readable size string (e.g. ``"187MB"``)
* created (string) - creation timestamp string
* version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
* volumes (list) - Docker volumes, each with:
* name (string) - volume name
* driver (string) - volume driver
* mountpoint (string) - host filesystem path
* scope (string) - ``"local"`` or ``"global"``
* networks (list) - Docker networks, each with:
* id (string) - short network ID
* name (string) - network name
* driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
* scope (string) - network scope
* ipv6 (string) - ``"true"`` if IPv6 is enabled
* internal (string) - ``"true"`` if the network is internal
* outdated_images (list of strings) - image names where the local digest
differs from the latest remote digest; empty list if all images are
current or update checks could not be performed.
Example::
# Docker not installed:
{"available": False}
# Docker installed, no socket permission:
{"available": False, "permission_denied": True}
# Docker available:
{
"available": True,
"version": "Docker version 27.3.1, build ce12230",
"containers": [
{
"id": "a1b2c3d4e5f6",
"name": "my-app",
"image": "nginx:latest",
"image_version": "1.27.0",
"command": "nginx -g 'daemon off;'",
"created": "2026-05-28 10:00:00 +0000 UTC",
"status": "Up 3 days",
"ports": "0.0.0.0:80->80/tcp",
"state": "running",
},
],
"images": [...],
"volumes": [],
"networks": [...],
"outdated_images": ["nginx:latest"],
}
"""
...
# ------------------------------------------------------------------
# Generic device actions
# ------------------------------------------------------------------
+64
View File
@@ -0,0 +1,64 @@
"""
What provisioning a VM from a cloud image needs to know about the guest.
The same for every hypervisor: which agent a guest runs so a QEMU-based
hypervisor can read its IP, the network-config cloud-init and FreeBSD's
nuageinit both read, and how a packed cloud image is unpacked. How a
hypervisor attaches, boots and talks to the VM stays in its driver.
"""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
#: ``{guest_os: (packages, runcmd)}``: what cloud-init installs and runs so
#: the hypervisor can read the guest's IP address back.
GuestAgents = Mapping[str, tuple[tuple[str, ...], tuple[str, ...]]]
#: The QEMU guest agent per guest OS, for hypervisors built on QEMU. Package
#: names and service commands were verified on FreeBSD 15.1 and OpenBSD 7.9
#: cloud images (NetOrk/netork#793).
QEMU_GUEST_AGENTS: GuestAgents = {
"linux": (("qemu-guest-agent",), ("systemctl enable --now qemu-guest-agent",)),
"freebsd": (
("qemu-guest-agent",),
("sysrc qemu_guest_agent_enable=YES", "service qemu-guest-agent start"),
),
"openbsd": (("qemu-ga",), ("rcctl enable qemu_ga", "rcctl start qemu_ga")),
}
#: Suffix of a packed image -> the command that writes it unpacked to stdout.
_DECOMPRESSORS = {
"xz": "xz -dc",
"gz": "gzip -dc",
"bz2": "bzip2 -dc",
"zst": "zstd -dc",
}
def network_config(nics: list[tuple[str, bool]]) -> dict[str, Any] | None:
"""Network-config v2: DHCP on every ``(mac, dhcp)`` NIC that asks for it.
Version 2 because FreeBSD's nuageinit reads no other: given the v1 that
Proxmox generates, it fails and skips the rest of its first stage,
runcmd included. ``None`` when no NIC wants DHCP, so the guest is left
alone rather than told to configure nothing.
"""
ethernets = {
f"nic{i}": {"match": {"macaddress": mac.lower()}, "dhcp4": True}
for i, (mac, dhcp) in enumerate(nics)
if dhcp
}
return {"version": 2, "ethernets": ethernets} if ethernets else None
def split_compression(filename: str) -> tuple[str, str | None]:
"""``(unpacked filename, command that unpacks to stdout)`` for *filename*.
The command is None for an image that is not packed, which is then
imported as it is.
"""
stem, dot, suffix = filename.rpartition(".")
command = _DECOMPRESSORS.get(suffix.lower()) if dot and stem else None
return (stem, command) if command else (filename, None)
+435
View File
@@ -0,0 +1,435 @@
# -*- coding: utf-8 -*-
"""OpenSSH server hardening: taking algorithms out of what sshd offers.
A scanner such as ssh-audit names algorithms a server should stop offering --
``hmac-sha1``, the NIST curves. Taking them out is the same on every host that
runs OpenSSH, so the reading, the plan and the write live here once, and a
driver only mixes in :class:`OpenSshServerMixin` and supplies ``run_command``.
**What the server offers is read, not assumed.** ``sshd -T`` prints the lists
sshd actually uses, after its defaults, ``sshd_config``, every ``Include`` and
first-match-wins. The new list is that one minus what was asked, so nothing the
host already restricted comes back, and nothing it allows is dropped by
accident.
**A drop-in, not an edit.** Where ``sshd_config`` includes
``/etc/ssh/sshd_config.d/*.conf`` -- Debian and Ubuntu do, at the top -- the
lists go to :data:`DROPIN_PATH`, which sorts first and so wins. sshd_config
stays as the distribution shipped it, and removing the file undoes the change.
Only without such an ``Include`` is sshd_config itself rewritten, after a copy.
**Nobody is locked out, by three checks.**
1. The plan never removes the last algorithm of a class, and -- given the
lists of the client that manages the host -- never the last one that client
speaks. Each one kept back is reported, with why.
2. ``sshd -t`` must accept the result, and ``sshd -T`` must show the new lists.
A file read before ours can set the same directive; ours is then valid and
ignored, and that is reported rather than called a success.
3. After the reload, *verify* -- the caller's -- opens a new session. The
running one survives a reload, so if the new one cannot get in, this one
puts the old state back and reloads again.
"""
from __future__ import annotations
import time
from dataclasses import dataclass, field
from shlex import quote
from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple, TYPE_CHECKING
if TYPE_CHECKING: # pragma: no cover - for type checkers only
from napalm_device_types.channel import CommandResult
#: The algorithm classes: key -> (``sshd -T`` keyword, sshd_config directive).
SSH_ALGORITHM_CLASSES: Dict[str, Tuple[str, str]] = {
"kex": ("kexalgorithms", "KexAlgorithms"),
"hostkey": ("hostkeyalgorithms", "HostKeyAlgorithms"),
"cipher": ("ciphers", "Ciphers"),
"mac": ("macs", "MACs"),
}
MAIN_CONFIG = "/etc/ssh/sshd_config"
DROPIN_DIR = "/etc/ssh/sshd_config.d"
#: ``00-`` so it sorts before every other drop-in: sshd keeps the first value it reads.
DROPIN_PATH = f"{DROPIN_DIR}/00-netork-ssh-hardening.conf"
#: What marks a line this module disabled in sshd_config.
DISABLED_PREFIX = "# netork: "
_HEADER = (
"# Written by netOrk: the SSH algorithms a security assessment flagged,\n"
"# taken out of what this server offers. Delete this file to undo it.\n"
)
_RELOAD = (
"if systemctl is-active --quiet ssh 2>/dev/null; then systemctl reload ssh;"
" elif systemctl is-active --quiet sshd 2>/dev/null; then systemctl reload sshd;"
" else for f in /run/sshd.pid /var/run/sshd.pid; do"
' [ -r "$f" ] && kill -HUP "$(cat "$f")" && exit 0; done; exit 0; fi'
)
def _directive_key(line: str) -> Optional[str]:
"""The algorithm class a global sshd_config line sets, if it sets one."""
parts = line.split(None, 1)
if not parts:
return None
word = parts[0].lower()
for kind, (keyword, _directive) in SSH_ALGORITHM_CLASSES.items():
if word == keyword:
return kind
return None
def parse_sshd_effective(text: str) -> Dict[str, List[str]]:
"""The four algorithm lists from ``sshd -T`` output, in the server's order.
:raises ValueError: if a list is missing -- not sshd -T output, or cut short.
"""
found: Dict[str, List[str]] = {}
for raw in text.splitlines():
parts = raw.strip().split(None, 1)
if len(parts) != 2:
continue
kind = _directive_key(parts[0])
if kind and kind not in found:
found[kind] = [a for a in parts[1].split(",") if a]
missing = [k for k in SSH_ALGORITHM_CLASSES if k not in found]
if missing:
raise ValueError(f"sshd -T printed no {', '.join(missing)} list")
return found
@dataclass(frozen=True)
class Refusal:
"""One algorithm that was asked to go and stays, with why.
*reason* is ``"last"`` (the class would be empty), ``"client"`` (the managing
client would share none of the class with the server) or ``"not_offered"``
(the server does not offer it, so there is nothing to remove).
"""
kind: str
algorithm: str
reason: str
@dataclass(frozen=True)
class SshHardeningPlan:
keep: Dict[str, List[str]]
remove: Dict[str, List[str]]
refused: List[Refusal] = field(default_factory=list)
not_offered: List[Refusal] = field(default_factory=list)
@property
def changed_classes(self) -> List[str]:
return [k for k in SSH_ALGORITHM_CLASSES if self.remove.get(k)]
def plan_ssh_hardening(
effective: Mapping[str, Sequence[str]],
remove: Mapping[str, Iterable[str]],
*,
client: Optional[Mapping[str, Iterable[str]]] = None,
) -> SshHardeningPlan:
"""What the lists become when *remove* is taken out of *effective*.
*client* is what the client managing this host speaks, per class. A class it
has no list for is not guarded beyond "never empty".
When a removal would leave a class empty, or without an algorithm *client*
speaks, the first of the asked-for algorithms in the server's order that
fixes it stays, and is reported in ``refused``.
:raises ValueError: for a class that is not one of :data:`SSH_ALGORITHM_CLASSES`.
"""
unknown = [k for k in remove if k not in SSH_ALGORITHM_CLASSES]
if unknown:
raise ValueError(f"unknown SSH algorithm class: {', '.join(unknown)}")
keep: Dict[str, List[str]] = {}
removed: Dict[str, List[str]] = {}
refused: List[Refusal] = []
not_offered: List[Refusal] = []
for kind in SSH_ALGORITHM_CLASSES:
offered = list(effective.get(kind, []))
asked = set(remove.get(kind, []))
not_offered += [Refusal(kind, a, "not_offered") for a in sorted(asked - set(offered))]
going = [a for a in offered if a in asked]
staying = [a for a in offered if a not in asked]
speaks = set(client[kind]) if client and kind in client else None
if going and speaks is not None and not speaks & set(staying):
spared = next((a for a in going if a in speaks), None)
if spared:
going.remove(spared)
refused.append(Refusal(kind, spared, "client"))
if going and not [a for a in offered if a not in going]:
spared = going.pop(0)
refused.append(Refusal(kind, spared, "last"))
keep[kind] = [a for a in offered if a not in going]
if going:
removed[kind] = going
return SshHardeningPlan(keep=keep, remove=removed, refused=refused, not_offered=not_offered)
def render_dropin(keep: Mapping[str, Sequence[str]], kinds: Iterable[str]) -> str:
"""The drop-in file stating the full list of each class in *kinds*."""
lines = [
f"{SSH_ALGORITHM_CLASSES[k][1]} {','.join(keep[k])}\n"
for k in SSH_ALGORITHM_CLASSES
if k in set(kinds)
]
return _HEADER + "".join(lines)
def dropin_classes(text: str) -> List[str]:
"""The classes an existing drop-in sets."""
return [k for k in (_directive_key(line.strip()) for line in text.splitlines()) if k]
def _is_match(line: str) -> bool:
parts = line.split(None, 1)
return bool(parts) and parts[0].lower() == "match"
def has_dropin_include(text: str) -> bool:
"""Whether sshd_config's global section includes :data:`DROPIN_DIR`."""
for raw in text.splitlines():
line = raw.strip()
if _is_match(line):
return False
parts = line.split()
if len(parts) >= 2 and parts[0].lower() == "include":
if any(p.startswith(DROPIN_DIR + "/") for p in parts[1:]):
return True
return False
def _split_our_block(text: str) -> Tuple[List[str], str]:
"""(the classes an earlier rewrite put above the file, the file without them)."""
if not text.startswith(_HEADER):
return [], text
lines = text[len(_HEADER):].splitlines(keepends=True)
ours = 0
while ours < len(lines) and _directive_key(lines[ours].strip()):
ours += 1
return dropin_classes("".join(lines[:ours])), "".join(lines[ours:])
def render_main_config(text: str, keep: Mapping[str, Sequence[str]], kinds: Iterable[str]) -> str:
"""sshd_config with the lists of *kinds* first and their old global lines disabled.
First because sshd keeps the first value it reads. A ``Match`` block is left
alone -- it is not the global setting. A class an earlier rewrite set stays
set, and rewriting a file this already rewrote gives the same file.
"""
earlier, body = _split_our_block(text)
wanted = set(kinds) | set(earlier)
out: List[str] = []
in_match = False
for raw in body.splitlines():
if _is_match(raw.strip()):
in_match = True
if not in_match and _directive_key(raw.strip()) in wanted:
out.append(DISABLED_PREFIX + raw)
else:
out.append(raw)
return render_dropin(keep, wanted) + "\n".join(out) + ("\n" if body.endswith("\n") else "")
@dataclass
class SshHardeningResult:
"""What :meth:`OpenSshServerMixin.harden_ssh_server` did.
*applied* is True only when the new lists are live and a new session got
in. *rolled_back* says the host was changed and then put back. *backup* is
where sshd_config was copied to, when it was the file changed.
"""
plan: SshHardeningPlan
applied: bool = False
rolled_back: bool = False
path: Optional[str] = None
backup: Optional[str] = None
message: str = ""
class OpenSshServerMixin:
"""Adds :meth:`harden_ssh_server` to a driver that can run commands as root on an OpenSSH host.
Mixed in by the drivers whose host runs a stock OpenSSH, not by a base class,
so ``hasattr(driver, "harden_ssh_server")`` stays a truthful answer. A driver
whose platform writes sshd_config itself (an appliance that regenerates it
from its own database) must not inherit this, or must opt out with
``harden_ssh_server = None``: its file would be overwritten on the next
apply.
"""
if TYPE_CHECKING: # pragma: no cover - declared for type checkers only
def run_command(
self,
command: str,
*,
privileged: bool = False,
timeout: float = 60,
stdin: Optional[bytes] = None,
) -> CommandResult: ...
# ── the host ────────────────────────────────────────────────────────
def _ssh_run(self, command: str, stdin: Optional[bytes] = None) -> "CommandResult":
return self.run_command(command, privileged=True, timeout=30, stdin=stdin)
def _ssh_read(self, path: str) -> Optional[str]:
result = self._ssh_run(f"cat -- {quote(path)}")
return result.stdout if result.exit_code == 0 else None
def _ssh_write(self, path: str, content: str) -> None:
script = (
f'tmp={quote(path + ".netork-tmp")} && cat > "$tmp" && chmod 644 "$tmp"'
f' && mv -f -- "$tmp" {quote(path)} || {{ s=$?; rm -f -- "$tmp"; exit $s; }}'
)
result = self._ssh_run(script, stdin=content.encode())
if result.exit_code != 0:
raise RuntimeError(f"could not write {path}: {(result.stderr or result.stdout).strip()}")
def _ssh_remove(self, path: str) -> None:
self._ssh_run(f"rm -f -- {quote(path)}")
def _reload_ssh_server(self) -> None:
"""Make the running sshd read its configuration again; sessions stay up."""
result = self._ssh_run(_RELOAD)
if result.exit_code != 0:
raise RuntimeError(f"could not reload sshd: {(result.stderr or result.stdout).strip()}")
def get_ssh_server_algorithms(self) -> Dict[str, List[str]]:
"""The algorithm lists sshd uses now (``sshd -T``), per class.
:raises RuntimeError: if sshd -T fails -- no OpenSSH, or no root.
"""
result = self._ssh_run("PATH=$PATH:/usr/sbin:/sbin sshd -T")
if result.exit_code != 0:
raise RuntimeError(f"sshd -T failed: {(result.stderr or result.stdout).strip()}")
return parse_sshd_effective(result.stdout)
def _sshd_accepts(self) -> Optional[str]:
"""None when ``sshd -t`` accepts the configuration, else what it said."""
result = self._ssh_run("PATH=$PATH:/usr/sbin:/sbin sshd -t")
return None if result.exit_code == 0 else (result.stderr or result.stdout).strip()
# ── the change ──────────────────────────────────────────────────────
def harden_ssh_server(
self,
remove: Mapping[str, Iterable[str]],
*,
client: Optional[Mapping[str, Iterable[str]]] = None,
verify: Optional[Callable[[], bool]] = None,
log: Optional[Callable[[str], None]] = None,
) -> SshHardeningResult:
"""Stop the server offering the algorithms in *remove* (class -> names).
*client* guards the managing client's algorithms (see
:func:`plan_ssh_hardening`); *verify* opens a new session after the
reload and returns whether it got in; *log* receives a line per step.
Never raises after the host was touched: a failure from there on puts
the previous state back and says so in the result.
:raises RuntimeError: if the current state cannot be read.
"""
say = log or (lambda _line: None)
say("Reading the algorithms sshd offers (sshd -T)")
plan = plan_ssh_hardening(self.get_ssh_server_algorithms(), remove, client=client)
for r in plan.refused:
why = "it is the last one" if r.reason == "last" else "netOrk could no longer connect"
say(f"Keeping {r.algorithm}: {why}")
kinds = plan.changed_classes
result = SshHardeningResult(plan=plan)
if not kinds:
result.message = "Nothing to change: the server offers none of the selected algorithms."
say(result.message)
return result
for kind in kinds:
say(f"Removing {', '.join(plan.remove[kind])} from {SSH_ALGORITHM_CLASSES[kind][1]}")
main = self._ssh_read(MAIN_CONFIG)
if main is None:
raise RuntimeError(f"cannot read {MAIN_CONFIG}")
restore = self._write_change(plan, main, result, say)
return self._apply_or_restore(plan, kinds, result, restore, verify, say)
def _write_change(
self, plan: SshHardeningPlan, main: str, result: SshHardeningResult, say: Callable[[str], None]
) -> Callable[[], None]:
"""Write the new lists; return what puts the previous state back."""
if has_dropin_include(main):
previous = self._ssh_read(DROPIN_PATH)
kinds = set(plan.changed_classes) | set(dropin_classes(previous or ""))
self._ssh_run(f"mkdir -p -- {quote(DROPIN_DIR)}")
self._ssh_write(DROPIN_PATH, render_dropin(plan.keep, kinds))
result.path = DROPIN_PATH
say(f"Wrote {DROPIN_PATH}")
def restore() -> None:
if previous is None:
self._ssh_remove(DROPIN_PATH)
else:
self._ssh_write(DROPIN_PATH, previous)
return restore
backup = f"{MAIN_CONFIG}.netork-{time.strftime('%Y%m%d-%H%M%S')}.bak"
self._ssh_write(backup, main)
self._ssh_write(MAIN_CONFIG, render_main_config(main, plan.keep, plan.changed_classes))
result.path, result.backup = MAIN_CONFIG, backup
say(f"{MAIN_CONFIG} has no Include of {DROPIN_DIR}; rewrote it (copy at {backup})")
def restore_main() -> None:
self._ssh_write(MAIN_CONFIG, main)
return restore_main
def _apply_or_restore(
self,
plan: SshHardeningPlan,
kinds: List[str],
result: SshHardeningResult,
restore: Callable[[], None],
verify: Optional[Callable[[], bool]],
say: Callable[[str], None],
) -> SshHardeningResult:
def back(message: str, reload: bool) -> SshHardeningResult:
say(message)
say("Putting the previous configuration back")
restore()
if reload:
self._reload_ssh_server()
result.rolled_back = True
result.message = message
return result
try:
say("Checking the configuration (sshd -t)")
complaint = self._sshd_accepts()
if complaint:
return back(f"sshd rejected the new configuration: {complaint}", reload=False)
now = self.get_ssh_server_algorithms()
ignored = [SSH_ALGORITHM_CLASSES[k][1] for k in kinds if now[k] != plan.keep[k]]
if ignored:
return back(
f"sshd does not use the new {', '.join(ignored)}: a setting read before"
f" {result.path} decides them",
reload=False,
)
say("Reloading sshd")
self._reload_ssh_server()
if verify is not None:
say("Opening a new SSH session to check netOrk still gets in")
if not verify():
return back("A new SSH session could not log in after the reload", reload=True)
except Exception as exc: # noqa: BLE001 - whatever went wrong, the host goes back
return back(f"The change failed: {exc}", reload=True)
result.applied = True
result.message = "The server no longer offers the selected algorithms."
say(result.message)
return result
+406
View File
@@ -0,0 +1,406 @@
# -*- coding: utf-8 -*-
"""Virtual IPs: the addresses a host's HA configuration lets float between machines.
vip-manager and keepalived put a service address on whichever member is master
*right now*. The address list of a standby shows no trace of it; what every
member has is the configuration that declares it. That is read the same way on
every Linux host, so the command and its parse live here once, and a driver
only carries the command across.
**Declared, not held.** The reading says which addresses the configuration
lets onto this host. Whether the host holds one at the moment is in its
interface addresses, which the caller already has.
**Filtered on the host.** The same files hold secrets -- vip-manager's etcd
password, keepalived's ``auth_pass``, notify scripts with their arguments. The
command therefore never prints a configuration: an awk program reads the unit's
properties and files and prints allowlisted fields only. A secret cannot reach
the caller's memory, its log or an exception text, because it never leaves the
host.
**What is read.**
- *vip-manager*: every ``vip-manager.service`` and ``vip-manager@*.service`` unit
systemd knows. The address comes from the first of: a command-line flag
(``--ip``; 1.x's ``-ip=${VIP_IP}`` resolved through the unit's environment), the
environment (``VIP_IP``, an ``EnvironmentFile`` over ``Environment``), the
``--config`` YAML (5.x's ``ip``). Mask, interface and trigger key alike.
- *keepalived*: ``keepalived.service``, its ``-f``/``--use-file`` or
``/etc/keepalived/keepalived.conf``. The ``virtual_ipaddress`` and
``virtual_ipaddress_excluded`` entries of every ``vrrp_instance``, with the
instance's ``interface`` and ``virtual_router_id``. ``include`` is followed
four levels deep, a relative pattern from the including file's directory. A
pattern that is not a plain path glob is not followed: it would otherwise be
handed to a root shell.
**None is "did not look".** A file that exists but cannot be read, an include
that cannot be followed, a host without systemd: that mechanism is None, and the
caller keeps what it knew. ``[]`` is a host that has none.
**Root, and without it.** The configuration may be root's alone. The whole
script goes to the host as one ``sh -c`` argument; when the privileged call
brings no report back, it runs again without privilege and reads what it can.
**An exec channel, not a terminal.** With its awk program the command is about
4 kB, past the line an interactive shell reliably takes. A driver sends it the
way it runs any command that needs no PTY -- napalm-linux's ``run_command``.
"""
from __future__ import annotations
import ipaddress
import re
from dataclasses import dataclass, field
from shlex import quote
from typing import Callable, Dict, List, Optional, Set, Tuple, TYPE_CHECKING
from napalm_device_types.models import VirtualIpDict, VirtualIpsDict
from napalm_device_types.terminal import strip_terminal_codes
_BEGIN = "VIPS_BEGIN"
_END = "VIPS_END"
_NO_SYSTEMD = "no-systemd"
#: A record that says a mechanism was not fully read.
_BLIND = frozenset({"unreadable", "too-deep"})
_KEEPALIVED_CONF = "/etc/keepalived/keepalived.conf"
_ENV_KEYS = "VIP_(IP|MASK|NETMASK|IFACE|INTERFACE|KEY|TRIGGER_KEY)"
_CFG_KEYS = "(ip|netmask|mask|interface|iface|trigger-key|key)"
#: One line of awk, fed a unit's ``systemctl show`` output with ``-v m=<mechanism>``.
#: It prints records, never a line of configuration: ``active <state>``,
#: ``arg <flag> <value>``, ``env <VIP_*> <value>``, ``cfg <key> <value>``,
#: ``instance <name>`` / ``interface`` / ``virtual_router_id`` / ``vip <addr> [dev <if>]``
#: / ``end``, and ``unreadable``/``too-deep``/``missing <path>``. No ``>`` but in
#: ``2>/dev/null``: comparisons are written the other way round. ``\047`` is a
#: single quote, so the program itself needs none.
_AWK = " ".join(
(
r'function q(p) { return "\047" p "\047" }',
r'function ok(f) { if (f == "" || f ~ /\047/) { print "unreadable " f; return 0 }'
r' if (system("test -r " q(f)) == 0) return 1;'
r' if (system("test -e " q(f)) == 0) print "unreadable " f; else print "missing " f;'
r" return 0 }",
r"function rf(f, c, l, r, k) { if (!ok(f)) return;"
r" while (0 < (r = (getline l < f))) {"
rf' if (c == "cfg" && l ~ /^[ \t]*{_CFG_KEYS}[ \t]*:/) {{ sub(/^[ \t]*/, "", l); k = l;'
r' sub(/[ \t]*:.*/, "", k); sub(/^[^:]*:[ \t]*/, "", l); print "cfg " k " " l }'
rf' else if (c == "env" && l ~ /^[ \t]*(export[ \t]+)?{_ENV_KEYS}[ \t]*=/) {{'
r' sub(/^[ \t]*(export[ \t]+)?/, "", l); k = l; sub(/[ \t]*=.*/, "", k);'
r' sub(/^[^=]*=[ \t]*/, "", l); print "env " k " " l } }'
r' if (r < 0) print "unreadable " f; close(f) }',
r'function push(n, a) { st[++sd] = n; if (n == "vrrp_instance") print "instance " a }',
r'function pop() { if (0 < sd) { if (st[sd] == "vrrp_instance") print "end"; sd-- } }',
r"function stmt(w, nw, f, d, x, i) { if (!nw) return;"
r' if (w[1] == "include") { if (1 < nw) inc(w[2], f, d); return }'
r' if (sd == 1 && st[1] == "vrrp_instance") {'
r' if (1 < nw && (w[1] == "interface" || w[1] == "virtual_router_id")) print w[1] " " w[2];'
r" return }"
r' if (sd == 2 && st[1] == "vrrp_instance" && st[2] ~ /^virtual_ipaddress(_excluded)?$/) {'
r' x = "vip " w[1]; for (i = 2; i < nw; i++) if (w[i] == "dev") x = x " dev " w[i + 1];'
r" print x } }",
r"function rk(f, d, l, r, n, t, k, w, nw) {"
r' if (4 < d || (f in seen)) { print "too-deep " f; return }'
r" if (!ok(f)) return; seen[f] = 1;"
r' while (0 < (r = (getline l < f))) { sub(/[#!].*/, "", l); gsub(/[{}]/, " & ", l);'
r' n = split(l, t, " "); nw = 0; for (k = 1; k <= n; k++) {'
r' if (t[k] == "{") { push(nw ? w[1] : "", 1 < nw ? w[2] : ""); nw = 0 }'
r' else if (t[k] == "}") { stmt(w, nw, f, d); nw = 0; pop() } else w[++nw] = t[k] }'
r" stmt(w, nw, f, d) }"
r' if (r < 0) print "unreadable " f; close(f); delete seen[f] }',
r"function inc(p, f, d, c, g, dir, fs, n, i) {"
r' if (p !~ /^\//) { dir = f; sub(/[^\/]*$/, "", dir); p = dir p }'
r' if (p !~ /^[A-Za-z0-9_.\/*?-]+$/) { print "unreadable " p; return }'
r' c = "for g in " p "; do [ -e \"$g\" ] && echo \"$g\"; done"; n = 0;'
r" while (0 < (c | getline g)) fs[++n] = g; close(c);"
r" for (i = 1; i <= n; i++) rk(fs[i], d + 1) }",
r'/^ActiveState=/ { print "active " substr($0, 13); next }',
r'/^ExecStart=/ { s = $0; i = index(s, "argv[]="); if (!i) next; s = substr(s, i + 7);'
r' j = index(s, " ;"); if (j) s = substr(s, 1, j - 1); n = split(s, t, " ");'
r' for (k = 2; k <= n; k++) { f = t[k]; if (f !~ /^-/) continue; v = ""; x = index(f, "=");'
r" if (x) { v = substr(f, x + 1); f = substr(f, 1, x - 1) }"
r' else if (k < n && t[k + 1] !~ /^-/) v = t[++k]; sub(/^--?/, "", f);'
r" if (f ~ /^(config|ip|netmask|mask|interface|iface|trigger-key|key|f|use-file)$/) {"
r' print "arg " f " " v; a[f] = v } } next }',
r'/^EnvironmentFiles=/ { s = substr($0, 18); sub(/ \(ignore_errors=[a-z]*\)$/, "", s);'
r' if (s != "") ef[++ne] = s; next }',
r'/^Environment=/ { n = split(substr($0, 13), t, " "); for (k = 1; k <= n; k++) {'
rf' x = index(t[k], "="); if (x && substr(t[k], 1, x - 1) ~ /^{_ENV_KEYS}$/)'
r' print "env " substr(t[k], 1, x - 1) " " substr(t[k], x + 1) } next }',
rf'END {{ if (m == "keepalived") {{ p = "{_KEEPALIVED_CONF}";'
r' if ("use-file" in a) p = a["use-file"]; if ("f" in a) p = a["f"]; rk(p, 0) }'
r' else { for (k = 1; k <= ne; k++) rf(ef[k], "env");'
r' if ("config" in a) rf(a["config"], "cfg") } }',
)
)
#: One line, POSIX ``sh``, read-only. The frame markers are printed in two halves
#: so that a transport which echoes the command does not show them early.
VIRTUAL_IPS_COMMAND = (
"PATH=$PATH:/usr/sbin:/sbin; "
"printf '%s%s\\n' VIPS_ BEGIN; "
"if command -v systemctl >/dev/null 2>&1; then echo '[systemd]'; "
"for u in $(systemctl list-units --all --plain --no-legend --type=service "
"vip-manager.service 'vip-manager@*' keepalived.service 2>/dev/null | awk '{print $1}'); do "
"case $u in keepalived*) m=keepalived;; *) m=vip-manager;; esac; "
'echo "[$m $u]"; '
'systemctl show -p ActiveState -p ExecStart -p EnvironmentFiles -p Environment "$u" '
f"2>/dev/null | awk -v m=$m {quote(_AWK)}; done; "
"else echo '[no-systemd]'; fi; "
"printf '%s%s\\n' VIPS_ END"
)
_COMMENT_RE = re.compile(r"\s+#.*$")
_VARIABLE_RE = re.compile(r"^\$\{?(\w+)\}?$")
_Record = Tuple[str, str]
def _frame(output: str) -> List[str]:
lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
try:
start = lines.index(_BEGIN)
end = lines.index(_END, start)
except ValueError:
raise ValueError("no intact virtual IP report in the output") from None
return lines[start + 1 : end]
def _sections(lines: List[str]) -> List[Tuple[str, str, List[_Record]]]:
"""``[<mechanism> <unit>]`` headers, each with its records as ``(kind, rest)``."""
sections: List[Tuple[str, str, List[_Record]]] = []
for line in lines:
if line.startswith("[") and line.endswith("]"):
mechanism, _, unit = line[1:-1].partition(" ")
sections.append((mechanism, unit, []))
elif line and sections:
kind, _, rest = line.partition(" ")
sections[-1][2].append((kind, rest.strip()))
return sections
def _clean(value: str) -> str:
"""Without a trailing ``# comment`` and the quotes around it."""
value = _COMMENT_RE.sub("", value.strip())
if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
value = value[1:-1]
return value.strip()
def _address(text: str) -> Optional[Tuple[str, Optional[int]]]:
"""``"10.0.0.10/24"`` -> ``("10.0.0.10", 24)``; None for what is no address."""
host, slash, prefix = text.partition("/")
try:
address = ipaddress.ip_address(host)
except ValueError:
return None
if not slash:
return str(address), None
if not prefix.isdigit() or int(prefix) > address.max_prefixlen:
return None
return str(address), int(prefix)
def _prefix(mask: Optional[str]) -> Optional[int]:
"""``"24"`` or ``"255.255.255.0"`` -> 24."""
if not mask:
return None
if mask.isdigit():
return int(mask) if int(mask) <= 128 else None
try:
return ipaddress.IPv4Network(f"0.0.0.0/{mask}").prefixlen
except ValueError:
return None
def _running(records: List[_Record]) -> Optional[bool]:
states = [rest for kind, rest in records if kind == "active"]
return states[-1] == "active" if states else None
def _vip_manager(unit: str, records: List[_Record]) -> List[VirtualIpDict]:
values: Dict[str, Dict[str, str]] = {"arg": {}, "env": {}, "cfg": {}}
for kind, rest in records:
if kind in values:
key, _, value = rest.partition(" ")
values[kind][key] = value
env = values["env"]
def resolve(value: str) -> Optional[str]:
value = _clean(value)
variable = _VARIABLE_RE.match(value)
if variable:
value = _clean(env.get(variable.group(1), ""))
return value if value and "$" not in value else None
def pick(args: Tuple[str, ...], envs: Tuple[str, ...], cfgs: Tuple[str, ...]) -> Optional[str]:
for kind, names in (("arg", args), ("env", envs), ("cfg", cfgs)):
for name in names:
if name in values[kind]:
found = resolve(values[kind][name])
if found:
return found
return None
ip = pick(("ip",), ("VIP_IP",), ("ip",))
parsed = _address(ip) if ip else None
if parsed is None:
return []
address, prefix = parsed
mask = pick(("netmask", "mask"), ("VIP_NETMASK", "VIP_MASK"), ("netmask", "mask"))
return [
{
"mechanism": "vip-manager",
"instance": unit[: -len(".service")] if unit.endswith(".service") else unit,
"address": address,
"prefix_length": _prefix(mask) if mask else prefix,
"interface": pick(
("interface", "iface"), ("VIP_INTERFACE", "VIP_IFACE"), ("interface", "iface")
),
"group_key": pick(
("trigger-key", "key"), ("VIP_TRIGGER_KEY", "VIP_KEY"), ("trigger-key", "key")
),
"running": _running(records),
}
]
@dataclass
class _Instance:
name: str
interface: Optional[str] = None
vrid: str = ""
vips: List[str] = field(default_factory=list)
def _keepalived(unit: str, records: List[_Record]) -> List[VirtualIpDict]:
running = _running(records)
instances: List[_Instance] = []
current: Optional[_Instance] = None
for kind, rest in records:
if kind == "instance":
current = _Instance(rest or unit)
instances.append(current)
elif current is None:
continue
elif kind == "interface":
current.interface = rest or None
elif kind == "virtual_router_id":
current.vrid = rest
elif kind == "vip":
current.vips.append(rest)
elif kind == "end":
current = None
return [entry for instance in instances for entry in _instance_entries(instance, running)]
def _instance_entries(instance: _Instance, running: Optional[bool]) -> List[VirtualIpDict]:
entries: List[VirtualIpDict] = []
for vip in instance.vips:
tokens = vip.split()
parsed = _address(tokens[0]) if tokens else None
if parsed is None:
continue
address, prefix = parsed
dev = tokens[tokens.index("dev") + 1] if "dev" in tokens[1:-1] else None
entries.append(
{
"mechanism": "keepalived",
"instance": instance.name,
"address": address,
"prefix_length": prefix if prefix is not None else _host_prefix(address),
"interface": dev or instance.interface,
"group_key": f"vrid:{instance.vrid}" if instance.vrid.isdigit() else None,
"running": running,
}
)
return entries
def _host_prefix(address: str) -> int:
"""keepalived's default for an entry without one: the single address."""
return ipaddress.ip_address(address).max_prefixlen
_READERS: Dict[str, Tuple[str, Callable[[str, List[_Record]], List[VirtualIpDict]]]] = {
"vip-manager": ("vip_manager", _vip_manager),
"keepalived": ("keepalived", _keepalived),
}
def _order(entry: VirtualIpDict) -> Tuple[int, int, str]:
address = ipaddress.ip_address(entry["address"])
return address.version, int(address), entry["instance"]
def parse_virtual_ips(output: str) -> VirtualIpsDict:
"""Parse what :data:`VIRTUAL_IPS_COMMAND` printed, each mechanism's entries
sorted by address.
:raises ValueError: when the output carries no intact report.
"""
sections = _sections(_frame(output))
if any(mechanism == _NO_SYSTEMD for mechanism, _, _ in sections):
return {"vip_manager": None, "keepalived": None}
found: Dict[str, List[VirtualIpDict]] = {"vip_manager": [], "keepalived": []}
blind: Set[str] = set()
for mechanism, unit, records in sections:
if mechanism not in _READERS:
continue
key, read = _READERS[mechanism]
if any(kind in _BLIND for kind, _ in records):
blind.add(key)
else:
found[key].extend(read(unit, records))
def result(key: str) -> Optional[List[VirtualIpDict]]:
return None if key in blind else sorted(found[key], key=_order)
return {"vip_manager": result("vip_manager"), "keepalived": result("keepalived")}
class VirtualIpsMixin:
"""Adds :meth:`get_virtual_ips` to a driver that can run a command on a Linux host.
The template form (README, "Function classes"): the command and its parse are
the same everywhere, so they are concrete here, and a driver supplies only
:meth:`_run_virtual_ips_command` -- how a command reaches its host, and how it
gains root there. Mixed in by the drivers that can, not by
:class:`~napalm_device_types.os.OSDriver`, so ``hasattr(driver,
"get_virtual_ips")`` stays a truthful answer.
"""
if TYPE_CHECKING: # pragma: no cover - declared for type checkers only
def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
"""Run *command* on the host and return what it printed; as root
when *privileged*. The command is a single ``sh -c`` invocation of
about 4 kB: send it on an exec channel, not typed into a shell."""
...
def get_virtual_ips(self) -> VirtualIpsDict:
"""
Returns the virtual IPs the host's vip-manager and keepalived configuration
declares -- whether or not the host holds them at the moment.
* vip_manager (list or None) - see :class:`~napalm_device_types.models.VirtualIpDict`
* keepalived (list or None)
None for a mechanism means it could not be read; ``[]`` that the host has none.
Example::
{
"vip_manager": [
{"mechanism": "vip-manager", "instance": "vip-manager",
"address": "10.7.224.10", "prefix_length": 24, "interface": "ens7",
"group_key": "/service/netork-db/leader", "running": True},
],
"keepalived": [],
}
:raises ValueError: if neither reading carried an intact report.
"""
command = f"sh -c {quote(VIRTUAL_IPS_COMMAND)}"
try:
return parse_virtual_ips(self._run_virtual_ips_command(command, privileged=True))
except ValueError:
pass
return parse_virtual_ips(self._run_virtual_ips_command(command, privileged=False))
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "napalm-device-types"
version = "2.3.0"
version = "4.5.0"
description = "Abstract device-type base classes for NAPALM drivers"
readme = "README.md"
requires-python = ">=3.10"
+169
View File
@@ -0,0 +1,169 @@
"""The command channel: how a driver runs a command and opens a byte stream.
netOrk's container runtime driver speaks the Docker Engine API over a stream
from ``docker system dial-stdio`` (NetOrk/netork#765). The driver only provides
the way there; these are the SSH pieces every SSH driver can reuse. The
paramiko channel is faked, so the tests pin behaviour, not paramiko.
"""
from __future__ import annotations
import socket
import pytest
from napalm_device_types.channel import (
CommandResult,
ParamikoExecStream,
open_stream_on_transport,
run_on_transport,
)
class FakeChannel:
"""Just enough of paramiko.Channel: scripted stdout/stderr chunks, an exit code."""
def __init__(self, stdout=(), stderr=(), exit_code=0, hang=False):
self.out = list(stdout)
self.err = list(stderr)
self.exit_code = exit_code
self.hang = hang
self.sent = b""
self.command = None
self.write_closed = False
self.closed = False
self.timeout = None
def exec_command(self, command):
self.command = command
def settimeout(self, timeout):
self.timeout = timeout
def sendall(self, data):
self.sent += data
def shutdown_write(self):
self.write_closed = True
def close(self):
self.closed = True
def recv_ready(self):
return bool(self.out)
def recv(self, n):
if self.out:
return self.out.pop(0)
if self.hang:
raise socket.timeout()
return b""
def recv_stderr_ready(self):
return bool(self.err)
def recv_stderr(self, n):
return self.err.pop(0) if self.err else b""
def exit_status_ready(self):
return not self.hang and not self.out and not self.err
def recv_exit_status(self):
return self.exit_code
class FakeTransport:
def __init__(self, channel):
self.channel = channel
def open_session(self):
return self.channel
def test_run_collects_stdout_stderr_and_the_exit_code():
ch = FakeChannel(stdout=[b"hel", b"lo\n"], stderr=[b"warn\n"], exit_code=3)
result = run_on_transport(FakeTransport(ch), "echo hello", timeout=5)
assert result == CommandResult(stdout="hello\n", stderr="warn\n", exit_code=3)
assert ch.command == "echo hello"
assert ch.closed
def test_run_sends_stdin_and_then_closes_the_write_side():
ch = FakeChannel(stdout=[b"ok"])
run_on_transport(FakeTransport(ch), "sudo -S true", stdin=b"pw\n", timeout=5)
assert ch.sent == b"pw\n"
assert ch.write_closed
def test_run_raises_when_the_command_does_not_finish_in_time():
ch = FakeChannel(hang=True)
with pytest.raises(TimeoutError, match="did not finish"):
run_on_transport(FakeTransport(ch), "sleep 999", timeout=0.05)
assert ch.closed
def test_a_stream_reads_until_eof_and_drains_stderr_on_the_way():
ch = FakeChannel(stdout=[b"HTTP/1.1 200 OK\r\n"], stderr=[b"note\n"])
stream = open_stream_on_transport(FakeTransport(ch), "docker system dial-stdio")
assert stream.read(4096, timeout=5) == b"HTTP/1.1 200 OK\r\n"
assert stream.read(4096, timeout=5) == b""
assert stream.stderr == "note\n"
assert ch.command == "docker system dial-stdio"
def test_a_stream_writes_and_closes_its_write_side_only_when_asked():
"""dial-stdio answers HTTP 499 to a request whose writer closed early
(netork#771), so write() never implies close_write()."""
ch = FakeChannel()
stream = open_stream_on_transport(FakeTransport(ch), "docker system dial-stdio")
stream.write(b"GET /_ping HTTP/1.1\r\n\r\n")
assert not ch.write_closed
stream.close_write()
assert ch.write_closed and ch.sent.startswith(b"GET /_ping")
def test_a_stream_sends_its_stdin_prefix_first():
"""How a sudo password reaches `sudo -S` before the stream's own bytes."""
ch = FakeChannel()
open_stream_on_transport(FakeTransport(ch), "sudo -S -p '' cmd", stdin_prefix=b"pw\n")
assert ch.sent == b"pw\n"
def test_a_stream_times_out_as_timeout_error():
stream = ParamikoExecStream(FakeChannel(hang=True))
with pytest.raises(TimeoutError):
stream.read(10, timeout=0.01)
def test_the_stream_keeps_only_a_bounded_tail_of_stderr():
ch = FakeChannel(stderr=[b"x" * 100, b"y" * 100])
stream = ParamikoExecStream(ch, stderr_limit=50)
assert stream.stderr == "y" * 50
def test_exit_status_is_none_while_running_and_the_code_after():
running = ParamikoExecStream(FakeChannel(hang=True))
finished = ParamikoExecStream(FakeChannel(exit_code=1))
assert running.exit_status is None
assert finished.exit_status == 1
def test_close_closes_the_channel():
ch = FakeChannel()
ParamikoExecStream(ch).close()
assert ch.closed
+178
View File
@@ -0,0 +1,178 @@
"""Access to a container engine: which engines a host has, and a way to each.
This package abstracts the connection, the driver builds it, and netOrk does
the talking (NetOrk/netork#765, decided 2026-10-07). So nothing here knows what
a container is. It knows how to reach the engine's API (``dial-stdio``) and how
to run the engine's CLI with arguments netOrk chooses, and the driver alone
knows where that CLI lives.
"""
from __future__ import annotations
import pytest
from napalm_device_types import OSDriver
from napalm_device_types.channel import CommandResult
from napalm_device_types.container_engine import (
ContainerEngineConnection,
ContainerEngineMixin,
ContainerEngineUnavailable,
)
class FakeDriver(ContainerEngineMixin):
"""A driver with a channel: records what it is asked to run."""
def __init__(self, present=("docker",), binary=None):
self.present = set(present)
self.binary = binary
self.commands = []
self.streams = []
def run_command(self, command, *, privileged=False, timeout=60, stdin=None):
self.commands.append((command, privileged, timeout, stdin))
found = any(command == f"command -v {b}" for b in self._binaries())
return CommandResult(stdout="/usr/bin/x\n" if found else "", stderr="", exit_code=0 if found else 1)
def open_stream(self, command, *, privileged=False):
self.streams.append((command, privileged))
return object()
def _binaries(self):
return {self._container_engine_binary(e) for e in self.present}
def _container_engine_binary(self, engine):
return self.binary or super()._container_engine_binary(engine)
def test_a_host_with_docker_lists_it():
assert FakeDriver().container_engines() == [{"engine": "docker", "api": "docker-engine"}]
def test_a_host_without_the_cli_lists_nothing():
assert FakeDriver(present=()).container_engines() == []
def test_the_api_is_reached_through_dial_stdio():
driver = FakeDriver()
driver.open_container_engine("docker").open_api()
assert driver.streams == [("docker system dial-stdio", False)]
def test_the_driver_decides_where_the_binary_lives():
"""QNAP keeps docker under Container Station's path; netOrk never sees it."""
path = "/share/CACHEDEV1_DATA/.qpkg/container-station/bin/docker"
driver = FakeDriver(binary=path)
assert driver.container_engines() == [{"engine": "docker", "api": "docker-engine"}]
driver.open_container_engine("docker").open_api()
assert driver.streams == [(f"{path} system dial-stdio", False)]
def test_cli_arguments_are_quoted_and_prefixed_by_the_binary():
driver = FakeDriver()
conn = driver.open_container_engine("docker")
conn.run_cli(["compose", "-f", "/srv/my stack/compose.yml", "config", "--format", "json"])
command, privileged, timeout, stdin = driver.commands[-1]
assert command == "docker compose -f '/srv/my stack/compose.yml' config --format json"
assert privileged is False and stdin is None
def test_cli_stdin_and_timeout_reach_the_channel():
driver = FakeDriver()
driver.open_container_engine("docker").run_cli(["login", "--password-stdin"], stdin=b"x", timeout=30)
_command, _privileged, timeout, stdin = driver.commands[-1]
assert (timeout, stdin) == (30, b"x")
def test_a_streamed_cli_call_can_merge_stderr():
"""compose writes its progress to stderr; a merged stream carries both."""
driver = FakeDriver()
conn = driver.open_container_engine("docker")
conn.stream_cli(["pull", "redis:7-alpine"])
conn.stream_cli(["compose", "up", "-d", "db"], merge_stderr=True)
assert driver.streams == [
("docker pull redis:7-alpine", False),
("docker compose up -d db 2>&1", False),
]
def test_a_shell_metacharacter_in_an_argument_stays_an_argument():
driver = FakeDriver()
driver.open_container_engine("docker").run_cli(["inspect", "x; rm -rf /"])
assert driver.commands[-1][0] == "docker inspect 'x; rm -rf /'"
def test_an_unknown_engine_is_refused():
with pytest.raises(ContainerEngineUnavailable):
FakeDriver().open_container_engine("rkt")
def test_the_connection_names_its_engine():
conn = FakeDriver().open_container_engine("docker")
assert isinstance(conn, ContainerEngineConnection)
assert conn.engine == "docker"
def test_an_os_driver_does_not_get_container_engine_access_by_role():
"""Mixed in by the driver, not by OSDriver: a Windows host is an OS driver
too, and `hasattr(driver, "open_container_engine")` has to stay truthful."""
assert not issubclass(OSDriver, ContainerEngineMixin)
assert not hasattr(OSDriver, "open_container_engine")
class TestPrivilegedCli:
"""A refused engine call can be repeated as root (netork#773): the driver
knows the binary and how it gains root, so the connection carries it."""
def test_run_cli_passes_privileged_to_the_channel(self):
driver = FakeDriver()
driver.open_container_engine("docker").run_cli(["restart", "web"], privileged=True)
assert driver.commands == [("docker restart web", True, 120, None)]
def test_run_cli_is_unprivileged_by_default(self):
driver = FakeDriver()
driver.open_container_engine("docker").run_cli(["ps"])
assert driver.commands[0][1] is False
def test_stream_cli_passes_privileged_too(self):
driver = FakeDriver()
driver.open_container_engine("docker").stream_cli(["pull", "nginx"], privileged=True)
assert driver.streams == [("docker pull nginx", True)]
class TestNoDockerModel:
"""There is no container model here (NetOrk/netork#765): netOrk reads and
handles containers over the Engine API, so the CLI-shaped types and the
``get_docker_info()`` declaration that served napalm-linux' old reading are gone."""
def test_the_docker_types_are_gone(self):
import napalm_device_types
from napalm_device_types import models
for name in ("DockerInfoDict", "DockerContainerDict", "DockerImageDict"):
assert not hasattr(models, name), name
assert not hasattr(napalm_device_types, name), name
def test_an_os_driver_declares_no_docker_reading(self):
from napalm_device_types import os as os_role
assert "get_docker_info" not in os_role.__dict__.get("OSDriver", object).__dict__
+309 -1
View File
@@ -11,10 +11,14 @@ The fixtures are the real states of six hosts on netOrk's test server.
from __future__ import annotations
import re
import subprocess
import pytest
from napalm_device_types import OSDriver
from napalm_device_types.host_status import (
BUSY_CHECK,
HOST_STATUS_COMMAND,
HostStatusMixin,
kernel_reboot_pending,
@@ -29,8 +33,12 @@ def _wire(
modules=("6.8.0-139-generic", "6.8.0-142-generic"),
needs_restarting=None,
periodic=None,
stamp=None,
timer="enabled",
dnf_timers=("not-found", "not-found"),
dpkg=None,
journal=(),
busy=False,
):
lines = ["HSTAT_BEGIN"]
if reboot_file:
@@ -39,7 +47,11 @@ def _wire(
lines += ["[needs-restarting]", str(needs_restarting)]
lines += ["[kernel]", running, "[modules]", *modules]
if periodic is not None:
lines += ["[apt-config]", *periodic]
lines += ["[apt-config]", *periodic, "[upgrade-stamp]", *([str(stamp)] if stamp else [])]
if dpkg is not None:
if busy:
lines.append("[dpkg-busy]")
lines += ["[dpkg-journal]", *journal, "[dpkg-audit]", *dpkg]
lines += [
"[timers]",
f"apt-daily-upgrade.timer {timer}",
@@ -198,3 +210,299 @@ class TestHostStatusMixin:
def test_not_every_os_driver_has_it(self):
assert not hasattr(OSDriver, "get_host_status")
class TestFreeBSD:
"""freebsd-version -k names the installed kernel, -r the running one; a
patch that replaced the kernel shows as the two differing until the host
boots it. Recorded on a FreeBSD 15.1 host (NetOrk/netork#799)."""
#: What the command printed there before this section existed.
BEFORE = (
"HSTAT_BEGIN\n[kernel]\n15.1-RELEASE\n[modules]\n[timers]\n"
"apt-daily-upgrade.timer \ndnf-automatic.timer \ndnf-automatic-install.timer \n"
"HSTAT_END\n"
)
@staticmethod
def _report(installed: str, running: str) -> str:
return (
f"HSTAT_BEGIN\n[kernel]\n{running}\n[modules]\n"
f"[freebsd-kernel]\n{installed}\n[freebsd-running]\n{running}\n"
"[timers]\napt-daily-upgrade.timer \nHSTAT_END\n"
)
def test_without_the_section_it_is_unknown(self):
assert parse_host_status(self.BEFORE)["reboot_required"] is None
def test_the_running_kernel_is_the_installed_one(self):
status = parse_host_status(self._report("15.1-RELEASE-p4", "15.1-RELEASE-p4"))
assert status["reboot_required"] is False
assert status["reboot_reason"] is None
def test_a_patched_kernel_waits_for_a_reboot(self):
status = parse_host_status(self._report("15.1-RELEASE-p5", "15.1-RELEASE-p4"))
assert status["reboot_required"] is True
assert status["reboot_reason"] == (
"kernel 15.1-RELEASE-p5 installed, 15.1-RELEASE-p4 running"
)
def test_freebsd_does_not_patch_itself_as_far_as_netork_can_tell(self):
status = parse_host_status(self._report("15.1-RELEASE", "15.1-RELEASE"))
assert status["auto_updates"] is None
def test_the_command_asks_only_where_freebsd_version_exists(self):
assert "command -v freebsd-version" in HOST_STATUS_COMMAND
assert "freebsd-version -k" in HOST_STATUS_COMMAND
assert "freebsd-version -r" in HOST_STATUS_COMMAND
class TestInterruptedDpkg:
"""dpkg left halfway: apt refuses to install anything until
``dpkg --configure -a`` has run, and the daily apt timers report success
all the same (NetOrk/netork#738).
The fixture is aris (Ubuntu 24.04): unattended-upgrades was cut off on
2026-07-03 while update-initramfs built the initrd for 6.8.0-134, and the
kernel package stayed half-configured for three months.
"""
ARIS = ("iF linux-image-6.8.0-134-generic",)
def test_a_half_configured_kernel_is_an_interrupted_dpkg(self):
status = parse_host_status(_wire(dpkg=self.ARIS))
assert status["package_manager_interrupted"] is True
assert status["interrupted_packages"] == [
{"name": "linux-image-6.8.0-134-generic", "state": "half-configured"}
]
def test_a_clean_dpkg_is_not_interrupted(self):
status = parse_host_status(_wire(dpkg=()))
assert status["package_manager_interrupted"] is False
assert status["interrupted_packages"] == []
@pytest.mark.parametrize(
("line", "state"),
[
("iH base-files", "half-installed"),
("iU base-files", "unpacked"),
("iF base-files", "half-configured"),
("iW base-files", "triggers-awaited"),
("it base-files", "triggers-pending"),
("iUR base-files", "reinstall-required"),
("hF base-files", "half-configured"),
],
)
def test_every_state_an_interrupted_run_leaves(self, line, state):
status = parse_host_status(_wire(dpkg=(line,)))
assert status["interrupted_packages"] == [{"name": "base-files", "state": state}]
def test_a_journal_left_behind_is_interrupted_on_its_own(self):
"""What apt itself checks: numbered files in /var/lib/dpkg/updates."""
status = parse_host_status(_wire(dpkg=(), journal=("0000", "0001")))
assert status["package_manager_interrupted"] is True
assert status["interrupted_packages"] == []
def test_only_numbered_journal_files_count(self):
"""apt ignores anything else there, such as dpkg's tmp.i."""
status = parse_host_status(_wire(dpkg=(), journal=("tmp.i",)))
assert status["package_manager_interrupted"] is False
def test_a_running_package_manager_makes_it_unknown(self):
"""Mid-upgrade, packages are unpacked and not yet configured, also
between two of apt's dpkg calls. That is work in progress, not an
interruption."""
status = parse_host_status(_wire(dpkg=("iU base-files",), busy=True))
assert status["package_manager_interrupted"] is None
assert status["interrupted_packages"] == []
def test_without_dpkg_it_is_unknown(self):
status = parse_host_status(_wire())
assert status["package_manager_interrupted"] is None
assert status["interrupted_packages"] == []
def test_the_command_asks_dpkg_only_where_it_exists(self):
assert "command -v dpkg-query" in HOST_STATUS_COMMAND
assert "/var/lib/dpkg/updates" in HOST_STATUS_COMMAND
def test_it_never_repairs(self):
assert "--configure" not in HOST_STATUS_COMMAND
class TestAutoUpdatesLastSuccess:
"""apt.systemd.daily touches upgrade-stamp only after unattended-upgrade
succeeded. update-success-stamp says nothing: netOrk's own daily
``apt-get update`` touches it as well (NetOrk/netork#738)."""
def test_the_upgrade_stamp_is_the_last_successful_run(self):
status = parse_host_status(_wire(periodic=UNATTENDED, stamp=1751525820))
assert status["auto_updates_last_success"] == 1751525820
assert status["auto_updates_interval"] == 86400
def test_no_stamp_is_unknown(self):
status = parse_host_status(_wire(periodic=UNATTENDED))
assert status["auto_updates_last_success"] is None
def test_without_apt_both_are_unknown(self):
status = parse_host_status(_wire())
assert status["auto_updates_last_success"] is None
assert status["auto_updates_interval"] is None
@pytest.mark.parametrize(
("value", "seconds"),
[
('"1"', 86400),
('"7d"', 7 * 86400),
('"12h"', 43200),
('"30m"', 1800),
('"90s"', 90),
('"always"', 0),
('"0"', None),
('"weekly"', None),
],
)
def test_the_interval_in_apts_units(self, value, seconds):
periodic = [f"APT::Periodic::Unattended-Upgrade {value};"]
assert parse_host_status(_wire(periodic=periodic))["auto_updates_interval"] == seconds
def test_the_command_reads_the_upgrade_stamp_only(self):
assert "/var/lib/apt/periodic/upgrade-stamp" in HOST_STATUS_COMMAND
assert "update-success-stamp" not in HOST_STATUS_COMMAND
class TestBusyCheck:
"""Whether a package manager is working right now, run through ``sh`` with a
stand-in ``ps``. On Ubuntu, unattended-upgrades keeps
``unattended-upgrade-shutdown --wait-for-signal`` running all the time; its
process name is cut to 15 characters, ``unattended-upgr``, the same as a real
run's. Taken for a run, it made every such host's dpkg state unknown -- aris
among them (NetOrk/netork#738)."""
UBUNTU_IDLE = [
("systemd", "/sbin/init"),
("unattended-upgr", "/usr/bin/python3 /usr/share/unattended-upgrades/unattended-upgrade-shutdown --wait-for-signal"),
("sshd", "sshd: netork [priv]"),
]
@staticmethod
def _busy(tmp_path, processes) -> bool:
comm = "\n".join(c for c, _ in processes)
args = "\n".join(a for _, a in processes)
ps = tmp_path / "ps"
ps.write_text(
"#!/bin/sh\n"
f"case \"$*\" in *comm=*) printf '%s\\n' '{comm}';; *) printf '%s\\n' '{args}';; esac\n"
)
ps.chmod(0o755)
env = {"PATH": f"{tmp_path}:/usr/bin:/bin"}
out = subprocess.run(["sh", "-c", BUSY_CHECK], capture_output=True, text=True, env=env)
return "[dpkg-busy]" in out.stdout
def test_the_shutdown_helper_of_an_idle_ubuntu_host_is_no_run(self, tmp_path):
assert self._busy(tmp_path, self.UBUNTU_IDLE) is False
def test_unattended_upgrade_running(self, tmp_path):
run = ("unattended-upgr", "/usr/bin/python3 /usr/bin/unattended-upgrade")
assert self._busy(tmp_path, [*self.UBUNTU_IDLE, run]) is True
@pytest.mark.parametrize("name", ["dpkg", "apt", "apt-get", "aptitude"])
def test_dpkg_and_apt(self, tmp_path, name):
assert self._busy(tmp_path, [*self.UBUNTU_IDLE, (name, f"/usr/bin/{name} upgrade")]) is True
def test_it_is_part_of_the_command(self):
assert BUSY_CHECK in HOST_STATUS_COMMAND
def test_the_command_does_not_match_itself(self):
"""Run over an exec channel, the command is the command line of the shell
that runs it, and ``ps -o args`` lists it."""
assert not re.search(r"(^|[ /])unattended-upgrade( |$)", HOST_STATUS_COMMAND)
class TestFilesystems:
"""How full the filesystems an upgrade writes to are: ``/``, ``/boot`` and
``/var``, read with ``df -Pk`` and no root.
pve-dual's root filesystem stood at 94 % with about 400 MB free for a week;
a kernel upgrade failed on it with "No space left on device" and left dpkg
halfway, and nothing had said the disk was full (NetOrk/netork#862). The
figures below are shaped after it (6.5 GiB, 400 MiB available): ext4 keeps
5 % for root, so what is available is less than total minus used.
"""
HEADER = "Filesystem 1024-blocks Used Available Capacity Mounted on"
PVE_ROOT = "/dev/mapper/pve-root 6815744 6065357 409600 94% /"
BOOT = "/dev/sda2 1046512 104860 941652 11% /boot"
@staticmethod
def _report(*df_lines: str) -> str:
return _wire().replace("[timers]", "\n".join(["[filesystems]", *df_lines, "[timers]"]))
def test_the_root_filesystem_in_bytes(self):
status = parse_host_status(self._report(self.HEADER, self.PVE_ROOT))
assert status["filesystems"] == [
{
"mountpoint": "/",
"total": 6815744 * 1024,
"used": 6065357 * 1024,
"available": 409600 * 1024,
}
]
def test_a_separate_boot_partition(self):
status = parse_host_status(self._report(self.HEADER, self.PVE_ROOT, self.BOOT))
assert [fs["mountpoint"] for fs in status["filesystems"]] == ["/", "/boot"]
def test_a_directory_on_root_is_root_once(self):
"""``df /boot /var`` names the filesystem they live on, here ``/`` three times."""
status = parse_host_status(
self._report(self.HEADER, self.PVE_ROOT, self.PVE_ROOT, self.PVE_ROOT)
)
assert [fs["mountpoint"] for fs in status["filesystems"]] == ["/"]
def test_a_long_device_name_and_a_blank_in_the_mountpoint(self):
line = "/dev/disk/by-id/a very long name 1000 600 400 60% /mnt/my disk"
(fs,) = parse_host_status(self._report(self.HEADER, line))["filesystems"]
assert (fs["mountpoint"], fs["available"]) == ("/mnt/my disk", 400 * 1024)
def test_a_filesystem_without_size_is_left_out(self):
"""What a pseudo filesystem reports; it cannot run full."""
status = parse_host_status(self._report(self.HEADER, "none 0 0 0 - /"))
assert status["filesystems"] is None
def test_without_the_section_it_is_unknown(self):
"""An older command, or no df: None, not "nothing is full"."""
assert parse_host_status(_wire())["filesystems"] is None
def test_df_that_printed_nothing_is_unknown(self):
assert parse_host_status(self._report())["filesystems"] is None
def test_the_command_asks_for_the_three_without_root(self):
assert "df -Pk / /boot /var" in HOST_STATUS_COMMAND
assert "sudo" not in HOST_STATUS_COMMAND
def test_the_command_reads_them_on_a_real_shell(self):
"""POSIX df the way a host's sh runs it: ``/`` is always there, whatever it is."""
out = subprocess.run(
["sh", "-c", HOST_STATUS_COMMAND], capture_output=True, text=True, timeout=30
)
mountpoints = [fs["mountpoint"] for fs in parse_host_status(out.stdout)["filesystems"]]
assert "/" in mountpoints
+396
View File
@@ -0,0 +1,396 @@
"""get_listening_sockets: what listens on which address, and which service it is.
Whether a service is reachable from outside its host is decided by what it
listens on -- ``0.0.0.0:5432`` is, ``127.0.0.1:5432`` is not. Reading that is
the same on every Linux host: ``ss`` for the sockets, ``/proc/<pid>/cgroup``
for the systemd unit or container a process belongs to. So both live here once,
and a driver only carries the command across (#658 in netOrk).
A host without ``ss`` -- OpenWrt's busybox, an old net-tools box -- is read
with ``netstat -lntup`` instead, and on OpenWrt the cgroup names the procd
service (#673 in netOrk).
"""
from __future__ import annotations
import os
import shutil
import subprocess
import pytest
from napalm_device_types import OSDriver
from napalm_device_types.listening import (
LISTENING_SOCKETS_COMMAND,
ListeningSocketsMixin,
ListeningSocketsUnavailable,
parse_listening_sockets,
)
CID = "4f1c" + "0" * 60
SS = """Netid State Recv-Q Send-Q Local Address:Port Peer Address:PortProcess
udp UNCONN 0 0 127.0.0.53%lo:53 0.0.0.0:* users:(("systemd-resolve",pid=612,fd=13))
udp UNCONN 0 0 0.0.0.0%ens18:68 0.0.0.0:* users:(("systemd-network",pid=590,fd=22))
tcp LISTEN 0 4096 0.0.0.0:5432 0.0.0.0:* users:(("postgres",pid=812,fd=6))
tcp LISTEN 0 128 [::]:22 [::]:* users:(("sshd",pid=700,fd=4))
tcp LISTEN 0 511 *:80 *:* users:(("nginx",pid=901,fd=6),("nginx",pid=900,fd=6))
tcp LISTEN 0 4096 [::ffff:127.0.0.1]:8125 *:* users:(("statsd",pid=950,fd=3))
tcp LISTEN 0 4096 0.0.0.0:8080 0.0.0.0:* users:(("docker-proxy",pid=1200,fd=4))
tcp LISTEN 0 4096 0.0.0.0:9100 0.0.0.0:* users:(("node_exporter",pid=1300,fd=3))
tcp LISTEN 0 64 0.0.0.0:2049 0.0.0.0:*
"""
CGROUPS = f"""612 0::/system.slice/systemd-resolved.service
590 0::/system.slice/systemd-networkd.service
812 0::/system.slice/system-postgresql.slice/postgresql@16-main.service
700 0::/system.slice/ssh.service
900 0::/system.slice/nginx.service
901 0::/system.slice/nginx.service
950 0::/user.slice/user-1000.slice/session-3.scope
1200 0::/system.slice/docker.service
1300 0::/system.slice/docker-{CID}.scope
"""
def _wire(ss: str = SS, cgroups: str = CGROUPS, *, rc: int = 0, noise: str = "") -> str:
"""The report as the command prints it, framed."""
return f"{noise}SOCK_BEGIN\n[ss]\n{ss}__SS_RC={rc}\n[cgroups]\n{cgroups}SOCK_END\n"
def _by_port(sockets: list) -> dict:
return {(s["proto"], s["port"], s["address"]): s for s in sockets}
class TestParsing:
def test_a_socket_comes_with_its_process_and_unit(self):
sockets = _by_port(parse_listening_sockets(_wire()))
assert sockets[("tcp", 5432, "0.0.0.0")] == {
"proto": "tcp",
"address": "0.0.0.0",
"port": 5432,
"interface": None,
"process": "postgres",
"pid": 812,
"unit": "postgresql@16-main",
"container_id": None,
}
@pytest.mark.parametrize(
"local, address, interface",
[
("[::]:22", "::", None),
("*:80", "*", None),
("127.0.0.53%lo:53", "127.0.0.53", "lo"),
("0.0.0.0%ens18:68", "0.0.0.0", "ens18"),
("[::ffff:127.0.0.1]:8125", "::ffff:127.0.0.1", None),
("[fe80::1%eth0]:546", "fe80::1", "eth0"),
(":::22", "::", None), # iproute2 4.9 prints IPv6 without brackets
],
)
def test_every_address_form(self, local: str, address: str, interface):
line = f"tcp LISTEN 0 128 {local} *:* users:((\"x\",pid=1,fd=3))\n"
[socket] = parse_listening_sockets(_wire(line, "1 0::/system.slice/x.service\n"))
assert (socket["address"], socket["interface"]) == (address, interface)
def test_the_first_of_several_processes_names_the_socket(self):
nginx = _by_port(parse_listening_sockets(_wire()))[("tcp", 80, "*")]
assert (nginx["process"], nginx["pid"], nginx["unit"]) == ("nginx", 901, "nginx")
def test_a_socket_without_a_process_is_kept(self):
"""The kernel's nfsd, or every socket of another user without root."""
nfs = _by_port(parse_listening_sockets(_wire()))[("tcp", 2049, "0.0.0.0")]
assert (nfs["process"], nfs["pid"], nfs["unit"], nfs["container_id"]) == (
None,
None,
None,
None,
)
def test_the_header_is_no_socket(self):
sockets = parse_listening_sockets(_wire())
assert len(sockets) == 9
assert all(s["proto"] in ("tcp", "udp") for s in sockets)
def test_sorted_by_protocol_port_and_address(self):
keys = [(s["proto"], s["port"], s["address"]) for s in parse_listening_sockets(_wire())]
assert keys == sorted(keys)
def test_nothing_listening_is_an_empty_list(self):
assert parse_listening_sockets(_wire(SS.splitlines(True)[0], "")) == []
class TestCgroups:
def _unit_and_container(self, cgroup_lines: str) -> tuple:
line = 'tcp LISTEN 0 128 0.0.0.0:1 0.0.0.0:* users:(("p",pid=5,fd=3))\n'
[socket] = parse_listening_sockets(_wire(line, cgroup_lines))
return socket["unit"], socket["container_id"]
def test_a_container_on_the_host_network(self):
assert self._unit_and_container(f"5 0::/system.slice/docker-{CID}.scope\n") == (
None,
CID,
)
def test_a_container_under_the_cgroupfs_driver(self):
assert self._unit_and_container(f"5 0::/docker/{CID}\n") == (None, CID)
def test_cgroup_v1_reads_the_systemd_hierarchy(self):
lines = "5 12:pids:/user.slice\n5 1:name=systemd:/system.slice/ssh.service\n"
assert self._unit_and_container(lines) == ("ssh", None)
def test_an_escaped_template_instance(self):
lines = "5 0::/system.slice/system-wg\\x2dquick.slice/wg-quick@wg0.service\n"
assert self._unit_and_container(lines) == ("wg-quick@wg0", None)
def test_a_procd_service_on_openwrt(self):
"""procd puts every instance into /services/<name>/<instance>, a jailed
one too -- its PID is not the one procd reports, its cgroup is."""
assert self._unit_and_container("5 0::/services/dnsmasq/cfg01411c\n") == (
"dnsmasq",
None,
)
def test_a_login_session_is_no_unit(self):
assert self._unit_and_container("5 0::/user.slice/user-1000.slice/session-3.scope\n") == (
None,
None,
)
def test_a_process_gone_before_its_cgroup_was_read(self):
assert self._unit_and_container("") == (None, None)
def test_the_docker_daemons_own_proxy_stays_raw(self):
"""docker-proxy runs in docker.service. Telling it apart from a real
listener is the consumer's business; the reading reports what is there."""
proxy = _by_port(parse_listening_sockets(_wire()))[("tcp", 8080, "0.0.0.0")]
assert (proxy["process"], proxy["unit"]) == ("docker-proxy", "docker")
class TestFailures:
def test_whatever_surrounds_the_frame_is_ignored(self):
noisy = _wire(noise="$ sh -c '...'\nWelcome to Ubuntu\n")
assert len(parse_listening_sockets(noisy)) == 9
def test_output_without_the_frame_raises(self):
with pytest.raises(ValueError):
parse_listening_sockets("sudo: a password is required\n")
def test_a_report_cut_short_raises(self):
with pytest.raises(ValueError):
parse_listening_sockets(_wire()[:-len("SOCK_END\n")])
def test_a_host_without_ss_or_netstat_is_unavailable(self):
with pytest.raises(ListeningSocketsUnavailable):
parse_listening_sockets("SOCK_BEGIN\n[no-ss]\nSOCK_END\n")
def test_ss_failing_raises(self):
with pytest.raises(ValueError):
parse_listening_sockets(_wire("ss: invalid option -- 'p'\n", "", rc=1))
def test_unavailable_is_not_a_value_error(self):
"""A caller retries a broken report without privilege; a host without
ss has nothing to retry."""
assert not issubclass(ListeningSocketsUnavailable, ValueError)
# busybox netstat on OpenWrt 25.12, addresses replaced by documentation ones.
BUSYBOX = """Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address Foreign Address State PID/Program name
tcp 0 0 0.0.0.0:22 0.0.0.0:* LISTEN 1604/dropbear
tcp 0 0 0.0.0.0:443 0.0.0.0:* LISTEN 1969/uhttpd
tcp 0 0 192.0.2.15:53 0.0.0.0:* LISTEN 1495/dnsmasq
tcp 0 0 127.0.0.1:53 0.0.0.0:* LISTEN 1495/dnsmasq
tcp 0 0 :::22 :::* LISTEN 1604/dropbear
tcp 0 0 fe80::1:53 :::* LISTEN 1495/dnsmasq
tcp 0 0 ::1:53 :::* LISTEN 1495/dnsmasq
udp 0 0 192.0.2.15:53 0.0.0.0:* 1495/dnsmasq
udp 0 0 0.0.0.0:161 0.0.0.0:* 3173/snmpd
udp 0 0 0.0.0.0:5353 0.0.0.0:* -
"""
PROCD = """1495 0::/services/dnsmasq/cfg01411c
1604 0::/services/dropbear/instance1
1969 0::/services/uhttpd/instance1
3173 0::/services/snmpd/instance1
"""
# net-tools netstat on an old Debian: tcp6/udp6, and a program name with a space.
NET_TOOLS = """Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address Foreign Address State PID/Program name
tcp 0 0 0.0.0.0:22 0.0.0.0:* LISTEN 700/sshd: /usr/sbin
tcp6 0 0 :::22 :::* LISTEN 700/sshd: /usr/sbin
udp6 0 0 :::5353 :::* -
"""
def _netstat(out: str = BUSYBOX, cgroups: str = PROCD, *, rc: int = 0) -> str:
return f"SOCK_BEGIN\n[netstat]\n{out}__SS_RC={rc}\n[cgroups]\n{cgroups}SOCK_END\n"
class TestNetstat:
"""A host without ss (#673 in netOrk): OpenWrt's busybox, old net-tools."""
def test_a_socket_comes_with_its_process_and_procd_service(self):
sockets = _by_port(parse_listening_sockets(_netstat()))
assert sockets[("tcp", 22, "0.0.0.0")] == {
"proto": "tcp",
"address": "0.0.0.0",
"port": 22,
"interface": None,
"process": "dropbear",
"pid": 1604,
"unit": "dropbear",
"container_id": None,
}
@pytest.mark.parametrize(
"proto, port, address",
[("tcp", 22, "::"), ("tcp", 53, "fe80::1"), ("tcp", 53, "::1"), ("tcp", 53, "192.0.2.15")],
)
def test_every_address_form(self, proto, port, address):
assert (proto, port, address) in _by_port(parse_listening_sockets(_netstat()))
def test_a_udp_socket_has_no_state_column(self):
snmpd = _by_port(parse_listening_sockets(_netstat()))[("udp", 161, "0.0.0.0")]
assert (snmpd["process"], snmpd["pid"], snmpd["unit"]) == ("snmpd", 3173, "snmpd")
def test_a_socket_without_a_process_is_kept(self):
mdns = _by_port(parse_listening_sockets(_netstat()))[("udp", 5353, "0.0.0.0")]
assert (mdns["process"], mdns["pid"], mdns["unit"]) == (None, None, None)
def test_the_headers_are_no_sockets(self):
assert len(parse_listening_sockets(_netstat())) == 10
def test_net_tools_names_ipv6_and_programs_its_own_way(self):
sockets = _by_port(parse_listening_sockets(_netstat(NET_TOOLS, "")))
assert sockets[("tcp", 22, "::")]["process"] == "sshd"
assert sockets[("tcp", 22, "::")]["pid"] == 700
assert ("udp", 5353, "::") in sockets
def test_netstat_failing_raises(self):
with pytest.raises(ValueError):
parse_listening_sockets(_netstat("netstat: invalid option -- 'p'\n", "", rc=1))
class TestTheNetstatFallback:
"""The command itself, on a host where ss is missing and netstat is not."""
@pytest.mark.skipif(
any(os.path.exists(f"{d}/ss") for d in ("/usr/sbin", "/sbin")),
reason="ss sits on the PATH the command always adds",
)
def test_it_reads_netstat_when_there_is_no_ss(self, tmp_path):
for tool in ("grep", "cut", "sort", "sed", "tr", "cat"):
(tmp_path / tool).symlink_to(shutil.which(tool))
stub = tmp_path / "netstat"
stub.write_text(f"#!/bin/sh\ncat <<'EOF'\n{BUSYBOX}EOF\n")
stub.chmod(0o755)
out = subprocess.run(
["/bin/sh", "-c", LISTENING_SOCKETS_COMMAND],
capture_output=True,
text=True,
env={"PATH": str(tmp_path)},
timeout=30,
).stdout
assert "[netstat]" in out
sockets = _by_port(parse_listening_sockets(out))
assert sockets[("tcp", 443, "0.0.0.0")]["process"] == "uhttpd"
assert len(sockets) == 10
class TestTheCommand:
def test_the_frame_is_not_in_the_command_itself(self):
"""An echoing transport prints the command back; the markers must only
appear once the command has run."""
assert "SOCK_BEGIN" not in LISTENING_SOCKETS_COMMAND
assert "SOCK_END" not in LISTENING_SOCKETS_COMMAND
def test_it_writes_and_changes_nothing(self):
for verb in ("sudo", " > ", ">>", "kill", "rm ", "systemctl "):
assert verb not in LISTENING_SOCKETS_COMMAND
def test_it_is_posix_sh(self):
result = subprocess.run(
["sh", "-n", "-c", LISTENING_SOCKETS_COMMAND], capture_output=True, text=True
)
assert result.returncode == 0, result.stderr
@pytest.mark.skipif(shutil.which("ss") is None, reason="needs ss on this host")
def test_it_runs_and_parses_on_this_host(self):
out = subprocess.run(
["sh", "-c", LISTENING_SOCKETS_COMMAND], capture_output=True, text=True, timeout=60
).stdout
sockets = parse_listening_sockets(out)
assert all(s["proto"] in ("tcp", "udp") and 0 < s["port"] < 65536 for s in sockets)
class _Driver(ListeningSocketsMixin):
def __init__(self, privileged: str, unprivileged: str = "") -> None:
self.answers = {True: privileged, False: unprivileged}
self.calls: list = []
def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
self.calls.append((command, privileged))
return self.answers[privileged]
class TestTheTemplate:
def test_a_driver_supplies_only_the_transport(self):
driver = _Driver(_wire())
reading = driver.get_listening_sockets()
assert reading["attributed"] is True
assert len(reading["sockets"]) == 9
[(command, privileged)] = driver.calls
assert privileged is True
def test_the_command_reaches_root_as_one_shell(self):
"""``sudo -n a; b`` runs only ``a`` as root: the whole script goes as
one ``sh -c`` argument."""
driver = _Driver(_wire())
driver.get_listening_sockets()
[(command, _privileged)] = driver.calls
assert command.startswith("sh -c '")
assert subprocess.run(["sh", "-n", "-c", command]).returncode == 0
def test_without_root_it_reads_what_it_can(self):
"""Without sudo, ss names only the user's own processes: the sockets
are still worth having, marked as not attributed."""
driver = _Driver("sudo: a password is required\n", _wire(cgroups=""))
reading = driver.get_listening_sockets()
assert reading["attributed"] is False
assert len(reading["sockets"]) == 9
assert [p for _c, p in driver.calls] == [True, False]
def test_a_host_without_either_is_not_asked_twice(self):
driver = _Driver("SOCK_BEGIN\n[no-ss]\nSOCK_END\n")
with pytest.raises(ListeningSocketsUnavailable):
driver.get_listening_sockets()
assert len(driver.calls) == 1
def test_not_every_os_driver_has_it(self):
"""A Windows host is an OSDriver too and has no ss: ``hasattr`` has to
stay a truthful answer, so the drivers that can mix this in themselves."""
assert not issubclass(OSDriver, ListeningSocketsMixin)
assert not hasattr(OSDriver, "get_listening_sockets")
+102
View File
@@ -0,0 +1,102 @@
"""What provisioning a VM from a cloud image needs to know about the guest,
the same for every hypervisor (NetOrk/netork#794)."""
from __future__ import annotations
import pytest
from napalm_device_types import HypervisorDriver
from napalm_device_types.provisioning import (
QEMU_GUEST_AGENTS,
network_config,
split_compression,
)
class TestGuestAgentDeclaration:
"""netOrk's cloud-init installs the agent through which the hypervisor
reads the new VM's IP. Which agent depends on the hypervisor and on the
guest; the guests a driver lists are the ones it can provision."""
def test_the_base_provisions_linux_with_qemu_guest_agent(self):
assert HypervisorDriver.GUEST_AGENTS == {
"linux": (("qemu-guest-agent",), ("systemctl enable --now qemu-guest-agent",)),
}
def test_the_declaration_is_data_netork_reads_off_the_class(self):
assert not callable(vars(HypervisorDriver)["GUEST_AGENTS"])
def test_the_per_hypervisor_attributes_are_gone(self):
assert not hasattr(HypervisorDriver, "GUEST_AGENT_PACKAGES")
assert not hasattr(HypervisorDriver, "GUEST_AGENT_RUNCMD")
class TestQemuGuestAgents:
"""Verified on FreeBSD 15.1 and OpenBSD 7.9 cloud images (#793)."""
def test_linux_is_the_base_default(self):
assert QEMU_GUEST_AGENTS["linux"] == HypervisorDriver.GUEST_AGENTS["linux"]
def test_freebsd(self):
assert QEMU_GUEST_AGENTS["freebsd"] == (
("qemu-guest-agent",),
("sysrc qemu_guest_agent_enable=YES", "service qemu-guest-agent start"),
)
def test_openbsd(self):
assert QEMU_GUEST_AGENTS["openbsd"] == (
("qemu-ga",),
("rcctl enable qemu_ga", "rcctl start qemu_ga"),
)
@pytest.mark.parametrize("guest_os", ["freebsd", "openbsd"])
def test_no_bsd_guest_is_told_to_use_systemd(self, guest_os):
_, runcmd = QEMU_GUEST_AGENTS[guest_os]
assert not any("systemctl" in command for command in runcmd)
class TestNetworkConfig:
"""cloud-init's network-config v2. FreeBSD's nuageinit reads only this
version; the v1 Proxmox generates makes it skip runcmd (#793)."""
def test_dhcp_nics_matched_by_mac(self):
cfg = network_config([("00:50:56:aa:bb:cc", True), ("00:50:56:aa:bb:dd", False)])
assert cfg == {
"version": 2,
"ethernets": {
"nic0": {"match": {"macaddress": "00:50:56:aa:bb:cc"}, "dhcp4": True},
},
}
def test_no_dhcp_nic_means_no_network_config(self):
assert network_config([("00:50:56:aa:bb:cc", False)]) is None
def test_macs_are_written_in_lower_case(self):
"""Proxmox reports them upper-case; nuageinit compares them as given."""
cfg = network_config([("BC:24:11:AA:BB:02", True)])
assert cfg["ethernets"]["nic0"]["match"]["macaddress"] == "bc:24:11:aa:bb:02"
class TestSplitCompression:
"""Official FreeBSD images come packed; hypervisors import unpacked ones."""
def test_xz(self):
assert split_compression("FreeBSD-15.1-RELEASE-amd64-BASIC-CLOUDINIT-ufs.qcow2.xz") == (
"FreeBSD-15.1-RELEASE-amd64-BASIC-CLOUDINIT-ufs.qcow2",
"xz -dc",
)
@pytest.mark.parametrize(
("name", "command"),
[("a.raw.gz", "gzip -dc"), ("a.raw.bz2", "bzip2 -dc"), ("a.qcow2.zst", "zstd -dc")],
)
def test_other_packers(self, name, command):
assert split_compression(name) == (name.rsplit(".", 1)[0], command)
@pytest.mark.parametrize(
"name", ["debian-13-genericcloud-amd64.qcow2", "ubuntu-24.04-server-cloudimg-amd64.img"]
)
def test_an_unpacked_image_is_left_as_it_is(self, name):
assert split_compression(name) == (name, None)
def test_the_suffix_is_matched_case_insensitively(self):
assert split_compression("IMAGE.QCOW2.XZ") == ("IMAGE.QCOW2", "xz -dc")
+303
View File
@@ -0,0 +1,303 @@
"""OpenSSH server hardening: removing algorithms a scanner flagged, without locking anyone out.
The plan is pure and tested on its own. The mixin is tested against a fake host
that keeps files and answers ``sshd -T`` from them, so every path -- the
drop-in, the sshd_config fallback, a refused config, a reload the new session
cannot get through -- runs without a machine.
"""
from __future__ import annotations
import re
from typing import Dict, List, Optional
import pytest
from napalm_device_types.channel import CommandResult
from napalm_device_types.ssh_server import (
DROPIN_PATH,
MAIN_CONFIG,
OpenSshServerMixin,
has_dropin_include,
parse_sshd_effective,
plan_ssh_hardening,
render_dropin,
render_main_config,
)
SSHD_T = """\
port 22
kexalgorithms sntrup761x25519-sha512@openssh.com,curve25519-sha256,curve25519-sha256@libssh.org,ecdh-sha2-nistp256,ecdh-sha2-nistp384,diffie-hellman-group14-sha256
hostkeyalgorithms ssh-ed25519,ecdsa-sha2-nistp256,rsa-sha2-512,rsa-sha2-256
ciphers chacha20-poly1305@openssh.com,aes128-ctr,aes256-gcm@openssh.com
macs umac-64-etm@openssh.com,hmac-sha2-256-etm@openssh.com,hmac-sha1,hmac-sha2-256
passwordauthentication no
"""
PARAMIKO = {
"kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "diffie-hellman-group14-sha256"],
"hostkey": ["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-512"],
"cipher": ["aes128-ctr", "aes256-gcm@openssh.com"],
"mac": ["hmac-sha2-256", "hmac-sha2-256-etm@openssh.com", "hmac-sha1"],
}
# ── parse ────────────────────────────────────────────────────────────────
def test_parse_reads_the_four_lists_in_server_order():
eff = parse_sshd_effective(SSHD_T)
assert eff["kex"][:2] == ["sntrup761x25519-sha512@openssh.com", "curve25519-sha256"]
assert eff["hostkey"] == ["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-512", "rsa-sha2-256"]
assert eff["cipher"] == ["chacha20-poly1305@openssh.com", "aes128-ctr", "aes256-gcm@openssh.com"]
assert eff["mac"][-1] == "hmac-sha2-256"
def test_parse_refuses_output_without_the_lists():
with pytest.raises(ValueError):
parse_sshd_effective("port 22\n")
# ── plan ─────────────────────────────────────────────────────────────────
def test_plan_removes_what_was_asked_and_keeps_the_order():
plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"mac": ["hmac-sha1", "umac-64-etm@openssh.com"]})
assert plan.remove == {"mac": ["umac-64-etm@openssh.com", "hmac-sha1"]}
assert plan.keep["mac"] == ["hmac-sha2-256-etm@openssh.com", "hmac-sha2-256"]
assert plan.changed_classes == ["mac"]
assert plan.refused == []
def test_plan_ignores_an_algorithm_the_server_no_longer_offers():
plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"cipher": ["3des-cbc"]})
assert plan.changed_classes == []
assert [r.algorithm for r in plan.not_offered] == ["3des-cbc"]
def test_plan_never_removes_the_last_algorithm_of_a_class():
plan = plan_ssh_hardening(
parse_sshd_effective(SSHD_T),
{"cipher": ["chacha20-poly1305@openssh.com", "aes128-ctr", "aes256-gcm@openssh.com"]},
)
# The first in the server's order stays, the rest go.
assert plan.keep["cipher"] == ["chacha20-poly1305@openssh.com"]
assert plan.remove["cipher"] == ["aes128-ctr", "aes256-gcm@openssh.com"]
assert [(r.kind, r.algorithm, r.reason) for r in plan.refused] == [
("cipher", "chacha20-poly1305@openssh.com", "last")
]
def test_plan_keeps_one_algorithm_the_managing_client_still_speaks():
"""paramiko knows no sntrup or plain curve25519: removing everything it does
know would leave the server reachable, but not by netOrk."""
plan = plan_ssh_hardening(
parse_sshd_effective(SSHD_T),
{"kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "ecdh-sha2-nistp384",
"diffie-hellman-group14-sha256"]},
client=PARAMIKO,
)
assert "curve25519-sha256@libssh.org" in plan.keep["kex"]
assert [(r.algorithm, r.reason) for r in plan.refused] == [("curve25519-sha256@libssh.org", "client")]
assert plan.remove["kex"] == ["ecdh-sha2-nistp256", "ecdh-sha2-nistp384", "diffie-hellman-group14-sha256"]
def test_plan_without_a_client_list_only_guards_the_last_algorithm():
plan = plan_ssh_hardening(
parse_sshd_effective(SSHD_T),
{"kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "ecdh-sha2-nistp384",
"diffie-hellman-group14-sha256"]},
)
assert plan.refused == []
assert plan.keep["kex"] == ["sntrup761x25519-sha512@openssh.com", "curve25519-sha256"]
def test_plan_rejects_an_unknown_class():
with pytest.raises(ValueError):
plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"compression": ["zlib"]})
# ── render ───────────────────────────────────────────────────────────────
def test_dropin_states_full_lists_for_the_changed_classes_only():
plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"mac": ["hmac-sha1"]})
text = render_dropin(plan.keep, ["mac"])
assert "MACs umac-64-etm@openssh.com,hmac-sha2-256-etm@openssh.com,hmac-sha2-256\n" in text
assert "KexAlgorithms" not in text
assert text.startswith("#") # says who wrote it
def test_include_is_found_only_in_the_global_section():
assert has_dropin_include("Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n")
assert has_dropin_include(" include /etc/ssh/sshd_config.d/*.conf\n")
assert not has_dropin_include("Port 22\n")
assert not has_dropin_include("Port 22\nMatch User x\n Include /etc/ssh/sshd_config.d/*.conf\n")
def test_main_config_gets_the_lists_first_and_the_old_lines_disabled():
main = "# defaults\nPort 22\nMACs hmac-sha1,hmac-sha2-256\nMatch User backup\n MACs hmac-sha1\n"
text = render_main_config(main, {"mac": ["hmac-sha2-256"]}, ["mac"])
lines = text.splitlines()
first_directive = next(line for line in lines if line and not line.startswith("#"))
assert first_directive == "MACs hmac-sha2-256"
assert "# netork: MACs hmac-sha1,hmac-sha2-256" in lines
# A Match block is left alone: it is not the global setting.
assert " MACs hmac-sha1" in lines
def test_main_config_rewrite_is_idempotent():
main = "Port 22\nMACs hmac-sha1,hmac-sha2-256\n"
once = render_main_config(main, {"mac": ["hmac-sha2-256"]}, ["mac"])
twice = render_main_config(once, {"mac": ["hmac-sha2-256"]}, ["mac"])
assert once == twice
# ── the mixin, against a fake host ───────────────────────────────────────
_DIRECTIVE = {"kexalgorithms": "kex", "hostkeyalgorithms": "hostkey", "ciphers": "cipher", "macs": "mac"}
class FakeHost(OpenSshServerMixin):
"""A host with files, an sshd that reads them first-match-wins, and a log of what ran."""
def __init__(self, main: str = "Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n") -> None:
self.files: Dict[str, str] = {MAIN_CONFIG: main}
self.defaults = parse_sshd_effective(SSHD_T)
self.ran: List[str] = []
self.syntax_ok = True
self.reloads = 0
def _effective(self) -> Dict[str, List[str]]:
eff = {k: list(v) for k, v in self.defaults.items()}
seen = set()
lines: List[str] = []
for line in self.files[MAIN_CONFIG].splitlines():
if re.match(r"\s*include\s", line, re.I):
lines += self.files.get(DROPIN_PATH, "").splitlines()
elif re.match(r"\s*match\s", line, re.I):
break
else:
lines.append(line)
for line in lines:
parts = line.split()
if len(parts) == 2 and parts[0].lower() in _DIRECTIVE:
kind = _DIRECTIVE[parts[0].lower()]
if kind not in seen:
seen.add(kind)
eff[kind] = parts[1].split(",")
return eff
def run_command(self, command: str, *, privileged: bool = False, timeout: float = 60,
stdin: Optional[bytes] = None) -> CommandResult:
self.ran.append(command)
assert privileged, command
if command.startswith("sshd -T") or "sshd -T" in command:
eff = self._effective()
out = "".join(f"{d} {','.join(eff[k])}\n" for d, k in _DIRECTIVE.items())
return CommandResult(out, "", 0)
if "sshd -t" in command:
return CommandResult("", "" if self.syntax_ok else "bad line 3", 0 if self.syntax_ok else 255)
m = re.match(r"cat -- '?([^' ]+)'?$", command)
if m:
path = m.group(1)
if path in self.files:
return CommandResult(self.files[path], "", 0)
return CommandResult("", "No such file", 1)
m = re.search(r"mv -f -- \"\$tmp\" '?([^' ]+)'?", command)
if m:
self.files[m.group(1)] = (stdin or b"").decode()
return CommandResult("", "", 0)
m = re.match(r"rm -f -- '?([^' ]+)'?$", command)
if m:
self.files.pop(m.group(1), None)
return CommandResult("", "", 0)
if "reload" in command or "HUP" in command:
self.reloads += 1
return CommandResult("", "", 0)
if command.startswith("mkdir"):
return CommandResult("", "", 0)
raise AssertionError(f"unexpected command: {command}")
def test_mixin_writes_the_dropin_checks_it_and_reloads():
host = FakeHost()
lines: List[str] = []
result = host.harden_ssh_server({"mac": ["hmac-sha1"]}, client=PARAMIKO, log=lines.append)
assert result.applied and not result.rolled_back
assert result.path == DROPIN_PATH
assert "hmac-sha1" not in host._effective()["mac"]
assert host.reloads == 1
assert any("sshd -t" in c for c in host.ran)
assert lines # it says what it does
def test_mixin_keeps_what_an_earlier_fix_wrote():
host = FakeHost()
host.harden_ssh_server({"hostkey": ["ecdsa-sha2-nistp256"]})
host.harden_ssh_server({"mac": ["hmac-sha1"]})
eff = host._effective()
assert "ecdsa-sha2-nistp256" not in eff["hostkey"]
assert "hmac-sha1" not in eff["mac"]
def test_mixin_falls_back_to_sshd_config_without_an_include():
host = FakeHost(main="Port 22\nMACs hmac-sha1,hmac-sha2-256\n")
result = host.harden_ssh_server({"mac": ["hmac-sha1"]})
assert result.path == MAIN_CONFIG
assert result.backup and host.files[result.backup] == "Port 22\nMACs hmac-sha1,hmac-sha2-256\n"
assert host._effective()["mac"] == ["hmac-sha2-256"]
def test_mixin_restores_when_sshd_rejects_the_config():
host = FakeHost()
host.syntax_ok = False
result = host.harden_ssh_server({"mac": ["hmac-sha1"]})
assert not result.applied and result.rolled_back
assert DROPIN_PATH not in host.files
assert host.reloads == 0
assert "bad line 3" in result.message
def test_mixin_restores_when_the_effective_lists_do_not_change():
"""Another file read before ours sets the list: ours is valid and ignored."""
host = FakeHost(main="MACs hmac-sha1,hmac-sha2-256\nInclude /etc/ssh/sshd_config.d/*.conf\n")
result = host.harden_ssh_server({"mac": ["hmac-sha1"]})
assert not result.applied and result.rolled_back
assert DROPIN_PATH not in host.files
assert "MACs" in result.message
def test_mixin_rolls_back_when_a_new_session_cannot_get_in():
host = FakeHost()
host.files[DROPIN_PATH] = "# netOrk\nHostKeyAlgorithms ssh-ed25519\n"
before = host.files[DROPIN_PATH]
result = host.harden_ssh_server({"mac": ["hmac-sha1"]}, verify=lambda: False)
assert not result.applied and result.rolled_back
assert host.files[DROPIN_PATH] == before
assert host.reloads == 2 # the change, and the way back
def test_mixin_does_nothing_when_nothing_would_change():
host = FakeHost()
result = host.harden_ssh_server({"cipher": ["3des-cbc"]})
assert not result.applied and not result.rolled_back
assert host.reloads == 0
assert DROPIN_PATH not in host.files
def test_declared_on_no_base_class():
"""hasattr is the capability probe: only a driver that mixes it in answers True."""
from napalm_device_types import DeviceTypeDriver, OSDriver
assert not hasattr(DeviceTypeDriver, "harden_ssh_server")
assert not hasattr(OSDriver, "harden_ssh_server")
def test_main_config_second_fix_keeps_what_the_first_set():
host = FakeHost(main="Port 22\n")
host.harden_ssh_server({"hostkey": ["ecdsa-sha2-nistp256"]})
host.harden_ssh_server({"mac": ["hmac-sha1"]})
eff = host._effective()
assert "ecdsa-sha2-nistp256" not in eff["hostkey"]
assert "hmac-sha1" not in eff["mac"]
+617
View File
@@ -0,0 +1,617 @@
"""get_virtual_ips: the addresses a host's HA configuration lets float between machines.
A virtual IP is not visible from the address list alone: vip-manager and
keepalived put it on whichever node is master *right now*, and a standby has no
trace of it in ``ip addr``. What every member has is the configuration that
declares it. Reading that is the same on every Linux host, so the command and
its parse live here once, and a driver only carries the command across
(netOrk #829).
The configuration also holds secrets -- vip-manager's etcd password, keepalived's
``auth_pass`` -- so the command filters on the host: only allowlisted fields
ever leave it. The tests below run the command for real to hold it to that.
"""
from __future__ import annotations
import os
import shutil
import subprocess
from pathlib import Path
import pytest
from napalm_device_types import OSDriver
from napalm_device_types.virtual_ips import (
VIRTUAL_IPS_COMMAND,
VirtualIpsMixin,
parse_virtual_ips,
)
SECRET = "s3cr3t-do-not-leak"
def _wire(*sections: str, noise: str = "") -> str:
"""The report as the command prints it, framed."""
return f"{noise}VIPS_BEGIN\n{''.join(sections)}VIPS_END\n"
def _section(mechanism: str, unit: str, *records: str) -> str:
return f"[{mechanism} {unit}]\n" + "".join(f"{r}\n" for r in records)
def _vip_manager(*records: str, unit: str = "vip-manager.service") -> str:
return _section("vip-manager", unit, *records)
def _keepalived(*records: str, unit: str = "keepalived.service") -> str:
return _section("keepalived", unit, *records)
DB01 = _vip_manager(
"active active",
"arg config /etc/default/vip-manager.yml",
"cfg ip 10.7.224.10",
"cfg netmask 24",
"cfg interface ens7",
'cfg trigger-key "/service/netork-db/leader"',
)
class TestVipManager:
def test_the_5x_yaml_declares_the_address(self):
reading = parse_virtual_ips(_wire(DB01))
assert reading == {
"vip_manager": [
{
"mechanism": "vip-manager",
"instance": "vip-manager",
"address": "10.7.224.10",
"prefix_length": 24,
"interface": "ens7",
"group_key": "/service/netork-db/leader",
"running": True,
}
],
"keepalived": [],
}
def test_the_1x_unit_takes_its_values_from_the_environment_file(self):
"""Ubuntu's 1.0.2 package: ``-ip=${VIP_IP}`` on the command line,
the values in /etc/default/vip-manager."""
section = _vip_manager(
"active active",
"arg ip ${VIP_IP}",
"arg mask $VIP_MASK",
"arg iface ${VIP_IFACE}",
"arg key ${VIP_KEY}",
"env VIP_IP 10.0.0.10",
"env VIP_MASK 255.255.255.0",
"env VIP_IFACE eth0",
'env VIP_KEY "/service/pg/leader"',
)
[entry] = parse_virtual_ips(_wire(section))["vip_manager"]
assert (entry["address"], entry["prefix_length"], entry["interface"]) == (
"10.0.0.10",
24,
"eth0",
)
assert entry["group_key"] == "/service/pg/leader"
def test_a_flag_beats_the_environment_which_beats_the_config_file(self):
section = _vip_manager(
"active active",
"arg ip 10.0.0.1",
"env VIP_IP 10.0.0.2",
"env VIP_NETMASK 16",
"cfg ip 10.0.0.3",
"cfg netmask 24",
"cfg interface eth1",
)
[entry] = parse_virtual_ips(_wire(section))["vip_manager"]
assert (entry["address"], entry["prefix_length"], entry["interface"]) == (
"10.0.0.1",
16,
"eth1",
)
@pytest.mark.parametrize(
"value, expected",
[
("10.0.0.10 # the database", "10.0.0.10"),
("'10.0.0.10'", "10.0.0.10"),
('"fd00::10"', "fd00::10"),
("FD00:0:0::10", "fd00::10"),
],
)
def test_values_lose_quotes_and_comments(self, value, expected):
section = _vip_manager("active active", f"cfg ip {value}")
[entry] = parse_virtual_ips(_wire(section))["vip_manager"]
assert entry["address"] == expected
def test_a_stopped_unit_still_declares_its_address(self):
"""A standby whose vip-manager is down is still a member -- one that
cannot take the address over, which is worth knowing."""
section = _vip_manager("active inactive", "cfg ip 10.0.0.10")
[entry] = parse_virtual_ips(_wire(section))["vip_manager"]
assert entry["running"] is False
assert entry["prefix_length"] is None
def test_a_unit_without_an_address_declares_nothing(self):
section = _vip_manager("active active", "arg ip ${VIP_IP}", "cfg netmask 24")
assert parse_virtual_ips(_wire(section))["vip_manager"] == []
def test_a_template_unit_is_named_after_its_instance(self):
section = _vip_manager(
"active active", "cfg ip 10.0.0.10", unit="vip-manager@pg16.service"
)
[entry] = parse_virtual_ips(_wire(section))["vip_manager"]
assert entry["instance"] == "vip-manager@pg16"
def test_an_unreadable_config_is_not_looked_at(self):
"""None, not []: the reading must not say the address is gone."""
section = _vip_manager("active active", "unreadable /etc/default/vip-manager.yml")
reading = parse_virtual_ips(_wire(section, _keepalived("active inactive")))
assert reading["vip_manager"] is None
assert reading["keepalived"] == []
class TestKeepalived:
TWO_INSTANCES = _keepalived(
"active active",
"instance VI_DB",
"interface eth0",
"virtual_router_id 51",
"vip 10.0.0.10/24 dev eth0",
"vip 10.0.0.11",
"end",
"instance web",
"vip fd00::80",
"interface eth1",
"end",
)
def test_each_address_of_each_instance(self):
reading = parse_virtual_ips(_wire(self.TWO_INSTANCES))
assert reading["vip_manager"] == []
assert reading["keepalived"] == [
{
"mechanism": "keepalived",
"instance": "VI_DB",
"address": "10.0.0.10",
"prefix_length": 24,
"interface": "eth0",
"group_key": "vrid:51",
"running": True,
},
{
"mechanism": "keepalived",
"instance": "VI_DB",
"address": "10.0.0.11",
"prefix_length": 32,
"interface": "eth0",
"group_key": "vrid:51",
"running": True,
},
{
"mechanism": "keepalived",
"instance": "web",
"address": "fd00::80",
"prefix_length": 128,
"interface": "eth1",
"group_key": None,
"running": True,
},
]
def test_dev_beats_the_instance_interface(self):
section = _keepalived(
"active active", "instance a", "interface eth0", "vip 10.0.0.10/24 dev eth9", "end"
)
[entry] = parse_virtual_ips(_wire(section))["keepalived"]
assert entry["interface"] == "eth9"
@pytest.mark.parametrize("line", ["vip $VIP", "vip @node1", "vip not-an-address"])
def test_what_is_no_address_is_skipped(self, line):
"""keepalived's ``$VAR`` substitution and ``@host`` conditionals are
not resolved here; an entry that is no address is no entry."""
section = _keepalived("active active", "instance a", line, "vip 10.0.0.12", "end")
assert [e["address"] for e in parse_virtual_ips(_wire(section))["keepalived"]] == [
"10.0.0.12"
]
def test_an_instance_cut_off_by_the_end_of_the_file_still_counts(self):
section = _keepalived("active active", "instance a", "vip 10.0.0.10")
assert len(parse_virtual_ips(_wire(section))["keepalived"]) == 1
def test_without_a_config_file_it_declares_nothing(self):
section = _keepalived("active inactive", "missing /etc/keepalived/keepalived.conf")
assert parse_virtual_ips(_wire(section))["keepalived"] == []
@pytest.mark.parametrize(
"record", ["unreadable /etc/keepalived/conf.d/x.conf", "too-deep /etc/keepalived/a.conf"]
)
def test_an_include_it_could_not_follow_is_not_looked_at(self, record):
section = _keepalived("active active", "instance a", "vip 10.0.0.10", "end", record)
assert parse_virtual_ips(_wire(section))["keepalived"] is None
class TestTheReport:
def test_no_units_is_looked_and_found_nothing(self):
assert parse_virtual_ips(_wire("[systemd]\n")) == {"vip_manager": [], "keepalived": []}
def test_a_host_without_systemd_is_not_looked_at(self):
assert parse_virtual_ips(_wire("[no-systemd]\n")) == {
"vip_manager": None,
"keepalived": None,
}
def test_noise_before_the_report_is_ignored(self):
reading = parse_virtual_ips(_wire(DB01, noise="$ sh -c '...'\nWelcome to Ubuntu\n"))
assert reading["vip_manager"][0]["address"] == "10.7.224.10"
def test_a_report_cut_short_raises_without_quoting_it(self):
"""The output may carry configuration; an error message must not."""
with pytest.raises(ValueError) as caught:
parse_virtual_ips(f"VIPS_BEGIN\n{DB01}")
assert "10.7.224.10" not in str(caught.value)
def test_entries_are_sorted(self):
reading = parse_virtual_ips(
_wire(
_vip_manager("active active", "cfg ip 10.0.0.20", unit="vip-manager@b.service"),
_vip_manager("active active", "cfg ip 10.0.0.3", unit="vip-manager@a.service"),
)
)
assert [e["address"] for e in reading["vip_manager"]] == ["10.0.0.3", "10.0.0.20"]
# ---------------------------------------------------------------------------
# The command itself, run against a stub systemctl and real files
# ---------------------------------------------------------------------------
DB01_YAML = f"""# managed by hand, see infrastructure/docs/DATABASE_CLUSTER.md
ip: 10.7.224.10
netmask: 24
interface: ens7
trigger-key: "/service/netork-db/leader"
trigger-value: "db-01"
dcs-type: etcd
dcs-endpoints:
- http://10.7.234.12:2379
etcd-user: patroni
etcd-password: {SECRET}
"""
KEEPALIVED_CONF = f"""! Configuration File for keepalived
global_defs {{
notification_email {{ ops@example.org }}
router_id LVS_{SECRET}
}}
vrrp_script chk_haproxy {{
script "/usr/local/bin/check {SECRET}"
}}
vrrp_instance VI_WEB {{
state MASTER
interface eth0
virtual_router_id 51
priority 101
authentication {{
auth_type PASS
auth_pass {SECRET}
}}
virtual_ipaddress {{
192.0.2.10/24 dev eth0 label eth0:1
}}
notify_master "/etc/keepalived/master.sh {SECRET}"
}}
include conf.d/*.conf
"""
INCLUDED_CONF = """vrrp_instance VI_DB { interface eth1
virtual_router_id 52
virtual_ipaddress { 192.0.2.20 }
virtual_ipaddress_excluded {
2001:db8::20/64
}
}
"""
def _awks() -> list:
found = []
if shutil.which("mawk"):
found.append(pytest.param(["mawk"], id="mawk"))
if shutil.which("busybox"):
found.append(pytest.param(["busybox", "awk"], id="busybox"))
if shutil.which("gawk"):
found.append(pytest.param(["gawk"], id="gawk"))
return found or [pytest.param(None, marks=pytest.mark.skip(reason="no awk to run"))]
class _Host:
"""A host as far as the command can tell: units, their properties, files."""
def __init__(self, root: Path, awk: list) -> None:
self.root = root
self.bin = root / "bin"
self.bin.mkdir()
self.state = root / "systemd"
self.state.mkdir()
(self.state / "units").write_text("")
stub = self.bin / "systemctl"
stub.write_text(
"#!/bin/sh\n"
f"d={self.state}\n"
'case "$1" in\n'
f' list-units) {shutil.which("cat")} "$d/units" ;;\n'
f' show) for last; do :; done; {shutil.which("cat")} "$d/show-$last" ;;\n'
"esac\n"
)
stub.chmod(0o755)
wrapper = self.bin / "awk"
program = " ".join([shutil.which(awk[0]) or awk[0], *awk[1:]])
wrapper.write_text(f'#!/bin/sh\nexec {program} "$@"\n')
wrapper.chmod(0o755)
def unit(self, name: str, *properties: str) -> None:
with (self.state / "units").open("a") as units:
units.write(f"{name} loaded active running {name}\n")
(self.state / f"show-{name}").write_text("".join(f"{p}\n" for p in properties))
def file(self, relative: str, content: str) -> Path:
path = self.root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
return path
def run(self, *, systemctl: bool = True) -> str:
if not systemctl:
(self.bin / "systemctl").unlink()
return subprocess.run(
["/bin/sh", "-c", VIRTUAL_IPS_COMMAND],
capture_output=True,
text=True,
env={"PATH": str(self.bin)},
timeout=30,
).stdout
def _exec_start(*argv: str) -> str:
return (
f"ExecStart={{ path={argv[0]} ; argv[]={' '.join(argv)} ; ignore_errors=no ; "
"start_time=[Mon 2026-10-05 06:50:00 UTC] ; stop_time=[n/a] ; pid=758 ; "
"code=(null) ; status=0/0 }"
)
@pytest.fixture(params=_awks())
def host(request, tmp_path) -> _Host:
return _Host(tmp_path, request.param)
class TestTheCommandOnAHost:
def test_vip_manager_5x_as_on_db_01(self, host):
config = host.file("etc/default/vip-manager.yml", DB01_YAML)
host.unit(
"vip-manager.service",
"ActiveState=active",
_exec_start("/usr/bin/vip-manager", f"--config={config}"),
f"Environment=VIP_ETCD_PASSWORD={SECRET} VIP_INTERFACE=ens7",
)
out = host.run()
assert SECRET not in out
assert "10.7.234.12" not in out # the etcd endpoints stay on the host
assert parse_virtual_ips(out)["vip_manager"] == [
{
"mechanism": "vip-manager",
"instance": "vip-manager",
"address": "10.7.224.10",
"prefix_length": 24,
"interface": "ens7",
"group_key": "/service/netork-db/leader",
"running": True,
}
]
def test_vip_manager_1x_with_its_environment_file(self, host):
envfile = host.file(
"etc/default/vip-manager",
f'VIP_IP="10.0.0.10"\nexport VIP_MASK=24\nVIP_IFACE=eth0\nVIP_KEY=/service/pg/leader\n'
f"VIP_ETCD_PASSWORD={SECRET}\n",
)
host.unit(
"vip-manager.service",
"ActiveState=active",
_exec_start(
"/usr/bin/vip-manager",
"-ip=${VIP_IP}",
"-mask=${VIP_MASK}",
"-iface=${VIP_IFACE}",
"-key=${VIP_KEY}",
"-etcd_password",
SECRET,
),
f"EnvironmentFiles={envfile} (ignore_errors=yes)",
f"EnvironmentFiles={host.root}/etc/default/missing (ignore_errors=yes)",
)
out = host.run()
assert SECRET not in out
[entry] = parse_virtual_ips(out)["vip_manager"]
assert (entry["address"], entry["prefix_length"], entry["interface"]) == (
"10.0.0.10",
24,
"eth0",
)
def test_keepalived_with_includes(self, host):
config = host.file("etc/keepalived/keepalived.conf", KEEPALIVED_CONF)
host.file("etc/keepalived/conf.d/db.conf", INCLUDED_CONF)
host.unit(
"keepalived.service",
"ActiveState=active",
_exec_start("/usr/sbin/keepalived", "--dont-fork", "-f", str(config)),
)
out = host.run()
assert SECRET not in out
assert "MASTER" not in out and "priority" not in out
reading = parse_virtual_ips(out)
assert [(e["instance"], e["address"], e["prefix_length"], e["interface"], e["group_key"])
for e in reading["keepalived"]] == [
("VI_WEB", "192.0.2.10", 24, "eth0", "vrid:51"),
("VI_DB", "192.0.2.20", 32, "eth1", "vrid:52"),
("VI_DB", "2001:db8::20", 64, "eth1", "vrid:52"),
]
assert reading["vip_manager"] == []
def test_an_include_that_includes_itself_is_not_followed_forever(self, host):
config = host.file(
"etc/keepalived/keepalived.conf",
"vrrp_instance a {\n virtual_ipaddress {\n 192.0.2.30\n }\n}\n"
"include keepalived.conf\n",
)
host.unit(
"keepalived.service",
"ActiveState=active",
_exec_start("/usr/sbin/keepalived", f"--use-file={config}"),
)
assert parse_virtual_ips(host.run())["keepalived"] is None
def test_an_include_pattern_is_never_run_as_shell(self, host):
"""The include line comes from a file and the command runs as root."""
marker = host.root / "PWNED"
config = host.file(
"etc/keepalived/keepalived.conf",
f"include /etc/x.conf;touch${{IFS}}{marker}\ninclude $(touch {marker})\n",
)
host.unit(
"keepalived.service",
"ActiveState=active",
_exec_start("/usr/sbin/keepalived", "-f", str(config)),
)
out = host.run()
assert not marker.exists()
assert parse_virtual_ips(out)["keepalived"] is None
@pytest.mark.skipif(os.geteuid() == 0, reason="root reads every file")
def test_a_config_it_may_not_read_is_reported_as_such(self, host):
config = host.file("etc/default/vip-manager.yml", DB01_YAML)
config.chmod(0o000)
host.unit(
"vip-manager.service",
"ActiveState=active",
_exec_start("/usr/bin/vip-manager", "--config", str(config)),
)
out = host.run()
assert parse_virtual_ips(out)["vip_manager"] is None
def test_no_units_at_all(self, host):
assert parse_virtual_ips(host.run()) == {"vip_manager": [], "keepalived": []}
def test_a_host_without_systemd(self, host):
assert parse_virtual_ips(host.run(systemctl=False)) == {
"vip_manager": None,
"keepalived": None,
}
class TestTheCommand:
def test_the_frame_is_not_in_the_command_itself(self):
assert "VIPS_BEGIN" not in VIRTUAL_IPS_COMMAND
assert "VIPS_END" not in VIRTUAL_IPS_COMMAND
def test_it_writes_and_changes_nothing(self):
for verb in ("sudo", " > ", ">>", "kill", "rm ", "start", "stop", "restart", "reload"):
assert verb not in VIRTUAL_IPS_COMMAND
def test_it_is_posix_sh(self):
result = subprocess.run(
["sh", "-n", "-c", VIRTUAL_IPS_COMMAND], capture_output=True, text=True
)
assert result.returncode == 0, result.stderr
def test_it_is_one_line(self):
"""About 4 kB with its awk program -- past the line a terminal takes, so a
driver sends it on an exec channel. One argument there, one line."""
driver = _Driver(_wire())
driver.get_virtual_ips()
[(command, _privileged)] = driver.calls
assert "\n" not in command
class _Driver(VirtualIpsMixin):
def __init__(self, privileged: str, unprivileged: str = "") -> None:
self.answers = {True: privileged, False: unprivileged}
self.calls: list = []
def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
self.calls.append((command, privileged))
return self.answers[privileged]
class TestTheTemplate:
def test_a_driver_supplies_only_the_transport(self):
driver = _Driver(_wire(DB01))
reading = driver.get_virtual_ips()
assert reading["vip_manager"][0]["address"] == "10.7.224.10"
[(command, privileged)] = driver.calls
assert privileged is True
assert command.startswith("sh -c '")
assert subprocess.run(["sh", "-n", "-c", command]).returncode == 0
def test_without_root_it_reads_what_it_can(self):
"""The config may be root's alone; what is readable is still worth having."""
driver = _Driver(
"sudo: a password is required\n",
_wire(_vip_manager("active active", "unreadable /etc/default/vip-manager.yml")),
)
reading = driver.get_virtual_ips()
assert reading["vip_manager"] is None
assert [p for _, p in driver.calls] == [True, False]
def test_not_every_os_driver_has_it(self):
assert not issubclass(OSDriver, VirtualIpsMixin)
assert not hasattr(OSDriver, "get_virtual_ips")
-17
View File
@@ -51,20 +51,3 @@ class TestVMConfigCarriesWhatAHardwareViewShows:
from napalm_device_types.models import VMPassthroughDict
assert get_type_hints(VMPassthroughDict) == {"slot": str, "kind": str, "config": str}
class TestGuestAgentDeclaration:
"""netOrk's cloud-init installed qemu-guest-agent on every new VM. A VMware
guest reports its IP through open-vm-tools instead; the hypervisor says
which, and netOrk stops hard-coding one of them."""
def test_default_is_qemu_guest_agent(self):
from napalm_device_types import HypervisorDriver
assert HypervisorDriver.GUEST_AGENT_PACKAGES == ("qemu-guest-agent",)
assert HypervisorDriver.GUEST_AGENT_RUNCMD == ("systemctl enable --now qemu-guest-agent",)
def test_attributes_are_not_methods(self):
from napalm_device_types import HypervisorDriver
assert not callable(vars(HypervisorDriver)["GUEST_AGENT_PACKAGES"])