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

nixos-rebuild-ng: add --elevate={sudo,run0} and polkit-stdin-agent (#512018)

Browse Source
This commit is contained in:
Ramses
2026-05-30 20:01:40 +00:00
committed by GitHub
parent 6bc2984826 dbbee0dd3a
commit 82141a1cd5
15 changed files with 1057 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nixos/modules/installer/tools/tools.nix
+21
View File
@@ -314,6 +314,27 @@ in
name = "nixos-rebuild";
package = config.system.build.nixos-rebuild;
})
(
{ config, ... }:
{
options.system.tools.nixos-rebuild.enableRun0Elevation = lib.mkEnableOption ''
support for being targeted by `nixos-rebuild --elevate=run0
--ask-elevate-password`.
This enables polkit and adds {command}`polkit-stdin-agent` to
{option}`environment.systemPackages` so that a deploying host
can find a target-architecture agent at
{file}`<toplevel>/sw/bin/polkit-stdin-agent` after copying the
closure (which is required for cross-architecture deploys and
mismatched nixpkgs revisions to work).
'';
config = lib.mkIf config.system.tools.nixos-rebuild.enableRun0Elevation {
security.polkit.enable = lib.mkDefault true;
environment.systemPackages = [ pkgs.polkit-stdin-agent ];
};
}
)
(mkToolModule {
name = "nixos-version";
package = nixos-version;
nixos/tests/nixos-rebuild-target-host.nix
+22
View File
@@ -22,6 +22,8 @@
};
system.includeBuildDependencies = true;
# Needed so the offline build of the target config succeeds.
system.extraDependencies = [ pkgs.polkit-stdin-agent ];
virtualisation = {
cores = 2;
@@ -49,6 +51,11 @@
users.users.alice.extraGroups = [ "wheel" ];
users.users.bob.extraGroups = [ "wheel" ];
# Needed for --elevate=run0. NixOS's default polkit admin rule is
# `unix-group:wheel`, so bob (in wheel) can authenticate with his
# own password via polkit-stdin-agent.
system.tools.nixos-rebuild.enableRun0Elevation = true;
# Disable sudo for root to ensure sudo isn't called without `--sudo`
security.sudo.extraRules = lib.mkForce [
{
@@ -142,6 +149,7 @@
deployer.copy_from_host("${configFile "config-1-deployed"}", "/root/configuration-1.nix")
deployer.copy_from_host("${configFile "config-2-deployed"}", "/root/configuration-2.nix")
deployer.copy_from_host("${configFile "config-3-deployed"}", "/root/configuration-3.nix")
deployer.copy_from_host("${configFile "config-4-deployed"}", "/root/configuration-4.nix")
deployer.copy_from_host("${targetNetworkJSON}", "/root/target-network.json")
deployer.copy_from_host("${targetConfigJSON}", "/root/target-configuration.json")
@@ -168,6 +176,20 @@
target_hostname = deployer.succeed("ssh alice@target cat /etc/hostname").rstrip()
assert target_hostname == "config-3-deployed", f"{target_hostname=}"
with subtest("Deploy to bob@target with run0 and password"):
# polkit-stdin-agent registers an agent for systemd-run on the
# target and answers the PAM conversation with the password we
# supply locally. The agent is resolved on the target from
# <toplevel>/sw/bin (see Run0Elevator._remote_agent_argv).
deployer.send_chars("nixos-rebuild switch -I nixos-config=/root/configuration-4.nix --target-host bob@target --elevate=run0 --ask-elevate-password\n")
deployer.wait_until_tty_matches("1", "\\[run0\\] password for bob@target")
deployer.send_chars("${nodes.target.users.users.bob.password}\n")
deployer.wait_until_tty_matches("1", "Done. The new configuration is /nix/store/.*config-4-deployed")
target_hostname = deployer.succeed("ssh alice@target cat /etc/hostname").rstrip()
assert target_hostname == "config-4-deployed", f"{target_hostname=}"
# The target-arch agent is reachable at the stable sw/bin path.
target.succeed("test -x /run/current-system/sw/bin/polkit-stdin-agent")
with subtest("Deploy works with very long TMPDIR"):
tmp_dir = "/var/folder/veryveryveryveryverylongpathnamethatdoesnotworkwithcontrolpath"
deployer.succeed(f"mkdir -p {tmp_dir}")
pkgs/by-name/ni/nixos-rebuild-ng/nixos-rebuild.8.scd
+33 -9
View File
@@ -21,7 +21,7 @@ nixos-rebuild - reconfigure a NixOS machine
_nixos-rebuild_ \[--verbose] [--quiet] [--max-jobs MAX_JOBS] [--cores CORES] [--log-format LOG_FORMAT] [--keep-going] [--keep-failed] [--fallback] [--repair] [--option OPTION OPTION] [--builders BUILDERS] [--include INCLUDE]++
\[--print-build-logs] [--show-trace] [--accept-flake-config] [--refresh] [--impure] [--offline] [--no-net] [--recreate-lock-file] [--no-update-lock-file] [--no-write-lock-file] [--no-registries] [--commit-lock-file]++
\[--update-input UPDATE_INPUT] [--override-input OVERRIDE_INPUT OVERRIDE_INPUT] [--no-build-output] [--use-substitutes] [--help] [--debug] [--file FILE] [--attr ATTR] [--flake [FLAKE]] [--no-flake] [--install-bootloader]++
\[--profile-name PROFILE_NAME] [--specialisation SPECIALISATION] [--rollback] [--store-path STORE_PATH] [--upgrade] [--upgrade-all] [--json] [--ask-sudo-password] [--sudo] [--no-reexec]++
\[--profile-name PROFILE_NAME] [--specialisation SPECIALISATION] [--rollback] [--store-path STORE_PATH] [--upgrade] [--upgrade-all] [--json] [--elevate {none,sudo,run0}] [--ask-elevate-password] [--no-reexec]++
\[--build-host BUILD_HOST] [--target-host TARGET_HOST] [--no-build-nix] [--image-variant IMAGE_VARIANT]++
\[{switch,boot,test,build,edit,repl,dry-build,dry-run,dry-activate,build-image,build-vm,build-vm-with-bootloader,list-generations}]
@@ -269,17 +269,41 @@ It must be one of the following:
target-host connection to cache.nixos.org is faster than the connection
between hosts.
*--elevate* {none,sudo,run0}
Privilege-elevation method used for activation commands. Setting this
option allows deploying as a non-root user.
_sudo_ prefixes commands with *sudo*. Additional sudo options can be
passed via the NIX_SUDOOPTS environment variable.
_run0_ uses systemd's polkit-based elevation. Locally this runs *run0*
directly so the user's normal polkit agent handles any prompts. With
*--target-host* the equivalent _systemd-run --uid=0 --pipe_ form is
used (no remote TTY is allocated). Unless *--ask-elevate-password* is
also passed, the deploying user must be granted the polkit action
_org.freedesktop.systemd1.manage-units_ on the target host without
authentication, e.g. via _security.polkit.extraConfig_.
*--ask-elevate-password*, *-S*
Prompt locally for a password and feed it to the elevation method.
Implies *--elevate=sudo* if *--elevate* is not given.
For _sudo_ this uses *sudo --stdin*. For _run0_ the command is wrapped
in *polkit-stdin-agent*, which registers a per-process polkit agent
and answers the PAM conversation from the supplied password. The
machine performing the elevation (the local host, or the target host
with *--target-host*) must set
_system.tools.nixos-rebuild.enableRun0Elevation = true_.
*--elevate=run0 --ask-elevate-password* is not usable otherwise.
*--sudo*
When set, *nixos-rebuild* prefixes activation commands with sudo.
Setting this option allows deploying as a non-root user.
Alias for *--elevate=sudo*.
You can set sudo options by defining the NIX_SUDOOPTS environment
variable.
*--ask-sudo-password*
Alias for *--elevate=sudo --ask-elevate-password*.
*--ask-sudo-password*, *-S*
When set, *nixos-rebuild* will ask for sudo password for remote
activation (i.e.: on *--target-host*) at the start of the build process.
Implies *--sudo*.
*--use-remote-sudo*
Deprecated, use *--elevate=sudo* instead.
*--file* _path_, *-f* _path_
Build the NixOS system from the specified file. The file must
pkgs/by-name/ni/nixos-rebuild-ng/package.nix
-1
View File
@@ -1,6 +1,5 @@
{
lib,
stdenv,
callPackage,
installShellFiles,
mkShell,
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/__init__.py
+49 -14
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, ElevatorKind
from .models import Action, BuildAttr, Flake, GroupedNixArgs, Profile
from .process import Remote
from .utils import LogFormatter
@@ -163,18 +164,32 @@ def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentPa
help="JSON output, only implemented for 'list-generations' right now",
)
main_parser.add_argument(
"--ask-sudo-password",
"-S",
action="store_true",
help="Asks for sudo password for remote activation, implies --sudo",
"--elevate",
choices=ElevatorKind.choices(),
default=None,
help="Privilege-elevation method for activation commands",
)
main_parser.add_argument(
"--sudo", action="store_true", help="Prefixes activation commands with sudo"
"--ask-elevate-password",
"-S",
action="store_true",
help="Prompt locally for the password to feed to the elevation "
"method, implies --elevate=sudo if --elevate is not given",
)
main_parser.add_argument(
"--sudo",
action="store_true",
help="Alias for '--elevate=sudo'",
)
main_parser.add_argument(
"--ask-sudo-password",
action="store_true",
help="Alias for '--elevate=sudo --ask-elevate-password'",
)
main_parser.add_argument(
"--use-remote-sudo",
action="store_true",
help="Deprecated, use '--sudo' instead",
help="Deprecated, use '--elevate=sudo' instead",
)
main_parser.add_argument("--no-ssh-tty", action="store_true", help="Deprecated")
main_parser.add_argument(
@@ -245,16 +260,32 @@ def parse_args(
args.action = Action.DRY_BUILD.value
if args.ask_sudo_password:
args.sudo = True
args.ask_elevate_password = True
if args.use_remote_sudo:
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_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:
args.elevator = ElevatorKind.SUDO.make()
elif args.ask_elevate_password:
# -S historically implied --sudo. Keep that for muscle memory
# but be explicit now that there is more than one backend.
parser_warn(
"--ask-elevate-password without --elevate, falling back to --elevate=sudo"
)
args.elevator = ElevatorKind.SUDO.make()
else:
args.elevator = NO_ELEVATOR
if args.install_grub:
parser_warn("--install-grub is deprecated, use --install-bootloader instead")
args.install_bootloader = True
if args.use_remote_sudo:
parser_warn("--use-remote-sudo is deprecated, use --sudo instead")
args.sudo = True
if args.fast:
parser_warn("--fast is deprecated, use --no-reexec instead")
args.no_reexec = True
@@ -321,7 +352,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
@@ -337,8 +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_sudo_password)
build_host = Remote.from_arg(args.build_host, False, validate_opts=False)
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
+418
View File
@@ -0,0 +1,418 @@
"""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.
The remote case has no controlling terminal and the elevated command's
environment depends on the backend (``sudo`` inherits the SSH login env,
while the run0 backend starts a transient unit with only systemd's
default ``PATH``), so each backend builds the full remote argv itself
via :meth:`Elevator.wrap_remote`.
"""
from __future__ import annotations
import getpass
import os
import shlex
from abc import ABC, abstractmethod
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field, replace
from enum import Enum
from pathlib import Path, PurePosixPath
from typing import ClassVar, Final, Literal, Self, override
# Kept here (rather than in process.py) so that elevators can build remote
# argvs without a circular import.
type Arg = str | bytes | os.PathLike[str] | os.PathLike[bytes]
type Args = Sequence[Arg]
class _Env(Enum):
PRESERVE_ENV = "PRESERVE"
@override
def __repr__(self) -> str:
return self.value
#: Sentinel meaning "copy this variable from the environment the wrapped
#: command would naturally see" (``os.environ`` locally, the SSH login
#: shell's environment remotely).
PRESERVE_ENV: Final = _Env.PRESERVE_ENV
type EnvValue = str | Literal[_Env.PRESERVE_ENV]
def _remote_env_shell_argv(
prefix: Sequence[str],
env: Mapping[str, EnvValue],
args: Args,
) -> list[Arg]:
"""Build ``<prefix> /bin/sh -c 'exec /usr/bin/env -i K=V… "$@"' sh <args>``.
The wrapper runs in the SSH login session, resolves ``PRESERVE_ENV``
variables against that session's environment, and re-execs the command
with exactly that set. ``/usr/bin/env`` is referenced by absolute path so
the wrapper does not depend on ``PATH`` itself (provided on NixOS via
``environment.usrbinenv``).
"""
assigns: list[str] = []
for k, v in env.items():
if v is PRESERVE_ENV:
assigns.append(f'{k}="${{{k}-}}"')
else:
assigns.append(f"{k}={shlex.quote(v)}")
script = f'exec /usr/bin/env -i {" ".join(assigns)} "$@"'
return [*prefix, "/bin/sh", "-c", script, "sh", *args]
@dataclass(frozen=True)
class Wrapped:
"""Result of wrapping a command for local elevation."""
#: Arguments to prepend to the command.
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
@dataclass(frozen=True)
class RemoteWrapped:
"""Result of wrapping a command for remote elevation over SSH.
Unlike :class:`Wrapped` this carries the *full* remote argv: backends
differ in where the env-resolution shell wrapper must sit relative to
the elevator (inside ``sudo``, but *around* ``systemd-run``), so a
plain prefix is not expressive enough.
"""
argv: list[Arg]
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, env: Mapping[str, EnvValue], args: Args) -> RemoteWrapped:
"""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.
*env* is the environment to establish for the elevated command.
:data:`PRESERVE_ENV` values are resolved against the SSH login
shell's environment, and the backend must do so before any step
that replaces it with a service-style one.
"""
@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 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 for_target_config(self, toplevel: PurePosixPath | Path) -> Self:
"""Return a copy bound to the toplevel being activated on the target.
Backends that need a helper binary on the remote
(:class:`Run0Elevator`'s ``polkit-stdin-agent``) use this to find
a target-architecture copy inside the just-copied closure. No-op
by default.
"""
del toplevel # unused in the base implementation
return self
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, env: Mapping[str, EnvValue], args: Args) -> RemoteWrapped:
return RemoteWrapped(argv=_remote_env_shell_argv([], env, args))
@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.
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 wrap_remote(self, env: Mapping[str, EnvValue], args: Args) -> RemoteWrapped:
# sudo runs inside the SSH login session, so the env wrapper can
# sit *inside* it and ${VAR-} resolves against the login env.
if self.password is not None:
prefix = ["sudo", "--prompt=", "--stdin", *self.extra_opts]
stdin = self.password + "\n"
else:
prefix = ["sudo", *self.extra_opts]
stdin = None
return RemoteWrapped(
argv=_remote_env_shell_argv(prefix, env, args),
stdin=stdin,
)
@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
@dataclass(frozen=True)
class Run0Elevator(Elevator):
"""Wrap with systemd's polkit-based ``run0``.
Locally, ``run0`` is used directly and the user's polkit agent
(graphical or ``pkttyagent``) handles any prompts.
Remotely we spell out the explicit ``systemd-run --uid=0 --pipe``
form instead. ``run0`` would internally do the same thing when stdio
is not a TTY (see systemd ``src/run/run.c``), but going through
``systemd-run`` directly gives us ``--setenv K=V``, which we need to
forward the SSH login environment into the transient unit (whose own
``PATH`` on NixOS is just the systemd store path), and keeps the
argv independent of whether the SSH session happens to have a TTY.
Authorisation comes from either a polkit rule on the target granting
``org.freedesktop.systemd1.manage-units`` to the deploying user, or
from ``polkit-stdin-agent`` for ``--ask-elevate-password``.
"""
name: str = "run0"
password: str | None = None
#: ``${toplevel}/sw/bin/polkit-stdin-agent`` on the target, set via
#: :meth:`for_target_config`. ``None`` falls back to the target's ``PATH``.
remote_agent: str | None = None
#: Non-interactive equivalent of ``run0`` (see the class docstring).
REMOTE_BASE: ClassVar[tuple[str, ...]] = (
"systemd-run",
"--uid=0",
"--pipe",
"--quiet",
"--wait",
"--collect",
"--service-type=exec",
"--send-sighup",
)
@override
def wrap_local(self) -> Wrapped:
if self.password is not None:
# Resolved from PATH, same requirement as the remote case: the
# machine doing the elevation needs
# system.tools.nixos-rebuild.enableRun0Elevation.
return Wrapped(
prefix=["polkit-stdin-agent", "--password-fd=0", "--", "run0", "--"],
stdin=self.password + "\n",
)
return Wrapped(prefix=["run0", "--"])
@override
def wrap_remote(self, env: Mapping[str, EnvValue], args: Args) -> RemoteWrapped:
# /bin/sh wrapper resolves PRESERVE_ENV in the SSH login session
# and forwards the result into the unit via --setenv.
setenvs: list[str] = []
for k, v in env.items():
if v is PRESERVE_ENV:
setenvs.append(f'--setenv={k}="${{{k}-}}"')
else:
setenvs.append(f"--setenv={shlex.quote(f'{k}={v}')}")
script = f'exec {shlex.join(self.REMOTE_BASE)} {" ".join(setenvs)} -- "$@"'
argv: list[Arg] = ["/bin/sh", "-c", script, "sh", *args]
if self.password is not None:
# polkit has no `sudo --stdin` equivalent. polkit-stdin-agent
# registers a per-process agent for the wrapped command and
# answers the PAM conversation from its stdin.
argv = self._remote_agent_argv(argv)
return RemoteWrapped(argv=argv, stdin=self.password + "\n")
return RemoteWrapped(argv=argv)
#: POSIX sh fragment that picks the first runnable agent from the
#: positional parameters up to ``--`` and execs it with the remainder.
#: ``command -v`` covers both absolute paths and ``PATH`` lookups.
_AGENT_PICKER: ClassVar[str] = (
"agent=; "
"for a; do "
"shift; "
'[ "$a" = -- ] && break; '
'[ -z "$agent" ] && command -v "$a" >/dev/null 2>&1 && agent="$a"; '
"done; "
'[ -n "$agent" ] && exec "$agent" --password-fd=0 -- "$@"; '
'echo "nixos-rebuild: polkit-stdin-agent not found on target host '
'(set system.tools.nixos-rebuild.enableRun0Elevation = true)" >&2; '
"exit 127"
)
def _remote_agent_argv(self, inner: list[Arg]) -> list[Arg]:
"""Wrap *inner* in a target-side agent lookup.
The deployer's own agent may be the wrong arch/nixpkgs (cross-arch
deploys, Darwin deployers, ``--no-reexec``), so resolve on the
target instead: first ``${toplevel}/sw/bin/polkit-stdin-agent``
(present when ``system.tools.nixos-rebuild.enableRun0Elevation``
is set), then bare ``polkit-stdin-agent`` on the SSH login PATH.
:data:`_AGENT_PICKER` exits 127 with a hint if neither is found.
"""
candidates: list[str] = []
if self.remote_agent is not None:
candidates.append(self.remote_agent)
candidates.append("polkit-stdin-agent")
return ["/bin/sh", "-c", self._AGENT_PICKER, "sh", *candidates, "--", *inner]
@override
def with_password(self, password: str) -> Self:
return replace(self, password=password)
@override
def for_target_config(self, toplevel: PurePosixPath | Path) -> Self:
return replace(
self, remote_agent=str(toplevel / "sw" / "bin" / "polkit-stdin-agent")
)
@override
def on_remote_failure(self) -> str | None:
if self.password is None:
return (
"while running command with remote run0. Either pass "
"--ask-elevate-password, or grant the deploying user the "
"polkit action 'org.freedesktop.systemd1.manage-units' on "
"the target host (security.polkit.extraConfig)."
)
return (
"while running command with remote run0. If the error above "
"mentions polkit-stdin-agent or PolicyKit1, the target host "
"needs system.tools.nixos-rebuild.enableRun0Elevation = true."
)
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.
"""
NONE = NoElevator
SUDO = SudoElevator
RUN0 = Run0Elevator
@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()
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/nix.py
+16 -15
View File
@@ -14,6 +14,7 @@ from textwrap import dedent
from typing import Final, Literal
from . import tmpdir
from .elevate import NO_ELEVATOR, PRESERVE_ENV, Elevator
from .models import (
Action,
BuildAttr,
@@ -25,7 +26,7 @@ from .models import (
Profile,
Remote,
)
from .process import PRESERVE_ENV, SSH_DEFAULT_OPTS, run_wrapper
from .process import SSH_DEFAULT_OPTS, run_wrapper
from .utils import Args, dict_to_flags
FLAKE_FLAGS: Final = ["--extra-experimental-features", "nix-command flakes"]
@@ -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
+36 -93
View File
@@ -1,17 +1,23 @@
import atexit
import getpass
import logging
import os
import re
import shlex
import subprocess
from collections.abc import Mapping, Sequence
from collections.abc import Mapping
from dataclasses import dataclass
from enum import Enum
from ipaddress import AddressValueError, IPv6Address
from typing import Final, Literal, Self, TextIO, TypedDict, Unpack, override
from typing import Final, Self, TextIO, TypedDict, Unpack
from . import tmpdir
from .elevate import (
NO_ELEVATOR,
PRESERVE_ENV,
Arg,
Args,
Elevator,
EnvValue,
)
logger: Final = logging.getLogger(__name__)
@@ -25,34 +31,16 @@ SSH_DEFAULT_OPTS: Final = [
]
class _Env(Enum):
PRESERVE_ENV = "PRESERVE"
@override
def __repr__(self) -> str:
return self.value
PRESERVE_ENV: Final = _Env.PRESERVE_ENV
type Arg = str | bytes | os.PathLike[str] | os.PathLike[bytes]
type Args = Sequence[Arg]
type EnvValue = str | Literal[_Env.PRESERVE_ENV]
@dataclass(frozen=True)
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:
@@ -65,25 +53,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.
@@ -130,7 +114,7 @@ def run_wrapper(
env: Mapping[str, EnvValue] | None = None,
append_local_env: Mapping[str, str] | None = None,
remote: Remote | None = None,
sudo: bool = False,
elevate: Elevator = NO_ELEVATOR,
**kwargs: Unpack[RunKwargs],
) -> subprocess.CompletedProcess[str]:
"Wrapper around `subprocess.run` that supports extra functionality."
@@ -142,27 +126,9 @@ def run_wrapper(
resolved_env = _resolve_env_local(normalized_env)
if remote:
remote_run_args: list[Arg] = [
"/bin/sh",
"-c",
_remote_shell_script(normalized_env),
"sh",
*run_args,
]
if sudo:
sudo_args = shlex.split(os.getenv("NIX_SUDOOPTS", ""))
if remote.sudo_password:
remote_run_args = [
"sudo",
"--prompt=",
"--stdin",
*sudo_args,
*remote_run_args,
]
process_input = remote.sudo_password + "\n"
else:
remote_run_args = ["sudo", *sudo_args, *remote_run_args]
rwrapped = elevate.wrap_remote(normalized_env, run_args)
process_input = rwrapped.stdin
remote_run_args: list[Arg] = rwrapped.argv
ssh_args: list[Arg] = [
"ssh",
@@ -176,22 +142,19 @@ def run_wrapper(
popen_env = None # keep ssh's environment normal
else:
if sudo:
# subprocess.run(env=...) would affect sudo, but sudo may drop env
# for the target command.
# So we inject env via `sudo env ... cmd`.
wrapped = elevate.wrap_local()
process_input = wrapped.stdin
if elevate.elevates:
# subprocess.run(env=...) would affect the elevator process,
# which may then drop env for the target command. Inject env
# via `env -i ... cmd` instead so it survives.
if env is not None and resolved_env:
run_args = _prefix_env_cmd(run_args, resolved_env)
sudo_args = shlex.split(os.getenv("NIX_SUDOOPTS", ""))
final_args = ["sudo", *sudo_args, *run_args]
# No need to pass env to subprocess.run; keep sudo's own env
# default.
final_args = [*wrapped.prefix, *run_args]
popen_env = None
else:
# Non-sudo local: we can fully control the environment with
# subprocess.run(env=...)
# Unprivileged local: we can fully control the environment
# with subprocess.run(env=...)
final_args = run_args
popen_env = None if env is None else resolved_env
@@ -225,16 +188,14 @@ def run_wrapper(
return r
except KeyboardInterrupt:
# sudo commands are activation only and unlikely to be long running
if remote and not sudo:
# elevated commands are activation only and unlikely to be long
# running
if remote and not elevate.elevates:
_kill_long_running_ssh_process(args, remote)
raise
except subprocess.CalledProcessError:
if sudo and remote and remote.sudo_password is None:
logger.error(
"while running command with remote sudo, did you forget to use "
"--ask-sudo-password?"
)
if remote and (hint := elevate.on_remote_failure()):
logger.error(hint)
raise
@@ -268,7 +229,7 @@ def _resolve_env_local(env: dict[str, EnvValue]) -> dict[str, str]:
return result
def _prefix_env_cmd(cmd: Sequence[Arg], resolved_env: dict[str, str]) -> list[Arg]:
def _prefix_env_cmd(cmd: Args, resolved_env: dict[str, str]) -> list[Arg]:
"""
Prefix a command with `env -i K=V ... -- <cmd...>` to set vars for the
command.
@@ -280,24 +241,6 @@ def _prefix_env_cmd(cmd: Sequence[Arg], resolved_env: dict[str, str]) -> list[Ar
return ["env", "-i", *assigns, *cmd]
def _remote_shell_script(env: Mapping[str, EnvValue]) -> str:
"""
Build the POSIX shell wrapper used for remote execution over SSH.
SSH sends the remote command as a shell-interpreted command line, so we
need a wrapper to establish a clean environment before `exec`-ing the real
command. This wrapper is always run under `/bin/sh -c` so preserved
variables like `${PATH-}` do not depend on the remote user's login shell.
"""
shell_assigns: list[str] = []
for k, v in env.items():
if v is PRESERVE_ENV:
shell_assigns.append(f'{k}="${{{k}-}}"')
else:
shell_assigns.append(f"{k}={shlex.quote(v)}")
return f'exec env -i {" ".join(shell_assigns)} "$@"'
def _quote_remote_arg(arg: Arg) -> str:
return shlex.quote(str(arg))
pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/services.py
+16 -10
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,
@@ -132,12 +128,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 +227,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 +243,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,
)
@@ -302,6 +298,11 @@ def build_and_activate_system(
copy_flags=grouped_nix_args.copy_flags,
)
elif args.rollback:
if target_host is not None:
# The elevated `nix-env --rollback` runs before path_to_config
# is known, so point the elevator at the profile to find a
# target-arch helper in the *current* generation's sw/bin.
args.elevator = args.elevator.for_target_config(profile.path)
path_to_config = _rollback_system(
action=action,
args=args,
@@ -319,6 +320,11 @@ def build_and_activate_system(
grouped_nix_args=grouped_nix_args,
)
if target_host is not None and not args.rollback:
# Prefer the helper from the toplevel we just copied to the
# target (correct arch, independent of re-exec / nixpkgs pin).
args.elevator = args.elevator.for_target_config(path_to_config)
current_config = Path("/run/current-system")
if args.diff:
if current_config.exists():
pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_elevate.py
+155
View File
@@ -0,0 +1,155 @@
from pathlib import PurePosixPath
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=[])
rw = n.wrap_remote({"PATH": e.PRESERVE_ENV}, ["cmd"])
assert rw.stdin is None
assert rw.argv[:2] == ["/bin/sh", "-c"]
assert rw.argv[-1] == "cmd"
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"])
rw = s.wrap_remote({"PATH": e.PRESERVE_ENV}, ["cmd"])
assert rw.argv[0] == "sudo"
assert rw.argv[1:3] == ["/bin/sh", "-c"]
assert rw.stdin is None
assert s.on_remote_failure() is not None
sp = s.with_password("hunter2")
rw = sp.wrap_remote({"PATH": e.PRESERVE_ENV}, ["cmd"])
assert rw.argv[:3] == ["sudo", "--prompt=", "--stdin"]
assert rw.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"])
rw = s.with_password("p").wrap_remote({"PATH": e.PRESERVE_ENV}, ["cmd"])
assert rw.argv[:5] == ["sudo", "--prompt=", "--stdin", "--preserve-env=FOO", "-H"]
assert rw.stdin == "p\n"
def test_run0_elevator() -> None:
r = e.Run0Elevator()
assert r.elevates
assert r.wrap_local() == e.Wrapped(prefix=["run0", "--"])
assert r.on_remote_failure() is not None
rp = r.with_password("hunter2")
w = rp.wrap_local()
assert w.prefix[0] == "polkit-stdin-agent"
assert "run0" in w.prefix
assert w.stdin == "hunter2\n"
# With a password the failure hint points at the agent, not at -S.
hint = rp.on_remote_failure()
assert hint is not None and "polkit-stdin-agent" in hint
def test_run0_elevator_remote() -> None:
r = e.Run0Elevator()
# No password: /bin/sh wrapper around systemd-run, env passed via
# --setenv so it is resolved in the SSH login shell rather than
# inside the transient unit (which has a useless PATH on NixOS).
rw = r.wrap_remote(
{"PATH": e.PRESERVE_ENV, "NIXOS_INSTALL_BOOTLOADER": "1"},
["nix-env", "-p", "/profile"],
)
assert rw.stdin is None
assert rw.argv[:2] == ["/bin/sh", "-c"]
script = rw.argv[2]
assert isinstance(script, str)
assert script.startswith("exec systemd-run --uid=0 --pipe ")
assert '--setenv=PATH="${PATH-}"' in script
assert "--setenv=NIXOS_INSTALL_BOOTLOADER=1" in script
assert script.endswith(' -- "$@"')
assert rw.argv[3:] == ["sh", "nix-env", "-p", "/profile"]
# With password: an agent-picker /bin/sh wraps the inner /bin/sh, so
# the agent is registered for the inner shell (and the systemd-run it
# execs into). With no toplevel bound the only candidate is bare-name
# PATH lookup.
rw = r.with_password("pw").wrap_remote({"PATH": e.PRESERVE_ENV}, ["cmd"])
assert rw.stdin == "pw\n"
assert rw.argv[:3] == ["/bin/sh", "-c", e.Run0Elevator._AGENT_PICKER]
sep = rw.argv.index("--")
assert rw.argv[3:sep] == ["sh", "polkit-stdin-agent"]
assert rw.argv[sep + 1 : sep + 3] == ["/bin/sh", "-c"]
# Explicit values containing spaces are shell-quoted inside the
# script (the whole thing is later shlex.quoted again for SSH).
rw = r.wrap_remote({"FOO": "a b"}, ["cmd"])
script = rw.argv[2]
assert isinstance(script, str)
assert "--setenv='FOO=a b'" in script
def test_run0_for_target_config() -> None:
toplevel = PurePosixPath("/nix/store/aaaa-nixos-system")
r = e.Run0Elevator().with_password("pw").for_target_config(toplevel)
assert r.remote_agent == f"{toplevel}/sw/bin/polkit-stdin-agent"
rw = r.wrap_remote({"PATH": e.PRESERVE_ENV}, ["cmd"])
sep = rw.argv.index("--")
# Order matters: target-arch toplevel first, then PATH.
assert rw.argv[4:sep] == [
f"{toplevel}/sw/bin/polkit-stdin-agent",
"polkit-stdin-agent",
]
# Inner argv is preserved verbatim after the separator.
assert rw.argv[sep + 1 : sep + 3] == ["/bin/sh", "-c"]
assert rw.argv[-1] == "cmd"
# Non-run0 elevators ignore the toplevel.
s = e.SudoElevator()
assert s.for_target_config(toplevel) is s
def test_elevator_kind() -> None:
assert isinstance(e.ElevatorKind.from_name("sudo"), e.SudoElevator)
assert isinstance(e.ElevatorKind.from_name("run0"), e.Run0Elevator)
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", "run0"}
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_main.py
+47 -16
View File
@@ -21,6 +21,37 @@ DEFAULT_RUN_KWARGS = {
}
def test_parse_args_elevate() -> None:
r, _ = nr.parse_args(["nixos-rebuild", "switch"])
assert r.elevator is nr.elevate.NO_ELEVATOR
r, _ = nr.parse_args(["nixos-rebuild", "switch", "--elevate=sudo"])
assert isinstance(r.elevator, nr.elevate.SudoElevator)
r, _ = nr.parse_args(["nixos-rebuild", "switch", "--elevate=run0"])
assert isinstance(r.elevator, nr.elevate.Run0Elevator)
# back-compat aliases
for flag in ("--sudo", "--use-remote-sudo"):
r, _ = nr.parse_args(["nixos-rebuild", "switch", flag])
assert isinstance(r.elevator, nr.elevate.SudoElevator)
r, _ = nr.parse_args(["nixos-rebuild", "switch", "--ask-sudo-password"])
assert isinstance(r.elevator, nr.elevate.SudoElevator)
assert r.ask_elevate_password
# -S without --elevate implies sudo
r, _ = nr.parse_args(["nixos-rebuild", "switch", "-S"])
assert isinstance(r.elevator, nr.elevate.SudoElevator)
# explicit --elevate wins over the --sudo alias
r, _ = nr.parse_args(["nixos-rebuild", "switch", "--elevate=none", "--sudo"])
assert isinstance(r.elevator, nr.elevate.NoElevator)
with pytest.raises(SystemExit):
nr.parse_args(["nixos-rebuild", "switch", "--elevate=doas"])
def test_parse_args() -> None:
with pytest.raises(SystemExit) as e:
nr.parse_args(["nixos-rebuild", "unknown-action"])
@@ -663,7 +694,7 @@ def test_execute_nix_switch_build_target_host(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"mktemp",
"-d",
@@ -682,7 +713,7 @@ def test_execute_nix_switch_build_target_host(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"nix-store",
"--realise",
@@ -702,7 +733,7 @@ def test_execute_nix_switch_build_target_host(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"readlink",
"-f",
@@ -720,7 +751,7 @@ def test_execute_nix_switch_build_target_host(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"rm",
"-rf",
@@ -753,7 +784,7 @@ def test_execute_nix_switch_build_target_host(
"sudo",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"nix-env",
"-p",
@@ -772,7 +803,7 @@ def test_execute_nix_switch_build_target_host(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"test",
"-d",
@@ -790,7 +821,7 @@ def test_execute_nix_switch_build_target_host(
"sudo",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" LOCALE_ARCHIVE="${LOCALE_ARCHIVE-}" NIXOS_NO_CHECK="${NIXOS_NO_CHECK-}" NIXOS_INSTALL_BOOTLOADER=0 "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" LOCALE_ARCHIVE="${LOCALE_ARCHIVE-}" NIXOS_NO_CHECK="${NIXOS_NO_CHECK-}" NIXOS_INSTALL_BOOTLOADER=0 "$@"'""",
"sh",
*nr.nix.SWITCH_TO_CONFIGURATION_CMD_PREFIX,
str(config_path / "bin/switch-to-configuration"),
@@ -879,7 +910,7 @@ def test_execute_nix_switch_flake_target_host(
"sudo",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"nix-env",
"-p",
@@ -898,7 +929,7 @@ def test_execute_nix_switch_flake_target_host(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"test",
"-d",
@@ -916,7 +947,7 @@ def test_execute_nix_switch_flake_target_host(
"sudo",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" LOCALE_ARCHIVE="${LOCALE_ARCHIVE-}" NIXOS_NO_CHECK="${NIXOS_NO_CHECK-}" NIXOS_INSTALL_BOOTLOADER=0 "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" LOCALE_ARCHIVE="${LOCALE_ARCHIVE-}" NIXOS_NO_CHECK="${NIXOS_NO_CHECK-}" NIXOS_INSTALL_BOOTLOADER=0 "$@"'""",
"sh",
*nr.nix.SWITCH_TO_CONFIGURATION_CMD_PREFIX,
str(config_path / "bin/switch-to-configuration"),
@@ -1004,7 +1035,7 @@ def test_execute_nix_switch_flake_build_host(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"nix",
"--extra-experimental-features",
@@ -1243,7 +1274,7 @@ def test_execute_build_dry_run_build_and_target_remote(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"nix",
"--extra-experimental-features",
@@ -1498,7 +1529,7 @@ def test_execute_switch_store_path_target_host(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"test",
"-f",
@@ -1516,7 +1547,7 @@ def test_execute_switch_store_path_target_host(
"sudo",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"nix-env",
"-p",
@@ -1535,7 +1566,7 @@ def test_execute_switch_store_path_target_host(
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"test",
"-d",
@@ -1553,7 +1584,7 @@ def test_execute_switch_store_path_target_host(
"sudo",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" LOCALE_ARCHIVE="${LOCALE_ARCHIVE-}" NIXOS_NO_CHECK="${NIXOS_NO_CHECK-}" NIXOS_INSTALL_BOOTLOADER=0 "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" LOCALE_ARCHIVE="${LOCALE_ARCHIVE-}" NIXOS_NO_CHECK="${NIXOS_NO_CHECK-}" NIXOS_INSTALL_BOOTLOADER=0 "$@"'""",
"sh",
*nr.nix.SWITCH_TO_CONFIGURATION_CMD_PREFIX,
str(config_path / "bin/switch-to-configuration"),
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
+64 -58
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),
@@ -83,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(
@@ -177,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(
@@ -237,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(
@@ -531,15 +534,15 @@ 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")
remote = m.Remote("user@host", [], "ssh")
with patch(
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
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],
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")
target_host = m.Remote("user@localhost", [], "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,
@@ -764,29 +767,29 @@ def test_switch_to_configuration_without_systemd_run(
mock_run.assert_called_with(
[profile_path / "bin/switch-to-configuration", "switch"],
env={
"LOCALE_ARCHIVE": p.PRESERVE_ENV,
"NIXOS_NO_CHECK": p.PRESERVE_ENV,
"LOCALE_ARCHIVE": e.PRESERVE_ENV,
"NIXOS_NO_CHECK": e.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'"
)
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")
@@ -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",
@@ -806,11 +809,11 @@ def test_switch_to_configuration_without_systemd_run(
"test",
],
env={
"LOCALE_ARCHIVE": p.PRESERVE_ENV,
"NIXOS_NO_CHECK": p.PRESERVE_ENV,
"LOCALE_ARCHIVE": e.PRESERVE_ENV,
"NIXOS_NO_CHECK": e.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,
@@ -842,16 +845,16 @@ def test_switch_to_configuration_with_systemd_run(
"switch",
],
env={
"LOCALE_ARCHIVE": p.PRESERVE_ENV,
"NIXOS_NO_CHECK": p.PRESERVE_ENV,
"LOCALE_ARCHIVE": e.PRESERVE_ENV,
"NIXOS_NO_CHECK": e.PRESERVE_ENV,
"NIXOS_INSTALL_BOOTLOADER": "0",
},
sudo=False,
elevate=e.NO_ELEVATOR,
remote=None,
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")
@@ -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",
@@ -872,11 +875,11 @@ def test_switch_to_configuration_with_systemd_run(
"test",
],
env={
"LOCALE_ARCHIVE": p.PRESERVE_ENV,
"NIXOS_NO_CHECK": p.PRESERVE_ENV,
"LOCALE_ARCHIVE": e.PRESERVE_ENV,
"NIXOS_NO_CHECK": e.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
+130 -41
View File
@@ -3,28 +3,46 @@ 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
def test_remote_shell_script() -> None:
assert p._remote_shell_script({"PATH": p.PRESERVE_ENV}) == (
'''exec env -i PATH="${PATH-}" "$@"'''
)
assert p._remote_shell_script(
def test_remote_env_shell_argv() -> None:
assert e._remote_env_shell_argv([], {"PATH": e.PRESERVE_ENV}, ["cmd"]) == [
"/bin/sh",
"-c",
'''exec /usr/bin/env -i PATH="${PATH-}" "$@"''',
"sh",
"cmd",
]
assert e._remote_env_shell_argv(
["sudo"],
{
"PATH": p.PRESERVE_ENV,
"LOCALE_ARCHIVE": p.PRESERVE_ENV,
"NIXOS_NO_CHECK": p.PRESERVE_ENV,
"PATH": e.PRESERVE_ENV,
"LOCALE_ARCHIVE": e.PRESERVE_ENV,
"NIXOS_NO_CHECK": e.PRESERVE_ENV,
"NIXOS_INSTALL_BOOTLOADER": "0",
}
) == (
"""exec env -i PATH="${PATH-}" LOCALE_ARCHIVE="${LOCALE_ARCHIVE-}" """
'''NIXOS_NO_CHECK="${NIXOS_NO_CHECK-}" NIXOS_INSTALL_BOOTLOADER=0 "$@"'''
)
assert p._remote_shell_script({"PATH": p.PRESERVE_ENV, "FOO": "some value"}) == (
'''exec env -i PATH="${PATH-}" FOO='some value' "$@"'''
)
},
["cmd", "arg"],
) == [
"sudo",
"/bin/sh",
"-c",
"""exec /usr/bin/env -i PATH="${PATH-}" LOCALE_ARCHIVE="${LOCALE_ARCHIVE-}" """
'''NIXOS_NO_CHECK="${NIXOS_NO_CHECK-}" NIXOS_INSTALL_BOOTLOADER=0 "$@"''',
"sh",
"cmd",
"arg",
]
assert e._remote_env_shell_argv(
[], {"PATH": e.PRESERVE_ENV, "FOO": "some value"}, []
) == [
"/bin/sh",
"-c",
'''exec /usr/bin/env -i PATH="${PATH-}" FOO='some value' "$@"''',
"sh",
]
@patch.dict(p.os.environ, {"PATH": "/path/to/bin"}, clear=True)
@@ -43,7 +61,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 +85,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 +106,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"},
)
@@ -112,7 +128,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(
[
@@ -124,7 +140,7 @@ def test_run_wrapper(mock_run: Any) -> None:
"--",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"test",
"--with",
@@ -140,9 +156,9 @@ 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"),
remote=m.Remote("user@localhost", ["--ssh", "opt"], "ssh"),
)
mock_run.assert_called_with(
[
@@ -157,7 +173,7 @@ def test_run_wrapper(mock_run: Any) -> None:
"--stdin",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" FOO=bar "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" FOO=bar "$@"'""",
"sh",
"test",
"--with",
@@ -181,7 +197,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(
[
@@ -204,21 +220,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:
@@ -240,18 +253,94 @@ 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"
@patch.dict(p.os.environ, {"PATH": "/path/to/bin"}, clear=True)
@patch("subprocess.run", autospec=True)
def test_run_wrapper_run0(mock_run: Any) -> None:
p.run_wrapper(["cmd", "arg"], elevate=e.Run0Elevator())
mock_run.assert_called_with(
["run0", "--", "cmd", "arg"],
check=True,
text=True,
errors="surrogateescape",
env=None,
input=None,
)
run0_script = (
"exec systemd-run --uid=0 --pipe --quiet --wait --collect "
"--service-type=exec --send-sighup "
'--setenv=PATH="${PATH-}" -- "$@"'
)
p.run_wrapper(
["cmd", "arg"],
elevate=e.Run0Elevator(),
remote=m.Remote("user@host", [], "ssh"),
)
mock_run.assert_called_with(
[
"ssh",
*p.SSH_DEFAULT_OPTS,
"user@host",
"--",
"/bin/sh",
"-c",
p._quote_remote_arg(run0_script),
"sh",
"cmd",
"arg",
],
check=True,
text=True,
errors="surrogateescape",
env=None,
input=None,
)
p.run_wrapper(
["cmd"],
elevate=e.Run0Elevator().with_password("pw"),
remote=m.Remote("user@host", [], "ssh"),
)
mock_run.assert_called_with(
[
"ssh",
*p.SSH_DEFAULT_OPTS,
"user@host",
"--",
"/bin/sh",
"-c",
p._quote_remote_arg(e.Run0Elevator._AGENT_PICKER),
"sh",
# No toplevel bound, so the only candidate is PATH lookup.
"polkit-stdin-agent",
"--",
"/bin/sh",
"-c",
p._quote_remote_arg(run0_script),
"sh",
"cmd",
],
check=True,
text=True,
errors="surrogateescape",
env=None,
input="pw\n",
)
@patch("subprocess.run", autospec=True)
def test_custom_sudo_args(mock_run: Any) -> None:
with patch.dict(
@@ -262,7 +351,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,8 +376,8 @@ def test_custom_sudo_args(mock_run: Any) -> None:
p.run_wrapper(
["test"],
check=False,
sudo=True,
remote=m.Remote("user@localhost", [], None, "ssh"),
elevate=e.SudoElevator(),
remote=m.Remote("user@localhost", [], "ssh"),
)
mock_run.assert_called_with(
[
@@ -302,7 +391,7 @@ def test_custom_sudo_args(mock_run: Any) -> None:
"--args",
"/bin/sh",
"-c",
"""'exec env -i PATH="${PATH-}" "$@"'""",
"""'exec /usr/bin/env -i PATH="${PATH-}" "$@"'""",
"sh",
"test",
],
pkgs/by-name/po/polkit-stdin-agent/package.nix
+48
View File
@@ -0,0 +1,48 @@
{
lib,
rustPlatform,
fetchFromGitea,
nix-update-script,
nixosTests,
}:
rustPlatform.buildRustPackage (finalAttrs: {
pname = "polkit-stdin-agent";
version = "0.3.0";
src = fetchFromGitea {
domain = "codeberg.org";
owner = "r-vdp";
repo = "polkit-stdin-agent";
tag = "v${finalAttrs.version}";
hash = "sha256-Nl/+IBbUEsxSKSWLXwUB3mV4iAG0z9mv+Bl6CSeFzR4=";
};
cargoHash = "sha256-Eb/7ejVmtG5FNSh66gZO3337KCPNi+xtYVC5qyFKJzg=";
strictDeps = true;
__structuredAttrs = true;
passthru = {
updateScript = nix-update-script { };
tests = { inherit (nixosTests) nixos-rebuild-target-host; };
};
meta = {
description = "Non-interactive polkit authentication agent that answers PAM prompts from a file descriptor";
longDescription = ''
Registers a per-process polkit authentication agent for a wrapped
command and answers the PAM conversation from a file descriptor
instead of /dev/tty, giving run0 / systemd-run the same
"password on stdin" ergonomics as `sudo --stdin`.
Used by `nixos-rebuild --elevate=run0 --ask-elevate-password` to
authenticate on a target host over SSH without allocating a TTY.
'';
homepage = "https://codeberg.org/r-vdp/polkit-stdin-agent";
license = lib.licenses.mit;
maintainers = with lib.maintainers; [ rvdp ];
platforms = lib.platforms.linux;
mainProgram = "polkit-stdin-agent";
};
})