Merge pull request 'vars: global metadata paths for all store backends' (#2032) from DavHau/clan-core:DavHau-dave into main

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

@@ -18,14 +18,19 @@ let
 
   metaData = sopsFile: if pathExists (metaFile sopsFile) then importJSON (metaFile sopsFile) else { };
 
-  toDeploy = secret: (metaData secret.sopsFile).deploy or true;
+  isSopsSecret =
+    secret:
+    let
+      meta = metaData secret.sopsFile;
+    in
+    meta.store or null == "sops" && meta.deployed or true && meta.secret or true;
 
-  varsDirMachines = config.clan.core.clanDir + "/sops/vars/per-machine/${machineName}";
+  varsDirMachines = config.clan.core.clanDir + "/vars/per-machine/${machineName}";
-  varsDirShared = config.clan.core.clanDir + "/sops/vars/shared";
+  varsDirShared = config.clan.core.clanDir + "/vars/shared";
 
   vars' = (listVars varsDirMachines) ++ (listVars varsDirShared);
 
-  vars = lib.filter (secret: toDeploy secret) vars';
+  vars = lib.filter isSopsSecret vars';
 in
 {
   config.clan.core.vars.settings = lib.mkIf (config.clan.core.vars.settings.secretStore == "sops") {

pkgs/clan-cli/clan_cli/secrets/secrets.py

@@ -88,10 +88,7 @@ def encrypt_secret(
     add_users: list[str] | None = None,
     add_machines: list[str] | None = None,
     add_groups: list[str] | None = None,
-    meta: dict | None = None,
 ) -> None:
-    if meta is None:
-        meta = {}
     if add_groups is None:
         add_groups = []
     if add_machines is None:
@@ -146,7 +143,7 @@ def encrypt_secret(
         )
 
     secret_path = secret_path / "secret"
-    encrypt_file(secret_path, value, sorted(recipient_keys), meta)
+    encrypt_file(secret_path, value, sorted(recipient_keys))
     files_to_commit.append(secret_path)
     commit_files(
         files_to_commit,

pkgs/clan-cli/clan_cli/secrets/sops.py

@@ -145,10 +145,7 @@ def encrypt_file(
     secret_path: Path,
     content: IO[str] | str | bytes | None,
     pubkeys: list[str],
-    meta: dict | None = None,
 ) -> None:
-    if meta is None:
-        meta = {}
     folder = secret_path.parent
     folder.mkdir(parents=True, exist_ok=True)
 
@@ -190,9 +187,6 @@ def encrypt_file(
                 with NamedTemporaryFile(dir=folder, delete=False) as f2:
                     shutil.copyfile(f.name, f2.name)
                     Path(f2.name).rename(secret_path)
-                meta_path = secret_path.parent / "meta.json"
-                with meta_path.open("w") as f_meta:
-                    json.dump(meta, f_meta, indent=2)
             finally:
                 with suppress(OSError):
                     Path(f.name).unlink()

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

@@ -1,4 +1,5 @@
 # !/usr/bin/env python3
+import json
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from pathlib import Path
@@ -25,31 +26,81 @@ class StoreBase(ABC):
     def __init__(self, machine: Machine) -> None:
         self.machine = machine
 
+    @property
     @abstractmethod
-    def exists(self, service: str, name: str, shared: bool = False) -> bool:
+    def store_name(self) -> str:
         pass
 
     # get a single fact
     @abstractmethod
-    def get(self, service: str, name: str, shared: bool = False) -> bytes:
+    def get(self, generator_name: str, name: str, shared: bool = False) -> bytes:
         pass
 
     @abstractmethod
-    def set(
+    def _set(
         self,
-        service: str,
+        generator_name: str,
         name: str,
         value: bytes,
         shared: bool = False,
         deployed: bool = True,
     ) -> Path | None:
-        pass
+        """
+        override this method to implement the actual creation of the file
+        """
 
     @property
     @abstractmethod
     def is_secret_store(self) -> bool:
         pass
 
+    def directory(
+        self, generator_name: str, var_name: str, shared: bool = False
+    ) -> Path:
+        if shared:
+            base_path = self.machine.flake_dir / "vars" / "shared"
+        else:
+            base_path = (
+                self.machine.flake_dir / "vars" / "per-machine" / self.machine.name
+            )
+        return base_path / generator_name / var_name
+
+    def exists(self, generator_name: str, name: str, shared: bool = False) -> bool:
+        directory = self.directory(generator_name, name, shared)
+        if not (directory / "meta.json").exists():
+            return False
+        with (directory / "meta.json").open() as f:
+            meta = json.load(f)
+        # check if is secret, as secret and public store names could collide (eg. 'vm')
+        if meta.get("secret") != self.is_secret_store:
+            return False
+        return meta.get("store") == self.store_name
+
+    def set(
+        self,
+        generator_name: str,
+        var_name: str,
+        value: bytes,
+        shared: bool = False,
+        deployed: bool = True,
+    ) -> Path | None:
+        directory = self.directory(generator_name, var_name, shared)
+        # delete directory
+        if directory.exists():
+            for f in directory.glob("*"):
+                f.unlink()
+        # re-create directory
+        directory.mkdir(parents=True, exist_ok=True)
+        new_file = self._set(generator_name, var_name, value, shared, deployed)
+        meta = {
+            "deployed": deployed,
+            "secret": self.is_secret_store,
+            "store": self.store_name,
+        }
+        with (directory / "meta.json").open("w") as file:
+            json.dump(meta, file, indent=2)
+        return new_file
+
     def get_all(self) -> list[Var]:
         all_vars = []
         for gen_name, generator in self.machine.vars_generators.items():

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

@@ -17,10 +17,10 @@ def check_vars(machine: Machine, generator_name: None | str = None) -> bool:
     missing_secret_vars = []
     missing_public_vars = []
     if generator_name:
-        services = [generator_name]
+        generators = [generator_name]
     else:
-        services = list(machine.vars_generators.keys())
+        generators = list(machine.vars_generators.keys())
-    for generator_name in services:
+    for generator_name in generators:
         for name, file in machine.vars_generators[generator_name]["files"].items():
             if file["secret"] and not secret_vars_store.exists(generator_name, name):
                 log.info(

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

@@ -10,17 +10,12 @@ class FactStore(FactStoreBase):
     def __init__(self, machine: Machine) -> None:
         self.machine = machine
         self.works_remotely = False
-        self.per_machine_folder = (
-            self.machine.flake_dir / "vars" / "per-machine" / self.machine.name
-        )
-        self.shared_folder = self.machine.flake_dir / "vars" / "shared"
 
-    def _var_path(self, generator_name: str, name: str, shared: bool) -> Path:
+    @property
-        if shared:
+    def store_name(self) -> str:
-            return self.shared_folder / generator_name / name
+        return "in_repo"
-        return self.per_machine_folder / generator_name / name
 
-    def set(
+    def _set(
         self,
         generator_name: str,
         name: str,
@@ -29,17 +24,14 @@ class FactStore(FactStoreBase):
         deployed: bool = True,
     ) -> Path | None:
         if self.machine.flake.is_local():
-            fact_path = self._var_path(generator_name, name, shared)
+            file_path = self.directory(generator_name, name, shared) / "value"
-            fact_path.parent.mkdir(parents=True, exist_ok=True)
+            file_path.parent.mkdir(parents=True, exist_ok=True)
-            fact_path.touch()
+            file_path.touch()
-            fact_path.write_bytes(value)
+            file_path.write_bytes(value)
-            return fact_path
+            return file_path
         msg = f"in_flake fact storage is only supported for local flakes: {self.machine.flake}"
         raise ClanError(msg)
 
-    def exists(self, generator_name: str, name: str, shared: bool = False) -> bool:
-        return self._var_path(generator_name, name, shared).exists()
-
     # get a single fact
     def get(self, generator_name: str, name: str, shared: bool = False) -> bytes:
-        return self._var_path(generator_name, name, shared).read_bytes()
+        return (self.directory(generator_name, name, shared) / "value").read_bytes()

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

@@ -17,11 +17,15 @@ class FactStore(FactStoreBase):
         self.dir = vm_state_dir(str(machine.flake), machine.name) / "facts"
         log.debug(f"FactStore initialized with dir {self.dir}")
 
+    @property
+    def store_name(self) -> str:
+        return "vm"
+
     def exists(self, service: str, name: str, shared: bool = False) -> bool:
         fact_path = self.dir / service / name
         return fact_path.exists()
 
-    def set(
+    def _set(
         self,
         service: str,
         name: str,

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

@@ -13,6 +13,10 @@ class SecretStore(SecretStoreBase):
     def __init__(self, machine: Machine) -> None:
         self.machine = machine
 
+    @property
+    def store_name(self) -> str:
+        return "password_store"
+
     @property
     def _password_store_dir(self) -> str:
         return os.environ.get(
@@ -24,7 +28,7 @@ class SecretStore(SecretStoreBase):
             return Path(f"shared/{generator_name}/{name}")
         return Path(f"machines/{self.machine.name}/{generator_name}/{name}")
 
-    def set(
+    def _set(
         self,
         generator_name: str,
         name: str,
@@ -62,6 +66,8 @@ class SecretStore(SecretStoreBase):
         ).stdout
 
     def exists(self, generator_name: str, name: str, shared: bool = False) -> bool:
+        if not super().exists(generator_name, name, shared):
+            return False
         return (
             Path(self._password_store_dir)
             / f"{self._var_path(generator_name, name, shared)}.gpg"

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

@@ -36,22 +36,16 @@ class SecretStore(SecretStoreBase):
         )
         add_machine(self.machine.flake_dir, self.machine.name, pub_key, False)
 
+    @property
+    def store_name(self) -> str:
+        return "sops"
+
     def secret_path(
         self, generator_name: str, secret_name: str, shared: bool = False
     ) -> Path:
-        if shared:
+        return self.directory(generator_name, secret_name, shared=shared)
-            base_path = self.machine.flake_dir / "sops" / "vars" / "shared"
-        else:
-            base_path = (
-                self.machine.flake_dir
-                / "sops"
-                / "vars"
-                / "per-machine"
-                / self.machine.name
-            )
-        return base_path / generator_name / secret_name
 
-    def set(
+    def _set(
         self,
         generator_name: str,
         name: str,
@@ -66,9 +60,6 @@ class SecretStore(SecretStoreBase):
             value,
             add_machines=[self.machine.name],
             add_groups=self.machine.deployment["sops"]["defaultGroups"],
-            meta={
-                "deploy": deployed,
-            },
         )
         return path
 
@@ -77,11 +68,6 @@ class SecretStore(SecretStoreBase):
             self.machine.flake_dir, self.secret_path(generator_name, name, shared)
         ).encode("utf-8")
 
-    def exists(self, generator_name: str, name: str, shared: bool = False) -> bool:
-        return has_secret(
-            self.secret_path(generator_name, name, shared),
-        )
-
     def upload(self, output_dir: Path) -> None:
         key_name = f"{self.machine.name}-age.key"
         if not has_secret(sops_secrets_folder(self.machine.flake_dir) / key_name):

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

@@ -13,7 +13,11 @@ class SecretStore(SecretStoreBase):
         self.dir = vm_state_dir(str(machine.flake), machine.name) / "secrets"
         self.dir.mkdir(parents=True, exist_ok=True)
 
-    def set(
+    @property
+    def store_name(self) -> str:
+        return "vm"
+
+    def _set(
         self,
         service: str,
         name: str,
@@ -30,9 +34,6 @@ class SecretStore(SecretStoreBase):
         secret_file = self.dir / service / name
         return secret_file.read_bytes()
 
-    def exists(self, service: str, name: str, shared: bool = False) -> bool:
-        return (self.dir / service / name).exists()
-
     def upload(self, output_dir: Path) -> None:
         if output_dir.exists():
             shutil.rmtree(output_dir)