diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py index bff1ccab7ff6..c1ad2bde3d00 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py @@ -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) 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 index e75fd099d678..be9e35fce613 100644 --- 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 @@ -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 diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py index 121c33980459..bad599b36643 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py @@ -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. diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/services.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/services.py index 5237075a17ec..96d9d16e76f2 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/services.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/services.py @@ -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, 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 index 349d36827448..6d6231408adb 100644 --- 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 @@ -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") diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_models.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_models.py index 64ffce686976..fa9a41f93a42 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_models.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_models.py @@ -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"') diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py index ef198fdac275..3d5a5093230b 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py @@ -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") diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py index af8f510c5f49..4d6421146a8d 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py @@ -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( [