secrets upload: skip on exit 23, cleanup

This commit is contained in:
lassulus
2023-10-04 21:29:19 +02:00
parent 718deb9c70
commit 528d0ee45a
3 changed files with 33 additions and 62 deletions

View File

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

View File

@@ -62,7 +62,7 @@ class Machine:
self.name, self.deployment_address, meta={"machine": self} 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 Upload the secrets to the provided directory
@secrets_dir: the directory to store the secrets in @secrets_dir: the directory to store the secrets in
@@ -73,14 +73,22 @@ class Machine:
":".join(sys.path) ":".join(sys.path)
) # TODO do this in the clanCore module ) # TODO do this in the clanCore module
env["SECRETS_DIR"] = str(secrets_dir) env["SECRETS_DIR"] = str(secrets_dir)
subprocess.run( print(f"uploading secrets... {self.upload_secrets}")
proc = subprocess.run(
[self.upload_secrets], [self.upload_secrets],
env=env, env=env,
check=True,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
text=True, 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: def eval_nix(self, attr: str) -> str:
""" """
eval a nix attribute of the machine eval a nix attribute of the machine

View File

@@ -1,58 +1,21 @@
import argparse import argparse
import json
import logging import logging
import shlex
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from tempfile import TemporaryDirectory from tempfile import TemporaryDirectory
from ..errors import ClanError
from ..machines.machines import Machine from ..machines.machines import Machine
from ..nix import nix_build, nix_config, nix_shell from ..nix import nix_shell
log = logging.getLogger(__name__) 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: def upload_secrets(machine: Machine) -> None:
with TemporaryDirectory() as tempdir_: with TemporaryDirectory() as tempdir_:
tempdir = Path(tempdir_) tempdir = Path(tempdir_)
machine.run_upload_secrets(tempdir) should_upload = machine.run_upload_secrets(tempdir)
if should_upload:
host = machine.host host = machine.host
ssh_cmd = host.ssh_cmd() ssh_cmd = host.ssh_cmd()