yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Refactor StoreBase to take machine name string instead of Machine object

Browse Source 
- Updated StoreBase.__init__ to accept machine: str and flake: Flake
- Modified all StoreBase subclasses (in_repo, vm, fs, sops, password_store) to match new signature
- Added select_machine method to Flake class for machine-specific attribute selection
- Updated Machine.select to use the new Flake.select_machine method
- Fixed all test cases to pass machine name and flake to store constructors
- Maintained backward compatibility by keeping the same external API

This reduces coupling between the store system and the Machine class,
making the architecture more modular and flexible.
This commit is contained in:
DavHau
2025-07-07 17:11:55 +07:00
parent 979d5dcdd1
commit ed0b86385b
10 changed files with 137 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
pkgs/clan-cli/clan_cli/tests/test_vars.py
View File

@@ -185,12 +185,10 @@ def test_generate_public_and_secret_vars(
)
vars_text = stringify_all_vars(machine)
in_repo_store = in_repo.FactStore(
Machine(name="my_machine", flake=Flake(str(flake.path)))
machine="my_machine", flake=Flake(str(flake.path))
)
assert not in_repo_store.exists(Generator("my_generator"), "my_secret")
sops_store = sops.SecretStore(
Machine(name="my_machine", flake=Flake(str(flake.path)))
)
sops_store = sops.SecretStore(machine="my_machine", flake=Flake(str(flake.path)))
assert sops_store.exists(Generator("my_generator"), "my_secret")
assert sops_store.get(Generator("my_generator"), "my_secret").decode() == "secret"
assert sops_store.exists(Generator("dependent_generator"), "my_secret")
@@ -265,12 +263,10 @@ def test_generate_secret_var_sops_with_default_group(
cli.run(["secrets", "groups", "add-user", "my_group", sops_setup.user])
cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"])
in_repo_store = in_repo.FactStore(
Machine(name="my_machine", flake=Flake(str(flake.path)))
machine="my_machine", flake=Flake(str(flake.path))
)
assert not in_repo_store.exists(Generator("first_generator"), "my_secret")
sops_store = sops.SecretStore(
Machine(name="my_machine", flake=Flake(str(flake.path)))
)
sops_store = sops.SecretStore(machine="my_machine", flake=Flake(str(flake.path)))
assert sops_store.exists(Generator("first_generator"), "my_secret")
assert (
sops_store.get(Generator("first_generator"), "my_secret").decode() == "hello\n"
@@ -355,8 +351,8 @@ def test_generated_shared_secret_sops(
cli.run(["vars", "generate", "--flake", str(flake.path), "machine2"])
assert check_vars(machine2.name, machine2.flake)
assert check_vars(machine2.name, machine2.flake)
m1_sops_store = sops.SecretStore(machine1)
m2_sops_store = sops.SecretStore(machine2)
m1_sops_store = sops.SecretStore(machine1.name, machine1.flake)
m2_sops_store = sops.SecretStore(machine2.name, machine2.flake)
assert m1_sops_store.exists(
Generator("my_shared_generator", share=True), "my_shared_secret"
)
@@ -421,7 +417,7 @@ def test_generate_secret_var_password_store(
cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"])
assert check_vars(machine.name, machine.flake)
store = password_store.SecretStore(
Machine(name="my_machine", flake=Flake(str(flake.path)))
machine="my_machine", flake=Flake(str(flake.path))
)
assert store.exists(Generator("my_generator", share=False, files=[]), "my_secret")
assert not store.exists(
@@ -496,12 +492,8 @@ def test_generate_secret_for_multiple_machines(
monkeypatch.chdir(flake.path)
cli.run(["vars", "generate", "--flake", str(flake.path)])
# check if public vars have been created correctly
in_repo_store1 = in_repo.FactStore(
Machine(name="machine1", flake=Flake(str(flake.path)))
)
in_repo_store2 = in_repo.FactStore(
Machine(name="machine2", flake=Flake(str(flake.path)))
)
in_repo_store1 = in_repo.FactStore(machine="machine1", flake=Flake(str(flake.path)))
in_repo_store2 = in_repo.FactStore(machine="machine2", flake=Flake(str(flake.path)))
assert in_repo_store1.exists(Generator("my_generator"), "my_value")
assert in_repo_store2.exists(Generator("my_generator"), "my_value")
assert (
@@ -513,12 +505,8 @@ def test_generate_secret_for_multiple_machines(
== "machine2\n"
)
# check if secret vars have been created correctly
sops_store1 = sops.SecretStore(
Machine(name="machine1", flake=Flake(str(flake.path)))
)
sops_store2 = sops.SecretStore(
Machine(name="machine2", flake=Flake(str(flake.path)))
)
sops_store1 = sops.SecretStore(machine="machine1", flake=Flake(str(flake.path)))
sops_store2 = sops.SecretStore(machine="machine2", flake=Flake(str(flake.path)))
assert sops_store1.exists(Generator("my_generator"), "my_secret")
assert sops_store2.exists(Generator("my_generator"), "my_secret")
assert (
@@ -563,7 +551,7 @@ def test_prompt(
)
cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"])
in_repo_store = in_repo.FactStore(
Machine(name="my_machine", flake=Flake(str(flake.path)))
machine="my_machine", flake=Flake(str(flake.path))
)
assert in_repo_store.exists(Generator("my_generator"), "line_value")
assert (
@@ -576,9 +564,7 @@ def test_prompt(
in_repo_store.get(Generator("my_generator"), "multiline_value").decode()
== "my\nmultiline\ninput\n"
)
sops_store = sops.SecretStore(
Machine(name="my_machine", flake=Flake(str(flake.path)))
)
sops_store = sops.SecretStore(machine="my_machine", flake=Flake(str(flake.path)))
assert sops_store.exists(
Generator(name="my_generator", share=False, files=[]), "prompt_persist"
)
@@ -620,10 +606,10 @@ def test_multi_machine_shared_vars(
monkeypatch.chdir(flake.path)
machine1 = Machine(name="machine1", flake=Flake(str(flake.path)))
machine2 = Machine(name="machine2", flake=Flake(str(flake.path)))
sops_store_1 = sops.SecretStore(machine1)
sops_store_2 = sops.SecretStore(machine2)
in_repo_store_1 = in_repo.FactStore(machine1)
in_repo_store_2 = in_repo.FactStore(machine2)
sops_store_1 = sops.SecretStore(machine1.name, machine1.flake)
sops_store_2 = sops.SecretStore(machine2.name, machine2.flake)
in_repo_store_1 = in_repo.FactStore(machine1.name, machine1.flake)
in_repo_store_2 = in_repo.FactStore(machine2.name, machine2.flake)
generator = Generator("shared_generator", share=True)
# generate for machine 1
cli.run(["vars", "generate", "--flake", str(flake.path), "machine1"])
@@ -679,7 +665,7 @@ def test_api_set_prompts(
},
)
machine = Machine(name="my_machine", flake=Flake(str(flake.path)))
store = in_repo.FactStore(machine)
store = in_repo.FactStore(machine.name, machine.flake)
assert store.exists(Generator("my_generator"), "prompt1")
assert store.get(Generator("my_generator"), "prompt1").decode() == "input1"
create_machine_vars(
@@ -830,11 +816,9 @@ def test_migration(
assert "Migrated var my_generator/my_value" in caplog.text
assert "Migrated secret var my_generator/my_secret" in caplog.text
in_repo_store = in_repo.FactStore(
Machine(name="my_machine", flake=Flake(str(flake.path)))
)
sops_store = sops.SecretStore(
Machine(name="my_machine", flake=Flake(str(flake.path)))
machine="my_machine", flake=Flake(str(flake.path))
)
sops_store = sops.SecretStore(machine="my_machine", flake=Flake(str(flake.path)))
assert in_repo_store.exists(Generator("my_generator"), "my_value")
assert in_repo_store.get(Generator("my_generator"), "my_value").decode() == "hello"
assert sops_store.exists(Generator("my_generator"), "my_secret")

9
pkgs/clan-cli/clan_cli/vars/_types.py
View File

@@ -6,7 +6,7 @@ from pathlib import Path
from typing import TYPE_CHECKING
from clan_lib.errors import ClanError
from clan_lib.machines import machines
from clan_lib.flake import Flake
from clan_lib.ssh.remote import Remote
if TYPE_CHECKING:
@@ -29,8 +29,9 @@ class GeneratorUpdate:
class StoreBase(ABC):
def __init__(self, machine: "machines.Machine") -> None:
def __init__(self, machine: str, flake: Flake) -> None:
self.machine = machine
self.flake = flake
@property
@abstractmethod
@@ -86,10 +87,10 @@ class StoreBase(ABC):
def rel_dir(self, generator: "Generator", var_name: str) -> Path:
if generator.share:
return Path("shared") / generator.name / var_name
return Path("per-machine") / self.machine.name / generator.name / var_name
return Path("per-machine") / self.machine / generator.name / var_name
def directory(self, generator: "Generator", var_name: str) -> Path:
return Path(self.machine.flake_dir) / "vars" / self.rel_dir(generator, var_name)
return self.flake.path / "vars" / self.rel_dir(generator, var_name)
def set(
self,

14
pkgs/clan-cli/clan_cli/vars/public_modules/in_repo.py
View File

@@ -5,7 +5,7 @@ from pathlib import Path
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_lib.errors import ClanError
from clan_lib.machines.machines import Machine
from clan_lib.flake import Flake
from clan_lib.ssh.remote import Remote
@@ -14,8 +14,8 @@ class FactStore(StoreBase):
def is_secret_store(self) -> bool:
return False
def __init__(self, machine: Machine) -> None:
self.machine = machine
def __init__(self, machine: str, flake: Flake) -> None:
super().__init__(machine, flake)
self.works_remotely = False
@property
@@ -28,8 +28,8 @@ class FactStore(StoreBase):
var: Var,
value: bytes,
) -> Path | None:
if not self.machine.flake.is_local:
msg = f"in_flake fact storage is only supported for local flakes: {self.machine.flake}"
if not self.flake.is_local:
msg = f"in_flake fact storage is only supported for local flakes: {self.flake}"
raise ClanError(msg)
folder = self.directory(generator, var.name)
file_path = folder / "value"
@@ -62,8 +62,8 @@ class FactStore(StoreBase):
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
flake_root = self.flake.path
store_folder = flake_root / "vars/per-machine" / self.machine
if not store_folder.exists():
return []
shutil.rmtree(store_folder)

13
pkgs/clan-cli/clan_cli/vars/public_modules/vm.py
View File

@@ -7,7 +7,7 @@ from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_lib.dirs import vm_state_dir
from clan_lib.errors import ClanError
from clan_lib.machines.machines import Machine
from clan_lib.flake import Flake
from clan_lib.ssh.remote import Remote
log = logging.getLogger(__name__)
@@ -18,11 +18,14 @@ class FactStore(StoreBase):
def is_secret_store(self) -> bool:
return False
def __init__(self, machine: Machine) -> None:
self.machine = machine
def __init__(self, machine: str, flake: Flake) -> None:
super().__init__(machine, flake)
self.works_remotely = False
self.dir = vm_state_dir(machine.flake.identifier, machine.name) / "facts"
machine.debug(f"FactStore initialized with dir {self.dir}")
self.dir = vm_state_dir(flake.identifier, machine) / "facts"
log.debug(
f"FactStore initialized with dir {self.dir}",
extra={"command_prefix": machine},
)
@property
def store_name(self) -> str:

17
pkgs/clan-cli/clan_cli/vars/secret_modules/fs.py
View File

@@ -4,7 +4,7 @@ from pathlib import Path
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_lib.machines.machines import Machine
from clan_lib.flake import Flake
from clan_lib.ssh.remote import Remote
@@ -13,8 +13,8 @@ class SecretStore(StoreBase):
def is_secret_store(self) -> bool:
return True
def __init__(self, machine: Machine) -> None:
self.machine = machine
def __init__(self, machine: str, flake: Flake) -> None:
super().__init__(machine, flake)
self.dir = Path(tempfile.gettempdir()) / "clan_secrets"
self.dir.mkdir(parents=True, exist_ok=True)
@@ -46,6 +46,17 @@ class SecretStore(StoreBase):
shutil.copytree(self.dir, output_dir)
shutil.rmtree(self.dir)
def delete(self, generator: Generator, name: str) -> list[Path]:
secret_file = self.dir / generator.name / name
if secret_file.exists():
secret_file.unlink()
return []
def delete_store(self) -> list[Path]:
if self.dir.exists():
shutil.rmtree(self.dir)
return []
def upload(self, host: Remote, phases: list[str]) -> None:
msg = "Cannot upload secrets with FS backend"
raise NotImplementedError(msg)

31
pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py
View File

@@ -9,7 +9,7 @@ from tempfile import TemporaryDirectory
from clan_cli.ssh.upload import upload
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_lib.machines.machines import Machine
from clan_lib.flake import Flake
from clan_lib.ssh.remote import Remote
log = logging.getLogger(__name__)
@@ -20,8 +20,8 @@ class SecretStore(StoreBase):
def is_secret_store(self) -> bool:
return True
def __init__(self, machine: Machine) -> None:
self.machine = machine
def __init__(self, machine: str, flake: Flake) -> None:
super().__init__(machine, flake)
self.entry_prefix = "clan-vars"
self._store_dir: Path | None = None
@@ -42,12 +42,13 @@ class SecretStore(StoreBase):
@property
def _pass_command(self) -> str:
out_path = self.machine.select(
"config.clan.core.vars.password-store.passPackage.outPath"
out_path = self.flake.select_machine(
self.machine, "config.clan.core.vars.password-store.passPackage.outPath"
)
main_program = (
self.machine.select(
"config.clan.core.vars.password-store.passPackage.?meta.?mainProgram"
self.flake.select_machine(
self.machine,
"config.clan.core.vars.password-store.passPackage.?meta.?mainProgram",
)
.get("meta", {})
.get("mainProgram")
@@ -119,7 +120,7 @@ class SecretStore(StoreBase):
return []
def delete_store(self) -> Iterable[Path]:
machine_dir = Path(self.entry_prefix) / "per-machine" / self.machine.name
machine_dir = Path(self.entry_prefix) / "per-machine" / self.machine
# Check if the directory exists in the password store before trying to delete
result = self._run_pass("ls", str(machine_dir), check=False)
if result.returncode == 0:
@@ -138,9 +139,7 @@ class SecretStore(StoreBase):
from clan_cli.vars.generate import Generator
manifest = []
generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake
)
generators = Generator.generators_from_flake(self.machine, self.flake)
for generator in generators:
for file in generator.files:
manifest.append(f"{generator.name}/{file.name}".encode())
@@ -158,7 +157,7 @@ class SecretStore(StoreBase):
remote_hash = host.run(
[
"cat",
f"{self.machine.select('config.clan.core.vars.password-store.secretLocation')}/.pass_info",
f"{self.flake.select_machine(self.machine, 'config.clan.core.vars.password-store.secretLocation')}/.pass_info",
],
RunOpts(log=Log.STDERR, check=False),
).stdout.strip()
@@ -171,9 +170,7 @@ class SecretStore(StoreBase):
def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
from clan_cli.vars.generate import Generator
vars_generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake
)
vars_generators = Generator.generators_from_flake(self.machine, self.flake)
if "users" in phases:
with tarfile.open(
output_dir / "secrets_for_users.tar.gz", "w:gz"
@@ -247,8 +244,8 @@ class SecretStore(StoreBase):
pass_dir = Path(_tempdir).resolve()
self.populate_dir(pass_dir, phases)
upload_dir = Path(
self.machine.select(
"config.clan.core.vars.password-store.secretLocation"
self.flake.select_machine(
self.machine, "config.clan.core.vars.password-store.secretLocation"
)
)
upload(host, pass_dir, upload_dir)

91
pkgs/clan-cli/clan_cli/vars/secret_modules/sops.py
View File

@@ -27,7 +27,7 @@ from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator
from clan_cli.vars.var import Var
from clan_lib.errors import ClanError
from clan_lib.machines.machines import Machine
from clan_lib.flake import Flake
from clan_lib.ssh.remote import Remote
@@ -48,15 +48,13 @@ class SecretStore(StoreBase):
def is_secret_store(self) -> bool:
return True
def __init__(self, machine: Machine) -> None:
self.machine = machine
def __init__(self, machine: str, flake: Flake) -> None:
super().__init__(machine, flake)
# no need to generate keys if we don't manage secrets
from clan_cli.vars.generate import Generator
vars_generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake
)
vars_generators = Generator.generators_from_flake(self.machine, self.flake)
if not vars_generators:
return
has_secrets = False
@@ -67,18 +65,19 @@ class SecretStore(StoreBase):
if not has_secrets:
return
if has_machine(self.machine.flake_dir, self.machine.name):
if has_machine(self.flake.path, self.machine):
return
priv_key, pub_key = sops.generate_private_key()
encrypt_secret(
self.machine.flake_dir,
sops_secrets_folder(self.machine.flake_dir)
/ f"{self.machine.name}-age.key",
self.flake.path,
sops_secrets_folder(self.flake.path) / f"{self.machine}-age.key",
priv_key,
add_groups=self.machine.select("config.clan.core.sops.defaultGroups"),
age_plugins=load_age_plugins(self.machine.flake),
add_groups=self.flake.select_machine(
self.machine, "config.clan.core.sops.defaultGroups"
),
age_plugins=load_age_plugins(self.flake),
)
add_machine(self.machine.flake_dir, self.machine.name, pub_key, False)
add_machine(self.flake.path, self.machine, pub_key, False)
@property
def store_name(self) -> str:
@@ -87,11 +86,11 @@ class SecretStore(StoreBase):
def user_has_access(
self, user: str, generator: Generator, secret_name: str
) -> bool:
key_dir = sops_users_folder(self.machine.flake_dir) / user
key_dir = sops_users_folder(self.flake.path) / user
return self.key_has_access(key_dir, generator, secret_name)
def machine_has_access(self, generator: Generator, secret_name: str) -> bool:
key_dir = sops_machines_folder(self.machine.flake_dir) / self.machine.name
key_dir = sops_machines_folder(self.flake.path) / self.machine
return self.key_has_access(key_dir, generator, secret_name)
def key_has_access(
@@ -117,9 +116,7 @@ class SecretStore(StoreBase):
if generator is None:
from clan_cli.vars.generate import Generator
generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake
)
generators = Generator.generators_from_flake(self.machine, self.flake)
else:
generators = [generator]
file_found = False
@@ -144,7 +141,7 @@ class SecretStore(StoreBase):
if outdated:
msg = (
"The local state of some secret vars is inconsistent and needs to be updated.\n"
f"Run 'clan vars fix {self.machine.name}' to apply the necessary changes."
f"Run 'clan vars fix {self.machine}' to apply the necessary changes."
"Problems to fix:\n"
"\n".join(o[2] for o in outdated if o[2])
)
@@ -162,20 +159,22 @@ class SecretStore(StoreBase):
secret_folder.mkdir(parents=True, exist_ok=True)
# initialize the secret
encrypt_secret(
self.machine.flake_dir,
self.flake.path,
secret_folder,
value,
add_machines=[self.machine.name] if var.deploy else [],
add_groups=self.machine.select("config.clan.core.sops.defaultGroups"),
add_machines=[self.machine] if var.deploy else [],
add_groups=self.flake.select_machine(
self.machine, "config.clan.core.sops.defaultGroups"
),
git_commit=False,
age_plugins=load_age_plugins(self.machine.flake),
age_plugins=load_age_plugins(self.flake),
)
return secret_folder
def get(self, generator: Generator, name: str) -> bytes:
return decrypt_secret(
self.secret_path(generator, name),
age_plugins=load_age_plugins(self.machine.flake),
age_plugins=load_age_plugins(self.flake),
).encode("utf-8")
def delete(self, generator: "Generator", name: str) -> Iterable[Path]:
@@ -184,8 +183,8 @@ class SecretStore(StoreBase):
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
flake_root = self.flake.path
store_folder = flake_root / "vars/per-machine" / self.machine
if not store_folder.exists():
return []
shutil.rmtree(store_folder)
@@ -194,17 +193,15 @@ class SecretStore(StoreBase):
def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
from clan_cli.vars.generate import Generator
vars_generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake
)
vars_generators = Generator.generators_from_flake(self.machine, self.flake)
if "users" in phases or "services" in phases:
key_name = f"{self.machine.name}-age.key"
if not has_secret(sops_secrets_folder(self.machine.flake_dir) / key_name):
key_name = f"{self.machine}-age.key"
if not has_secret(sops_secrets_folder(self.flake.path) / key_name):
# skip uploading the secret, not managed by us
return
key = decrypt_secret(
sops_secrets_folder(self.machine.flake_dir) / key_name,
age_plugins=load_age_plugins(self.machine.flake),
sops_secrets_folder(self.flake.path) / key_name,
age_plugins=load_age_plugins(self.flake),
)
(output_dir / "key.txt").touch(mode=0o600)
(output_dir / "key.txt").write_text(key)
@@ -258,10 +255,10 @@ class SecretStore(StoreBase):
return
secret_folder = self.secret_path(generator, name)
add_secret(
self.machine.flake_dir,
self.machine.name,
self.flake.path,
self.machine,
secret_folder,
age_plugins=load_age_plugins(self.machine.flake),
age_plugins=load_age_plugins(self.flake),
)
def collect_keys_for_secret(self, path: Path) -> set[sops.SopsKey]:
@@ -271,15 +268,17 @@ class SecretStore(StoreBase):
)
keys = collect_keys_for_path(path)
for group in self.machine.select("config.clan.core.sops.defaultGroups"):
for group in self.flake.select_machine(
self.machine, "config.clan.core.sops.defaultGroups"
):
keys.update(
collect_keys_for_type(
self.machine.flake_dir / "sops" / "groups" / group / "machines"
self.flake.path / "sops" / "groups" / group / "machines"
)
)
keys.update(
collect_keys_for_type(
self.machine.flake_dir / "sops" / "groups" / group / "users"
self.flake.path / "sops" / "groups" / group / "users"
)
)
@@ -296,7 +295,7 @@ class SecretStore(StoreBase):
f"One or more recipient keys were added to secret{' shared' if generator.share else ''} var '{var_id}', but it was never re-encrypted.\n"
f"This could have been a malicious actor trying to add their keys, please investigate.\n"
f"Added keys: {', '.join(f'{r.key_type.name}:{r.pubkey}' for r in recipients_to_add)}\n"
f"If this is intended, run 'clan vars fix {self.machine.name}' to re-encrypt the secret."
f"If this is intended, run 'clan vars fix {self.machine}' to re-encrypt the secret."
)
return needs_update, msg
@@ -309,9 +308,7 @@ class SecretStore(StoreBase):
if generator is None:
from clan_cli.vars.generate import Generator
generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake
)
generators = Generator.generators_from_flake(self.machine, self.flake)
else:
generators = [generator]
file_found = False
@@ -328,12 +325,14 @@ class SecretStore(StoreBase):
secret_path = self.secret_path(generator, file.name)
age_plugins = load_age_plugins(self.machine.flake)
age_plugins = load_age_plugins(self.flake)
for group in self.machine.select("config.clan.core.sops.defaultGroups"):
for group in self.flake.select_machine(
self.machine, "config.clan.core.sops.defaultGroups"
):
allow_member(
groups_folder(secret_path),
sops_groups_folder(self.machine.flake_dir),
sops_groups_folder(self.flake.path),
group,
# we just want to create missing symlinks, we call update_keys below:
do_update_keys=False,

8
pkgs/clan-cli/clan_cli/vars/secret_modules/vm.py
View File

@@ -5,7 +5,7 @@ from pathlib import Path
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_lib.dirs import vm_state_dir
from clan_lib.machines.machines import Machine
from clan_lib.flake import Flake
from clan_lib.ssh.remote import Remote
@@ -14,9 +14,9 @@ class SecretStore(StoreBase):
def is_secret_store(self) -> bool:
return True
def __init__(self, machine: Machine) -> None:
self.machine = machine
self.dir = vm_state_dir(machine.flake.identifier, machine.name) / "secrets"
def __init__(self, machine: str, flake: Flake) -> None:
super().__init__(machine, flake)
self.dir = vm_state_dir(flake.identifier, machine) / "secrets"
self.dir.mkdir(parents=True, exist_ok=True)
@property

17
pkgs/clan-cli/clan_lib/flake/flake.py
View File

@@ -906,3 +906,20 @@ nix repl --expr 'rec {{
self.get_from_nix([selector])
value = self._cache.select(selector)
return value
def select_machine(self, machine_name: str, selector: str) -> Any:
"""
Select a nix attribute for a specific machine.
Args:
machine_name: The name of the machine
selector: The attribute selector string relative to the machine config
apply: Optional function to apply to the result
"""
from clan_lib.nix import nix_config
config = nix_config()
system = config["system"]
full_selector = f'clanInternals.machines."{system}"."{machine_name}".{selector}'
return self.select(full_selector)

13
pkgs/clan-cli/clan_lib/machines/machines.py
View File

@@ -13,7 +13,6 @@ from clan_cli.vars._types import StoreBase
from clan_lib.api import API
from clan_lib.errors import ClanCmdError, ClanError
from clan_lib.flake import Flake
from clan_lib.nix import nix_config
from clan_lib.nix_models.clan import InventoryMachine
from clan_lib.ssh.remote import Remote
@@ -105,13 +104,13 @@ class Machine:
def secret_vars_store(self) -> StoreBase:
secret_module = self.select("config.clan.core.vars.settings.secretModule")
module = importlib.import_module(secret_module)
return module.SecretStore(machine=self)
return module.SecretStore(machine=self.name, flake=self.flake)
@cached_property
def public_vars_store(self) -> StoreBase:
public_module = self.select("config.clan.core.vars.settings.publicModule")
module = importlib.import_module(public_module)
return module.FactStore(machine=self)
return module.FactStore(machine=self.name, flake=self.flake)
@property
def facts_data(self) -> dict[str, dict[str, Any]]:
@@ -160,13 +159,7 @@ class Machine:
Select a nix attribute of the machine
@attr: the attribute to get
"""
config = nix_config()
system = config["system"]
return self.flake.select(
f'clanInternals.machines."{system}"."{self.name}".{attr}'
)
return self.flake.select_machine(self.name, attr)
@dataclass(frozen=True)