# Copyright Kevin Deldycke <[email protected]> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Unit tests for the privilege-escalation machinery.
These exercise {func}`meta_package_manager.sudo.prime_sudo` and its keepalive with
a mocked `subprocess.run` (no test ever launches a real `sudo`), the escalation
policy inventories across the manager pool, and the
`_StallWatchdog` end to end through
{class}`tests.fake_manager.FakeManager` (whose CLI is the Python interpreter, so
its subprocesses are real but harmless). The `sudo --non-interactive` command
wrapping of
`build_cli()`, the authentication-failure hint of `run()` and the CLI wiring
of `prime_sudo` stay in {mod}`tests.test_execution`.
"""
from __future__ import annotations
import logging
import os
import signal
import subprocess
import sys
import threading
import time
from contextlib import ExitStack, contextmanager
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
import click
import pytest
from click_extra import COLOR_ENVVARS, Spinner
from extra_platforms import is_any_windows
from meta_package_manager.capabilities import Operations
from meta_package_manager.cli import package_task
from meta_package_manager.definitions import OVERRIDABLE_FIELDS
from meta_package_manager.execution import (
STILL_CALL_HINT,
STILL_CALL_MARKER,
CLIExecutor,
)
from meta_package_manager.pool import pool
from meta_package_manager.specifier import Specifier
from meta_package_manager.sudo import (
_SUDO_CACHE_WARM,
ESCALATION,
ESCALATORS,
UNANSWERED_PROMPT_DETAIL,
Escalator,
_escalation_is_passwordless,
_hidden_prompt_risk,
_installed_release,
_is_permission_failure,
_is_sudo_auth_failure,
_is_sudo_denied,
_probe_credentials,
_restore_terminal_modes,
_StallWatchdog,
_terminal_echo,
_terminal_modes,
diagnose_escalation,
inspect_install_root,
prime_sudo,
resolve_escalator,
)
from .fake_manager import FakeManager
# Platform-dependent imports: `pty`, `pwd` and `termios` do not exist on Windows,
# where the tests consuming them are skipped.
if not is_any_windows():
import pty
import pwd
import termios
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Callable
INSTALL_RUN = (Operations.install,)
"""Operations of the run the fakes below are primed for: `install`, the one
operation `_escalating_manager` marks privileged."""
# Escalation policy inventories: which managers escalate through mpm and which run
# sudo themselves, pinned across the whole pool so a new manager cannot silently
# change the escalation story.
[docs]
def test_default_sudo_matches_system_managers():
"""Exactly the system package managers escalate by default.
Classes, their subclasses and bundled definitions alike. User-level managers do not,
the dual-scope language managers (npm, pip, gem, cpan) keep their privileged markers
dormant, and the polkit-native daemon clients (flatpak, fwupd, pkcon) never mark an
operation at all.
`choco` is the lone Windows entry, and belongs on the same grounds as the
rest: Chocolatey installs into a `ProgramData` tree it locks to
administrators on purpose, so each of its four mutating operations fails
unprivileged. `scoop` stays out, being built around a user-owned prefix.
"""
escalating = {mid for mid, manager in pool.items() if type(manager).default_sudo}
assert escalating == {
"apk",
"apt",
"apt-mint",
"aptitude",
"cave",
"choco",
"dkp-pacman",
"deb-get",
"dnf",
"dnf5",
"emerge",
"eopkg",
"ips",
"macports",
"microdnf",
"nala",
"pacman",
"pkg",
"pkg-tools",
"pkgin",
"ports",
"prt-get",
"slapt-get",
"snap",
"sorcery",
"sun-tools",
"swupd",
"tazpkg",
"urpmi",
"xbps",
"yum",
"zypper",
}
for mid in (
"brew",
"cask",
"npm",
"pip",
"gem",
"cpan",
"cargo",
"flatpak",
"fwupd",
"pkcon",
):
assert pool[mid].default_sudo is False
[docs]
def test_internal_sudo_matches_internal_escalators():
"""The markers name exactly the managers whose CLI runs `sudo` mid-run.
None of them escalates through mpm: an internal escalator is never wrapped.
"""
internal = {mid for mid, manager in pool.items() if type(manager).internal_sudo}
assert internal == {
"am",
"aura",
"cask",
"fink",
"pacaur",
"pacstall",
"paru",
"pikaur",
"shelly",
"topgrade",
"trizen",
"yay",
}
for mid in sorted(internal):
assert pool[mid].default_sudo is False
[docs]
def test_internal_sudo_operations_are_narrowed_by_cask_alone():
"""An internal escalator may run its `sudo` in every mutating operation.
Except `cask`: `brew update` and `brew cleanup` never escalate.
"""
narrowed = {
mid: manager.internal_sudo_operations
for mid, manager in pool.items()
if type(manager).internal_sudo
and manager.internal_sudo_operations != CLIExecutor.internal_sudo_operations
}
assert narrowed == {
"cask": frozenset({"install", "remove", "upgrade", "upgrade_all"}),
}
# Sudo priming: probe the credential cache non-interactively first, prompt once up
# front only for the managers mpm itself escalates, then keep the cache warm so a
# password prompt never stalls the concurrent fan-out.
@pytest.fixture(autouse=True)
def _clear_escalator_cache():
"""Keep the `PATH`-derived escalator choice from leaking between tests.
`resolve_escalator` caches on the assumption that `PATH` does not move
mid-run, which a test patching `shutil.which` breaks in both directions.
`_installed_release` caches the same way, on the binary not changing.
"""
resolve_escalator.cache_clear()
_installed_release.cache_clear()
yield
resolve_escalator.cache_clear()
_installed_release.cache_clear()
[docs]
@contextmanager
def only_escalator(escalator_id: str | None, *, selected: str | None = None):
"""Pretend the host carries exactly `escalator_id`, or none at all.
`selected` drives the process-wide choice the CLI would have recorded, so a
test can force an escalator the way `--sudo-command` does.
The binary is also pinned as genuine, so `resolve_escalator` never spends a
`subprocess.run` on {meth}`~meta_package_manager.sudo.Escalator.is_genuine`,
and as a current release accepting every
{attr}`~meta_package_manager.sudo.Escalator.gated_options`. That keeps the
call counts these tests assert on measuring credential probes alone, and
keeps a pretended host from probing the real one. A test about an impostor
or an old release patches those methods itself instead.
A test that plants a stand-in escalator binary on `PATH` runs under this
context. `write_fake_executable` answers every argv the same way, so the
stand-in fails the identity probe and `resolve_escalator` falls through to
the escalator of the host: `run0` on a systemd runner, none on macOS. The
test then passes locally and fails in CI. Clearing the cache alone makes
that failure deterministic and does not remove it.
"""
with (
patch(
"meta_package_manager.sudo.shutil.which",
side_effect=lambda name: (
f"/usr/bin/{name}" if name == escalator_id else None
),
),
patch.object(Escalator, "is_genuine", return_value=True),
patch.object(
Escalator,
"supported_options",
lambda self: tuple(option for option, _ in self.gated_options),
),
):
resolve_escalator.cache_clear()
ESCALATION.select(selected)
try:
yield
finally:
ESCALATION.select(None)
# Entries computed under the patched `shutil.which` must not
# outlive it: the helper is imported by other test modules that
# carry no cache-clearing fixture of their own.
resolve_escalator.cache_clear()
def _prompt_argv(run):
"""The argv of the interactive prompt, found by shape rather than by index.
`prime_sudo` spawns a variable number of non-interactive probes before it
prompts, so a positional lookup breaks whenever one is added. The prompt is
the only call carrying `--prompt` (or, for `doas`, the only one without a
non-interactive flag).
"""
for call in run.call_args_list:
argv = call.args[0]
if "--prompt" in argv:
return argv
raise AssertionError(f"no branded prompt among {run.call_args_list}")
class _EscalatingFakeManager(FakeManager):
"""A fake manager marking its `install` privileged.
The set is a class attribute on every real manager, so it is declared here
rather than assigned to an instance. The ID stays the parent's, which the
notice and prompt assertions read.
"""
id = FakeManager.id
privileged_operations = frozenset({Operations.install.name})
def _escalating_manager() -> FakeManager:
"""A fake manager whose policy escalates its `install`.
To trip prime_sudo in an {data}`INSTALL_RUN`.
"""
manager = _EscalatingFakeManager()
manager.sudo = True
return manager
def _internal_manager() -> FakeManager:
"""A fake manager that escalates internally, like cask and fink.
mpm never wraps it in `sudo`, so prime_sudo only probes on its behalf.
"""
manager = FakeManager()
manager.internal_sudo = True
return manager
[docs]
@contextmanager
def prime_sudo_env(
*,
root: bool = False,
stdin_tty: bool | None = None,
stderr_tty: bool | None = None,
):
"""Patch the whole environment `prime_sudo` probes, yielding the `run` mock.
Pins the effective user (`root`) and the terminal
state (`None` leaves the real descriptor unpatched, for tests that never reach
the TTY check), and replaces `subprocess.run` so no test ever launches a real
`sudo`. Callers set the mock's `return_value`/`side_effect` to shape the
probe and prompt outcomes.
The escalator is pinned to `sudo` rather than detected, since a Windows
runner carries none on `PATH` and detection finding nothing would return
from `prime_sudo` before the paths these tests exercise. A test wanting a
different host nests its own {func}`only_escalator`, whose patch then wins.
"""
with ExitStack() as stack:
stack.enter_context(only_escalator("sudo"))
stack.enter_context(
patch(
"meta_package_manager.sudo.os.geteuid",
return_value=0 if root else 1000,
create=True,
),
)
if stdin_tty is not None:
stack.enter_context(patch("sys.stdin.isatty", return_value=stdin_tty))
if stderr_tty is not None:
stack.enter_context(patch("sys.stderr.isatty", return_value=stderr_tty))
yield stack.enter_context(
patch("meta_package_manager.sudo.subprocess.run"),
)
[docs]
def test_prime_sudo_skips_when_no_manager_escalates():
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env() as run:
prime_sudo(ctx, [FakeManager()], operations=INSTALL_RUN)
run.assert_not_called()
[docs]
def test_prime_sudo_skips_a_run_of_unprivileged_operations():
"""A manager forced to escalate asks for nothing in a run without need.
A run that reaches none of its privileged operations.
The case of `gem` under `[mpm.overrides.gem] sudo = true`: its `sync` runs
`gem sources --update`, which carries no marker, so the password `mpm sync`
would collect on a terminal serves no command.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(stdin_tty=True, stderr_tty=True) as run:
prime_sudo(ctx, [_escalating_manager()], operations=(Operations.sync,))
run.assert_not_called()
[docs]
@pytest.mark.parametrize(
("resets", "prompted"),
(
pytest.param(False, True, id="policy-keeps-it-in-the-prompt"),
pytest.param(True, False, id="tool-drops-the-cache"),
),
)
def test_prime_sudo_counts_an_internal_escalator_on_its_policy(
caplog, resets, prompted
):
"""A forced policy keeps an internal escalator in the prompt of every run.
`mpm` cannot see which of its commands call `sudo` (`fink` does it on `sync`),
so the operations of the run leave it in.
A tool that drops the cached credentials before its own `sudo` is left out:
the password would serve nothing, so one warning says the policy does
nothing for it.
"""
ctx = click.Context(click.Command("mpm"))
manager = _internal_manager()
manager.sudo = True
manager.resets_sudo_timestamp = resets
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
caplog.at_level(logging.WARNING),
):
run.return_value = subprocess.CompletedProcess((), 1)
prime_sudo(ctx, [manager], operations=(Operations.sync,))
asked = any("--prompt" in call.args[0] for call in run.call_args_list)
assert asked is prompted
futile = (
f"A forced sudo policy does nothing for {manager.id}: it drops the "
"cached credentials before its own sudo, so its password prompt still "
"comes."
)
warned = futile in [record.getMessage() for record in caplog.records]
assert warned is not prompted
[docs]
def test_prime_sudo_leaves_a_stock_windows_run_alone():
"""A stock Windows run spends no subprocess, with no platform guard in the way.
What returns early is the empty selection, not the platform: no manager
escalates by default on Windows, so only `--sudo` or a
`[mpm.overrides.<id>] sudo = true` entry reaches the probe, which is what
lets `gsudo` be probed at all when one does.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env() as run, only_escalator("gsudo"):
prime_sudo(ctx, [FakeManager()], operations=INSTALL_RUN)
run.assert_not_called()
[docs]
def test_prime_sudo_probes_gsudo_when_windows_escalation_is_asked_for():
"""A manager forced to escalate reaches the `gsudo` probe.
The probe asks its status rather than trying an elevation that would raise a UAC
dialog.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(stdin_tty=False, stderr_tty=False) as run:
run.return_value = subprocess.CompletedProcess(
(), 0, stdout=b'{"IsElevated":false,"CacheAvailable":false}', stderr=b""
)
with only_escalator("gsudo"):
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert run.call_args_list, "the gsudo probe never ran"
assert run.call_args_list[0].args[0] == ("gsudo", "status", "--json")
[docs]
def test_prime_sudo_skips_when_root():
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(root=True) as run:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
run.assert_not_called()
[docs]
@pytest.mark.parametrize("simulation_flag", ("dry_run", "plan"))
def test_prime_sudo_skips_on_simulation(simulation_flag):
"""Dry and plan runs both skip the sudo prompt.
Neither executes a state-changing CLI.
"""
ctx = click.Context(click.Command("mpm"))
manager = _escalating_manager()
setattr(manager, simulation_flag, True)
with prime_sudo_env() as run:
prime_sudo(ctx, [manager], operations=INSTALL_RUN)
run.assert_not_called()
[docs]
def test_prime_sudo_warns_without_tty(caplog):
"""A cold cache off-terminal: the probe runs, then one warning.
The warning names the managers left to fail fast instead of blocking on a prompt.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(stdin_tty=False) as run, caplog.at_level(logging.WARNING):
run.return_value = subprocess.CompletedProcess((), 1)
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
# Two non-interactive probes and no prompt: the cache is cold, the policy
# grants nothing unauthenticated, and nothing could be answered off-terminal.
assert [call.args[0][:3] for call in run.call_args_list] == [
("sudo", "--non-interactive", "--validate"),
("sudo", "--non-interactive", "--list"),
]
assert any(
"fakemanager needs administrator rights" in record.getMessage()
and "no terminal" in record.getMessage()
and "`NOPASSWD` rule" in record.getMessage()
for record in caplog.records
)
[docs]
@pytest.mark.host_environment
def test_prime_sudo_skips_the_prompt_under_a_passwordless_policy(caplog):
"""A cold cache whose policy still runs every escalated command unauthenticated.
`sudo --validate` refuses whenever *any* matching `sudoers` entry wants a
password, so the `NOPASSWD` rule `docs/sudo.md` recommends reads as a cold
cache once a distribution's stock `ALL ALL=(ALL) ALL` sits beside it.
openSUSE ships exactly that pair, where mpm would otherwise warn that
managers "may fail" and then watch them succeed. Off-terminal on purpose:
that is the path such a warning comes from, and a policy needing no password
needs no terminal either.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(stdin_tty=False) as run, caplog.at_level(logging.WARNING):
def answer(argv, **kwargs):
"""A cold credential cache over a policy that clears the command."""
return subprocess.CompletedProcess((), 0 if "--list" in argv else 1)
run.side_effect = answer
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert _SUDO_CACHE_WARM.is_set()
finally:
ctx.close()
assert not _SUDO_CACHE_WARM.is_set()
assert not caplog.records
assert not any("--prompt" in call.args[0] for call in run.call_args_list)
[docs]
def test_passwordless_probe_asks_about_escalated_binaries_alone(tmp_path):
"""The policy probe names the binary of each manager mpm escalates.
It skips a manager whose commands run unescalated.
"""
sudo = next(e for e in ESCALATORS if e.id == "sudo")
assert sudo.passwordless_probe_args is not None
unescalated = FakeManager()
unescalated.cli_path = tmp_path / "fake-mpm"
escalated = _escalating_manager()
with patch("meta_package_manager.sudo.subprocess.run") as run:
run.return_value = subprocess.CompletedProcess((), 0)
assert _escalation_is_passwordless(sudo, [unescalated, escalated])
assert [call.args[0] for call in run.call_args_list] == [
(*sudo.passwordless_probe_args, str(escalated.cli_path)),
]
[docs]
@pytest.mark.parametrize(
("binary_found", "probe_error"),
(
pytest.param(False, None, id="binary-not-found"),
pytest.param(True, FileNotFoundError("sudo"), id="probe-cannot-run"),
),
)
def test_passwordless_probe_answers_no_when_unsure(
monkeypatch, binary_found, probe_error
):
"""A manager without a binary, or a failed policy probe, answers `False`.
An escalated manager whose binary was not found, or a policy probe that cannot run:
the cold-cache path then handles the run.
"""
sudo = next(e for e in ESCALATORS if e.id == "sudo")
manager = _escalating_manager()
if not binary_found:
monkeypatch.setattr(manager, "cli_path", None)
with patch(
"meta_package_manager.sudo.subprocess.run", side_effect=probe_error
) as run:
assert not _escalation_is_passwordless(sudo, [manager])
# A binary that was not found is no command to ask the policy about.
assert run.call_count == int(binary_found)
[docs]
def test_prime_sudo_authenticates_and_keeps_alive_on_tty():
"""A cold cache on a terminal: probe, prompt, then keepalive.
One branded password prompt, then the keepalive until the context closes.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(stdin_tty=True, stderr_tty=True) as run:
authenticated = []
def answer(argv, **kwargs):
"""A cold cache until the interactive prompt succeeds.
And a policy granting nothing unauthenticated, as sudo behaves. Keyed on
argv rather than on call order, which the probes ahead of it may shift.
"""
if "--prompt" in argv:
authenticated.append(argv)
return subprocess.CompletedProcess((), 0)
code = 0 if authenticated and "--validate" in argv else 1
return subprocess.CompletedProcess((), code)
run.side_effect = answer
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
# The non-interactive probe runs first, then authenticates once, up
# front, before the fan-out, with the branded prompt.
assert run.call_args_list[0].args[0] == (
"sudo",
"--non-interactive",
"--validate",
)
prompt_argv = _prompt_argv(run)
assert prompt_argv[:3] == ("sudo", "--validate", "--prompt")
assert prompt_argv[3].startswith("[mpm] password for ")
assert "fakemanager" in prompt_argv[3]
assert _SUDO_CACHE_WARM.is_set()
finally:
# Stop the keep-alive (a stop callback was registered on the context)
# while subprocess.run is still patched, so no real sudo escapes the
# test.
ctx.close()
assert not _SUDO_CACHE_WARM.is_set()
[docs]
def test_prime_sudo_is_idempotent():
"""A second call on the same context is a no-op: one probe, ever."""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(stdin_tty=True, stderr_tty=True) as run:
run.return_value = subprocess.CompletedProcess((), 0)
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
finally:
ctx.close()
assert run.call_count == 1
assert run.call_args.args[0] == ("sudo", "--non-interactive", "--validate")
[docs]
@pytest.mark.host_environment
def test_prime_sudo_warm_probe_stays_silent_on_tty(capsys):
"""A warm credential cache gets no notice and no prompt, just a keepalive.
Pre-authenticated or NOPASSWD: the keepalive runs until the context closes.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(stdin_tty=True, stderr_tty=True) as run:
run.return_value = subprocess.CompletedProcess((), 0)
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert run.call_count == 1
assert run.call_args.args[0] == ("sudo", "--non-interactive", "--validate")
assert _SUDO_CACHE_WARM.is_set()
finally:
ctx.close()
assert not _SUDO_CACHE_WARM.is_set()
assert capsys.readouterr().err == ""
[docs]
@pytest.mark.host_environment
def test_prime_sudo_warm_probe_keeps_alive_off_tty(caplog):
"""The probe short-circuits before the terminal check.
A CI job with pre-cached credentials gets the keepalive instead of the no-terminal
warning.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(stdin_tty=False) as run, caplog.at_level(logging.WARNING):
run.return_value = subprocess.CompletedProcess((), 0)
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert run.call_count == 1
assert _SUDO_CACHE_WARM.is_set()
finally:
ctx.close()
assert not _SUDO_CACHE_WARM.is_set()
assert not caplog.records
def _wait_for(predicate: Callable[[], bool], timeout: float = 5.0) -> bool:
"""Poll `predicate` until it holds or `timeout` seconds pass."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if predicate():
return True
time.sleep(0.005)
return False
[docs]
def test_keepalive_drop_warns_once_and_rearms(monkeypatch, caplog):
"""A refresh finding the credentials gone clears the warm flag.
Homebrew resets them on purpose. One warning per drop, and a later successful
refresh sets the flag back, saying so at INFO.
"""
monkeypatch.setattr("meta_package_manager.sudo._SUDO_KEEPALIVE_INTERVAL", 0.01)
ctx = click.Context(click.Command("mpm"))
refresh_rc = [0]
def fake_run(argv, **kwargs):
return subprocess.CompletedProcess(argv, refresh_rc[0])
def drop_warnings():
return [
record
for record in caplog.records
if "primed for this run are gone" in record.getMessage()
]
with prime_sudo_env() as run, caplog.at_level(logging.INFO):
run.side_effect = fake_run
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert _SUDO_CACHE_WARM.is_set()
# Drop the credentials: the next refresh finds them gone.
refresh_rc[0] = 1
assert _wait_for(lambda: not _SUDO_CACHE_WARM.is_set())
# Let several more failing refreshes pass: the drop warns only once.
time.sleep(0.1)
assert len(drop_warnings()) == 1
# A re-authentication in this terminal re-warms the cache.
refresh_rc[0] = 0
assert _wait_for(_SUDO_CACHE_WARM.is_set)
finally:
ctx.close()
assert not _SUDO_CACHE_WARM.is_set()
assert len(drop_warnings()) == 1
assert all(record.levelno == logging.WARNING for record in drop_warnings())
assert any(
"warm again" in record.getMessage() and record.levelno == logging.INFO
for record in caplog.records
)
[docs]
def test_keepalive_refresh_of_warm_credentials_stays_at_debug(monkeypatch, caplog):
"""A refresh finding the credentials still there keeps the flag set.
It says so at DEBUG alone: a routine tick is no news for the user.
"""
monkeypatch.setattr("meta_package_manager.sudo._SUDO_KEEPALIVE_INTERVAL", 0.01)
ctx = click.Context(click.Command("mpm"))
def refreshes():
return [
record
for record in list(caplog.records)
if record.getMessage() == "Refreshed the sudo credentials."
]
with prime_sudo_env() as run, caplog.at_level(logging.DEBUG):
run.return_value = subprocess.CompletedProcess((), 0)
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert _wait_for(lambda: bool(refreshes()))
assert _SUDO_CACHE_WARM.is_set()
finally:
ctx.close()
assert not _SUDO_CACHE_WARM.is_set()
assert all(record.levelno == logging.DEBUG for record in refreshes())
[docs]
def test_keepalive_drops_a_refresh_answered_after_the_stop_request(monkeypatch, caplog):
"""A refresh still running when the run ends drops its answer.
The teardown clears the warm flag once it has joined the thread. An answer
that lands after the stop request is about credentials the run no longer
holds, so it must neither set the flag back nor report on them.
The refresh waits for the stop request before it answers, as a `sudo` slow
to answer would. The module's `threading` is replaced by a stand-in that
records the stop event the keepalive creates, so the fake probe can wait for
it.
"""
monkeypatch.setattr("meta_package_manager.sudo._SUDO_KEEPALIVE_INTERVAL", 0.01)
stop_requests: list[threading.Event] = []
def recorded_event() -> threading.Event:
event = threading.Event()
stop_requests.append(event)
return event
monkeypatch.setattr(
"meta_package_manager.sudo.threading",
SimpleNamespace(Event=recorded_event, Thread=threading.Thread),
)
in_flight = threading.Event()
def fake_run(argv, **kwargs):
# The probe of prime_sudo runs before the keepalive creates its event.
# Every later call is a refresh of the keepalive.
if stop_requests:
in_flight.set()
stop_requests[0].wait(5)
return subprocess.CompletedProcess(argv, 0)
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env() as run, caplog.at_level(logging.DEBUG):
run.side_effect = fake_run
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert in_flight.wait(5)
finally:
ctx.close()
assert not _SUDO_CACHE_WARM.is_set()
# A warm answer handled as usual would log this line, the flag still set.
assert "Refreshed the sudo credentials." not in [
record.getMessage() for record in caplog.records
]
[docs]
def test_prime_sudo_narrates_warm_probe_at_info(caplog):
"""A warm probe and the keepalive arming both tell their story at INFO.
So a `--verbosity INFO` run shows why no password prompt appeared.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env() as run, caplog.at_level(logging.INFO):
run.return_value = subprocess.CompletedProcess((), 0)
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
finally:
ctx.close()
messages = [record.getMessage() for record in caplog.records]
assert any("credential cache warm" in message for message in messages)
assert any("Keeping the sudo credentials fresh" in message for message in messages)
[docs]
def test_prime_sudo_narrates_internal_only_skip_at_info(caplog):
"""Not prompting for an internal-only selection surfaces at INFO.
Never as a warning.
"""
ctx = click.Context(click.Command("mpm"))
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
caplog.at_level(logging.INFO),
):
run.return_value = subprocess.CompletedProcess((), 1)
prime_sudo(ctx, [_internal_manager()], operations=INSTALL_RUN)
skips = [
record
for record in caplog.records
if "no up-front prompt" in record.getMessage()
]
assert len(skips) == 1
assert skips[0].levelno == logging.INFO
[docs]
def test_prime_sudo_cold_internal_only_never_prompts_on_tty(capsys, caplog):
"""A cold cache with only internal escalators probes, then never prompts.
Like a stock cask/fink selection: most such runs never escalate, and the stall
notice covers the rare mid-run prompt instead.
"""
ctx = click.Context(click.Command("mpm"))
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
caplog.at_level(logging.WARNING),
):
run.return_value = subprocess.CompletedProcess((), 1)
try:
prime_sudo(ctx, [_internal_manager()], operations=INSTALL_RUN)
finally:
ctx.close()
assert run.call_count == 1
assert run.call_args.args[0] == ("sudo", "--non-interactive", "--validate")
assert not _SUDO_CACHE_WARM.is_set()
assert not caplog.records
assert capsys.readouterr().err == ""
[docs]
def test_prime_sudo_cold_internal_only_stays_silent_off_tty(caplog):
"""Off-terminal, an internal-only selection gets no warning.
Each manager's own sudo fails fast and surfaces through its error path.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env(stdin_tty=False) as run, caplog.at_level(logging.WARNING):
run.return_value = subprocess.CompletedProcess((), 1)
try:
prime_sudo(ctx, [_internal_manager()], operations=INSTALL_RUN)
finally:
ctx.close()
assert run.call_count == 1
assert not _SUDO_CACHE_WARM.is_set()
assert not caplog.records
[docs]
def test_prime_sudo_denied_user_skips_prompt(caplog):
"""A probe reporting a sudoers denial skips the password prompt.
Even on a terminal: a password could only be collected then rejected. One warning
names the managers and the remedy.
"""
ctx = click.Context(click.Command("mpm"))
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
caplog.at_level(logging.WARNING),
):
run.return_value = subprocess.CompletedProcess(
(),
1,
stderr=b"Sorry, user kevin may not run sudo on host.\n",
)
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
# The non-interactive probe is the only subprocess: no prompt is raised.
assert run.call_count == 1
assert not _SUDO_CACHE_WARM.is_set()
assert any(
"not authorized to run sudo" in record.getMessage()
and "fakemanager" in record.getMessage()
and "--no-sudo" in record.getMessage()
for record in caplog.records
)
[docs]
def test_prime_sudo_denied_doas_probe_speaks_for_its_command_alone(caplog):
"""doas rules can match single commands, so the warning stays mild.
`permit nopass tester as root cmd /sbin/apk` denies the `doas -n true` probe while
it runs apk unprompted, measured on Alpine's doas 6.8.2. So the warning does not say
the managers will fail, and no prompt is raised.
"""
ctx = click.Context(click.Command("mpm"))
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
only_escalator("doas", selected="doas"),
caplog.at_level(logging.WARNING),
):
run.return_value = subprocess.CompletedProcess(
(), 1, stderr=b"doas: Operation not permitted\n"
)
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert run.call_count == 1
messages = [record.getMessage() for record in caplog.records]
assert any("denies the `true` command it is probed with" in m for m in messages)
assert not any("will fail" in m for m in messages)
[docs]
def test_prime_sudo_denied_internal_only_stays_silent(caplog):
"""A sudoers denial with only internal escalators selected stays silent.
Each manager's own sudo surfaces the denial through its error path.
"""
ctx = click.Context(click.Command("mpm"))
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
caplog.at_level(logging.WARNING),
):
run.return_value = subprocess.CompletedProcess(
(),
1,
stderr=b"Sorry, user kevin may not run sudo on host.\n",
)
prime_sudo(ctx, [_internal_manager()], operations=INSTALL_RUN)
assert run.call_count == 1
assert not caplog.records
[docs]
@pytest.mark.skipif(
is_any_windows(),
reason="The POSIX ownership check is skipped on Windows.",
)
@pytest.mark.parametrize(
("mode", "expect_warning"),
(
pytest.param(0o755, False, id="trusted"),
pytest.param(0o777, True, id="world-writable"),
),
)
def test_prime_sudo_audits_escalated_binaries(tmp_path, caplog, mode, expect_warning):
"""The binary of an escalated manager gets the config-file tamper test.
One warning per manager when others can modify it, and the escalation still proceeds
(the probe still runs).
"""
ctx = click.Context(click.Command("mpm"))
binary = tmp_path / "fake-mpm"
binary.write_text("#!/bin/sh\n", encoding="UTF-8")
binary.chmod(mode)
manager = _escalating_manager()
manager.cli_path = binary
with prime_sudo_env(stdin_tty=False) as run, caplog.at_level(logging.WARNING):
run.return_value = subprocess.CompletedProcess((), 0)
try:
prime_sudo(ctx, [manager], operations=INSTALL_RUN)
finally:
ctx.close()
assert run.call_count == 1
tamper_warnings = [
record
for record in caplog.records
if "others can write to it" in record.getMessage()
]
if expect_warning:
assert len(tamper_warnings) == 1
assert str(binary) in tamper_warnings[0].getMessage()
assert tamper_warnings[0].label == manager.subject
else:
assert not tamper_warnings
[docs]
@pytest.mark.parametrize(
"probe_error",
(
pytest.param(FileNotFoundError, id="not-found"),
pytest.param(PermissionError, id="not-executable"),
),
)
def test_prime_sudo_warns_when_sudo_cannot_run(caplog, probe_error):
"""A UNIX host with a missing or non-executable sudo gets one warning.
The probe does not crash prime_sudo. Both cases raise an OSError subclass from the
probe.
"""
ctx = click.Context(click.Command("mpm"))
with prime_sudo_env() as run, caplog.at_level(logging.WARNING):
run.side_effect = probe_error
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert not _SUDO_CACHE_WARM.is_set()
assert any(
"sudo could not be run" in record.getMessage() for record in caplog.records
)
[docs]
@pytest.mark.parametrize(
("manager_ids", "expected_notice", "expected_prompt"),
(
pytest.param(
("alpha",),
"alpha needs administrator rights to upgrade.",
"[mpm] password for %p (running alpha): ",
id="singular",
),
pytest.param(
("beta", "alpha"),
"alpha, beta need administrator rights to upgrade.",
"[mpm] password for %p (running alpha, beta): ",
id="plural-sorted",
),
),
)
def test_prime_sudo_notice_names_managers_and_subcommand(
capsys, manager_ids, expected_notice, expected_prompt
):
"""The password notice names the escalating managers and the subcommand.
The managers are sorted, with a verb agreeing in number. The sudo prompt is branded
alike.
"""
ctx = click.Context(click.Command("upgrade"))
managers = []
for manager_id in manager_ids:
manager = _escalating_manager()
manager.id = manager_id
managers.append(manager)
with prime_sudo_env(stdin_tty=True, stderr_tty=True) as run:
# Cold probe, then a failed password prompt: no keepalive to tear down.
run.return_value = subprocess.CompletedProcess((), 1)
try:
prime_sudo(ctx, managers, operations=INSTALL_RUN)
finally:
ctx.close()
assert expected_notice in capsys.readouterr().err
assert _prompt_argv(run) == (
"sudo",
"--validate",
"--prompt",
expected_prompt,
)
[docs]
@pytest.mark.parametrize(
"manager_ids",
(
pytest.param(("alpha", "beta"), id="plain-slugs"),
# Adversarial ids a creative future prompt must still neutralize: sudo expands
# %h/%H/%p/%u/%U in --prompt, and %% collapses to a literal %.
pytest.param(("we%ird", "100%"), id="percent"),
pytest.param(("%p", "%u%H"), id="sudo-escapes"),
pytest.param(("a%%b",), id="pre-doubled"),
),
)
def test_sudo_prompt_respects_sudo_constraints(manager_ids):
"""The `--prompt` argument of `sudo` stays within sudo's constraints.
Whatever the prompt copy becomes.
Locks the properties rather than the wording, so a future rewording trips here
only if it breaks sudo: the only expandable escapes left are the ones the
prompt hands to sudo on purpose, every other `%` being doubled (`sudo
--prompt` expands `%h`/`%H`/`%p`/`%u`/`%U`, and a lone `%` is undefined), the
prompt must be a single line (a newline would detach the ask from the input
cursor), and it must end with a space so the typed password is not glued to
the text.
A manager id is the untrusted half here: whatever it contains, it may never
reach sudo as an escape.
"""
ctx = click.Context(click.Command("upgrade"))
managers = []
for manager_id in manager_ids:
manager = _escalating_manager()
manager.id = manager_id
managers.append(manager)
with prime_sudo_env(stdin_tty=True, stderr_tty=True) as run:
# Cold probe, then a failed password prompt: no keepalive to tear down.
run.return_value = subprocess.CompletedProcess((), 1)
try:
prime_sudo(ctx, managers, operations=INSTALL_RUN)
finally:
ctx.close()
prompt_argv = _prompt_argv(run)
assert prompt_argv[:3] == ("sudo", "--validate", "--prompt")
prompt = prompt_argv[3]
assert prompt
# Doubling check: with every %% pair collapsed away, the only escapes left may
# be the ones the prompt spells itself. Anything else came from a manager id.
residual = prompt.replace("%%", "")
for deliberate in ("%p",):
residual = residual.replace(deliberate, "")
assert "%" not in residual
assert "\n" not in prompt
assert "\r" not in prompt
assert prompt.endswith(" ")
[docs]
def test_sudo_prompt_names_the_account_that_owns_the_password():
"""The prompt defers to sudo's `%p` for whose password is wanted.
Not the caller: a `targetpw`, `rootpw` or `runaspw` policy asks for another
account, and openSUSE ships `Defaults targetpw`, so a prompt naming the
invoking user sends every one of its users to type the wrong password.
"""
ctx = click.Context(click.Command("upgrade"))
manager = _escalating_manager()
manager.id = "zypper"
with prime_sudo_env(stdin_tty=True, stderr_tty=True) as run:
run.return_value = subprocess.CompletedProcess((), 1)
try:
prime_sudo(ctx, [manager], operations=INSTALL_RUN)
finally:
ctx.close()
prompt = _prompt_argv(run)[3]
assert prompt.startswith("[mpm] password for %p ")
assert "zypper" in prompt
RUN0_REFUSAL = (
"Failed to start transient service unit: Access denied as the requested "
"operation requires interactive authentication. However, interactive "
"authentication has not been enabled by the calling program."
)
"""What `run0` writes to `<stderr>` when polkit refuses a `--no-ask-password`
call, measured on Arch Linux running systemd 261.
Signed by neither `run0` nor any `sudo` wording, because systemd's bus layer
words it: the matchers need a branch of their own to see it at all.
"""
PKEXEC_NO_AGENT = (
"Error creating textual authentication agent: Error opening current "
"controlling terminal for the process (`/dev/tty'): No such device or "
"address"
)
"""What `pkexec` writes to `<stderr>` off a terminal, measured over SSH on Arch
Linux running polkit 127.
It raises its own textual agent when no other answers, and that agent needs a
controlling terminal. Like run0's, the message names neither the escalator nor
a `sudo` wording.
"""
[docs]
@pytest.mark.parametrize(
("error", "expected"),
(
("sudo: a password is required", True),
("sudo: a terminal is required to read the password", True),
("sudo: no tty present and no askpass program specified", True),
("SUDO: A PASSWORD IS REQUIRED", True),
# sudo-rs, the default sudo of Ubuntu 25.10 and newer.
("sudo: interactive authentication is required", True),
# The three doas wordings, shared by OpenBSD's doas and opendoas. The
# first is what Alpine's doas 6.8.2 answers under `-n` for both a bare
# `permit` rule and a cold `permit persist` one.
("doas: Authentication required", True),
("doas: a tty is required", True),
("doas: Authentication failed", True),
# Both of run0's refusal paths share a prefix: the non-interactive one
# measured above, and the bare `Access denied` an agent reports when it
# asked and was refused.
(RUN0_REFUSAL, True),
("Failed to start transient service unit: Access denied", True),
# pkexec, whose three authentication failures name neither pkexec nor
# sudo. The first was measured over SSH on Arch with no agent running.
(PKEXEC_NO_AGENT, True),
(
"Error executing command as another user: No authentication agent found.",
True,
),
("Error executing command as another user: Request dismissed", True),
# polkit refusing outright is a denial instead, so it stays False here.
("Error executing command as another user: Not authorized", False),
# Unsigned by any escalator: a command of its own saying the same thing.
("build.sh: authentication required", False),
("", False),
("error: package not found", False),
("sudo: command not found", False),
),
)
def test_is_sudo_auth_failure(error, expected):
assert _is_sudo_auth_failure(error) is expected
[docs]
@pytest.mark.parametrize(
("error", "expected"),
(
# sudo 1.9 and newer, per its message catalog.
("User kevin is not allowed to run sudo on host.", True),
# sudo before 1.9, and sudo-rs's validate denial.
("Sorry, user kevin may not run sudo on host.", True),
("kevin is not in the sudoers file.", True),
# The per-command denial, sudo and sudo-rs alike.
(
"Sorry, user kevin is not allowed to execute '/bin/sh' as root on host.",
True,
),
("SORRY, USER KEVIN MAY NOT RUN SUDO ON HOST.", True),
# sudo-rs denies a `--validate` by quoting HAL 9000 rather than naming
# sudo, and `--validate` is the probe mpm runs first. Matching only the
# `--list` wording above left a non-sudoer on Ubuntu 25.10 and later
# being prompted for a password that could never authorize them.
("sudo: I'm sorry kevin. I'm afraid I can't do that", True),
# The same quote from a command rather than an escalator stays inert.
("hal: I'm afraid I can't do that", False),
# opendoas prints the bare errno string of EPERM for an unmatched rule.
# Confirmed on Alpine's doas 6.8.2 against a config permitting another
# user only.
("doas: Operation not permitted", True),
# An installed but unconfigured doas, measured on a runner carrying no
# `/etc/doas.conf`, and confirmed verbatim on Alpine's doas 6.8.2. No
# password can fix it either.
(
"doas: doas is not enabled, /etc/doas.conf: No such file or directory",
True,
),
# run0 defers to polkit, whose refusal is an authentication failure
# rather than a denial: a prompt may still authorize it.
(RUN0_REFUSAL, False),
# pkexec separates the two: polkit refusing outright is a denial no
# password can lift, where a dismissed prompt is not.
("Error executing command as another user: Not authorized", True),
("Error executing command as another user: Request dismissed", False),
(PKEXEC_NO_AGENT, False),
# A cold cache is an authentication failure, never a denial.
("sudo: a password is required", False),
("sudo: interactive authentication is required", False),
("doas: Authentication required", False),
# The same errno string, from a command rather than from an escalator,
# must not read as a refusal: only a line the escalator signed counts.
("rm: cannot remove '/etc/hosts': Operation not permitted", False),
("", False),
("error: package not found", False),
),
)
def test_is_sudo_denied(error, expected):
assert _is_sudo_denied(error) is expected
[docs]
@pytest.mark.parametrize(
("error", "expected"),
(
# npm, on a root-owned global prefix.
(
(
"npm error Error: EACCES: permission denied, "
"access '/usr/local/lib/node_modules'"
),
True,
),
# gem, on a system Ruby.
(
(
"You don't have write permissions for the "
"/Library/Ruby/Gems/3.4.0 directory."
),
True,
),
# pip and cpan, from the interpreter and the shell.
("[Errno 13] Permission denied: '/usr/lib/python3.12'", True),
("PERMISSION DENIED", True),
("error: package not found", False),
("sudo: a password is required", False),
("", False),
),
)
def test_is_permission_failure(error, expected):
assert _is_permission_failure(error) is expected
[docs]
@pytest.mark.skipif(
is_any_windows(),
reason="The POSIX ownership model does not apply on Windows.",
)
def test_inspect_install_root(tmp_path):
"""The snapshot carries the root's owner; every dead end answers `None`."""
manager = FakeManager()
manager.install_root = tmp_path
root = inspect_install_root(manager)
assert root is not None
assert root.path == tmp_path
assert root.owner_uid == os.getuid()
assert root.owner_name == pwd.getpwuid(os.getuid()).pw_name
manager.install_root = tmp_path / "not-there"
assert inspect_install_root(manager) is None
manager.install_root = None
assert inspect_install_root(manager) is None
[docs]
@pytest.mark.skipif(
is_any_windows(),
reason="The POSIX ownership model does not apply on Windows.",
)
def test_inspect_install_root_swallows_probe_failures():
"""A probe that dies means "unknown", never a crash of the command it decorates."""
class _BrokenProbe(FakeManager):
@property
def install_root(self):
raise RuntimeError("probe exploded")
assert inspect_install_root(_BrokenProbe()) is None
[docs]
def test_inspect_install_root_needs_posix_ownership(monkeypatch, tmp_path):
"""A host whose `os` has no `getuid` reports no install root.
As on Windows, even for a root that exists.
The module's `os` is replaced by a stand-in without `getuid`, so the check
runs on every platform: on POSIX, the real module would report the root.
"""
manager = FakeManager()
manager.install_root = tmp_path
monkeypatch.setattr("meta_package_manager.sudo.os", SimpleNamespace())
assert inspect_install_root(manager) is None
[docs]
@pytest.mark.skipif(
is_any_windows(),
reason="The POSIX ownership model does not apply on Windows.",
)
def test_inspect_install_root_names_an_unknown_owner_by_uid(monkeypatch, tmp_path):
"""A root owned by a uid with no account names its owner by the bare uid.
Like files unpacked from a container image.
"""
def no_account(uid):
raise KeyError(f"getpwuid(): uid not found: {uid}")
monkeypatch.setattr(pwd, "getpwuid", no_account)
manager = FakeManager()
manager.install_root = tmp_path
root = inspect_install_root(manager)
assert root is not None
assert root.owner_uid == os.getuid()
assert root.owner_name == str(os.getuid())
[docs]
@pytest.mark.parametrize(
("manager_id", "expected_argv"),
(
pytest.param("npm", ("prefix",), id="npm"),
pytest.param("gem", ("environment", "gemdir"), id="gem"),
pytest.param(
"pip",
("-c", "import sysconfig; print(sysconfig.get_paths()['purelib'])"),
id="pip",
),
),
)
def test_install_root_probes_run_the_documented_verb(manager_id, expected_argv):
"""Each override resolves its root through its own discovery verb.
Forced to run for real (`force_exec`), like the version probe.
"""
manager = type(pool[manager_id])()
calls = []
def fake_run_cli(*args, **kwargs):
calls.append((args, kwargs))
return "/fake/root\n"
manager.run_cli = fake_run_cli
assert manager.install_root == Path("/fake/root")
assert len(calls) == 1
args, kwargs = calls[0]
assert args == expected_argv
assert kwargs.get("force_exec") is True
# Escalator selection: which binary drives the escalation, and how each dialect
# reaches the argv, the probe and the prompt.
[docs]
def test_escalator_registry_prefers_sudo():
"""`sudo` stays the first choice.
So a host carrying several keeps the behavior it has today, and only an explicit
override moves it.
`run0` sits behind them on purpose: it authorizes through polkit and needs
one running, so it answers for the systemd hosts shipping neither of the
others rather than displacing a working escalator.
`win-sudo` closes the list, behind `gsudo`. Microsoft's `sudo.exe` is inbox
from `24H2` where `gsudo` must be installed, but it caches nothing, so it
is the fallback for a host carrying no `gsudo` rather than the default for
one carrying both.
"""
assert [e.id for e in ESCALATORS] == [
"sudo",
"doas",
"run0",
"pkexec",
"gsudo",
"win-sudo",
]
[docs]
def test_escalator_binary_name_defaults_to_its_id():
"""Only `win-sudo` looks for a file under a name other than its own.
Microsoft's `sudo.exe` ships under the name the Unix escalator claims while
sharing none of its dialect, so the two carry distinct ids and
`resolve_escalator` has to be told which file to look for.
"""
by_id = {e.id: e.binary_name for e in ESCALATORS}
assert by_id["win-sudo"] == "sudo"
assert all(e.binary_name == e.id for e in ESCALATORS if e.id != "win-sudo")
[docs]
@pytest.mark.parametrize(
("returncode", "stdout", "expected"),
(
# Measured on Windows 11 24H2, build 26100.1742, running sudo 1.0.0.
pytest.param(0, "Sudo for Windows. \nCan be used to r", True, id="win-sudo"),
# The Unix escalator's banner, which must not claim this entry: both
# answer to the same file name, and only one takes `--inline`.
pytest.param(0, "Sudo version 1.9.17p2\n", False, id="unix-sudo"),
pytest.param(0, "sudo-rs 0.2.13-0ubuntu1\n", False, id="sudo-rs"),
# gsudo installs an optional `sudo` alias of its own.
pytest.param(0, "gsudo v2.6.1 (Branch)\n", False, id="gsudo-alias"),
pytest.param(1, "Sudo for Windows.\n", False, id="non-zero-exit"),
),
)
def test_win_sudo_identity_rejects_every_other_sudo(returncode, stdout, expected):
"""The help banner names the tool outright.
`--version` prints a bare `sudo 1.0.0` that another implementation shipping a `1.x`
would also match.
"""
win_sudo = next(e for e in ESCALATORS if e.id == "win-sudo")
with patch("meta_package_manager.sudo.subprocess.run") as run:
run.return_value = subprocess.CompletedProcess((), returncode, stdout=stdout)
assert win_sudo.is_genuine() is expected
[docs]
@pytest.mark.parametrize(
("returncode", "stdout", "expected"),
(
pytest.param(0, " Enabled REG_DWORD 0x3\n", True, id="inline"),
pytest.param(0, " Enabled REG_DWORD 0x2\n", False, id="input-closed"),
pytest.param(0, " Enabled REG_DWORD 0x1\n", False, id="new-window"),
pytest.param(0, " Enabled REG_DWORD 0x0\n", False, id="disabled"),
pytest.param(1, "", False, id="value-missing"),
),
)
def test_win_sudo_probe_only_accepts_inline_mode(returncode, stdout, expected):
"""`--inline` errors against any other mode rather than falling back.
So a mode mpm cannot drive has to read as cold before a manager reaches it.
"""
win_sudo = next(e for e in ESCALATORS if e.id == "win-sudo")
probe = subprocess.CompletedProcess((), returncode, stdout=stdout.encode())
assert win_sudo.probe_says_warm(probe) is expected
[docs]
@pytest.mark.parametrize(
"error",
(
# Microsoft's `sudo.exe`, whose whole diagnostic this is: it names
# neither itself nor a reason. Measured on build 26100.1742 from an SSH
# session, whose token is elevated but of type `Default`.
pytest.param("You are not allowed to run sudo", id="win-sudo"),
# The Unix message catalog, in the third person.
pytest.param("kde is not allowed to run sudo on farm.", id="unix-sudo"),
),
)
def test_denial_matches_both_subjects_of_the_sudo_wording(error):
"""The marker is unanchored on the subject so one phrase covers both.
Keyed on `is not allowed`, Microsoft's second-person wording went
unrecognized, and `prime_sudo` would prompt for a consent that cannot
authorize the run.
"""
assert _is_sudo_denied(error) is True
[docs]
@pytest.mark.parametrize(
("returncode", "stdout", "expected"),
(
# Captured from gsudo 2.6.1 on Windows 11 21H2. An SSH session there
# already carries an elevated token, which is the first case.
pytest.param(
0,
b'{\n "IsElevated":true,\n "CacheAvailable":false\n}',
True,
id="elevated",
),
pytest.param(
0,
b'{\n "IsElevated":false,\n "CacheAvailable":true\n}',
True,
id="cache-warm",
),
pytest.param(
0,
b'{\n "IsElevated":false,\n "CacheAvailable":false\n}',
False,
id="cold",
),
# A probe that could not run at all is never warm, whatever it printed.
pytest.param(1, b'{\n "IsElevated":true\n}', False, id="failed-probe-wins"),
),
)
def test_gsudo_probe_reads_its_output_not_its_exit_code(returncode, stdout, expected):
"""`gsudo status` exits `0` whether or not elevation is ready.
So the answer is only in its JSON.
Reading the exit code alone would report every Windows host as warm, and
the up-front prompt that keeps a UAC dialog out of the concurrent fan-out
would never fire.
"""
gsudo = next(e for e in ESCALATORS if e.id == "gsudo")
probe = subprocess.CompletedProcess((), returncode, stdout=stdout)
assert gsudo.probe_says_warm(probe) is expected
[docs]
def test_run0_forwards_a_forced_environment():
"""`run0` forks its payload off the service manager, which inherits nothing.
So a forced variable has to travel as an argument or be lost.
Spliced ahead of the `--` the prefix closes on, leaving a manager's own
flags shielded, and sorted so the disclosed command is stable.
"""
run0 = next(e for e in ESCALATORS if e.id == "run0")
argv = (*run0.escalate_args, "/usr/bin/nala", "upgrade")
assert run0.forward_env(
argv, {"LC_ALL": "C", "DEBIAN_FRONTEND": "noninteractive"}
) == (
"run0",
"--pipe",
"--no-ask-password",
"--setenv=DEBIAN_FRONTEND=noninteractive",
"--setenv=LC_ALL=C",
"--",
"/usr/bin/nala",
"upgrade",
)
# Nothing to carry leaves the command exactly as it was.
assert run0.forward_env(argv, {}) == argv
assert run0.forward_env(argv, None) == argv
# A variable to unset is not carried: the far side never had it.
assert run0.forward_env(argv, {"LC_ALL": None}) == argv
[docs]
@pytest.mark.parametrize(
"escalator",
[e for e in ESCALATORS if e.env_forward_template is None],
ids=[e.id for e in ESCALATORS if e.env_forward_template is None],
)
def test_escalator_without_a_template_forwards_nothing(escalator):
"""An escalator declaring no template leaves the argv alone.
`sudo` and `doas` defer to a `sudoers` policy the host owns.
"""
argv = (*escalator.escalate_args, "/usr/bin/apt", "upgrade")
assert escalator.forward_env(argv, {"LC_ALL": "C"}) == argv
[docs]
@pytest.mark.parametrize(
"escalator",
[e for e in ESCALATORS if e.probe_success_markers is None],
ids=[e.id for e in ESCALATORS if e.probe_success_markers is None],
)
def test_probe_without_markers_reads_the_exit_code(escalator):
"""An escalator declaring no marker keeps the exit code as the answer.
That is what `sudo`, `doas`, `run0` and `pkexec` report through.
"""
assert escalator.probe_says_warm(subprocess.CompletedProcess((), 0, stdout=b""))
assert not escalator.probe_says_warm(subprocess.CompletedProcess((), 1, stdout=b""))
[docs]
def test_pkexec_probe_substitutes_the_running_pid():
"""`pkcheck` asks about a subject and refuses to guess one.
So the probe carries a `{pid}` token mpm fills with its own.
Left unsubstituted it would reach polkit as a literal, and every probe
would report a cold cache on a host that authorizes the action.
"""
pkexec = next(e for e in ESCALATORS if e.id == "pkexec")
assert "{pid}" in pkexec.probe_args
resolved = pkexec.resolved_probe_args()
assert "{pid}" not in resolved
assert str(os.getpid()) in resolved
[docs]
@pytest.mark.parametrize("escalator", ESCALATORS, ids=[e.id for e in ESCALATORS])
def test_probe_args_resolve_for_every_escalator(escalator):
"""Resolving a probe leaves no placeholder behind, whether it carried one or not."""
resolved = escalator.resolved_probe_args()
assert len(resolved) == len(escalator.probe_args)
assert not any("{pid}" in arg for arg in resolved)
[docs]
@pytest.mark.parametrize(
("installed", "expected"),
(
pytest.param("sudo", "sudo", id="sudo-only"),
pytest.param("doas", "doas", id="doas-only"),
pytest.param(None, None, id="neither"),
),
)
def test_resolve_escalator_detects_what_the_host_carries(installed, expected):
with only_escalator(installed):
escalator = resolve_escalator()
assert (escalator.id if escalator else None) == expected
def _both_escalators_installed():
"""Pretend the host carries both `sudo` and `doas`, neither pinned genuine.
So the identity probe is what decides between them.
"""
return patch(
"meta_package_manager.sudo.shutil.which",
side_effect=lambda name: f"/usr/bin/{name}",
)
[docs]
@pytest.mark.parametrize(
("returncode", "stdout", "expected"),
(
# Real sudo, measured on Alpine 3.24.1 running sudo 1.9.17_p2.
pytest.param(0, "Sudo version 1.9.17p2\n", True, id="real-sudo"),
# sudo-rs brands its own banner, measured on Ubuntu 26.04, where it is
# the default `sudo`. Matching `Sudo version` alone rejected it as a
# stand-in and handed a host carrying doas too the wrong escalator.
pytest.param(0, "sudo-rs 0.2.13-0ubuntu1\n", True, id="sudo-rs"),
# Alpine's doas-sudo-shim 0.2.0-r0 rejects the option outright, its
# parser sending every unknown long option to `die`.
pytest.param(1, "", False, id="doas-sudo-shim"),
# A stand-in free to accept the option is still not sudo, so the
# marker is checked and not just the exit status.
pytest.param(0, "some other tool 1.0\n", False, id="silent-impostor"),
# SliTaz 5.0's `su -c` wrapper, run as a non-root user with no terminal:
# BusyBox `su` prints its prompt, reads nothing, and the wrapper exits 0.
pytest.param(0, "Password: \n", False, id="slitaz-su-wrapper"),
# Nothing captured at all.
pytest.param(0, None, False, id="no-output"),
),
)
def test_escalator_identity_probe_reads_the_binary(returncode, stdout, expected):
sudo = ESCALATORS[0]
assert sudo.id == "sudo"
with patch("meta_package_manager.sudo.subprocess.run") as run:
run.return_value = subprocess.CompletedProcess((), returncode, stdout=stdout)
assert sudo.is_genuine() is expected
run.assert_called_once()
assert run.call_args.args[0] == ("sudo", "--version")
[docs]
def test_escalator_without_identity_args_is_genuine():
"""An escalator opts into the identity check by declaring its argv.
So the ones with no known stand-in spend no subprocess.
"""
doas = ESCALATORS[1]
assert doas.id == "doas"
assert doas.identity_args is None
with patch("meta_package_manager.sudo.subprocess.run") as run:
assert doas.is_genuine() is True
run.assert_not_called()
[docs]
def test_escalator_identity_probe_survives_a_missing_binary():
"""A binary that vanished before the probe cannot be driven.
It vanished between `which()` and the probe, and the run does not crash.
"""
with patch("meta_package_manager.sudo.subprocess.run", side_effect=OSError):
assert ESCALATORS[0].is_genuine() is False
[docs]
@pytest.mark.parametrize(
"probe",
(
pytest.param(ESCALATORS[0].is_genuine, id="identity"),
pytest.param(lambda: _installed_release(("pkexec", "--version")), id="release"),
),
)
def test_escalator_probes_never_reach_the_terminal(probe):
"""A stand-in may prompt, so a probe gets no terminal to prompt on.
SliTaz's `/usr/bin/sudo` wraps `su -c`, whose root password prompt landed
in the captured output and blocked every command of a non-root user.
"""
with patch("meta_package_manager.sudo.subprocess.run") as run:
run.return_value = subprocess.CompletedProcess((), 0, stdout="")
probe()
assert run.call_args.kwargs["stdin"] is subprocess.DEVNULL
assert run.call_args.kwargs["start_new_session"] is True
[docs]
@pytest.mark.parametrize(
("returncode", "stdout", "expected"),
(
pytest.param(0, "pkexec version 0.104\n", (), id="slitaz-0.104"),
pytest.param(0, "pkexec version 0.120\n", (), id="last-without"),
pytest.param(0, "pkexec version 121\n", ("--keep-cwd",), id="first-with"),
pytest.param(0, "pkexec version 127\n", ("--keep-cwd",), id="arch-127"),
# An unreadable release keeps the option, so a rejection stays loud.
pytest.param(0, "", ("--keep-cwd",), id="no-version"),
pytest.param(1, "pkexec version 0.104\n", ("--keep-cwd",), id="failed"),
),
)
def test_pkexec_keeps_cwd_from_polkit_121(returncode, stdout, expected):
"""An older `pkexec` reads `--keep-cwd` as the program to run."""
pkexec = next(e for e in ESCALATORS if e.id == "pkexec")
with patch("meta_package_manager.sudo.subprocess.run") as run:
run.return_value = subprocess.CompletedProcess((), returncode, stdout=stdout)
assert pkexec.supported_options() == expected
assert run.call_args.args[0] == ("pkexec", "--version")
[docs]
def test_supported_options_survive_a_missing_binary():
"""A release that cannot be read keeps every option."""
pkexec = next(e for e in ESCALATORS if e.id == "pkexec")
with patch("meta_package_manager.sudo.subprocess.run", side_effect=OSError):
assert pkexec.supported_options() == ("--keep-cwd",)
[docs]
def test_escalators_without_gated_options_spend_no_probe():
"""Only an escalator gating an option reads its release."""
with patch("meta_package_manager.sudo.subprocess.run") as run:
for escalator in ESCALATORS:
if not escalator.gated_options:
assert escalator.supported_options() == ()
run.assert_not_called()
[docs]
def test_resolve_escalator_skips_an_impostor_for_the_real_thing():
"""Alpine's `doas-sudo-shim` puts a forwarding `sudo` on `PATH`.
It forwards to `doas`, accepting `--non-interactive` and no other option mpm sends.
Driving `doas` directly is what keeps the probes answering.
"""
with (
_both_escalators_installed(),
patch.object(
Escalator,
"is_genuine",
autospec=True,
side_effect=lambda self: self.id != "sudo",
),
):
resolve_escalator.cache_clear()
escalator = resolve_escalator()
resolve_escalator.cache_clear()
assert escalator is not None
assert escalator.id == "doas"
[docs]
def test_resolve_escalator_keeps_sudo_on_a_sudo_rs_host():
"""A host whose `sudo` is sudo-rs keeps escalating through it.
Even when it carries `doas` too.
The identity probe reads the binary's own banner, and sudo-rs prints
`sudo-rs <version>` where the original prints `Sudo version <version>`.
Matching only the latter made `is_genuine` reject Ubuntu's default `sudo`,
and the search then settled on `doas`, inverting the preference
{data}`~meta_package_manager.sudo.ESCALATORS` documents.
"""
with (
_both_escalators_installed(),
patch("meta_package_manager.sudo.subprocess.run") as run,
):
run.return_value = subprocess.CompletedProcess(
(), 0, stdout="sudo-rs 0.2.13-0ubuntu1\n"
)
resolve_escalator.cache_clear()
escalator = resolve_escalator()
resolve_escalator.cache_clear()
assert escalator is not None
assert escalator.id == "sudo"
[docs]
@pytest.mark.parametrize(
("windows", "expected"),
(
pytest.param(False, "sudo", id="unix-drives-the-stand-in"),
pytest.param(True, None, id="windows-runs-unescalated"),
),
)
def test_resolve_escalator_falls_back_to_a_lone_impostor_off_windows(windows, expected):
"""An impostor still escalates, so it beats reporting no escalator.
Except on Windows.
Disabled, Microsoft's `sudo.exe` fails both identity probes: its `--help`
prints only how to enable it. The fallback then drove it as POSIX `sudo`,
whose `--non-interactive` it rejects as an unknown argument, which failed
every `choco` install on the `windows-11-arm` runner image. The platform is
pinned both ways, so neither case depends on the host running the suite.
"""
with (
only_escalator("sudo"),
patch.object(Escalator, "is_genuine", return_value=False),
patch("meta_package_manager.sudo.is_any_windows", return_value=windows),
):
resolve_escalator.cache_clear()
escalator = resolve_escalator()
resolve_escalator.cache_clear()
assert (escalator.id if escalator else None) == expected
[docs]
def test_resolve_escalator_override_wins_over_detection():
"""An override is honored even when its binary is missing.
So the failure names the escalator the user asked for.
"""
with only_escalator("sudo"):
escalator = resolve_escalator("doas")
assert escalator is not None
assert escalator.id == "doas"
[docs]
def test_resolve_escalator_rejects_an_unknown_name(caplog):
with only_escalator("sudo"), caplog.at_level(logging.WARNING):
assert resolve_escalator("please") is None
assert any("Unknown sudo_command" in r.getMessage() for r in caplog.records)
[docs]
def test_escalator_choice_is_machine_level():
"""The escalator is process-wide, never a manager attribute.
A machine runs one of them, and every manager escalating on it uses that one.
Guards both halves. No manager may carry the choice as an attribute, and the
per-manager config section must refuse it, so `[mpm.overrides.<id>]` can never
make one manager escalate differently from its peers.
"""
assert not hasattr(FakeManager(), "sudo_command")
assert "sudo_command" not in OVERRIDABLE_FIELDS
assert "sudo_command" not in pool.ALLOWED_EXTRA_OPTION
[docs]
def test_per_manager_sudo_command_is_refused(invoke, tmp_path):
"""A `[mpm.overrides.<id>] sudo_command` entry is rejected outright.
The error names the fields that are per-manager, rather than silently escalating one
manager through another binary.
"""
config = tmp_path / "config.toml"
config.write_text(
'[mpm.overrides.apt]\nsudo_command = "doas"\n',
encoding="UTF-8",
)
result = invoke("--config", str(config), "managers")
assert result.exit_code != 0
assert "mpm.overrides.apt.sudo_command: unknown field" in result.stderr
[docs]
def test_sudo_command_reaches_the_config_file(invoke, tmp_path):
"""`[mpm] sudo_command` is honored, and validated like the flag.
Against the same choices: an unknown escalator is a usage error, not a silent
fallback.
"""
config = tmp_path / "config.toml"
config.write_text('[mpm]\nsudo_command = "bogus"\n', encoding="UTF-8")
result = invoke("--config", str(config), "managers")
assert result.exit_code == 2
assert "'bogus' is not one of 'sudo', 'doas'" in result.stderr
[docs]
@pytest.mark.parametrize(
("escalator_id", "expected_prefix"),
(
pytest.param("sudo", ("sudo", "--non-interactive"), id="sudo"),
# `doas` parses short options only: `-n` is its whole "do not prompt"
# vocabulary, so the long-form convention cannot reach it.
pytest.param("doas", ("doas", "-n"), id="doas"),
),
)
def test_build_cli_escalates_with_the_selected_binary(escalator_id, expected_prefix):
manager = _escalating_manager()
with only_escalator(escalator_id, selected=escalator_id):
args = manager.build_cli("install", "pkg", sudo=True)
assert tuple(str(a) for a in args[: len(expected_prefix)]) == expected_prefix
[docs]
def test_build_cli_skips_escalation_without_an_escalator():
"""A host carrying neither binary runs the command unprivileged.
Rather than prefixing a `sudo` that is not there.
"""
manager = _escalating_manager()
with only_escalator(None):
args = manager.build_cli("install", "pkg", sudo=True)
assert "sudo" not in [str(a) for a in args]
assert "doas" not in [str(a) for a in args]
[docs]
def test_prime_sudo_warns_without_any_escalator(caplog):
ctx = click.Context(click.Command("mpm"))
with (
prime_sudo_env() as run,
only_escalator(None),
caplog.at_level(logging.WARNING),
):
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
run.assert_not_called()
assert any("Found none of sudo, doas" in r.getMessage() for r in caplog.records)
[docs]
def test_prime_sudo_probes_and_prompts_through_doas(capsys):
"""A doas host is probed with `doas -n true`.
On a cold cache, it is prompted with a bare `doas true`: it cannot be told what
prompt to print, so the echoed notice carries the explanation instead.
"""
ctx = click.Context(click.Command("upgrade"))
manager = _escalating_manager()
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
only_escalator("doas", selected="doas"),
):
run.side_effect = (
subprocess.CompletedProcess((), 1), # Cold-cache probe.
subprocess.CompletedProcess((), 0), # Successful password prompt.
subprocess.CompletedProcess((), 0), # A `persist` rule kept it.
)
try:
prime_sudo(ctx, [manager], operations=INSTALL_RUN)
assert run.call_args_list[0].args[0] == ("doas", "-n", "true")
assert run.call_args_list[1].args[0] == ("doas", "true")
assert run.call_args_list[2].args[0] == ("doas", "-n", "true")
assert _SUDO_CACHE_WARM.is_set()
finally:
ctx.close()
assert "needs administrator rights to upgrade" in capsys.readouterr().err
[docs]
def test_doas_gets_no_keepalive_thread(caplog):
"""A doas cache is marked warm, without a thread refreshing it.
doas persistence is opt-in per rule, so a recurring probe would report a drop on
every tick of a host that never asked for it.
"""
ctx = click.Context(click.Command("mpm"))
manager = _escalating_manager()
with (
prime_sudo_env() as run,
only_escalator("doas", selected="doas"),
caplog.at_level(logging.INFO),
):
run.return_value = subprocess.CompletedProcess((), 0)
try:
prime_sudo(ctx, [manager], operations=INSTALL_RUN)
assert _SUDO_CACHE_WARM.is_set()
# The probe is the only subprocess: no refresh tick follows it.
assert run.call_count == 1
finally:
ctx.close()
assert not _SUDO_CACHE_WARM.is_set()
assert any(
"cannot be refreshed on a schedule" in r.getMessage() for r in caplog.records
)
[docs]
@pytest.mark.parametrize("escalator", ESCALATORS, ids=[e.id for e in ESCALATORS])
def test_prime_sudo_names_the_remedy_without_tty(escalator, caplog):
"""Off a terminal, the warning closes on what the policy needs.
What the escalator's policy needs, rather than a pre-authentication that most
escalators do not keep.
"""
ctx = click.Context(click.Command("mpm"))
with (
prime_sudo_env(stdin_tty=False) as run,
only_escalator(escalator.id, selected=escalator.id),
caplog.at_level(logging.WARNING),
):
run.return_value = subprocess.CompletedProcess((), 1)
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert any(
"no terminal" in record.getMessage() and escalator.remedy in record.getMessage()
for record in caplog.records
)
PROMPT_BLIND = [e for e in ESCALATORS if e.retention_markers is None]
"""Escalators whose probe cannot tell ahead of a prompt whether it would keep
the authorization."""
[docs]
@pytest.mark.parametrize("escalator", PROMPT_BLIND, ids=[e.id for e in PROMPT_BLIND])
def test_prime_sudo_warns_when_the_prompt_keeps_nothing(escalator, caplog):
"""A prompt answered, then a probe still cold: the policy keeps nothing.
Like `doas` without `persist` or `run0` over SSH. One warning names the remedy, and
no keepalive pretends otherwise.
"""
ctx = click.Context(click.Command("mpm"))
prompt_args = escalator.prompt_args
def answer(argv, **kwargs):
"""Only the interactive prompt succeeds."""
return subprocess.CompletedProcess(
(), 0 if tuple(argv[: len(prompt_args)]) == prompt_args else 1
)
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
only_escalator(escalator.id, selected=escalator.id),
caplog.at_level(logging.WARNING),
):
run.side_effect = answer
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert not _SUDO_CACHE_WARM.is_set()
finally:
ctx.close()
assert run.call_args.args[0] == escalator.resolved_probe_args()
assert any(
"keeps no authorization past a password prompt" in record.getMessage()
and escalator.remedy in record.getMessage()
for record in caplog.records
)
# `pkcheck` answers, on polkit 0.104 under SliTaz 5.0, a cold probe of each policy.
AUTH_ADMIN_KEEP = b"polkit\\56retains_authorization_after_challenge=1\n"
AUTH_ADMIN = b""
[docs]
@pytest.mark.parametrize(
("probe_stdout", "prompted"),
(
pytest.param(AUTH_ADMIN_KEEP, True, id="auth_admin_keep"),
pytest.param(AUTH_ADMIN, False, id="auth_admin"),
),
)
def test_prime_sudo_prompts_only_when_polkit_keeps_it(caplog, probe_stdout, prompted):
"""polkit says ahead of a prompt whether it keeps what the password buys.
Under `auth_admin` it keeps nothing, so no password is asked for at all.
"""
pkexec = next(e for e in ESCALATORS if e.id == "pkexec")
ctx = click.Context(click.Command("mpm"))
authenticated = []
def answer(argv, **kwargs):
"""A cold probe until the prompt authenticates, as polkit keeps it."""
if tuple(argv) == pkexec.prompt_args:
authenticated.append(argv)
return subprocess.CompletedProcess((), 0)
return subprocess.CompletedProcess((), 0 if authenticated else 2, probe_stdout)
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
only_escalator("pkexec", selected="pkexec"),
caplog.at_level(logging.WARNING),
):
run.side_effect = answer
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert _SUDO_CACHE_WARM.is_set() is prompted
finally:
ctx.close()
assert bool(authenticated) is prompted
warned = any(
"keeps no authorization past a password prompt" in record.getMessage()
for record in caplog.records
)
assert warned is not prompted
[docs]
@pytest.mark.parametrize(
("pkcheck_stdout", "prompted"),
(
# `pkcheck` on `org.freedesktop.systemd1.manage-units`, measured on
# Fedora 44 with polkit 127, under each policy systemd ships.
pytest.param(
"polkit\\56result=auth_admin_keep\n"
"polkit\\56retains_authorization_after_challenge=1\n",
True,
id="local-session",
),
pytest.param("polkit\\56result=auth_admin\n", False, id="ssh-session"),
# No `pkcheck` to ask: the prompt goes ahead, as before.
pytest.param(None, True, id="no-pkcheck"),
),
)
def test_prime_sudo_asks_polkit_before_prompting_for_run0(
caplog, pkcheck_stdout, prompted
):
"""`pkcheck` asks polkit about the action run0 authorizes against.
run0's own probe cannot tell whether polkit keeps what a password buys.
"""
run0 = next(e for e in ESCALATORS if e.id == "run0")
ctx = click.Context(click.Command("mpm"))
authenticated = []
def answer(argv, **kwargs):
"""A cold probe until the prompt authenticates, as polkit keeps it."""
argv = tuple(argv)
if argv[0] == "pkcheck":
if pkcheck_stdout is None:
raise FileNotFoundError(argv[0])
return subprocess.CompletedProcess(argv, 2, stdout=pkcheck_stdout)
if argv == run0.prompt_args:
authenticated.append(argv)
return subprocess.CompletedProcess(argv, 0)
return subprocess.CompletedProcess(argv, 0 if authenticated else 1)
with (
prime_sudo_env(stdin_tty=True, stderr_tty=True) as run,
only_escalator("run0", selected="run0"),
caplog.at_level(logging.WARNING),
):
run.side_effect = answer
try:
prime_sudo(ctx, [_escalating_manager()], operations=INSTALL_RUN)
assert _SUDO_CACHE_WARM.is_set() is prompted
finally:
ctx.close()
assert bool(authenticated) is prompted
assert any(call.args[0][0] == "pkcheck" for call in run.call_args_list)
[docs]
@pytest.mark.parametrize("escalator", ESCALATORS, ids=[e.id for e in ESCALATORS])
def test_every_escalator_names_a_remedy(escalator):
"""Warnings go on after the remedy, so it is a sentence of its own."""
assert escalator.remedy[0].isupper()
assert escalator.remedy.endswith(".")
[docs]
@pytest.mark.parametrize(
("on_path", "genuine", "warns"),
(
pytest.param(True, False, True, id="stand-in"),
pytest.param(True, True, False, id="genuine"),
pytest.param(False, False, False, id="missing"),
),
)
def test_override_warns_on_a_stand_in(caplog, on_path, genuine, warns):
"""`--sudo-command` is honored even for a stand-in, with one warning.
SliTaz's `sudo` wraps `su -c` and exits 0 whatever happened.
"""
with (
patch(
"meta_package_manager.sudo.shutil.which",
return_value="/usr/bin/sudo" if on_path else None,
),
patch.object(Escalator, "is_genuine", return_value=genuine),
caplog.at_level(logging.WARNING),
):
escalator = resolve_escalator("sudo")
assert escalator is not None
assert escalator.id == "sudo"
warned = any(
"does not identify as sudo" in record.getMessage() for record in caplog.records
)
assert warned is warns
# `mpm doctor` escalation report: what the probes find, without ever prompting.
PKEXEC = next(e for e in ESCALATORS if e.id == "pkexec")
PKCHECK_COLD = b"Authorization requires authentication and -u wasn't passed.\n"
def _slitaz_answer(argv, **kwargs):
"""How a non-root user's SliTaz 5.0 answers each probe.
Its `sudo` is a BusyBox `su -c` wrapper, and its polkit `0.104` grants nothing by
default.
"""
if tuple(argv) == ("sudo", "--version"):
return subprocess.CompletedProcess(argv, 0, stdout="Password: \n")
if tuple(argv) == ("pkexec", "--version"):
return subprocess.CompletedProcess(argv, 0, stdout="pkexec version 0.104\n")
return subprocess.CompletedProcess(argv, 2, stdout=b"", stderr=PKCHECK_COLD)
[docs]
def test_diagnose_escalation_on_slitaz():
"""The whole report of a SliTaz host.
Auto-detection passes over the `su` wrapper there, and polkit would keep nothing a
prompt buys.
"""
with (
patch(
"meta_package_manager.sudo.shutil.which",
side_effect=lambda name: (
f"/usr/bin/{name}" if name in ("sudo", "pkexec") else None
),
),
patch("meta_package_manager.sudo.subprocess.run", side_effect=_slitaz_answer),
patch("meta_package_manager.sudo.os.geteuid", return_value=1000, create=True),
):
diagnosis = diagnose_escalation([_escalating_manager()])
assert diagnosis.lines() == [
"Escalator: pkexec at /usr/bin/pkexec, detected.",
"Skipped: the sudo on PATH does not identify as sudo.",
"Release: 0.104, too old for --keep-cwd.",
(
"Credentials: a password is needed "
"(Authorization requires authentication and -u wasn't passed.)."
),
(
"Prompt: the authorization would not hold past it, so mpm asks for "
"no password."
),
f"Fix: {PKEXEC.remedy}",
"Escalated by mpm: fakemanager.",
]
[docs]
def test_diagnose_escalation_as_root():
"""Root runs every command as is, so nothing is probed."""
with (
patch("meta_package_manager.sudo.subprocess.run") as run,
patch("meta_package_manager.sudo.os.geteuid", return_value=0, create=True),
):
diagnosis = diagnose_escalation([_escalating_manager()])
run.assert_not_called()
assert diagnosis.lines() == ["Running as root: no command needs an escalator."]
[docs]
@pytest.mark.parametrize(
("probe", "expected"),
(
pytest.param(
subprocess.CompletedProcess((), 0, stdout=b"", stderr=b""),
["Credentials: ready, escalated commands run with no prompt."],
id="warm",
),
pytest.param(
subprocess.CompletedProcess(
(), 1, stdout=b"", stderr=b"doas: Operation not permitted\n"
),
[
"Credentials: denied (doas: Operation not permitted).",
(
"Probe: it runs `true`, and a rule can still permit the "
"managers' own commands."
),
(
"Fix: Give this user a doas rule for the escalated commands, "
"or drop escalation with `--no-sudo`."
),
],
id="scoped-rule",
),
),
)
def test_diagnose_doas_credentials(probe, expected):
"""What the doas credential probe found, as the report words it."""
with (
only_escalator("doas"),
patch("meta_package_manager.sudo.subprocess.run", return_value=probe),
patch("meta_package_manager.sudo.os.geteuid", return_value=1000, create=True),
):
lines = diagnose_escalation([_escalating_manager()]).lines()
assert lines[0] == "Escalator: doas at /usr/bin/doas, detected."
assert lines[1 : 1 + len(expected)] == expected
[docs]
def test_diagnose_escalation_without_escalator():
"""A host carrying none names the ones mpm looks for, and the way out."""
with (
only_escalator(None),
patch("meta_package_manager.sudo.os.geteuid", return_value=1000, create=True),
):
lines = diagnose_escalation([_escalating_manager()]).lines()
assert lines[0] == (
"Escalator: none found among sudo, doas, run0, pkexec, gsudo, win-sudo."
)
assert lines[1].startswith("Fix: Install one")
[docs]
def test_diagnose_escalation_tells_unrunnable_from_stand_in():
"""A binary that cannot run is no stand-in: the report says which it is."""
with (
patch(
"meta_package_manager.sudo.shutil.which",
side_effect=lambda name: (
f"/usr/bin/{name}" if name in ("sudo", "doas") else None
),
),
patch("meta_package_manager.sudo.subprocess.run", side_effect=OSError),
patch("meta_package_manager.sudo.os.geteuid", return_value=1000, create=True),
):
lines = diagnose_escalation([_escalating_manager()]).lines()
assert "Skipped: the sudo on PATH could not run." in lines
assert "Credentials: doas could not run." in lines
[docs]
def test_diagnose_escalation_under_a_passwordless_policy():
"""A cold cache over a policy that clears every command is passwordless.
The credential cache is cold, but the policy clears every escalated command: no
password to type, so no fix to apply.
The internal escalator beside it gets a line of its own, and stays out of the
policy probe, which asks about the commands mpm escalates alone.
"""
def answer(argv, **kwargs):
"""A genuine `sudo`, with a cold cache and a policy clearing the command."""
argv = tuple(argv)
if argv == ("sudo", "--version"):
return subprocess.CompletedProcess(
argv, 0, stdout="Sudo version 1.9.17p2\n"
)
return subprocess.CompletedProcess(
argv, 0 if "--list" in argv else 1, stdout=b"", stderr=b""
)
sudo = next(e for e in ESCALATORS if e.id == "sudo")
assert sudo.passwordless_probe_args is not None
escalated = _escalating_manager()
with (
only_escalator("sudo"),
patch("meta_package_manager.sudo.subprocess.run", side_effect=answer) as run,
patch("meta_package_manager.sudo.os.geteuid", return_value=1000, create=True),
):
lines = diagnose_escalation([escalated, _internal_manager()]).lines()
assert lines == [
"Escalator: sudo at /usr/bin/sudo, detected.",
"Credentials: the policy runs every escalated command with no password.",
"Escalated by mpm: fakemanager.",
"Escalating on their own: fakemanager.",
]
policy_probes = [
call.args[0] for call in run.call_args_list if "--list" in call.args[0]
]
assert policy_probes == [
(*sudo.passwordless_probe_args, str(escalated.cli_path)),
]
[docs]
def test_diagnose_escalation_when_polkit_keeps_the_authorization():
"""polkit under `auth_admin_keep` keeps what a password buys.
So the report says a prompt would hold for the whole run.
The `pkcheck` answer of the credential probe is the one measured on Fedora 44
with polkit 127.
"""
def answer(argv, **kwargs):
"""A cold `pkcheck` whose policy keeps the authorization it grants."""
if tuple(argv) == ("pkexec", "--version"):
return subprocess.CompletedProcess(argv, 0, stdout="pkexec version 127\n")
return subprocess.CompletedProcess(
argv,
2,
stdout=b"polkit\\56result=auth_admin_keep\n" + AUTH_ADMIN_KEEP,
stderr=PKCHECK_COLD,
)
with (
only_escalator("pkexec"),
patch("meta_package_manager.sudo.subprocess.run", side_effect=answer),
patch("meta_package_manager.sudo.os.geteuid", return_value=1000, create=True),
):
lines = diagnose_escalation([_escalating_manager()]).lines()
assert lines == [
"Escalator: pkexec at /usr/bin/pkexec, detected.",
"Release: 127.",
(
"Credentials: a password is needed "
"(Authorization requires authentication and -u wasn't passed.)."
),
"Prompt: the authorization would hold for the whole run.",
f"Fix: {PKEXEC.remedy}",
"Escalated by mpm: fakemanager.",
]
def _slitaz_sudo_answer(argv, **kwargs):
"""SliTaz's `sudo`, a BusyBox `su -c` wrapper.
It asks for the root password whatever the arguments, and exits 0 whatever happened.
"""
prompt = "Password: \n"
return subprocess.CompletedProcess(
argv, 0, stdout=prompt if kwargs.get("text") else prompt.encode()
)
[docs]
def test_diagnose_escalation_flags_a_chosen_stand_in():
"""A stand-in named with `--sudo-command` is driven anyway.
So the report warns that its answers may be misread.
SliTaz's wrapper exits 0 on the credential probe, which reads as ready: the
identity line is what tells the user not to trust it.
"""
with (
only_escalator("sudo", selected="sudo"),
patch(
"meta_package_manager.sudo.subprocess.run",
side_effect=_slitaz_sudo_answer,
),
patch("meta_package_manager.sudo.os.geteuid", return_value=1000, create=True),
):
lines = diagnose_escalation([_escalating_manager()]).lines()
assert lines == [
"Escalator: sudo at /usr/bin/sudo, chosen with --sudo-command.",
(
"Identity: the sudo on PATH does not identify as sudo, so mpm may "
"misread its answers."
),
"Credentials: ready, escalated commands run with no prompt.",
"Escalated by mpm: fakemanager.",
]
# Stall watchdog: a call of an internal escalator (cask, fink) that goes silent
# may be blocked on a hidden sudo password prompt. run() arms _StallWatchdog
# around the spawn to flag the silence.
def _stalling_script(line: str, sleep: float) -> str:
"""A one-liner that prints `line`, flushes, then stays silent.
It stays silent for `sleep` seconds, mimicking an installer blocked on a hidden
prompt.
"""
return f"import sys, time; print({line!r}); sys.stdout.flush(); time.sleep({sleep})"
[docs]
def test_stall_watchdog_notices_silent_internal_escalator(monkeypatch, caplog):
"""A mutating escalator call on a cold cache warns once per silence.
A mutating call of an internal escalator, on a terminal with a cold cache, warns
once per silence episode, naming the manager and quoting its last output line.
"""
assert not _SUDO_CACHE_WARM.is_set()
manager = FakeManager()
manager.internal_sudo = True
manager._active_operation = "install"
monkeypatch.setattr("meta_package_manager.sudo._STALL_NOTICE_DELAY", 0.2)
with (
patch("sys.stderr.isatty", return_value=True),
caplog.at_level(logging.WARNING),
):
manager.run_cli("-c", _stalling_script("installer may ask", 2))
notices = [
record
for record in caplog.records
if "hidden password prompt" in record.getMessage()
]
assert notices
assert all(record.label == manager.subject for record in notices)
assert all(record.getMessage().startswith("No output for ") for record in notices)
# A slow interpreter startup may trip an extra no-output-yet notice before the
# line arrives; the episode quoting the line warns exactly once.
quoting = [
record
for record in notices
if 'Last output: "installer may ask"' in record.getMessage()
]
assert len(quoting) == 1
[docs]
def test_stall_watchdog_ignores_the_command_line(monkeypatch, caplog):
"""The command line `run_cli` discloses is no output of the child.
A child that prints nothing gets the notice of a silence since the start, which
quotes no line.
"""
assert not _SUDO_CACHE_WARM.is_set()
manager = FakeManager()
manager.internal_sudo = True
manager._active_operation = "install"
monkeypatch.setattr("meta_package_manager.sudo._STALL_NOTICE_DELAY", 0.2)
with (
patch("sys.stderr.isatty", return_value=True),
caplog.at_level(logging.WARNING),
):
manager.run_cli("-c", "import time; time.sleep(1)")
notices = [
record.getMessage()
for record in caplog.records
if "hidden password prompt" in record.getMessage()
]
assert notices
for notice in notices:
assert notice.endswith("No output since the command started.")
[docs]
def test_stall_watchdog_notice_starts_on_a_fresh_line(monkeypatch, capsys):
"""The notice breaks the line before writing, sparing a prompt sitting on it.
A tool's own `sudo` writes `Password:` to `/dev/tty` with no trailing
newline, and that never reaches the pipe mpm reads, so a notice written
straight to `stderr` runs on from it as `Password:warning:cask: No output
for 30s…` and buries the prompt mid-sentence.
The child here prints nothing, so its single silence episode leaves `stderr`
carrying the notice alone: a leading newline is then the whole difference.
"""
assert not _SUDO_CACHE_WARM.is_set()
manager = FakeManager()
manager.internal_sudo = True
manager._active_operation = "install"
monkeypatch.setattr("meta_package_manager.sudo._STALL_NOTICE_DELAY", 0.2)
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter("%(message)s"))
root = logging.getLogger()
root.addHandler(handler)
try:
with patch("sys.stderr.isatty", return_value=True):
manager.run_cli("-c", "import time; time.sleep(1)")
finally:
root.removeHandler(handler)
captured = capsys.readouterr().err
assert "No output for " in captured
assert captured.startswith("\n")
[docs]
def test_stall_watchdog_notice_without_an_output_line(monkeypatch, caplog):
"""A silence that started with the call has no line to quote, and says so.
The watchdog is armed alone, with no call behind it to feed it a line, and
with no terminal to read: an echo the host terminal holds off would hold the
notice back.
"""
monkeypatch.setattr("meta_package_manager.sudo._STALL_NOTICE_DELAY", 0.05)
monkeypatch.setattr("meta_package_manager.sudo._terminal_modes", lambda: None)
def notices():
return [
record
for record in list(caplog.records)
if "hidden password prompt" in record.getMessage()
]
label = FakeManager().subject
with caplog.at_level(logging.WARNING):
watchdog = _StallWatchdog(label)
try:
assert _wait_for(lambda: bool(notices()))
finally:
watchdog.stop()
# One silence episode, so one notice.
assert len(notices()) == 1
assert notices()[0].getMessage().endswith("No output since the command started.")
assert notices()[0].label == label
[docs]
@pytest.fixture
def terminal(monkeypatch):
"""A pseudo-terminal that answers for the controlling one, as its path.
A test run has no terminal to hold in a password read, and under CI it has
none at all, so the watchdog is pointed at this one.
"""
try:
master, slave = pty.openpty()
except OSError:
pytest.skip("This host allocates no pseudo-terminal, as a build sandbox.")
path = os.ttyname(slave)
monkeypatch.setattr("meta_package_manager.sudo.os.ctermid", lambda: path)
yield path
os.close(master)
os.close(slave)
def _echo(terminal: str) -> bool:
"""Whether `terminal` echoes the keys typed on it."""
descriptor = os.open(terminal, os.O_RDONLY | os.O_NOCTTY)
try:
return bool(termios.tcgetattr(descriptor)[3] & termios.ECHO)
finally:
os.close(descriptor)
def _set_echo(terminal: str, echo: bool) -> None:
"""Turn the echo of `terminal` on or off, the flag a password read holds off."""
descriptor = os.open(terminal, os.O_RDWR | os.O_NOCTTY)
try:
attributes = termios.tcgetattr(descriptor)
if echo:
attributes[3] |= termios.ECHO
else:
attributes[3] &= ~termios.ECHO
# TCSANOW: a drain waits forever on a terminal nobody reads the other
# end of.
termios.tcsetattr(descriptor, termios.TCSANOW, attributes)
finally:
os.close(descriptor)
def _prompting_script(terminal: str, line: str, sleep: float) -> str:
"""A one-liner that turns the echo off, prints `line`, then stays silent.
It turns the echo of `terminal` off and stays silent for `sleep` seconds, mimicking
a tool that waits on the password prompt of its own `sudo`.
"""
return (
"import os, sys, termios, time; "
f"descriptor = os.open({terminal!r}, os.O_RDWR | os.O_NOCTTY); "
"attributes = termios.tcgetattr(descriptor); "
"attributes[3] &= ~termios.ECHO; "
"termios.tcsetattr(descriptor, termios.TCSANOW, attributes); "
f"print({line!r}); sys.stdout.flush(); time.sleep({sleep})"
)
[docs]
@pytest.mark.skipif(is_any_windows(), reason="A pseudo-terminal is POSIX-only.")
@pytest.mark.parametrize(
("echo_at_start", "prompts", "held_back"),
(
pytest.param(True, True, True, id="prompt-on-screen"),
pytest.param(True, False, False, id="silent-without-a-prompt"),
pytest.param(False, True, False, id="echo-already-off-at-start"),
),
)
def test_stall_watchdog_holds_the_notice_back_for_a_prompt_on_screen(
monkeypatch, caplog, terminal, echo_at_start, prompts, held_back
):
"""A password prompt the terminal shows gets no notice written below it.
The notice ends on a newline, so it leaves the cursor on an empty line under
the warning, with the prompt it must answer two lines above. The watchdog
reads a prompt from the echo the child turned off, and stays silent.
The two controls prove the echo is what holds the notice back. A child that
leaves the echo on draws the notice. So does one that finds it off already,
since an echo that was never on proves nothing about a prompt.
A slow interpreter startup may trip a no-output-yet notice before the child
reaches the terminal, so the assertions read the episode that quotes its
line: the echo is off by the time that line is printed.
"""
assert not _SUDO_CACHE_WARM.is_set()
manager = FakeManager()
manager.internal_sudo = True
manager._active_operation = "install"
monkeypatch.setattr("meta_package_manager.sudo._STALL_NOTICE_DELAY", 0.2)
_set_echo(terminal, echo_at_start)
if prompts:
script = _prompting_script(terminal, "installer asks", 1)
else:
script = _stalling_script("installer asks", 1)
with (
patch("sys.stderr.isatty", return_value=True),
caplog.at_level(logging.DEBUG),
):
manager.run_cli("-c", script)
messages = [record.getMessage() for record in caplog.records]
quoting = [
message
for message in messages
if "hidden password prompt" in message
and 'Last output: "installer asks"' in message
]
assert len(quoting) == (0 if held_back else 1)
assert (
"Held the stall notice back: a password prompt was on screen." in messages
) is held_back
[docs]
@pytest.mark.skipif(is_any_windows(), reason="A pseudo-terminal is POSIX-only.")
@pytest.mark.parametrize(
("echo_at_start", "prompts", "named"),
(
pytest.param(True, True, True, id="prompt-left-unanswered"),
pytest.param(True, False, False, id="silent-without-a-prompt"),
pytest.param(False, True, False, id="echo-already-off-at-start"),
),
)
def test_timeout_names_an_unanswered_password_prompt(
caplog, capsys, terminal, echo_at_start, prompts, named
):
"""A call that times out on a password prompt says so, and how to go on.
The kill leaves the echo as the child held it, so the timeout reads the
prompt from the terminal the way the watchdog does. The prompt also holds
the terminal line with no line break after it: the call ends that line, so
the warning does not run on from `Password:`. The `✘` trail line of the
attempt names the cause too. The killed child could not turn the echo back
on, so the call does: it leaves the terminal as it found it.
The two controls time out the same way and keep the plain wording: a child
that never touches the echo, and one that finds it off already.
"""
assert not _SUDO_CACHE_WARM.is_set()
manager = FakeManager()
manager.internal_sudo = True
# Two seconds leave the child the time to start and to reach the terminal.
manager.timeout = 2
_set_echo(terminal, echo_at_start)
if prompts:
script = _prompting_script(terminal, "installer asks", 30)
else:
script = _stalling_script("installer asks", 30)
failures: list[Specifier] = []
task = package_task(
manager,
Specifier(raw_spec="mango", package_id="mango"),
threading.Lock(),
action=lambda m, s: m.run_cli("-c", script),
verb="install",
operation="install",
record_failure=failures.append,
)
with (
patch("sys.stderr.isatty", return_value=True),
caplog.at_level(logging.WARNING),
):
succeeded, label = task()
assert succeeded is False
assert len(failures) == 1
message = "Timed out after 2s."
if named:
message = (
"Timed out after 2s on an unanswered password prompt. "
"Run the command again and answer the prompt."
)
assert [error.error for error in manager.cli_errors] == [message]
assert message in [record.getMessage() for record in caplog.records]
assert capsys.readouterr().err.startswith("\n") is named
detail = UNANSWERED_PROMPT_DETAIL if named else None
assert manager.cli_errors[0].detail == detail
assert _echo(terminal) is echo_at_start
assert label.endswith(f"mango ({detail})" if named else "mango")
[docs]
def test_terminal_helpers_answer_nothing_on_windows(monkeypatch):
"""Windows has no `termios`: the terminal reads as no modes.
Nothing is given back to it, and no echo is reported, before any terminal is opened.
The module's `sys` is replaced by a stand-in naming Windows, so the check
runs on every platform. On Windows, `os.ctermid` does not even exist.
"""
monkeypatch.setattr(
"meta_package_manager.sudo.sys", SimpleNamespace(platform="win32")
)
opened: list[None] = []
monkeypatch.setattr(
"meta_package_manager.sudo.os.ctermid",
lambda: opened.append(None),
raising=False,
)
modes = [0, 0, 0, 0, 0, 0, []]
assert _terminal_modes() is None
_restore_terminal_modes(modes)
assert _terminal_echo(modes) is None
assert not opened
[docs]
@pytest.mark.skipif(is_any_windows(), reason="Windows has no terminal modes.")
@pytest.mark.parametrize("echoes", (True, False))
def test_terminal_echo_reads_the_local_flags(echoes):
"""The echo is the `ECHO` bit of the local flags.
The fourth item of what `termios.tcgetattr` returns: a password read clears it, a
line read keeps it.
"""
local_flags = termios.ICANON | (termios.ECHO if echoes else 0)
assert _terminal_echo([0, 0, 0, local_flags, 0, 0, []]) is echoes
[docs]
@pytest.mark.skipif(is_any_windows(), reason="Windows has no terminal modes.")
@pytest.mark.parametrize("present", (True, False), ids=("not-a-terminal", "gone"))
def test_terminal_modes_survive_a_terminal_that_cannot_answer(
monkeypatch, tmp_path, present
):
"""A terminal that cannot answer reads as no modes, and gets none back.
The watchdog only reads the controlling terminal to word a warning, so it must never
fail the call it watches.
A regular file stands in for a device that refuses the terminal requests, and
a missing path for a terminal that went away during the call. The signal mask
blocked around the write comes back as it was.
"""
device = tmp_path / "tty"
if present:
device.write_bytes(b"")
monkeypatch.setattr("meta_package_manager.sudo.os.ctermid", lambda: str(device))
mask = signal.pthread_sigmask(signal.SIG_BLOCK, ())
assert _terminal_modes() is None
# The shape `termios.tcgetattr` returns. The write fails before it reads them.
_restore_terminal_modes([0, 0, 0, termios.ECHO, 0, 0, [b"\x00"] * termios.NCCS])
assert signal.pthread_sigmask(signal.SIG_BLOCK, ()) == mask
[docs]
@pytest.mark.parametrize(
("active_operation", "internal_sudo", "warm_cache", "tty"),
(
pytest.param("installed", True, False, True, id="read-only-operation"),
pytest.param("install", False, False, True, id="no-internal-sudo"),
pytest.param("install", True, True, True, id="warm-cache"),
pytest.param("install", True, False, False, id="off-tty"),
),
)
def test_stall_watchdog_negative_gates(
monkeypatch, caplog, active_operation, internal_sudo, warm_cache, tty
):
"""Each arming gate individually disarms the watchdog.
No notice fires even when the call stays silent for far longer than the (shortened)
delay.
"""
manager = FakeManager()
manager.internal_sudo = internal_sudo
manager._active_operation = active_operation
monkeypatch.setattr("meta_package_manager.sudo._STALL_NOTICE_DELAY", 0.2)
if warm_cache:
_SUDO_CACHE_WARM.set()
try:
with (
patch("sys.stderr.isatty", return_value=tty),
caplog.at_level(logging.WARNING),
):
manager.run_cli("-c", _stalling_script("quiet stretch", 1))
finally:
_SUDO_CACHE_WARM.clear()
assert not any(
"hidden password prompt" in record.getMessage() for record in caplog.records
)
[docs]
@pytest.mark.parametrize(
("internal_sudo", "warm", "resets", "expected_live"),
(
pytest.param(True, False, False, "never", id="armed-call-stays-still"),
pytest.param(False, False, False, "auto", id="ordinary-call-animates"),
pytest.param(True, True, False, "auto", id="warm-cache-serves-the-call"),
pytest.param(True, True, True, "never", id="tool-drops-the-warm-cache"),
),
)
def test_hidden_prompt_risk_holds_the_spinner_still(
internal_sudo, warm, resets, expected_live
):
"""A call that may hide a `sudo` prompt builds its spinner disabled.
This leaves the terminal line the tool writes that prompt on: an animation
repainting it erases the prompt and leaves the run to die at the mutating timeout.
The control differs by the `internal_sudo` gate alone, and keeps `live="auto"`, the
auto-detected animation of an ordinary call.
A warm cache animates the call again, since the tool's own `sudo` spends it
without a prompt. A tool that drops the cache before that `sudo` prompts
whatever the cache holds, so its call stays still.
"""
assert not _SUDO_CACHE_WARM.is_set()
manager = FakeManager()
manager.internal_sudo = internal_sudo
manager.resets_sudo_timestamp = resets
manager._active_operation = "install"
manager.progress = True
built: list[str] = []
class RecordingSpinner(Spinner):
def __init__(self, *args, **kwargs):
built.append(kwargs["live"])
super().__init__(*args, **kwargs)
if warm:
_SUDO_CACHE_WARM.set()
try:
with (
patch("sys.stderr.isatty", return_value=True),
patch("meta_package_manager.execution.Spinner", RecordingSpinner),
):
manager.run_cli("-c", "print('quick call')")
finally:
_SUDO_CACHE_WARM.clear()
assert built == [expected_live]
[docs]
@pytest.mark.parametrize(
("internal_sudo", "progress", "announced"),
(
pytest.param(True, True, True, id="still-call-names-itself"),
pytest.param(False, True, False, id="animated-call-leaves-it-to-the-spinner"),
pytest.param(True, False, False, id="progress-off-stays-silent"),
),
)
def test_still_call_prints_its_label_once(
monkeypatch, capsys, internal_sudo, progress, announced
):
"""A call whose spinner is held still names itself on a static line instead.
Without the line the terminal stays blank for the whole call. The line says
a password may be asked, and ends before the child starts, so a prompt the
tool raises lands on a fresh line below that hint.
"""
assert not _SUDO_CACHE_WARM.is_set()
for envvar in COLOR_ENVVARS:
monkeypatch.delenv(envvar, raising=False)
monkeypatch.setenv("NO_COLOR", "1")
manager = FakeManager()
manager.internal_sudo = internal_sudo
manager._active_operation = "install"
manager.progress = progress
script = "print('quick call')"
with patch("sys.stderr.isatty", return_value=True):
manager.run_cli("-c", script)
label = manager._call_label((str(manager.cli_path), "-c", script))
line = f"{STILL_CALL_MARKER} {label} ({STILL_CALL_HINT})"
assert (f"{line}\n" in capsys.readouterr().err) is announced
[docs]
def test_stall_watchdog_tee_gates_debug_lines_at_default_verbosity(caplog):
"""An armed call still hides child output at the default WARNING level.
The tee forwards lines through the root logger's level gate instead of bypassing it.
"""
manager = FakeManager()
manager.internal_sudo = True
manager._active_operation = "install"
with (
patch("sys.stderr.isatty", return_value=True),
caplog.at_level(logging.WARNING),
):
manager.run_cli("-c", "print('tee gated line')")
assert not any("tee gated line" in record.getMessage() for record in caplog.records)
[docs]
def test_stall_watchdog_tee_forwards_verbatim_at_debug(caplog):
"""At DEBUG, an armed call streams child lines through the tee.
With the same message, level and manager label as an un-teed run.
"""
manager = FakeManager()
manager.internal_sudo = True
manager._active_operation = "install"
with (
patch("sys.stderr.isatty", return_value=True),
caplog.at_level(logging.DEBUG),
):
manager.run_cli("-c", "print('tee forwarded line')")
records = [
record
for record in caplog.records
if record.getMessage() == "tee forwarded line"
]
assert len(records) == 1
assert records[0].levelno == logging.DEBUG
assert records[0].label == manager.subject