vars: support secrets for partitioning the disk

nixosModules/clanCore/vars/interface.nix

@@ -207,11 +207,14 @@ in
                       description = ''
                         This option determines when the secret will be decrypted and deployed to the target machine.
 
+                        By setting this to `partitioning`, the secret will be deployed prior to running `disko` allowing
+                        you to manage filesystem encryption keys. These will only be deployed when installing the system.
                         By setting this to `activation`, the secret will be deployed prior to running `nixos-rebuild` or `nixos-install`.
                         By setting this to `user`, the secret will be deployed prior to users and groups are created, allowing
                         users' passwords to be managed by vars. The secret will be stored in `/run/secrets-for-users` and `owner` and `group` must be `root`.
                       '';
                       type = lib.types.enum [
+                        "partitioning"
                         "activation"
                         "users"
                         "services"

nixosModules/clanCore/vars/secret/password-store.nix

@@ -42,6 +42,7 @@ let
   useSystemdActivation =
     (options.systemd ? sysusers && config.systemd.sysusers.enable)
     || (options.services ? userborn && config.services.userborn.enable);
+
   normalSecrets = lib.any (
     gen: lib.any (file: file.neededFor == "services") (lib.attrValues gen.files)
   ) (lib.attrValues config.clan.core.vars.generators);
@@ -75,7 +76,9 @@ in
                 else if file.config.neededFor == "services" then
                   "/run/secrets/${file.config.generatorName}/${file.config.name}"
                 else if file.config.neededFor == "activation" then
-                  "${config.clan.password-store.secretLocation}/${file.config.generatorName}/${file.config.name}"
+                  "${config.clan.password-store.secretLocation}/activation/${file.config.generatorName}/${file.config.name}"
+                else if file.config.neededFor == "partitioning" then
+                  "/run/partitioning-secrets/${file.config.generatorName}/${file.config.name}"
                 else
                   throw "unknown neededFor ${file.config.neededFor}";
 

nixosModules/clanCore/vars/secret/sops/default.nix

@@ -25,8 +25,10 @@ in
     # Before we generate a secret we cannot know the path yet, so we need to set it to an empty string
     fileModule = file: {
       path = lib.mkIf file.config.secret (
-        if file.config.neededFor == "activation" then
+        if file.config.neededFor == "partitioning" then
-          "/var/lib/sops-nix/${file.config.generatorName}/${file.config.name}"
+          "/run/partitioning-secrets/${file.config.generatorName}/${file.config.name}"
+        else if file.config.neededFor == "activation" then
+          "/var/lib/sops-nix/activation/${file.config.generatorName}/${file.config.name}"
         else
           config.sops.secrets.${"vars/${file.config.generatorName}/${file.config.name}"}.path
             or "/no-such-path"

nixosModules/clanCore/vars/secret/sops/funcs.nix

@@ -17,7 +17,9 @@ in
     let
       relevantFiles =
         generator:
-        filterAttrs (_name: f: f.secret && f.deploy && (f.neededFor != "activation")) generator.files;
+        filterAttrs (
+          _name: f: f.secret && f.deploy && (f.neededFor == "users" || f.neededFor == "services")
+        ) generator.files;
       allFiles = flatten (
         mapAttrsToList (
           gen_name: generator:

nixosModules/clanCore/vars/secret/vm.nix

@@ -6,7 +6,11 @@
 {
   config.clan.core.vars.settings = lib.mkIf (config.clan.core.vars.settings.secretStore == "vm") {
     fileModule = file: {
-      path = "/etc/secrets/${file.config.generatorName}/${file.config.name}";
+      path =
+        if file.config.neededFor == "partitioning" then
+          "/run/partitioning-secrets/${file.config.generatorName}/${file.config.name}"
+        else
+          "/etc/secrets/${file.config.generatorName}/${file.config.name}";
     };
     secretModule = "clan_cli.vars.secret_modules.vm";
   };

pkgs/clan-cli/clan_cli/machines/install.py

@@ -53,16 +53,20 @@ def install_machine(opts: InstallOptions) -> None:
     generate_facts([machine])
     generate_vars([machine])
 
-    with TemporaryDirectory(prefix="nixos-install-") as tmpdir_:
+    with TemporaryDirectory(prefix="nixos-install-") as base_directory:
-        tmpdir = Path(tmpdir_)
+        activation_secrets = Path(base_directory) / "activation_secrets"
-        upload_dir_ = machine.secrets_upload_directory
+        upload_dir = activation_secrets / machine.secrets_upload_directory.lstrip("/")
-
-        if upload_dir_.startswith("/"):
-            upload_dir_ = upload_dir_[1:]
-        upload_dir = tmpdir / upload_dir_
         upload_dir.mkdir(parents=True)
         machine.secret_facts_store.upload(upload_dir)
-        machine.secret_vars_store.populate_dir(upload_dir)
+        machine.secret_vars_store.populate_dir(
+            upload_dir, phases=["activation", "users", "services"]
+        )
+
+        partitioning_secrets = Path(base_directory) / "partitioning_secrets"
+        partitioning_secrets.mkdir(parents=True)
+        machine.secret_vars_store.populate_dir(
+            partitioning_secrets, phases=["partitioning"]
+        )
 
         if opts.password:
             os.environ["SSHPASS"] = opts.password
@@ -72,9 +76,22 @@ def install_machine(opts: InstallOptions) -> None:
             "--flake",
             f"{machine.flake}#{machine.name}",
             "--extra-files",
-            str(tmpdir),
+            str(activation_secrets),
         ]
 
+        for path in partitioning_secrets.rglob("*"):
+            if path.is_file():
+                cmd.extend(
+                    [
+                        "--disk-encryption-keys",
+                        str(
+                            "/run/partitioning-secrets"
+                            / path.relative_to(partitioning_secrets)
+                        ),
+                        str(path),
+                    ]
+                )
+
         if opts.no_reboot:
             cmd.append("--no-reboot")
 

pkgs/clan-cli/clan_cli/vars/_types.py

@@ -155,5 +155,9 @@ class StoreBase(ABC):
         return stored_hash == target_hash
 
     @abstractmethod
-    def populate_dir(self, output_dir: Path) -> None:
+    def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
+        pass
+
+    @abstractmethod
+    def upload(self, phases: list[str]) -> None:
         pass

pkgs/clan-cli/clan_cli/vars/public_modules/in_repo.py

@@ -50,6 +50,10 @@ class FactStore(StoreBase):
     def exists(self, generator: Generator, name: str) -> bool:
         return (self.directory(generator, name) / "value").exists()
 
-    def populate_dir(self, output_dir: Path) -> None:
+    def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
         msg = "populate_dir is not implemented for public vars stores"
         raise NotImplementedError(msg)
+
+    def upload(self, phases: list[str]) -> None:
+        msg = "upload is not implemented for public vars stores"
+        raise NotImplementedError(msg)

pkgs/clan-cli/clan_cli/vars/public_modules/vm.py

@@ -48,6 +48,10 @@ class FactStore(StoreBase):
         msg = f"Fact {name} for service {generator.name} not found"
         raise ClanError(msg)
 
-    def populate_dir(self, output_dir: Path) -> None:
+    def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
         msg = "populate_dir is not implemented for public vars stores"
         raise NotImplementedError(msg)
+
+    def upload(self, phases: list[str]) -> None:
+        msg = "upload is not implemented for public vars stores"
+        raise NotImplementedError(msg)

pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py

@@ -152,52 +152,77 @@ class SecretStore(StoreBase):
 
         return local_hash.decode() != remote_hash
 
-    def populate_dir(self, output_dir: Path) -> None:
+    def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
-        with (
+        if "users" in phases:
-            tarfile.open(output_dir / "secrets.tar.gz", "w:gz") as tar,
+            with tarfile.open(
-            tarfile.open(output_dir / "secrets_for_users.tar.gz", "w:gz") as user_tar,
+                output_dir / "secrets_for_users.tar.gz", "w:gz"
-        ):
+            ) as user_tar:
+                for generator in self.machine.vars_generators:
+                    dir_exists = False
+                    for file in generator.files:
+                        if not file.deploy:
+                            continue
+                        if not file.secret:
+                            continue
+                        tar_file = tarfile.TarInfo(name=f"{generator.name}/{file.name}")
+                        content = self.get(generator, file.name)
+                        tar_file.size = len(content)
+                        tar_file.mode = file.mode
+                        user_tar.addfile(tarinfo=tar_file, fileobj=io.BytesIO(content))
+
+        if "services" in phases:
+            with tarfile.open(output_dir / "secrets.tar.gz", "w:gz") as tar:
+                for generator in self.machine.vars_generators:
+                    dir_exists = False
+                    for file in generator.files:
+                        if not file.deploy:
+                            continue
+                        if not file.secret:
+                            continue
+                        if not dir_exists:
+                            tar_dir = tarfile.TarInfo(name=generator.name)
+                            tar_dir.type = tarfile.DIRTYPE
+                            tar_dir.mode = 0o511
+                            tar.addfile(tarinfo=tar_dir)
+                            dir_exists = True
+                        tar_file = tarfile.TarInfo(name=f"{generator.name}/{file.name}")
+                        content = self.get(generator, file.name)
+                        tar_file.size = len(content)
+                        tar_file.mode = file.mode
+                        tar_file.uname = file.owner
+                        tar_file.gname = file.group
+                        tar.addfile(tarinfo=tar_file, fileobj=io.BytesIO(content))
+        if "activation" in phases:
             for generator in self.machine.vars_generators:
-                dir_exists = False
                 for file in generator.files:
                     if file.needed_for == "activation":
-                        (output_dir / generator.name / file.name).parent.mkdir(
+                        out_file = (
-                            parents=True,
+                            output_dir / "activation" / generator.name / file.name
-                            exist_ok=True,
                         )
-                        (output_dir / generator.name / file.name).write_bytes(
+                        out_file.parent.mkdir(parents=True, exist_ok=True)
-                            self.get(generator, file.name)
+                        out_file.write_bytes(self.get(generator, file.name))
+        if "partitioning" in phases:
+            for generator in self.machine.vars_generators:
+                for file in generator.files:
+                    if file.needed_for == "partitioning":
+                        out_file = (
+                            output_dir / "partitioning" / generator.name / file.name
                         )
-                        continue
+                        out_file.parent.mkdir(parents=True, exist_ok=True)
-                    if not file.deploy:
+                        out_file.write_bytes(self.get(generator, file.name))
-                        continue
+
-                    if not file.secret:
-                        continue
-                    if not dir_exists and file.needed_for == "services":
-                        tar_dir = tarfile.TarInfo(name=generator.name)
-                        tar_dir.type = tarfile.DIRTYPE
-                        tar_dir.mode = 0o511
-                        tar.addfile(tarinfo=tar_dir)
-                        dir_exists = True
-                    tar_file = tarfile.TarInfo(name=f"{generator.name}/{file.name}")
-                    content = self.get(generator, file.name)
-                    tar_file.size = len(content)
-                    tar_file.mode = file.mode
-                    tar_file.uname = file.owner
-                    tar_file.gname = file.group
-                    if file.needed_for == "users":
-                        user_tar.addfile(tarinfo=tar_file, fileobj=io.BytesIO(content))
-                    else:
-                        tar.addfile(tarinfo=tar_file, fileobj=io.BytesIO(content))
         (output_dir / ".pass_info").write_bytes(self.generate_hash())
 
-    def upload(self) -> None:
+    def upload(self, phases: list[str]) -> None:
+        if "partitioning" in phases:
+            msg = "Cannot upload partitioning secrets"
+            raise NotImplementedError(msg)
         if not self.needs_upload():
             log.info("Secrets already uploaded")
             return
         with TemporaryDirectory(prefix="vars-upload-") as tempdir:
             pass_dir = Path(tempdir)
-            self.populate_dir(pass_dir)
+            self.populate_dir(pass_dir, phases)
             upload_dir = Path(
                 self.machine.deployment["password-store"]["secretLocation"]
             )

pkgs/clan-cli/clan_cli/vars/secret_modules/sops.py

@@ -163,34 +163,56 @@ class SecretStore(StoreBase):
             self.machine.flake_dir, self.secret_path(generator, name)
         ).encode("utf-8")
 
-    def populate_dir(self, output_dir: Path) -> None:
+    def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
-        key_name = f"{self.machine.name}-age.key"
+        if "users" in phases or "services" in phases:
-        if not has_secret(sops_secrets_folder(self.machine.flake_dir) / key_name):
+            key_name = f"{self.machine.name}-age.key"
-            # skip uploading the secret, not managed by us
+            if not has_secret(sops_secrets_folder(self.machine.flake_dir) / key_name):
-            return
+                # skip uploading the secret, not managed by us
-        key = decrypt_secret(
+                return
-            self.machine.flake_dir,
+            key = decrypt_secret(
-            sops_secrets_folder(self.machine.flake_dir) / key_name,
+                self.machine.flake_dir,
-        )
+                sops_secrets_folder(self.machine.flake_dir) / key_name,
-        (output_dir / "key.txt").touch(mode=0o600)
+            )
-        (output_dir / "key.txt").write_text(key)
+            (output_dir / "key.txt").touch(mode=0o600)
-        for generator in self.machine.vars_generators:
+            (output_dir / "key.txt").write_text(key)
-            for file in generator.files:
-                if file.needed_for == "activation":
-                    target_path = output_dir / generator.name / file.name
-                    target_path.parent.mkdir(
-                        parents=True,
-                        exist_ok=True,
-                    )
-                    # chmod after in case it doesn't have u+w
-                    target_path.touch(mode=0o600)
-                    target_path.write_bytes(self.get(generator, file.name))
-                    target_path.chmod(file.mode)
 
-    def upload(self) -> None:
+        if "activation" in phases:
+            for generator in self.machine.vars_generators:
+                for file in generator.files:
+                    if file.needed_for == "activation":
+                        target_path = (
+                            output_dir / "activation" / generator.name / file.name
+                        )
+                        target_path.parent.mkdir(
+                            parents=True,
+                            exist_ok=True,
+                        )
+                        # chmod after in case it doesn't have u+w
+                        target_path.touch(mode=0o600)
+                        target_path.write_bytes(self.get(generator, file.name))
+                        target_path.chmod(file.mode)
+
+        if "partitioning" in phases:
+            for generator in self.machine.vars_generators:
+                for file in generator.files:
+                    if file.needed_for == "partitioning":
+                        target_path = output_dir / generator.name / file.name
+                        target_path.parent.mkdir(
+                            parents=True,
+                            exist_ok=True,
+                        )
+                        # chmod after in case it doesn't have u+w
+                        target_path.touch(mode=0o600)
+                        target_path.write_bytes(self.get(generator, file.name))
+                        target_path.chmod(file.mode)
+
+    def upload(self, phases: list[str]) -> None:
+        if "partitioning" in phases:
+            msg = "Cannot upload partitioning secrets"
+            raise NotImplementedError(msg)
         with TemporaryDirectory(prefix="sops-upload-") as tempdir:
             sops_upload_dir = Path(tempdir)
-            self.populate_dir(sops_upload_dir)
+            self.populate_dir(sops_upload_dir, phases)
             upload(self.machine.target_host, sops_upload_dir, Path("/var/lib/sops-nix"))
 
     def exists(self, generator: Generator, name: str) -> bool:

pkgs/clan-cli/clan_cli/vars/secret_modules/vm.py

@@ -32,11 +32,18 @@ class SecretStore(StoreBase):
         secret_file.write_bytes(value)
         return None  # we manage the files outside of the git repo
 
+    def exists(self, generator: "Generator", name: str) -> bool:
+        return (self.dir / generator.name / name).exists()
+
     def get(self, generator: Generator, name: str) -> bytes:
         secret_file = self.dir / generator.name / name
         return secret_file.read_bytes()
 
-    def populate_dir(self, output_dir: Path) -> None:
+    def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
         if output_dir.exists():
             shutil.rmtree(output_dir)
         shutil.copytree(self.dir, output_dir)
+
+    def upload(self, phases: list[str]) -> None:
+        msg = "Cannot upload secrets to VMs"
+        raise NotImplementedError(msg)

pkgs/clan-cli/clan_cli/vars/upload.py

@@ -11,7 +11,7 @@ log = logging.getLogger(__name__)
 def upload_secret_vars(machine: Machine) -> None:
     secret_store_module = importlib.import_module(machine.secret_vars_module)
     secret_store = secret_store_module.SecretStore(machine=machine)
-    secret_store.upload()
+    secret_store.upload(phases=["activation", "users", "services"])
 
 
 def upload_command(args: argparse.Namespace) -> None: