diff --git a/pkgs/clan-cli/clan_cli/vars/__init__.py b/pkgs/clan-cli/clan_cli/vars/__init__.py index 923580785..888bdea45 100644 --- a/pkgs/clan-cli/clan_cli/vars/__init__.py +++ b/pkgs/clan-cli/clan_cli/vars/__init__.py @@ -5,6 +5,7 @@ from clan_cli.hyperlink import help_hyperlink 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 .upload import register_upload_parser @@ -65,6 +66,25 @@ For more detailed information, visit: {help_hyperlink("secrets", "https://docs.c ) register_list_parser(list_parser) + get_parser = subparser.add_parser( + "get", + help="get a specific var", + epilog=( + f""" +This subcommand allows getting a specific var for a specific machine. + +Examples: + + $ clan vars get my-server zerotier/vpn-ip + Will get 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_get_parser(get_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 e1ff4cf64..c19a5cf4a 100644 --- a/pkgs/clan-cli/clan_cli/vars/_types.py +++ b/pkgs/clan-cli/clan_cli/vars/_types.py @@ -12,14 +12,36 @@ class Var: store: "StoreBase" generator: str name: str + id: str secret: bool shared: bool deployed: bool + @property + def value(self) -> bytes: + if not self.store.exists(self.generator, self.name, self.shared): + msg = f"Var {self.id} has not been generated yet" + raise ValueError(msg) + # try decode the value or return + return self.store.get(self.generator, self.name, self.shared) + + @property + def printable_value(self) -> str: + try: + return self.value.decode() + except UnicodeDecodeError: + return "" + + @property + def exists(self) -> bool: + return self.store.exists(self.generator, self.name, self.shared) + def __str__(self) -> str: if self.secret: - return f"{self.generator}/{self.name}: ********" - return f"{self.generator}/{self.name}: {self.store.get(self.generator, self.name, self.shared).decode()}" + return f"{self.id}: ********" + if self.store.exists(self.generator, self.name, self.shared): + return f"{self.id}: {self.printable_value}" + return f"{self.id}: " class StoreBase(ABC): @@ -113,6 +135,7 @@ class StoreBase(ABC): store=self, generator=gen_name, name=f_name, + id=f"{gen_name}/{f_name}", secret=file["secret"], shared=generator["share"], deployed=file["deploy"], diff --git a/pkgs/clan-cli/clan_cli/vars/get.py b/pkgs/clan-cli/clan_cli/vars/get.py new file mode 100644 index 000000000..6bb37ea27 --- /dev/null +++ b/pkgs/clan-cli/clan_cli/vars/get.py @@ -0,0 +1,74 @@ +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.errors import ClanError +from clan_cli.machines.machines import Machine + +from ._types import Var +from .list import all_vars + +log = logging.getLogger(__name__) + + +def get_var(machine: Machine, var_id: str) -> Var | None: + vars_ = all_vars(machine) + results = [] + for var in vars_: + if var_id in var.id: + results.append(var) + if len(results) == 0: + return None + if len(results) > 1: + error = ( + f"Found multiple vars for {var_id}:\n - " + + "\n - ".join([str(var) for var in results]) + + "\nBe more specific." + ) + 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}" + raise ClanError(msg) + + +def get_command( + machine: str, var_id: str, flake: FlakeId, quiet: bool, **kwargs: dict +) -> 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: + sys.stdout.buffer.write(var.value) + else: + print(f"{var.id}: {var.printable_value}") + + +def register_get_parser(parser: argparse.ArgumentParser) -> None: + machines_arg = parser.add_argument( + "machine", + help="The machine to print vars for", + ) + add_dynamic_completer(machines_arg, complete_machines) + + parser.add_argument( + "var_id", + 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))) diff --git a/pkgs/clan-cli/clan_cli/vars/list.py b/pkgs/clan-cli/clan_cli/vars/list.py index 24efb23fd..58d305e74 100644 --- a/pkgs/clan-cli/clan_cli/vars/list.py +++ b/pkgs/clan-cli/clan_cli/vars/list.py @@ -11,7 +11,7 @@ log = logging.getLogger(__name__) # TODO get also secret facts -def get_all_vars(machine: Machine) -> list[Var]: +def all_vars(machine: Machine) -> list[Var]: public_vars_module = importlib.import_module(machine.public_vars_module) public_vars_store = public_vars_module.FactStore(machine=machine) secret_vars_module = importlib.import_module(machine.secret_vars_module) @@ -24,19 +24,19 @@ def stringify_vars(_vars: list[Var]) -> str: def stringify_all_vars(machine: Machine) -> str: - return stringify_vars(get_all_vars(machine)) + return stringify_vars(all_vars(machine)) -def get_command(args: argparse.Namespace) -> None: +def list_command(args: argparse.Namespace) -> None: machine = Machine(name=args.machine, flake=args.flake) print(stringify_all_vars(machine)) def register_list_parser(parser: argparse.ArgumentParser) -> None: - machines_parser = parser.add_argument( + machines_arg = parser.add_argument( "machine", - help="The machine to print facts for", + help="The machine to print vars for", ) - add_dynamic_completer(machines_parser, complete_machines) + add_dynamic_completer(machines_arg, complete_machines) - parser.set_defaults(func=get_command) + parser.set_defaults(func=list_command)