Christian Manivong
c00deef3d7
Backport of the 0.15.1 fix onto 0ba5081, the commit netOrk pins, which
cannot take 0.15.x before it moves to napalm-device-types 4.22.
systemd 258 (Ubuntu 26.04) ships /etc/profile.d/80-systemd-osc-context.sh:
an interactive bash wraps every command in OSC 3008 sequences from PS0 and
PROMPT_COMMAND, each with a fresh UUID. They ended up in every value read
through netmiko's shell, and because the prompt never looked the same twice,
the prompt line stayed in the output too. os_version grew past 255
characters and netOrk could not store the poll.
The session now unsets PS0 and PROMPT_COMMAND right after switching the
history off and reads its prompt anew; _send() strips terminal codes from
what it returns.
263 lines
8.7 KiB
Python
263 lines
8.7 KiB
Python
"""The sudo password never goes into a command line or a shell history (NetOrk/netork#1039).
|
|
|
|
``_sudo`` typed ``echo '<password>' | sudo -S -p "" <command>`` into netmiko's
|
|
interactive shell. That shell keeps a history, so every poll appended the
|
|
password to the login user's ``~/.bash_history`` -- 121 lines of it on one host
|
|
-- and while the command ran it was in the process list for every local user.
|
|
|
|
Privileged commands now run on an exec channel next to netmiko's PTY, the way
|
|
``run_command(privileged=True)`` does: the password reaches ``sudo -S`` on
|
|
stdin, and an exec channel has no history and no PTY to echo it back. The
|
|
interactive session switches its history off before anything else, in case
|
|
anything ever slips through again.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shlex
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from napalm_linux import LinuxDriver
|
|
from napalm_linux.linux import _HISTORY_OFF
|
|
|
|
_PW = "s3cr3t 'q' $x"
|
|
|
|
|
|
class Channel:
|
|
"""An exec channel answering with *output* and *status*; it refuses a line
|
|
that holds the password, which is what ``ps`` would show."""
|
|
|
|
def __init__(self, host: "Transport"):
|
|
self.host = host
|
|
self.command = ""
|
|
self.stdin = b""
|
|
self.timeout = None
|
|
self._out = b""
|
|
|
|
def exec_command(self, command):
|
|
assert _PW not in command, f"sudo password in a command line: {command!r}"
|
|
self.command = command
|
|
output, status = self.host.answer
|
|
self._out = output.encode()
|
|
self.status = status
|
|
|
|
def settimeout(self, timeout):
|
|
self.timeout = timeout
|
|
|
|
def sendall(self, data):
|
|
self.stdin += data
|
|
|
|
def shutdown_write(self):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
def recv_ready(self):
|
|
return bool(self._out)
|
|
|
|
def recv(self, size):
|
|
chunk, self._out = self._out[:size], self._out[size:]
|
|
return chunk
|
|
|
|
def recv_stderr_ready(self):
|
|
return False
|
|
|
|
def recv_stderr(self, size):
|
|
return b""
|
|
|
|
def exit_status_ready(self):
|
|
return not self._out
|
|
|
|
def recv_exit_status(self):
|
|
return self.status
|
|
|
|
|
|
class Transport:
|
|
def __init__(self, output="", status=0):
|
|
self.answer = (output, status)
|
|
self.channels: list[Channel] = []
|
|
|
|
def open_session(self):
|
|
self.channels.append(Channel(self))
|
|
return self.channels[-1]
|
|
|
|
|
|
def _driver(*, sudo_password=_PW, root=False, output="", status=0):
|
|
driver = LinuxDriver("h", "admin", "p", optional_args={"sudo_password": sudo_password})
|
|
transport = Transport(output, status)
|
|
driver._device = MagicMock()
|
|
driver._device.remote_conn_pre = SimpleNamespace(get_transport=lambda: transport)
|
|
driver._root = root
|
|
return driver, transport
|
|
|
|
|
|
def _script(command_line: str) -> str:
|
|
"""What ``sh -c`` runs in *command_line*, unquoted."""
|
|
parts = shlex.split(command_line)
|
|
return parts[parts.index("-c") + 1]
|
|
|
|
|
|
class TestSudo:
|
|
def test_the_password_goes_to_sudo_on_stdin(self):
|
|
driver, transport = _driver(output="ok\n")
|
|
|
|
assert driver._sudo("apt-get update 2>&1") == "ok"
|
|
|
|
channel = transport.channels[-1]
|
|
assert channel.command.startswith("sudo -S -p '' sh -c ")
|
|
assert channel.stdin == f"{_PW}\n".encode()
|
|
|
|
def test_nothing_goes_through_the_interactive_shell(self):
|
|
"""Its history is where the password ended up."""
|
|
driver, _ = _driver()
|
|
|
|
driver._sudo("apt-get update")
|
|
|
|
driver._device.send_command.assert_not_called()
|
|
|
|
def test_the_whole_command_runs_as_root(self):
|
|
"""``echo pw | sudo -S mv a b && chown …`` ran only ``mv`` as root."""
|
|
driver, transport = _driver()
|
|
|
|
driver._sudo("mv /tmp/x /etc/x && chown root:root /etc/x")
|
|
|
|
assert _script(transport.channels[-1].command).endswith(
|
|
"mv /tmp/x /etc/x && chown root:root /etc/x"
|
|
)
|
|
|
|
def test_a_sudo_that_does_not_prompt_leaves_the_password_to_nobody(self):
|
|
"""NOPASSWD or a cached timestamp leave the password on stdin; the
|
|
command's stdin is /dev/null, so nothing it runs reads it."""
|
|
driver, transport = _driver()
|
|
|
|
driver._sudo("dpkg --configure -a")
|
|
|
|
assert _script(transport.channels[-1].command).startswith("exec </dev/null; ")
|
|
|
|
def test_sudos_own_complaints_are_in_the_output(self):
|
|
"""``sudo: a password is required`` goes to stderr; callers read it from the output."""
|
|
driver, transport = _driver()
|
|
|
|
driver._sudo("true")
|
|
|
|
assert transport.channels[-1].command.endswith(" 2>&1")
|
|
|
|
def test_without_a_password_sudo_never_waits_for_one(self):
|
|
driver, transport = _driver(sudo_password=None)
|
|
|
|
driver._sudo("apt-get update")
|
|
|
|
assert transport.channels[-1].command.startswith("sudo -n sh -c ")
|
|
assert transport.channels[-1].stdin == b""
|
|
|
|
def test_a_root_login_needs_no_sudo(self):
|
|
driver, transport = _driver(root=True)
|
|
|
|
driver._sudo("apt-get update")
|
|
|
|
assert transport.channels[-1].command.startswith("sh -c ")
|
|
assert transport.channels[-1].stdin == b""
|
|
|
|
def test_the_read_timeout_is_the_channels(self):
|
|
driver, transport = _driver()
|
|
|
|
driver._sudo("apt-get upgrade -y", read_timeout=600)
|
|
|
|
assert transport.channels[-1].timeout == 600
|
|
|
|
|
|
class TestSudoStatus:
|
|
def test_the_exit_status_is_the_commands_own(self):
|
|
driver, _ = _driver(output="E: Unable to locate package nope\n", status=100)
|
|
|
|
assert driver._sudo_status("apt-get remove -y nope 2>&1") == (
|
|
"E: Unable to locate package nope",
|
|
100,
|
|
)
|
|
|
|
def test_a_refused_sudo_is_a_failure(self):
|
|
driver, _ = _driver(output="Sorry, try again.\n", status=1)
|
|
|
|
assert driver._sudo_status("apt-get remove -y x")[1] == 1
|
|
|
|
|
|
class TestRunPrivileged:
|
|
def test_with_a_password_it_takes_the_exec_channel(self):
|
|
driver, transport = _driver(output="__NETORK_RC=0\n")
|
|
|
|
driver._run_privileged("apt-get update -q 2>&1; echo __NETORK_RC=$?", 180)
|
|
|
|
driver._device.send_command.assert_not_called()
|
|
assert transport.channels[-1].stdin == f"{_PW}\n".encode()
|
|
|
|
def test_a_refresh_reads_its_status_from_the_exec_channel(self):
|
|
driver, _ = _driver(output="Hit:1 http://deb.debian.org\n__NETORK_RC=0\n")
|
|
driver._pkg_manager = "apt"
|
|
|
|
assert driver.refresh_available_updates()["success"] is True
|
|
|
|
|
|
class TestWireGuardWithSudo:
|
|
def test_the_dump_comes_straight_from_the_exec_channel(self):
|
|
"""No PTY to turn its tabs into spaces, so no file in /tmp holding the
|
|
interfaces' private keys."""
|
|
dump = "wg0\tPRIV\tPUB\t51820\toff\nwg0\tPEER\t(none)\t1.2.3.4:51820\t10.0.0.2/32\t0\t1\t2\toff\n"
|
|
driver, transport = _driver(output=dump)
|
|
|
|
tunnels = driver.get_vpn_tunnels()
|
|
|
|
assert list(tunnels) == ["wireguard-wg0-PEER"]
|
|
assert "/tmp" not in transport.channels[-1].command
|
|
driver._device.send_command.assert_not_called()
|
|
|
|
|
|
class TestHistoryOff:
|
|
"""Defense in depth: the interactive session keeps no history at all."""
|
|
|
|
def test_it_is_the_first_thing_the_session_sends(self):
|
|
driver = LinuxDriver("h", "admin", "p", optional_args={"sudo_password": _PW})
|
|
device = MagicMock()
|
|
device.send_command.return_value = "/usr/bin/apt"
|
|
with patch("napalm_linux.linux.ConnectHandler", return_value=device):
|
|
driver.open()
|
|
|
|
first = device.send_command.call_args_list[0].args[0]
|
|
assert first == _HISTORY_OFF
|
|
|
|
def test_it_leaves_no_history_file_and_records_nothing_more(self):
|
|
assert _HISTORY_OFF.startswith(" ") # HISTCONTROL=ignorespace skips the line itself
|
|
assert "unset HISTFILE" in _HISTORY_OFF
|
|
assert "set +o history" in _HISTORY_OFF
|
|
|
|
def test_a_shell_that_refuses_it_does_not_stop_the_connection(self):
|
|
driver = LinuxDriver("h", "admin", "p")
|
|
device = MagicMock()
|
|
# history off (refused), prompt hooks off (#41), stty, package manager
|
|
device.send_command.side_effect = [Exception("no prompt"), "", "", "/usr/bin/apt"]
|
|
with patch("napalm_linux.linux.ConnectHandler", return_value=device):
|
|
driver.open()
|
|
|
|
assert driver._device is device
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"method, args",
|
|
[
|
|
("install_package", ("curl",)),
|
|
("uninstall_package", ("curl",)),
|
|
("apply_updates", ([],)),
|
|
],
|
|
)
|
|
def test_package_work_never_types_the_password(method, args):
|
|
driver, transport = _driver(output="done\n")
|
|
driver._pkg_manager = "apt"
|
|
|
|
getattr(driver, method)(*args)
|
|
|
|
driver._device.send_command.assert_not_called()
|
|
assert transport.channels and all(c.stdin == f"{_PW}\n".encode() for c in transport.channels)
|