dev/nixpkgs
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

nixos-rebuild-ng: do not use TTY for --target-host

Browse Source 
Instead this commit introduces the `--ask-sudo-password` that stores the
password in memory and injects it via `stdin` if the user wants.
This commit is contained in:
Thiago Kenji Okada 2024-11-26 11:10:53 +00:00
parent 10f2b080c3
commit 2e4d755351
8 changed files with 83 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py
View File

@ -50,8 +50,9 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
parser.add_argument("--upgrade-all", action="store_true")
parser.add_argument("--json", action="store_true")
parser.add_argument("--sudo", action="store_true")
parser.add_argument("--ask-sudo-password", action="store_true")
parser.add_argument("--use-remote-sudo", action="store_true") # deprecated
parser.add_argument("--no-ssh-tty", action="store_true")
parser.add_argument("--no-ssh-tty", action="store_true") # deprecated
# parser.add_argument("--build-host") # TODO
parser.add_argument("--target-host")
parser.add_argument("--verbose", "-v", action="count", default=0)
@ -101,6 +102,9 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
if args.action == Action.DRY_RUN.value:
args.action = Action.DRY_BUILD.value
if args.ask_sudo_password:
args.sudo = True
# TODO: use deprecated=True in Python >=3.13
if args.install_grub:
info(
@ -115,6 +119,12 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
)
args.sudo = True
# TODO: use deprecated=True in Python >=3.13
if args.no_ssh_tty:
info(
f"{parser.prog}: warning: --no-ssh-tty deprecated, SSH's pseudo-TTY is never used anymore"
)
if args.action == Action.EDIT.value and (args.file or args.attr):
parser.error("--file and --attr are not supported with 'edit'")
@ -190,7 +200,7 @@ def execute(argv: list[str]) -> None:
atexit.register(cleanup_ssh, tmpdir_path)
profile = Profile.from_name(args.profile_name)
target_host = Remote.from_arg(args.target_host, not args.no_ssh_tty, tmpdir_path)
target_host = Remote.from_arg(args.target_host, args.ask_sudo_password, tmpdir_path)
flake = Flake.from_arg(args.flake, target_host)
if args.upgrade or args.upgrade_all:

1
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/models.py
View File

@ -76,7 +76,6 @@ class Flake:
["uname", "-n"],
check=True,
capture_output=True,
allow_tty=False,
remote=target_host,
).stdout.strip()
except (AttributeError, subprocess.CalledProcessError):

6
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/nix.py
View File

