diff --git a/pkgs/clan-cli/clan_cli/vars/_types.py b/pkgs/clan-cli/clan_cli/vars/_types.py index 7269cd625..b058afe6c 100644 --- a/pkgs/clan-cli/clan_cli/vars/_types.py +++ b/pkgs/clan-cli/clan_cli/vars/_types.py @@ -1,5 +1,6 @@ import logging from abc import ABC, abstractmethod +from collections.abc import Iterable from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING @@ -122,6 +123,29 @@ class StoreBase(ABC): ) return new_file + @abstractmethod + def delete(self, generator: "Generator", name: str) -> Iterable[Path]: + """Remove a var from the store. + + :return: An iterable of affected paths in the git repository. This + may be empty if the store is outside of the repository. + """ + + @abstractmethod + def delete_store(self) -> Iterable[Path]: + """Delete the store (all vars) for this machine. + + .. note:: + + This does not make the distinction between public and private vars. + Since the public and private store of a machine can be co-located + under the same directory, this method's implementation has to be + idempotent. + + :return: An iterable of affected paths in the git repository. This + may be empty if the store was outside of the repository. + """ + def get_validation(self, generator: "Generator") -> str | None: """ Return the invalidation hash that indicates if a generator needs to be re-run diff --git a/pkgs/clan-cli/clan_cli/vars/public_modules/in_repo.py b/pkgs/clan-cli/clan_cli/vars/public_modules/in_repo.py index 51219635a..566d84dec 100644 --- a/pkgs/clan-cli/clan_cli/vars/public_modules/in_repo.py +++ b/pkgs/clan-cli/clan_cli/vars/public_modules/in_repo.py @@ -1,4 +1,5 @@ import shutil +from collections.abc import Iterable from pathlib import Path from clan_cli.errors import ClanError @@ -30,15 +31,14 @@ class FactStore(StoreBase): msg = f"in_flake fact storage is only supported for local flakes: {self.machine.flake}" raise ClanError(msg) folder = self.directory(generator, var.name) + file_path = folder / "value" if folder.exists(): - if not (folder / "value").exists(): + if not file_path.exists(): # another backend has used that folder before -> error out self.backend_collision_error(folder) shutil.rmtree(folder) # re-create directory - file_path = folder / "value" folder.mkdir(parents=True, exist_ok=True) - file_path.parent.mkdir(parents=True, exist_ok=True) file_path.touch() file_path.write_bytes(value) return file_path @@ -50,6 +50,24 @@ class FactStore(StoreBase): def exists(self, generator: Generator, name: str) -> bool: return (self.directory(generator, name) / "value").exists() + def delete(self, generator: Generator, name: str) -> Iterable[Path]: + fact_folder = self.directory(generator, name) + fact_file = fact_folder / "value" + fact_file.unlink() + empty = None + if next(fact_folder.iterdir(), empty) is not empty: + return [fact_file] + fact_folder.rmdir() + return [fact_folder] + + def delete_store(self) -> Iterable[Path]: + flake_root = Path(self.machine.flake_dir) + store_folder = flake_root / "vars/per-machine" / self.machine.name + if not store_folder.exists(): + return [] + shutil.rmtree(store_folder) + return [store_folder] + def populate_dir(self, output_dir: Path, phases: list[str]) -> None: msg = "populate_dir is not implemented for public vars stores" raise NotImplementedError(msg) diff --git a/pkgs/clan-cli/clan_cli/vars/public_modules/vm.py b/pkgs/clan-cli/clan_cli/vars/public_modules/vm.py index 315a4e9b6..a2719c8fb 100644 --- a/pkgs/clan-cli/clan_cli/vars/public_modules/vm.py +++ b/pkgs/clan-cli/clan_cli/vars/public_modules/vm.py @@ -1,4 +1,6 @@ import logging +import shutil +from collections.abc import Iterable from pathlib import Path from clan_cli.dirs import vm_state_dir @@ -48,6 +50,21 @@ class FactStore(StoreBase): msg = f"Fact {name} for service {generator.name} not found" raise ClanError(msg) + def delete(self, generator: Generator, name: str) -> Iterable[Path]: + fact_dir = self.dir / generator.name + fact_file = fact_dir / name + fact_file.unlink() + empty = None + if next(fact_dir.iterdir(), empty) is empty: + fact_dir.rmdir() + return [fact_file] + + def delete_store(self) -> Iterable[Path]: + if not self.dir.exists(): + return [] + shutil.rmtree(self.dir) + return [self.dir] + def populate_dir(self, output_dir: Path, phases: list[str]) -> None: msg = "populate_dir is not implemented for public vars stores" raise NotImplementedError(msg) diff --git a/pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py b/pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py index 61e4e699f..c82b59fa4 100644 --- a/pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py +++ b/pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py @@ -2,11 +2,12 @@ import io import logging import os import tarfile +from collections.abc import Iterable from itertools import chain from pathlib import Path from tempfile import TemporaryDirectory -from clan_cli.cmd import Log, RunOpts, run +from clan_cli.cmd import CmdOut, Log, RunOpts, run from clan_cli.machines.machines import Machine from clan_cli.nix import nix_shell from clan_cli.ssh.upload import upload @@ -35,52 +36,51 @@ class SecretStore(StoreBase): return backend @property - def _password_store_dir(self) -> str: + def _password_store_dir(self) -> Path: if self._store_backend == "passage": - return os.environ.get("PASSAGE_DIR", f"{os.environ['HOME']}/.passage/store") - return os.environ.get( - "PASSWORD_STORE_DIR", f"{os.environ['HOME']}/.password-store" - ) + lookup = os.environ.get("PASSAGE_DIR") + default = Path.home() / ".passage/store" + else: + lookup = os.environ.get("PASSWORD_STORE_DIR") + default = Path.home() / ".password-store" + return Path(lookup) if lookup else default def entry_dir(self, generator: Generator, name: str) -> Path: return Path(self.entry_prefix) / self.rel_dir(generator, name) + def _run_pass(self, *args: str, options: RunOpts | None = None) -> CmdOut: + cmd = nix_shell(packages=["nixpkgs#pass"], cmd=[self._store_backend, *args]) + return run(cmd, options) + def _set( self, generator: Generator, var: Var, value: bytes, ) -> Path | None: - run( - nix_shell( - [f"nixpkgs#{self._store_backend}"], - [ - f"{self._store_backend}", - "insert", - "-m", - str(self.entry_dir(generator, var.name)), - ], - ), - RunOpts(input=value, check=True), - ) + pass_call = ["insert", "-m", str(self.entry_dir(generator, var.name))] + self._run_pass(*pass_call, options=RunOpts(input=value, check=True)) return None # we manage the files outside of the git repo def get(self, generator: Generator, name: str) -> bytes: - return run( - nix_shell( - [f"nixpkgs#{self._store_backend}"], - [ - f"{self._store_backend}", - "show", - str(self.entry_dir(generator, name)), - ], - ), - ).stdout.encode() + pass_name = str(self.entry_dir(generator, name)) + return self._run_pass("show", pass_name).stdout.encode() def exists(self, generator: Generator, name: str) -> bool: extension = "age" if self._store_backend == "passage" else "gpg" filename = f"{self.entry_dir(generator, name)}.{extension}" - return (Path(self._password_store_dir) / filename).exists() + return (self._password_store_dir / filename).exists() + + def delete(self, generator: Generator, name: str) -> Iterable[Path]: + pass_name = str(self.entry_dir(generator, name)) + self._run_pass("rm", "--force", pass_name, options=RunOpts(check=True)) + return [] + + def delete_store(self) -> Iterable[Path]: + machine_pass_dir = Path(self.entry_prefix) / "per-machine" / self.machine.name + pass_call = ["rm", "--force", "--recursive", str(machine_pass_dir)] + self._run_pass(*pass_call, options=RunOpts(check=True)) + return [] def generate_hash(self) -> bytes: hashes = [] @@ -91,7 +91,7 @@ class SecretStore(StoreBase): [ "git", "-C", - self._password_store_dir, + str(self._password_store_dir), "log", "-1", "--format=%H", @@ -103,9 +103,9 @@ class SecretStore(StoreBase): .stdout.strip() .encode() ) - shared_dir = Path(self._password_store_dir) / self.entry_prefix / "shared" + shared_dir = self._password_store_dir / self.entry_prefix / "shared" machine_dir = ( - Path(self._password_store_dir) + self._password_store_dir / self.entry_prefix / "per-machine" / self.machine.name @@ -119,7 +119,7 @@ class SecretStore(StoreBase): [ "git", "-C", - self._password_store_dir, + str(self._password_store_dir), "log", "-1", "--format=%H", diff --git a/pkgs/clan-cli/clan_cli/vars/secret_modules/sops.py b/pkgs/clan-cli/clan_cli/vars/secret_modules/sops.py index f6dd1f437..3ef05ce2b 100644 --- a/pkgs/clan-cli/clan_cli/vars/secret_modules/sops.py +++ b/pkgs/clan-cli/clan_cli/vars/secret_modules/sops.py @@ -1,3 +1,5 @@ +import shutil +from collections.abc import Iterable from dataclasses import dataclass from pathlib import Path from tempfile import TemporaryDirectory @@ -162,6 +164,19 @@ class SecretStore(StoreBase): self.machine.flake_dir, self.secret_path(generator, name) ).encode("utf-8") + def delete(self, generator: "Generator", name: str) -> Iterable[Path]: + secret_dir = self.directory(generator, name) + shutil.rmtree(secret_dir) + return [secret_dir] + + def delete_store(self) -> Iterable[Path]: + flake_root = Path(self.machine.flake_dir) + store_folder = flake_root / "vars/per-machine" / self.machine.name + if not store_folder.exists(): + return [] + shutil.rmtree(store_folder) + return [store_folder] + def populate_dir(self, output_dir: Path, phases: list[str]) -> None: if "users" in phases or "services" in phases: key_name = f"{self.machine.name}-age.key" diff --git a/pkgs/clan-cli/clan_cli/vars/secret_modules/vm.py b/pkgs/clan-cli/clan_cli/vars/secret_modules/vm.py index 2fefcce5f..d8c1aa59d 100644 --- a/pkgs/clan-cli/clan_cli/vars/secret_modules/vm.py +++ b/pkgs/clan-cli/clan_cli/vars/secret_modules/vm.py @@ -1,4 +1,5 @@ import shutil +from collections.abc import Iterable from pathlib import Path from clan_cli.dirs import vm_state_dir @@ -39,6 +40,21 @@ class SecretStore(StoreBase): secret_file = self.dir / generator.name / name return secret_file.read_bytes() + def delete(self, generator: Generator, name: str) -> Iterable[Path]: + secret_dir = self.dir / generator.name + secret_file = secret_dir / name + secret_file.unlink() + empty = None + if next(secret_dir.iterdir(), empty) is empty: + secret_dir.rmdir() + return [secret_file] + + def delete_store(self) -> Iterable[Path]: + if not self.dir.exists(): + return [] + shutil.rmtree(self.dir) + return [self.dir] + def populate_dir(self, output_dir: Path, phases: list[str]) -> None: if output_dir.exists(): shutil.rmtree(output_dir)