From 595496549bb33dfcdca9d39a6c2fe2af6b3f139d Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 18 Oct 2026 14:08:09 +0200 Subject: [PATCH] feat: a public command channel; no warnings of its own (0.4.0) run_command and open_stream (napalm-device-types CommandChannelMixin, which AccessPointDriver carries) run on an SSH exec channel next to netmiko's shell: stdin, stderr apart, a real exit code. netOrk sent UCI scripts and the wifi channel through the private _send_command, typing a file in base64 chunks or printf lines; with stdin it writes it with cat, and base64 is needed for nothing (netOrk #1113). OpenWrt has no sudo, so a privileged command from a login other than root is refused before it runs. get_device_warnings goes with it: no_base64 was its last code. --- napalm_openwrt/openwrt.py | 40 +++++++++++++++ napalm_openwrt/system_mixin.py | 13 ----- pyproject.toml | 2 +- tests/unit/test_command_channel.py | 79 ++++++++++++++++++++++++++++++ tests/unit/test_warnings.py | 43 ---------------- 5 files changed, 120 insertions(+), 57 deletions(-) create mode 100644 tests/unit/test_command_channel.py delete mode 100644 tests/unit/test_warnings.py diff --git a/napalm_openwrt/openwrt.py b/napalm_openwrt/openwrt.py index 739ab47..988e00e 100644 --- a/napalm_openwrt/openwrt.py +++ b/napalm_openwrt/openwrt.py @@ -29,10 +29,13 @@ from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationExc from napalm_device_types import ( AccessPointDriver, + ByteStream, + CommandResult, EnsurePackageMixin, FingerprintRule, SnmpAgentMixin, ) +from napalm_device_types.channel import open_stream_on_transport, run_on_transport from napalm_device_types.listening import ListeningSocketsMixin from napalm.base.exceptions import ( ConnectionException, @@ -183,6 +186,43 @@ class OpenWrtDriver( except (socket.error, EOFError) as exc: raise ConnectionClosedException(str(exc)) from exc + # ------------------------------------------------------------------ + # Command channel (napalm-device-types CommandChannelMixin, via AccessPointDriver) + # ------------------------------------------------------------------ + + def _transport(self) -> Any: + """The SSH transport netmiko already holds, for exec channels next to its shell.""" + if self.device is None: + raise ConnectionClosedException("Not connected") + return self.device.remote_conn_pre.get_transport() + + def _check_privilege(self, privileged: bool) -> None: + if privileged and self.username != "root": + raise PermissionError( + f"{self.hostname}: OpenWrt has no sudo; {self.username!r} cannot run as root" + ) + + def run_command( + self, + command: str, + *, + privileged: bool = False, + timeout: float = 60, + stdin: bytes | None = None, + ) -> CommandResult: + """Run *command* on an SSH exec channel: no PTY, stderr apart, a real exit code. + + The login is root on OpenWrt as a rule, and there is no sudo: a + privileged command from any other login is refused before it runs. + """ + self._check_privilege(privileged) + return run_on_transport(self._transport(), command, stdin=stdin, timeout=timeout) + + def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream: + """Start *command* on an SSH exec channel and return a stream to it.""" + self._check_privilege(privileged) + return open_stream_on_transport(self._transport(), command) + def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str: """Run the listening-socket command for ``ListeningSocketsMixin``. diff --git a/napalm_openwrt/system_mixin.py b/napalm_openwrt/system_mixin.py index 0449116..db91e0c 100644 --- a/napalm_openwrt/system_mixin.py +++ b/napalm_openwrt/system_mixin.py @@ -390,19 +390,6 @@ class OpenWrtSystemMixin: output = self._send_command(f"/etc/init.d/{name} {action} 2>&1") return {"success": True, "output": output} - def get_device_warnings(self) -> list[dict[str, Any]]: - """Only a missing ``base64`` binary, which BusyBox may lack. - - It is not a package the caller can look up -- BusyBox may provide it as - an applet -- so it stays here until the shell commands callers send get - a public method. lldpd, auc and the pending updates are facts - (``get_packages``, ``get_available_updates``); whether they are a - problem is the caller's to decide. - """ - if self._send_command("command -v base64 2>/dev/null").strip(): - return [] - return [{"code": "no_base64"}] - def get_available_updates(self) -> list[dict[str, Any]]: """Return list of upgradable packages from the local package manager cache. diff --git a/pyproject.toml b/pyproject.toml index 7d1369c..605870b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-openwrt" -version = "0.3.0" +version = "0.4.0" description = "NAPALM driver for OpenWrt routers/access-points" readme = "README.md" license = { text = "Apache-2.0" } diff --git a/tests/unit/test_command_channel.py b/tests/unit/test_command_channel.py new file mode 100644 index 0000000..56060fc --- /dev/null +++ b/tests/unit/test_command_channel.py @@ -0,0 +1,79 @@ +"""A public command channel (napalm-device-types ``CommandChannelMixin``, netOrk #1113). + +netOrk sent UCI scripts and the wifi channel through the private +``_send_command``: an interactive shell with no exit code, where a file had to +be typed in base64 chunks or ``printf`` lines. ``run_command`` runs on an SSH +exec channel of its own, with stdin, stderr and the exit code -- a script is +written with ``cat`` from stdin, and ``base64`` is needed for nothing. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from napalm.base.exceptions import ConnectionClosedException +from napalm_device_types import CommandResult + +from napalm_openwrt.openwrt import OpenWrtDriver + + +def _driver(username: str = "root") -> OpenWrtDriver: + with patch("napalm_openwrt.openwrt.ConnectHandler"): + drv = OpenWrtDriver(hostname="192.0.2.15", username=username, password="") + drv.device = MagicMock() + return drv + + +def test_a_command_runs_on_the_ssh_transport_with_its_stdin(): + drv = _driver() + transport = drv.device.remote_conn_pre.get_transport.return_value + done = CommandResult("", "", 0) + with patch("napalm_openwrt.openwrt.run_on_transport", return_value=done) as run: + result = drv.run_command("cat > /tmp/x.sh", stdin=b"uci commit\n", timeout=30) + + assert result is done + run.assert_called_once_with(transport, "cat > /tmp/x.sh", stdin=b"uci commit\n", timeout=30) + + +def test_root_runs_privileged_commands_as_they_are(): + drv = _driver() + with patch("napalm_openwrt.openwrt.run_on_transport") as run: + drv.run_command("wifi", privileged=True) + + assert run.call_args.args[1] == "wifi" + + +def test_a_login_other_than_root_cannot_gain_root(): + """OpenWrt has no sudo: refused before anything runs, not half done.""" + drv = _driver(username="admin") + with patch("napalm_openwrt.openwrt.run_on_transport") as run, pytest.raises(PermissionError): + drv.run_command("wifi", privileged=True) + + run.assert_not_called() + + +def test_a_stream_on_the_same_transport(): + drv = _driver() + transport = drv.device.remote_conn_pre.get_transport.return_value + with patch("napalm_openwrt.openwrt.open_stream_on_transport") as stream: + drv.open_stream("logread -f") + + stream.assert_called_once_with(transport, "logread -f") + + +def test_not_connected(): + drv = _driver() + drv.device = None + with pytest.raises(ConnectionClosedException): + drv.run_command("true") + + +def test_no_warnings_of_its_own(): + """base64 is no longer needed for anything, so a missing one is no warning; + lldpd, auc and the updates are facts the caller reads (netOrk #1111).""" + assert not any( + "get_device_warnings" in vars(c) + for c in OpenWrtDriver.__mro__ + if c.__module__.startswith("napalm_openwrt") + ) diff --git a/tests/unit/test_warnings.py b/tests/unit/test_warnings.py deleted file mode 100644 index 06ab904..0000000 --- a/tests/unit/test_warnings.py +++ /dev/null @@ -1,43 +0,0 @@ -"""What the driver still names itself (netOrk #1111). - -lldpd, auc and the pending updates are facts the caller reads (``get_packages``, -``get_available_updates``) and decides on. A missing ``base64`` stays: BusyBox -may provide it as an applet, so no package says whether it is there. -""" - -from __future__ import annotations - -from napalm_openwrt import OpenWrtDriver - - -def _driver(answers: dict[str, str]) -> OpenWrtDriver: - d = OpenWrtDriver("ap", "root", "pw") - issued: list[str] = [] - - def send(command: str) -> str: - issued.append(command) - return next((out for frag, out in answers.items() if frag in command), "") - - d._send_command = send - d.issued = issued - return d - - -def test_nothing_installed_names_only_base64(): - """No lldpd, no auc and an opkg system: the caller names those from the packages.""" - d = _driver({"opkg list-upgradable": "luci - 1.0 - 1.1\n"}) - - assert [w["code"] for w in d.get_device_warnings()] == ["no_base64"] - - -def test_base64_present_names_nothing(): - d = _driver({"command -v base64": "/usr/bin/base64\n"}) - - assert d.get_device_warnings() == [] - - -def test_one_command_per_poll(): - d = _driver({}) - d.get_device_warnings() - - assert d.issued == ["command -v base64 2>/dev/null"]