From 707e0487927b4820f0cd13376beb93b47ffeaf76 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 16 Oct 2026 11:48:25 +0200 Subject: [PATCH] fix: keep the sudo password out of files written as root (0.2.1) run_command prepended the sudo password to the caller's stdin. A sudo that does not prompt left it for the command, and fix_snmp's 'cat > snmpd.conf' wrote it as the configuration's first line before making the file world-readable. run_command now refuses stdin to a privileged command while a sudo password is set. fix_snmp stages the configuration in a mktemp file without privilege, and root installs it next to the target and renames it over, so nothing root runs follows a symlink. Closes #8 --- CHANGELOG.md | 11 +++++++ napalm_bsd/base.py | 69 +++++++++++++++++++++++++++++++++++-------- pyproject.toml | 2 +- tests/test_drivers.py | 60 +++++++++++++++++++++++++++++++++---- 4 files changed, 122 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d70753..40f9182 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/napalm_bsd/base.py b/napalm_bsd/base.py index 65a5acc..5c0edd0 100644 --- a/napalm_bsd/base.py +++ b/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: diff --git a/pyproject.toml b/pyproject.toml index 7a9a632..6d60c8c 100644 --- a/pyproject.toml +++ b/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" diff --git a/tests/test_drivers.py b/tests/test_drivers.py index 0cf7eb3..1eb5615 100644 --- a/tests/test_drivers.py +++ b/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).""" -- 2.54.0