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

nixos-rebuild-ng: move password prompt out of Remote

Browse Source 
Remote no longer carries auth state, the elevator owns the password
end-to-end. The prompt itself lives in Elevator.with_prompted_password()
so the CLI entry point stays a thin orchestrator and all
elevation-related plumbing is in one module.
This commit is contained in:
r-vdp
2026-05-29 00:22:18 +03:00
parent a769269aef
commit 0273aad113
8 changed files with 76 additions and 57 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 -10
View File
@@ -6,7 +6,7 @@ from typing import Final, assert_never
from . import nix, services
from .constants import EXECUTABLE, WITH_SHELL_FILES
from .elevate import NO_ELEVATOR, ElevateError, ElevatorKind
from .elevate import NO_ELEVATOR, ElevatorKind
from .models import Action, BuildAttr, Flake, GroupedNixArgs, Profile
from .process import Remote
from .utils import LogFormatter
@@ -266,8 +266,8 @@ def parse_args(
parser_warn("--use-remote-sudo is deprecated, use --elevate=sudo instead")
# Map the elevate flags onto an Elevator. The password itself is
# attached later via Elevator.with_password() once the target host
# (used in the prompt) is known.
# attached later via Elevator.with_prompted_password() once the
# target host (used in the prompt) is known.
if args.elevate is not None:
args.elevator = ElevatorKind.from_name(args.elevate)
elif args.sudo or args.use_remote_sudo or args.ask_sudo_password:
@@ -368,13 +368,12 @@ def execute(argv: list[str]) -> None:
services.reexec(argv, args, grouped_nix_args)
profile = Profile.from_arg(args.profile_name)
target_host = Remote.from_arg(args.target_host, args.ask_elevate_password)
build_host = Remote.from_arg(args.build_host, False, validate_opts=False)
if target_host and target_host.sudo_password:
try:
args.elevator = args.elevator.with_password(target_host.sudo_password)
except ElevateError as ex:
sys.exit(f"error: {ex}")
target_host = Remote.from_arg(args.target_host)
build_host = Remote.from_arg(args.build_host, validate_opts=False)
args.elevator = args.elevator.with_prompted_password(
ask=args.ask_elevate_password,
host_label=target_host.host if target_host else "localhost",
)
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/elevate.py
+12
View File
@@ -10,6 +10,7 @@ prefix and stdin.
from __future__ import annotations
import getpass
import os
import shlex
from abc import ABC, abstractmethod
@@ -71,6 +72,17 @@ class Elevator(ABC):
(e.g. a polkit rule).
"""
def with_prompted_password(self, *, ask: bool, host_label: str) -> Self:
"""Prompt locally for a password and return a copy carrying it.
No-op when *ask* is false. May raise :class:`ElevateError` (e.g.
on :class:`NoElevator`).
"""
if not ask:
return self
password = getpass.getpass(f"[{self.name}] password for {host_label}: ")
return self.with_password(password)
def on_remote_failure(self) -> str | None:
"""Optional hint to print when a remote elevated command fails."""
return None
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py
+7 -14
View File
@@ -1,5 +1,4 @@
import atexit
import getpass
import logging
import os
import re
@@ -46,14 +45,12 @@ type EnvValue = str | Literal[_Env.PRESERVE_ENV]
class Remote:
host: str
opts: list[str]
sudo_password: str | None
store_type: str
@classmethod
def from_arg(
cls,
host: str | None,
ask_sudo_password: bool | None,
validate_opts: bool = True,
) -> Self | None:
if not host:
@@ -66,25 +63,21 @@ class Remote:
opts = shlex.split(os.getenv("NIX_SSHOPTS", ""))
if validate_opts:
cls._validate_opts(opts, ask_sudo_password)
sudo_password = None
if ask_sudo_password:
sudo_password = getpass.getpass(f"[sudo] password for {host}: ")
return cls(host, opts, sudo_password, store_type)
cls._validate_opts(opts)
return cls(host, opts, store_type)
@staticmethod
def _validate_opts(opts: list[str], ask_sudo_password: bool | None) -> None:
def _validate_opts(opts: list[str]) -> None:
for o in opts:
if o in ["-t", "-tt", "RequestTTY=yes", "RequestTTY=force"]:
logger.warning(
f"detected option '{o}' in NIX_SSHOPTS. SSH's TTY may "
"cause issues, it is recommended to remove this option"
)
if not ask_sudo_password:
logger.warning(
"if you want to prompt for sudo password use "
"'--ask-sudo-password' option instead"
)
logger.warning(
"if you want to prompt for a password for remote "
"elevation use '--ask-elevate-password' instead"
)
def ssh_host(self) -> str:
"""Fix up host string for SSH.
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/services.py
+1 -5
View File
@@ -35,11 +35,7 @@ def reexec(
return
drv = None
# Parsing the args here but ignore ask_sudo_password since it is not
# needed and we would end up asking sudo password twice
if flake := Flake.from_arg(
args.flake, Remote.from_arg(args.target_host, ask_sudo_password=None)
):
if flake := Flake.from_arg(args.flake, Remote.from_arg(args.target_host)):
drv = nix.build_flake(
NIXOS_REBUILD_ATTR,
flake,
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_elevate.py
+22
View File
@@ -48,3 +48,25 @@ def test_elevator_kind() -> None:
with pytest.raises(e.ElevateError):
e.ElevatorKind.from_name("doas")
assert set(e.ElevatorKind.choices()) == {"none", "sudo"}
def test_with_prompted_password(monkeypatch: MonkeyPatch) -> None:
prompts: list[str] = []
def fake_getpass(prompt: str) -> str:
prompts.append(prompt)
return "hunter2"
monkeypatch.setattr(e.getpass, "getpass", fake_getpass)
s = e.SudoElevator()
assert s.with_prompted_password(ask=False, host_label="x") is s
assert prompts == []
sp = s.with_prompted_password(ask=True, host_label="user@host")
assert isinstance(sp, e.SudoElevator)
assert sp.password == "hunter2"
assert prompts == ["[sudo] password for user@host: "]
with pytest.raises(e.ElevateError):
e.NoElevator().with_prompted_password(ask=True, host_label="localhost")
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_models.py
+2 -2
View File
@@ -81,7 +81,7 @@ def test_flake_parse(mock_node: Mock, tmpdir: Path, monkeypatch: MonkeyPatch) ->
autospec=True,
return_value=subprocess.CompletedProcess([], 0, stdout="remote\n"),
):
target_host = m.Remote("target@remote", [], None, "ssh")
target_host = m.Remote("target@remote", [], "ssh")
assert m.Flake.parse("/path/to/flake", target_host) == m.Flake(
"/path/to/flake", 'nixosConfigurations."remote"'
)
@@ -201,7 +201,7 @@ def test_flake_from_arg(
),
):
assert m.Flake.from_arg(
"/path/to", m.Remote("user@host", [], None, "ssh")
"/path/to", m.Remote("user@host", [], "ssh")
) == m.Flake("/path/to", 'nixosConfigurations."remote-hostname"')
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py
+10 -10
View File
@@ -86,7 +86,7 @@ def test_build_flake(mock_run: Mock, monkeypatch: MonkeyPatch, tmpdir: Path) ->
def test_build_remote(
mock_uuid4: Mock, mock_run: Mock, monkeypatch: MonkeyPatch
) -> None:
build_host = m.Remote("user@host", [], None, "ssh")
build_host = m.Remote("user@host", [], "ssh")
monkeypatch.setenv("NIX_SSHOPTS", "--ssh opts")
def run_wrapper_side_effect(
@@ -180,7 +180,7 @@ def test_build_remote_flake(
) -> None:
monkeypatch.chdir(tmpdir)
flake = m.Flake.parse("/flake.nix#hostname")
build_host = m.Remote("user@host", [], None, "ssh")
build_host = m.Remote("user@host", [], "ssh")
monkeypatch.setenv("NIX_SSHOPTS", "--ssh opts")
assert n.build_remote_flake(
@@ -240,9 +240,9 @@ def test_copy_closure(monkeypatch: MonkeyPatch) -> None:
n.copy_closure(closure, None)
mock_run.assert_not_called()
target_host = m.Remote("user@target.host", [], None, "ssh")
build_host = m.Remote("user@build.host", [], None, "ssh")
target_host_ng = m.Remote("user@target.host", [], None, "ssh-ng")
target_host = m.Remote("user@target.host", [], "ssh")
build_host = m.Remote("user@build.host", [], "ssh")
target_host_ng = m.Remote("user@target.host", [], "ssh-ng")
with patch(get_qualified_name(n.run_wrapper, n), autospec=True) as mock_run:
n.copy_closure(closure, target_host)
mock_run.assert_called_with(
@@ -537,7 +537,7 @@ def test_get_generations_from_nix_env(tmp_path: Path) -> None:
elevate=e.NO_ELEVATOR,
)
remote = m.Remote("user@host", [], "password", "ssh")
remote = m.Remote("user@host", [], "ssh")
with patch(
get_qualified_name(n.run_wrapper, n), autospec=True, return_value=return_value
) as mock_run:
@@ -650,7 +650,7 @@ def test_rollback(mock_run: Mock, tmp_path: Path) -> None:
elevate=e.NO_ELEVATOR,
)
target_host = m.Remote("user@localhost", [], None, "ssh")
target_host = m.Remote("user@localhost", [], "ssh")
assert n.rollback(profile, target_host, SUDO) == profile.path
mock_run.assert_called_with(
["nix-env", "--rollback", "-p", path],
@@ -690,7 +690,7 @@ def test_rollback_temporary_profile(tmp_path: Path) -> None:
elevate=e.NO_ELEVATOR,
)
target_host = m.Remote("user@localhost", [], None, "ssh")
target_host = m.Remote("user@localhost", [], "ssh")
assert (
n.rollback_temporary_profile(m.Profile("foo", path), target_host, SUDO)
== path.parent / "foo-2083-link"
@@ -789,7 +789,7 @@ def test_switch_to_configuration_without_systemd_run(
== "error: '--specialisation' can only be used with 'switch' and 'test'"
)
target_host = m.Remote("user@localhost", [], None, "ssh")
target_host = m.Remote("user@localhost", [], "ssh")
with monkeypatch.context() as mp:
mp.setenv("LOCALE_ARCHIVE", "/path/to/locale")
mp.setenv("PATH", "/path/to/bin")
@@ -854,7 +854,7 @@ def test_switch_to_configuration_with_systemd_run(
stdout=sys.stderr,
)
target_host = m.Remote("user@localhost", [], None, "ssh")
target_host = m.Remote("user@localhost", [], "ssh")
with monkeypatch.context() as mp:
mp.setenv("LOCALE_ARCHIVE", "/path/to/locale")
mp.setenv("PATH", "/path/to/bin")
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py
+13 -16
View File
@@ -111,7 +111,7 @@ def test_run_wrapper(mock_run: Any) -> None:
p.run_wrapper(
["test", "--with", "some flags"],
check=True,
remote=m.Remote("user@localhost", ["--ssh", "opt"], "password", "ssh"),
remote=m.Remote("user@localhost", ["--ssh", "opt"], "ssh"),
)
mock_run.assert_called_with(
[
@@ -141,7 +141,7 @@ def test_run_wrapper(mock_run: Any) -> None:
check=True,
elevate=e.SudoElevator(password="password"),
env={"FOO": "bar"},
remote=m.Remote("user@localhost", ["--ssh", "opt"], "password", "ssh"),
remote=m.Remote("user@localhost", ["--ssh", "opt"], "ssh"),
)
mock_run.assert_called_with(
[
@@ -180,7 +180,7 @@ def test__kill_long_running_ssh_process(mock_run: Any) -> None:
"build",
"/nix/store/la0c8nmpr9xfclla0n4f3qq9iwgdrq4g-nixos-system-sankyuu-nixos-25.05.20250424.f771eb4.drv^*",
],
m.Remote("user@localhost", opts=[], sudo_password=None, store_type="ssh"),
m.Remote("user@localhost", opts=[], store_type="ssh"),
)
mock_run.assert_called_with(
[
@@ -203,21 +203,18 @@ def test__kill_long_running_ssh_process(mock_run: Any) -> None:
def test_remote_from_name(monkeypatch: MonkeyPatch) -> None:
monkeypatch.setenv("NIX_SSHOPTS", "")
assert m.Remote.from_arg("user@localhost", None, False) == m.Remote(
assert m.Remote.from_arg("user@localhost", validate_opts=False) == m.Remote(
"user@localhost",
opts=[],
sudo_password=None,
store_type="ssh",
)
with patch("getpass.getpass", autospec=True, return_value="password"):
monkeypatch.setenv("NIX_SSHOPTS", "-f foo -b bar -t")
assert m.Remote.from_arg("user@localhost", True, True) == m.Remote(
"user@localhost",
opts=["-f", "foo", "-b", "bar", "-t"],
sudo_password="password",
store_type="ssh",
)
monkeypatch.setenv("NIX_SSHOPTS", "-f foo -b bar -t")
assert m.Remote.from_arg("user@localhost") == m.Remote(
"user@localhost",
opts=["-f", "foo", "-b", "bar", "-t"],
store_type="ssh",
)
def test_ssh_host() -> None:
@@ -239,13 +236,13 @@ def test_ssh_host() -> None:
}
for host_input, expected in ssh_remotes.items():
remote = m.Remote.from_arg(host_input, None, False)
remote = m.Remote.from_arg(host_input, validate_opts=False)
assert remote is not None
assert remote.ssh_host() == expected
assert remote.store_type == "ssh"
for host_input, expected in ssh_ng_remotes.items():
remote = m.Remote.from_arg(host_input, None, False)
remote = m.Remote.from_arg(host_input, validate_opts=False)
assert remote is not None
assert remote.ssh_host() == expected
assert remote.store_type == "ssh-ng"
@@ -287,7 +284,7 @@ def test_custom_sudo_args(mock_run: Any) -> None:
["test"],
check=False,
elevate=e.SudoElevator(),
remote=m.Remote("user@localhost", [], None, "ssh"),
remote=m.Remote("user@localhost", [], "ssh"),
)
mock_run.assert_called_with(
[