From 59cdf0f127dd6bbbcfe044bbe17f2b7ed951c9a4 Mon Sep 17 00:00:00 2001 From: r-vdp Date: Thu, 16 Apr 2026 20:12:01 +0200 Subject: [PATCH] nixos-rebuild-ng: introduce Elevator abstraction Adds elevate.py with an Elevator base, NoElevator and SudoElevator. Nothing uses it yet. This is a pure addition so the next commit can swap run_wrapper over without mixing new code with the move. The SudoElevator wrapping is byte-for-byte what run_wrapper currently emits (including NIX_SUDOOPTS handling), so behaviour is unchanged once the switch happens. Motivated by #507054: threading a second `run0: bool` alongside `sudo: bool` through every signature does not scale and makes invalid combinations representable. --- .../src/nixos_rebuild/elevate.py | 190 ++++++++++++++++++ .../src/tests/test_elevate.py | 50 +++++ 2 files changed, 240 insertions(+) create mode 100644 pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/elevate.py create mode 100644 pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_elevate.py diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/elevate.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/elevate.py new file mode 100644 index 000000000000..e75fd099d678 --- /dev/null +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/elevate.py @@ -0,0 +1,190 @@ +"""Privilege-elevation backends for activation commands. + +An :class:`Elevator` describes how to wrap a command so it runs as root, +both on the local machine and on a remote target host over SSH (where no +controlling terminal is available), and how to feed it a pre-supplied +password when the backend supports that. ``run_wrapper`` and its callers +carry a single ``elevate: Elevator`` value and let it produce the command +prefix and stdin. +""" + +from __future__ import annotations + +import os +import shlex +from abc import ABC, abstractmethod +from dataclasses import dataclass, field, replace +from enum import Enum +from typing import Final, Self, override + + +@dataclass(frozen=True) +class Wrapped: + """Result of wrapping a command for elevation.""" + + #: Arguments to prepend to the command. Kept as ``list[str]`` rather than + #: the wider ``process.Args`` because elevators only ever prepend plain + #: strings; callers splice this into their existing ``Args`` list. + prefix: list[str] + #: Text to send on the wrapped command's stdin (typically a password + #: followed by a newline), or ``None`` to leave stdin alone. + stdin: str | None = None + + +class Elevator(ABC): + """How to gain root for activation commands.""" + + #: CLI name, e.g. ``sudo`` or ``run0``. + name: str + + @property + def elevates(self) -> bool: + """Whether this elevator actually changes privileges. + + ``run_wrapper`` uses this to decide between passing ``env`` to + :func:`subprocess.run` directly (unprivileged local case) and + injecting it via ``env -i`` inside the wrapped command (where the + elevator may otherwise scrub the environment). + """ + return True + + @abstractmethod + def wrap_local(self) -> Wrapped: + """Wrap a command run on the local machine.""" + + @abstractmethod + def wrap_remote(self) -> Wrapped: + """Wrap a command run on a target host over SSH. + + The remote side has no controlling terminal, so backends that need + interactive prompts must either accept a pre-supplied password + (see :meth:`with_password`) or rely on a passwordless policy on + the target. + """ + + @abstractmethod + def with_password(self, password: str) -> Self: + """Return a copy that will feed *password* to the backend. + + Backends that have no stdin path for credentials must raise + :class:`ElevateError` with a hint pointing at the alternative + (e.g. a polkit rule). + """ + + def on_remote_failure(self) -> str | None: + """Optional hint to print when a remote elevated command fails.""" + return None + + +class ElevateError(Exception): + """Raised for invalid elevator/flag combinations.""" + + +@dataclass(frozen=True) +class NoElevator(Elevator): + name: str = "none" + + @property + @override + def elevates(self) -> bool: + return False + + @override + def wrap_local(self) -> Wrapped: + return Wrapped(prefix=[]) + + @override + def wrap_remote(self) -> Wrapped: + return Wrapped(prefix=[]) + + @override + def with_password(self, password: str) -> Self: + raise ElevateError( + "--ask-elevate-password requires --elevate to select an elevation method" + ) + + +@dataclass(frozen=True) +class SudoElevator(Elevator): + """Wrap with ``sudo``, optionally feeding the password on stdin. + + Extra arguments come from ``NIX_SUDOOPTS`` for backwards + compatibility with the previous implementation in ``run_wrapper``. + """ + + name: str = "sudo" + password: str | None = None + extra_opts: list[str] = field( + default_factory=lambda: shlex.split(os.getenv("NIX_SUDOOPTS", "")) + ) + + @override + def wrap_local(self) -> Wrapped: + # Local sudo can prompt on /dev/tty itself, so the password is + # only piped when one was supplied explicitly. + return self._wrap() + + @override + def wrap_remote(self) -> Wrapped: + return self._wrap() + + def _wrap(self) -> Wrapped: + if self.password is not None: + return Wrapped( + prefix=["sudo", "--prompt=", "--stdin", *self.extra_opts], + stdin=self.password + "\n", + ) + return Wrapped(prefix=["sudo", *self.extra_opts]) + + @override + def with_password(self, password: str) -> Self: + return replace(self, password=password) + + @override + def on_remote_failure(self) -> str | None: + if self.password is None: + return ( + "while running command with remote sudo, did you forget to " + "use --ask-elevate-password?" + ) + return None + + +class ElevatorKind(Enum): + """CLI-selectable elevation backends. + + The enum *value* is the :class:`Elevator` subclass to instantiate, + ``str(member)`` is what ``--elevate`` accepts on the command line. + Extended by later commits. + """ + + NONE = NoElevator + SUDO = SudoElevator + + @override + def __str__(self) -> str: + return self.name.lower() + + def make(self) -> Elevator: + """Instantiate a fresh, unparameterised elevator of this kind.""" + cls: type[Elevator] = self.value + return cls() + + @classmethod + def choices(cls) -> list[str]: + """All ``--elevate`` values, for argparse ``choices=``.""" + return [str(m) for m in cls] + + @classmethod + def from_name(cls, name: str) -> Elevator: + try: + return cls[name.upper()].make() + except KeyError: + raise ElevateError( + f"unknown elevation method {name!r}, choose from: " + + ", ".join(cls.choices()) + ) from None + + +#: Singleton used as the default ``elevate=`` argument throughout. +NO_ELEVATOR: Final[Elevator] = NoElevator() diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_elevate.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_elevate.py new file mode 100644 index 000000000000..349d36827448 --- /dev/null +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_elevate.py @@ -0,0 +1,50 @@ +import pytest +from pytest import MonkeyPatch + +import nixos_rebuild.elevate as e + + +def test_no_elevator() -> None: + n = e.NoElevator() + assert not n.elevates + assert n.wrap_local() == e.Wrapped(prefix=[]) + assert n.wrap_remote() == e.Wrapped(prefix=[]) + with pytest.raises(e.ElevateError): + n.with_password("x") + + +def test_sudo_elevator(monkeypatch: MonkeyPatch) -> None: + monkeypatch.delenv("NIX_SUDOOPTS", raising=False) + + s = e.SudoElevator() + assert s.elevates + assert s.wrap_local() == e.Wrapped(prefix=["sudo"]) + assert s.wrap_remote() == e.Wrapped(prefix=["sudo"]) + assert s.on_remote_failure() is not None + + sp = s.with_password("hunter2") + assert sp.wrap_remote() == e.Wrapped( + prefix=["sudo", "--prompt=", "--stdin"], + stdin="hunter2\n", + ) + assert sp.on_remote_failure() is None + # original unchanged + assert s.password is None + + +def test_sudo_elevator_extra_opts(monkeypatch: MonkeyPatch) -> None: + monkeypatch.setenv("NIX_SUDOOPTS", "--preserve-env=FOO -H") + s = e.SudoElevator() + assert s.wrap_local() == e.Wrapped(prefix=["sudo", "--preserve-env=FOO", "-H"]) + assert s.with_password("p").wrap_remote() == e.Wrapped( + prefix=["sudo", "--prompt=", "--stdin", "--preserve-env=FOO", "-H"], + stdin="p\n", + ) + + +def test_elevator_kind() -> None: + assert isinstance(e.ElevatorKind.from_name("sudo"), e.SudoElevator) + assert isinstance(e.ElevatorKind.from_name("none"), e.NoElevator) + with pytest.raises(e.ElevateError): + e.ElevatorKind.from_name("doas") + assert set(e.ElevatorKind.choices()) == {"none", "sudo"}