From 38e2c29f08fccbe891a8f5d0cf24affa6beeb461 Mon Sep 17 00:00:00 2001 From: Qubasa Date: Fri, 12 Jan 2024 17:01:46 +0100 Subject: [PATCH 1/2] cmd.py refactor part 6 --- pkgs/clan-cli/clan_cli/secrets/sops.py | 19 ++++------- .../clan_cli/secrets/sops_generate.py | 4 +-- pkgs/clan-cli/clan_cli/secrets/upload.py | 6 ++-- pkgs/clan-cli/clan_cli/ssh/cli.py | 32 ++++++++----------- 4 files changed, 24 insertions(+), 37 deletions(-) diff --git a/pkgs/clan-cli/clan_cli/secrets/sops.py b/pkgs/clan-cli/clan_cli/secrets/sops.py index 5f4493ac9..4c6f70a7c 100644 --- a/pkgs/clan-cli/clan_cli/secrets/sops.py +++ b/pkgs/clan-cli/clan_cli/secrets/sops.py @@ -8,6 +8,7 @@ from pathlib import Path from tempfile import NamedTemporaryFile from typing import IO +from ..cmd import Log, run from ..dirs import user_config_dir from ..errors import ClanError from ..nix import nix_shell @@ -36,7 +37,7 @@ def get_public_key(privkey: str) -> str: def generate_private_key() -> tuple[str, str]: cmd = nix_shell(["nixpkgs#age"], ["age-keygen"]) try: - proc = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, text=True) + proc = run(cmd) res = proc.stdout.strip() pubkey = None private_key = None @@ -129,11 +130,7 @@ def update_keys(secret_path: Path, keys: list[str]) -> None: str(secret_path / "secret"), ], ) - res = subprocess.run(cmd) - if res.returncode != 0: - raise ClanError( - f"Failed to update keys for {secret_path}: sops exited with {res.returncode}" - ) + run(cmd, log=Log.BOTH, error_msg=f"Could not update keys for {secret_path}") def encrypt_file( @@ -147,7 +144,7 @@ def encrypt_file( args = ["sops", "--config", str(manifest)] args.extend([str(secret_path)]) cmd = nix_shell(["nixpkgs#sops"], args) - p = subprocess.run(cmd) + p = run(cmd, log=Log.BOTH, check=False) # returns 200 if the file is changed if p.returncode != 0 and p.returncode != 200: raise ClanError( @@ -167,7 +164,7 @@ def encrypt_file( args = ["sops", "--config", str(manifest)] args.extend(["-i", "--encrypt", str(f.name)]) cmd = nix_shell(["nixpkgs#sops"], args) - subprocess.run(cmd, check=True) + run(cmd, log=Log.BOTH) # atomic copy of the encrypted file with NamedTemporaryFile(dir=folder, delete=False) as f2: shutil.copyfile(f.name, f2.name) @@ -185,11 +182,7 @@ def decrypt_file(secret_path: Path) -> str: ["nixpkgs#sops"], ["sops", "--config", str(manifest), "--decrypt", str(secret_path)], ) - res = subprocess.run(cmd, stdout=subprocess.PIPE, text=True) - if res.returncode != 0: - raise ClanError( - f"Failed to decrypt {secret_path}: sops exited with {res.returncode}" - ) + res = run(cmd, error_msg=f"Could not decrypt {secret_path}") return res.stdout diff --git a/pkgs/clan-cli/clan_cli/secrets/sops_generate.py b/pkgs/clan-cli/clan_cli/secrets/sops_generate.py index b4aadb83c..9ac91cf93 100644 --- a/pkgs/clan-cli/clan_cli/secrets/sops_generate.py +++ b/pkgs/clan-cli/clan_cli/secrets/sops_generate.py @@ -7,7 +7,7 @@ from pathlib import Path from tempfile import TemporaryDirectory from typing import Any -from clan_cli.cmd import run +from clan_cli.cmd import Log, run from clan_cli.nix import nix_shell from ..errors import ClanError @@ -61,7 +61,7 @@ export secrets={shlex.quote(str(secrets_dir))} {generator} """ cmd = nix_shell(["nixpkgs#bash"], ["bash", "-c", text]) - run(cmd) + run(cmd, log=Log.BOTH) for name in secrets: secret_file = secrets_dir / name diff --git a/pkgs/clan-cli/clan_cli/secrets/upload.py b/pkgs/clan-cli/clan_cli/secrets/upload.py index a4cacf138..4e3a97740 100644 --- a/pkgs/clan-cli/clan_cli/secrets/upload.py +++ b/pkgs/clan-cli/clan_cli/secrets/upload.py @@ -1,9 +1,9 @@ import argparse import logging -import subprocess from pathlib import Path from tempfile import TemporaryDirectory +from ..cmd import Log, run from ..machines.machines import Machine from ..nix import nix_shell @@ -19,7 +19,7 @@ def upload_secrets(machine: Machine) -> None: host = machine.host ssh_cmd = host.ssh_cmd() - subprocess.run( + run( nix_shell( ["nixpkgs#rsync"], [ @@ -32,7 +32,7 @@ def upload_secrets(machine: Machine) -> None: f"{host.user}@{host.host}:{machine.secrets_upload_directory}/", ], ), - check=True, + log=Log.BOTH, ) diff --git a/pkgs/clan-cli/clan_cli/ssh/cli.py b/pkgs/clan-cli/clan_cli/ssh/cli.py index fe97d6082..4c3112687 100644 --- a/pkgs/clan-cli/clan_cli/ssh/cli.py +++ b/pkgs/clan-cli/clan_cli/ssh/cli.py @@ -1,7 +1,7 @@ import argparse import json -import subprocess +from ..cmd import Log, run from ..nix import nix_shell @@ -30,27 +30,21 @@ def ssh( f"{user}@{host}", ] cmd = nix_shell(packages, ["torify", *password_args, *_ssh_args]) - subprocess.run(cmd) + run(cmd, log=Log.BOTH) def qrcode_scan(picture_file: str) -> str: - return ( - subprocess.run( - nix_shell( - ["nixpkgs#zbar"], - [ - "zbarimg", - "--quiet", - "--raw", - picture_file, - ], - ), - stdout=subprocess.PIPE, - check=True, - ) - .stdout.decode() - .strip() - ) + return run( + nix_shell( + ["nixpkgs#zbar"], + [ + "zbarimg", + "--quiet", + "--raw", + picture_file, + ], + ), + ).stdout.strip() def main(args: argparse.Namespace) -> None: From a72477e574f4a614d7b7a8334f43c2f9cc8d3c4b Mon Sep 17 00:00:00 2001 From: Qubasa Date: Fri, 12 Jan 2024 17:21:48 +0100 Subject: [PATCH 2/2] cmd.py refactor part 6 --- pkgs/clan-cli/clan_cli/ssh/cli.py | 32 ++++++++++++++++++------------- 1 file changed, 19 insertions(+), 13 deletions(-) diff --git a/pkgs/clan-cli/clan_cli/ssh/cli.py b/pkgs/clan-cli/clan_cli/ssh/cli.py index 4c3112687..fe97d6082 100644 --- a/pkgs/clan-cli/clan_cli/ssh/cli.py +++ b/pkgs/clan-cli/clan_cli/ssh/cli.py @@ -1,7 +1,7 @@ import argparse import json +import subprocess -from ..cmd import Log, run from ..nix import nix_shell @@ -30,21 +30,27 @@ def ssh( f"{user}@{host}", ] cmd = nix_shell(packages, ["torify", *password_args, *_ssh_args]) - run(cmd, log=Log.BOTH) + subprocess.run(cmd) def qrcode_scan(picture_file: str) -> str: - return run( - nix_shell( - ["nixpkgs#zbar"], - [ - "zbarimg", - "--quiet", - "--raw", - picture_file, - ], - ), - ).stdout.strip() + return ( + subprocess.run( + nix_shell( + ["nixpkgs#zbar"], + [ + "zbarimg", + "--quiet", + "--raw", + picture_file, + ], + ), + stdout=subprocess.PIPE, + check=True, + ) + .stdout.decode() + .strip() + ) def main(args: argparse.Namespace) -> None: