diff --git a/pkgs/clan-cli/clan_cli/vars/__init__.py b/pkgs/clan-cli/clan_cli/vars/__init__.py index 888bdea45..66a1d6fa0 100644 --- a/pkgs/clan-cli/clan_cli/vars/__init__.py +++ b/pkgs/clan-cli/clan_cli/vars/__init__.py @@ -7,6 +7,7 @@ from .check import register_check_parser from .generate import register_generate_parser from .get import register_get_parser from .list import register_list_parser +from .set import register_set_parser from .upload import register_upload_parser @@ -85,6 +86,25 @@ For more detailed information, visit: {help_hyperlink("secrets", "https://docs.c ) register_get_parser(get_parser) + set_parser = subparser.add_parser( + "set", + help="set a specific var", + epilog=( + f""" +This subcommand allows setting a specific var for a specific machine. + +Examples: + + $ clan vars set my-server zerotier/vpn-ip + Will set the var for the specified machine. + +For more detailed information, visit: {help_hyperlink("secrets", "https://docs.clan.lol/getting-started/secrets")} + """ + ), + formatter_class=argparse.RawTextHelpFormatter, + ) + register_set_parser(set_parser) + parser_generate = subparser.add_parser( "generate", help="(re-)generate vars for specific or all machines", diff --git a/pkgs/clan-cli/clan_cli/vars/_types.py b/pkgs/clan-cli/clan_cli/vars/_types.py index c19a5cf4a..b86a1b403 100644 --- a/pkgs/clan-cli/clan_cli/vars/_types.py +++ b/pkgs/clan-cli/clan_cli/vars/_types.py @@ -1,5 +1,6 @@ # !/usr/bin/env python3 import json +import shutil from abc import ABC, abstractmethod from dataclasses import dataclass from pathlib import Path @@ -32,6 +33,9 @@ class Var: except UnicodeDecodeError: return "" + def set(self, value: bytes) -> None: + self.store.set(self.generator, self.name, value, self.shared, self.deployed) + @property def exists(self) -> bool: return self.store.exists(self.generator, self.name, self.shared) @@ -109,8 +113,7 @@ class StoreBase(ABC): directory = self.directory(generator_name, var_name, shared) # delete directory if directory.exists(): - for f in directory.glob("*"): - f.unlink() + shutil.rmtree(directory) # re-create directory directory.mkdir(parents=True, exist_ok=True) new_file = self._set(generator_name, var_name, value, shared, deployed) diff --git a/pkgs/clan-cli/clan_cli/vars/generate.py b/pkgs/clan-cli/clan_cli/vars/generate.py index da5349059..8d2715fce 100644 --- a/pkgs/clan-cli/clan_cli/vars/generate.py +++ b/pkgs/clan-cli/clan_cli/vars/generate.py @@ -3,7 +3,6 @@ import importlib import logging import os import sys -from getpass import getpass from graphlib import TopologicalSorter from pathlib import Path from tempfile import TemporaryDirectory @@ -22,6 +21,7 @@ from clan_cli.machines.machines import Machine from clan_cli.nix import nix_shell from .check import check_vars +from .prompt import prompt from .public_modules import FactStoreBase from .secret_modules import SecretStoreBase @@ -133,11 +133,11 @@ def execute_generator( if machine.vars_generators[generator_name]["prompts"]: tmpdir_prompts.mkdir() env["prompts"] = str(tmpdir_prompts) - for prompt_name, prompt in machine.vars_generators[generator_name][ + for prompt_name, prompt_ in machine.vars_generators[generator_name][ "prompts" ].items(): prompt_file = tmpdir_prompts / prompt_name - value = prompt_func(prompt["description"], prompt["type"]) + value = prompt(prompt_["description"], prompt_["type"]) prompt_file.write_text(value) if sys.platform == "linux": @@ -184,21 +184,6 @@ def execute_generator( return True -def prompt_func(description: str, input_type: str) -> str: - if input_type == "line": - result = input(f"Enter the value for {description}: ") - elif input_type == "multiline": - print(f"Enter the value for {description} (Finish with Ctrl-D): ") - result = sys.stdin.read() - elif input_type == "hidden": - result = getpass(f"Enter the value for {description} (hidden): ") - else: - msg = f"Unknown input type: {input_type} for prompt {description}" - raise ClanError(msg) - log.info("Input received. Processing...") - return result - - def _get_subgraph(graph: dict[str, set], vertex: str) -> dict[str, set]: visited = set() queue = [vertex] diff --git a/pkgs/clan-cli/clan_cli/vars/get.py b/pkgs/clan-cli/clan_cli/vars/get.py index 6bb37ea27..4a065c5b0 100644 --- a/pkgs/clan-cli/clan_cli/vars/get.py +++ b/pkgs/clan-cli/clan_cli/vars/get.py @@ -13,14 +13,15 @@ from .list import all_vars log = logging.getLogger(__name__) -def get_var(machine: Machine, var_id: str) -> Var | None: +def get_var(machine: Machine, var_id: str) -> Var: vars_ = all_vars(machine) results = [] for var in vars_: if var_id in var.id: results.append(var) if len(results) == 0: - return None + msg = f"No var found for search string: {var_id}" + raise ClanError(msg) if len(results) > 1: error = ( f"Found multiple vars for {var_id}:\n - " @@ -29,28 +30,33 @@ def get_var(machine: Machine, var_id: str) -> Var | None: ) raise ClanError(error) # we have exactly one result at this point - result = results[0] - if var_id == result.id: - return result - msg = f"Did you mean: {result.id}" + var = results[0] + if var_id == var.id: + return var + msg = f"Did you mean: {var.id}" raise ClanError(msg) -def get_command( - machine: str, var_id: str, flake: FlakeId, quiet: bool, **kwargs: dict -) -> None: +def get_command(machine: str, var_id: str, flake: FlakeId) -> None: _machine = Machine(name=machine, flake=flake) var = get_var(_machine, var_id) - if var is None: - msg = f"No var found for search string: {var_id}" - raise ClanError(msg) if not var.exists: msg = f"Var {var.id} has not been generated yet" raise ClanError(msg) - if quiet: + if sys.stdout.isatty(): sys.stdout.buffer.write(var.value) else: - print(f"{var.id}: {var.printable_value}") + print(var.printable_value) + + +def _get_command( + args: argparse.Namespace, +) -> None: + get_command( + machine=args.machine, + var_id=args.var_id, + flake=args.flake, + ) def register_get_parser(parser: argparse.ArgumentParser) -> None: @@ -65,10 +71,4 @@ def register_get_parser(parser: argparse.ArgumentParser) -> None: help="The var id to get the value for. Example: ssh-keys/pubkey", ) - parser.add_argument( - "--quiet", - "-q", - help="Only print the value of the var", - action="store_true", - ) - parser.set_defaults(func=lambda args: get_command(**vars(args))) + parser.set_defaults(func=_get_command) diff --git a/pkgs/clan-cli/clan_cli/vars/prompt.py b/pkgs/clan-cli/clan_cli/vars/prompt.py new file mode 100644 index 000000000..4ac1aebe4 --- /dev/null +++ b/pkgs/clan-cli/clan_cli/vars/prompt.py @@ -0,0 +1,22 @@ +import logging +import sys +from getpass import getpass + +from clan_cli.errors import ClanError + +log = logging.getLogger(__name__) + + +def prompt(description: str, input_type: str) -> str: + if input_type == "line": + result = input(f"Enter the value for {description}: ") + elif input_type == "multiline": + print(f"Enter the value for {description} (Finish with Ctrl-D): ") + result = sys.stdin.read() + elif input_type == "hidden": + result = getpass(f"Enter the value for {description} (hidden): ") + else: + msg = f"Unknown input type: {input_type} for prompt {description}" + raise ClanError(msg) + log.info("Input received. Processing...") + return result diff --git a/pkgs/clan-cli/clan_cli/vars/set.py b/pkgs/clan-cli/clan_cli/vars/set.py new file mode 100644 index 000000000..d374330ca --- /dev/null +++ b/pkgs/clan-cli/clan_cli/vars/set.py @@ -0,0 +1,41 @@ +import argparse +import logging +import sys + +from clan_cli.clan_uri import FlakeId +from clan_cli.completions import add_dynamic_completer, complete_machines +from clan_cli.machines.machines import Machine +from clan_cli.vars.get import get_var + +from .prompt import prompt + +log = logging.getLogger(__name__) + + +def get_command(machine: str, var_id: str, flake: FlakeId) -> None: + _machine = Machine(name=machine, flake=flake) + var = get_var(_machine, var_id) + if sys.stdin.isatty(): + new_value = prompt(var.id, "hidden").encode("utf-8") + else: + new_value = sys.stdin.buffer.read() + var.set(new_value) + + +def _get_command(args: argparse.Namespace) -> None: + get_command(args.machine, args.var_id, args.flake) + + +def register_set_parser(parser: argparse.ArgumentParser) -> None: + machines_arg = parser.add_argument( + "machine", + help="The machine to set a var for", + ) + add_dynamic_completer(machines_arg, complete_machines) + + parser.add_argument( + "var_id", + help="The var id for which to set the value. Example: ssh-keys/pubkey", + ) + + parser.set_defaults(func=_get_command)