use machine.{secrets,public}_{vars,fact}_store everywhere

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

@@ -1,5 +1,4 @@
 import argparse
-import importlib
 import logging
 
 from clan_cli.completions import add_dynamic_completer, complete_machines
@@ -9,11 +8,6 @@ log = logging.getLogger(__name__)
 
 
 def check_secrets(machine: Machine, service: None | str = None) -> bool:
-    secret_facts_module = importlib.import_module(machine.secret_facts_module)
-    secret_facts_store = secret_facts_module.SecretStore(machine=machine)
-    public_facts_module = importlib.import_module(machine.public_facts_module)
-    public_facts_store = public_facts_module.FactStore(machine=machine)
-
     missing_secret_facts = []
     missing_public_facts = []
     services = [service] if service else list(machine.facts_data.keys())
@@ -23,14 +17,14 @@ def check_secrets(machine: Machine, service: None | str = None) -> bool:
                 secret_name = secret_fact
             else:
                 secret_name = secret_fact["name"]
-            if not secret_facts_store.exists(service, secret_name):
+            if not machine.secret_facts_store.exists(service, secret_name):
                 machine.info(
                     f"Secret fact '{secret_fact}' for service '{service}' is missing."
                 )
                 missing_secret_facts.append((service, secret_name))
 
         for public_fact in machine.facts_data[service]["public"]:
-            if not public_facts_store.exists(service, public_fact):
+            if not machine.public_facts_store.exists(service, public_fact):
                 machine.info(
                     f"Public fact '{public_fact}' for service '{service}' is missing."
                 )

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

@@ -1,5 +1,4 @@
 import argparse
-import importlib
 import logging
 import os
 import sys
@@ -161,11 +160,6 @@ def _generate_facts_for_machine(
 ) -> bool:
     local_temp = tmpdir / machine.name
     local_temp.mkdir()
-    secret_facts_module = importlib.import_module(machine.secret_facts_module)
-    secret_facts_store = secret_facts_module.SecretStore(machine=machine)
-
-    public_facts_module = importlib.import_module(machine.public_facts_module)
-    public_facts_store = public_facts_module.FactStore(machine=machine)
 
     machine_updated = False
 
@@ -184,8 +178,8 @@ def _generate_facts_for_machine(
             machine=machine,
             service=service,
             regenerate=regenerate,
-            secret_facts_store=secret_facts_store,
-            public_facts_store=public_facts_store,
+            secret_facts_store=machine.secret_facts_store,
+            public_facts_store=machine.public_facts_store,
             tmpdir=local_temp,
             prompt=prompt,
         )

pkgs/clan-cli/clan_cli/facts/list.py

@@ -1,8 +1,6 @@
 import argparse
-import importlib
 import json
 import logging
-from typing import Any
 
 from clan_cli.completions import add_dynamic_completer, complete_machines
 from clan_cli.machines.machines import Machine
@@ -10,24 +8,11 @@ from clan_cli.machines.machines import Machine
 log = logging.getLogger(__name__)
 
 
-# TODO get also secret facts
-def get_all_facts(machine: Machine) -> dict:
-    public_facts_store = get_public_facts_store(machine)
-
-    return public_facts_store.get_all()
-
-
-def get_public_facts_store(machine: Machine) -> Any:
-    public_facts_module = importlib.import_module(machine.public_facts_module)
-    public_facts_store = public_facts_module.FactStore(machine=machine)
-    return public_facts_store
-
-
 def get_command(args: argparse.Namespace) -> None:
     machine = Machine(name=args.machine, flake=args.flake)
 
     # the raw_facts are bytestrings making them not json serializable
-    raw_facts = get_all_facts(machine)
+    raw_facts = machine.public_facts_store.get_all()
     facts = {}
     for key in raw_facts["TODO"]:
         facts[key] = raw_facts["TODO"][key].decode("utf8")

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

@@ -1,5 +1,4 @@
 import argparse
-import importlib
 import logging
 from pathlib import Path
 from tempfile import TemporaryDirectory
@@ -12,16 +11,13 @@ log = logging.getLogger(__name__)
 
 
 def upload_secrets(machine: Machine) -> None:
-    secret_facts_module = importlib.import_module(machine.secret_facts_module)
-    secret_facts_store = secret_facts_module.SecretStore(machine=machine)
-
-    if not secret_facts_store.needs_upload():
+    if not machine.secret_facts_store.needs_upload():
         machine.info("Secrets already uploaded")
         return
 
     with TemporaryDirectory(prefix="facts-upload-") as _tempdir:
         local_secret_dir = Path(_tempdir).resolve()
