From 647e058b8a96590597c400593e132cd079dcdab3 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Mon, 12 Oct 2026 20:41:06 +0200 Subject: [PATCH] fix: refuse stdin to a privileged command while a sudo password is set run_command(privileged=True, stdin=data) sent `sudo -S` with stdin = \n + data. A sudo that does not prompt (NOPASSWD, a cached timestamp) leaves the password line on stdin, and the command reads it as the first line of the data - `cat > file` wrote it into the file (NetOrk/netork#1029). run_command now raises ValueError for that combination before opening a channel. Stdin still reaches the command under `sudo -n` (no password set), for a root login, and for unprivileged commands. Callers stage the data without privilege and have root install the file. open_stream documents the same caveat. --- CHANGELOG.md | 10 +++++++++ napalm_linux/linux.py | 28 +++++++++++++++++++++--- tests/test_command_channel.py | 41 +++++++++++++++++++++++++++++++---- 3 files changed, 72 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b7833bf..1d28d49 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Security + +- `run_command(privileged=True, stdin=...)` raises `ValueError` while a sudo + password is set. `sudo -S` reads the password from stdin, and a sudo that does + not prompt (NOPASSWD, a cached timestamp) left it for the command, which read + it as the first line of the data: netOrk wrote it into a file that way + (NetOrk/netork#1029). Stage the data without privilege and have root install + that file. Without a sudo password (`sudo -n`) or for a root login, stdin + still goes to the command. + ## [0.4.0] – 2026-10-08 ### Added diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 43e8551..31d17f0 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -132,6 +132,16 @@ _YUM_SECURITY_COMMAND = "LC_ALL=C yum updateinfo list security -q 2>/dev/null" #: cannot take the restart with it. _REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; /sbin/reboot) /dev/null 2>&1 &'" +#: Why ``run_command`` refuses stdin to a privileged command while a sudo +#: password is set (NetOrk/netork#1029). +_PRIVILEGED_STDIN = ( + "run_command(privileged=True) takes no stdin while a sudo password is set: " + "sudo -S reads the password from stdin, and a sudo that does not prompt " + "(NOPASSWD, a cached timestamp) leaves it for the command, which would read it " + "as the first line of the data. Stage the data without privilege (umask 077, " + "mktemp, cat) and have root install that file." +) + #: What to do when sudo wants a password netOrk does not have. _SUDO_PASSWORD_HINT = ( "sudo requires a password on this device but none is configured in netOrk. " @@ -428,13 +438,25 @@ class LinuxDriver( timeout: float = 60, stdin: bytes | None = None, ) -> CommandResult: - """Run *command* on an exec channel: no PTY, stderr apart, a real exit code.""" + """Run *command* on an exec channel: no PTY, stderr apart, a real exit code. + + :raises ValueError: for *stdin* to a privileged command while a sudo + password is set (see ``_PRIVILEGED_STDIN``). + """ line, prefix = self._privileged(command, privileged) - data = (prefix or b"") + (stdin or b"") if (prefix or stdin) else None + if prefix and stdin is not None: + raise ValueError(_PRIVILEGED_STDIN) + data = prefix or stdin or None return run_on_transport(self._transport(), line, stdin=data, timeout=timeout) def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream: - """Start *command* on an exec channel and return a stream to it.""" + """Start *command* on an exec channel and return a stream to it. + + A privileged stream with a sudo password starts with that password, for + ``sudo -S``; a sudo that does not prompt leaves it for *command*. Write + nothing to such a stream that must not follow the password, as + ``_PRIVILEGED_STDIN`` says for ``run_command``. + """ line, prefix = self._privileged(command, privileged) return open_stream_on_transport(self._transport(), line, stdin_prefix=prefix) diff --git a/tests/test_command_channel.py b/tests/test_command_channel.py index c59b822..6f421ca 100644 --- a/tests/test_command_channel.py +++ b/tests/test_command_channel.py @@ -107,21 +107,54 @@ def test_a_root_login_needs_no_sudo(): assert transport.channels[-1].sent == b"" -def test_stdin_follows_the_sudo_password(): +def test_a_privileged_command_with_a_sudo_password_takes_no_stdin(): + """sudo -S reads the password from stdin. A sudo that does not prompt -- + NOPASSWD, a cached timestamp -- leaves it there, and the command reads it as + its first line: it wrote the password into an sshd drop-in (NetOrk/netork#1029).""" driver, transport = _driver(sudo_password="pw") + with pytest.raises(ValueError, match="stdin"): + driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5) + + assert transport.channels == [] # nothing reached the host + + +def test_privileged_stdin_without_a_sudo_password_goes_to_the_command(): + """sudo -n reads nothing from stdin: the data is the command's alone.""" + driver, transport = _driver() + driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5) - assert transport.channels[-1].sent == b"pw\ndata" + channel = transport.channels[-1] + assert channel.command == "sudo -n sh -c 'tee /etc/x'" + assert channel.sent == b"data" + + +def test_privileged_stdin_for_a_root_login_goes_to_the_command(): + driver, transport = _driver(root=True, sudo_password="pw") + + driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5) + + assert transport.channels[-1].command == "tee /etc/x" + assert transport.channels[-1].sent == b"data" + + +def test_unprivileged_stdin_with_a_sudo_password_goes_to_the_command(): + driver, transport = _driver(sudo_password="pw") + + driver.run_command("cat > /tmp/x", stdin=b"data", timeout=5) + + assert transport.channels[-1].command == "cat > /tmp/x" + assert transport.channels[-1].sent == b"data" def test_a_privileged_stream_gets_the_password_first(): driver, transport = _driver(sudo_password="pw") - driver.open_stream("cat > /tmp/x", privileged=True) + driver.open_stream("journalctl -f", privileged=True) channel = transport.channels[-1] - assert channel.command == "sudo -S -p '' sh -c 'cat > /tmp/x'" + assert channel.command == "sudo -S -p '' sh -c 'journalctl -f'" assert channel.sent == b"pw\n"