Merge pull request 'fix: keep the sudo password out of files written as root (0.2.1)' (#15) from fix/privileged-stdin-8 into main

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.2.1] - 2026-10-16
+
+### Fixed
+- A sudo password no longer ends up in `snmpd.conf`. `run_command` sent the
+  password and the data on the same stdin, and a sudo that does not prompt
+  (NOPASSWD, a cached timestamp) left the password for `cat`, which wrote it
+  as the first line of a file then made world-readable (#8). `run_command` now
+  refuses stdin to a privileged command while a sudo password is set, and
+  `fix_snmp` stages the configuration without privilege before root installs
+  it, without following a symlink at the target.
+
 ## [0.2.0] - 2026-10-13
 
 ### Changed

napalm_bsd/base.py

@@ -44,6 +44,25 @@ _SNMPD_CONF = (
     "sysLocation    Managed by netOrk\n"
     "sysContact     netork@localhost\n"
 )
+#: Why ``run_command`` refuses stdin to a privileged command while a sudo
+#: password is set (#8).
+_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 and have root "
+    "install that file."
+)
+
+#: Stages content without privilege: a fresh ``mktemp`` file only the login user
+#: may read, never a fixed name another user could create first. Prints its path.
+_STAGE_TEMPLATE = "/tmp/napalm-bsd-stage.XXXXXX"
+_STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")]
+_STAGE = (
+    f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && "
+    '{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }'
+)
+
 #: What a working agent answers sysDescr.0 with.
 _SNMP_TYPES = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
 
@@ -155,10 +174,38 @@ class BsdDriver(HostStatusMixin, OSDriver):
         timeout: float = 60,
         stdin: Optional[bytes] = 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
-        return run_on_transport(self._transport(), line, stdin=data, timeout=timeout)
+        if prefix and stdin is not None:
+            raise ValueError(_PRIVILEGED_STDIN)
+        return run_on_transport(self._transport(), line, stdin=prefix or stdin, timeout=timeout)
+
+    def _install_file_as_root(self, path: str, content: str) -> CommandResult:
+        """Replace *path* with *content*, as a new file owned by root, mode 644.
+
+        The content is staged without privilege (``_STAGE``), so it never follows
+        a sudo password on stdin. Root then installs the staged file next to *path*
+        and renames it over *path*; nothing root runs follows a symlink at *path*.
+        Nothing runs as root when staging failed.
+        """
+        staged = self.run_command(_STAGE, timeout=30, stdin=content.encode())
+        lines = staged.stdout.strip().splitlines()
+        tmp = lines[-1] if lines else ""
+        if staged.exit_code != 0 or not tmp.startswith(_STAGE_PREFIX) or "/" in tmp[len(_STAGE_PREFIX):]:
+            reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}"
+            return CommandResult(reason, "", staged.exit_code or 1)
+        tmp_q, part_q, path_q = (quote(p) for p in (tmp, f"{path}.napalm-tmp", path))
+        directory = quote(path.rsplit("/", 1)[0] or "/")
+        return self.run_command(
+            f"mkdir -p {directory} && install -m 644 -o root -g wheel {tmp_q} {part_q}"
+            f" && mv -f {part_q} {path_q}; s=$?; rm -f {tmp_q} {part_q}; exit $s",
+            privileged=True,
+            timeout=30,
+        )
 
     def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
         line, prefix = self._privileged(command, privileged)
