Christian Manivong
d89bb38688
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 23s
CI / test (3.12) (push) Successful in 26s
CI / test (3.10) (pull_request) Successful in 25s
CI / test (3.11) (pull_request) Successful in 24s
CI / test (3.12) (pull_request) Successful in 25s
HostRebootMixin's reboot_host over the command channel: reboot detached and two seconds late, so the answer comes back before the session goes down, and a refusal is raised instead of being mistaken for the dropped session. netOrk sent /sbin/reboot through the private _send_command and could not tell the two apart (netOrk #1113). A login other than root is refused before anything runs.
110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
"""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")
|
|
)
|
|
|
|
|
|
class TestRebootHost:
|
|
"""``reboot_host`` (napalm-device-types ``HostRebootMixin``): netOrk sent
|
|
``/sbin/reboot`` through the private ``_send_command`` and could not tell a
|
|
refusal from the session dropping (netOrk #1113)."""
|
|
|
|
def test_detached_so_the_answer_comes_back_first(self):
|
|
drv = _driver()
|
|
with patch("napalm_openwrt.openwrt.run_on_transport", return_value=CommandResult("", "", 0)) as run:
|
|
drv.reboot_host()
|
|
|
|
command = run.call_args.args[1]
|
|
assert "reboot" in command and command.rstrip().endswith("&'")
|
|
assert "trap" in command and "sleep" in command
|
|
|
|
def test_a_refusal_is_raised(self):
|
|
drv = _driver()
|
|
failed = CommandResult("", "reboot: Operation not permitted", 1)
|
|
with patch("napalm_openwrt.openwrt.run_on_transport", return_value=failed), pytest.raises(
|
|
RuntimeError, match="not permitted"
|
|
):
|
|
drv.reboot_host()
|
|
|
|
def test_a_login_other_than_root_is_refused(self):
|
|
drv = _driver(username="admin")
|
|
with patch("napalm_openwrt.openwrt.run_on_transport") as run, pytest.raises(RuntimeError):
|
|
drv.reboot_host()
|
|
|
|
run.assert_not_called()
|