# Copyright Kevin Deldycke <kevin@deldycke.com> 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.
"""Privilege-escalation machinery for the mutating fan-outs.
This module owns `sudo` credential priming ({func}`prime_sudo`) and its
background keepalive (`_start_sudo_keepalive`), escalation-policy
resolution (`_resolved_sudo`), sudo-failure detection
(`_is_sudo_auth_failure`), and the hidden-prompt stall watchdog
(`_StallWatchdog`). The execution engine
({mod}`meta_package_manager.execution`) consumes the policy pieces to wrap and
diagnose escalated commands; the CLI calls {func}`prime_sudo` at the top of
each mutating subcommand.
Why priming exists: a concurrent state-changing command mutes per-manager output
and feeds each child `stdin=/dev/null`, so a `sudo` password prompt raised
mid-run (by mpm's own `sudo --non-interactive` or by a manager that escalates
internally, like
Homebrew `cask`) lands invisibly on `/dev/tty` and can stall the run up to the
mutating timeout. Priming first probes the credential cache non-interactively:
found warm, it is silently kept alive for the whole run; found cold on a terminal,
the managers mpm itself escalates get a single up-front password prompt, naming
them and branded `[mpm]`. Internal escalators never prompt up front: their rare
cold-cache escalation is covered by the silent-call stall notice instead, raised
while the hidden prompt can still be answered.
```{note}
Windows reaches only a corner of this module. `gsudo` gives it an escalator to
probe and a cache to warm, and Microsoft's own `sudo.exe` backs it up from
`24H2`, but no manager there escalates by default, so {func}`prime_sudo`
returns on the empty selection unless `--sudo` or a `[mpm.overrides.<id>] sudo =
true` entry asks for it. The internal escalators stay macOS-only managers, so
the watchdog is not armed there either.
Neither reaches an unattended run the way a Unix escalator does. `sudo.exe`
refuses every token whose elevation type is `Default`, which is what a network
logon carries: an SSH session holding a full administrator token at High
integrity is still turned away with `You are not allowed to run sudo`. So a
scripted or remote Windows run escalates through `gsudo` or not at all.
```
```{todo}
Prime Microsoft's `sudo.exe` once there is anything to prime. It caches
nothing, so every escalation of a run raises a UAC dialog of its own, which is
why {data}`ESCALATORS` ranks it behind `gsudo` rather than beside it.
[microsoft/sudo#7](https://github.com/microsoft/sudo/issues/7) is the request
that would give it a cache for {func}`prime_sudo` to warm; `gsudo` has the same
gap open at
[gerardog/gsudo#378](https://github.com/gerardog/gsudo/issues/378) for its own
password path. Nothing else blocks the backend: the command line, the
`--preserve-env` environment and the child's exit code all survive the
elevation, measured on build `26100.1742` from a Medium-integrity shell, the
one path that really brokers it, so
[microsoft/sudo#117](https://github.com/microsoft/sudo/issues/117) never
reaches the shapes this project builds.
Emulate an option a backend cannot express rather than failing on it: topgrade
returns a hard error there, which its users report as a bug
([topgrade-rs/topgrade#1435](https://github.com/topgrade-rs/topgrade/issues/1435)).
```
```{todo}
Escalate to the user owning a manager's tree, not only to root: every
{attr}`~Escalator.escalate_args` reaches root alone, where `sudo --user` and
`doas -u` could reach the owner. The one legitimate case is a multi-user nix
install, whose foreign-owned profiles are a first-class upstream
configuration. A shared Homebrew prefix is not: Homebrew's
[support tiers](https://docs.brew.sh/Support-Tiers) file "Multi-user Homebrew
environments where multiple users share the same installation" as
unsupported, so smoothing that setup over (as topgrade does for its brew
step) would carry a burden upstream itself refuses. Stays unbuilt until a
nix user asks.
```
```{todo}
Rebrand the hidden password prompt of an internal escalator with a
`SUDO_ASKPASS` helper, once the stall notice of `_StallWatchdog` proves
insufficient in the field. It is also the only route serving a hardened
`sudoers` policy, whose timestamps the primed cache cannot reach (see
`_SUDO_CACHE_WARM`). That class records why the helper was rejected, and any
implementation has to answer both of its points: the raw password it handles,
and the tools it never reaches (`brew` honors the variable, `fink`'s plain
`sudo` re-exec does not).
```
"""
from __future__ import annotations
import logging
import os
import re
import shutil
import signal
import subprocess
import sys
import threading
import time
from dataclasses import dataclass, replace
from functools import cache
from typing import Final
from click_extra import echo
from .version import parse_version
# Windows has no `termios`: `_terminal_echo` answers `None` there.
if sys.platform != "win32":
import termios
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Iterable, Mapping
from pathlib import Path
from typing import Any
from click import Context
from .capabilities import Operations
from .execution import CLIExecutor
from .manager import PackageManager
from .version import TokenizedString
UNANSWERED_PROMPT_DETAIL: Final = "no password"
"""How a `✘` trail line words a call that ended on an unanswered password prompt.
{meth}`CLIExecutor.run <meta_package_manager.execution.CLIExecutor.run>` files it
as the {attr}`~meta_package_manager.execution.CLIError.detail` of two failures:
its own timeout, when `_StallWatchdog` still finds the prompt on the terminal,
and the failure of a tool whose own `sudo` gave up on the prompt
(`_is_prompt_timeout`).
"""
_STALL_NOTICE_DELAY: Final = 30
"""Seconds of child silence before an armed stall watchdog raises its notice.
Counted during a call that may hide a `sudo` password prompt
(`_hidden_prompt_risk`). Long enough that ordinary quiet stretches (dependency
resolution, download lulls that still tick progress lines) rarely trip it, yet
far below {data}`~meta_package_manager.execution.MUTATING_TIMEOUT`, so the user
gets the hint while the hidden password prompt can still be answered. See
`_StallWatchdog`.
"""
_SUDO_CACHE_WARM: Final = threading.Event()
"""Set while the priming keepalive believes the credential cache is warm.
Armed by `_start_sudo_keepalive` and cleared when the context closes. A warm
cache serves internal escalations ({attr}`CLIExecutor.internal_sudo
<meta_package_manager.execution.CLIExecutor.internal_sudo>`) silently, as
long as their call stays in the session of the terminal (`_spends_warm_cache`).
So the silent-call stall watchdog skips arming while this flag is set, except
for a manager that drops the cache itself
({attr}`CLIExecutor.resets_sudo_timestamp
<meta_package_manager.execution.CLIExecutor.resets_sudo_timestamp>`).
The keepalive keeps the flag honest mid-run: a refresh that finds the
credentials gone clears it, with one warning per drop, so later spawns arm the
watchdog again, and a refresh that succeeds after a re-authentication sets it
back. Homebrew is the known cache killer: it resets the `sudo` timestamp on
purpose, at the start of every command before `7.0.8` and when a command first
needs `sudo` since (see `docs/sudo.md`).
```{note}
The flag records that `mpm` holds a validated credential, assuming sudo's default
timestamp semantics. Under a hardened sudoers policy (`timestamp_timeout=0`, or a
`timestamp_type` keyed to the process rather than the terminal) a manager's own
child `sudo` may not be able to spend that credential, and its mid-run prompt then
goes unflagged. Priming still authenticates; only the watchdog is suppressed.
```
"""
def _with_pid(args: tuple[str, ...]) -> tuple[str, ...]:
"""`args`, with any `{pid}` token replaced by mpm's own process id."""
pid = str(os.getpid())
return tuple(arg.replace("{pid}", pid) for arg in args)
def _run_probe(args: tuple[str, ...]) -> subprocess.CompletedProcess[str] | None:
"""Ask an escalator binary a question, detached from the terminal.
`stdin` is closed and the probe runs in a session of its own, so it has no
terminal to read a password from. A stand-in for an escalator may prompt,
and the prompt would land in the output captured here, where nobody sees
it. SliTaz's `/usr/bin/sudo` is such a stand-in, a BusyBox `su -c` wrapper:
its root password prompt blocked every command a non-root user ran in a
terminal, before the first manager call.
Returns `None` when the binary cannot run: it vanished between `which()`
and here, or is not executable.
"""
try:
return subprocess.run(
args,
capture_output=True,
check=False,
text=True,
encoding="UTF-8",
stdin=subprocess.DEVNULL,
start_new_session=True,
)
except OSError:
return None
@cache
def _installed_release(version_args: tuple[str, ...]) -> TokenizedString | None:
"""The release `version_args` reports, or `None` when it reports none.
Cached like {func}`resolve_escalator`, since the binary does not change
mid-run.
"""
probe = _run_probe(version_args)
if probe is None or probe.returncode != 0:
return None
match = re.search(r"\d+(?:\.\d+)*", probe.stdout or "")
return parse_version(match.group()) if match else None
[docs]
@dataclass(frozen=True)
class Escalator:
"""One privilege-escalation binary, and the argv dialects mpm drives it with.
Everything specific to an escalator lives here, so the rest of the module
reasons about escalation without naming a binary. The dialects are not
interchangeable: `doas` is not a `sudo` clone with another name, it takes
short options only and has no way to authenticate without running a command.
"""
id: str
"""The binary's name, and the value the `sudo_command` override selects it by."""
escalate_args: tuple[str, ...]
"""Argv prefix escalating a manager command, non-interactively.
Non-interactive on purpose: a prompt raised inside the concurrent fan-out
lands on a terminal nobody is watching (see {func}`prime_sudo`). Emitted by
{meth}`CLIExecutor.build_cli
<meta_package_manager.execution.CLIExecutor.build_cli>` and matched back by
{meth}`CLIExecutor.run <meta_package_manager.execution.CLIExecutor.run>` to
recognize an escalation failure, so the two sites must stay in lockstep.
"""
probe_args: tuple[str, ...]
"""Argv reading the credential cache without ever prompting.
`sudo` answers this without running anything (`--validate`); `doas` has no
such mode, so it runs `true` as the cheapest harmless command.
"""
passwordless_probe_args: tuple[str, ...] | None
"""Argv asking whether *one named command* runs without a password, or `None`.
Completed with the command's path by `_escalation_is_passwordless`.
`sudo --list -- <command>` answers precisely, reporting the command when the
policy grants it unauthenticated and failing under `--non-interactive` when a
password would be wanted. `doas` has no such query and gets `None`.
Needed because {attr}`probe_args` answers a *different* question. `sudo
--validate` refuses whenever any matching `sudoers` entry requires a
password, even when the entry that would actually run the command is tagged
`NOPASSWD`. openSUSE stacks exactly that pair: its stock `ALL ALL=(ALL) ALL`
sits under the `NOPASSWD` rule `docs/sudo.md` recommends, so `--validate`
reports a cold cache on a host where every escalation in fact runs untouched,
and without this probe mpm warns that managers "may fail" before they go on
to succeed.
That is a property of `--validate` rather than of one distribution's policy,
and Fedora reaches it by an unrelated route: its stock `%wheel ALL=(ALL) ALL`
under the same `NOPASSWD` rule leaves `sudo 1.9.17p2` answering `a password
is required` and exiting `1`, while `--list --` exits `0` and a real
`sudo --non-interactive id -u` returns `0`. Two policies with nothing in
common, one false cold cache, which is what this second probe exists to
catch.
"""
prompt_args: tuple[str, ...]
"""Argv authenticating the user up front, interactively, once per run."""
refreshable: bool
"""Whether {attr}`probe_args` can also serve as a keepalive tick.
True only where the escalator caches credentials on a schedule mpm can
reason about. `doas` persistence is opt-in per rule in `doas.conf`, so a
recurring probe would report a drop on every tick for the majority of hosts,
where no rule asks for it. Those escalators get no keepalive thread at all.
"""
brands_prompt: bool
"""Whether the escalator can be told what password prompt to print.
`sudo --prompt` can; `doas` cannot. The notice {func}`prime_sudo` echoes
before authenticating names the managers either way, so an unbranded prompt
loses the `[mpm]` marker, not the explanation.
"""
remedy: str
"""What lets escalated commands run with no password prompt mid-run, as the
sentence closing each warning about a missing authorization.
It names what the host owns: a `sudoers` or `doas.conf` rule, or a polkit
grant. A password typed up front cannot stand in for one where the
escalator keeps no authorization for later commands: `doas` without
`persist`, or polkit under `auth_admin`, which is what `pkexec` gets by
default and `run0` gets over SSH.
"""
binary: str | None = None
"""The name to look for on `PATH`, where it differs from {attr}`id`.
Only Microsoft's `sudo.exe` needs one. It ships under a name the Unix
escalator already claims, and shares none of its dialect: `--non-interactive`
and `--validate` are not options it has. So the two need distinct ids, for
the `sudo_command` override to name either, while
{func}`resolve_escalator` still has to look for the same file name on both
platforms.
"""
identity_args: tuple[str, ...] | None = None
"""Argv proving the binary on `PATH` is this escalator, not a stand-in.
A name on `PATH` is not proof of the dialect behind it: Alpine's
`doas-sudo-shim` installs `/usr/bin/sudo` as a shell script forwarding to
`doas`, and it accepts `--non-interactive` alone out of everything mpm
sends. So {attr}`escalate_args` works there while every probe dies on
`unrecognized option`, and mpm reads that as a cold credential cache on a
host where escalation in fact runs untouched.
Run by {meth}`is_genuine`, which needs the argv to authenticate nothing and
to cost nothing: `sudo --version` reports the build and exits `0` on real
sudo, where the shim rejects the option and exits `1`. `None` where no
stand-in is known, which is every escalator but `sudo`.
"""
identity_markers: tuple[str, ...] | None = None
"""Substrings {attr}`identity_args` prints when the binary is genuine, any
one of which is proof.
Matched against `<stdout>` on a zero exit. Kept beside the argv because
exit status alone is too weak a signal: a stand-in free to accept the
option would pass on the returncode.
Several markers rather than one because a reimplementation brands its own
banner: `sudo --version` prints `Sudo version 1.9.17` on the original and
`sudo-rs 0.2.13` on the Rust rewrite Ubuntu ships as its default `sudo`
since `25.10`. Matching the first alone would reject sudo-rs as a stand-in,
{func}`resolve_escalator` then falling through to `doas` on a host carrying
both and inverting the documented preference. Upstream's `SUDO_RS_VERSION`
override replaces the number and never the `sudo-rs` prefix, so the prefix
is what the second marker keys on.
Both markers are the ones sudo-rs itself sorts the two implementations by:
its test framework reads the same banner, stripping `Sudo version ` to
recognize the original and treating everything else as its own
(`test-framework/sudo-test/src/lib.rs`).
"""
retention_markers: tuple[str, ...] | None = None
"""Substrings a cold {attr}`probe_args` answer prints when a password prompt
would authorize the commands after it too, or `None` where the probe cannot
tell.
Only polkit's `pkcheck` can, for `pkexec` and `run0`: it adds
`polkit\\56retains_authorization_after_challenge=1` to its `<stdout>` under
`auth_admin_keep`, and nothing under `auth_admin`, where a prompt authorizes
its own command alone. {func}`prime_sudo` then skips the prompt, since the
password would buy nothing. Measured on polkit `0.104`, SliTaz 5.0.
"""
retention_probe_args: tuple[str, ...] | None = None
"""Argv asking polkit about {attr}`retention_markers`, where the credential
probe cannot answer, or `None` to read the answer of that probe.
Only `run0` needs one: its probe runs `true` through systemd, while
`pkcheck` asks polkit about `org.freedesktop.systemd1.manage-units`, the
action run0 authorizes against. systemd's policy grants it
`auth_admin_keep` for an active local session and `auth_admin` for an SSH
one. Measured on Fedora 44, systemd `259`, polkit `127`.
"""
probe_command: str | None = None
"""The stand-in command {attr}`probe_args` runs, where the escalator's rules
can match single commands, or `None`.
Only `doas`: `permit nopass tester as root cmd /sbin/apk` denies the
`doas -n true` probe with `Operation not permitted` and still runs
`doas -n /sbin/apk` unprompted, measured on Alpine's doas `6.8.2`. A
denial of the probe then speaks for that command alone, so
{func}`prime_sudo` does not say the managers will fail. `doas -C` would
answer for the managers' own commands, but it reads `doas.conf` as the
caller, and Alpine installs that file mode `640`, owned by `root:root`.
"""
probe_success_markers: tuple[str, ...] | None = None
"""Substrings proving {attr}`probe_args` found escalation ready, for an
escalator whose probe cannot say so through its exit code. Any one is enough.
`None` keeps the exit code as the whole answer, which is what `sudo`, `doas`
and `run0` report through. `gsudo` is the exception: `gsudo status --json`
asks exactly the question {func}`prime_sudo` needs and always exits `0`, so
the answer lives in its output alone. Two of its fields settle it, and
either is sufficient: `IsElevated` because a process already elevated needs
no escalation, and `CacheAvailable` because a warm credentials cache spends
without raising a UAC dialog.
"""
env_forward_template: str | None = None
"""Argument forwarding one environment variable across the escalation, or
`None` for an escalator the variables survive on their own.
Only `run0` needs it. It runs the command in a transient service forked
from the service manager, which inherits nothing from the caller, so the
environment {meth}`CLIExecutor.run
<meta_package_manager.execution.CLIExecutor.run>` forces on a call would
reach `run0` and stop there. That silently costs the managers relying on
one: `nala`, `tazpkg` and `urpmi` pin their parsers against a translated
locale with `LC_ALL=C`, and `ports` keeps out of an interactive dialog with
`BATCH=yes`.
`sudo` and `doas` need no template: both reset the environment too, but
through a `sudoers` policy whose `env_keep` the host owns, which is where
such a decision belongs. `pkexec` clears it down to a fixed list, which
holds `LANG` and every `LC_*` variable, so the `LC_ALL=C` above survives
it and only `BATCH=yes` is lost. Carrying that one would need the
`pkexec env NAME=VALUE ...` idiom rather than a per-variable flag, so it
keeps `None` until a manager on a polkit-only host asks for one.
"""
gated_options: tuple[tuple[str, str], ...] = ()
"""Options following {attr}`escalate_args` on the releases accepting them,
each paired with the first such release.
{meth}`supported_options` reads the installed release through
{attr}`version_args`, and keeps the options it accepts. Only `pkexec` has
one: `--keep-cwd` arrived in polkit `121`
([polkit-org/polkit@af533784a330](https://github.com/polkit-org/polkit/commit/af533784a33040ce91b03744c6fd5ba9aff3e007)),
and an older `pkexec` reads it as the program to run, failing every
escalation on `Cannot run program --keep-cwd`. SliTaz 5.0 ships polkit
`0.104`.
"""
version_args: tuple[str, ...] | None = None
"""Argv printing the installed release, for {attr}`gated_options`, or `None`
where no option is gated."""
@property
def binary_name(self) -> str:
"""The file {func}`resolve_escalator` looks for on `PATH`."""
return self.binary or self.id
[docs]
def forward_env(
self,
args: tuple[str, ...],
env: Mapping[str, str | None] | None,
) -> tuple[str, ...]:
"""Splice {attr}`env_forward_template` for each variable into `args`.
Inserted ahead of the end-of-options separator the prefix closes on, so
a manager's own flags stay shielded from the escalator's parser. Sorted
for a stable command disclosure. Returns `args` untouched for an
escalator declaring no template, or for a call forcing no variable.
A `None` value asks for a variable to be *unset*, which needs carrying
nowhere: the fresh environment on the far side of the escalation never
had it.
"""
if not self.env_forward_template or not env:
return args
forwarded = tuple(
self.env_forward_template.format(name=name, value=value)
for name, value in sorted(env.items())
if value is not None
)
if not forwarded:
return args
cut = len(self.escalate_args)
if self.escalate_args and self.escalate_args[-1] == "--":
cut -= 1
return (*args[:cut], *forwarded, *args[cut:])
[docs]
def probe_says_warm(self, probe: subprocess.CompletedProcess[bytes]) -> bool:
"""Whether the probe found escalation ready to run unprompted."""
if probe.returncode != 0:
return False
if not self.probe_success_markers:
return True
stdout = (probe.stdout or b"").decode("UTF-8", errors="replace")
return any(marker in stdout for marker in self.probe_success_markers)
[docs]
def resolved_probe_args(self) -> tuple[str, ...]:
"""{attr}`probe_args`, with any `{pid}` token replaced by mpm's own id.
Only `pkexec` needs it. Its probe is `pkcheck`, polkit's own
authorization query, which asks about a *subject* rather than about the
caller and refuses to guess one: without `--process` it exits `126` on
`Subject not specified`. Every other escalator answers for whoever runs
it and carries no token, so the substitution is a no-op there.
"""
return _with_pid(self.probe_args)
[docs]
def supported_options(self) -> tuple[str, ...]:
"""The {attr}`gated_options` the installed release accepts.
A release that cannot be read keeps every option. A rejected option
then fails the escalation with the escalator's own error, where
dropping it would change the command without a word.
"""
if not self.gated_options or not self.version_args:
return ()
release = _installed_release(self.version_args)
return tuple(
option
for option, since in self.gated_options
if release is None or release >= parse_version(since)
)
[docs]
def identity(self) -> str:
"""What the identity probe found: `genuine`, `stand-in` or `unrunnable`.
`genuine` when the escalator declares no {attr}`identity_args`, so an
escalator opts into the check rather than out of it. `unrunnable` keeps
a binary that cannot run apart from one that runs and answers as
something else, which `mpm doctor` reports differently.
"""
if not self.identity_args:
return "genuine"
probe = _run_probe(self.identity_args)
if probe is None:
return "unrunnable"
# `stdout` is `None` whenever the output was not captured, so it is
# normalized rather than trusted to be a string.
stdout = probe.stdout or ""
if probe.returncode == 0 and any(
marker in stdout for marker in self.identity_markers or ()
):
return "genuine"
return "stand-in"
[docs]
def is_genuine(self) -> bool:
"""Whether the binary on `PATH` really is this escalator."""
return self.identity() == "genuine"
ESCALATORS: Final[tuple[Escalator, ...]] = (
Escalator(
id="sudo",
escalate_args=("sudo", "--non-interactive"),
probe_args=("sudo", "--non-interactive", "--validate"),
passwordless_probe_args=("sudo", "--non-interactive", "--list", "--"),
prompt_args=("sudo", "--validate"),
refreshable=True,
brands_prompt=True,
remedy=(
"Allow the escalated commands without a password with a `NOPASSWD` "
"rule, or authenticate first with `sudo --validate` in the same "
"terminal."
),
identity_args=("sudo", "--version"),
identity_markers=("Sudo version", "sudo-rs"),
),
Escalator(
id="doas",
# `doas` parses short options only, so the long-form convention of
# `docs/add-new-manager.md` cannot apply here: `-n` is the whole
# vocabulary for "do not prompt".
escalate_args=("doas", "-n"),
probe_args=("doas", "-n", "true"),
passwordless_probe_args=None,
prompt_args=("doas", "true"),
refreshable=False,
brands_prompt=False,
remedy=(
"Add `persist` to the matching `doas.conf` rule to keep the "
"authorization, or `nopass` to need no password."
),
probe_command="true",
),
Escalator(
id="run0",
# `--pipe` passes the caller's file descriptors straight through rather
# than allocating a pseudo TTY, which is what keeps a captured listing
# byte-clean and its exit code intact. run0 picks that mode on its own
# when no descriptor is a TTY, so the switch only pins what it would
# infer. The trailing `--` shields a manager's own flags from run0's
# getopt, which would otherwise claim any it recognizes.
escalate_args=("run0", "--pipe", "--no-ask-password", "--"),
# No `--validate` before systemd 262, so the cheapest harmless command
# stands in for one, exactly as it does for `doas`.
probe_args=("run0", "--pipe", "--no-ask-password", "true"),
# A bare `--setenv=NAME` would import the caller's value; the pair is
# spelled out because mpm forces values the caller may not carry.
env_forward_template="--setenv={name}={value}",
passwordless_probe_args=None,
prompt_args=("run0", "--pipe", "true"),
# polkit owns the authorization and retains it per session, so there is
# no timestamp for mpm to extend: `-v` creates one and `-k` revokes it,
# and nothing refreshes one.
refreshable=False,
# The password prompt belongs to whichever polkit agent answers, built
# from the action's own message. run0 has no `--prompt`, and
# systemd/systemd#33902 asks for control over that text.
brands_prompt=False,
# systemd's own policy keeps the authorization for an active local
# session only (`auth_admin_keep`): an SSH session gets `auth_admin`.
remedy="Grant `org.freedesktop.systemd1.manage-units` with a polkit rule.",
retention_markers=("retains_authorization_after_challenge=1",),
retention_probe_args=(
"pkcheck",
"--action-id",
"org.freedesktop.systemd1.manage-units",
"--process",
"{pid}",
),
identity_args=("run0", "--version"),
# run0 reports the systemd version it ships with, not one of its own.
identity_markers=("systemd",),
),
Escalator(
id="pkexec",
# No `--` separator: pkexec stops parsing at the first non-option and
# would try to execute `--` itself. Nothing is at risk without one,
# since the first argument mpm appends is the manager's absolute path,
# which is already a non-option. `--disable-internal-agent` makes it
# fail on `No authentication agent found` where it would otherwise
# raise its textual agent on the terminal, inside the fan-out.
# `--keep-cwd` follows where the release accepts it (see
# `gated_options`), and holds the working directory, which pkexec
# otherwise resets to the target user's home.
escalate_args=("pkexec", "--disable-internal-agent"),
# pkexec carries no validate mode: it always executes a program, and
# asking it anything either prompts or dies for want of an agent.
# `pkcheck` is polkit's own query tool and the only way to read the
# answer without doing either, reporting `0` when the action is
# authorized and `2` when it is not. Verified on
# polkit `127`, where it answers `2` and `polkit\56result=auth_admin`
# for a `wheel` user over SSH, that session being remote.
#
# A misspelled action id exits `127` rather than `2`, which this probe
# reads as a cold cache like any other non-zero: the id above is a
# polkit constant and cannot drift, but a future one is worth spelling
# against `pkcheck` directly, since nothing here would report a typo.
# `pkexec` itself also exits `127` when authentication fails and `126`
# when the prompt is dismissed, both colliding with the shell's own
# meanings for those codes, so its failures are told apart by the
# wording `_is_sudo_auth_failure` matches and never by exit code.
#
# A host carrying polkit with no authority reachable answers `126` and
# `Error getting authority`, a fourth code and the one `pkexec` also
# spends on a dismissed prompt. Only non-zero is read here, so the
# collision costs nothing. Seen on an OpenRC Gentoo guest running
# polkit version `126`, where the package ships a systemd unit alone
# and `sys-apps/dbus` was absent, leaving nothing to activate
# `polkitd`: `pkexec` is on `PATH` and passes `is_genuine` there while
# being unable to escalate at all.
probe_args=(
"pkcheck",
"--action-id",
"org.freedesktop.policykit.exec",
"--process",
"{pid}",
),
passwordless_probe_args=None,
# `true` ignores its working directory, so the prompt needs no
# `--keep-cwd` on any release.
prompt_args=("pkexec", "true"),
# polkit owns retention, and the action pkexec defaults to is
# `auth_admin` rather than `auth_admin_keep`, so nothing is kept at all
# unless the host says otherwise.
refreshable=False,
brands_prompt=False,
remedy=(
"Grant `org.freedesktop.policykit.exec` with a polkit rule, or with a "
"`.pkla` file before polkit `0.106`."
),
retention_markers=("retains_authorization_after_challenge=1",),
gated_options=(("--keep-cwd", "121"),),
# `pkexec version 0.104` on SliTaz 5.0, `pkexec version 127` on Arch.
version_args=("pkexec", "--version"),
),
Escalator(
id="gsudo",
# Options must precede the command: gsudo claims any leading `-...`
# token as its own and takes no `--` separator, which is why a manager's
# flags stay behind its absolute path. `-n` is deliberately absent here,
# meaning *new window* rather than non-interactive.
escalate_args=("gsudo",),
# `status` never prompts and never fails, so the answer is in the JSON
# rather than the exit code. See `probe_success_markers`.
probe_args=("gsudo", "status", "--json"),
probe_success_markers=('"IsElevated":true', '"CacheAvailable":true'),
passwordless_probe_args=None,
# Warming the cache is the one moment a UAC dialog belongs: answered
# once up front, every later elevation of the run spends it silently.
# `--pid 0` widens the cache past the process that opened it, which is
# what lets a later manager subprocess use it at all.
prompt_args=("gsudo", "cache", "on", "--pid", "0"),
# The cache expires on idle rather than on a timestamp mpm can extend,
# and re-running `cache on` against a cold one raises a dialog.
refreshable=False,
# The prompt is the UAC consent dialog, which Windows words itself.
brands_prompt=False,
remedy="Warm its cache first with `gsudo cache on --pid 0`.",
identity_args=("gsudo", "--version"),
# `gsudo v2.6.1 (Branch...)`, measured on Windows 11 21H2.
identity_markers=("gsudo",),
),
Escalator(
id="win-sudo",
binary="sudo",
# `--inline` overrides the `forceNewWindow` default, which runs the
# command in a window of its own and leaves nothing to capture. No `--`
# separator: the first argument mpm appends is the manager's absolute
# path, already a non-option, exactly as for `pkexec`.
#
# `--preserve-env` is the only environment forwarding this escalator
# offers, and it is all-or-nothing rather than per-variable, so it sits
# in the prefix instead of an `env_forward_template`: the variables
# `CLIExecutor.run` forces are already in the environment it spawns
# `sudo` with, and this is what carries them across the elevation.
# Without it the escalated child loses the `LC_ALL=C` a parser is pinned
# against.
escalate_args=("sudo", "--inline", "--preserve-env"),
# There is no credential cache to read: every escalation raises its own
# UAC dialog. So the probe answers the only question mpm can act on
# ahead of time, which is whether an escalation would be allowed to run
# at all, and `sudo config` cannot serve: it exits `0` and reports the
# mode even from a session the elevation gate goes on to refuse, and it
# words its answer in the display language.
#
# The registry value it reflects is neither localized nor ambiguous.
# `0x3` is inline mode, the only one mpm can drive: `0x0` is disabled,
# `0x1` opens a new window and `0x2` closes the child's input, and
# `--inline` exits with an error against any of them rather than
# falling back. A missing key or value exits non-zero, which reads as
# cold like any other.
probe_args=(
"reg",
"query",
r"HKLM\SOFTWARE\Microsoft\Windows\CurrentVersion\Sudo",
"/v",
"Enabled",
),
probe_success_markers=("0x3",),
passwordless_probe_args=None,
# Nothing to authenticate up front, so this only ever reports the mode.
# It is reached solely when the probe found a mode mpm cannot drive, and
# no argv fixes that: enabling inline mode is the user's call to make in
# Settings, not one mpm may take on their behalf.
prompt_args=("sudo", "config"),
refreshable=False,
# The prompt is the UAC consent dialog, which Windows words itself.
brands_prompt=False,
remedy=(
"Set Windows `sudo` to inline mode in Settings, or install `gsudo`, "
"whose cache one UAC dialog warms for the whole run."
),
# `--version` prints a bare `sudo 1.0.0`, which a reimplementation
# shipping its own `1.x` under this name would also match. The help
# banner names the tool outright and has no version in it to drift.
identity_args=("sudo", "--help"),
identity_markers=("Sudo for Windows",),
),
)
"""Every escalator mpm can drive, in the order it prefers them.
`sudo` comes first so a host carrying both keeps the behavior it has today, and
the `sudo_command` override exists for the user who wants the other one. The
order only decides auto-detection: an explicit override always wins.
`run0` comes after them for the same reason, one step further: it needs a
running polkit to authorize anything, so a host carrying a working `sudo` or
`doas` keeps it, and run0 answers for the systemd hosts that ship neither.
`win-sudo` closes the list, behind `gsudo`, and the two are the whole of
Windows. Microsoft's is inbox from `24H2` where `gsudo` has to be installed,
which is exactly why it ranks second: it caches nothing, so each escalation of
a run raises its own UAC dialog, where one answered `gsudo cache on` covers
every later one. It is the fallback for a host carrying no `gsudo`, not the
default for a host carrying both.
`pkexec` sits before them, and auto-detection rarely reaches it: it ships
wherever polkit does, which is nearly every desktop Linux, and those carry
`sudo` too. SliTaz is the exception, its `sudo` being a `su -c` wrapper that
fails the identity probe. It is also there for `--sudo-command pkexec`, and
it only works where a polkit rule already grants
`org.freedesktop.policykit.exec`, since it cannot escalate without prompting.
The probe is what keeps that honest: a host without the rule reports a cold
cache and its managers decline to run, rather than each of them stopping on a
prompt inside the fan-out.
"""
_SUDO_KEEPALIVE_INTERVAL: Final = 60
"""Seconds between `sudo --non-interactive --validate` credential-cache
refreshes during a run.
Comfortably under sudo's default `timestamp_timeout` (5 minutes), so the cache warmed
by {func}`prime_sudo` stays valid for the whole command. A host configured with a
shorter `timestamp_timeout` may still see a mid-run escalation re-prompt or fail.
"""
_SUDO_PRIMED: Final = "mpm_sudo_primed"
"""`ctx.meta` key marking that {func}`prime_sudo` already ran this invocation."""
[docs]
@cache
def resolve_escalator(override: str | None = None) -> Escalator | None:
"""The escalator mpm drives, or `None` when the host carries none.
With no `override`, returns the first entry of {data}`ESCALATORS` whose
binary is on `PATH` and passes {meth}`~Escalator.is_genuine`, so a host
without `sudo` still escalates through whatever it does have, and one whose
`sudo` is a stand-in for another escalator drives that other one directly.
An `override` names one by its {attr}`~Escalator.id` and is honored even
when the binary is missing, so the failure names the escalator the user
asked for instead of silently falling back to another one.
Cached on the override, since `PATH` does not move mid-run and the identity
probe costs a subprocess. Tests changing what is installed must call
`resolve_escalator.cache_clear()`.
"""
if override is not None:
for escalator in ESCALATORS:
if escalator.id == override:
# Honored all the same, but a stand-in answers differently:
# SliTaz's `sudo` wraps `su -c` and exits 0 whatever happened,
# and Alpine's `doas-sudo-shim` rejects every probe.
if shutil.which(escalator.binary_name) and not escalator.is_genuine():
logging.warning(
f"The {escalator.binary_name} on PATH does not identify "
f"as {escalator.id}: mpm may misread its answers.",
)
return escalator
# An unknown name is a configuration error, not a reason to escalate
# through something the user did not ask for.
logging.warning(
f"Unknown sudo_command {override!r}: expected one of "
f"{', '.join(e.id for e in ESCALATORS)}. Managers needing root may fail.",
)
return None
installed = tuple(e for e in ESCALATORS if shutil.which(e.binary_name))
for escalator in installed:
if escalator.is_genuine():
return escalator
# Every installed escalator failed its identity probe. Falling back to the
# first one still beats reporting none: a stand-in that answers no probe
# usually still escalates, and Alpine's `doas-sudo-shim` only reaches this
# line on a host whose `doas` was removed from under it.
if installed:
logging.debug(
f"{installed[0].id} does not identify as itself: driving it anyway.",
)
return installed[0]
return None
class _EscalationChoice:
"""Which escalator this process drives, resolved once per invocation.
Escalation is a property of the machine, not of a manager: every manager
that escalates on a given host escalates through the same binary. Keeping
the choice here rather than on each manager is what makes that a statement
the code makes rather than one a reader has to infer from every copy
holding the same value.
"""
def __init__(self) -> None:
self._override: str | None = None
def select(self, override: str | None) -> None:
"""Record the user's `--sudo-command`, or `None` to auto-detect.
Called once at the top of the CLI group. It always assigns, so a
previous in-process invocation (the test suite drives the CLI
repeatedly) cannot leak its choice into this one.
"""
self._override = override
def resolve(self) -> Escalator | None:
"""The escalator to drive, or `None` when the host carries none."""
return resolve_escalator(self._override)
@property
def chosen(self) -> bool:
"""Whether the user named the escalator, instead of letting mpm detect it."""
return self._override is not None
ESCALATION: Final = _EscalationChoice()
"""Process-wide escalator selection.
A module-level singleton for the same reason as
{data}`~meta_package_manager.execution.PLAN_RECORDER`:
{meth}`CLIExecutor.build_cli
<meta_package_manager.execution.CLIExecutor.build_cli>` needs it from the
fan-out's worker threads, where the click context is not reliably reachable.
"""
def _resolved_sudo(manager: CLIExecutor) -> bool:
"""Whether `manager` escalates: its
{attr}`~meta_package_manager.execution.CLIExecutor.sudo` override if set,
else its built-in
{attr}`~meta_package_manager.execution.CLIExecutor.default_sudo`."""
return manager.sudo if manager.sudo is not None else manager.default_sudo
def _running_as_root() -> bool:
"""Whether `mpm` runs as root, where no command needs an escalator.
Windows has no `os.geteuid`, and never counts as root.
"""
return getattr(os, "geteuid", lambda: 1)() == 0
def _escalates_in(manager: CLIExecutor, operations: frozenset[str]) -> bool:
"""Whether `manager` escalates a command of a run over `operations`.
`operations` holds the names of the
{class}`~meta_package_manager.capabilities.Operations` the run can reach.
The policy must escalate (`_resolved_sudo`), and one of those
operations must be in its
{attr}`~meta_package_manager.execution.CLIExecutor.privileged_operations`.
An internal escalator counts on its policy alone: `mpm` cannot see which of
its commands call `sudo`, and `fink` does it on `sync` and `cleanup` too. A
forced `sudo = true` on one keeps it in the up-front prompt of every run (see
{attr}`~meta_package_manager.execution.CLIExecutor.internal_sudo`).
A manager that drops the cached credentials itself
({attr}`~meta_package_manager.execution.CLIExecutor.resets_sudo_timestamp`)
is the exception. A password typed up front cannot serve its own `sudo`, so
a forced policy leaves it out, and {func}`prime_sudo` says so.
"""
if not _resolved_sudo(manager):
return False
if manager.internal_sudo:
return not manager.resets_sudo_timestamp
return not operations.isdisjoint(manager.privileged_operations)
def _names_an_escalator(error: str) -> bool:
"""Whether `error` is prefixed by one of the escalators mpm drives.
Both matchers below key on wordings plain enough to appear in an unrelated
command's output (`doas` reports an unauthorized user as the bare errno
string `Operation not permitted`), so they only trust a line the escalator
signed with its own name.
"""
return any(f"{escalator.id}:" in error for escalator in ESCALATORS)
def _is_sudo_auth_failure(error: str) -> bool:
"""Whether the escalator is refusing to authenticate non-interactively.
`sudo --non-interactive` and `doas -n` write one of these to `<stderr>` when
they have no cached credentials
and cannot prompt for a password (nothing cached, no controlling terminal, no
askpass helper). Lets {meth}`CLIExecutor.run
<meta_package_manager.execution.CLIExecutor.run>` turn an opaque escalation
failure into an actionable hint.
The wordings are not interchangeable across implementations: `sudo-rs`, the
Rust rewrite Ubuntu ships as the default `sudo` since `25.10`, answers
`sudo: interactive authentication is required` where the original says
`sudo: a password is required`. Matching only the latter would leave every
escalation failure on a current Ubuntu unrecognized, and the hint unprinted.
`run0` needs its own branch. It hands the refusal to systemd's bus layer,
which signs it with neither `run0` nor any sudo wording, so the guard below
would drop it. The prefix is stable across both refusal paths: polkit
denying a `--no-ask-password` call answers `... Access denied as the
requested operation requires interactive authentication`, and a denied
interactive one answers the same `Access denied` alone. Both read as an
authentication failure rather than a denial, which is the conservative
call: a prompt may still authorize, and a user polkit grants nothing simply
fails it, the same prompt-then-fail path a `sudo` hiding its denial takes.
"""
lowered = error.lower()
# The polkit-brokered pair sign nothing with their own name, so each is
# matched on a phrase distinctive enough to stand without the guard below:
# systemd's bus layer words run0's refusal, while pkexec's comes from the
# authentication agent it failed to raise, or from the prompt a user
# dismissed. A dismissal is an authentication failure and not a denial,
# since answering the next one would authorize the call.
if any(
marker in lowered
for marker in (
"failed to start transient service unit: access denied",
"error creating textual authentication agent",
"error executing command as another user: request dismissed",
"error executing command as another user: no authentication agent",
)
):
return True
return _names_an_escalator(lowered) and any(
marker in lowered
for marker in (
"a password is required",
"a terminal is required",
"interactive authentication is required",
"no tty present",
"askpass",
# Three `doas` wordings, identical in OpenBSD's own `doas.c` and in
# opendoas: `-n` meeting a rule without `nopass` answers
# `Authentication required`, a password prompt with no terminal to
# read from answers `a tty is required` (where sudo says *terminal*,
# already matched above), and a rejected password answers
# `Authentication failed`.
"authentication required",
"a tty is required",
"authentication failed",
)
)
def _is_prompt_timeout(error: str) -> bool:
"""Whether a `sudo` gave up on a password prompt nobody answered.
A tool that escalates on its own relays the line when its `sudo` stops
waiting: `sudo: timed out reading password` from the original, whose
prompt lasts [5 minutes by
default](https://github.com/sudo-project/sudo/blob/f2f340938e51f9f12b29b72507d9edca8cb98abc/configure.ac#L198),
and `sudo: timed out` from `sudo-rs`, which words it [as its PAM
error](https://github.com/trifectatechfoundation/sudo-rs/blob/c42055954861eb6cba42b5286682689fb15d8f30/src/pam/error.rs#L241).
The stock `sudo` of macOS has no prompt timeout, so it never prints it: see
`_StallWatchdog.prompt_unanswered` for the path its prompt takes.
"""
return "sudo: timed out" in error.lower()
def _is_sudo_denied(error: str) -> bool:
"""Whether `error` is `sudo` reporting the user is not authorized to run it.
Distinct from `_is_sudo_auth_failure`: an authentication failure means
the cache is cold and a password would unblock, while a denial means the
`sudoers` policy grants this user nothing, so a prompt could only collect a
password `sudo` then rejects. {func}`prime_sudo` skips its up-front prompt on
a denial.
The detection is opportunistic: some `sudo` configurations reveal the denial
to a non-interactive `--validate` while others hide it behind `a password is
required` (authenticating before disclosing authorization), and the hidden
case simply keeps today's prompt-then-fail path. The markers cover, in
order: one `sudo` wording (`is not allowed to run sudo on`, per its message
catalog), a second that `sudo-rs` also uses for its *list* denial (`may not
run sudo on`, `src/common/error.rs`), the historic sudoers lecture (`is not
in the sudoers file`), and both implementations' per-command denial (`is not
allowed to execute`), and `doas`, which reports the bare errno string of
`EPERM` instead.
Neither `sudo` wording supersedes the other, so both are matched rather than
one being kept for old releases: `1.9.17p2` on Fedora 44 answers a
non-sudoer's `--validate` with `Sorry, user plain may not run sudo on
localhost`, which reading that phrasing as the pre-`1.9` one would drop.
`sudo-rs` denies a `--validate` differently from a `--list`, which matters
because {attr}`~meta_package_manager.sudo.Escalator.probe_args` runs the former first: a user no
`sudoers` rule matches gets `I'm sorry {user}. I'm afraid I can't do that`
(`Error::Authorization`) where `--list` says `may not run sudo`. Matching
only the list wording would leave a non-sudoer on Ubuntu `25.10` and later
being prompted for a password that could never authorize them.
"""
lowered = error.lower()
if any(
marker in lowered
for marker in (
# Left unanchored on the subject so it covers both the third-person
# `<user> is not allowed to run sudo` of the Unix message catalog
# and the second-person `You are not allowed to run sudo` Microsoft's
# `sudo.exe` prints, which is its whole diagnostic: it names neither
# itself nor a reason.
"not allowed to run sudo",
"may not run sudo",
"is not in the sudoers file",
"is not allowed to execute",
# pkexec, where polkit answered rather than the user: a prompt
# cannot change this one, unlike the dismissal it words otherwise.
"error executing command as another user: not authorized",
)
):
return True
# Two opendoas answers, both measured on a runner and both leaving nothing
# for a password to fix. An unmatched rule ends in `errc(1, EPERM, NULL)`,
# printing the bare errno string of EPERM and nothing more specific; a
# missing `doas.conf` ends in `err(1, "doas is not enabled, %s")`, which is
# what an installed-but-unconfigured host answers (`doas.c`). The errno
# string is one any command could print, so both count only on a line doas
# signed. The sudo wordings above need no such guard: each already names
# sudo itself, and sudo prefixes none of them.
#
# `sudo-rs`'s validate denial names no tool either, quoting HAL 9000
# instead, so it joins the guarded group and rides the `sudo: ` prefix its
# own diagnostic path writes.
return _names_an_escalator(lowered) and (
"operation not permitted" in lowered
or "is not enabled" in lowered
or "afraid i can't do that" in lowered
)
def _is_permission_failure(error: str) -> bool:
"""Whether `error` reads as a filesystem permission refusal.
Consumed by the failure gate of {meth}`CLIExecutor.run
<meta_package_manager.execution.CLIExecutor.run>` to recognize a dormant
privileged marker ({attr}`CLIExecutor._dormant_sudo
<meta_package_manager.execution.CLIExecutor._dormant_sudo>`) meeting the
root-owned tree it exists for. No `stat` of mpm's own is needed: the tool
already performed the probe, and its own message usually names the very
directory. The markers cover the wordings of the managers carrying dormant
markers today: npm (`EACCES: permission denied, access
'/usr/local/lib/node_modules'`), gem (`You don't have write permissions
for the /Library/Ruby/Gems/3.4.0 directory`), and pip and cpan (plain
`Permission denied` from the interpreter and the shell).
"""
lowered = error.lower()
return any(
marker in lowered
for marker in (
"permission denied",
"write permissions",
"eacces",
)
)
[docs]
@dataclass(frozen=True)
class InstallRoot:
"""Ownership snapshot of the tree a manager's global installs write into.
Built by {func}`inspect_install_root` from a manager's own
{attr}`~meta_package_manager.manager.PackageManager.install_root` probe.
Diagnosis only, by decision: deciding escalation from this snapshot was
assessed and rejected, since silently running a manager as root on
filesystem evidence nobody reviewed is a posture change no diagnostic
payoff justifies. The scoped `sudo = true` override stays the one road to
escalating a dormant marker, and the failure-gate hint
(`_is_permission_failure`) is what names it.
"""
path: Path
"""The install root itself."""
owner_uid: int
"""Numeric owner of the root directory."""
owner_name: str
"""The owner's account name, or the bare uid when no account matches."""
[docs]
def inspect_install_root(manager: PackageManager) -> InstallRoot | None:
"""Resolve and stat `manager`'s install root, or `None` when unknowable.
`None` covers every dead end: a non-POSIX host (the ownership model does
not apply, and `pwd` does not exist), a manager with no discovery verb, a
probe that fails, and a resolved path that does not exist. The probe shells
out, so failures of any kind are swallowed: this is diagnosis, and it must
never break the command it decorates.
"""
if not hasattr(os, "getuid"):
return None
try:
path = manager.install_root
except Exception: # noqa: BLE001
# The probe runs the manager's own CLI: any failure means "unknown".
return None
if path is None:
return None
try:
uid = path.stat().st_uid
except OSError:
return None
# Deferred on purpose: `pwd` does not exist on Windows, returned above.
import pwd
try:
owner_name = pwd.getpwuid(uid).pw_name
except KeyError:
owner_name = str(uid)
return InstallRoot(path=path, owner_uid=uid, owner_name=owner_name)
def _probe_credentials(
escalator: Escalator,
) -> subprocess.CompletedProcess[bytes] | None:
"""Run the credential probe of `escalator`, or `None` when it cannot run.
Unlike `_run_probe`, it keeps the controlling terminal, since `sudo` keys
its credential cache on it. Only `stdin` is closed, so a stand-in escalator
that asks for a password reads nothing and fails instead of waiting.
"""
try:
return subprocess.run(
escalator.resolved_probe_args(),
capture_output=True,
check=False,
stdin=subprocess.DEVNULL,
)
except OSError:
# Not on PATH (FileNotFoundError), or one that cannot be run: not
# executable for this user (PermissionError), not a valid binary.
return None
def _prompt_kept(
escalator: Escalator,
probe: subprocess.CompletedProcess[bytes],
) -> bool | None:
"""Whether a password prompt would authorize the commands after it, read
from polkit ahead of any prompt, or `None` where nothing tells.
`probe` is the cold answer of the credential probe, which carries the
answer for `pkexec`. A `pkcheck` that cannot run leaves the question open.
"""
if escalator.retention_markers is None:
return None
if escalator.retention_probe_args:
answer = _run_probe(_with_pid(escalator.retention_probe_args))
if answer is None:
return None
output = answer.stdout or ""
else:
output = (probe.stdout or b"").decode("UTF-8", errors="replace")
return any(marker in output for marker in escalator.retention_markers)
def _start_sudo_keepalive(ctx: Context, escalator: Escalator) -> None:
"""Keep the credential cache of `escalator` fresh for the rest of the invocation.
Marks the cache warm whatever the escalator, but only spawns the refreshing
thread for a {attr}`~Escalator.refreshable` one: an escalator whose
persistence is opt-in per rule would report a drop on every tick of a host
that never asked for it, which is noise rather than news. The flag can then
go stale, exactly as its own docstring already admits for a hardened
`sudoers` policy.
Refreshes the cache every {data}`_SUDO_KEEPALIVE_INTERVAL` seconds so a long
fan-out does not outlast sudo's timestamp and re-prompt mid-flight. Output is
captured so a failed refresh cannot smear the aggregate spinner drawing on
stderr. Sets {data}`_SUDO_CACHE_WARM` for the run, and keeps it honest: a
refresh finding the credentials gone (a manager reset them, as Homebrew does
on purpose, or a strict `sudoers` policy expired them) clears the flag and
warns, once per drop, so the stall watchdog re-arms for later spawns; a
refresh succeeding again, after the user re-authenticates in this terminal,
sets it back. The refresh never re-prompts by design: only a command that
needs the credentials may. The daemon thread is stopped and the flag cleared
when the context closes (normal exit or Ctrl+C both run close callbacks).
"""
stop = threading.Event()
def keepalive() -> None:
while not stop.wait(_SUDO_KEEPALIVE_INTERVAL):
refresh = _probe_credentials(escalator)
# A refresh racing the teardown must not touch the flag the
# teardown just cleared.
if stop.is_set():
break
if refresh is not None and escalator.probe_says_warm(refresh):
if not _SUDO_CACHE_WARM.is_set():
logging.info("The sudo credentials are warm again.")
else:
logging.debug("Refreshed the sudo credentials.")
_SUDO_CACHE_WARM.set()
elif _SUDO_CACHE_WARM.is_set():
_SUDO_CACHE_WARM.clear()
logging.warning(
"The sudo credentials primed for this run are gone: a "
"manager reset them (Homebrew does, on purpose) or the "
"sudoers policy expired them. Managers needing root may "
"prompt or fail.",
)
else:
logging.debug("The sudo credentials are still gone.")
_SUDO_CACHE_WARM.set()
if not escalator.refreshable:
logging.info(
f"{escalator.id} credentials cannot be refreshed on a schedule: "
"they hold for as long as its own persistence rules say.",
)
ctx.call_on_close(_SUDO_CACHE_WARM.clear)
return
logging.info(f"Keeping the {escalator.id} credentials fresh for the whole run.")
thread = threading.Thread(target=keepalive, daemon=True)
thread.start()
def teardown() -> None:
stop.set()
# The thread leaves its wait promptly once stopped; join it so a
# refresh still in flight cannot set the flag back after the clear
# below. The timeout bounds a Ctrl+C exit if sudo itself wedges.
thread.join(timeout=2)
_SUDO_CACHE_WARM.clear()
ctx.call_on_close(teardown)
def _escalation_is_passwordless(
escalator: Escalator,
managers: Iterable[PackageManager],
) -> bool:
"""Whether every command mpm escalates already runs without a password.
Asked only after {attr}`~meta_package_manager.sudo.Escalator.probe_args` reported a cold cache, and
answering the question that probe cannot: `sudo --validate` refuses while any
matching `sudoers` entry wants a password, where
{attr}`~Escalator.passwordless_probe_args` names one command and reports the
rule that would actually run it. A host whose policy grants every escalated
command unauthenticated therefore needs no prompt, no warning and no
keepalive, having no credential to keep.
Conservative on every uncertainty: an escalator with no such query, a manager
whose binary was not found, or a single command the policy does not clear
unauthenticated all answer `False` and leave the cold-cache path to handle it.
"""
if escalator.passwordless_probe_args is None:
return False
cli_paths: set[Path] = set()
for manager in managers:
if not _resolved_sudo(manager):
continue
if manager.cli_path is None:
return False
cli_paths.add(manager.cli_path)
if not cli_paths:
return False
for cli_path in sorted(cli_paths):
probe_cli = (*escalator.passwordless_probe_args, str(cli_path))
logging.debug(f"Probe the {escalator.id} policy: {' '.join(probe_cli)}")
try:
probe = subprocess.run(
probe_cli,
capture_output=True,
check=False,
stdin=subprocess.DEVNULL,
)
except OSError:
return False
if probe.returncode != 0:
return False
return True
[docs]
def prime_sudo(
ctx: Context,
managers: Iterable[PackageManager],
*,
operations: Iterable[Operations],
) -> None:
"""Warm the `sudo` credential cache, up front, for a mutating fan-out.
Probes the cache non-interactively (`sudo --non-interactive --validate`)
before considering any
prompt. A warm cache (pre-authenticated `sudo --validate`, a `NOPASSWD` rule, a
recent run) is silently kept fresh for the whole invocation by
`_start_sudo_keepalive`, so every later escalation on the same
terminal, mpm's own `sudo --non-interactive` as well as a manager's internal
`sudo`
({attr}`CLIExecutor.internal_sudo
<meta_package_manager.execution.CLIExecutor.internal_sudo>`), spends the cache
instead of blocking on an invisible prompt inside the concurrent fan-out. Only
a cold cache, on an interactive terminal, with managers that mpm itself
escalates in this run (`_escalates_in`), triggers the interactive path: a
notice naming the managers and the subcommand, then a single branded `sudo`
password prompt.
`operations` lists every operation the run of the subcommand can reach, the
reads included. A manager escalates in this run only when one of them is in
its {attr}`~meta_package_manager.execution.CLIExecutor.privileged_operations`,
so no password is asked for a manager whose commands stay unprivileged.
Before probing, the binary of each manager that escalates in this run is
audited with the same tamper test the config loader applies
({func}`~meta_package_manager.config.config_file_is_trusted`): a binary that
others can modify, handed to `sudo`, runs their code as root, so it draws one
warning per manager (see `docs/security.md`).
Call at the top of each mutating subcommand, before the fan-out draws its
spinner. Never prompts when:
- the process is already root,
- no selected manager escalates a command of this run, through mpm or
internally,
- a dry run or a plan run (no state-changing CLI is executed),
- already primed once this invocation (idempotent),
- a forced policy names a manager that drops the cached credentials itself:
one warning says the policy does nothing for it, and the manager stays
out of the prompt,
- no escalator is on `PATH`, or `sudo_command` names an unknown one (one
warning is logged),
- the probe finds the cache already warm (keepalive only, fully silent),
- the probe reports the user is not authorized to run `sudo` at all
(`_is_sudo_denied`): one warning names the managers mpm escalates and
the remedy, since a prompt could only collect a password `sudo` then
rejects, while an internal-only selection stays silent,
- no interactive terminal is available: one warning names the managers mpm
escalates and leaves them to fail fast rather than block on a prompt no one
can answer, while an internal-only selection stays silent, or
- only internal escalators are selected on a cold cache: most such runs never
escalate, so the rare mid-run prompt is covered by the silent-call stall
notice instead.
"""
managers = list(managers)
# Root has nothing to escalate. Windows passes this guard like any other
# host, since `gsudo` gives it an escalator to probe and a cache to warm:
# a stock run there returns on the empty selection below instead, no
# manager escalating by default on Windows, so only `--sudo`, or a
# `[mpm.overrides.<id>] sudo = true` entry, reaches the probe.
if _running_as_root():
return
run_operations = frozenset(operation.name for operation in operations)
escalating_managers = [m for m in managers if _escalates_in(m, run_operations)]
escalating = sorted({m.id for m in escalating_managers})
# The verb and pronouns of the messages below, agreeing with the managers
# they name.
needs = "needs" if len(escalating) == 1 else "need"
they = "it" if len(escalating) == 1 else "they"
their = "its" if len(escalating) == 1 else "their"
internal = any(m.internal_sudo for m in managers)
if not escalating and not internal:
return
if any(manager.dry_run or manager.plan for manager in managers):
return
if ctx.meta.get(_SUDO_PRIMED):
return
ctx.meta[_SUDO_PRIMED] = True
# A forced policy cannot serve a tool that drops the cached credentials
# before its own sudo: say so, where the up-front prompt would have been.
futile = sorted(
m.id
for m in managers
if m.internal_sudo and m.resets_sudo_timestamp and _resolved_sudo(m)
)
if futile:
one = len(futile) == 1
logging.warning(
f"A forced sudo policy does nothing for {', '.join(futile)}: "
f"{'it drops' if one else 'they drop'} the cached credentials before "
f"{'its' if one else 'their'} own sudo, so "
f"{'its' if one else 'their'} password prompt still comes.",
)
escalator = ESCALATION.resolve()
if escalator is None:
# No escalator on PATH at all, or an override naming an unknown one
# (which logged its own warning). Let unprivileged managers proceed.
if escalating:
logging.warning(
f"Found none of {', '.join(e.id for e in ESCALATORS)} to escalate "
f"{', '.join(escalating)} with: {they} may fail. Install one, or "
"drop escalation with `--no-sudo` or a `[mpm] sudo = false` "
"entry in your configuration file.",
)
return
# Deferred on purpose: `config` pulls in the definitions machinery, which
# imports back into this module through `manager` and `execution`.
from .config import config_file_is_trusted
for manager in escalating_managers:
# The same file-plus-parent tamper test the config loader applies (see
# `docs/security.md`): a binary that others can modify, handed to
# `sudo`, runs their code as root. The warning does not block: like the
# risky-override warning, existing setups keep working.
cli_path = manager.cli_path
if cli_path is not None and not config_file_is_trusted(cli_path):
logging.warning(
f"About to run {cli_path} as root, but it is not owned by you "
"or root, or others can write to it or its directory. Fix its "
"ownership and permissions, or drop escalation with "
"`--no-sudo`.",
extra={"label": manager.subject},
)
logging.debug(
f"Probe the {escalator.id} credential cache: "
f"{' '.join(escalator.resolved_probe_args())}",
)
probe = _probe_credentials(escalator)
if probe is None:
# Degrade to a warning and let unprivileged managers proceed rather
# than crash.
logging.warning(
f"{escalator.id} could not be run: managers needing root may fail. "
"Drop escalation with `--no-sudo` or a `[mpm] sudo = false` entry in "
"your configuration file.",
)
return
if escalator.probe_says_warm(probe):
# Cache already warm (a prior authentication, a passwordless rule): keep
# it fresh, silently. A CI job with pre-cached credentials thus gets the
# keepalive instead of the no-terminal warning.
logging.info(
f"Found the {escalator.id} credential cache warm: no password prompt "
"needed.",
)
_start_sudo_keepalive(ctx, escalator)
return
ids = ", ".join(escalating)
probe_error = (probe.stderr or b"").decode("UTF-8", errors="replace")
# The raw answer settles which cold case this is, and catches a wording no
# matcher knows yet (the sudo-rs precedent, see _is_sudo_auth_failure).
logging.debug(f"The {escalator.id} probe answered: {probe_error.strip()!r}")
if _is_sudo_denied(probe_error):
if escalating and escalator.probe_command:
logging.warning(
f"{ids} {needs} administrator rights, and {escalator.id} denies "
f"the `{escalator.probe_command}` command it is probed with: "
f"{they} run only where a rule permits {their} own commands with "
"no password.",
)
elif escalating:
logging.warning(
f"{ids} {needs} administrator rights, but you are not authorized "
f"to run {escalator.id} on this host: {they} will fail. Drop "
"escalation with `--no-sudo` or a `[mpm] sudo = false` entry in "
"your configuration file.",
)
# An internal-only selection stays silent, as on the no-terminal path:
# each manager's own sudo surfaces the denial through its error path.
return
if _escalation_is_passwordless(escalator, escalating_managers):
# No credential is involved, so there is none to prompt for, refresh or
# lose mid-run. The flag still goes up: it gates the stall watchdog, and
# nothing can stall on a password the policy never asks for.
logging.info(
f"The {escalator.id} policy runs every escalated command without a "
"password: no prompt needed.",
)
_SUDO_CACHE_WARM.set()
ctx.call_on_close(_SUDO_CACHE_WARM.clear)
return
if not (sys.stdin.isatty() and sys.stderr.isatty()):
if escalating:
logging.warning(
f"{ids} {needs} administrator rights, but no terminal is "
f"available to prompt for a password: {they} may fail. "
f"{escalator.remedy} Or drop escalation with `--no-sudo` or a "
"`[mpm] sudo = false` entry in your configuration file.",
)
# An internal-only selection stays silent: each manager's own sudo fails
# fast and surfaces through its error path.
return
if not escalating:
# Internal-only selection on a cold cache (a stock macOS cask/fink run):
# return without prompting. Most such runs never escalate, so an up-front
# password prompt on every run would be the mirror-image regression. The
# silent-call stall notice covers the rare mid-run prompt instead.
logging.info(
"Only managers running sudo internally are selected, on a cold "
"credential cache: no up-front prompt, the stall notice covers a "
"hidden one.",
)
return
no_retention = (
f"{escalator.id} keeps no authorization past a password prompt: {ids} "
f"may fail. {escalator.remedy}"
)
if _prompt_kept(escalator, probe) is False:
logging.warning(no_retention)
return
echo(f"{ids} {needs} administrator rights to {ctx.command.name}.", err=True)
prompt_cli = escalator.prompt_args
if escalator.brands_prompt:
# `sudo --prompt` expands %-escapes, and `%p` is the account whose password
# is wanted. That is not always the invoking user: a `targetpw`, `rootpw` or
# `runaspw` policy asks for another one, and openSUSE ships `targetpw` by
# default, so a prompt naming the caller would send its users to type the
# wrong password. Manager IDs are plain slugs, but are escaped anyway so one
# carrying a `%` cannot smuggle in an escape of its own. An escalator that
# cannot be told what to print falls back to its own prompt, under the
# notice echoed above.
escaped_ids = ids.replace("%", "%%")
prompt = f"[mpm] password for %p (running {escaped_ids}): "
prompt_cli = (*prompt_cli, "--prompt", prompt)
if subprocess.run(prompt_cli, check=False).returncode != 0:
logging.warning(
f"Could not acquire {escalator.id} credentials: managers needing root "
"may fail.",
)
return
# The prompt authorized its own command. Whether the authorization holds
# for the commands to come is up to the host's policy, and `doas` without
# `persist` or polkit under `auth_admin` keep nothing: ask again.
reprobe = _probe_credentials(escalator)
if reprobe is None or not escalator.probe_says_warm(reprobe):
logging.warning(no_retention)
return
_start_sudo_keepalive(ctx, escalator)
[docs]
@dataclass(frozen=True)
class EscalationDiagnosis:
"""How escalation stands on this host, as `mpm doctor` reports it.
Gathered by {func}`diagnose_escalation`, which never prompts: it runs the
non-interactive probes {func}`prime_sudo` runs.
"""
escalated: tuple[str, ...]
"""The selected managers whose privileged commands mpm escalates."""
internal: tuple[str, ...]
"""The selected managers that run `sudo` from inside their own commands."""
is_root: bool = False
"""Whether `mpm` runs as root, which leaves nothing to escalate."""
escalator: Escalator | None = None
"""The escalator mpm drives, or `None` when the host carries none."""
chosen: bool = False
"""Whether `--sudo-command` or the `sudo_command` key named the escalator."""
path: str | None = None
"""Where the binary of the escalator resolves on `PATH`."""
skipped: tuple[tuple[str, str], ...] = ()
"""Escalators on `PATH` that auto-detection passed over, each with what
{meth}`Escalator.identity` found: `stand-in` or `unrunnable`."""
identity: str = "genuine"
"""What {meth}`Escalator.identity` found for the escalator."""
release: TokenizedString | None = None
"""The installed release, for an escalator that reads one."""
missing_options: tuple[str, ...] = ()
"""The gated options the installed release predates."""
credentials: str = ""
"""What the credential probe found: `ready`, `passwordless`, `denied`,
`cold` or `unrunnable`, or empty when nothing was probed."""
answer: str = ""
"""The first line the credential probe printed."""
prompt_kept: bool | None = None
"""Whether a password prompt would authorize the commands after it, or
`None` where the probe cannot tell."""
[docs]
def lines(self) -> list[str]:
"""The report, one fact per line."""
if self.is_root:
return ["Running as root: no command needs an escalator."]
lines = []
escalator = self.escalator
if escalator is None:
lines.append(
f"Escalator: none found among {', '.join(e.id for e in ESCALATORS)}.",
)
if self.escalated:
lines.append(
"Fix: Install one, or drop escalation with `--no-sudo` or a "
"`[mpm] sudo = false` entry in your configuration file.",
)
else:
how = "chosen with --sudo-command" if self.chosen else "detected"
where = f" at {self.path}" if self.path else ", not found on PATH"
lines.append(f"Escalator: {escalator.id}{where}, {how}.")
lines.extend(
f"Skipped: the {skipped} on PATH could not run."
if verdict == "unrunnable"
else f"Skipped: the {skipped} on PATH does not identify as {skipped}."
for skipped, verdict in self.skipped
)
if self.identity == "stand-in":
lines.append(
f"Identity: the {escalator.binary_name} on PATH does not "
f"identify as {escalator.id}, so mpm may misread its answers.",
)
if self.release is not None:
too_old = (
f", too old for {', '.join(self.missing_options)}"
if self.missing_options
else ""
)
lines.append(f"Release: {self.release}{too_old}.")
lines.extend(self._credential_lines(escalator))
escalated = ", ".join(self.escalated) or "no selected manager"
lines.append(f"Escalated by mpm: {escalated}.")
if self.internal:
lines.append(f"Escalating on their own: {', '.join(self.internal)}.")
return lines
def _credential_lines(self, escalator: Escalator) -> list[str]:
"""The lines on what the credential probe found, and the fix."""
answer = f" ({self.answer})" if self.answer else ""
if self.credentials == "ready":
return ["Credentials: ready, escalated commands run with no prompt."]
if self.credentials == "passwordless":
return [
(
"Credentials: the policy runs every escalated command with no "
"password."
),
]
if self.credentials == "unrunnable":
return [f"Credentials: {escalator.id} could not run."]
lines = []
if self.credentials == "denied":
lines.append(f"Credentials: denied{answer}.")
if escalator.probe_command:
lines.append(
f"Probe: it runs `{escalator.probe_command}`, and a rule can "
"still permit the managers' own commands.",
)
else:
lines.append(f"Credentials: a password is needed{answer}.")
if self.prompt_kept is True:
lines.append(
"Prompt: the authorization would hold for the whole run.",
)
elif self.prompt_kept is False:
lines.append(
"Prompt: the authorization would not hold past it, so mpm "
"asks for no password.",
)
if self.escalated or self.internal:
# A denial needs a rule, which no authentication can stand in for.
if self.credentials == "denied":
lines.append(
f"Fix: Give this user a {escalator.id} rule for the escalated "
"commands, or drop escalation with `--no-sudo`.",
)
else:
lines.append(f"Fix: {escalator.remedy}")
return lines
[docs]
def diagnose_escalation(managers: Iterable[PackageManager]) -> EscalationDiagnosis:
"""Report how escalation stands on this host, without prompting.
Runs the probes {func}`prime_sudo` runs, in the same non-interactive shape:
the identity check of the escalators auto-detection passed over, the
release read and the credential probe. `managers` are the selected ones,
all operations included: escalation belongs to the host, not to the
managers one subcommand reaches.
"""
managers = list(managers)
escalated = tuple(
sorted(
manager.id
for manager in managers
if _resolved_sudo(manager)
and not manager.internal_sudo
and manager.privileged_operations
),
)
internal = tuple(sorted(m.id for m in managers if m.internal_sudo))
if _running_as_root():
return EscalationDiagnosis(escalated, internal, is_root=True)
escalator = ESCALATION.resolve()
if escalator is None:
return EscalationDiagnosis(escalated, internal)
chosen = ESCALATION.chosen
skipped: list[tuple[str, str]] = []
if not chosen:
for earlier in ESCALATORS[: ESCALATORS.index(escalator)]:
if shutil.which(earlier.binary_name):
verdict = earlier.identity()
if verdict != "genuine":
skipped.append((earlier.id, verdict))
supported = escalator.supported_options()
found = EscalationDiagnosis(
escalated,
internal,
escalator=escalator,
chosen=chosen,
path=shutil.which(escalator.binary_name),
skipped=tuple(skipped),
identity=escalator.identity(),
release=(
_installed_release(escalator.version_args)
if escalator.version_args
else None
),
missing_options=tuple(
option for option, _ in escalator.gated_options if option not in supported
),
)
probe = _probe_credentials(escalator)
if probe is None:
return replace(found, credentials="unrunnable")
output = (probe.stdout or b"").decode("UTF-8", errors="replace")
error = (probe.stderr or b"").decode("UTF-8", errors="replace")
answer = next(
(
stripped
for line in (*error.splitlines(), *output.splitlines())
if (stripped := line.strip())
),
"",
)
if escalator.probe_says_warm(probe):
return replace(found, credentials="ready")
if _is_sudo_denied(error):
return replace(found, credentials="denied", answer=answer)
if _escalation_is_passwordless(
escalator,
[manager for manager in managers if manager.id in escalated],
):
return replace(found, credentials="passwordless")
return replace(
found,
credentials="cold",
answer=answer,
prompt_kept=_prompt_kept(escalator, probe),
)
def _hidden_prompt_risk(manager: CLIExecutor, operation: str | None) -> bool:
"""Whether a call may block on a `sudo` password prompt the user cannot see.
True for a call of a manager that escalates internally
({attr}`CLIExecutor.internal_sudo
<meta_package_manager.execution.CLIExecutor.internal_sudo>`), in an
operation where its tool may run `sudo`
({attr}`CLIExecutor.internal_sudo_operations
<meta_package_manager.execution.CLIExecutor.internal_sudo_operations>`),
made on a terminal while the credential cache is cold. Those are the
conditions under which the tool's own `sudo` prompts for a password nothing
has primed. A manager that drops the cache itself
({attr}`CLIExecutor.resets_sudo_timestamp
<meta_package_manager.execution.CLIExecutor.resets_sudo_timestamp>`)
meets them whatever the cache holds.
{meth}`CLIExecutor.run <meta_package_manager.execution.CLIExecutor.run>`
reads it once per call and spends it on every response, so they cannot drift
apart: it arms `_StallWatchdog`, and it holds the call's spinner still,
naming the call once on a static line instead (see
{data}`~meta_package_manager.execution.STILL_CALL_MARKER`).
The still spinner is what makes the prompt answerable. `sudo` writes its
prompt to `/dev/tty` with no trailing newline, so the prompt sits on the live
terminal line. An animated call repaints that line every
{data}`~meta_package_manager.execution.SPINNER_DELAY` seconds, which erases
the prompt within one frame. Nothing repaints it, so the user reads a notice
about a prompt they cannot see, and the run dies at
{data}`~meta_package_manager.execution.MUTATING_TIMEOUT`.
A concurrent batch draws one aggregate indicator instead of these per-call
spinners, and that one is never in the way: the predicate also holds such a
manager back to the sequential tail of
{func}`~meta_package_manager.dispatch.dispatch`, which runs once the batch,
and its indicator, are done.
"""
return (
manager.internal_sudo
and operation in manager.internal_sudo_operations
and sys.stderr.isatty()
and (manager.resets_sudo_timestamp or not _SUDO_CACHE_WARM.is_set())
)
def _spends_warm_cache(manager: CLIExecutor, operation: str | None) -> bool:
"""Whether a call may run the manager's own `sudo` on the cache mpm keeps warm.
True for a call of a manager that escalates internally, in an operation
where its tool may run `sudo`
({attr}`CLIExecutor.internal_sudo_operations
<meta_package_manager.execution.CLIExecutor.internal_sudo_operations>`),
while the cache is warm.
{meth}`CLIExecutor.run <meta_package_manager.execution.CLIExecutor.run>` then
keeps the call in the session of the terminal, where a call otherwise gets
a session of its own.
`sudo` keys a ticket on the terminal [and on its
session](https://github.com/apple-oss-distributions/sudo/blob/9d15835617972783e565818fae7ea047626b1e5c/sudo/plugins/sudoers/timestamp.c#L887-L892),
and falls back on its parent process [when it has no
terminal](https://github.com/apple-oss-distributions/sudo/blob/9d15835617972783e565818fae7ea047626b1e5c/sudo/plugins/sudoers/timestamp.c#L404-L420).
A detached child therefore finds no ticket, and no terminal to prompt on:
its privileged step fails on the very cache meant to serve it. Measured on
macOS 27: right after `sudo --validate`, `sudo --non-interactive true`
answers `a password is required` from a new session.
A manager that drops the cache itself has nothing to spend:
`_hidden_prompt_risk` covers its prompt.
"""
return (
manager.internal_sudo
and not manager.resets_sudo_timestamp
and operation in manager.internal_sudo_operations
and _SUDO_CACHE_WARM.is_set()
)
def _terminal_modes() -> list[Any] | None:
"""The modes of the controlling terminal, as `termios.tcgetattr` reads them.
:return: `None` when the terminal cannot answer: on Windows, which has no
`termios`, and in a process with no controlling terminal.
"""
if sys.platform == "win32":
return None
try:
fd = os.open(os.ctermid(), os.O_RDONLY | os.O_NOCTTY)
except OSError:
return None
try:
return termios.tcgetattr(fd)
except termios.error:
return None
finally:
os.close(fd)
def _restore_terminal_modes(modes: list[Any]) -> None:
"""Give the controlling terminal back the `modes` `_terminal_modes` read.
Best effort: a terminal that is gone has nothing to get back.
"""
if sys.platform == "win32":
return
try:
fd = os.open(os.ctermid(), os.O_RDWR | os.O_NOCTTY)
except OSError:
return
# A process outside the foreground group that sets the modes of its
# terminal is sent SIGTTOU, which stops it, unless its thread blocks that
# signal.
blocked = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGTTOU})
try:
termios.tcsetattr(fd, termios.TCSANOW, modes)
except termios.error:
pass
finally:
signal.pthread_sigmask(signal.SIG_SETMASK, blocked)
os.close(fd)
def _terminal_echo(modes: list[Any] | None) -> bool | None:
"""Whether a terminal in these `modes` echoes the keys the user types.
`sudo` turns the echo off on `/dev/tty` [before it writes its password
prompt](https://github.com/sudo-project/sudo/blob/f2f340938e51f9f12b29b72507d9edca8cb98abc/src/tgetpass.c#L179-L229),
and restores it once it has the answer. An echo that goes off during a call
therefore means a prompt is on screen and waits at the cursor, which is what
`_StallWatchdog` reads it for. Measured on macOS 27 with the system `sudo`:
`stty -a` reports `-echo` while `Password:` waits.
:param modes: what `_terminal_modes` read.
:return: `None` when `_terminal_modes` had no answer.
"""
if sys.platform == "win32" or modes is None:
return None
return bool(modes[3] & termios.ECHO)
class _StallWatchdog(logging.Handler):
"""Warn when a CLI call that may hide a `sudo` password prompt goes silent.
A manager that escalates internally ({attr}`CLIExecutor.internal_sudo
<meta_package_manager.execution.CLIExecutor.internal_sudo>`) can raise a
`sudo` prompt from inside its own commands. The child reads `stdin` from
`/dev/null` and its output streams to `DEBUG` logs, so on a cold
credential cache the prompt lands invisibly on `/dev/tty`: the run looks
stuck until the mutating timeout kills it. When {func}`prime_sudo` left the
cache cold, the keepalive later found it dropped mid-run, or the tool drops
it before its own `sudo`, {meth}`CLIExecutor.run
<meta_package_manager.execution.CLIExecutor.run>` arms this watchdog around
the spawn: once {data}`_STALL_NOTICE_DELAY` seconds pass without a fresh
output line, a daemon thread logs one `WARNING` naming the manager and
quoting its last line, so the user can tell a hidden prompt from a slow
download. Each silence episode warns at most once; a fresh line starts a new
episode.
The notice only pays off because of the still spinner:
`_hidden_prompt_risk` arms both for the same call, so a prompt the tool
prints stays on screen to be answered. A notice raised while an animation
erases that prompt names the stall but cannot end it.
A prompt the terminal shows gets no notice. A program that reads a password
holds the terminal echo off for as long as it waits (see `_terminal_echo`),
so an echo that went off during the call means the prompt is on screen, with
the cursor right after it. A notice written then moves the cursor below the
warning, away from the prompt the user has to answer. The prompt settles its
silence episode instead, the way a notice does, so the silence after the
answer draws no warning either: the tool is assumed to run what it asked the
password for. Where the terminal cannot answer, or its echo was already off
when the call started, the notice fires as it does for a prompt nothing
shows.
The watchdog doubles as the sole handler of {attr}`tee`, the logger
{meth}`CLIExecutor.run <meta_package_manager.execution.CLIExecutor.run>`
hands to {func}`click_extra.execution.run_cli` in place of the root logger:
{meth}`emit` tracks the child's activity, then forwards every record
verbatim to the root logger, whose level click-extra's `--verbosity`
manages, keeping the display byte-identical to an un-teed run at every
verbosity.
```{note}
Considered alternative: a `SUDO_ASKPASS` helper. `brew` documents
passing `--askpass` to its internal `sudo` whenever that variable is
set, so mpm could export a helper into the child environment and rebrand
the hidden prompt itself ("[mpm] cask needs your password..."). Rejected:
the helper reads the raw password and pipes it to `sudo` (a security
surface this notice avoids entirely), and it only covers tools honoring
the variable (`brew` does, `fink`'s plain `sudo` re-exec does not). The
scoped `sudo = true` opt-in documented in `docs/sudo.md` already covers
users wanting a guaranteed up-front prompt.
```
```{note}
One terminal state defeats the notice, and it is not a defect this class
can repair. Keeping the child in mpm's process group is what lets its
`sudo` reach the terminal at all, so when that group is *not* the
terminal's foreground group, the read earns `SIGTTIN` and the kernel stops
the whole group, mpm included: the notice thread is stopped along with the
process it would warn about. Measured on a shell without job control
(`ssh -tt host 'mpm …'`), where `ps` reports `mpm`, the helper and `sudo`
all in state `T` at `do_signal_stop`. An interactive run puts mpm in the
foreground group and the notice fires normally, so this reaches the
frontends that spawn mpm from a pty without making it the foreground job.
```
"""
tee: logging.Logger
"""Stand-in destination for `run_cli`'s streamed records while armed.
Deliberately a direct {class}`logging.Logger` construction, never
{func}`logging.getLogger`: unregistered, each armed call gets a private tee
that concurrent calls cannot cross-contaminate; parentless, its records cannot
propagate straight to the root handlers, which would bypass the root level
gate and leak `DEBUG` lines at default verbosity. Its `DEBUG` level lets
every record reach {meth}`emit`: dropping is the root logger's decision.
"""
prompt_unanswered: bool
"""Whether the call ended while a password prompt still held the terminal.
Set by {meth}`stop`.
{meth}`CLIExecutor.run <meta_package_manager.execution.CLIExecutor.run>`
reads it when the call times out, to name the prompt in its warning.
```{note}
That is how an unanswered prompt ends on macOS, whose `sudo` is built
[with no prompt
timeout](https://github.com/apple-oss-distributions/sudo/blob/9d15835617972783e565818fae7ea047626b1e5c/sudo/config.h#L1233).
Upstream [defaults to 5
minutes](https://github.com/sudo-project/sudo/blob/f2f340938e51f9f12b29b72507d9edca8cb98abc/configure.ac#L198),
well under {data}`~meta_package_manager.execution.MUTATING_TIMEOUT`:
such a `sudo` gives up first, and the tool then fails with an account of
it that `_is_prompt_timeout` recognizes.
```
"""
def __init__(self, label: str) -> None:
"""Arm the watchdog for one CLI call, its notices labeled with `label`."""
super().__init__()
self._label = label
self._started = time.monotonic()
# Latest child activity, one `(monotonic timestamp, output line)` pair.
# Written by emit() in a single reference assignment and read the same way
# by the notice thread, so the pair stays consistent without a lock
# (free-threading safe). Starts at arming time, with no line seen yet.
self._activity: tuple[float, str | None] = (self._started, None)
# Activity timestamp of the silence episode already settled, by a notice
# or by a prompt seen on the terminal, so each episode warns at most
# once. Touched by the notice thread only.
self._noticed: float | None = None
# Modes of the terminal before the child starts. Only an echo found on
# can go off under a prompt, see _prompt_on_screen(), and stop() gives
# these modes back when a killed prompt left its own behind.
self._modes = _terminal_modes()
self._echoed = _terminal_echo(self._modes) is True
self.prompt_unanswered = False
# Whether a prompt held a notice back, for the DEBUG line of stop().
self._prompt_seen = False
# Instantiated directly, not via getLogger: the tee must stay out of the
# registry (re-arming would reuse it, stacking handlers) and have no
# parent, so this handler is its only sink and nothing double-emits.
self.tee = logging.Logger( # noqa: LOG001
f"mpm-stall-tee-{label}", logging.DEBUG
)
self.tee.addHandler(self)
self._stop = threading.Event()
self._thread = threading.Thread(target=self._watch, daemon=True)
self._thread.start()
def emit(self, record: logging.LogRecord) -> None:
"""Track child activity, then forward `record` verbatim to the root.
Only the streamed output lines refresh the activity state: they are the
records carrying a `label` attribute (`run_cli` labels the child's
output lines only, never its own prompt-disclosure or PID-tracking lines),
and only genuine output vouches that the child is not blocked on a prompt.
Forwarding re-enters {meth}`logging.Logger.log` on the root logger so
its level gate (the one click-extra's `--verbosity` manages) and its
handlers apply exactly as if `run_cli` had logged there directly.
"""
try:
message = record.getMessage()
label = getattr(record, "label", None)
root = logging.getLogger()
if label is None:
root.log(record.levelno, message)
else:
self._activity = (time.monotonic(), message)
root.log(record.levelno, message, extra={"label": label})
except Exception: # noqa: BLE001
self.handleError(record)
def _prompt_on_screen(self) -> bool:
"""Whether a program holds the terminal echo off to read a password.
Read against the echo found when the call started: a terminal whose echo
was already off then, or that cannot answer, proves nothing about a prompt.
"""
return self._echoed and _terminal_echo(_terminal_modes()) is False
def _watch(self) -> None:
"""Notice-thread body: warn once per silence episode.
A silence episode is identified by the timestamp of the latest activity: a
fresh output line moves it, which re-arms the notice for the next silent
stretch. Wakes at most every second (sooner when the delay itself is
shorter) so a stop request is honored promptly, and so a prompt is seen
on the terminal long before the notice is due.
"""
while not self._stop.wait(min(1.0, _STALL_NOTICE_DELAY)):
last_stamp, last_line = self._activity
if self._noticed == last_stamp:
continue
if self._prompt_on_screen():
# The prompt is the notice of this episode. Nothing is written,
# not even at DEBUG: any line here moves the cursor away from it.
self._noticed = last_stamp
self._prompt_seen = True
continue
silence = time.monotonic() - last_stamp
if silence < _STALL_NOTICE_DELAY:
continue
self._noticed = last_stamp
if last_line is None:
detail = "No output since the command started."
else:
# Cap the quoted line at 120 characters, ellipsis included.
if len(last_line) > 120:
last_line = last_line[:119] + "…"
detail = f'Last output: "{last_line}"'
# Break the line a prompt may be sitting on, so the notice starts on
# a fresh one instead of running on from "Password:". The tool's
# prompt goes to `/dev/tty` without a trailing newline and never
# reaches the pipe mpm reads. _prompt_on_screen() rules a prompt out
# only where the terminal answers, and never one that echoes its
# answer: the watchdog is armed only when a prompt is possible
# (`_hidden_prompt_risk`), which is what makes an unconditional break
# the right trade. The cost is a blank line on a call that stayed
# silent without prompting.
sys.stderr.write("\n")
sys.stderr.flush()
# WARNING survives the default verbosity, and click-extra's handler
# prints it above any animating spinner frame. The wording never
# instructs the user to type blindly: the prompt may not exist.
logging.warning(
f"No output for {int(silence)}s: may be waiting on a hidden "
f"password prompt. {detail}",
extra={"label": self._label},
)
def stop(self) -> None:
"""Disarm: stop the notice thread, join it, and detach the handler.
The call is over, so no prompt waits at the cursor: this is where a
notice held back for a prompt gets its `DEBUG` record.
A prompt that still holds the terminal belongs to a program the call
took down with it, on a timeout or an interrupt. That kill is a
`SIGKILL`, which leaves the program no chance to restore the echo it
turned off. So {attr}`prompt_unanswered` records the prompt, and the
terminal gets back the modes it had when the call started.
```{note}
The restored echo shows no password, because the kill reaches a
`sudo` at its prompt. Read in the sources, not measured: `sudo`
asks the password under [the credentials it started
with](https://github.com/apple-oss-distributions/sudo/blob/9d15835617972783e565818fae7ea047626b1e5c/sudo/plugins/sudoers/set_perms.c#L890-L902),
its real UID still the caller's, and the kernel [lets a process
signal](https://github.com/apple-oss-distributions/xnu/blob/f6217f891ac0bb64f3d375211650a4c1ff8ca1ea/bsd/kern/kern_sig.c#L343-L352)
another whose real UID is its own.
```
"""
self._stop.set()
self._thread.join()
self.close()
self.prompt_unanswered = self._prompt_on_screen()
if self.prompt_unanswered and self._modes is not None:
_restore_terminal_modes(self._modes)
if self._prompt_seen:
logging.debug(
"Held the stall notice back: a password prompt was on screen.",
extra={"label": self._label},
)