@@ -420,18 +467,14 @@ class BsdDriver(HostStatusMixin, OSDriver):
         (NetOrk/netork#800). Stops at the first step that fails.
         """
         lines: list[str] = []
-        conf_dir = self.SNMPD_CONF.rsplit("/", 1)[0]
         steps = (
-            ("install", self.INSTALL_COMMAND.format(name="net-snmp"), None),
-            (
-                "config",
-                f"mkdir -p {conf_dir} && cat > {self.SNMPD_CONF} && chmod 644 {self.SNMPD_CONF}",
-                _SNMPD_CONF.encode(),
-            ),
-            ("service", self.SNMPD_START, None),
+            ("install", lambda: self.run_command(
+                self.INSTALL_COMMAND.format(name="net-snmp"), privileged=True, timeout=600)),
+            ("config", lambda: self._install_file_as_root(self.SNMPD_CONF, _SNMPD_CONF)),
+            ("service", lambda: self.run_command(self.SNMPD_START, privileged=True, timeout=600)),
         )
-        for label, command, stdin in steps:
-            result = self.run_command(command, privileged=True, timeout=600, stdin=stdin)
+        for label, step in steps:
+            result = step()
             output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip())))
             lines.append(f"[{label}] {output[-300:]}".rstrip())
             if result.exit_code != 0:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-bsd"
-version = "0.2.0"
+version = "0.2.1"
 description = "NAPALM drivers for FreeBSD and OpenBSD hosts via SSH"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_drivers.py

@@ -9,6 +9,7 @@ from napalm.base import NetworkDriver
 from napalm_device_types import OSDriver
 
 from napalm_bsd import FreeBSDDriver, OpenBSDDriver
+from napalm_bsd import base
 from napalm_bsd.base import BsdDriver
 
 FIXTURES = Path(__file__).parent / "fixtures"
@@ -463,6 +464,7 @@ class TestSnmp:
     """net-snmp from packages, as decided in NetOrk/netork#800: UCD-SNMP-MIB,
     the same health metrics as on Linux."""
 
+    STAGED = "/tmp/napalm-bsd-stage.Ab12Cd"
     PROBE_OK = "STRING: FreeBSD host 15.1-RELEASE-p4 FreeBSD 15.1-RELEASE-p4 GENERIC amd64"
 
     @pytest.mark.parametrize(
@@ -483,13 +485,13 @@ class TestSnmp:
         ],
     )
     def test_fix_snmp_installs_configures_and_starts_net_snmp(self, cls, install, conf, start):
-        driver = _channel(cls, {BsdDriver.SNMP_PROBE: self.PROBE_OK})
-        stdin_seen = {}
+        driver = _channel(cls, {base._STAGE: self.STAGED, BsdDriver.SNMP_PROBE: self.PROBE_OK})
+        stdin_seen = []
         run = driver.run_command
 
         def recording(command, *, privileged=False, timeout=60, stdin=None):
             if stdin is not None:
-                stdin_seen[command] = stdin
+                stdin_seen.append((command, privileged, stdin))
             return run(command, privileged=privileged, timeout=timeout, stdin=stdin)
 
         driver.run_command = recording  # type: ignore[method-assign]
@@ -498,13 +500,22 @@ class TestSnmp:
         assert result["success"] is True
         commands = [c for c, _ in driver.calls]
         assert commands[0] == install
-        assert any(conf in c and "cat >" in c for c in commands)
+        installed = next(c for c, p in driver.calls if p and "install -m 644" in c)
+        assert conf in installed and self.STAGED in installed
         assert start in commands
         assert commands[-1] == BsdDriver.SNMP_PROBE
-        assert all(p for c, p in driver.calls if c != BsdDriver.SNMP_PROBE)  # root for the setup
-        written = next(v for k, v in stdin_seen.items() if conf in k).decode()
+        assert all(p for c, p in driver.calls if c not in (BsdDriver.SNMP_PROBE, base._STAGE))
+        # The configuration reaches the host only through the unprivileged stage (#8).
+        assert [(c, p) for c, p, _ in stdin_seen] == [(base._STAGE, False)]
+        written = stdin_seen[0][2].decode()
         assert "agentAddress udp:161" in written and "rocommunity public" in written
 
+    def test_a_failed_stage_runs_nothing_as_root(self):
+        driver = _channel(FreeBSDDriver, {base._STAGE: ("mktemp: permission denied", 1)})
+        result = driver.run_device_action("fix_snmp")
+        assert result["success"] is False
+        assert not any(p and "install -m 644" in c for c, p in driver.calls)
+
     def test_a_failed_install_stops_and_says_so(self):
         driver = _channel(
             FreeBSDDriver,
@@ -550,6 +561,43 @@ class TestSnmp:
         assert driver.get_snmp_config() is None
 
 
+class TestPrivilegedStdin:
+    """A privileged command never gets stdin while a sudo password is set (#8).
+
+    ``sudo -S`` reads the password from stdin; a sudo that does not prompt
+    (NOPASSWD, a cached timestamp) leaves it for the command, which would write
+    it into whatever the data goes to."""
+
+    def _driver(self, monkeypatch, *, sudo_password, root=False):
+        sent = []
+        monkeypatch.setattr(
+            base, "run_on_transport",
+            lambda transport, line, stdin=None, timeout=60: sent.append((line, stdin)) or _Result(),
+        )
+        args = {"sudo_password": sudo_password} if sudo_password else {}
+        driver = FreeBSDDriver("host", "netork", "", optional_args=args)
+        driver._transport = lambda: object()  # type: ignore[method-assign]
+        driver._root = root
+        return driver, sent
+
+    def test_refuses_stdin_to_a_privileged_command_with_a_sudo_password(self, monkeypatch):
+        driver, sent = self._driver(monkeypatch, sudo_password="s3cret")
+        with pytest.raises(ValueError, match="no stdin"):
+            driver.run_command("cat > /etc/x", privileged=True, stdin=b"data")
+        assert sent == []
+
+    def test_a_privileged_command_without_stdin_gets_only_the_password(self, monkeypatch):
+        driver, sent = self._driver(monkeypatch, sudo_password="s3cret")
+        driver.run_command("id -u", privileged=True)
+        assert sent == [("sudo -S -p '' sh -c 'id -u'", b"s3cret\n")]
+
+    @pytest.mark.parametrize("sudo_password, root", [(None, False), ("s3cret", True)])
+    def test_stdin_is_fine_where_no_password_goes_first(self, monkeypatch, sudo_password, root):
+        driver, sent = self._driver(monkeypatch, sudo_password=sudo_password, root=root)
+        driver.run_command("cat > /tmp/x", privileged=True, stdin=b"data")
+        assert sent[0][1] == b"data"
+
+
 class TestHostStatus:
     """napalm-device-types' host status, carried over the exec channel; on
     FreeBSD it compares the installed with the running kernel (4.1.0)."""