diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 7460dc6..a996405 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -50,6 +50,12 @@ logger = logging.getLogger("napalm_linux") # Package managers in detection order _PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"] +#: Printed after a command by ``_sudo_status`` so its exit status survives the +#: trip through an interactive shell. Matched only on a line of its own with a +#: number after it — an echoed command line carries the literal ``$?`` instead. +_RC_MARKER = "__NETORK_RC=" +_RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) + # DMI field values that carry no useful information (OEM defaults, blanks) _BAD_DMI: frozenset[str] = frozenset({ "", "none", "n/a", "not specified", "not applicable", @@ -267,6 +273,25 @@ class LinuxDriver(OSDriver): return self._send(wrapped, read_timeout=read_timeout) return self._send(f'sudo {command}', read_timeout=read_timeout) + def _sudo_status(self, command: str, read_timeout: float = 100) -> tuple[str, int | None]: + """Run *command* via sudo and return ``(output, exit_status)``. + + ``_sudo`` callers append ``|| true`` so a failing command yields output + instead of an error, which throws the exit status away. This variant + echoes ``$?`` straight after the sudo pipeline instead — sudo passes + the command's status through, and a failed password is non-zero too. + + The status is ``None`` when the marker never arrived (output cut short), + so a caller can tell "unknown" from "succeeded". + """ + raw = self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout) + matches = list(_RC_MARKER_RE.finditer(raw)) + if not matches: + return raw, None + last = matches[-1] + output = (raw[: last.start()] + raw[last.end():]).strip() + return output, int(last.group(1)) + def _detect_pkg_manager(self) -> str | None: """Return the first package manager binary found on PATH.""" for pm in _PKG_MANAGERS: @@ -1046,6 +1071,7 @@ class LinuxDriver(OSDriver): return {"success": success, "output": raw.strip()} #: Words in a package manager's output that mean it did not do the job. + #: Only consulted when the exit status is unknown; see ``_uninstall_failed``. _UNINSTALL_FAILED = ("error:", "failed", "not found", "is not installed", "no packages") def uninstall_package(self, name: str, purge: bool = False) -> dict[str, Any]: @@ -1077,42 +1103,49 @@ class LinuxDriver(OSDriver): pm = self._pkg_manager if pm == "apt": action = "purge" if purge else "remove" - raw = self._sudo( - f"DEBIAN_FRONTEND=noninteractive apt-get {action} -y {safe} 2>&1 || true" - ) + cmd = f"DEBIAN_FRONTEND=noninteractive apt-get {action} -y {safe} 2>&1" elif pm in ("dnf", "yum"): - raw = self._sudo(f"{pm} remove -y {safe} 2>&1 || true") + cmd = f"{pm} remove -y {safe} 2>&1" elif pm == "apk": # apk and pacman have no separate purge; asking for one is not an # error, it simply has nothing extra to do. - raw = self._sudo(f"apk del {safe} 2>&1 || true") + cmd = f"apk del {safe} 2>&1" elif pm == "pacman": - raw = self._sudo(f"pacman -R --noconfirm {safe} 2>&1 || true") + cmd = f"pacman -R --noconfirm {safe} 2>&1" else: return {"success": False, "output": f"Unsupported package manager: {pm}"} - if self._uninstall_failed(raw) and pm == "apt": - forced = self._sudo(f"dpkg --purge --force-all {safe} 2>&1 || true") - if not self._uninstall_failed(forced): - return { - "success": True, - "output": f"{raw.strip()}\n--- dpkg --purge --force-all ---\n{forced.strip()}", - } + raw, rc = self._sudo_status(cmd) + failed = self._uninstall_failed(raw, rc) + + if failed and pm == "apt": + forced, forced_rc = self._sudo_status(f"dpkg --purge --force-all {safe} 2>&1") raw = f"{raw.strip()}\n--- dpkg --purge --force-all ---\n{forced.strip()}" + failed = self._uninstall_failed(forced, forced_rc) - return {"success": not self._uninstall_failed(raw), "output": raw.strip()} + return {"success": not failed, "output": raw.strip()} - def _uninstall_failed(self, output: str) -> bool: - """Whether the package manager said it did not do the job. + def _uninstall_failed(self, output: str, rc: int | None = None) -> bool: + """Whether the package manager did not do the job. - apt prefixes its own errors with ``E: `` at the start of a line, and - the commonest of them — ``E: Sub-process /usr/bin/dpkg returned an - error code (1)`` — contains neither "error:" nor "failed". The keyword - list alone therefore read a failed removal as a success, which is the - worst direction for this particular answer to be wrong in. + The exit status decides whenever there is one (netork#267): it is the + answer the package manager actually gives, where the output is prose + that every tool phrases differently. A prerm printing "Failed to stop + …" while the removal completes is a success; a non-zero exit with + nothing alarming in the output is not. + + Only when the status is unknown (``rc is None``) is the output read, + as the best answer left. apt prefixes its own errors with ``E: `` at + the start of a line, and the commonest of them — ``E: Sub-process + /usr/bin/dpkg returned an error code (1)`` — contains neither "error:" + nor "failed". The keyword list alone therefore read a failed removal + as a success, which is the worst direction for this particular answer + to be wrong in. Matched at line start rather than anywhere: "note: " ends in "e: ". """ + if rc is not None: + return rc != 0 low = output.lower() if any(line.lstrip().startswith("e: ") for line in low.splitlines()): return True diff --git a/tests/test_linux.py b/tests/test_linux.py index 898aa6f..8559323 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -984,3 +984,164 @@ class TestUninstallPackage: assert result["success"] is True assert "apk del" in driver._device.send_command.call_args[0][0] + + +# --------------------------------------------------------------------------- +# uninstall_package – success from the exit status, not from prose (netork#267) +# --------------------------------------------------------------------------- + + +def _with_rc(output: str, rc: int) -> str: + """What the shell prints for a command run through ``_sudo_status``.""" + return f"{output}\n__NETORK_RC={rc}" + + +class TestSudoStatus: + """``_sudo_status`` keeps the exit status that ``|| true`` throws away.""" + + def test_returns_output_and_exit_status(self, driver): + _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0)) + + assert driver._sudo_status("apt-get remove -y wazuh-agent") == ( + "Removing wazuh-agent ...", + 0, + ) + + def test_a_non_zero_exit_status_is_reported(self, driver): + _mock_send(driver, _with_rc("E: Unable to locate package nope", 100)) + + assert driver._sudo_status("apt-get remove -y nope")[1] == 100 + + def test_the_status_is_read_right_after_sudo_returns(self, driver): + """``$?`` must be read straight after the sudo pipeline — with an + ``|| true`` in between, every command would report 0.""" + driver._sudo_password = "pw" # noqa: S105 + _mock_send(driver, _with_rc("", 0)) + + driver._sudo_status("apt-get remove -y x 2>&1") + + sent = driver._device.send_command.call_args[0][0] + assert sent.startswith("echo pw | sudo -S") + assert sent.endswith("apt-get remove -y x 2>&1; echo __NETORK_RC=$?") + assert "|| true" not in sent + + def test_a_missing_marker_means_unknown_not_success(self, driver): + """Output cut short before the marker arrived says nothing about the + exit status; ``None`` says so instead of guessing 0.""" + _mock_send(driver, "Removing wazuh-agent ...") + + assert driver._sudo_status("apt-get remove -y wazuh-agent") == ( + "Removing wazuh-agent ...", + None, + ) + + def test_the_command_echo_is_not_mistaken_for_the_marker(self, driver): + """A terminal may echo the command line back; its literal ``$?`` is not + a number, and only the marker on a line of its own counts.""" + _mock_send( + driver, + "sudo apt-get remove -y x; echo __NETORK_RC=$?\nRemoving x ...\n__NETORK_RC=1", + ) + + output, rc = driver._sudo_status("apt-get remove -y x") + + assert rc == 1 + assert "__NETORK_RC=1" not in output + + +class TestUninstallExitStatus: + """Whether a removal worked is what the package manager's exit status says. + + Reading it out of human-readable output was guesswork in both directions: + apt's commonest failure (``E: Sub-process /usr/bin/dpkg returned an error + code (1)``) read as success until #240, and a successful removal whose + prerm merely *mentions* a failure read as a failure. + """ + + def test_a_non_zero_exit_is_a_failure_whatever_the_output_says(self, driver): + """Nothing in this output matches a failure keyword; only the exit + status knows.""" + driver._pkg_manager = "dnf" + _mock_send(driver, _with_rc("Removing: wazuh-agent", 1)) + + result = driver.uninstall_package("wazuh-agent") + + assert result["success"] is False + + def test_a_zero_exit_is_a_success_even_if_the_output_mentions_failure(self, driver): + """A prerm that cannot stop an already-dead unit prints "Failed" and + still lets the removal complete.""" + _mock_send( + driver, + _with_rc( + "Removing wazuh-agent (4.14.7-1) ...\n" + "Failed to stop wazuh-agent.service: Unit wazuh-agent.service not loaded.", + 0, + ), + ) + + result = driver.uninstall_package("wazuh-agent") + + assert result["success"] is True + + def test_the_marker_does_not_reach_the_caller(self, driver): + _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0)) + + result = driver.uninstall_package("wazuh-agent") + + assert result["output"] == "Removing wazuh-agent ..." + + def test_the_uninstall_command_keeps_its_exit_status(self, driver): + _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0)) + + driver.uninstall_package("wazuh-agent") + + sent = driver._device.send_command.call_args[0][0] + assert "|| true" not in sent + assert sent.endswith("; echo __NETORK_RC=$?") + + def test_apt_failing_by_exit_status_falls_back_to_dpkg(self, driver): + driver._device.send_command.side_effect = [ + _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100), + _with_rc("Removing wazuh-agent (4.14.7-1) ...", 0), + ] + + result = driver.uninstall_package("wazuh-agent", purge=True) + + assert result["success"] is True + second = driver._device.send_command.call_args_list[1][0][0] + assert "dpkg --purge --force-all" in second + assert "|| true" not in second + assert "__NETORK_RC" not in result["output"] + + def test_the_dpkg_fallback_failing_is_a_failure(self, driver): + driver._device.send_command.side_effect = [ + _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100), + _with_rc("dpkg: error processing package wazuh-agent (--purge):", 1), + ] + + result = driver.uninstall_package("wazuh-agent", purge=True) + + assert result["success"] is False + assert "dpkg --purge --force-all" in result["output"] + + def test_a_zero_exit_does_not_trigger_the_fallback(self, driver): + """Even when the output contains words that used to mean failure: apt + exits 0 for a package that is already gone, which is the state the + caller asked for.""" + _mock_send(driver, _with_rc("Package 'x' is not installed, so not removed", 0)) + + result = driver.uninstall_package("x", purge=True) + + assert result["success"] is True + assert driver._device.send_command.call_count == 1 + + def test_without_an_exit_status_the_output_is_read_as_before(self, driver): + """If the marker never arrived, the keyword check is still the best + answer available — and it errs towards failure on apt's ``E:``.""" + driver._pkg_manager = "dnf" + _mock_send(driver, "E: Sub-process /usr/bin/dpkg returned an error code (1)") + + result = driver.uninstall_package("wazuh-agent") + + assert result["success"] is False