add flash command

checks/flake-module.nix

@@ -3,6 +3,7 @@
     ./impure/flake-module.nix
     ./backups/flake-module.nix
     ./installation/flake-module.nix
+    ./flash/flake-module.nix
   ];
   perSystem = { pkgs, lib, self', ... }: {
     checks =

checks/flash/flake-module.nix

@@ -0,0 +1,46 @@
+{ self, ... }:
+{
+  perSystem = { nodes, pkgs, lib, ... }:
+    let
+      dependencies = [
+        self
+        pkgs.stdenv.drvPath
+        self.clanInternals.machines.${pkgs.hostPlatform.system}.test_install_machine.config.system.build.toplevel
+        self.clanInternals.machines.${pkgs.hostPlatform.system}.test_install_machine.config.system.build.diskoScript
+        self.clanInternals.machines.${pkgs.hostPlatform.system}.test_install_machine.config.system.clan.deployment.file
+        self.inputs.nixpkgs.legacyPackages.${pkgs.hostPlatform.system}.disko
+      ] ++ builtins.map (i: i.outPath) (builtins.attrValues self.inputs);
+      closureInfo = pkgs.closureInfo { rootPaths = dependencies; };
+    in
+    {
+      checks = pkgs.lib.mkIf (pkgs.stdenv.isLinux) {
+        flash =
+          (import ../lib/test-base.nix)
+            {
+              name = "flash";
+              nodes.target = {
+                virtualisation.emptyDiskImages = [ 4096 ];
+                virtualisation.memorySize = 3000;
+                environment.systemPackages = [ self.packages.${pkgs.system}.clan-cli ];
+                environment.etc."install-closure".source = "${closureInfo}/store-paths";
+
+                nix.settings = {
+                  substituters = lib.mkForce [ ];
+                  hashed-mirrors = null;
+                  connect-timeout = lib.mkForce 3;
+                  flake-registry = pkgs.writeText "flake-registry" ''{"flakes":[],"version":2}'';
+                  experimental-features = [
+                    "nix-command"
+                    "flakes"
+                  ];
+                };
+              };
+              testScript = ''
+                start_all()
+                machine.succeed("clan --flake ${../..} flash --debug --yes --disk main /dev/vdb test_install_machine")
+              '';
+            }
+            { inherit pkgs self; };
+      };
+    };
+}

flake.lock

@@ -78,11 +78,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1708847675,
+        "lastModified": 1709764733,
-        "narHash": "sha256-RUZ7KEs/a4EzRELYDGnRB6i7M1Izii3JD/LyzH0c6Tg=",
+        "narHash": "sha256-GptBnEUy8IcRrnd8X5WBJPDXG7M4bjj8OG4Wjg8dCDs=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "2a34566b67bef34c551f204063faeecc444ae9da",
+        "rev": "edf9f14255a7ac20f8da7b70609e980a964fca7a",
         "type": "github"
       },
       "original": {

pkgs/clan-cli/clan_cli/flash.py

@@ -1,51 +1,110 @@
 import argparse
 import importlib
 import logging
+import os
+import shlex
+import shutil
+from collections.abc import Sequence
 from dataclasses import dataclass
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from typing import Any
 
+from .cmd import Log, run
+from .errors import ClanError
 from .machines.machines import Machine
-from .secrets.generate import generate_secrets
+from .nix import nix_shell
+from .secrets.modules import SecretStoreBase
 
 log = logging.getLogger(__name__)
 
 
-def flash_machine(machine: Machine, device: str | None = None) -> None:
+def flash_machine(
+    machine: Machine, disks: dict[str, str], dry_run: bool, debug: bool
+) -> None:
     secrets_module = importlib.import_module(machine.secrets_module)
-    secret_store = secrets_module.SecretStore(machine=machine)
+    secret_store: SecretStoreBase = secrets_module.SecretStore(machine=machine)
-
-    generate_secrets(machine)
-
     with TemporaryDirectory() as tmpdir_:
         tmpdir = Path(tmpdir_)
-        upload_dir_ = machine.secrets_upload_directory
+        upload_dir = machine.secrets_upload_directory
 
-        if upload_dir_.startswith("/"):
+        if upload_dir.startswith("/"):
-            upload_dir_ = upload_dir_[1:]
+            local_dir = tmpdir / upload_dir[1:]
-        upload_dir = tmpdir / upload_dir_
+        else:
-        upload_dir.mkdir(parents=True)
+            local_dir = tmpdir / upload_dir
-        secret_store.upload(upload_dir)
 
-        fs_image = machine.build_nix("config.system.clan.iso")
+        local_dir.mkdir(parents=True)
-        print(fs_image)
+        secret_store.upload(local_dir)
+        disko_install = []
+
+        if os.geteuid() != 0:
+            if shutil.which("sudo") is None:
+                raise ClanError(
+                    "sudo is required to run disko-install as a non-root user"
+                )
+            disko_install.append("sudo")
+
+        disko_install.append("disko-install")
+        if dry_run:
+            disko_install.append("--dry-run")
+        if debug:
+            disko_install.append("--debug")
+        for name, device in disks.items():
+            disko_install.extend(["--disk", name, device])
+
+        disko_install.extend(["--extra-files", str(local_dir), upload_dir])
+        disko_install.extend(["--flake", str(machine.flake) + "#" + machine.name])
+
+        cmd = nix_shell(
+            ["nixpkgs#disko"],
+            disko_install,
+        )
+        print("$", " ".join(map(shlex.quote, cmd)))
+        run(cmd, log=Log.BOTH, error_msg=f"Failed to flash {machine}")
 
 
 @dataclass
 class FlashOptions:
     flake: Path
     machine: str
-    device: str | None
+    disks: dict[str, str]
+    dry_run: bool
+    confirm: bool
+    debug: bool
+
+
+class AppendDiskAction(argparse.Action):
+    def __init__(self, option_strings: str, dest: str, **kwargs: Any) -> None:
+        super().__init__(option_strings, dest, **kwargs)
+
+    def __call__(
+        self,
+        parser: argparse.ArgumentParser,
+        namespace: argparse.Namespace,
+        values: str | Sequence[str] | None,
+        option_string: str | None = None,
+    ) -> None:
+        disks = getattr(namespace, self.dest)
+        assert isinstance(values, list), "values must be a list"
+        disks[values[0]] = values[1]
 
 
 def flash_command(args: argparse.Namespace) -> None:
     opts = FlashOptions(
         flake=args.flake,
         machine=args.machine,
-        device=args.device,
+        disks=args.disk,
+        dry_run=args.dry_run,
+        confirm=not args.yes,
+        debug=args.debug,
     )
     machine = Machine(opts.machine, flake=opts.flake)
-    flash_machine(machine, device=opts.device)
+    if opts.confirm and not opts.dry_run:
+        disk_str = ", ".join(f"{name}={device}" for name, device in opts.disks.items())
+        ask = input(f"Install {machine.name} to {disk_str}? [y/N] ")
+        if ask != "y":
+            return
+    flash_machine(machine, disks=opts.disks, dry_run=opts.dry_run, debug=opts.debug)
 
 
 def register_parser(parser: argparse.ArgumentParser) -> None:
@@ -55,8 +114,30 @@ def register_parser(parser: argparse.ArgumentParser) -> None:
         help="machine to install",
     )
     parser.add_argument(
-        "--device",
+        "--disk",
         type=str,
-        help="device to flash the system to",
+        nargs=2,
+        metavar=("name", "value"),
+        action=AppendDiskAction,
+        help="device to flash to",
+        default={},
+    )
+    parser.add_argument(
+        "--yes",
+        action="store_true",
+        help="do not ask for confirmation",
+        default=False,
+    )
+    parser.add_argument(
+        "--dry-run",
+        help="Only build the system, don't flash it",
+        default=False,
+        action="store_true",
+    )
+    parser.add_argument(
+        "--debug",
+        help="Print debug information",
+        default=False,
+        action="store_true",
     )
     parser.set_defaults(func=flash_command)