From 1a6983e0318016abac09ac053b1f65117b637d4c 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 5f4493ac..4c6f70a7 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 b4aadb83..9ac91cf9 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 a4cacf13..4e3a9774 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 fe97d608..4c311268 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 0c1d4a1d4102e4ff03530a6d3446c7e40d98934f 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 4c311268..fe97d608 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: