nixos-rebuild-ng: validate NixOS configuration path (#418243)

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

···
 import sys
 from pathlib import Path
 from subprocess import CalledProcessError, run
+from textwrap import dedent
 from typing import Final, assert_never
 
 from . import nix, tmpdir
 from .constants import EXECUTABLE, WITH_NIX_2_18, WITH_REEXEC, WITH_SHELL_FILES
-from .models import Action, BuildAttr, Flake, ImageVariants, NRError, Profile
+from .models import Action, BuildAttr, Flake, ImageVariants, NixOSRebuildError, Profile
 from .process import Remote, cleanup_ssh
 from .utils import Args, LogFormatter, tabulate
 
···
         "--attr",
         "-A",
         help="Enable and build the NixOS system from nix file and use the "
-        + "specified attribute path from file specified by the --file option",
+        "specified attribute path from file specified by the --file option",
     )
     main_parser.add_argument(
         "--flake",
···
         "--install-bootloader",
         action="store_true",
         help="Causes the boot loader to be (re)installed on the device specified "
-        + "by the relevant configuration options",
+        "by the relevant configuration options",
     )
     main_parser.add_argument(
         "--install-grub",
···
         "--upgrade",
         action="store_true",
         help="Update the root user's channel named 'nixos' before rebuilding "
-        + "the system and channels which have a file named '.update-on-nixos-rebuild'",
+        "the system and channels which have a file named '.update-on-nixos-rebuild'",
     )
     main_parser.add_argument(
         "--upgrade-all",
···
     main_parser.add_argument(
         "--image-variant",
         help="Selects an image variant to build from the "
-        + "config.system.build.images attribute of the given configuration",
+        "config.system.build.images attribute of the given configuration",
     )
     main_parser.add_argument("action", choices=Action.values(), nargs="?")
 
···
                 # - Exec format error (e.g.: another OS/CPU arch)
                 logger.warning(
                     "could not re-exec in a newer version of nixos-rebuild, "
-                    + "using current version",
+                    "using current version",
                     exc_info=logger.isEnabledFor(logging.DEBUG),
                 )
                 # We already run clean-up, let's re-exec in the current version
···
                 os.execve(current, argv, os.environ | {"_NIXOS_REBUILD_REEXEC": "1"})
 
 
+def validate_image_variant(image_variant: str, variants: ImageVariants) -> None:
+    if image_variant not in variants:
+        raise NixOSRebuildError(
+            "please specify one of the following supported image variants via "
+            "--image-variant:\n" + "\n".join(f"- {v}" for v in variants)
+        )
+
+
+def validate_nixos_config(path_to_config: Path) -> None:
+    if not (path_to_config / "nixos-version").exists() and not os.environ.get(
+        "NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM"
+    ):
+        msg = dedent(
+            # the lowercase for the first letter below is proposital
+            f"""
+                your NixOS configuration path seems to be missing essential files.
+                To avoid corrupting your current NixOS installation, the activation will abort.
+
+                This could be caused by Nix bug: https://github.com/NixOS/nix/issues/13367.
+                This is the evaluated NixOS configuration path: {path_to_config}.
+                Change the directory to somewhere else (e.g., `cd $HOME`) before trying again.
+
+                If you think this is a mistake, you can set the environment variable
+                NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM to 1
+                and re-run the command to continue.
+                Please open an issue if this is the case.
+            """
+        ).strip()
+        raise NixOSRebuildError(msg)
+
+
 def execute(argv: list[str]) -> None:
     args, args_groups = parse_args(argv)
 
···
             no_link = action in (Action.SWITCH, Action.BOOT)
             rollback = bool(args.rollback)
 
-            def validate_image_variant(variants: ImageVariants) -> None:
-                if args.image_variant not in variants:
-                    raise NRError(
-                        "please specify one of the following "
-                        + "supported image variants via --image-variant:\n"
-                        + "\n".join(f"- {v}" for v in variants)
-                    )
-
             match action:
                 case Action.BUILD_IMAGE if flake:
                     variants = nix.get_build_image_variants_flake(
                         flake,
                         eval_flags=flake_common_flags,
                     )
-                    validate_image_variant(variants)
+                    validate_image_variant(args.image_variant, variants)
                     attr = f"config.system.build.images.{args.image_variant}"
                 case Action.BUILD_IMAGE:
                     variants = nix.get_build_image_variants(
                         build_attr,
                         instantiate_flags=common_flags,
                     )
-                    validate_image_variant(variants)
+                    validate_image_variant(args.image_variant, variants)
                     attr = f"config.system.build.images.{args.image_variant}"
                 case Action.BUILD_VM:
                     attr = "config.system.build.vm"
···
                     if maybe_path_to_config:  # kinda silly but this makes mypy happy
                         path_to_config = maybe_path_to_config
                     else:
-                        raise NRError("could not find previous generation")
+                        raise NixOSRebuildError("could not find previous generation")
                 case (_, True, _, _):
-                    raise NRError(f"--rollback is incompatible with '{action}'")
+                    raise NixOSRebuildError(
+                        f"--rollback is incompatible with '{action}'"
+                    )
                 case (_, False, Remote(_), Flake(_)):
                     path_to_config = nix.build_remote_flake(
                         attr,
···
                     copy_flags=copy_flags,
                 )
                 if action in (Action.SWITCH, Action.BOOT):
+                    validate_nixos_config(path_to_config)
                     nix.set_profile(
                         profile,
                         path_to_config,

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

···
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
-from typing import Any, Callable, ClassVar, Self, TypedDict, override
+from typing import Any, ClassVar, Self, TypedDict, override
 
 from .process import Remote, run_wrapper
 
 type ImageVariants = list[str]
 
 
-class NRError(Exception):
+class NixOSRebuildError(Exception):
     "nixos-rebuild general error."
 
     def __init__(self, message: str) -> None:
···
     return None
 
 
+def get_hostname(target_host: Remote | None) -> str | None:
+    if target_host:
+        try:
+            return run_wrapper(
+                ["uname", "-n"],
+                capture_output=True,
+                remote=target_host,
+            ).stdout.strip()
+        except (AttributeError, subprocess.CalledProcessError):
+            return None
+    else:
+        return platform.node()
+
+
 @dataclass(frozen=True)
 class Flake:
     path: Path | str
···
         return f"{self.path}#{self.attr}"
 
     @classmethod
-    def parse(
-        cls,
-        flake_str: str,
-        hostname_fn: Callable[[], str | None] = lambda: None,
-    ) -> Self:
+    def parse(cls, flake_str: str, target_host: Remote | None = None) -> Self:
         m = cls._re.match(flake_str)
         assert m is not None, f"got no matches for {flake_str}"
         attr = m.group("attr")
-        nixos_attr = f'nixosConfigurations."{attr or hostname_fn() or "default"}"'
+        nixos_attr = (
+            f'nixosConfigurations."{attr or get_hostname(target_host) or "default"}"'
+        )
         path_str = m.group("path")
         if ":" in path_str:
             return cls(path_str, nixos_attr)
···
 
     @classmethod
     def from_arg(cls, flake_arg: Any, target_host: Remote | None) -> Self | None:
-        def get_hostname() -> str | None:
-            if target_host:
-                try:
-                    return run_wrapper(
-                        ["uname", "-n"],
-                        stdout=subprocess.PIPE,
-                        remote=target_host,
-                    ).stdout.strip()
-                except (AttributeError, subprocess.CalledProcessError):
-                    return None
-            else:
-                return platform.node()
-
         match flake_arg:
             case str(s):
-                return cls.parse(s, get_hostname)
+                return cls.parse(s, target_host)
             case True:
-                return cls.parse(".", get_hostname)
+                return cls.parse(".", target_host)
             case False:
                 return None
             case _:
···
                 if default_path.exists():
                     # It can be a symlink to the actual flake.
                     default_path = default_path.resolve()
-                    return cls.parse(str(default_path.parent), get_hostname)
+                    return cls.parse(str(default_path.parent), target_host)
                 else:
                     return None
 

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

···
     Generation,
     GenerationJson,
     ImageVariants,
-    NRError,
+    NixOSRebuildError,
     Profile,
     Remote,
 )
···
         )
     else:
         if flake_flags:
