diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d2f02dc..71d27e3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -642,10 +642,13 @@ jobs: # interfaces are never handed to it. `--require` makes a runner that cannot run # the check a failure, not a quiet SKIP. Not in the branch ruleset: a red one says # Linux capture broke, and whether that blocks a merge is the owner's call. + # + # Its last step is the window's way to root on Linux (linuxenv.py), for the same + # reason: only a real pkexec and polkitd show what pkexec does to the program. linux-capture: name: Linux capture (experimental, real kernel) runs-on: ubuntu-latest - timeout-minutes: 10 + timeout-minutes: 15 steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: @@ -659,6 +662,37 @@ jobs: run: uname -r - name: Real capture inside a network namespace of its own run: sudo "$(command -v python)" tools/ci_linux_capture_smoke.py --require + # The window asking for root through polkit, with this image's own pkexec and + # polkitd (Ubuntu 24.04: polkit 124, which names the person by PKEXEC_UID + # only). The runner is thrown away after the job, so a polkit rule that lets + # its user run pkexec without a password stands in for the person typing it. + # The smoke first asks without the rule and must see pkexec refuse. + - name: The window asks for root through pkexec (real polkit, Xvfb) + run: | + missing="" + for package in pkexec polkitd python3-tk xvfb; do + dpkg -s "$package" >/dev/null 2>&1 || missing="$missing $package" + done + if [ -n "$missing" ]; then + echo "installing:$missing" + # The mirror and the two-step install: see "GUI render check". + for source in /etc/apt/apt-mirrors.txt /etc/apt/sources.list \ + /etc/apt/sources.list.d/*.sources \ + /etc/apt/sources.list.d/*.list; do + if [ -f "$source" ]; then + sudo sed -i 's|azure.archive.ubuntu.com|archive.ubuntu.com|g' "$source" + fi + done + if ! sudo timeout 180 apt-get install -y --no-install-recommends $missing; then + sudo timeout 240 apt-get update \ + -o Acquire::Retries=2 \ + -o Acquire::http::Timeout=15 -o Acquire::https::Timeout=15 + sudo timeout 240 apt-get install -y --no-install-recommends $missing + fi + fi + pkexec --version + timeout 120 xvfb-run -a --server-args="-screen 0 1280x800x24" \ + python -u tools/ci_linux_pkexec_smoke.py --require build: name: build the executable (Windows) diff --git a/CHANGELOG.md b/CHANGELOG.md index 616ab47..adbd095 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,12 @@ The format follows [Keep a Changelog](https://keepachangelog.com/); versions fol row of its own. The lowest round trip is the lowest of the last 5 minutes, and timeouts are counted from Linux 6.7 on. +- **Linux, experimental: the window asks for the administrator password.** Started without + sudo, it shows the desktop's own password dialog, as Windows shows UAC, and opens as root. + **Cancel** opens it without root, as before. Where there is no such dialog, as in WSL or over + ssh, it opens without root at once. `BEAN_NO_ELEVATE=1` turns the question off. The files it + saves stay yours, as under sudo. + - **Ping on the Tools tab, to see the session's latency and loss arrive.** Type an address or a name and press **Start**: each ping is a row with its round trip and what came back, the last 60 are charted, and the totals say what was lost. The ping goes on while another tab is in diff --git a/README.md b/README.md index 31952f1..e22f2b9 100644 --- a/README.md +++ b/README.md @@ -190,6 +190,12 @@ checkout, which you could then not delete without `sudo`. Plain `sudo` keeps the the window needs to reach the desktop, `DISPLAY` and `XAUTHORITY`, on Ubuntu 24.04 and 26.04, also under Wayland. Ubuntu 26.04's `sudo` refuses `-E`, and nothing here needs it. +Started without `sudo` (`.venv/bin/python bean_network_tester.py`), the window asks for an +administrator's password in the desktop's own dialog, through polkit's `pkexec`, as Windows +asks through UAC, and opens as root. **Cancel** opens it without root, as before. Where no such +dialog exists, such as in WSL or over ssh, it opens without root at once, and +`BEAN_NO_ELEVATE=1` turns the question off. + `--doctor` lists what the machine still lacks, and while something stops a capture it ends with exit code 1, as on Windows without administrator rights. Where Linux differs from Windows: @@ -204,8 +210,8 @@ exit code 1, as on Windows without administrator rights. Where Linux differs fro - If the tool is killed, the traffic it was holding stays held for about 3 seconds, then flows again. There is nothing for `--cleanup-driver` to do: the kernel cleans up by itself. - IP fragments always pass through the tool, even those its filter does not match. -- In the window, the files it saves (window state, profiles, CSV) stay yours under `sudo`, and - GitHub and Donate print their address instead of starting a browser as root. +- In the window, the files it saves (window state, profiles, CSV) stay yours under `sudo` and + `pkexec`, and GitHub and Donate print their address instead of starting a browser as root. - On the Tools tab, Ping, Sockets, Port check, Filter tester, TCP health and Diagnostics work. Ping works without `sudo` where the system lets your group send pings (the setting `net.ipv4.ping_group_range`: Ubuntu 26.04 allows it, 24.04 does not). With `sudo` it sends @@ -1625,6 +1631,7 @@ beantester/ the implementation package exitcodes.py EXIT CODES - the CLI/CI-CD contract (single source of truth) clilog.py CLI output: log on stderr ([bean]), data on stdout (text/NDJSON) winenv.py Windows: admin, elevation (UAC), console detach, DPI + linuxenv.py Linux: the window asks for root through polkit (pkexec) driver.py WinDivert driver lifecycle + --doctor / --cleanup-driver fields.py FIELD REGISTRY - single source of truth: type, label, unit, range, form section, profile scope, CLI flag diff --git a/beantester/cli.py b/beantester/cli.py index 8aafd8d..1bce0ad 100644 --- a/beantester/cli.py +++ b/beantester/cli.py @@ -20,7 +20,7 @@ import threading import time -from . import appinfo, capture_backend, clilog, driver, exitcodes, winenv +from . import appinfo, capture_backend, clilog, driver, exitcodes, linuxenv, winenv from .appinfo import command_name, program_name, version_line from .clilog import CliLog from . import crashlog @@ -1323,6 +1323,14 @@ def _run_gui(argv, relaunched=False): # elevate_self never asks again, admin or not (see winenv.UAC_RELAUNCH). if winenv.elevate_self(argv): return exitcodes.OK # the elevated copy took over + elif linuxenv.wants_root(): + # The same question through polkit, and the copy's exit code back; None = + # it did not run (dismissed, no agent), so the window opens here without + # root (see linuxenv.py). + code = linuxenv.ask_for_root(argv) + if code is not None: + return code + linuxenv.take_desktop() # in the copy pkexec started: the person's screen if is_frozen(): winenv.detach_console() # no black window behind the GUI # Arm hard-crash capture for the GUI too, not only for a real capture session @@ -1338,6 +1346,7 @@ def _run_gui(argv, relaunched=False): # the import would also make the guard for this line untestable anywhere Tk is # missing, i.e. green here and vacuous on the Linux runner. crashlog.arm_native() + root = None try: import tkinter as tk @@ -1346,7 +1355,12 @@ def _run_gui(argv, relaunched=False): root = tk.Tk() App(root) except Exception as exc: - return _window_did_not_open(exc) + code = _window_did_not_open(exc) + # A copy pkexec started whose Tk cannot reach the person's display answers + # pkexec's own "did not run": the person's process then opens the window. + if root is None and type(exc).__name__ == "TclError" and linuxenv.started_by_pkexec(): + return linuxenv.NO_WINDOW + return code root.mainloop() return exitcodes.OK diff --git a/beantester/linuxenv.py b/beantester/linuxenv.py new file mode 100644 index 0000000..68a0dd2 --- /dev/null +++ b/beantester/linuxenv.py @@ -0,0 +1,291 @@ +"""Process-level Linux knobs for the window: asking for root through polkit. + +The Windows twin is ``winenv`` and its UAC prompt. Here the window, started +without root, runs itself again through ``pkexec``, which shows the desktop's +own password dialog (GNOME's, measured on Ubuntu 26.04), and waits for that +copy. A dismissed or impossible request leaves the window to open here, without +root, as it did before this module existed. + +What pkexec does to the program it starts - measured on Ubuntu 26.04 (polkit 127) +and 24.04 (polkit 124, in CI), and read in pkexec's own source - and what follows: + +* It keeps only a few variables and drops ``DISPLAY`` and ``XAUTHORITY``, so Tk + in the copy cannot reach the screen ("no display name"). The copy takes them, + and the working directory, from the process that asked - its parent, which + belongs to the person pkexec names in ``PKEXEC_UID`` (``take_desktop``). +* It refuses to run at all (exit 127) when ``SHELL`` is not in ``/etc/shells`` + or a kept variable holds ``/``, ``%`` or ``..``, so it is handed only the + locale variables that pass those rules (``pkexec_environment``). +* It closes every descriptor above 2, so the copy can answer only with its exit + code: 126 dismissed, 127 anything else that stopped it (``NOT_STARTED``). The + process that asked waits for that code (``ask_for_root``) - not to keep the copy + alive: pkexec sets ``PR_SET_PDEATHSIG`` while it asks, but for a person's call + the kernel clears it when pkexec takes root's group, so the copy outlives the + process that asked (measured: that process stopped, the root window stayed). +* Without ``--keep-cwd`` (polkit 121 and newer) it starts the program in root's + home, where ``python -m beantester`` from a checkout cannot import itself. +""" +import os +import posixpath +import re +import shutil +import signal +import stat +import subprocess +import sys + +from . import crashlog, winenv + +# What pkexec answers when the program did not run: 126 the person dismissed the +# dialog, 127 not authorised, no agent to ask through, or any other error. None of +# the program's own exit codes (exitcodes.py) is either. +NOT_STARTED = (126, 127) + +# What the copy answers when it ran as root but its window could not reach the +# person's display: pkexec's own "did not run", so the person's process opens the +# window itself, without root, instead of leaving the person with no window at all. +NO_WINDOW = 127 + +# The variables pkexec keeps from its caller, apart from SHELL, TERM, COLORTERM, +# DISPLAY and XAUTHORITY: nothing of those is needed, and each is a way to be refused. +LOCALE_NAMES = ("LANG", "LANGUAGE", "LINGUAS", "LC_ALL", "LC_COLLATE", "LC_CTYPE", + "LC_MESSAGES", "LC_MONETARY", "LC_NUMERIC", "LC_TIME") + +# pkexec's own test for a kept variable (validate_environment_variable). +_SUSPICIOUS = ("/", "%", "..") + +# host:display.screen, the host part empty for the local display. Anything else is +# not taken from the parent. +_DISPLAY = re.compile(r"[A-Za-z0-9._-]*:[0-9]+(\.[0-9]+)?") + +# The most of a parent's environment that is read. +_ENVIRON_LIMIT = 1 << 20 + + +def is_linux(): + return sys.platform.startswith("linux") + + +def wants_root(environ=None, euid=None, which=shutil.which): + """True when the window should ask for root before it opens. + + Only on Linux, and never as root: root that still cannot capture (a container + without CAP_BPF) would otherwise start copy after copy. Not with + ``BEAN_NO_ELEVATE`` set, as on Windows; not without a display, where there is + no window to ask from; not with the capabilities already there; and not + without pkexec. + """ + environ = os.environ if environ is None else environ + if not is_linux() or winenv.elevation_disabled(): + return False + if (os.geteuid() if euid is None else euid) == 0: + return False + if not environ.get("DISPLAY"): + return False + from . import capture_backend + if capture_backend.current().has_capture_rights(): + return False + return bool(which("pkexec")) + + +def pkexec_environment(environ=None): + """What pkexec is started with: the locale variables it will not refuse.""" + environ = os.environ if environ is None else environ + return {name: environ[name] for name in LOCALE_NAMES + if environ.get(name) and not any(bad in environ[name] for bad in _SUSPICIOUS)} + + +def relaunch_command(argv, pkexec, keep_cwd): + """The argv that starts this program again as root through pkexec, or None. + + ``winenv._relaunch_command`` repeats what this program was started with. From + sources ``-B`` is added when it is not there: root leaves no cache folders in + the person's checkout. ``python -m`` needs the working directory, which only + ``--keep-cwd`` keeps; without it there is nothing to repeat. + """ + program, args = winenv._relaunch_command(list(argv)) + if not winenv.is_frozen(): + head = args[:len(args) - len(argv)] + if "-m" in head and not keep_cwd: + return None + if "-B" not in head: + args = ["-B"] + args + return ([pkexec, "--disable-internal-agent"] + (["--keep-cwd"] if keep_cwd else []) + + [program] + args) + + +def _run(command, **kw): + """The one place this module starts a process: pkexec, for its usage and for + the copy (tests/test_no_telemetry.py SPAWN_ALLOWED).""" + return subprocess.run(command, check=False, **kw) + + +def ask_for_root(argv, run=_run, which=shutil.which, environ=None): + """Run this program again as root through pkexec and wait for it. + + Returns the copy's exit code, or None when it did not run (dismissed, not + allowed, nothing to ask through, the copy could not open its window): the + caller then opens the window itself, without root. A copy killed by a signal + answers 128 + the signal, as a shell would. + + Ctrl+C in the terminal reaches the copy as well, so here it is waited through + with a handler that does nothing - not SIG_IGN, which the copy would inherit + (pkexec executes it with the dispositions it was given). + """ + pkexec = which("pkexec") + if not pkexec: + return None + env = pkexec_environment(environ) + command = relaunch_command(argv, pkexec, _keeps_cwd(pkexec, env, run)) + if command is None: + return None + previous = signal.signal(signal.SIGINT, _wait_through) + try: + code = run(command, env=env).returncode + except OSError as exc: + crashlog.note(exc, "linuxenv.pkexec") + print(f"pkexec could not be started ({exc}): the window opens without " + "administrator rights", file=sys.stderr) + return None + finally: + signal.signal(signal.SIGINT, previous) + if code in NOT_STARTED: + print(f"pkexec did not start the window as root (exit {code}): " + "it opens without administrator rights", file=sys.stderr) + return None + return 128 - code if code < 0 else code + + +def _wait_through(_signum, _frame): + """Ctrl+C while the copy runs: the copy decides what it means.""" + + +def _keeps_cwd(pkexec, env, run): + """Whether this pkexec takes ``--keep-cwd`` - read from its usage, which names it + from polkit 121 on, rather than from a version a distribution may have patched.""" + try: + done = run([pkexec, "--help"], env=env, capture_output=True, text=True, + timeout=10) + except (OSError, subprocess.SubprocessError) as exc: + crashlog.note(exc, "linuxenv.keep_cwd") + return False + return "--keep-cwd" in (done.stdout or "") + (done.stderr or "") + + +def started_by_pkexec(environ=None): + """True in the copy pkexec started: root, with ``PKEXEC_UID`` naming the person.""" + environ = os.environ if environ is None else environ + return is_linux() and os.geteuid() == 0 and _uid(environ.get("PKEXEC_UID")) is not None + + +def take_desktop(environ=None, system=None, ppid=None): + """In the copy pkexec started: the person's display, X authority and working + directory, from the process that asked. Returns the names it set. + + Nothing is taken when ``DISPLAY`` is already there (sudo keeps it), outside + pkexec, or when the parent does not belong to the person in ``PKEXEC_UID``. + """ + environ = os.environ if environ is None else environ + if not started_by_pkexec(environ) or environ.get("DISPLAY"): + return [] + desktop = desktop_of_parent(environ, system or _System(), + os.getppid() if ppid is None else ppid) + cwd = desktop.pop("cwd", None) + environ.update(desktop) + if cwd is not None: + try: + os.chdir(cwd) + except OSError: + cwd = None + return sorted(desktop) + (["cwd"] if cwd is not None else []) + + +def desktop_of_parent(environ, system, ppid): + """``{"DISPLAY", "XAUTHORITY", "cwd"}`` of the parent, those that pass, or {}. + + The parent must be the person's: its real uid is ``PKEXEC_UID``. ``DISPLAY`` + must look like one. ``XAUTHORITY`` must be a regular file of the person's at an + absolute path - the parent's, else ``~/.Xauthority`` in the person's home, + which is where pkexec itself would look. Hygiene rather than a boundary: the + person has just typed an administrator's password. + """ + uid = _uid(environ.get("PKEXEC_UID")) + if uid is None or system.real_uid(ppid) != uid: + return {} + parent = system.environ(ppid) + display = parent.get("DISPLAY", "") + if not _DISPLAY.fullmatch(display): + return {} + found = {"DISPLAY": display} + home = system.home_of(uid) + for xauthority in (parent.get("XAUTHORITY"), + home and posixpath.join(home, ".Xauthority")): + if xauthority and posixpath.isabs(xauthority) and system.owns_file(uid, xauthority): + found["XAUTHORITY"] = xauthority + break + cwd = system.cwd(ppid) + if cwd: + found["cwd"] = cwd + return found + + +def _uid(text): + try: + uid = int(text) + except (TypeError, ValueError): + return None + return uid if uid >= 0 else None + + +class _System: + """What the copy reads about the person: its parent's ``/proc`` entries and the + person's files. ``root`` lets a test point it at a folder of its own.""" + + def __init__(self, root="/proc"): + self.root = root + + def real_uid(self, pid): + try: + with open(os.path.join(self.root, str(pid), "status"), encoding="ascii", + errors="replace") as f: + status = f.read() + except OSError: + return None + for line in status.splitlines(): + fields = line.split() + if fields[:1] == ["Uid:"] and len(fields) > 1: + return _uid(fields[1]) + return None + + def environ(self, pid): + try: + with open(os.path.join(self.root, str(pid), "environ"), "rb") as f: + raw = f.read(_ENVIRON_LIMIT) + except OSError: + return {} + found = {} + for entry in raw.split(b"\0"): + name, sep, value = entry.decode("utf-8", "replace").partition("=") + if sep and name not in found: + found[name] = value + return found + + def cwd(self, pid): + try: + place = os.readlink(os.path.join(self.root, str(pid), "cwd")) + except OSError: + return None + return place if os.path.isdir(place) else None + + @staticmethod + def home_of(uid): + from . import paths + return paths.home_of(uid) + + @staticmethod + def owns_file(uid, path): + try: + info = os.lstat(path) + except OSError: + return False + return stat.S_ISREG(info.st_mode) and info.st_uid == uid diff --git a/beantester/paths.py b/beantester/paths.py index 8a0c6ca..56ca04e 100644 --- a/beantester/paths.py +++ b/beantester/paths.py @@ -121,17 +121,34 @@ def running_as_root(): def invoking_user(): - """``(uid, gid)`` of the person who typed ``sudo`` for this root process, else ``None``. + """``(uid, gid)`` of the person who started this root process with ``sudo`` or + ``pkexec``, else ``None``. What sudo sets for the command it runs: sudo 1.9 (Ubuntu 24.04) and sudo-rs - (26.04) both set ``SUDO_UID`` and ``SUDO_GID`` (measured). ``None`` when this - process is not root, when nothing says who started it, and on Windows. + (26.04) both set ``SUDO_UID`` and ``SUDO_GID`` (measured). pkexec sets them too + from polkit 127 (Ubuntu 26.04); before that (124 on 24.04, 0.105 on 22.04) it + names the person only by ``PKEXEC_UID``, and the group is theirs from the + account database. ``None`` when this process is not root, when nothing says + who started it, and on Windows. """ if not running_as_root(): return None try: return int(os.environ["SUDO_UID"]), int(os.environ["SUDO_GID"]) except (KeyError, ValueError): + return _pkexec_user() + + +def _pkexec_user(): + """``(uid, gid)`` from ``PKEXEC_UID`` and the account database, else ``None``.""" + try: + uid = int(os.environ["PKEXEC_UID"]) + except (KeyError, ValueError): + return None + import pwd + try: + return uid, pwd.getpwuid(uid).pw_gid + except (KeyError, OverflowError): return None diff --git a/pyproject.toml b/pyproject.toml index 1a08c97..bbdde57 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -151,10 +151,12 @@ max-complexity = 26 # point is to survive exactly that. "beantester/gui/widgets/sortable_tree.py" = ["S112"] "beantester/winenv.py" = ["S112"] -# The one child process the package starts (tests/test_no_telemetry.py, SPAWN_ALLOWED): -# gsettings on Linux, a fixed argv of found programs and constants, never a shell, -# never text from outside. +# The two child processes the package starts (tests/test_no_telemetry.py, SPAWN_ALLOWED), +# both on Linux: gsettings, a fixed argv of found programs and constants, and pkexec, +# found on PATH, with this program's own command line. Never a shell, never text from +# outside. "beantester/proxy.py" = ["S603"] +"beantester/linuxenv.py" = ["S603"] # Maintainer and CI scripts: they run `git` and `python` by design, and the only # URL opened is a literal api.github.com endpoint whose path is validated first. "tools/**" = ["S603", "S607", "S310"] diff --git a/tests/test_ci_linux_pkexec_smoke.py b/tests/test_ci_linux_pkexec_smoke.py new file mode 100644 index 0000000..b8a418c --- /dev/null +++ b/tests/test_ci_linux_pkexec_smoke.py @@ -0,0 +1,77 @@ +"""tools/ci_linux_pkexec_smoke.py: its verdicts name every broken thing, and a machine +that cannot run it skips - which CI's ``--require`` turns into a failure. + +The real run is CI's (the Linux job); here only the judging, which runs anywhere. +""" +import os +import sys + +from fakes import ROOT, check + +sys.path.insert(0, os.path.join(ROOT, "tools")) + +import ci_linux_pkexec_smoke as smoke # noqa: E402 + +PERSON = {"uid": 1001, "gid": 118, "display": ":99", + "xauthority": "/tmp/xvfb-run.Ab/Xauthority", "cwd": "/w/repo"} + + +def good(keep=True): + return {"euid": 0, "pkexec_uid": "1001", "sudo_uid": None, "display_before": None, + "cwd_before": "/w/repo" if keep else "/root", "root_home": "/root", + "tk_before": "TclError: no display name and no $DISPLAY environment variable", + "taken": ["DISPLAY", "XAUTHORITY", "cwd"], "display_after": ":99", + "xauthority_after": "/tmp/xvfb-run.Ab/Xauthority", "cwd_after": "/w/repo", + "tk_after": "ok", "invoking_user": [1001, 118]} + + +def test_a_copy_that_did_its_job_passes(): + check("with --keep-cwd", smoke.judge_child("G", 0, good(), PERSON, True) == []) + check("without it", smoke.judge_child("G", 0, good(False), PERSON, False) == []) + + +def test_every_broken_copy_is_named(): + cases = {"euid": 1001, "pkexec_uid": "0", "display_before": ":99", "tk_before": "ok", + "display_after": None, "xauthority_after": None, + "tk_after": "TclError: couldn't connect to display", "cwd_after": "/root", + "invoking_user": None} + for field, value in cases.items(): + failures = smoke.judge_child("G", 0, dict(good(), **{field: value}), PERSON, True) + check(f"{field}={value!r} is named, alone", len(failures) == 1, f"({failures})") + kept = smoke.judge_child("G", 0, dict(good(), cwd_before="/root"), PERSON, True) + check("with --keep-cwd the copy starts in the person's folder", len(kept) == 1, f"({kept})") + fallback = smoke.judge_child("G", 0, dict(good(False), cwd_before="/w/repo"), PERSON, False) + check("without it in root's home - else the fallback was not what ran", + len(fallback) == 1, f"({fallback})") + check("no report", smoke.judge_child("G", 127, None, PERSON, True) + == ["G: the copy did not report (ask_for_root answered 127)"]) + + +def test_a_refusal_must_be_one_and_be_quick(): + check("refused at once: fine", smoke.judge_refusal(None, 0.05) == []) + check("ran without the rule: the canary fires", len(smoke.judge_refusal(0, 0.05)) == 1) + check("refused, but slowly", len(smoke.judge_refusal(None, 30)) == 1) + + +def test_a_machine_that_cannot_run_it_skips_and_ci_turns_that_into_a_failure(monkeypatch, capsys): + reason = smoke.skip_reason + asked = [] + + def sudo(answer): + return lambda: asked.append(answer) or answer + + check("Linux, a person, a display, pkexec, sudo: runs", + reason("linux", 1001, ":99", "/usr/bin/pkexec", sudo(True)) is None) + for args, said in ((("win32", None, ":0", "x"), "not Linux"), + (("linux", 0, ":0", "x"), "already root"), + (("linux", 1001, None, "x"), "no DISPLAY"), + (("linux", 1001, ":0", None), "pkexec")): + asked.clear() + got = reason(*args, sudo(True)) + check(f"{args}: skips, saying {said!r}, without asking sudo", + said in (got or "") and not asked, f"({got}, {asked})") + check("no sudo -n", "sudo" in (reason("linux", 1001, ":0", "x", sudo(False)) or "")) + monkeypatch.setattr(sys, "platform", "win32") + check("a skip is exit 0", smoke.main([]) == 0) + check("and with --require exit 1", smoke.main(["--require"]) == 1) + check("saying why", "SKIP: not Linux" in capsys.readouterr().out) diff --git a/tests/test_crashlog.py b/tests/test_crashlog.py index 5050f49..e6ecbe1 100644 --- a/tests/test_crashlog.py +++ b/tests/test_crashlog.py @@ -710,9 +710,10 @@ def test_the_gui_arms_native_capture_without_ever_starting_a_capture(isolated, m Reaching a real Tk root here would make it a Windows-only guard pretending to be a general one. """ - from beantester import cli, exitcodes, winenv + from beantester import cli, exitcodes, linuxenv, winenv monkeypatch.setattr(winenv, "is_windows", lambda: False) # skip the elevation dance + monkeypatch.setattr(linuxenv, "wants_root", lambda: False) # and pkexec's on Linux monkeypatch.setattr(cli, "is_frozen", lambda: False) monkeypatch.setitem(sys.modules, "tkinter", None) # import tkinter -> ImportError crashlog._arm_wanted[0] = True # what install(native=True) records @@ -1178,10 +1179,11 @@ def _gui_start(monkeypatch, frozen=False, gui=None, tkinter=None): on a runner without Tk as on this machine. ``detach_console`` is ALWAYS replaced: the real one would cut this test process off its console. """ - from beantester import cli, winenv + from beantester import cli, linuxenv, winenv shown = [] monkeypatch.setattr(winenv, "is_windows", lambda: False) # skip the elevation dance + monkeypatch.setattr(linuxenv, "wants_root", lambda: False) # and pkexec's on Linux monkeypatch.setattr(winenv, "detach_console", lambda: True) monkeypatch.setattr(winenv, "show_error", lambda title, text: shown.append((title, text))) monkeypatch.setattr(cli, "is_frozen", lambda: frozen) diff --git a/tests/test_linuxenv.py b/tests/test_linuxenv.py new file mode 100644 index 0000000..ba69ae2 --- /dev/null +++ b/tests/test_linuxenv.py @@ -0,0 +1,373 @@ +"""linuxenv: the window on Linux asks for root through pkexec (polkit). + +Nothing here starts pkexec: every call to the system is handed in. The real pkexec +runs in CI (``tools/ci_linux_pkexec_smoke.py``); what it does is in linuxenv's own +docstring. +""" +import os +import signal +import subprocess +import sys +import types + +import pytest + +from beantester import capture_backend, cli, crashlog, exitcodes, linuxenv, winenv +from fakes import check + +PKEXEC = "/usr/bin/pkexec" +USAGE = "pkexec --version |\n [--keep-cwd] [--user username] [PROGRAM] [ARGUMENTS...]\n" + + +def have(_name): + return PKEXEC + + +class _Rights: + def __init__(self, has): + self.has = has + + def has_capture_rights(self): + return self.has + + +class _Run: + """Stands in for ``subprocess.run``: records each command, answers pkexec's + usage, and ends the copy with ``code``.""" + + def __init__(self, code=0, usage=USAGE): + self.code, self.usage, self.calls = code, usage, [] + + def __call__(self, command, env=None, **kw): + self.calls.append((command, env, kw)) + if command[1:] == ["--help"]: + return subprocess.CompletedProcess(command, 0, stdout=self.usage, stderr="") + return subprocess.CompletedProcess(command, self.code) + + +@pytest.fixture +def started_from_script(monkeypatch, tmp_path): + """This test process started as ``python -u /bean_network_tester.py --gui``.""" + script = tmp_path / "bean_network_tester.py" + script.write_text("", encoding="utf-8") + monkeypatch.setattr(winenv, "is_frozen", lambda: False) + monkeypatch.setattr(sys, "executable", "/v/bin/python") + monkeypatch.setattr(sys, "orig_argv", ["/v/bin/python", "-u", str(script), "--gui"]) + monkeypatch.setattr(sys, "argv", [str(script), "--gui"]) + return os.path.abspath(str(script)) + + +def test_the_window_asks_for_root_only_where_it_can_and_root_would_help(monkeypatch): + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.delenv("BEAN_NO_ELEVATE", raising=False) + monkeypatch.setattr(capture_backend, "current", lambda: _Rights(False)) + desk = {"DISPLAY": ":0"} + check("a person at a desktop, without rights, with pkexec: asks", + linuxenv.wants_root(desk, 1000, have) is True) + check("root never asks: root without CAP_BPF (a container) would start copy after copy", + linuxenv.wants_root(desk, 0, have) is False) + check("no display: no window to ask from", linuxenv.wants_root({}, 1000, have) is False) + check("no pkexec", linuxenv.wants_root(desk, 1000, lambda name: None) is False) + monkeypatch.setenv("BEAN_NO_ELEVATE", "1") + check("BEAN_NO_ELEVATE, as on Windows", linuxenv.wants_root(desk, 1000, have) is False) + monkeypatch.delenv("BEAN_NO_ELEVATE") + monkeypatch.setattr(capture_backend, "current", lambda: _Rights(True)) + check("the capabilities are already there", linuxenv.wants_root(desk, 1000, have) is False) + monkeypatch.setattr(capture_backend, "current", lambda: _Rights(False)) + monkeypatch.setattr(sys, "platform", "win32") + check("not Linux: winenv asks there", linuxenv.wants_root(desk, 1000, have) is False) + + +def test_pkexec_is_given_only_the_locale_it_will_not_refuse(): + """pkexec exits 127 when SHELL is not in /etc/shells or a variable it keeps holds + `/`, `%` or `..` (validate_environment_variable) - before it asks anything.""" + env = {"LANG": "pl_PL.UTF-8", "LANGUAGE": "pl:en", "LC_ALL": "C/odd", "LC_TIME": "a%b", + "LC_CTYPE": "x..y", "LC_NUMERIC": "", "SHELL": "/usr/bin/fish", + "TERM": "xterm-256color", "DISPLAY": ":0", "HOME": "/u/ana", "PATH": "/usr/bin"} + given = linuxenv.pkexec_environment(env) + check("the locale that passes pkexec's rules, nothing else", + given == {"LANG": "pl_PL.UTF-8", "LANGUAGE": "pl:en"}, f"({given})") + + +def test_the_copy_is_started_the_way_this_program_was(monkeypatch, started_from_script): + script = started_from_script + cmd = linuxenv.relaunch_command(["--gui"], PKEXEC, True) + check("from sources: no agent of pkexec's own, -B added, the script absolute", + cmd == [PKEXEC, "--disable-internal-agent", "--keep-cwd", "/v/bin/python", "-B", "-u", + script, "--gui"], f"({cmd})") + old = linuxenv.relaunch_command(["--gui"], PKEXEC, False) + check("no --keep-cwd where pkexec has none", old == [PKEXEC, "--disable-internal-agent", + "/v/bin/python", "-B", "-u", script, + "--gui"], f"({old})") + monkeypatch.setattr(sys, "orig_argv", ["/v/bin/python", "-B", script]) + monkeypatch.setattr(sys, "argv", [script]) + once = linuxenv.relaunch_command([], PKEXEC, True) + check("-B given once", once.count("-B") == 1, f"({once})") + monkeypatch.setattr(sys, "orig_argv", ["/v/bin/python", "-m", "beantester"]) + monkeypatch.setattr(sys, "argv", ["/c/beantester/__main__.py"]) + module = linuxenv.relaunch_command([], PKEXEC, True) + check("python -m, with --keep-cwd", module == [PKEXEC, "--disable-internal-agent", + "--keep-cwd", "/v/bin/python", "-B", "-m", + "beantester"], f"({module})") + check("python -m without it: root's home cannot import it, so nothing to repeat", + linuxenv.relaunch_command([], PKEXEC, False) is None) + monkeypatch.setattr(winenv, "is_frozen", lambda: True) + monkeypatch.setattr(sys, "executable", "/opt/bnt/BeanNetworkTester") + frozen = linuxenv.relaunch_command(["--gui"], PKEXEC, True) + check("a frozen build: the executable and its arguments", frozen == [ + PKEXEC, "--disable-internal-agent", "--keep-cwd", "/opt/bnt/BeanNetworkTester", "--gui"], + f"({frozen})") + + +def test_what_pkexec_answers_decides_who_opens_the_window(capsys, started_from_script): + env = {"LANG": "C.UTF-8", "SHELL": "/bin/bash", "DISPLAY": ":0"} + for answer, expected in ((0, 0), (1, 1), (7, 7), (-15, 143), (126, None), (127, None)): + run = _Run(answer) + got = linuxenv.ask_for_root(["--gui"], run, have, env) + check(f"pkexec answers {answer}: {expected}", got == expected, f"({got})") + command, given, kw = run.calls[-1] + check("the copy is started through pkexec without its own agent", + command[:2] == [PKEXEC, "--disable-internal-agent"] and "--keep-cwd" in command, + f"({command})") + check("with the locale only", given == {"LANG": "C.UTF-8"}, f"({given})") + check("and keeps the terminal, for as long as it runs", kw == {}, f"({kw})") + check("a copy that did not run is said on stderr", + "did not start the window as root (exit 127)" in capsys.readouterr().err) + nothing = _Run() + check("no pkexec: nothing to ask through", + linuxenv.ask_for_root([], nothing, lambda name: None, env) is None and not nothing.calls) + + + +def test_a_pkexec_that_cannot_be_started_is_recorded_and_said(monkeypatch, capsys, + started_from_script): + noted = [] + monkeypatch.setattr(crashlog, "note", lambda exc, section: noted.append(section)) + + def broken(command, env=None, **kw): + raise OSError("gone") + check("pkexec that cannot be started: the window opens here", + linuxenv.ask_for_root([], broken, have, {}) is None) + check("recorded, and said on stderr", + "linuxenv.pkexec" in noted and "could not be started (gone)" in capsys.readouterr().err, + f"({noted})") + + started = [] + + def no_usage(command, env=None, **kw): + if command[1:] == ["--help"]: + raise subprocess.TimeoutExpired(command, 10) + started.append(command) + return subprocess.CompletedProcess(command, 3) + noted.clear() + check("a usage that cannot be read: recorded, and asked without --keep-cwd", + linuxenv.ask_for_root([], no_usage, have, {}) == 3 and noted == ["linuxenv.keep_cwd"] + and started and "--keep-cwd" not in started[0], f"({noted}, {started})") + + +def test_an_old_pkexec_and_python_m_ask_nothing(monkeypatch): + """polkit 0.105 (Ubuntu 22.04) has no --keep-cwd: `python -m beantester` from a + checkout would start in root's home and fail to import. Nothing is asked.""" + monkeypatch.setattr(winenv, "is_frozen", lambda: False) + monkeypatch.setattr(sys, "orig_argv", ["/v/bin/python", "-m", "beantester"]) + monkeypatch.setattr(sys, "argv", ["/c/beantester/__main__.py"]) + run = _Run(usage="pkexec --version |\n [--user username] PROGRAM [ARGUMENTS...]\n") + check("the window opens without root", linuxenv.ask_for_root([], run, have, {}) is None) + check("and only the usage was read", [c[0][1:] for c in run.calls] == [["--help"]], + f"({run.calls})") + + +def test_ctrl_c_while_the_copy_runs_is_waited_through(started_from_script): + """Ctrl+C in the terminal reaches the copy too. Here it must neither end the + wait (subprocess.run kills its child on an exception) nor become SIG_IGN, which + the copy would inherit through pkexec.""" + before = signal.getsignal(signal.SIGINT) + seen = [] + + def run(command, env=None, **kw): + if command[1:] == ["--help"]: + return subprocess.CompletedProcess(command, 0, stdout=USAGE, stderr="") + seen.append(signal.getsignal(signal.SIGINT)) + signal.raise_signal(signal.SIGINT) + return subprocess.CompletedProcess(command, 5) + + check("the copy's code comes back, no KeyboardInterrupt", + linuxenv.ask_for_root([], run, have, {}) == 5) + check("a handler that does nothing, not SIG_IGN", seen and seen[0] not in ( + signal.SIG_IGN, signal.SIG_DFL, signal.default_int_handler), f"({seen})") + check("and the old one is back", signal.getsignal(signal.SIGINT) is before) + + +class _Person: + """What the copy finds about the person: the parent process and their files.""" + + def __init__(self, uid=1000, env=None, cwd="/u/ana/bnt", files=(), home="/u/ana"): + self.uid, self.cwd_, self.files, self.home = uid, cwd, set(files), home + self.env = {"DISPLAY": ":0", "XAUTHORITY": "/run/user/1000/.mutter-Xwaylandauth.Q"} \ + if env is None else env + + def real_uid(self, pid): + return self.uid + + def environ(self, pid): + return dict(self.env) + + def cwd(self, pid): + return self.cwd_ + + def home_of(self, uid): + return self.home + + def owns_file(self, uid, path): + return uid == 1000 and path in self.files + + +XAUTH = "/run/user/1000/.mutter-Xwaylandauth.Q" +HOME_XAUTH = "/u/ana/.Xauthority" + + +def test_the_copy_takes_the_desktop_of_the_person_who_asked(): + asked = {"PKEXEC_UID": "1000"} + found = linuxenv.desktop_of_parent(asked, _Person(files=[XAUTH]), 42) + check("display, X authority and folder of the person's process", + found == {"DISPLAY": ":0", "XAUTHORITY": XAUTH, "cwd": "/u/ana/bnt"}, f"({found})") + check("a parent that is not the person's: nothing", + linuxenv.desktop_of_parent(asked, _Person(uid=1001, files=[XAUTH]), 42) == {}) + check("no PKEXEC_UID: nothing", linuxenv.desktop_of_parent({}, _Person(files=[XAUTH]), 42) == {}) + odd = _Person(env={"DISPLAY": "$(id):0", "XAUTHORITY": XAUTH}, files=[XAUTH]) + check("a DISPLAY that is not one: nothing", linuxenv.desktop_of_parent(asked, odd, 42) == {}) + for display in (":1", "localhost:10.0", "unix:0"): + ok = linuxenv.desktop_of_parent(asked, _Person(env={"DISPLAY": display}), 42) + check(f"{display!r} is a display", ok.get("DISPLAY") == display, f"({ok})") + foreign = linuxenv.desktop_of_parent(asked, _Person(files=[HOME_XAUTH]), 42) + check("an X authority file that is not the person's: ~/.Xauthority instead", + foreign.get("XAUTHORITY") == HOME_XAUTH, f"({foreign})") + x11 = linuxenv.desktop_of_parent(asked, _Person(env={"DISPLAY": ":0"}, files=[HOME_XAUTH]), 42) + check("an X11 session that sets none: ~/.Xauthority, as pkexec itself would", + x11.get("XAUTHORITY") == HOME_XAUTH, f"({x11})") + relative = _Person(env={"DISPLAY": ":0", "XAUTHORITY": "x/.Xauthority"}, files=["x/.Xauthority"]) + check("a relative path is not taken", + "XAUTHORITY" not in linuxenv.desktop_of_parent(asked, relative, 42)) + gone = linuxenv.desktop_of_parent(asked, _Person(cwd=None, files=[XAUTH]), 42) + check("a folder that cannot be read is left", "cwd" not in gone and gone["DISPLAY"] == ":0") + + +def test_only_the_copy_pkexec_started_takes_anything(monkeypatch): + moved = [] + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.setattr(os, "geteuid", lambda: 0, raising=False) + monkeypatch.setattr(os, "chdir", moved.append) + env = {"PKEXEC_UID": "1000"} + taken = linuxenv.take_desktop(env, _Person(files=[XAUTH]), 42) + check("the copy: display and X authority set, the folder entered", + taken == ["DISPLAY", "XAUTHORITY", "cwd"] and env["DISPLAY"] == ":0" + and env["XAUTHORITY"] == XAUTH and moved == ["/u/ana/bnt"], f"({taken}, {env}, {moved})") + moved.clear() + sudo = {"PKEXEC_UID": "1000", "DISPLAY": ":5"} + check("a DISPLAY already there (sudo keeps it): left as it is", + linuxenv.take_desktop(sudo, _Person(files=[XAUTH]), 42) == [] and sudo["DISPLAY"] == ":5" + and not moved) + check("not under pkexec: nothing", linuxenv.take_desktop({}, _Person(files=[XAUTH]), 42) == []) + monkeypatch.setattr(os, "geteuid", lambda: 1000, raising=False) + check("not root: nothing", linuxenv.take_desktop({"PKEXEC_UID": "1000"}, + _Person(files=[XAUTH]), 42) == []) + + +def test_the_parent_is_read_from_its_proc_entries(tmp_path): + entry = tmp_path / "4242" + entry.mkdir() + (entry / "status").write_text("Name:\tpython\nUid:\t1000\t1001\t1002\t1003\n", encoding="ascii") + (entry / "environ").write_bytes(b"A=1\0DISPLAY=:0\0DISPLAY=:9\0junk\0X=a=b\0") + system = linuxenv._System(str(tmp_path)) + check("the REAL uid", system.real_uid(4242) == 1000) + check("the environment, first value of a name wins, no '=' skipped", + system.environ(4242) == {"A": "1", "DISPLAY": ":0", "X": "a=b"}, f"({system.environ(4242)})") + check("a process that is gone: nothing", system.real_uid(9999) is None + and system.environ(9999) == {} and system.cwd(9999) is None) + + +@pytest.mark.skipif(not hasattr(os, "getuid"), reason="file owners are a POSIX question") +def test_an_x_authority_file_is_a_regular_file_of_the_person(tmp_path): + mine = tmp_path / "Xauthority" + mine.write_bytes(b"cookie") + link = tmp_path / "link" + link.symlink_to(mine) + owns = linuxenv._System.owns_file + check("a regular file of theirs", owns(os.getuid(), str(mine)) is True) + check("not someone else's", owns(os.getuid() + 1, str(mine)) is False) + check("not a link to one", owns(os.getuid(), str(link)) is False) + check("not a folder", owns(os.getuid(), str(tmp_path)) is False) + proc = tmp_path / "proc" / "77" + proc.mkdir(parents=True) + (proc / "cwd").symlink_to(tmp_path) + check("the parent's folder through its cwd link", + linuxenv._System(str(tmp_path / "proc")).cwd(77) == str(tmp_path)) + + +def _fake_tk(tk_fails): + tkinter = types.ModuleType("tkinter") + + class TclError(Exception): + pass + + def tk(): + if tk_fails: + raise TclError("no display name and no $DISPLAY environment variable") + return types.SimpleNamespace(mainloop=lambda: None) + + tkinter.TclError, tkinter.Tk = TclError, tk + return tkinter + + +def _gui(monkeypatch, tk_fails=True, app_fails=False): + """cli._run_gui up to the window, with nothing real; returns the window's failures + and the times the copy's desktop was taken.""" + failed, taken = [], [] + gui = types.ModuleType("beantester.gui") + + def app(root): + if app_fails: + raise sys.modules["tkinter"].TclError("a widget option") + gui.App = app + monkeypatch.setattr(winenv, "is_windows", lambda: False) + monkeypatch.setattr(winenv, "set_dpi_awareness", lambda: None) + monkeypatch.setattr(crashlog, "arm_native", lambda: None) + monkeypatch.setattr(cli, "is_frozen", lambda: False) + monkeypatch.setattr(cli, "_window_did_not_open", + lambda exc: failed.append(type(exc).__name__) or exitcodes.RUNTIME) + monkeypatch.setattr(linuxenv, "take_desktop", lambda: taken.append("taken") or []) + monkeypatch.setitem(sys.modules, "tkinter", _fake_tk(tk_fails)) + monkeypatch.setitem(sys.modules, "beantester.gui", gui) + return failed, taken + + +def test_the_window_waits_for_the_copy_and_opens_here_only_when_it_did_not_run(monkeypatch): + _failed, taken = _gui(monkeypatch, tk_fails=False) + asked = [] + monkeypatch.setattr(linuxenv, "wants_root", lambda: True) + monkeypatch.setattr(linuxenv, "ask_for_root", lambda argv: asked.append(argv) or 7) + check("the copy ran: its code, no window here", cli._run_gui(["--gui"]) == 7 + and asked == [["--gui"]] and not taken, f"({asked}, {taken})") + monkeypatch.setattr(linuxenv, "ask_for_root", lambda argv: None) + check("it did not run: the window opens here", cli._run_gui([]) == exitcodes.OK) + check("and the screen is taken before Tk (a no-op outside pkexec's copy)", + taken == ["taken"], f"({taken})") + monkeypatch.setattr(linuxenv, "wants_root", lambda: False) + monkeypatch.setattr(linuxenv, "ask_for_root", lambda argv: asked.append("asked") or 7) + asked.clear() + check("nothing to ask: nothing asked", cli._run_gui([]) == exitcodes.OK and not asked) + + +def test_a_copy_that_cannot_reach_the_screen_hands_the_window_back(monkeypatch): + failed, _taken = _gui(monkeypatch, tk_fails=True) + monkeypatch.setattr(linuxenv, "wants_root", lambda: False) + monkeypatch.setattr(linuxenv, "started_by_pkexec", lambda: True) + check("pkexec's copy without a screen: pkexec's own 'did not run'", + cli._run_gui([]) == linuxenv.NO_WINDOW == 127) + check("and it still says why", failed == ["TclError"], f"({failed})") + monkeypatch.setattr(linuxenv, "started_by_pkexec", lambda: False) + check("a window that cannot open anywhere else: as before", + cli._run_gui([]) == exitcodes.RUNTIME) + _gui(monkeypatch, tk_fails=False, app_fails=True) + monkeypatch.setattr(linuxenv, "started_by_pkexec", lambda: True) + check("a fault in the window itself is not handed back", cli._run_gui([]) == exitcodes.RUNTIME) diff --git a/tests/test_mutation_registry.py b/tests/test_mutation_registry.py index 5932871..b9c334f 100644 --- a/tests/test_mutation_registry.py +++ b/tests/test_mutation_registry.py @@ -1330,8 +1330,8 @@ # earlier session of leaving nothing at all behind. "label": "crash: the GUI entry point stops arming native capture", "file": "beantester/cli.py", - "old": " crashlog.arm_native()\n try:\n import tkinter as tk", - "new": " try:\n import tkinter as tk", + "old": " crashlog.arm_native()\n root = None\n try:\n import tkinter as tk", + "new": " root = None\n try:\n import tkinter as tk", "test": "test_the_gui_arms_native_capture_without_ever_starting_a_capture", }, { @@ -7410,7 +7410,12 @@ " root = tk.Tk()\n" " App(root)\n" " except Exception as exc:\n" - " return _window_did_not_open(exc)\n", + " code = _window_did_not_open(exc)\n" + " # A copy pkexec started whose Tk cannot reach the person's display answers\n" + " # pkexec's own \"did not run\": the person's process then opens the window.\n" + " if root is None and type(exc).__name__ == \"TclError\" and linuxenv.started_by_pkexec():\n" + " return linuxenv.NO_WINDOW\n" + " return code\n", "new": " except Exception as exc:\n" " return _window_did_not_open(exc)\n" " winenv.set_dpi_awareness()\n" @@ -9929,6 +9934,223 @@ "new": " return 0 if counts.get(\"SURVIVED\", 0) == 0 and canary_ok else 1\n", "test": "test_an_entry_that_does_not_compile_fails_every_kind_of_run", }, + # -- Linux: the window asks for root through pkexec (linuxenv.py) --------- # + { + # Root that cannot capture (a container without CAP_BPF) starts copy after copy. + "label": "pkexec: root asks for root again", + "file": "beantester/linuxenv.py", + "old": " if (os.geteuid() if euid is None else euid) == 0:\n return False\n", + "new": "", + "test": "test_the_window_asks_for_root_only_where_it_can_and_root_would_help", + }, + { + "label": "pkexec: asked for with no display to ask from", + "file": "beantester/linuxenv.py", + "old": " if not environ.get(\"DISPLAY\"):\n return False\n", + "new": "", + "test": "test_the_window_asks_for_root_only_where_it_can_and_root_would_help", + }, + { + "label": "pkexec: asked for with the capabilities already there", + "file": "beantester/linuxenv.py", + "old": " if capture_backend.current().has_capture_rights():\n", + "new": " if False:\n", + "test": "test_the_window_asks_for_root_only_where_it_can_and_root_would_help", + }, + { + # An automated GUI run on Linux would hang on a dialog nobody can answer. + "label": "pkexec: BEAN_NO_ELEVATE is not read", + "file": "beantester/linuxenv.py", + "old": " if not is_linux() or winenv.elevation_disabled():\n", + "new": " if not is_linux():\n", + "test": "test_the_window_asks_for_root_only_where_it_can_and_root_would_help", + }, + { + # pkexec exits 127 before it asks anything (validate_environment_variable). + "label": "pkexec: a locale value pkexec refuses is handed to it", + "file": "beantester/linuxenv.py", + "old": " if environ.get(name) and not any(bad in environ[name] for bad in _SUSPICIOUS)}", + "new": " if environ.get(name)}", + "test": "test_pkexec_is_given_only_the_locale_it_will_not_refuse", + }, + { + "label": "pkexec: the copy is handed the whole environment", + "file": "beantester/linuxenv.py", + "old": " code = run(command, env=env).returncode\n", + "new": " code = run(command).returncode\n", + "test": "test_what_pkexec_answers_decides_who_opens_the_window", + }, + { + # Root leaves cache folders in the person's checkout they cannot delete. + "label": "pkexec: the copy runs without -B", + "file": "beantester/linuxenv.py", + "old": " if \"-B\" not in head:\n args = [\"-B\"] + args\n", + "new": "", + "test": "test_the_copy_is_started_the_way_this_program_was", + }, + { + # In root's home `python -m beantester` cannot import itself: no window at all. + "label": "pkexec: python -m repeated without --keep-cwd", + "file": "beantester/linuxenv.py", + "old": " if \"-m\" in head and not keep_cwd:\n return None\n", + "new": "", + "test": "test_the_copy_is_started_the_way_this_program_was", + }, + { + # Measured in WSL: with a terminal and no agent, pkexec asks in the terminal and waits. + "label": "pkexec: its own terminal agent is allowed", + "file": "beantester/linuxenv.py", + "old": " return ([pkexec, \"--disable-internal-agent\"] + ", + "new": " return ([pkexec] + ", + "test": "test_the_copy_is_started_the_way_this_program_was", + }, + { + "label": "pkexec: --keep-cwd is assumed, not read from its usage", + "file": "beantester/linuxenv.py", + "old": " return \"--keep-cwd\" in (done.stdout or \"\") + (done.stderr or \"\")\n", + "new": " return True\n", + "test": "test_an_old_pkexec_and_python_m_ask_nothing", + }, + { + # A dismissed dialog left the person with no window at all. + "label": "pkexec: 126/127 taken for the copy's own exit code", + "file": "beantester/linuxenv.py", + "old": " if code in NOT_STARTED:\n", + "new": " if False:\n", + "test": "test_what_pkexec_answers_decides_who_opens_the_window", + }, + { + "label": "pkexec: a copy killed by a signal answers a negative code", + "file": "beantester/linuxenv.py", + "old": " return 128 - code if code < 0 else code\n", + "new": " return code\n", + "test": "test_what_pkexec_answers_decides_who_opens_the_window", + }, + { + # SIG_IGN is inherited through pkexec: the copy would ignore Ctrl+C for good. + "label": "pkexec: Ctrl+C is waited through with SIG_IGN", + "file": "beantester/linuxenv.py", + "old": " previous = signal.signal(signal.SIGINT, _wait_through)\n", + "new": " previous = signal.signal(signal.SIGINT, signal.SIG_IGN)\n", + "test": "test_ctrl_c_while_the_copy_runs_is_waited_through", + }, + { + "label": "pkexec: the desktop is taken from a parent that is not the person's", + "file": "beantester/linuxenv.py", + "old": " if uid is None or system.real_uid(ppid) != uid:\n", + "new": " if uid is None:\n", + "test": "test_the_copy_takes_the_desktop_of_the_person_who_asked", + }, + { + "label": "pkexec: any DISPLAY value is taken", + "file": "beantester/linuxenv.py", + "old": " if not _DISPLAY.fullmatch(display):\n", + "new": " if not display:\n", + "test": "test_the_copy_takes_the_desktop_of_the_person_who_asked", + }, + { + "label": "pkexec: an X authority file that is not the person's is taken", + "file": "beantester/linuxenv.py", + "old": " if xauthority and posixpath.isabs(xauthority) and system.owns_file(uid, xauthority):\n", + "new": " if xauthority and posixpath.isabs(xauthority):\n", + "test": "test_the_copy_takes_the_desktop_of_the_person_who_asked", + }, + { + # An X11 session sets no XAUTHORITY; pkexec itself would look in ~/.Xauthority. + "label": "pkexec: the ~/.Xauthority fallback is never looked at", + "file": "beantester/linuxenv.py", + "old": "home and posixpath.join(home, \".Xauthority\")):\n", + "new": "None):\n", + "test": "test_the_copy_takes_the_desktop_of_the_person_who_asked", + }, + { + "label": "pkexec: a pkexec that cannot be started is not recorded", + "file": "beantester/linuxenv.py", + "old": " crashlog.note(exc, \"linuxenv.pkexec\")\n", + "new": "", + "test": "test_a_pkexec_that_cannot_be_started_is_recorded_and_said", + }, + { + "label": "pkexec: a DISPLAY sudo kept is replaced", + "file": "beantester/linuxenv.py", + "old": " if not started_by_pkexec(environ) or environ.get(\"DISPLAY\"):\n", + "new": " if not started_by_pkexec(environ):\n", + "test": "test_only_the_copy_pkexec_started_takes_anything", + }, + { + "label": "pkexec: a process that is not root takes a desktop", + "file": "beantester/linuxenv.py", + "old": " return is_linux() and os.geteuid() == 0 and _uid(environ.get(\"PKEXEC_UID\")) is not None\n", + "new": " return is_linux() and _uid(environ.get(\"PKEXEC_UID\")) is not None\n", + "test": "test_only_the_copy_pkexec_started_takes_anything", + }, + { + # The effective uid of a setuid parent is not the person's. + "label": "pkexec: the parent's effective uid read for its real one", + "file": "beantester/linuxenv.py", + "old": " return _uid(fields[1])\n", + "new": " return _uid(fields[2])\n", + "test": "test_the_parent_is_read_from_its_proc_entries", + }, + { + "label": "pkexec: a later DISPLAY in the parent's environment wins", + "file": "beantester/linuxenv.py", + "old": " if sep and name not in found:\n", + "new": " if sep:\n", + "test": "test_the_parent_is_read_from_its_proc_entries", + }, + { + # polkit 124 (24.04) and 0.105 (22.04): every file the window saved stays root's. + "label": "paths: PKEXEC_UID does not name the person", + "file": "beantester/paths.py", + "old": " except (KeyError, ValueError):\n return _pkexec_user()\n", + "new": " except (KeyError, ValueError):\n return None\n", + "test": "test_who_ran_it_is_read_from_pkexec_where_polkit_sets_no_sudo_uid", + }, + { + "label": "paths: the pkexec person's group is taken to be their uid", + "file": "beantester/paths.py", + "old": " return uid, pwd.getpwuid(uid).pw_gid\n", + "new": " return uid, uid\n", + "test": "test_who_ran_it_is_read_from_pkexec_where_polkit_sets_no_sudo_uid", + }, + { + "label": "gui start: Linux never asks for root", + "file": "beantester/cli.py", + "old": " elif linuxenv.wants_root():\n", + "new": " elif False:\n", + "test": "test_the_window_waits_for_the_copy_and_opens_here_only_when_it_did_not_run", + }, + { + # A second window, without root, after the root one closed. + "label": "gui start: the window opens again after the copy ran", + "file": "beantester/cli.py", + "old": " code = linuxenv.ask_for_root(argv)\n if code is not None:\n return code\n", + "new": " code = linuxenv.ask_for_root(argv)\n", + "test": "test_the_window_waits_for_the_copy_and_opens_here_only_when_it_did_not_run", + }, + { + "label": "gui start: the copy never takes the person's screen", + "file": "beantester/cli.py", + "old": " linuxenv.take_desktop() # in the copy pkexec started: the person's screen\n", + "new": "", + "test": "test_the_window_waits_for_the_copy_and_opens_here_only_when_it_did_not_run", + }, + { + # Authenticated, and then no window anywhere. + "label": "gui start: pkexec's copy without a screen does not hand the window back", + "file": "beantester/cli.py", + "old": " if root is None and type(exc).__name__ == \"TclError\" and linuxenv.started_by_pkexec():\n", + "new": " if False:\n", + "test": "test_a_copy_that_cannot_reach_the_screen_hands_the_window_back", + }, + { + "label": "gui start: a fault in the window itself is handed back as well", + "file": "beantester/cli.py", + "old": " if root is None and type(exc).__name__ == \"TclError\" and linuxenv.started_by_pkexec():\n", + "new": " if type(exc).__name__ == \"TclError\" and linuxenv.started_by_pkexec():\n", + "test": "test_a_copy_that_cannot_reach_the_screen_hands_the_window_back", + }, ] # The runner's own check: a patch that cannot compile must be reported as BROKEN, not @@ -9995,6 +10217,10 @@ "test_the_command_survives_bash_with_history_expansion_on": "2026-10-28, WSL Ubuntu: red on the double-quoted command (`!`, `$( )`, a " "backtick, `$NAME`, two backslashes), green on the single-quoted one", + # File owners and symlinks: POSIX-only, skipped on the Windows mutation job. + "test_an_x_authority_file_is_a_regular_file_of_the_person": + "2026-10-28, WSL Ubuntu, four breaks of linuxenv._System each caught (any owner, " + "a link followed, a folder, the cwd link unread), green again restored", } # No mutation at all. Naming them is the point: an unproven guard and a guard nobody diff --git a/tests/test_no_telemetry.py b/tests/test_no_telemetry.py index eb2d20b..2522fa2 100644 --- a/tests/test_no_telemetry.py +++ b/tests/test_no_telemetry.py @@ -126,6 +126,8 @@ "list2cmdline, a pure string helper - nothing here spawns anything", ("beantester/proxy.py", "subprocess"): "Linux: gsettings, which reads the desktop's proxy settings (below)", + ("beantester/linuxenv.py", "subprocess"): + "Linux: pkexec, which asks for root and starts the window again (below)", } # The exceptions, each with the reason it exists. A call on one of these modules @@ -632,6 +634,12 @@ def test_the_wire_allowlist_names_a_place_not_a_function(): "2026-10-27 (strace -f -e connect,socket,execve, Ubuntu 26.04): its only " "sockets were AF_UNIX ones to the system's account database, no D-Bus, no " "AF_INET or AF_INET6", + ("beantester/linuxenv.py", "subprocess.run"): + "Linux: pkexec from PATH, twice at most per start of the window - `pkexec " + "--help` (10 s limit) to see whether it takes --keep-cwd, then pkexec running " + "this same program as root, waited for while it runs. A fixed argv, an " + "environment of the locale variables only, no shell. pkexec itself talks to " + "polkitd over the system D-Bus (AF_UNIX); nothing here opens AF_INET or AF_INET6", } @@ -640,12 +648,12 @@ def unallowed_spawns(found): if (rel, detail) not in SPAWN_ALLOWED] -def test_the_shipped_package_spawns_only_the_one_named_process(): - """No child process but the one named above, so no shelling out to curl, - bitsadmin or PowerShell.""" +def test_the_shipped_package_spawns_only_the_named_processes(): + """No child process but the ones named above (gsettings, pkexec), so no shelling + out to curl, bitsadmin or PowerShell.""" found = _scan().get("spawn", []) bad = ["%s:%d %s" % (rel, line, detail) for rel, detail, line in unallowed_spawns(found)] - check("the shipped package spawns no process but the named one", not bad, f"({bad})") + check("the shipped package spawns no process but the named ones", not bad, f"({bad})") seen = {(rel, detail) for rel, detail, _line in found} stale = sorted(str(key) for key in SPAWN_ALLOWED if key not in seen) check("every allowed process is still started where it is allowed", not stale, f"({stale})") diff --git a/tests/test_sudo_files.py b/tests/test_sudo_files.py index 445ebe3..d2de1bd 100644 --- a/tests/test_sudo_files.py +++ b/tests/test_sudo_files.py @@ -57,6 +57,7 @@ def fake_chown(path, uid, gid, *, follow_symlinks=True): monkeypatch.setattr(os, "stat", fake_stat) monkeypatch.setenv("SUDO_UID", "1000") monkeypatch.setenv("SUDO_GID", "1000") + monkeypatch.delenv("PKEXEC_UID", raising=False) # sudo, not pkexec return types.SimpleNamespace(chowned=chowned, followed=followed, owners=owners) @@ -82,6 +83,35 @@ def test_who_ran_it_is_read_from_sudo_and_only_by_root(monkeypatch, sudo): check("Windows: never root", paths.running_as_root() is False) +def test_who_ran_it_is_read_from_pkexec_where_polkit_sets_no_sudo_uid(monkeypatch, sudo): + """pkexec sets SUDO_UID and SUDO_GID from polkit 127 (Ubuntu 26.04) only. Before + that (124 on 24.04, 0.105 on 22.04) PKEXEC_UID names the person, and without it + every file the window saved stayed root's.""" + def getpwuid(uid): + if uid != 1000: + raise KeyError(uid) + return types.SimpleNamespace(pw_gid=1003) + + monkeypatch.setitem(sys.modules, "pwd", types.SimpleNamespace(getpwuid=getpwuid)) + monkeypatch.delenv("SUDO_UID") + monkeypatch.delenv("SUDO_GID") + monkeypatch.setenv("PKEXEC_UID", "1000") + check("polkit 124: the person from PKEXEC_UID, the group from the account database", + paths.invoking_user() == (1000, 1003), f"({paths.invoking_user()})") + monkeypatch.setenv("SUDO_UID", "1000") + monkeypatch.setenv("SUDO_GID", "1000") + check("sudo's pair where it is there (polkit 127 sets both)", + paths.invoking_user() == (1000, 1000)) + monkeypatch.delenv("SUDO_UID") + monkeypatch.setenv("PKEXEC_UID", "1001") + check("an account the system does not know: nobody", paths.invoking_user() is None) + monkeypatch.setenv("PKEXEC_UID", "x") + check("or not a number", paths.invoking_user() is None) + monkeypatch.setenv("PKEXEC_UID", "1000") + monkeypatch.setattr(os, "geteuid", lambda: 1000, raising=False) + check("not root: nobody to hand anything to", paths.invoking_user() is None) + + def test_a_file_is_handed_back_only_inside_the_persons_own_directory(tmp_path, sudo): mine, roots = tmp_path / "mine", tmp_path / "roots" mine.mkdir() diff --git a/tools/ci_linux_pkexec_smoke.py b/tools/ci_linux_pkexec_smoke.py new file mode 100644 index 0000000..3f77196 --- /dev/null +++ b/tools/ci_linux_pkexec_smoke.py @@ -0,0 +1,218 @@ +#!/usr/bin/env python3 +"""The window asking for root through pkexec on Linux, with the real pkexec and polkitd. + +Why this exists +--------------- +``beantester/linuxenv.py`` starts the window again as root through pkexec when it +opens without root, and the copy takes the person's display back from the process +that asked. Its tests hand every call to the system in; this runs the real thing on +GitHub's ubuntu runner (Ubuntu 24.04: polkit 124, which names the person only by +``PKEXEC_UID`` - the fallback ``paths.invoking_user`` needs there). + +What it checks +-------------- +* REFUSED - no rule yet, no agent to ask through: ``linuxenv.ask_for_root`` must + answer None (pkexec's 127) within seconds. The canary: if pkexec ran without + the rule, the phases below would prove nothing. +* ``linuxenv.wants_root`` says yes on this machine: not root, no capabilities, a + display (Xvfb), pkexec there. +* GRANTED - a polkit rule lets this runner's user run pkexec without a password, + standing in for the person typing it (the runner is thrown away after the job). + The copy, run as root by the product's own ``ask_for_root``, reports: pkexec took + ``DISPLAY`` away and Tk could not open (the problem is real here), then + ``take_desktop`` gave back this process's display, X authority and folder, Tk + opened, and ``invoking_user`` named this runner's user. +* The same once more as polkit 0.105 runs it, without ``--keep-cwd``: the copy + starts in root's home and must enter this process's folder by itself. + +Not Linux, root, no display, no pkexec or no ``sudo -n`` -> SKIP and exit 0, unless +``--require`` is given, which CI passes. +""" +import argparse +import json +import os +import shutil +import subprocess +import sys +import tempfile +import time + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, ROOT) + +RULE = "/etc/polkit-1/rules.d/49-bnt-ci-pkexec.rules" +RULE_TEXT = """polkit.addRule(function (action, subject) { + if (action.id == "org.freedesktop.policykit.exec" && subject.user == "%s") { + return polkit.Result.YES; + } +}); +""" +REFUSAL_SECONDS = 10 # pkexec without an agent answers in milliseconds (measured: 19-76 ms) +RULE_WAIT_SECONDS = 15 # polkitd reads a new rule file by itself + + +# -- verdicts (pure, tested in tests/test_ci_linux_pkexec_smoke.py) ------------------ # +def skip_reason(platform, euid, display, pkexec, can_sudo): + """Why this machine cannot run the check, or None. ``can_sudo`` is asked last + and only on Linux: Windows 11 has a ``sudo`` of its own, which shows UAC.""" + if not platform.startswith("linux"): + return "not Linux" + if euid == 0: + return "already root: there is nothing to ask for" + if not display: + return "no DISPLAY (run it under xvfb-run)" + if not pkexec: + return "pkexec is not installed" + if not can_sudo(): + return "sudo -n cannot write the polkit rule" + return None + + +def judge_refusal(code, seconds): + failures = [] + if code is not None: + failures.append(f"REFUSED: pkexec ran the copy without the rule (answered {code}) - " + "the granted phases would prove nothing") + if seconds > REFUSAL_SECONDS: + failures.append(f"REFUSED: pkexec took {seconds:.1f} s to refuse, not milliseconds") + return failures + + +def judge_child(phase, code, report, person, keep_cwd): + """``person``: uid, gid, display, xauthority and cwd of this process.""" + if code != 0 or report is None: + return [f"{phase}: the copy did not report (ask_for_root answered {code})"] + failures = [] + + def want(name, got, expected): + if got != expected: + failures.append(f"{phase}: {name} is {got!r}, expected {expected!r}") + + want("euid", report.get("euid"), 0) + want("PKEXEC_UID", report.get("pkexec_uid"), str(person["uid"])) + want("DISPLAY before the take (pkexec drops it)", report.get("display_before"), None) + if not str(report.get("tk_before", "")).startswith("TclError"): + failures.append(f"{phase}: Tk opened before the take ({report.get('tk_before')!r}) - " + "the take is not what made it work") + want("DISPLAY taken", report.get("display_after"), person["display"]) + want("XAUTHORITY taken", report.get("xauthority_after"), person["xauthority"]) + want("Tk after the take", report.get("tk_after"), "ok") + want("folder before the take", report.get("cwd_before"), + person["cwd"] if keep_cwd else report.get("root_home")) + want("folder after the take", report.get("cwd_after"), person["cwd"]) + want("invoking_user", report.get("invoking_user"), [person["uid"], person["gid"]]) + return failures + + +# -- the copy pkexec starts -------------------------------------------------------- # +def child_report(): + """Run as root by pkexec: what the copy finds, before and after taking the desktop.""" + import pwd + + from beantester import linuxenv, paths + report = {"euid": os.geteuid(), "pkexec_uid": os.environ.get("PKEXEC_UID"), + "sudo_uid": os.environ.get("SUDO_UID"), + "display_before": os.environ.get("DISPLAY"), "cwd_before": os.getcwd(), + "root_home": pwd.getpwuid(0).pw_dir} + report["tk_before"] = _tk() + report["taken"] = linuxenv.take_desktop() + report.update(display_after=os.environ.get("DISPLAY"), + xauthority_after=os.environ.get("XAUTHORITY"), cwd_after=os.getcwd(), + tk_after=_tk(), invoking_user=list(paths.invoking_user() or []) or None) + return report + + +def _tk(): + try: + import tkinter + root = tkinter.Tk() + root.update() + root.destroy() + except Exception as exc: + return f"{type(exc).__name__}: {exc}" + return "ok" + + +# -- the person's side ------------------------------------------------------------- # +def _sudo(*argv, stdin=None): + return subprocess.run(["sudo", "-n", *argv], input=stdin, capture_output=True, + text=True, timeout=30, check=False) + + +def _rule_takes(pkexec): + deadline = time.monotonic() + RULE_WAIT_SECONDS + while time.monotonic() < deadline: + done = subprocess.run([pkexec, "--disable-internal-agent", "/bin/true"], + env={}, capture_output=True, timeout=30, check=False) + if done.returncode == 0: + return True + time.sleep(0.5) + return False + + +def _ask(linuxenv, report_path): + code = linuxenv.ask_for_root(["--child", report_path]) + try: + with open(report_path, encoding="utf-8") as f: + return code, json.load(f) + except (OSError, ValueError): + return code, None + + +def run_phases(pkexec): + from beantester import linuxenv + failures = [] + person = {"uid": os.getuid(), "gid": os.getgid(), "display": os.environ.get("DISPLAY"), + "xauthority": os.environ.get("XAUTHORITY"), "cwd": os.getcwd()} + work = tempfile.mkdtemp(prefix="bnt-pkexec-") + _sudo("rm", "-f", RULE) + started = time.monotonic() + code = linuxenv.ask_for_root(["--child", os.path.join(work, "refused.json")]) + failures += judge_refusal(code, time.monotonic() - started) + print(f"REFUSED: ask_for_root answered {code}", flush=True) + if not linuxenv.wants_root(): + failures.append("wants_root says no on a runner without root, with a display and pkexec") + import pwd + _sudo("tee", RULE, stdin=RULE_TEXT % pwd.getpwuid(os.getuid()).pw_name) + try: + if not _rule_takes(pkexec): + return failures + [f"polkitd did not take {RULE} within {RULE_WAIT_SECONDS} s"] + for number, (phase, keep) in enumerate((("GRANTED", True), + ("GRANTED, no --keep-cwd", False))): + if not keep: + # As polkit 0.105's pkexec (Ubuntu 22.04), which has no --keep-cwd. + linuxenv._keeps_cwd = lambda *a: False + code, report = _ask(linuxenv, os.path.join(work, f"granted-{number}.json")) + print(f"{phase}: ask_for_root answered {code}, report {json.dumps(report)}", flush=True) + failures += judge_child(phase, code, report, person, keep) + finally: + _sudo("rm", "-f", RULE) + return failures + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("--require", action="store_true", + help="a machine that cannot run the check fails instead of skipping") + parser.add_argument("--child", metavar="REPORT", help=argparse.SUPPRESS) + args = parser.parse_args(argv) + if args.child: + with open(args.child, "w", encoding="utf-8") as f: + json.dump(child_report(), f) + return 0 + pkexec = shutil.which("pkexec") if sys.platform.startswith("linux") else None + reason = skip_reason(sys.platform, getattr(os, "geteuid", lambda: None)(), + os.environ.get("DISPLAY"), pkexec, + lambda: bool(shutil.which("sudo")) and _sudo("true").returncode == 0) + if reason: + print(f"SKIP: {reason}") + return 1 if args.require else 0 + failures = run_phases(pkexec) + for failure in failures: + print(f"FAIL {failure}") + print("pkexec smoke: " + ("FAILED" if failures else "OK")) + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main())