gador/nixpkgs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

nixos-rebuild-ng: thread Elevator through the call chain

Browse Source 
Mechanical replacement of `sudo: bool` with `elevate: Elevator` across
run_wrapper, nix.py and services.py, plus test updates. The temporary
shim in run_wrapper is removed.

`args.elevator` is constructed in execute() from the existing --sudo /
--ask-sudo-password flags so the CLI surface is unchanged in this
commit.
This commit is contained in:
r-vdp
2026-05-29 00:22:16 +03:00
parent b9a7b55898
commit c0941322a4
6 changed files with 81 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py
+9 -1
View File
@@ -6,6 +6,7 @@ from typing import Final, assert_never
from . import nix, services
from .constants import EXECUTABLE, WITH_SHELL_FILES
from .elevate import NO_ELEVATOR, SudoElevator
from .models import Action, BuildAttr, Flake, GroupedNixArgs, Profile
from .process import Remote
from .utils import LogFormatter
@@ -321,7 +322,7 @@ def execute(argv: list[str]) -> None:
args, grouped_nix_args = parse_args(argv)
if args.upgrade or args.upgrade_all:
nix.upgrade_channels(args.upgrade_all, args.sudo)
nix.upgrade_channels(args.upgrade_all, args.elevator)
action = Action(args.action)
# Only run shell scripts from the Nixpkgs tree if the action is
@@ -339,6 +340,13 @@ def execute(argv: list[str]) -> None:
profile = Profile.from_arg(args.profile_name)
target_host = Remote.from_arg(args.target_host, args.ask_sudo_password)
build_host = Remote.from_arg(args.build_host, False, validate_opts=False)
args.elevator = (
SudoElevator(
password=target_host.sudo_password if target_host else None,
)
if args.sudo
else NO_ELEVATOR
)
build_attr = BuildAttr.from_arg(args.attr, args.file)
flake = Flake.from_arg(args.flake, target_host)
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/nix.py
+15 -14
View File
@@ -14,6 +14,7 @@ from textwrap import dedent
from typing import Final, Literal
from . import tmpdir
from .elevate import NO_ELEVATOR, Elevator
from .models import (
Action,
BuildAttr,
@@ -453,7 +454,7 @@ def get_generations(profile: Profile) -> list[Generation]:
def get_generations_from_nix_env(
profile: Profile,
target_host: Remote | None = None,
sudo: bool = False,
elevate: Elevator = NO_ELEVATOR,
) -> list[Generation]:
"""Get all NixOS generations from profile with nix-env. Needs root.
@@ -468,7 +469,7 @@ def get_generations_from_nix_env(
["nix-env", "-p", profile.path, "--list-generations"],
stdout=PIPE,
remote=target_host,
sudo=sudo,
elevate=elevate,
)
def parse_line(line: str) -> Generation:
@@ -600,12 +601,12 @@ def repl_flake(flake: Flake, flake_flags: Args | None = None) -> None:
)
def rollback(profile: Profile, target_host: Remote | None, sudo: bool) -> Path:
def rollback(profile: Profile, target_host: Remote | None, elevate: Elevator) -> Path:
"Rollback Nix profile, like one created by `nixos-rebuild switch`."
run_wrapper(
["nix-env", "--rollback", "-p", profile.path],
remote=target_host,
sudo=sudo,
elevate=elevate,
)
# Rollback config PATH is the own profile
return profile.path
@@ -614,11 +615,11 @@ def rollback(profile: Profile, target_host: Remote | None, sudo: bool) -> Path:
def rollback_temporary_profile(
profile: Profile,
target_host: Remote | None,
sudo: bool,
elevate: Elevator,
) -> Path | None:
"Rollback a temporary Nix profile, like one created by `nixos-rebuild test`."
generations = get_generations_from_nix_env(
profile, target_host=target_host, sudo=sudo
profile, target_host=target_host, elevate=elevate
)
previous_gen_id = None
for generation in generations:
@@ -635,7 +636,7 @@ def set_profile(
profile: Profile,
path_to_config: Path,
target_host: Remote | None,
sudo: bool,
elevate: Elevator,
) -> None:
"Set a path as the current active Nix profile."
if not os.environ.get(
@@ -668,7 +669,7 @@ def set_profile(
run_wrapper(
["nix-env", "-p", profile.path, "--set", path_to_config],
remote=target_host,
sudo=sudo,
elevate=elevate,
)
@@ -676,7 +677,7 @@ def switch_to_configuration(
path_to_config: Path,
action: Literal[Action.SWITCH, Action.BOOT, Action.TEST, Action.DRY_ACTIVATE],
target_host: Remote | None,
sudo: bool,
elevate: Elevator,
install_bootloader: bool = False,
specialisation: str | None = None,
) -> None:
@@ -716,7 +717,7 @@ def switch_to_configuration(
"NIXOS_INSTALL_BOOTLOADER": "1" if install_bootloader else "0",
},
remote=target_host,
sudo=sudo,
elevate=elevate,
# switch-to-configuration is not expected to produce meaningful
# stdout, but if it (or any of its children) does, it would leak
# into our stdout and break the "only the store path on stdout"
@@ -728,7 +729,7 @@ def switch_to_configuration(
def upgrade_channels(
all_channels: bool = False,
sudo: bool = False,
elevate: Elevator = NO_ELEVATOR,
channels_dir: Path = Path("/nix/var/nix/profiles/per-user/root/channels/"),
) -> None:
"""Upgrade channels for classic Nix.
@@ -736,10 +737,10 @@ def upgrade_channels(
It will either upgrade just the `nixos` channel (including any channel
that has a `.update-on-nixos-rebuild` file) or all.
"""
if not sudo and os.geteuid() != 0:
if not elevate.elevates and os.geteuid() != 0:
raise NixOSRebuildError(
"if you pass the '--upgrade' or '--upgrade-all' flag, you must "
"also pass '--sudo' or run the command as root (e.g., with sudo)"
"also pass '--elevate' or run the command as root"
)
channel_updated = False
@@ -752,7 +753,7 @@ def upgrade_channels(
run_wrapper(
["nix-channel", "--update", channel_path.name],
check=False,
sudo=sudo,
elevate=elevate,
)
channel_updated = True
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py
+1 -7
View File
@@ -12,7 +12,7 @@ from ipaddress import AddressValueError, IPv6Address
from typing import Final, Literal, Self, TextIO, TypedDict, Unpack, override
from . import tmpdir
from .elevate import NO_ELEVATOR, Elevator, SudoElevator
from .elevate import NO_ELEVATOR, Elevator
logger: Final = logging.getLogger(__name__)
@@ -132,15 +132,9 @@ def run_wrapper(
append_local_env: Mapping[str, str] | None = None,
remote: Remote | None = None,
elevate: Elevator = NO_ELEVATOR,
sudo: bool = False,
**kwargs: Unpack[RunKwargs],
) -> subprocess.CompletedProcess[str]:
"Wrapper around `subprocess.run` that supports extra functionality."
# Back-compat shim while callers are migrated to ``elevate=``;
# removed in the next commit.
if sudo:
elevate = SudoElevator(password=remote.sudo_password if remote else None)
process_input = None
run_args: list[Arg] = list(args)
final_args: list[Arg]
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/services.py
+5 -5
View File
@@ -132,12 +132,12 @@ def _rollback_system(
) -> Path:
match action:
case Action.SWITCH | Action.BOOT:
path_to_config = nix.rollback(profile, target_host, sudo=args.sudo)
path_to_config = nix.rollback(profile, target_host, elevate=args.elevator)
case Action.TEST | Action.BUILD:
maybe_path_to_config = nix.rollback_temporary_profile(
profile,
target_host,
sudo=args.sudo,
elevate=args.elevator,
)
if maybe_path_to_config:
path_to_config = maybe_path_to_config
@@ -231,13 +231,13 @@ def _activate_system(
profile,
path_to_config,
target_host=target_host,
sudo=args.sudo,
elevate=args.elevator,
)
nix.switch_to_configuration(
path_to_config,
action,
target_host=target_host,
sudo=args.sudo,
elevate=args.elevator,
specialisation=args.specialisation,
install_bootloader=args.install_bootloader,
)
@@ -247,7 +247,7 @@ def _activate_system(
path_to_config,
action,
target_host=target_host,
sudo=args.sudo,
elevate=args.elevator,
specialisation=args.specialisation,
install_bootloader=args.install_bootloader,
)
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py
+46 -40
View File
@@ -9,12 +9,15 @@ from unittest.mock import ANY, Mock, call, patch
import pytest
from pytest import MonkeyPatch
import nixos_rebuild.elevate as e
import nixos_rebuild.models as m
import nixos_rebuild.nix as n
import nixos_rebuild.process as p
from .helpers import get_qualified_name
SUDO = e.SudoElevator()
@patch(
get_qualified_name(n.run_wrapper, n),
@@ -531,7 +534,7 @@ def test_get_generations_from_nix_env(tmp_path: Path) -> None:
["nix-env", "-p", path, "--list-generations"],
stdout=PIPE,
remote=None,
sudo=False,
elevate=e.NO_ELEVATOR,
)
remote = m.Remote("user@host", [], "password", "ssh")
@@ -539,7 +542,7 @@ def test_get_generations_from_nix_env(tmp_path: Path) -> None:
get_qualified_name(n.run_wrapper, n), autospec=True, return_value=return_value
) as mock_run:
assert n.get_generations_from_nix_env(
m.Profile("system", path), remote, True
m.Profile("system", path), remote, SUDO
) == [
m.Generation(id=2082, current=False, timestamp="2024-11-07 22:58:56"),
m.Generation(id=2083, current=False, timestamp="2024-11-07 22:59:41"),
@@ -549,7 +552,7 @@ def test_get_generations_from_nix_env(tmp_path: Path) -> None:
["nix-env", "-p", path, "--list-generations"],
stdout=PIPE,
remote=remote,
sudo=True,
elevate=SUDO,
)
@@ -640,19 +643,19 @@ def test_rollback(mock_run: Mock, tmp_path: Path) -> None:
profile = m.Profile("system", path)
assert n.rollback(profile, None, False) == profile.path
assert n.rollback(profile, None, e.NO_ELEVATOR) == profile.path
mock_run.assert_called_with(
["nix-env", "--rollback", "-p", path],
remote=None,
sudo=False,
elevate=e.NO_ELEVATOR,
)
target_host = m.Remote("user@localhost", [], None, "ssh")
assert n.rollback(profile, target_host, True) == profile.path
assert n.rollback(profile, target_host, SUDO) == profile.path
mock_run.assert_called_with(
["nix-env", "--rollback", "-p", path],
remote=target_host,
sudo=True,
elevate=SUDO,
)
@@ -672,7 +675,7 @@ def test_rollback_temporary_profile(tmp_path: Path) -> None:
"""),
)
assert (
n.rollback_temporary_profile(m.Profile("system", path), None, False)
n.rollback_temporary_profile(m.Profile("system", path), None, e.NO_ELEVATOR)
== path.parent / "system-2083-link"
)
mock_run.assert_called_with(
@@ -684,12 +687,12 @@ def test_rollback_temporary_profile(tmp_path: Path) -> None:
],
stdout=PIPE,
remote=None,
sudo=False,
elevate=e.NO_ELEVATOR,
)
target_host = m.Remote("user@localhost", [], None, "ssh")
assert (
n.rollback_temporary_profile(m.Profile("foo", path), target_host, True)
n.rollback_temporary_profile(m.Profile("foo", path), target_host, SUDO)
== path.parent / "foo-2083-link"
)
mock_run.assert_called_with(
@@ -701,12 +704,12 @@ def test_rollback_temporary_profile(tmp_path: Path) -> None:
],
stdout=PIPE,
remote=target_host,
sudo=True,
elevate=SUDO,
)
with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run:
mock_run.return_value = CompletedProcess([], 0, stdout="")
assert n.rollback_temporary_profile(profile, None, False) is None
assert n.rollback_temporary_profile(profile, None, e.NO_ELEVATOR) is None
@patch(get_qualified_name(n.run_wrapper, n), autospec=True)
@@ -719,25 +722,25 @@ def test_set_profile(mock_run: Mock) -> None:
m.Profile("system", profile_path),
config_path,
target_host=None,
sudo=False,
elevate=e.NO_ELEVATOR,
)
mock_run.assert_called_with(
["nix-env", "-p", profile_path, "--set", config_path],
remote=None,
sudo=False,
elevate=e.NO_ELEVATOR,
)
mock_run.return_value = CompletedProcess([], 1)
with pytest.raises(m.NixOSRebuildError) as e:
with pytest.raises(m.NixOSRebuildError) as exc:
n.set_profile(
m.Profile("system", profile_path),
config_path,
target_host=None,
sudo=False,
elevate=e.NO_ELEVATOR,
)
assert str(e.value).startswith(
assert str(exc.value).startswith(
"error: your NixOS configuration path seems to be missing essential files."
)
@@ -756,7 +759,7 @@ def test_switch_to_configuration_without_systemd_run(
n.switch_to_configuration(
profile_path,
m.Action.SWITCH,
sudo=False,
elevate=e.NO_ELEVATOR,
target_host=None,
specialisation=None,
install_bootloader=False,
@@ -768,21 +771,21 @@ def test_switch_to_configuration_without_systemd_run(
"NIXOS_NO_CHECK": p.PRESERVE_ENV,
"NIXOS_INSTALL_BOOTLOADER": "0",
},
sudo=False,
elevate=e.NO_ELEVATOR,
remote=None,
stdout=sys.stderr,
)
with pytest.raises(m.NixOSRebuildError) as e:
with pytest.raises(m.NixOSRebuildError) as exc:
n.switch_to_configuration(
config_path,
m.Action.BOOT,
sudo=False,
elevate=e.NO_ELEVATOR,
target_host=None,
specialisation="special",
)
assert (
str(e.value)
str(exc.value)
== "error: '--specialisation' can only be used with 'switch' and 'test'"
)
@@ -795,7 +798,7 @@ def test_switch_to_configuration_without_systemd_run(
n.switch_to_configuration(
Path("/path/to/config"),
m.Action.TEST,
sudo=True,
elevate=SUDO,
target_host=target_host,
install_bootloader=True,
specialisation="special",
@@ -810,7 +813,7 @@ def test_switch_to_configuration_without_systemd_run(
"NIXOS_NO_CHECK": p.PRESERVE_ENV,
"NIXOS_INSTALL_BOOTLOADER": "1",
},
sudo=True,
elevate=SUDO,
remote=target_host,
stdout=sys.stderr,
)
@@ -830,7 +833,7 @@ def test_switch_to_configuration_with_systemd_run(
n.switch_to_configuration(
profile_path,
m.Action.SWITCH,
sudo=False,
elevate=e.NO_ELEVATOR,
target_host=None,
specialisation=None,
install_bootloader=False,
@@ -846,7 +849,7 @@ def test_switch_to_configuration_with_systemd_run(
"NIXOS_NO_CHECK": p.PRESERVE_ENV,
"NIXOS_INSTALL_BOOTLOADER": "0",
},
sudo=False,
elevate=e.NO_ELEVATOR,
remote=None,
stdout=sys.stderr,
)
@@ -860,7 +863,7 @@ def test_switch_to_configuration_with_systemd_run(
n.switch_to_configuration(
Path("/path/to/config"),
m.Action.TEST,
sudo=True,
elevate=SUDO,
target_host=target_host,
install_bootloader=True,
specialisation="special",
@@ -876,7 +879,7 @@ def test_switch_to_configuration_with_systemd_run(
"NIXOS_NO_CHECK": p.PRESERVE_ENV,
"NIXOS_INSTALL_BOOTLOADER": "1",
},
sudo=True,
elevate=SUDO,
remote=target_host,
stdout=sys.stderr,
)
@@ -887,11 +890,13 @@ def test_switch_to_configuration_with_systemd_run(
def test_upgrade_channels(mock_run: Mock, mock_geteuid: Mock, tmpdir: Path) -> None:
tmp_path = Path(tmpdir)
with pytest.raises(m.NixOSRebuildError) as e:
n.upgrade_channels(all_channels=False, sudo=False, channels_dir=tmp_path)
assert str(e.value) == (
with pytest.raises(m.NixOSRebuildError) as exc:
n.upgrade_channels(
all_channels=False, elevate=e.NO_ELEVATOR, channels_dir=tmp_path
)
assert str(exc.value) == (
"error: if you pass the '--upgrade' or '--upgrade-all' flag, you must "
"also pass '--sudo' or run the command as root (e.g., with sudo)"
"also pass '--elevate' or run the command as root"
)
(tmp_path / "nixos").mkdir()
@@ -899,19 +904,20 @@ def test_upgrade_channels(mock_run: Mock, mock_geteuid: Mock, tmpdir: Path) -> N
(tmp_path / "nixos-hardware" / ".update-on-nixos-rebuild").touch()
(tmp_path / "home-manager").mkdir()
# should work because we are passing sudo=True even with os.geteuid == 1000
n.upgrade_channels(all_channels=False, sudo=True, channels_dir=tmp_path)
# should work because we are passing an elevator even with os.geteuid == 1000
n.upgrade_channels(all_channels=False, elevate=SUDO, channels_dir=tmp_path)
# Path.glob order is filesystem-dependent, so don't assert call order.
mock_run.assert_has_calls(
[
call(
["nix-channel", "--update", "nixos-hardware"],
check=False,
sudo=True,
elevate=SUDO,
),
call(
["nix-channel", "--update", "nixos"],
check=False,
sudo=True,
elevate=SUDO,
),
],
any_order=True,
@@ -921,23 +927,23 @@ def test_upgrade_channels(mock_run: Mock, mock_geteuid: Mock, tmpdir: Path) -> N
# root check
mock_geteuid.return_value = 0
n.upgrade_channels(all_channels=True, sudo=False, channels_dir=tmp_path)
n.upgrade_channels(all_channels=True, elevate=e.NO_ELEVATOR, channels_dir=tmp_path)
mock_run.assert_has_calls(
[
call(
["nix-channel", "--update", "home-manager"],
check=False,
sudo=False,
elevate=e.NO_ELEVATOR,
),
call(
["nix-channel", "--update", "nixos-hardware"],
check=False,
sudo=False,
elevate=e.NO_ELEVATOR,
),
call(
["nix-channel", "--update", "nixos"],
check=False,
sudo=False,
elevate=e.NO_ELEVATOR,
),
],
any_order=True,
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py
+5 -6
View File
@@ -3,6 +3,7 @@ from unittest.mock import patch
from pytest import MonkeyPatch
import nixos_rebuild.elevate as e
import nixos_rebuild.models as m
import nixos_rebuild.process as p
@@ -43,7 +44,7 @@ def test_run_wrapper(mock_run: Any) -> None:
p.run_wrapper(
["test", "--with", "flags"],
check=False,
sudo=True,
elevate=e.SudoElevator(),
env={"FOO": "bar"},
)
mock_run.assert_called_with(
@@ -67,7 +68,6 @@ def test_run_wrapper(mock_run: Any) -> None:
p.run_wrapper(
["test", "--with", "flags"],
check=False,
sudo=False,
append_local_env={"FOO": "bar"},
)
mock_run.assert_called_with(
@@ -89,7 +89,6 @@ def test_run_wrapper(mock_run: Any) -> None:
p.run_wrapper(
["test", "--with", "flags"],
check=False,
sudo=False,
env={"PATH": "/"},
append_local_env={"FOO": "bar"},
)
@@ -140,7 +139,7 @@ def test_run_wrapper(mock_run: Any) -> None:
p.run_wrapper(
["test", "--with", "flags"],
check=True,
sudo=True,
elevate=e.SudoElevator(password="password"),
env={"FOO": "bar"},
remote=m.Remote("user@localhost", ["--ssh", "opt"], "password", "ssh"),
)
@@ -262,7 +261,7 @@ def test_custom_sudo_args(mock_run: Any) -> None:
p.run_wrapper(
["test"],
check=False,
sudo=True,
elevate=e.SudoElevator(),
)
mock_run.assert_called_with(
[
@@ -287,7 +286,7 @@ def test_custom_sudo_args(mock_run: Any) -> None:
p.run_wrapper(
["test"],
check=False,
sudo=True,
elevate=e.SudoElevator(),
remote=m.Remote("user@localhost", [], None, "ssh"),
)
mock_run.assert_called_with(