@ -32,7 +32,7 @@ def copy_closure(
run_wrapper(
[
"nix-copy-closure",
*(dict_to_flags(copy_flags)),
*dict_to_flags(copy_flags),
"--to",
host.host,
closure,
@ -50,7 +50,7 @@ def edit(flake: Flake | None, **flake_flags: Args) -> None:
"nix",
*FLAKE_FLAGS,
"edit",
*(dict_to_flags(flake_flags)),
*dict_to_flags(flake_flags),
"--",
str(flake),
],
@ -281,8 +281,8 @@ def nixos_build_flake(
"build",
"--print-out-paths",
f"{flake}.config.system.build.{attr}",
*dict_to_flags(flake_flags),
]
run_args += dict_to_flags(flake_flags)
r = run_wrapper(run_args, check=True, stdout=PIPE)
return Path(r.stdout.strip())

39
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
import os
import subprocess
from dataclasses import dataclass
from getpass import getpass
from pathlib import Path
from typing import Self, Sequence, TypedDict, Unpack
@ -10,9 +11,7 @@ from typing import Self, Sequence, TypedDict, Unpack
# Not exhaustive, but we can always extend it later.
class RunKwargs(TypedDict, total=False):
capture_output: bool
input: str | None
stderr: int | None
stdin: int | None
stdout: int | None
@ -20,10 +19,15 @@ class RunKwargs(TypedDict, total=False):
class Remote:
host: str
opts: list[str]
tty: bool
sudo_password: str | None
@classmethod
def from_arg(cls, host: str | None, tty: bool | None, tmp_dir: Path) -> Self | None:
def from_arg(
cls,
host: str | None,
ask_sudo_password: bool | None,
tmp_dir: Path,
) -> Self | None:
if host:
opts = os.getenv("NIX_SSHOPTS", "").split() + [
# SSH ControlMaster flags, allow for faster re-connection
@ -34,7 +38,10 @@ class Remote:
"-o",
"ControlPersist=60",
]
return cls(host, opts, bool(tty))
sudo_password = None
if ask_sudo_password:
sudo_password = getpass(f"[sudo] password for {host}: ")
return cls(host, opts, sudo_password)
else:
return None
@ -50,31 +57,24 @@ def run_wrapper(
*,
check: bool, # make it explicit so we always know if the code is handling errors
extra_env: dict[str, str] | None = None,
allow_tty: bool = True,
remote: Remote | None = None,
sudo: bool = False,
**kwargs: Unpack[RunKwargs],
) -> subprocess.CompletedProcess[str]:
"Wrapper around `subprocess.run` that supports extra functionality."
if remote and allow_tty:
# SSH's TTY will redirect all output to stdout, that may cause
# unwanted effects when used
assert not kwargs.get(
"capture_output"
), "SSH's TTY is incompatible with capture_output"
assert not kwargs.get("stderr"), "SSH's TTY is incompatible with stderr"
assert not kwargs.get("stdout"), "SSH's TTY is incompatible with stdout"
env = None
input = None
if remote:
if extra_env:
extra_env_args = [f"{env}={value}" for env, value in extra_env.items()]
args = ["env", *extra_env_args, *args]
if sudo:
args = ["sudo", *args]
if allow_tty and remote.tty:
args = ["ssh", "-t", *remote.opts, remote.host, "--", *args]
else:
args = ["ssh", *remote.opts, remote.host, "--", *args]
if remote.sudo_password:
args = ["sudo", "--prompt=", "--stdin", *args]
input = remote.sudo_password + "\n"
else:
args = ["sudo", *args]
args = ["ssh", *remote.opts, remote.host, "--", *args]
else:
if extra_env:
env = os.environ | extra_env
@ -85,6 +85,7 @@ def run_wrapper(
args,
check=check,
env=env,
input=input,
# Hope nobody is using NixOS with non-UTF8 encodings, but "surrogateescape"
# should still work in those systems.
text=True,

9
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_main.py
View File

@ -10,7 +10,12 @@ import nixos_rebuild as nr
from .helpers import get_qualified_name
DEFAULT_RUN_KWARGS = {"text": True, "errors": "surrogateescape", "env": ANY}
DEFAULT_RUN_KWARGS = {
"env": ANY,
"input": None,
"text": True,
"errors": "surrogateescape",
}
def test_parse_args() -> None:
@ -267,7 +272,6 @@ def test_execute_nix_switch_flake_remote(
call(
[
"ssh",
"-t",
"-o",
"ControlMaster=auto",
"-o",
@ -289,7 +293,6 @@ def test_execute_nix_switch_flake_remote(
call(
[
"ssh",
"-t",
"-o",
"ControlMaster=auto",
"-o",

6
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_models.py
View File

@ -94,9 +94,9 @@ def test_flake_from_arg(mock_node: Any) -> None:
return_value=subprocess.CompletedProcess([], 0, "remote-hostname\n"),
),
):
assert m.Flake.from_arg(
"/path/to", m.Remote("user@host", [], False)
) == m.Flake(Path("/path/to"), "nixosConfigurations.remote-hostname")
assert m.Flake.from_arg("/path/to", m.Remote("user@host", [], None)) == m.Flake(
Path("/path/to"), "nixosConfigurations.remote-hostname"
)
@patch(get_qualified_name(m.Path.mkdir, m), autospec=True)

28
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py
View File

@ -18,7 +18,7 @@ def test_copy_closure(mock_run: Any) -> None:
n.copy_closure(closure, None)
mock_run.assert_not_called()
target_host = m.Remote("user@host", ["--ssh", "opt"], False)
target_host = m.Remote("user@host", ["--ssh", "opt"], None)
n.copy_closure(closure, target_host)
mock_run.assert_called_with(
["nix-copy-closure", "--to", "user@host", closure],
@ -276,18 +276,12 @@ def test_rollback(mock_run: Any, tmp_path: Path) -> None:
sudo=False,
)
assert (
n.rollback(
profile,
m.Remote("user@localhost", [], False),
True,
)
== profile.path
)
target_host = m.Remote("user@localhost", [], None)
assert n.rollback(profile, target_host, True) == profile.path
mock_run.assert_called_with(
["nix-env", "--rollback", "-p", path],
check=True,
remote=m.Remote("user@localhost", [], False),
remote=target_host,
sudo=True,
)
@ -324,12 +318,9 @@ def test_rollback_temporary_profile(tmp_path: Path) -> None:
sudo=False,
)
target_host = m.Remote("user@localhost", [], None)
assert (
n.rollback_temporary_profile(
m.Profile("foo", path),
m.Remote("user@localhost", [], False),
True,
)
n.rollback_temporary_profile(m.Profile("foo", path), target_host, True)
== path.parent / "foo-2083-link"
)
mock_run.assert_called_with(
@ -341,7 +332,7 @@ def test_rollback_temporary_profile(tmp_path: Path) -> None:
],
stdout=PIPE,
check=True,
remote=m.Remote("user@localhost", [], False),
remote=target_host,
sudo=True,
)
@ -406,6 +397,7 @@ def test_switch_to_configuration(mock_run: Any, monkeypatch: Any) -> None:
== "error: '--specialisation' can only be used with 'switch' and 'test'"
)
target_host = m.Remote("user@localhost", [], None)
with monkeypatch.context() as mp:
mp.setenv("LOCALE_ARCHIVE", "/path/to/locale")
mp.setenv("PATH", "/path/to/bin")
@ -415,7 +407,7 @@ def test_switch_to_configuration(mock_run: Any, monkeypatch: Any) -> None:
Path("/path/to/config"),
m.Action.TEST,
sudo=True,
target_host=m.Remote("user@localhost", [], False),
target_host=target_host,
install_bootloader=True,
specialisation="special",
)
@ -427,7 +419,7 @@ def test_switch_to_configuration(mock_run: Any, monkeypatch: Any) -> None:
extra_env={"NIXOS_INSTALL_BOOTLOADER": "1"},
check=True,
sudo=True,
remote=m.Remote("user@localhost", [], False),
remote=target_host,
)

61
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py
View File

@ -2,8 +2,6 @@ from pathlib import Path
from typing import Any
from unittest.mock import patch
import pytest
import nixos_rebuild.models as m
import nixos_rebuild.process as p
@ -19,6 +17,7 @@ def test_run(mock_run: Any) -> None:
text=True,
errors="surrogateescape",
env=None,
input=None,
)
with patch.dict(p.os.environ, {"PATH": "/path/to/bin"}, clear=True):
@ -37,12 +36,13 @@ def test_run(mock_run: Any) -> None:
"PATH": "/path/to/bin",
"FOO": "bar",
},
input=None,
)
p.run_wrapper(
["test", "--with", "flags"],
check=True,
remote=m.Remote("user@localhost", ["--ssh", "opt"], False),
remote=m.Remote("user@localhost", ["--ssh", "opt"], "password"),
)
mock_run.assert_called_with(
["ssh", "--ssh", "opt", "user@localhost", "--", "test", "--with", "flags"],
@ -50,6 +50,7 @@ def test_run(mock_run: Any) -> None:
text=True,
errors="surrogateescape",
env=None,
input=None,
)
p.run_wrapper(
@ -57,17 +58,18 @@ def test_run(mock_run: Any) -> None:
check=True,
sudo=True,
extra_env={"FOO": "bar"},
remote=m.Remote("user@localhost", ["--ssh", "opt"], True),
remote=m.Remote("user@localhost", ["--ssh", "opt"], "password"),
)
mock_run.assert_called_with(
[
"ssh",
"-t",
"--ssh",
"opt",
"user@localhost",
"--",
"sudo",
"--prompt=",
"--stdin",
"env",
"FOO=bar",
"test",
@ -78,17 +80,9 @@ def test_run(mock_run: Any) -> None:
env=None,
text=True,
errors="surrogateescape",
input="password\n",
)
with pytest.raises(AssertionError):
p.run_wrapper(
["test", "--with", "flags"],
check=False,
allow_tty=True,
remote=m.Remote("user@localhost", [], False),
capture_output=True,
)
def test_remote_from_name(monkeypatch: Any, tmpdir: Path) -> None:
monkeypatch.setenv("NIX_SSHOPTS", "")
@ -102,23 +96,26 @@ def test_remote_from_name(monkeypatch: Any, tmpdir: Path) -> None:
"-o",
"ControlPersist=60",
],
tty=False,
sudo_password=None,
)
monkeypatch.setenv("NIX_SSHOPTS", "-f foo -b bar")
assert m.Remote.from_arg("user@localhost", True, tmpdir) == m.Remote(
"user@localhost",
opts=[
"-f",
"foo",
"-b",
"bar",
"-o",
"ControlMaster=auto",
"-o",
f"ControlPath={tmpdir / "ssh-%n"}",
"-o",
"ControlPersist=60",
],
tty=True,
)
# get_qualified_name doesn't work because getpass is aliased to another
# function
with patch(f"{p.__name__}.getpass", return_value="password"):
monkeypatch.setenv("NIX_SSHOPTS", "-f foo -b bar")
assert m.Remote.from_arg("user@localhost", True, tmpdir) == m.Remote(
"user@localhost",
opts=[
"-f",
"foo",
"-b",
"bar",
"-o",
"ControlMaster=auto",
"-o",
f"ControlPath={tmpdir / "ssh-%n"}",
"-o",
"ControlPersist=60",
],
sudo_password="password",
)