Merge pull request 'fix: refuse stdin to a privileged command while a sudo password is set' (#24) from fix/privileged-stdin-1029 into master

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

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 >/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)
 

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"