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_<hash>`; the dump comes straight from the exec channel.
 
 ## [0.4.0] – 2026-10-08
 

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 </dev/null; "
 
+#: Where ``_install_file_as_root`` stages content: a fresh file from ``mktemp``
+#: under ``umask 077``, never a fixed name another user could create first.
+_STAGE_TEMPLATE = "/tmp/netork-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 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:
+        # ── 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_<hash> before: a predictable name,
+        # holding the interfaces' private keys (#1039).
         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()
 
         _perm_errors = (
             "operation not permitted", "permission denied",

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_<hash>``: 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 ")