Merge pull request 'make secrets stores inherit from an interface' (#857) from Mic92-target_host into main

pkgs/clan-cli/clan_cli/facts/modules/__init__.py

@@ -0,0 +1,28 @@
+from abc import ABC, abstractmethod
+from pathlib import Path
+
+from clan_cli.machines.machines import Machine
+
+
+class FactStoreBase(ABC):
+    @abstractmethod
+    def __init__(self, machine: Machine) -> None:
+        pass
+
+    @abstractmethod
+    def exists(self, service: str, name: str) -> bool:
+        pass
+
+    @abstractmethod
+    def set(self, service: str, name: str, value: bytes) -> Path | None:
+        pass
+
+    # get a single fact
+    @abstractmethod
+    def get(self, service: str, name: str) -> bytes:
+        pass
+
+    # get all facts
+    @abstractmethod
+    def get_all(self) -> dict[str, dict[str, bytes]]:
+        pass

pkgs/clan-cli/clan_cli/facts/modules/in_repo.py

@@ -3,13 +3,15 @@ from pathlib import Path
 from clan_cli.errors import ClanError
 from clan_cli.machines.machines import Machine
 
+from . import FactStoreBase
 
-class FactStore:
+
+class FactStore(FactStoreBase):
     def __init__(self, machine: Machine) -> None:
         self.machine = machine
         self.works_remotely = False
 
-    def set(self, _service: str, name: str, value: bytes) -> Path | None:
+    def set(self, service: str, name: str, value: bytes) -> Path | None:
         if isinstance(self.machine.flake, Path):
             fact_path = (
                 self.machine.flake / "machines" / self.machine.name / "facts" / name
@@ -23,14 +25,14 @@ class FactStore:
                 f"in_flake fact storage is only supported for local flakes: {self.machine.flake}"
             )
 
-    def exists(self, _service: str, name: str) -> bool:
+    def exists(self, service: str, name: str) -> bool:
         fact_path = (
             self.machine.flake_dir / "machines" / self.machine.name / "facts" / name
         )
         return fact_path.exists()
 
     # get a single fact
-    def get(self, _service: str, name: str) -> bytes:
+    def get(self, service: str, name: str) -> bytes:
         fact_path = (
             self.machine.flake_dir / "machines" / self.machine.name / "facts" / name
         )

pkgs/clan-cli/clan_cli/facts/modules/vm.py

@@ -5,10 +5,12 @@ from clan_cli.dirs import vm_state_dir
 from clan_cli.errors import ClanError
 from clan_cli.machines.machines import Machine
 
+from . import FactStoreBase
+
 log = logging.getLogger(__name__)
 
 
-class FactStore:
+class FactStore(FactStoreBase):
     def __init__(self, machine: Machine) -> None:
         self.machine = machine
         self.works_remotely = False

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

@@ -8,14 +8,93 @@ from tempfile import TemporaryDirectory
 from clan_cli.cmd import run
 
 from ..errors import ClanError
+from ..facts.modules import FactStoreBase
 from ..git import commit_files
 from ..machines.machines import Machine
 from ..nix import nix_shell
 from .check import check_secrets
+from .modules import SecretStoreBase
 
 log = logging.getLogger(__name__)
 
 
+def generate_service_secrets(
+    machine: Machine,
+    service: str,
+    secret_store: SecretStoreBase,
+    fact_store: FactStoreBase,
+    tmpdir: Path,
+) -> None:
+    service_dir = tmpdir / service
+    # check if all secrets exist and generate them if at least one is missing
+    needs_regeneration = not check_secrets(machine)
+    log.debug(f"{service} needs_regeneration: {needs_regeneration}")
+    if needs_regeneration:
+        if not isinstance(machine.flake, Path):
+            msg = f"flake is not a Path: {machine.flake}"
+            msg += "fact/secret generation is only supported for local flakes"
+
+        env = os.environ.copy()
+        facts_dir = service_dir / "facts"
+        facts_dir.mkdir(parents=True)
+        env["facts"] = str(facts_dir)
+        secrets_dir = service_dir / "secrets"
+        secrets_dir.mkdir(parents=True)
+        env["secrets"] = str(secrets_dir)
+        # fmt: off
+        cmd = nix_shell(
+            [
+                "nixpkgs#bash",
+                "nixpkgs#bubblewrap",
+            ],
+            [
+                "bwrap",
+                "--ro-bind", "/nix/store", "/nix/store",
+                "--tmpfs",  "/usr/lib/systemd",
+                "--dev", "/dev",
+                "--bind", str(facts_dir), str(facts_dir),
+                "--bind", str(secrets_dir), str(secrets_dir),
+                "--unshare-all",
+                "--unshare-user",
+                "--uid", "1000",
+                "--",
+                "bash", "-c", machine.secrets_data[service]["generator"]
+            ],
+        )
+        # fmt: on
+        run(
+            cmd,
+            env=env,
+        )
+        files_to_commit = []
+        # store secrets
+        for secret in machine.secrets_data[service]["secrets"]:
+            secret_file = secrets_dir / secret
+            if not secret_file.is_file():
+                msg = f"did not generate a file for '{secret}' when running the following command:\n"
+                msg += machine.secrets_data[service]["generator"]
+                raise ClanError(msg)
+            secret_path = secret_store.set(service, secret, secret_file.read_bytes())
+            if secret_path:
+                files_to_commit.append(secret_path)
+
+        # store facts
+        for name in machine.secrets_data[service]["facts"]:
+            fact_file = facts_dir / name
+            if not fact_file.is_file():
+                msg = f"did not generate a file for '{name}' when running the following command:\n"
+                msg += machine.secrets_data[service]["generator"]
+                raise ClanError(msg)
+            fact_file = fact_store.set(service, name, fact_file.read_bytes())
+            if fact_file:
+                files_to_commit.append(fact_file)
+        commit_files(
+            files_to_commit,
+            machine.flake_dir,
+            f"Update facts/secrets for service {service} in machine {machine.name}",
+        )
+
+
 def generate_secrets(machine: Machine) -> None:
     secrets_module = importlib.import_module(machine.secrets_module)
     secret_store = secrets_module.SecretStore(machine=machine)
@@ -23,78 +102,10 @@ def generate_secrets(machine: Machine) -> None:
     facts_module = importlib.import_module(machine.facts_module)
     fact_store = facts_module.FactStore(machine=machine)
 
-    with TemporaryDirectory() as d:
+    with TemporaryDirectory() as tmp:
+        tmpdir = Path(tmp)
         for service in machine.secrets_data:
-            tmpdir = Path(d) / service
-            # check if all secrets exist and generate them if at least one is missing
-            needs_regeneration = not check_secrets(machine)
-            log.debug(f"{service} needs_regeneration: {needs_regeneration}")
-            if needs_regeneration:
-                if not isinstance(machine.flake, Path):
-                    msg = f"flake is not a Path: {machine.flake}"
-                    msg += "fact/secret generation is only supported for local flakes"
-
-                env = os.environ.copy()
-                facts_dir = tmpdir / "facts"
-                facts_dir.mkdir(parents=True)
-                env["facts"] = str(facts_dir)
-                secrets_dir = tmpdir / "secrets"
-                secrets_dir.mkdir(parents=True)
-                env["secrets"] = str(secrets_dir)
-                # fmt: off
-                cmd = nix_shell(
-                    [
-                        "nixpkgs#bash",
-                        "nixpkgs#bubblewrap",
-                    ],
-                    [
-                        "bwrap",
-                        "--ro-bind", "/nix/store", "/nix/store",
-                        "--tmpfs",  "/usr/lib/systemd",
-                        "--dev", "/dev",
-                        "--bind", str(facts_dir), str(facts_dir),
-                        "--bind", str(secrets_dir), str(secrets_dir),
-                        "--unshare-all",
-                        "--unshare-user",
-                        "--uid", "1000",
-                        "--",
-                        "bash", "-c", machine.secrets_data[service]["generator"]
-                    ],
-                )
-                # fmt: on
-                run(
-                    cmd,
-                    env=env,
-                )
-                files_to_commit = []
-                # store secrets
-                for secret in machine.secrets_data[service]["secrets"]:
-                    secret_file = secrets_dir / secret
-                    if not secret_file.is_file():
-                        msg = f"did not generate a file for '{secret}' when running the following command:\n"
-                        msg += machine.secrets_data[service]["generator"]
-                        raise ClanError(msg)
-                    secret_path = secret_store.set(
-                        service, secret, secret_file.read_bytes()
-                    )
-                    if secret_path:
-                        files_to_commit.append(secret_path)
-
-                # store facts
-                for name in machine.secrets_data[service]["facts"]:
-                    fact_file = facts_dir / name
-                    if not fact_file.is_file():
-                        msg = f"did not generate a file for '{name}' when running the following command:\n"
-                        msg += machine.secrets_data[service]["generator"]
-                        raise ClanError(msg)
-                    fact_file = fact_store.set(service, name, fact_file.read_bytes())
-                    if fact_file:
-                        files_to_commit.append(fact_file)
-                commit_files(
-                    files_to_commit,
-                    machine.flake_dir,
-                    f"Update facts/secrets for service {service} in machine {machine.name}",
-                )
+            generate_service_secrets(machine, service, secret_store, fact_store, tmpdir)
 
     print("successfully generated secrets")
 

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

@@ -0,0 +1,34 @@
+from abc import ABC, abstractmethod
+from pathlib import Path
+
+from clan_cli.machines.machines import Machine
+
+
+class SecretStoreBase(ABC):
+    @abstractmethod
+    def __init__(self, machine: Machine) -> None:
+        pass
+
+    @abstractmethod
+    def set(self, service: str, name: str, value: bytes) -> Path | None:
+        pass
+
+    @abstractmethod
+    def get(self, service: str, name: str) -> bytes:
+        pass
+
+    @abstractmethod
+    def exists(self, service: str, name: str) -> bool:
+        pass
+
+    @abstractmethod
+    def generate_hash(self) -> bytes:
+        pass
+
+    @abstractmethod
+    def update_check(self) -> bool:
+        pass
+
+    @abstractmethod
+    def upload(self, output_dir: Path) -> None:
+        pass

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

@@ -5,12 +5,14 @@ from pathlib import Path
 from clan_cli.machines.machines import Machine
 from clan_cli.nix import nix_shell
 
+from . import SecretStoreBase
 
-class SecretStore:
+
+class SecretStore(SecretStoreBase):
     def __init__(self, machine: Machine) -> None:
         self.machine = machine
 
-    def set(self, _service: str, name: str, value: bytes) -> Path | None:
+    def set(self, service: str, name: str, value: bytes) -> Path | None:
         subprocess.run(
             nix_shell(
                 ["nixpkgs#pass"],
@@ -21,7 +23,7 @@ class SecretStore:
         )
         return None  # we manage the files outside of the git repo
 
-    def get(self, _service: str, name: str) -> bytes:
+    def get(self, service: str, name: str) -> bytes:
         return subprocess.run(
             nix_shell(
                 ["nixpkgs#pass"],
@@ -31,7 +33,7 @@ class SecretStore:
             stdout=subprocess.PIPE,
         ).stdout
 
-    def exists(self, _service: str, name: str) -> bool:
+    def exists(self, service: str, name: str) -> bool:
         password_store = os.environ.get(
             "PASSWORD_STORE_DIR", f"{os.environ['HOME']}/.password-store"
         )

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

@@ -28,7 +28,7 @@ class SecretStore:
         )
         add_machine(self.machine.flake_dir, self.machine.name, pub_key, False)
 
-    def set(self, _service: str, name: str, value: bytes) -> Path | None:
+    def set(self, service: str, name: str, value: bytes) -> Path | None:
         path = (
             sops_secrets_folder(self.machine.flake_dir) / f"{self.machine.name}-{name}"
         )
@@ -40,15 +40,18 @@ class SecretStore:
         )
         return path
 
-    def get(self, _service: str, _name: str) -> bytes:
+    def get(self, service: str, _name: str) -> bytes:
         raise NotImplementedError()
 
-    def exists(self, _service: str, name: str) -> bool:
+    def exists(self, service: str, name: str) -> bool:
         return has_secret(
             self.machine.flake_dir,
             f"{self.machine.name}-{name}",
         )
 
+    def update_check(self) -> bool:
+        return False
+
     def upload(self, output_dir: Path) -> None:
         key_name = f"{self.machine.name}-age.key"
         if not has_secret(self.machine.flake_dir, key_name):

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

@@ -5,8 +5,10 @@ from pathlib import Path
 from clan_cli.dirs import vm_state_dir
 from clan_cli.machines.machines import Machine
 
+from . import SecretStoreBase
 
-class SecretStore:
+
+class SecretStore(SecretStoreBase):
     def __init__(self, machine: Machine) -> None:
         self.machine = machine
         self.dir = vm_state_dir(str(machine.flake), machine.name) / "secrets"
@@ -25,6 +27,9 @@ class SecretStore:
     def exists(self, service: str, name: str) -> bool:
         return (self.dir / service / name).exists()
 
+    def update_check(self) -> bool:
+        return False
+
     def upload(self, output_dir: Path) -> None:
         if os.path.exists(output_dir):
             shutil.rmtree(output_dir)

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

@@ -15,11 +15,9 @@ def upload_secrets(machine: Machine) -> None:
     secrets_module = importlib.import_module(machine.secrets_module)
     secret_store = secrets_module.SecretStore(machine=machine)
 
-    update_check = getattr(secret_store, "update_check", None)
-    if callable(update_check):
-        if update_check():
-            log.info("Secrets already up to date")
-            return
+    if secret_store.update_check():
+        log.info("Secrets already up to date")
+        return
     with TemporaryDirectory() as tempdir:
         secret_store.upload(Path(tempdir))
         host = machine.target_host