nixos-rebuild-ng: validate NixOS configuration path

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

@@ -5,11 +5,12 @@
 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
 
@@ -99,7 +100,7 @@ def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentPa
         "--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",
@@ -117,7 +118,7 @@ def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentPa
         "--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",
@@ -142,7 +143,7 @@ def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentPa
         "--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",
@@ -186,7 +187,7 @@ def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentPa
     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="?")
 
@@ -321,14 +322,45 @@ def reexec(
                 # - 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
                 # to avoid issues
                 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)
 
@@ -393,28 +425,20 @@ def execute(argv: list[str]) -> None:
             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"
@@ -435,9 +459,11 @@ def validate_image_variant(variants: ImageVariants) -> None:
                     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,
@@ -488,6 +514,7 @@ def validate_image_variant(variants: ImageVariants) -> None:
                     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

@@ -4,14 +4,14 @@
 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:
@@ -100,6 +100,20 @@ def discover_closest_flake(location: Path) -> Path | 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
@@ -114,15 +128,13 @@ def __str__(self) -> 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)
@@ -143,24 +155,11 @@ def parse(
 
     @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 _:
@@ -169,7 +168,7 @@ def get_hostname() -> str | None:
                 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

@@ -20,7 +20,7 @@
     Generation,
     GenerationJson,
     ImageVariants,
-    NRError,
+    NixOSRebuildError,
     Profile,
     Remote,
 )
@@ -256,7 +256,7 @@ def edit(flake: Flake | None, flake_flags: Args | None = None) -> None:
         )
     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"
         )
@@ -266,7 +266,7 @@ def edit(flake: Flake | None, flake_flags: Args | None = None) -> None:
         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:
@@ -424,7 +424,7 @@ def get_generations(profile: Profile) -> list[Generation]:
     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]
@@ -456,7 +456,7 @@ def get_generations_from_nix_env(
     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(
@@ -635,13 +635,13 @@ def switch_to_configuration(
     """
     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"],
@@ -652,7 +652,7 @@ def switch_to_configuration(
     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

@@ -55,12 +55,12 @@ def _validate_opts(opts: list[str], ask_sudo_password: bool | None) -> None:
             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"
                     )
 
 
@@ -161,7 +161,7 @@ def run_wrapper(
         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

@@ -39,37 +39,38 @@ ignore_missing_imports = true
 
 [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 = ["."]
 addopts = "--import-mode=importlib"