Merge pull request 'vars/password-store: use standard paths; fix share flag' (#2061) from DavHau/clan-core:DavHau-dave into main

This commit is contained in:
clan-bot
2024-09-09 19:41:12 +00:00
4 changed files with 32 additions and 19 deletions

View File

@@ -103,16 +103,19 @@ class StoreBase(ABC):
def is_secret_store(self) -> bool: def is_secret_store(self) -> bool:
pass pass
def rel_dir(self, generator_name: str, var_name: str, shared: bool = False) -> Path:
if shared:
return Path(f"shared/{generator_name}/{var_name}")
return Path(f"per-machine/{self.machine.name}/{generator_name}/{var_name}")
def directory( def directory(
self, generator_name: str, var_name: str, shared: bool = False self, generator_name: str, var_name: str, shared: bool = False
) -> Path: ) -> Path:
if shared: return (
base_path = self.machine.flake_dir / "vars" / "shared" Path(self.machine.flake_dir)
else: / "vars"
base_path = ( / self.rel_dir(generator_name, var_name, shared)
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: def exists(self, generator_name: str, name: str, shared: bool = False) -> bool:
directory = self.directory(generator_name, name, shared) directory = self.directory(generator_name, name, shared)

View File

@@ -21,14 +21,17 @@ def check_vars(machine: Machine, generator_name: None | str = None) -> bool:
else: else:
generators = list(machine.vars_generators.keys()) generators = list(machine.vars_generators.keys())
for generator_name in generators: for generator_name in generators:
shared = machine.vars_generators[generator_name]["share"]
for name, file in machine.vars_generators[generator_name]["files"].items(): for name, file in machine.vars_generators[generator_name]["files"].items():
if file["secret"] and not secret_vars_store.exists(generator_name, name): if file["secret"] and not secret_vars_store.exists(
generator_name, name, shared=shared
):
log.info( log.info(
f"Secret fact '{name}' for service '{generator_name}' in machine {machine.name} is missing." f"Secret fact '{name}' for service '{generator_name}' in machine {machine.name} is missing."
) )
missing_secret_vars.append((generator_name, name)) missing_secret_vars.append((generator_name, name))
if not file["secret"] and not public_vars_store.exists( if not file["secret"] and not public_vars_store.exists(
generator_name, name generator_name, name, shared=shared
): ):
log.info( log.info(
f"Public fact '{name}' for service '{generator_name}' in machine {machine.name} is missing." f"Public fact '{name}' for service '{generator_name}' in machine {machine.name} is missing."

View File

@@ -23,11 +23,6 @@ class SecretStore(SecretStoreBase):
"PASSWORD_STORE_DIR", f"{os.environ['HOME']}/.password-store" "PASSWORD_STORE_DIR", f"{os.environ['HOME']}/.password-store"
) )
def _var_path(self, generator_name: str, name: str, shared: bool) -> Path:
if shared:
return Path(f"shared/{generator_name}/{name}")
return Path(f"machines/{self.machine.name}/{generator_name}/{name}")
def _set( def _set(
self, self,
generator_name: str, generator_name: str,
@@ -43,7 +38,7 @@ class SecretStore(SecretStoreBase):
"pass", "pass",
"insert", "insert",
"-m", "-m",
str(self._var_path(generator_name, name, shared)), str(self.rel_dir(generator_name, name, shared)),
], ],
), ),
input=value, input=value,
@@ -58,7 +53,7 @@ class SecretStore(SecretStoreBase):
[ [
"pass", "pass",
"show", "show",
str(self._var_path(generator_name, name, shared)), str(self.rel_dir(generator_name, name, shared)),
], ],
), ),
check=True, check=True,
@@ -70,7 +65,7 @@ class SecretStore(SecretStoreBase):
return False return False
return ( return (
Path(self._password_store_dir) Path(self._password_store_dir)
/ f"{self._var_path(generator_name, name, shared)}.gpg" / f"{self.rel_dir(generator_name, name, shared)}.gpg"
).exists() ).exists()
def generate_hash(self) -> bytes: def generate_hash(self) -> bytes:

View File

@@ -181,6 +181,12 @@ def test_generate_secret_var_password_store(
my_generator = config["clan"]["core"]["vars"]["generators"]["my_generator"] my_generator = config["clan"]["core"]["vars"]["generators"]["my_generator"]
my_generator["files"]["my_secret"]["secret"] = True my_generator["files"]["my_secret"]["secret"] = True
my_generator["script"] = "echo hello > $out/my_secret" my_generator["script"] = "echo hello > $out/my_secret"
my_shared_generator = config["clan"]["core"]["vars"]["generators"][
"my_shared_generator"
]
my_shared_generator["share"] = True
my_shared_generator["files"]["my_shared_secret"]["secret"] = True
my_shared_generator["script"] = "echo hello > $out/my_shared_secret"
flake = generate_flake( flake = generate_flake(
temporary_home, temporary_home,
flake_template=CLAN_CORE / "templates" / "minimal", flake_template=CLAN_CORE / "templates" / "minimal",
@@ -219,8 +225,11 @@ def test_generate_secret_var_password_store(
store = password_store.SecretStore( store = password_store.SecretStore(
Machine(name="my_machine", flake=FlakeId(str(flake.path))) Machine(name="my_machine", flake=FlakeId(str(flake.path)))
) )
assert store.exists("my_generator", "my_secret") assert store.exists("my_generator", "my_secret", shared=False)
assert store.get("my_generator", "my_secret").decode() == "hello\n" assert not store.exists("my_generator", "my_secret", shared=True)
assert store.exists("my_shared_generator", "my_shared_secret", shared=True)
assert not store.exists("my_shared_generator", "my_shared_secret", shared=False)
assert store.get("my_generator", "my_secret", shared=False).decode() == "hello\n"
vars_text = stringify_all_vars(machine) vars_text = stringify_all_vars(machine)
assert "my_generator/my_secret" in vars_text assert "my_generator/my_secret" in vars_text
@@ -382,7 +391,10 @@ def test_share_flag(
) )
monkeypatch.chdir(flake.path) monkeypatch.chdir(flake.path)
sops_setup.init() sops_setup.init()
machine = Machine(name="my_machine", flake=FlakeId(str(flake.path)))
assert not check_vars(machine)
cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"]) cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"])
assert check_vars(machine)
sops_store = sops.SecretStore( sops_store = sops.SecretStore(
Machine(name="my_machine", flake=FlakeId(str(flake.path))) Machine(name="my_machine", flake=FlakeId(str(flake.path)))
) )