-            raise NRError("'edit' does not support extra Nix flags")
+            raise NixOSRebuildError("'edit' does not support extra Nix flags")
         nixos_config = Path(
             os.getenv("NIXOS_CONFIG") or find_file("nixos-config") or "/etc/nixos"
         )
···
         if nixos_config.exists():
             run_wrapper([os.getenv("EDITOR", "nano"), nixos_config], check=False)
         else:
-            raise NRError("cannot find NixOS config file")
+            raise NixOSRebuildError("cannot find NixOS config file")
 
 
 def find_file(file: str, nix_flags: Args | None = None) -> Path | None:
···
     and if this is the current active profile or not.
     """
     if not profile.path.exists():
-        raise NRError(f"no profile '{profile.name}' found")
+        raise NixOSRebuildError(f"no profile '{profile.name}' found")
 
     def parse_path(path: Path, profile: Profile) -> Generation:
         entry_id = path.name.split("-")[1]
···
     and if this is the current active profile or not.
     """
     if not profile.path.exists():
-        raise NRError(f"no profile '{profile.name}' found")
+        raise NixOSRebuildError(f"no profile '{profile.name}' found")
 
     # Using `nix-env --list-generations` needs root to lock the profile
     r = run_wrapper(
···
     """
     if specialisation:
         if action not in (Action.SWITCH, Action.TEST):
-            raise NRError(
+            raise NixOSRebuildError(
                 "'--specialisation' can only be used with 'switch' and 'test'"
             )
         path_to_config = path_to_config / f"specialisation/{specialisation}"
 
         if not path_to_config.exists():
-            raise NRError(f"specialisation not found: {specialisation}")
+            raise NixOSRebuildError(f"specialisation not found: {specialisation}")
 
     r = run_wrapper(
         ["test", "-d", "/run/systemd/system"],
···
     if r.returncode:
         logger.debug(
             "skipping systemd-run to switch configuration since systemd is "
-            + "not working in target host"
+            "not working in target host"
         )
         cmd = []
 

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

···
             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"
+                    "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"
+                        "'--ask-sudo-password' option instead"
                     )
 
 
···
         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?"
+                "--ask-sudo-password?"
             )
         raise
 

pkgs/by-name/ni/nixos-rebuild-ng/src/pyproject.toml

···
 
 [tool.ruff.lint]
 extend-select = [
-    # Enforce type annotations
+    # enforce type annotations
     "ANN",
     # don't shadow built-in names
     "A",
-    # Better list/set/dict comprehensions
+    # better list/set/dict comprehensions
     "C4",
-    # Check for debugger statements
+    # check for debugger statements
     "T10",
     # ensure imports are sorted
     "I",
-    # Automatically upgrade syntax for newer versions
+    # automatically upgrade syntax for newer versions
     "UP",
     # detect common sources of bugs
     "B",
-    # Ruff specific rules
+    # ruff specific rules
     "RUF",
     # require `check` argument for `subprocess.run`
     "PLW1510",
     # check for needless exception names in raise statements
     "TRY201",
-    # Pythonic naming conventions
+    # pythonic naming conventions
     "N",
+    # string concatenation rules
+    "ISC001",
+    "ISC002",
+    "ISC003",
 ]
 ignore = [
     # allow Any type
    "ANN401"
 ]
-
-[tool.ruff.lint.per-file-ignores]
-"tests/" = ["FA102"]
 
 [tool.pytest.ini_options]
 pythonpath = ["."]

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

···
     )
 
 
[e-mail](os.environ, {}, clear=True)
[e-mail](
+    os.environ,
+    {"NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM": "1"},
+    clear=True,
+)
 @patch("subprocess.run", autospec=True)
 def test_execute_nix_boot(mock_run: Mock, tmp_path: Path) -> None:
     nixpkgs_path = tmp_path / "nixpkgs"
···
                     "boot",
                 ],
                 check=True,
-                **(DEFAULT_RUN_KWARGS | {"env": {"NIXOS_INSTALL_BOOTLOADER": "0"}}),
+                **(
+                    DEFAULT_RUN_KWARGS
+                    | {
+                        "env": {
+                            "NIXOS_INSTALL_BOOTLOADER": "0",
+                            "NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM": "1",
+                        }
+                    }
+                ),
             ),
         ]
     )
···
     )
 
 
[e-mail](os.environ, {}, clear=True)
[e-mail](
+    os.environ,
+    {"NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM": "1"},
+    clear=True,
+)
 @patch("subprocess.run", autospec=True)
 def test_execute_nix_switch_flake(mock_run: Mock, tmp_path: Path) -> None:
     config_path = tmp_path / "test"
···
                     "switch",
                 ],
                 check=True,
-                **(DEFAULT_RUN_KWARGS | {"env": {"NIXOS_INSTALL_BOOTLOADER": "1"}}),
+                **(
+                    DEFAULT_RUN_KWARGS
+                    | {
+                        "env": {
+                            "NIXOS_INSTALL_BOOTLOADER": "1",
+                            "NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM": "1",
+                        }
+                    }
+                ),
             ),
         ]
     )
 
 
[e-mail](os.environ, {}, clear=True)
[e-mail](
+    os.environ,
+    {"NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM": "1"},
+    clear=True,
+)
 @patch("subprocess.run", autospec=True)
 @patch("uuid.uuid4", autospec=True)
 @patch(get_qualified_name(nr.cleanup_ssh), autospec=True)
···
     )
 
 
[e-mail](os.environ, {}, clear=True)
[e-mail](
+    os.environ,
+    {"NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM": "1"},
+    clear=True,
+)
 @patch("subprocess.run", autospec=True)
 @patch(get_qualified_name(nr.cleanup_ssh), autospec=True)
 def test_execute_nix_switch_flake_target_host(
···
     )
 
 
[e-mail](os.environ, {}, clear=True)
[e-mail](
+    os.environ,
+    {"NIXOS_REBUILD_I_UNDERSTAND_THE_CONSEQUENCES_PLEASE_BREAK_MY_SYSTEM": "1"},
+    clear=True,
+)
 @patch("subprocess.run", autospec=True)
 @patch(get_qualified_name(nr.cleanup_ssh), autospec=True)
 def test_execute_nix_switch_flake_build_host(

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

···
     )
 
 
-def test_flake_parse(tmpdir: Path, monkeypatch: MonkeyPatch) -> None:
+@patch("platform.node", autospec=True, return_value="hostname")
+def test_flake_parse(mock_node: Mock, tmpdir: Path, monkeypatch: MonkeyPatch) -> None:
     assert m.Flake.parse("/path/to/flake#attr") == m.Flake(
         Path("/path/to/flake"), 'nixosConfigurations."attr"'
     )
-    assert m.Flake.parse("/path/ to /flake", lambda: "hostname") == m.Flake(
+    assert m.Flake.parse("/path/ to /flake") == m.Flake(
         Path("/path/ to /flake"), 'nixosConfigurations."hostname"'
     )
-    assert m.Flake.parse("/path/to/flake", lambda: "hostname") == m.Flake(
-        Path("/path/to/flake"), 'nixosConfigurations."hostname"'
-    )
+    with patch(
+        get_qualified_name(m.run_wrapper, m),
+        autospec=True,
+        return_value=subprocess.CompletedProcess([], 0, stdout="remote\n"),
+    ):
+        target_host = m.Remote("target@remote", [], None)
+        assert m.Flake.parse("/path/to/flake", target_host) == m.Flake(
+            Path("/path/to/flake"), 'nixosConfigurations."remote"'
+        )
     # change directory to tmpdir
     with monkeypatch.context() as patch_context:
         patch_context.chdir(tmpdir)
···
         assert m.Flake.parse("#attr") == m.Flake(
             Path("."), 'nixosConfigurations."attr"'
         )
-        assert m.Flake.parse(".") == m.Flake(Path("."), 'nixosConfigurations."default"')
+        assert m.Flake.parse(".") == m.Flake(
+            Path("."), 'nixosConfigurations."hostname"'
+        )
     assert m.Flake.parse("path:/to/flake#attr") == m.Flake(
         "path:/to/flake", 'nixosConfigurations."attr"'
     )
+
+    # from here on  we should return "default"
+    mock_node.return_value = None
+
     assert m.Flake.parse("github:user/repo/branch") == m.Flake(
         "github:user/repo/branch", 'nixosConfigurations."default"'
     )

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

···
         remote=None,
     )
 
-    with pytest.raises(m.NRError) as e:
+    with pytest.raises(m.NixOSRebuildError) as e:
         n.switch_to_configuration(
             config_path,
             m.Action.BOOT,