napalm_device_types/ssh_server.py

@@ -285,11 +285,28 @@ class OpenSshServerMixin:
         return result.stdout if result.exit_code == 0 else None
 
     def _ssh_write(self, path: str, content: str) -> None:
-        script = (
-            f'tmp={quote(path + ".netork-tmp")} && cat > "$tmp" && chmod 644 "$tmp"'
-            f' && mv -f -- "$tmp" {quote(path)} || {{ s=$?; rm -f -- "$tmp"; exit $s; }}'
+        """Replace *path* with *content*, as root, without root reading stdin.
+
+        A driver gains root with ``sudo -S`` and puts the password in front of
+        stdin. A sudo that does not prompt -- NOPASSWD, a cached timestamp --
+        leaves it there, and a ``cat`` under sudo writes it into the file. So the
+        content goes, without privilege, to a file only the login user can read;
+        root then installs that file (mode 644, owned by root) and removes it.
+        """
+        staged = self.run_command(
+            'umask 077 && t=$(mktemp /tmp/netork-ssh.XXXXXX) && cat > "$t" && printf %s "$t"',
+            privileged=False,
+            timeout=30,
+            stdin=content.encode(),
+        )
+        tmp = staged.stdout.strip()
+        if staged.exit_code != 0 or not tmp.startswith("/tmp/netork-ssh."):
+            raise RuntimeError(f"could not stage {path}: {(staged.stderr or staged.stdout).strip()}")
+        part = path + ".netork-tmp"
+        result = self._ssh_run(
+            f"install -m 644 -- {quote(tmp)} {quote(part)} && mv -f -- {quote(part)} {quote(path)};"
+            f" s=$?; rm -f -- {quote(tmp)} {quote(part)}; exit $s"
         )
-        result = self._ssh_run(script, stdin=content.encode())
         if result.exit_code != 0:
             raise RuntimeError(f"could not write {path}: {(result.stderr or result.stdout).strip()}")
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "4.5.0"
+version = "4.5.1"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_ssh_server.py

@@ -188,10 +188,29 @@ class FakeHost(OpenSshServerMixin):
                     eff[kind] = parts[1].split(",")
         return eff
 
+    #: What a real ``sudo -S`` leaves on stdin when it does not prompt (NOPASSWD,
+    #: or a cached timestamp): the password the driver put in front of the data.
+    SUDO_PASSWORD = "hunter2"
+
     def run_command(self, command: str, *, privileged: bool = False, timeout: float = 60,
                     stdin: Optional[bytes] = None) -> CommandResult:
         self.ran.append(command)
+        if privileged and stdin is not None:
+            stdin = f"{self.SUDO_PASSWORD}\n".encode() + stdin
+        m = re.search(r"mktemp", command)
+        if m:
+            assert not privileged, "the content must reach a file without sudo in the way"
+            path = f"/tmp/netork-ssh.{len(self.files)}"
+            self.files[path] = (stdin or b"").decode()
+            return CommandResult(path, "", 0)
         assert privileged, command
+        m = re.search(r"install -m 644 -- '?([^' ]+)'? '?([^' ]+)'?", command)
+        if m:
+            src, tmp = m.group(1), m.group(2)
+            target = re.search(r"mv -f -- '?[^' ]+'? '?([^' ;]+)'?", command).group(1)
+            self.files[target] = self.files.pop(src)
+            assert tmp.endswith(".netork-tmp")
+            return CommandResult("", "", 0)
         if command.startswith("sshd -T") or "sshd -T" in command:
             eff = self._effective()
             out = "".join(f"{d} {','.join(eff[k])}\n" for d, k in _DIRECTIVE.items())
@@ -204,10 +223,6 @@ class FakeHost(OpenSshServerMixin):
             if path in self.files:
                 return CommandResult(self.files[path], "", 0)
             return CommandResult("", "No such file", 1)
-        m = re.search(r"mv -f -- \"\$tmp\" '?([^' ]+)'?", command)
-        if m:
-            self.files[m.group(1)] = (stdin or b"").decode()
-            return CommandResult("", "", 0)
         m = re.match(r"rm -f -- '?([^' ]+)'?$", command)
         if m:
             self.files.pop(m.group(1), None)
@@ -301,3 +316,22 @@ def test_main_config_second_fix_keeps_what_the_first_set():
     eff = host._effective()
     assert "ecdsa-sha2-nistp256" not in eff["hostkey"]
     assert "hmac-sha1" not in eff["mac"]
+
+
+def test_the_sudo_password_never_reaches_a_file():
+    """napalm-linux puts the sudo password in front of stdin for ``sudo -S``. A
+    sudo that does not prompt leaves it there, and a ``cat`` under sudo wrote it
+    into the drop-in (NetOrk/netork#1027, first run on a real host). The content
+    now reaches a file of the login user first; root only installs that file."""
+    host = FakeHost()
+    host.harden_ssh_server({"mac": ["hmac-sha1"]})
+    assert host.SUDO_PASSWORD not in host.files[DROPIN_PATH]
+    assert host.files[DROPIN_PATH].startswith("# Written by netOrk")
+    assert not [p for p in host.files if p.startswith("/tmp/")]
+
+
+def test_restoring_an_earlier_dropin_leaks_nothing_either():
+    host = FakeHost()
+    host.files[DROPIN_PATH] = "# netOrk\nHostKeyAlgorithms ssh-ed25519\n"
+    host.harden_ssh_server({"mac": ["hmac-sha1"]}, verify=lambda: False)
+    assert host.files[DROPIN_PATH] == "# netOrk\nHostKeyAlgorithms ssh-ed25519\n"