secrets upload: skip on exit 23, cleanup

nixosModules/clanCore/secrets/password-store.nix

@@ -76,7 +76,7 @@ in
 
         if test "$local_pass_info" = "$remote_pass_info"; then
           echo secrets already match
-          exit 0
+          exit 23
         fi
       fi
 

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

@@ -62,7 +62,7 @@ class Machine:
             self.name, self.deployment_address, meta={"machine": self}
         )
 
-    def run_upload_secrets(self, secrets_dir: Path) -> None:
+    def run_upload_secrets(self, secrets_dir: Path) -> bool:
         """
         Upload the secrets to the provided directory
         @secrets_dir: the directory to store the secrets in
@@ -73,14 +73,22 @@ class Machine:
             ":".join(sys.path)
         )  # TODO do this in the clanCore module
         env["SECRETS_DIR"] = str(secrets_dir)
-        subprocess.run(
+        print(f"uploading secrets... {self.upload_secrets}")
+        proc = subprocess.run(
             [self.upload_secrets],
             env=env,
-            check=True,
             stdout=subprocess.PIPE,
             text=True,
         )
 
+        if proc.returncode == 23:
+            print("no secrets to upload")
+            return False
+        elif proc.returncode != 0:
+            print("failed generate secrets directory")
+            exit(1)
+        return True
+
     def eval_nix(self, attr: str) -> str:
         """
         eval a nix attribute of the machine

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

@@ -1,76 +1,39 @@
 import argparse
-import json
 import logging
-import shlex
 import subprocess
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
-from ..errors import ClanError
 from ..machines.machines import Machine
-from ..nix import nix_build, nix_config, nix_shell
+from ..nix import nix_shell
 
 log = logging.getLogger(__name__)
 
 
-def build_upload_script(machine: str, clan_dir: Path) -> str:
-    config = nix_config()
-    system = config["system"]
-
-    cmd = nix_build(
-        [
-            f'{clan_dir}#clanInternals.machines."{system}"."{machine}".config.system.clan.uploadSecrets'
-        ]
-    )
-    proc = subprocess.run(cmd, stdout=subprocess.PIPE, text=True)
-    if proc.returncode != 0:
-        raise ClanError(
-            f"failed to upload secrets:\n{shlex.join(cmd)}\nexited with {proc.returncode}"
-        )
-
-    return proc.stdout.strip()
-
-
-def get_deployment_info(machine: str, clan_dir: Path) -> dict:
-    config = nix_config()
-    system = config["system"]
-
-    cmd = nix_build(
-        [
-            f'{clan_dir}#clanInternals.machines."{system}"."{machine}".config.system.clan.deployment.file'
-        ]
-    )
-    proc = subprocess.run(cmd, stdout=subprocess.PIPE, text=True)
-    if proc.returncode != 0:
-        raise ClanError(
-            f"failed to get deploymentAddress:\n{shlex.join(cmd)}\nexited with {proc.returncode}"
-        )
-
-    return json.load(open(proc.stdout.strip()))
-
-
 def upload_secrets(machine: Machine) -> None:
     with TemporaryDirectory() as tempdir_:
         tempdir = Path(tempdir_)
-        machine.run_upload_secrets(tempdir)
-        host = machine.host
+        should_upload = machine.run_upload_secrets(tempdir)
 
-        ssh_cmd = host.ssh_cmd()
-        subprocess.run(
-            nix_shell(
-                ["rsync"],
-                [
-                    "rsync",
-                    "-e",
-                    " ".join(["ssh"] + ssh_cmd[2:]),
-                    "-az",
-                    "--delete",
-                    f"{str(tempdir)}/",
-                    f"{host.user}@{host.host}:{machine.secrets_upload_directory}/",
-                ],
-            ),
-            check=True,
-        )
+        if should_upload:
+            host = machine.host
+
+            ssh_cmd = host.ssh_cmd()
+            subprocess.run(
+                nix_shell(
+                    ["rsync"],
+                    [
+                        "rsync",
+                        "-e",
+                        " ".join(["ssh"] + ssh_cmd[2:]),
+                        "-az",
+                        "--delete",
+                        f"{str(tempdir)}/",
+                        f"{host.user}@{host.host}:{machine.secrets_upload_directory}/",
+                    ],
+                ),
+                check=True,
+            )
 
 
 def upload_command(args: argparse.Namespace) -> None: