Merge pull request 'fix: decide uninstall success by exit status, not by keywords' (#2) from fix/uninstall-exit-status into master
This commit was merged in pull request #2.
This commit is contained in:
+54
-21
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user