-        secret_facts_store.upload(local_secret_dir)
+        machine.secret_facts_store.upload(local_secret_dir)
         remote_secret_dir = Path(machine.secrets_upload_directory)
 
         upload(machine.target_host, local_secret_dir, remote_secret_dir)

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

@@ -1,4 +1,3 @@
-import importlib
 import json
 import logging
 import os
@@ -11,10 +10,10 @@ from clan_cli.api import API
 from clan_cli.cmd import Log, RunOpts, cmd_with_root, run
 from clan_cli.errors import ClanError
 from clan_cli.facts.generate import generate_facts
-from clan_cli.facts.secret_modules import SecretStoreBase
 from clan_cli.machines.machines import Machine
 from clan_cli.nix import nix_shell
 from clan_cli.vars.generate import generate_vars
+from clan_cli.vars.upload import upload_secret_vars
 
 from .automount import pause_automounting
 from .list import list_possible_keymaps, list_possible_languages
@@ -96,10 +95,6 @@ def flash_machine(
                     msg = f"Partitioning time secrets are not supported with `clan flash write`: clan.core.vars.generators.{generator.name}.files.{file.name}"
                     raise ClanError(msg)
 
-        secret_facts_module = importlib.import_module(machine.secret_facts_module)
-        secret_facts_store: SecretStoreBase = secret_facts_module.SecretStore(
-            machine=machine
-        )
         with TemporaryDirectory(prefix="disko-install-") as _tmpdir:
             tmpdir = Path(_tmpdir)
             upload_dir = machine.secrets_upload_directory
@@ -110,7 +105,8 @@ def flash_machine(
                 local_dir = tmpdir / upload_dir
 
             local_dir.mkdir(parents=True)
-            secret_facts_store.upload(local_dir)
+            machine.secret_facts_store.upload(local_dir)
+            upload_secret_vars(machine, local_dir)
             disko_install = []
 
             if os.geteuid() != 0:

pkgs/clan-cli/clan_cli/machines/delete.py

@@ -14,12 +14,6 @@ from clan_cli.secrets.machines import remove_machine as secrets_machine_remove
 from clan_cli.secrets.secrets import (
     list_secrets,
 )
-from clan_cli.vars.list import (
-    public_store as vars_public_store,
-)
-from clan_cli.vars.list import (
-    secret_store as vars_secret_store,
-)
 
 from .machines import Machine
 
@@ -55,8 +49,8 @@ def delete_machine(flake: Flake, name: str) -> None:
         shutil.rmtree(secret_path)
 
     machine = Machine(name, flake)
-    changed_paths.extend(vars_public_store(machine).delete_store())
-    changed_paths.extend(vars_secret_store(machine).delete_store())
+    changed_paths.extend(machine.public_vars_store.delete_store())
+    changed_paths.extend(machine.secret_vars_store.delete_store())
     # Remove the machine's key, and update secrets & vars that referenced it:
     if secrets_has_machine(flake.path, name):
         secrets_machine_remove(flake.path, name)

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

@@ -5,7 +5,7 @@ import re
 from dataclasses import dataclass, field
 from functools import cached_property
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any
 
 from clan_cli.cmd import Log, RunOpts, run_no_stdout
 from clan_cli.errors import ClanCmdError, ClanError
@@ -94,50 +94,24 @@ class Machine:
             )
         return val
 
-    @property
-    def secret_facts_module(
-        self,
-    ) -> Literal[
-        "clan_cli.facts.secret_modules.sops",
-        "clan_cli.facts.secret_modules.vm",
-        "clan_cli.facts.secret_modules.password_store",
-    ]:
-        return self.deployment["facts"]["secretModule"]
-
-    @property
-    def public_facts_module(
-        self,
-    ) -> Literal[
-        "clan_cli.facts.public_modules.in_repo", "clan_cli.facts.public_modules.vm"
-    ]:
-        return self.deployment["facts"]["publicModule"]
-
     @cached_property
     def secret_facts_store(self) -> facts_secret_modules.SecretStoreBase:
-        module = importlib.import_module(self.secret_facts_module)
+        module = importlib.import_module(self.deployment["facts"]["secretModule"])
         return module.SecretStore(machine=self)
 
     @cached_property
     def public_facts_store(self) -> facts_public_modules.FactStoreBase:
-        module = importlib.import_module(self.public_facts_module)
+        module = importlib.import_module(self.deployment["facts"]["publicModule"])
         return module.FactStore(machine=self)
 
-    @property
-    def secret_vars_module(self) -> str:
-        return self.deployment["vars"]["secretModule"]
-
-    @property
-    def public_vars_module(self) -> str:
-        return self.deployment["vars"]["publicModule"]
-
     @cached_property
     def secret_vars_store(self) -> StoreBase:
-        module = importlib.import_module(self.secret_vars_module)
+        module = importlib.import_module(self.deployment["vars"]["secretModule"])
         return module.SecretStore(machine=self)
 
     @cached_property
     def public_vars_store(self) -> StoreBase:
-        module = importlib.import_module(self.public_vars_module)
+        module = importlib.import_module(self.deployment["vars"]["publicModule"])
         return module.FactStore(machine=self)
 
     @property

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

@@ -1,11 +1,9 @@
 import argparse
-import importlib
 import logging
 
 from clan_cli.completions import add_dynamic_completer, complete_machines
 from clan_cli.errors import ClanError
 from clan_cli.machines.machines import Machine
-from clan_cli.vars._types import StoreBase
 
 log = logging.getLogger(__name__)
 
@@ -30,11 +28,6 @@ class VarStatus:
 
 
 def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatus:
-    secret_vars_module = importlib.import_module(machine.secret_vars_module)
-    secret_vars_store: StoreBase = secret_vars_module.SecretStore(machine=machine)
-    public_vars_module = importlib.import_module(machine.public_vars_module)
-    public_vars_store: StoreBase = public_vars_module.FactStore(machine=machine)
-
     missing_secret_vars = []
     missing_public_vars = []
     # signals if a var needs to be updated (eg. needs re-encryption due to new users added)
@@ -55,17 +48,19 @@ def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatu
     for generator in generators:
         generator.machine(machine)
         for file in generator.files:
-            file.store(secret_vars_store if file.secret else public_vars_store)
+            file.store(
+                machine.secret_vars_store if file.secret else machine.public_vars_store
+            )
             file.generator(generator)
 
             if file.secret:
-                if not secret_vars_store.exists(generator, file.name):
+                if not machine.secret_vars_store.exists(generator, file.name):
                     machine.info(
                         f"Secret var '{file.name}' for service '{generator.name}' in machine {machine.name} is missing."
                     )
                     missing_secret_vars.append(file)
                 else:
-                    msg = secret_vars_store.health_check(
+                    msg = machine.secret_vars_store.health_check(
                         generator=generator,
                         file_name=file.name,
                     )
@@ -75,15 +70,15 @@ def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatu
                         )
                         unfixed_secret_vars.append(file)
 
-            elif not public_vars_store.exists(generator, file.name):
+            elif not machine.public_vars_store.exists(generator, file.name):
                 machine.info(
                     f"Public var '{file.name}' for service '{generator.name}' in machine {machine.name} is missing."
                 )
                 missing_public_vars.append(file)
         # check if invalidation hash is up to date
         if not (
-            secret_vars_store.hash_is_valid(generator)
-            and public_vars_store.hash_is_valid(generator)
+            machine.secret_vars_store.hash_is_valid(generator)
+            and machine.public_vars_store.hash_is_valid(generator)
         ):
             invalid_generators.append(generator.name)
             machine.info(

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

@@ -1,21 +1,14 @@
 import argparse
-import importlib
 import logging
 
 from clan_cli.completions import add_dynamic_completer, complete_machines
 from clan_cli.errors import ClanError
 from clan_cli.machines.machines import Machine
-from clan_cli.vars._types import StoreBase
 
 log = logging.getLogger(__name__)
 
 
 def fix_vars(machine: Machine, generator_name: None | str = None) -> None:
-    secret_vars_module = importlib.import_module(machine.secret_vars_module)
-    secret_vars_store: StoreBase = secret_vars_module.SecretStore(machine=machine)
-    public_vars_module = importlib.import_module(machine.public_vars_module)
-    public_vars_store: StoreBase = public_vars_module.FactStore(machine=machine)
-
     generators = machine.vars_generators
     if generator_name:
         for generator in generators:
@@ -29,8 +22,8 @@ def fix_vars(machine: Machine, generator_name: None | str = None) -> None:
             raise ClanError(err_msg)
 
     for generator in generators:
-        public_vars_store.fix(generator=generator)
-        secret_vars_store.fix(generator=generator)
+        machine.public_vars_store.fix(generator=generator)
+        machine.secret_vars_store.fix(generator=generator)
 
 
 def fix_command(args: argparse.Namespace) -> None:

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

@@ -1,5 +1,4 @@
 import argparse
-import importlib
 import logging
 
 from clan_cli.api import API
@@ -7,7 +6,6 @@ from clan_cli.completions import add_dynamic_completer, complete_machines
 from clan_cli.errors import ClanError
 from clan_cli.flake import Flake
 from clan_cli.machines.machines import Machine
-from clan_cli.vars._types import StoreBase
 
 from ._types import GeneratorUpdate
 from .generate import Generator, Prompt, Var, execute_generator
@@ -15,21 +13,11 @@ from .generate import Generator, Prompt, Var, execute_generator
 log = logging.getLogger(__name__)
 
 
-def public_store(machine: Machine) -> StoreBase:
-    public_vars_module = importlib.import_module(machine.public_vars_module)
-    return public_vars_module.FactStore(machine=machine)
-
-
-def secret_store(machine: Machine) -> StoreBase:
-    secret_vars_module = importlib.import_module(machine.secret_vars_module)
-    return secret_vars_module.SecretStore(machine=machine)
-
-
 @API.register
 def get_vars(base_dir: str, machine_name: str) -> list[Var]:
     machine = Machine(name=machine_name, flake=Flake(base_dir))
-    pub_store = public_store(machine)
-    sec_store = secret_store(machine)
+    pub_store = machine.public_vars_store
+    sec_store = machine.secret_vars_store
     all_vars = []
     for generator in machine.vars_generators:
         for var in generator.files:
@@ -50,10 +38,10 @@ def _get_previous_value(
     if not prompt.persist:
         return None
 
-    pub_store = public_store(machine)
+    pub_store = machine.public_vars_store
     if pub_store.exists(generator, prompt.name):
         return pub_store.get(generator, prompt.name).decode()
-    sec_store = secret_store(machine)
+    sec_store = machine.secret_vars_store
     if sec_store.exists(generator, prompt.name):
         return sec_store.get(generator, prompt.name).decode()
     return None
@@ -87,8 +75,8 @@ def set_prompts(
         execute_generator(
             machine,
             generator,
-            secret_vars_store=secret_store(machine),
-            public_vars_store=public_store(machine),
+            secret_vars_store=machine.secret_vars_store,
+            public_vars_store=machine.public_vars_store,
             prompt_values=update.prompt_values,
         )
 

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

@@ -1,5 +1,4 @@
 import argparse
-import importlib
 import logging
 from pathlib import Path
 
@@ -10,12 +9,12 @@ log = logging.getLogger(__name__)
 
 
 def upload_secret_vars(machine: Machine, directory: Path | None = None) -> None:
-    secret_store_module = importlib.import_module(machine.secret_vars_module)
-    secret_store = secret_store_module.SecretStore(machine=machine)
     if directory:
-        secret_store.populate_dir(directory, phases=["activation", "users", "services"])
+        machine.secret_vars_store.populate_dir(
+            directory, phases=["activation", "users", "services"]
+        )
     else:
-        secret_store.upload(phases=["activation", "users", "services"])
+        machine.secret_vars_store.upload(phases=["activation", "users", "services"])
 
 
 def upload_command(args: argparse.Namespace) -> None:

pkgs/clan-cli/clan_cli/vms/run.py

@@ -1,5 +1,4 @@
 import argparse
-import importlib
 import json
 import logging
 import os
@@ -55,9 +54,7 @@ def build_vm(
         nix_options = []
     secrets_dir = get_secrets(machine, tmpdir)
 
-    public_facts_module = importlib.import_module(machine.public_facts_module)
-    public_facts_store = public_facts_module.FactStore(machine=machine)
-    public_facts = public_facts_store.get_all()
+    public_facts = machine.public_facts_store.get_all()
 
     nixos_config_file = machine.build_nix(
         "config.system.clan.vm.create",
@@ -81,12 +78,9 @@ def get_secrets(
     secrets_dir = tmpdir / "secrets"
     secrets_dir.mkdir(parents=True, exist_ok=True)
 
-    secret_facts_module = importlib.import_module(machine.secret_facts_module)
-    secret_facts_store = secret_facts_module.SecretStore(machine=machine)
-
     generate_facts([machine])
 
-    secret_facts_store.upload(secrets_dir)
+    machine.secret_facts_store.upload(secrets_dir)
     return secrets_dir