From 108f6fb3ca60af2b3fa54b54caaab39947282041 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Mon, 12 Oct 2026 22:56:06 +0200 Subject: [PATCH] fix: stage root writes in a mktemp file, never at a fixed /tmp path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Now that a whole _sudo line runs as root, `mv /tmp/netork_snmpd.conf /etc/snmp/snmpd.conf && chown root:root … && chmod 644 …` would have let a local user plant a symlink at the fixed /tmp path and have root's chown and chmod follow it (e.g. to /etc/shadow). The same held for 00proxy. _install_file_as_root stages the content unprivileged in a fresh mktemp file (umask 077), checks the path it got back, and root installs it as a new regular file next to the target and renames it over: nothing root runs follows a symlink or changes the owner or mode of an existing path. Nothing runs as root when staging failed. get_vpn_tunnels always reads `wg show all dump` from the exec channel; the predictable /tmp/.netork_wg_ file holding the private keys is gone. --- CHANGELOG.md | 11 ++ napalm_linux/linux.py | 78 ++++++---- tests/test_staged_root_writes.py | 246 +++++++++++++++++++++++++++++++ 3 files changed, 307 insertions(+), 28 deletions(-) create mode 100644 tests/test_staged_root_writes.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 63d569a..8fd41df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 login user's `~/.bash_history` on your hosts and consider changing the password. `_sudo_status` takes the exit status from the channel; it is `None` only when the channel closed without one. +- The snmpd and apt proxy fixes no longer stage their files at a fixed path in + /tmp (`/tmp/netork_snmpd.conf`, `/tmp/netork_00proxy`) and then `mv`, + `chown` and `chmod` them as root. Another local user could create that path + first as a symlink, for example to `/etc/shadow`, and root's `chown`/`chmod` + followed it. The content now goes to a fresh `mktemp` file under `umask 077`, + written without privilege, and root `install`s it as a new file owned by root + (mode 644) next to the target and renames it into place + (`_install_file_as_root`). Nothing runs as root when staging failed. +- `get_vpn_tunnels()` without a sudo password no longer writes `wg show all dump` + (which includes the interfaces' private keys) to the predictable + `/tmp/.netork_wg_`; the dump comes straight from the exec channel. ## [0.4.0] – 2026-10-08 diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 099f6f6..c35f79b 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -152,6 +152,15 @@ _HISTORY_OFF = " unset HISTFILE; set +o history 2>/dev/null; true" #: a cached timestamp) leaves the password on stdin, and nothing may read it. _NO_STDIN = "exec "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }' +) + #: 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. " @@ -385,6 +394,29 @@ class LinuxDriver( line, prefix = self._as_root(f"{_NO_STDIN}{command}") return run_on_transport(self._transport(), f"{line} 2>&1", stdin=prefix, 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 in a fresh ``mktemp`` file only + the login user may read; its stdin is the content's, not a privileged + command's (``_PRIVILEGED_STDIN``). Root then ``install``s that file as a + new regular file next to *path* and renames it over *path*: nothing root + runs follows a symlink or changes the owner or mode of an existing path + (NetOrk/netork#1039). 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 = (_shlex_quote(p) for p in (tmp, f"{path}.netork-tmp", path)) + return self._sudo_result( + f"install -m 644 -o root -g root -- {tmp_q} {part_q} && mv -f -- {part_q} {path_q};" + f" s=$?; rm -f -- {tmp_q} {part_q}; exit $s", + 30, + ) + def _sudo_status(self, command: str, read_timeout: float = 100) -> tuple[str, int | None]: """Run *command* as root and return ``(output, exit_status)``. @@ -1988,25 +2020,24 @@ class LinuxDriver( except Exception: pass - # Write snmpd.conf: - # 1. Write to /tmp (no sudo needed, avoids stdin conflict with sudo -S) - # 2. sudo mv to /etc/snmp/snmpd.conf + # Write snmpd.conf: staged unprivileged, installed by root (#1039). # agentAddress udp:161 overrides Debian's localhost-only default. - import base64 as _b64 conf_str = ( "agentAddress udp:161\n" "rocommunity public\n" "sysLocation Managed by netOrk\n" "sysContact netork@localhost\n" ) - conf_b64 = _b64.b64encode(conf_str.encode()).decode() - self._send(f"echo {conf_b64} | base64 -d > /tmp/netork_snmpd.conf") - self._sudo("mv /tmp/netork_snmpd.conf /etc/snmp/snmpd.conf && chown root:root /etc/snmp/snmpd.conf && chmod 644 /etc/snmp/snmpd.conf") + try: + written = self._install_file_as_root("/etc/snmp/snmpd.conf", conf_str) + install_error = written.stdout.strip() if written.exit_code else "" + except Exception as exc: + install_error = str(exc) verify = self._send("cat /etc/snmp/snmpd.conf 2>/dev/null").strip() if "agentAddress" in verify and "rocommunity" in verify: lines.append("[config] Wrote /etc/snmp/snmpd.conf — agentAddress udp:161, rocommunity public.") else: - lines.append(f"[warn] snmpd.conf write may have failed: {verify[:100]}") + lines.append(f"[warn] snmpd.conf write may have failed: {(install_error or verify)[:100]}") # 3. Open firewall for SNMP (UDP 161) — restrict to netOrk's source IP if netork_ip: @@ -2075,7 +2106,6 @@ class LinuxDriver( def _action_fix_apt_proxy(self) -> DeviceActionResultDict: """Write /etc/apt/apt.conf.d/00proxy with the configured proxy URL.""" - import base64 as _b64 proxy_url = self._apt_proxy_url if not proxy_url: return {"success": False, "output": "No apt_proxy_url configured."} @@ -2083,18 +2113,16 @@ class LinuxDriver( return {"success": False, "output": f"Package manager is {self._pkg_manager!r}, not apt — skipping."} content = f'Acquire::http::Proxy "{proxy_url}";\n' - content_b64 = _b64.b64encode(content.encode()).decode() - self._send(f"echo {content_b64} | base64 -d > /tmp/netork_00proxy") - self._sudo( - "mv /tmp/netork_00proxy /etc/apt/apt.conf.d/00proxy && " - "chown root:root /etc/apt/apt.conf.d/00proxy && " - "chmod 644 /etc/apt/apt.conf.d/00proxy" - ) + # Staged unprivileged, installed by root (#1039). + written = self._install_file_as_root("/etc/apt/apt.conf.d/00proxy", content) verify = self._send("cat /etc/apt/apt.conf.d/00proxy 2>/dev/null").strip() success = proxy_url in verify if success: return {"success": True, "output": f"Wrote /etc/apt/apt.conf.d/00proxy — proxy: {proxy_url}"} - return {"success": False, "output": f"Write may have failed. File content: {verify[:200]}"} + output = f"Write may have failed. File content: {verify[:200]}" + if written.exit_code: + output = f"{output}\n{written.stdout.strip()}" + return {"success": False, "output": output} def get_vpn_tunnels(self) -> dict[str, Any]: """Return WireGuard status via ``wg show all dump`` (requires root/sudo). @@ -2109,17 +2137,11 @@ class LinuxDriver( tunnels: dict[str, Any] = {} - # ── Attempt 1: wg show all dump via sudo ───────────────────────────── - # With a sudo password it runs on the exec channel, which keeps the - # tabs. The PTY expands them to spaces, breaking split("\t"), so - # without one the output goes through a temp file. - if self._sudo_password: - raw = self._sudo("wg show all dump 2>&1 || true") - else: - import hashlib as _hashlib - _tmp = "/tmp/.netork_wg_" + _hashlib.md5(self.hostname.encode()).hexdigest()[:8] - self._send(f"sudo -n wg show all dump > {_tmp} 2>&1 || true") - raw = self._send(f"cat {_tmp} 2>/dev/null || true; rm -f {_tmp}").strip() + # ── Attempt 1: wg show all dump as root ────────────────────────────── + # On the exec channel, which keeps the tabs the PTY expands to spaces. + # It went through /tmp/.netork_wg_ before: a predictable name, + # holding the interfaces' private keys (#1039). + raw = self._sudo("wg show all dump 2>&1 || true") _perm_errors = ( "operation not permitted", "permission denied", diff --git a/tests/test_staged_root_writes.py b/tests/test_staged_root_writes.py new file mode 100644 index 0000000..9e4f395 --- /dev/null +++ b/tests/test_staged_root_writes.py @@ -0,0 +1,246 @@ +"""Files root writes are staged in a fresh file, never at a fixed path in /tmp (NetOrk/netork#1039). + +The snmpd and apt proxy fixes wrote their content to ``/tmp/netork_snmpd.conf`` +or ``/tmp/netork_00proxy`` as the login user, then ran ``mv … && chown root:root +… && chmod 644 …``. Once the whole of a ``_sudo`` command ran as root, that was +a local privilege escalation: another user creates the fixed path first as a +symlink to ``/etc/shadow``, the login user's write fails (protected_symlinks), +root moves the symlink into place, and root's ``chown``/``chmod`` follow it -- +``/etc/shadow`` mode 644. + +Now the content goes to a ``mktemp`` file under ``umask 077``, unprivileged, and +root ``install``s that file as a new regular file next to the target, then +renames it over the target. Nothing root runs changes the owner or mode of a +path that could be a symlink. +""" + +from __future__ import annotations + +import re +import shlex +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from napalm_linux import LinuxDriver + +_PW = "pw-1039" +_STAGED = "/tmp/netork-stage.Ab12Cd" +#: What another local user planted where the fixes used to stage their content. +_PLANTED = { + "/tmp/netork_snmpd.conf": "/etc/shadow", + "/tmp/netork_00proxy": "/etc/shadow", +} + + +class Host: + """The host's exec channels: every line, its stdin, and whether it ran as root.""" + + def __init__(self, *, stage_output=_STAGED, stage_status=0): + self.runs: list[SimpleNamespace] = [] + self._stage = (stage_output, stage_status) + + def open_session(self): + return _Channel(self) + + def answer(self, command: str) -> tuple[str, int]: + if "mktemp" in command: + return self._stage + return "", 0 + + @property + def root_scripts(self) -> list[str]: + return [run.script for run in self.runs if run.root] + + @property + def unprivileged(self) -> list[SimpleNamespace]: + return [run for run in self.runs if not run.root] + + +class _Channel: + def __init__(self, host: Host): + self._host = host + self._run: SimpleNamespace | None = None + self._out = b"" + + def exec_command(self, command): + assert _PW not in command + root = command.startswith(("sudo ", "sh -c ")) + script = shlex.split(command)[shlex.split(command).index("-c") + 1] if root else command + self._run = SimpleNamespace(line=command, script=script, root=root, stdin=b"") + self._host.runs.append(self._run) + output, self._status = self._host.answer(command) + self._out = output.encode() + + def settimeout(self, timeout): + pass + + def sendall(self, data): + self._run.stdin += data + + def shutdown_write(self): + pass + + def close(self): + pass + + def recv_ready(self): + return bool(self._out) + + def recv(self, size): + chunk, self._out = self._out[:size], self._out[size:] + return chunk + + def recv_stderr_ready(self): + return False + + def exit_status_ready(self): + return not self._out + + def recv_exit_status(self): + return self._status + + +def _shell(command: str, **_kwargs) -> str: + """The interactive shell, for what the fixes still ask it.""" + if command.startswith("command -v apt"): + return "/usr/bin/apt" + if command.startswith("cat /etc/snmp/snmpd.conf"): + return "agentAddress udp:161\nrocommunity public\n" + if command.startswith("cat /etc/apt/apt.conf.d/00proxy"): + return 'Acquire::http::Proxy "http://proxy:3142";' + if command.startswith("snmpget"): + return "STRING: Linux" + return "" + + +def _driver(host: Host) -> LinuxDriver: + driver = LinuxDriver( + "h", + "admin", + "p", + optional_args={"sudo_password": _PW, "apt_proxy_url": "http://proxy:3142"}, + ) + driver._device = MagicMock() + driver._device.send_command.side_effect = _shell + driver._device.remote_conn_pre = SimpleNamespace(get_transport=lambda: host) + driver._root = False + driver._pkg_manager = "apt" + return driver + + +def _every_line(driver: LinuxDriver, host: Host) -> list[str]: + typed = [c.args[0] for c in driver._device.send_command.call_args_list] + return typed + [run.line for run in host.runs] + + +ACTIONS = [ + ("_action_fix_snmp", "/etc/snmp/snmpd.conf", b"agentAddress udp:161"), + ( + "_action_fix_apt_proxy", + "/etc/apt/apt.conf.d/00proxy", + b'Acquire::http::Proxy "http://proxy:3142";', + ), +] + + +@pytest.fixture(autouse=True) +def _no_waiting(): + with patch("time.sleep"): + yield + + +@pytest.mark.parametrize("action, target, content", ACTIONS) +class TestStagedWrite: + def test_the_old_fixed_path_is_never_touched(self, action, target, content): + host = Host() + driver = _driver(host) + + getattr(driver, action)() + + for planted in _PLANTED: + assert not any(planted in line for line in _every_line(driver, host)) + + def test_the_content_is_staged_unprivileged_in_a_fresh_private_file( + self, action, target, content + ): + host = Host() + driver = _driver(host) + + getattr(driver, action)() + + stage = next(run for run in host.unprivileged if "mktemp" in run.line) + assert "umask 077" in stage.line + assert "mktemp /tmp/netork-stage.XXXXXX" in stage.line + assert content in stage.stdin + + def test_root_installs_a_new_file_and_renames_it_over_the_target(self, action, target, content): + host = Host() + driver = _driver(host) + + getattr(driver, action)() + + install = next(script for script in host.root_scripts if "install -m" in script) + part = f"{target}.netork-tmp" + assert f"install -m 644 -o root -g root -- {_STAGED} {part}" in install + assert f"mv -f -- {part} {target}" in install + assert f"rm -f -- {_STAGED} {part}" in install + + def test_nothing_root_runs_changes_owner_or_mode_of_a_path(self, action, target, content): + host = Host() + driver = _driver(host) + + getattr(driver, action)() + + assert not any(re.search(r"\bch(own|mod)\b", script) for script in host.root_scripts) + + def test_nothing_runs_as_root_when_staging_failed(self, action, target, content): + host = Host(stage_output="mktemp: No space left on device\n", stage_status=1) + driver = _driver(host) + + getattr(driver, action)() + + assert not any("install -m" in script for script in host.root_scripts) + + def test_a_staged_path_it_did_not_ask_for_is_refused(self, action, target, content): + host = Host(stage_output="/etc/shadow") + driver = _driver(host) + + getattr(driver, action)() + + assert not any("/etc/shadow" in script for script in host.root_scripts) + + +def test_a_failed_install_is_reported(): + host = Host() + host.answer = lambda command: ( # type: ignore[method-assign] + (_STAGED, 0) if "mktemp" in command else ("install: cannot create\n", 1) + ) + driver = _driver(host) + driver._device.send_command.side_effect = lambda command, **_k: "" + + result = driver._action_fix_apt_proxy() + + assert result["success"] is False + assert "install: cannot create" in result["output"] + + +class TestWireGuardWithoutASudoPassword: + """It went through ``/tmp/.netork_wg_``: a name anyone can predict, and + a file holding the interfaces' private keys.""" + + def test_no_temp_file_and_the_dump_comes_from_the_exec_channel(self): + host = Host() + dump = ( + "wg0\tPRIV\tPUB\t51820\toff\nwg0\tPEER\t(none)\t1.2.3.4:1\t10.0.0.2/32\t0\t1\t2\toff\n" + ) + host.answer = lambda command: (dump, 0) # type: ignore[method-assign] + driver = _driver(host) + driver._sudo_password = None + + tunnels = driver.get_vpn_tunnels() + + assert list(tunnels) == ["wireguard-wg0-PEER"] + assert not any("/tmp" in line for line in _every_line(driver, host)) + assert host.runs[0].line.startswith("sudo -n sh -c ")