WIP: clan-cli secrets: add secret_store as python class

nixosModules/clanCore/outputs.nix

@@ -45,6 +45,28 @@
           '';
           default = "${pkgs.coreutils}/bin/true";
         };
+        secretsModule = lib.mkOption {
+          type = lib.types.path;
+          default = "${pkgs.coreutils}/bin/true";
+          description = ''
+            the module that generates secrets.
+            A needs to define a python class SecretStore which implements the following methods:
+            set, get, exists
+          '';
+        };
+        secretsData = lib.mkOption {
+          type = lib.types.path;
+          description = ''
+            secret data as json for the generator
+          '';
+          default = pkgs.writers.writeJSON "secrets.json" (lib.mapAttrs
+            (_name: secret: {
+              secrets = builtins.attrNames secret.secrets;
+              facts = lib.mapAttrs (_: secret: secret.path) secret.facts;
+              generator = secret.generator.finalScript;
+            })
+            config.clanCore.secrets);
+        };
         vm.create = lib.mkOption {
           type = lib.types.path;
           description = ''
@@ -60,7 +82,7 @@
   # optimization for faster secret generate/upload and machines update
   config = {
     system.clan.deployment.data = {
-      inherit (config.system.clan) uploadSecrets generateSecrets;
+      inherit (config.system.clan) uploadSecrets generateSecrets secretsModule secretsData;
       inherit (config.clan.networking) deploymentAddress;
       inherit (config.clanCore) secretsUploadDirectory;
     };

nixosModules/clanCore/secrets/password-store.nix

@@ -1,7 +1,4 @@
 { config, lib, pkgs, ... }:
-let
-  passwordstoreDir = "\${PASSWORD_STORE_DIR:-$HOME/.password-store}";
-in
 {
   options.clan.password-store.targetDirectory = lib.mkOption {
     type = lib.types.path;
@@ -13,103 +10,80 @@ in
   config = lib.mkIf (config.clanCore.secretStore == "password-store") {
     clanCore.secretsDirectory = config.clan.password-store.targetDirectory;
     clanCore.secretsUploadDirectory = config.clan.password-store.targetDirectory;
-    system.clan.generateSecrets = lib.mkIf (config.clanCore.secrets != { }) (
+    system.clan.secretsModule = pkgs.writeText "pass.py" ''
-      pkgs.writeScript "generate-secrets" ''
+      import os
-        #!/bin/sh
+      import subprocess
-        set -efu
+      from clan_cli.machines.machines import Machine
+      from pathlib import Path
 
-        test -d "$CLAN_DIR"
-        PATH=${lib.makeBinPath [
-          pkgs.pass
-        ]}:$PATH
 
-        # TODO maybe initialize password store if it doesn't exist yet
+      class SecretStore:
+          def __init__(self, machine: Machine) -> None:
+              self.machine = machine
 
-        ${lib.foldlAttrs (acc: n: v: ''
+          def set(self, service: str, name: str, value: str):
-          ${acc}
+              subprocess.run(
-          # ${n}
+                  ["${pkgs.pass}/bin/pass", "insert", "-m", f"machines/{self.machine.name}/{name}"],
-          # if any of the secrets are missing, we regenerate all connected facts/secrets
+                  input=value.encode("utf-8"),
-          (if ! (${lib.concatMapStringsSep " && " (x: "test -e ${passwordstoreDir}/machines/${config.clanCore.machineName}/${x.name}.gpg >/dev/null") (lib.attrValues v.secrets)}); then
+                  check=True,
+              )
 
-            tmpdir=$(mktemp -d)
+          def get(self, service: str, name: str) -> str:
-            trap "rm -rf $tmpdir" EXIT
+              return subprocess.run(
-            cd $tmpdir
+                  ["${pkgs.pass}/bin/pass", "show", f"machines/{self.machine.name}/{name}"],
+                  check=True,
+                  stdout=subprocess.PIPE,
+                  text=True,
+              ).stdout
 
-            facts=$(mktemp -d)
+          def exists(self, service: str, name: str) -> bool:
-            trap "rm -rf $facts" EXIT
+              password_store = os.environ.get("PASSWORD_STORE_DIR", f"{os.environ['HOME']}/.password-store")
-            secrets=$(mktemp -d)
+              secret_path = Path(password_store) / f"machines/{self.machine.name}/{name}.gpg"
-            trap "rm -rf $secrets" EXIT
+              print(f"checking {secret_path}")
-            ( ${v.generator.finalScript} )
+              return secret_path.exists()
 
-            ${lib.concatMapStrings (fact: ''
+          def generate_hash(self) -> str:
-              mkdir -p "$CLAN_DIR"/"$(dirname ${fact.path})"
+              password_store = os.environ.get("PASSWORD_STORE_DIR", f"{os.environ['HOME']}/.password-store")
-              cp "$facts"/${fact.name} "$CLAN_DIR"/${fact.path}
+              hashes = []
-            '') (lib.attrValues v.facts)}
+              hashes.append(
+                  subprocess.run(
+                      ["${pkgs.git}/bin/git", "-C", password_store, "log", "-1", "--format=%H", f"machines/{self.machine.name}"],
+                      stdout=subprocess.PIPE,
+                      text=True,
+                  ).stdout.strip()
+              )
+              for symlink in Path(password_store).glob(f"machines/{self.machine.name}/**/*"):
+                  if symlink.is_symlink():
+                      hashes.append(
+                          subprocess.run(
+                              ["${pkgs.git}/bin/git", "-C", password_store, "log", "-1", "--format=%H", symlink],
+                              stdout=subprocess.PIPE,
+                              text=True,
+                          ).stdout.strip()
+                      )
 
-            ${lib.concatMapStrings (secret: ''
+              # we sort the hashes to make sure that the order is always the same
-              cat "$secrets"/${secret.name} | pass insert -m machines/${config.clanCore.machineName}/${secret.name}
+              hashes.sort()
-            '') (lib.attrValues v.secrets)}
+              return "\n".join(hashes)
-          fi)
-        '') "" config.clanCore.secrets}
-      ''
-    );
-    system.clan.uploadSecrets = pkgs.writeScript "upload-secrets" ''
-      #!/bin/sh
-      set -efu
 
-      umask 0077
+          def update_check(self):
+              local_hash = self.generate_hash()
+              remote_hash = self.machine.host.run(
+                  ["cat", "${config.clan.password-store.targetDirectory}/.pass_info"],
+                  check=False,
+                  stdout=subprocess.PIPE,
+              ).stdout.strip()
 
-      PATH=${lib.makeBinPath [
+              if not remote_hash:
-        pkgs.pass
+                  print("remote hash is empty")
-        pkgs.git
+                  return False
-        pkgs.findutils
-        pkgs.rsync
-      ]}:$PATH:${lib.getBin pkgs.openssh}
 
-      if test -e ${passwordstoreDir}/.git; then
+              return local_hash == remote_hash
-        local_pass_info=$(
-          git -C ${passwordstoreDir} log -1 --format=%H machines/${config.clanCore.machineName}
-          # we append a hash for every symlink, otherwise we would miss updates on
-          # files where the symlink points to
-          find ${passwordstoreDir}/machines/${config.clanCore.machineName} -type l \
-              -exec realpath {} + |
-            sort |
-            xargs -r -n 1 git -C ${passwordstoreDir} log -1 --format=%H
-        )
-        remote_pass_info=$(ssh ${config.clan.networking.deploymentAddress} -- ${lib.escapeShellArg ''
-          cat ${config.clan.password-store.targetDirectory}/.pass_info || :
-        ''} || :)
 
-        if test "$local_pass_info" = "$remote_pass_info"; then
+          def upload(self, output_dir: Path, secrets: list[str, str]) -> None:
-          echo secrets already match
+              for service, secret in secrets:
-          exit 23
+                  (output_dir / secret).write_text(self.get(service, secret))
-        fi
+              (output_dir / ".pass_info").write_text(self.generate_hash())
-      fi
-
-      find ${passwordstoreDir}/machines/${config.clanCore.machineName} -type f -follow ! -name .gpg-id |
-      while read -r gpg_path; do
-
-        rel_name=''${gpg_path#${passwordstoreDir}}
-        rel_name=''${rel_name%.gpg}
-
-        pass_date=$(
-          if test -e ${passwordstoreDir}/.git; then
-            git -C ${passwordstoreDir} log -1 --format=%aI "$gpg_path"
-          fi
-        )
-        pass_name=$rel_name
-        tmp_path="$SECRETS_DIR"/$(basename $rel_name)
-
-        mkdir -p "$(dirname "$tmp_path")"
-        pass show "$pass_name" > "$tmp_path"
-        if [ -n "$pass_date" ]; then
-          touch -d "$pass_date" "$tmp_path"
-        fi
-      done
-
-      if test -n "''${local_pass_info-}"; then
-        echo "$local_pass_info" > "$SECRETS_DIR"/.pass_info
-      fi
     '';
   };
 }

nixosModules/clanCore/secrets/sops.nix

@@ -27,29 +27,55 @@ in
     clanCore.secretsPrefix = config.clanCore.machineName + "-";
     system.clan = lib.mkIf (config.clanCore.secrets != { }) {
 
-      generateSecrets = pkgs.writeScript "generate-secrets" ''
+      secretsModule = pkgs.writeText "sops.py" ''
-        #!${pkgs.python3}/bin/python
+        from pathlib import Path
-        import json
+
-        import sys
+        from clan_cli.secrets.folders import sops_secrets_folder
-        from clan_cli.secrets.sops_generate import generate_secrets_from_nix
+        from clan_cli.secrets.secrets import decrypt_secret, encrypt_secret, has_secret
-        args = json.loads(${builtins.toJSON (builtins.toJSON {
+        from clan_cli.secrets.sops import generate_private_key
-          machine_name = config.clanCore.machineName;
+        from clan_cli.secrets.machines import has_machine, add_machine
-          secret_submodules = lib.mapAttrs (_name: secret: {
+        from clan_cli.machines.machines import Machine
-            secrets = builtins.attrNames secret.secrets;
+
-            facts = lib.mapAttrs (_: secret: secret.path) secret.facts;
+
-            generator = secret.generator.finalScript;
+        class SecretStore:
-          }) config.clanCore.secrets;
+            def __init__(self, machine: Machine) -> None:
-        })})
+                self.machine = machine
-        generate_secrets_from_nix(**args)
+                if has_machine(self.machine.flake_dir, self.machine.name):
-      '';
+                    return
-      uploadSecrets = pkgs.writeScript "upload-secrets" ''
+                priv_key, pub_key = generate_private_key()
-        #!${pkgs.python3}/bin/python
+                encrypt_secret(
-        import json
+                    self.machine.flake_dir,
-        import sys
+                    sops_secrets_folder(self.machine.flake_dir) / f"{self.machine.name}-age.key",
-        from clan_cli.secrets.sops_generate import upload_age_key_from_nix
+                    priv_key,
-        # the second toJSON is needed to escape the string for the python
+                )
-        args = json.loads(${builtins.toJSON (builtins.toJSON { machine_name = config.clanCore.machineName; })})
+                add_machine(self.machine.flake_dir, self.machine.name, pub_key, False)
-        upload_age_key_from_nix(**args)
+
+            def set(self, service: str, name: str, value: str):
+                encrypt_secret(
+                    self.machine.flake_dir,
+                    sops_secrets_folder(self.machine.flake_dir) / f"{self.machine.name}-{name}",
+                    value,
+                    add_machines=[self.machine.name],
+                )
+
+            def get(self, service: str, name: str) -> bytes:
+                # TODO: add support for getting a secret
+                pass
+
+            def exists(self, service: str, name: str) -> bool:
+                return has_secret(
+                    self.machine.flake_dir,
+                    f"{self.machine.name}-{name}",
+                )
+
+            def upload(self, output_dir: Path, secrets: list[str, str]) -> None:
+                key_name = f"{self.machine.name}-age.key"
+                if not has_secret(self.machine.flake_dir, key_name):
+                    # skip uploading the secret, not managed by us
+                    return
+                key = decrypt_secret(self.machine.flake_dir, key_name)
+
+                (output_dir / "key.txt").write_text(key)
       '';
     };
     sops.secrets = builtins.mapAttrs

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

@@ -49,6 +49,10 @@ class Machine:
         self.deployment_address = self.machine_data["deploymentAddress"]
         self.upload_secrets = self.machine_data["uploadSecrets"]
         self.generate_secrets = self.machine_data["generateSecrets"]
+        self.secrets_module = self.machine_data["secretsModule"]
+        self.secrets_data = json.loads(
+            Path(self.machine_data["secretsData"]).read_text()
+        )
         self.secrets_upload_directory = self.machine_data["secretsUploadDirectory"]
         self.eval_cache: dict[str, str] = {}
         self.build_cache: dict[str, Path] = {}

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

@@ -1,27 +1,78 @@
 import argparse
 import logging
 import os
-import sys
+import shutil
+import types
+from importlib.machinery import SourceFileLoader
+from pathlib import Path
+from tempfile import TemporaryDirectory
 
-from clan_cli.cmd import Log, run
+from clan_cli.cmd import run
 
+from ..errors import ClanError
 from ..machines.machines import Machine
+from ..nix import nix_shell
 
 log = logging.getLogger(__name__)
 
 
 def generate_secrets(machine: Machine) -> None:
-    env = os.environ.copy()
+    # load secrets module from file
-    env["CLAN_DIR"] = str(machine.flake_dir)
+    loader = SourceFileLoader("secret_module", machine.secrets_module)
-    env["PYTHONPATH"] = ":".join(sys.path)  # TODO do this in the clanCore module
+    secrets_module = types.ModuleType(loader.name)
+    loader.exec_module(secrets_module)
+    secret_store = secrets_module.SecretStore(machine=machine)
 
-    print(f"generating secrets... {machine.generate_secrets}")
+    with TemporaryDirectory() as d:
-    run(
+        for service in machine.secrets_data:
-        [machine.generate_secrets],
+            print(service)
-        env=env,
+            tmpdir = Path(d) / service
-        error_msg="failed to generate secrets",
+            # check if all secrets exist and generate them if at least one is missing
-        log=Log.BOTH,
+            needs_regeneration = any(
-    )
+                not secret_store.exists(service, secret)
+                for secret in machine.secrets_data[service]["secrets"]
+            ) or any(
+                not (machine.flake_dir / fact).exists()
+                for fact in machine.secrets_data[service]["facts"].values()
+            )
+            for fact in machine.secrets_data[service]["facts"].values():
+                if not (machine.flake_dir / fact).exists():
+                    print(f"fact {fact} is missing")
+            if needs_regeneration:
+                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)
+                # TODO use bubblewrap here
+                cmd = nix_shell(
+                    ["nixpkgs#bash"],
+                    ["bash", "-c", machine.secrets_data[service]["generator"]],
+                )
+                run(
+                    cmd,
+                    env=env,
+                )
+                # 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_store.set(service, secret, secret_file.read_text())
+                # store facts
+                for name, fact_path in machine.secrets_data[service]["facts"].items():
+                    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_path = machine.flake_dir / fact_path
+                    fact_path.parent.mkdir(parents=True, exist_ok=True)
+                    shutil.copyfile(fact_file, fact_path)
 
     print("successfully generated secrets")
 

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

@@ -1,128 +0,0 @@
-import logging
-import os
-import shlex
-import shutil
-import sys
-from pathlib import Path
-from tempfile import TemporaryDirectory
-from typing import Any
-
-from clan_cli.cmd import Log, run
-from clan_cli.nix import nix_shell
-
-from ..errors import ClanError
-from .folders import sops_secrets_folder
-from .machines import add_machine, has_machine
-from .secrets import decrypt_secret, encrypt_secret, has_secret
-from .sops import generate_private_key
-
-log = logging.getLogger(__name__)
-
-
-def generate_host_key(flake_dir: Path, machine_name: str) -> None:
-    if has_machine(flake_dir, machine_name):
-        return
-    priv_key, pub_key = generate_private_key()
-    encrypt_secret(
-        flake_dir,
-        sops_secrets_folder(flake_dir) / f"{machine_name}-age.key",
-        priv_key,
-    )
-    add_machine(flake_dir, machine_name, pub_key, False)
-
-
-def generate_secrets_group(
-    flake_dir: Path,
-    secret_group: str,
-    machine_name: str,
-    tempdir: Path,
-    secret_options: dict[str, Any],
-) -> None:
-    clan_dir = flake_dir
-    secrets = secret_options["secrets"]
-    needs_regeneration = any(
-        not has_secret(flake_dir, f"{machine_name}-{name}") for name in secrets
-    ) or any(
-        not (flake_dir / fact).exists() for fact in secret_options["facts"].values()
-    )
-
-    generator = secret_options["generator"]
-    subdir = tempdir / secret_group
-    if needs_regeneration:
-        facts_dir = subdir / "facts"
-        facts_dir.mkdir(parents=True)
-        secrets_dir = subdir / "secrets"
-        secrets_dir.mkdir(parents=True)
-
-        text = f"""
-set -euo pipefail
-export facts={shlex.quote(str(facts_dir))}
-export secrets={shlex.quote(str(secrets_dir))}
-{generator}
-        """
-        cmd = nix_shell(["nixpkgs#bash"], ["bash", "-c", text])
-        run(cmd, log=Log.BOTH)
-
-        for name in secrets:
-            secret_file = secrets_dir / name
-            if not secret_file.is_file():
-                msg = f"did not generate a file for '{name}' when running the following command:\n"
-                msg += text
-                raise ClanError(msg)
-            encrypt_secret(
-                flake_dir,
-                sops_secrets_folder(flake_dir) / f"{machine_name}-{name}",
-                secret_file.read_text(),
-                add_machines=[machine_name],
-            )
-        for name, fact_path in secret_options["facts"].items():
-            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 += text
-                raise ClanError(msg)
-            fact_path = clan_dir / fact_path
-            fact_path.parent.mkdir(parents=True, exist_ok=True)
-            shutil.copyfile(fact_file, fact_path)
-
-
-# this is called by the sops.nix clan core module
-def generate_secrets_from_nix(
-    machine_name: str,
-    secret_submodules: dict[str, Any],
-) -> None:
-    flake_dir = Path(os.environ["CLAN_DIR"])
-    generate_host_key(flake_dir, machine_name)
-    errors = {}
-    log.debug("Generating secrets for machine %s and flake %s", machine_name, flake_dir)
-    with TemporaryDirectory() as d:
-        # if any of the secrets are missing, we regenerate all connected facts/secrets
-        for secret_group, secret_options in secret_submodules.items():
-            try:
-                generate_secrets_group(
-                    flake_dir, secret_group, machine_name, Path(d), secret_options
-                )
-            except ClanError as e:
-                errors[secret_group] = e
-    for secret_group, error in errors.items():
-        print(f"failed to generate secrets for {machine_name}/{secret_group}:")
-        print(error, file=sys.stderr)
-    if len(errors) > 0:
-        sys.exit(1)
-
-
-# this is called by the sops.nix clan core module
-def upload_age_key_from_nix(
-    machine_name: str,
-) -> None:
-    flake_dir = Path(os.environ["CLAN_DIR"])
-    log.debug("Uploading secrets for machine %s and flake %s", machine_name, flake_dir)
-    secret_name = f"{machine_name}-age.key"
-    if not has_secret(
-        flake_dir, secret_name
-    ):  # skip uploading the secret, not managed by us
-        return
-    secret = decrypt_secret(flake_dir, secret_name)
-
-    secrets_dir = Path(os.environ["SECRETS_DIR"])
-    (secrets_dir / "key.txt").write_text(secret)

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

@@ -1,5 +1,7 @@
 import argparse
 import logging
+import types
+from importlib.machinery import SourceFileLoader
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
@@ -11,29 +13,41 @@ log = logging.getLogger(__name__)
 
 
 def upload_secrets(machine: Machine) -> None:
-    with TemporaryDirectory() as tempdir_:
+    # load secrets module from file
-        tempdir = Path(tempdir_)
+    loader = SourceFileLoader("secret_module", machine.secrets_module)
-        should_upload = machine.run_upload_secrets(tempdir)
+    secrets_module = types.ModuleType(loader.name)
+    loader.exec_module(secrets_module)
+    secret_store = secrets_module.SecretStore(machine=machine)
 
-        if should_upload:
+    update_check = getattr(secret_store, "update_check", None)
-            host = machine.host
+    if callable(update_check):
+        if update_check():
+            log.info("Secrets already up to date")
+            return
+    with TemporaryDirectory() as tempdir:
+        secrets = []
+        for service in machine.secrets_data:
+            for secret in machine.secrets_data[service]["secrets"]:
+                secrets.append((service, secret))
+        secret_store.upload(Path(tempdir), secrets)
+        host = machine.host
 
-            ssh_cmd = host.ssh_cmd()
+        ssh_cmd = host.ssh_cmd()
-            run(
+        run(
-                nix_shell(
+            nix_shell(
-                    ["nixpkgs#rsync"],
+                ["nixpkgs#rsync"],
-                    [
+                [
-                        "rsync",
+                    "rsync",
-                        "-e",
+                    "-e",
-                        " ".join(["ssh"] + ssh_cmd[2:]),
+                    " ".join(["ssh"] + ssh_cmd[2:]),
-                        "-az",
+                    "-az",
-                        "--delete",
+                    "--delete",
-                        f"{tempdir!s}/",
+                    f"{tempdir!s}/",
-                        f"{host.user}@{host.host}:{machine.secrets_upload_directory}/",
+                    f"{host.user}@{host.host}:{machine.secrets_upload_directory}/",
-                    ],
+                ],
-                ),
+            ),
-                log=Log.BOTH,
+            log=Log.BOTH,
-            )
+        )
 
 
 def upload_command(args: argparse.Namespace) -> None: