feat: improve set-parameter and set-secret commands

.gitignore

@@ -8,3 +8,4 @@ coverage/
 .coverage
 venv
 .venv
+.DS_Store

aws_secrets/cli/cli.py

@@ -1,5 +1,5 @@
 import click
-import yaml
+
 from aws_secrets import __name__ as module_name
 from aws_secrets.cli.decrypt import decrypt
 from aws_secrets.cli.deploy import deploy
@@ -11,46 +11,32 @@
 from aws_secrets.cli.view_secret import view_secret
 from aws_secrets.helpers.catch_exceptions import catch_exceptions
 from aws_secrets.helpers.logging import setup_logging
-from aws_secrets.representers.literal import Literal, literal_presenter
-from aws_secrets.tags.cmd import CmdTag
-from aws_secrets.tags.file import FileTag
-from aws_secrets.tags.output_stack import OutputStackTag
-
-yaml.SafeLoader.add_constructor('!cf_output', OutputStackTag.from_yaml)
-yaml.SafeDumper.add_multi_representer(
-    OutputStackTag, OutputStackTag.to_yaml)
-
-yaml.SafeLoader.add_constructor('!cmd', CmdTag.from_yaml)
-yaml.SafeDumper.add_multi_representer(CmdTag, CmdTag.to_yaml)
-
-yaml.SafeLoader.add_constructor('!file', FileTag.from_yaml)
-yaml.SafeDumper.add_multi_representer(FileTag, FileTag.to_yaml)
-
-yaml.SafeDumper.add_representer(Literal, literal_presenter)
 
 
-@click.group(help='AWS Secrets CLI')
+@click.group(help="AWS Secrets CLI")
 @click.option(
-    '--loglevel',
-    help='Log level.',
+    "--loglevel",
+    help="Log level.",
     required=False,
-    default='WARNING',
+    default="WARNING",
     show_default=True,
-    type=click.Choice(['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], case_sensitive=False)
+    type=click.Choice(
+        ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
+    ),
 )
 @click.pass_context
 @catch_exceptions
 def cli(ctx, loglevel: str):
     """
-        Root CLI Group `aws-secrets --help`
+    Root CLI Group `aws-secrets --help`
 
-        Args:
-            ctx (`click.Context`): click context object
-            loglevel (`str`): log level
+    Args:
+        ctx (`click.Context`): click context object
+        loglevel (`str`): log level
     """
 
     ctx.ensure_object(dict)
-    ctx.obj['loglevel'] = loglevel
+    ctx.obj["loglevel"] = loglevel
 
     setup_logging(module_name, loglevel)
 

aws_secrets/cli/decrypt.py

@@ -2,10 +2,11 @@
 
 import click
 from click.core import Context
-import yaml
+
 from aws_secrets.config.config_reader import ConfigReader
 from aws_secrets.helpers.catch_exceptions import catch_exceptions
 from aws_secrets.miscellaneous import session
+from aws_secrets.yaml import yaml
 
 
 @click.command(name='decrypt', help='Decrypt a configuration file')
@@ -41,4 +42,4 @@ def decrypt(
 
     output_file = output if output else f"{env_file}.dec"
     with open(output_file, 'w') as outfile:
-        yaml.safe_dump(config.data, outfile)
+        yaml.dump(config.data, outfile)

aws_secrets/cli/set_parameter.py

@@ -1,76 +1,117 @@
 from typing import Optional
 
 import click
+from click.core import Context
+from prompt_toolkit import HTML, prompt
+
 from aws_secrets.config.config_reader import ConfigReader
 from aws_secrets.helpers.catch_exceptions import catch_exceptions
 from aws_secrets.miscellaneous import session
-from click.core import Context
+from aws_secrets.miscellaneous.prompt import (
+    DEFAULT_PROMPT_SYLE, get_multiline_input_save_keyboard, radiolist_prompt)
 
 
-@click.command(name='set-parameter', help='Add/Update SSM Parameters')
-@click.option('-e', '--env-file', type=click.Path(), required=True)
-@click.option('-n', '--name', prompt=True, required=True)
-@click.option('-d', '--description', help='SSM Parameter Description', required=False)
-@click.option('-t', '--type',
-              required=True, type=click.Choice(['String', 'SecureString'], case_sensitive=True),
-              default='SecureString')
-@click.option('-k', '--kms')
-@click.option('--profile', help="AWS Profile", envvar='AWS_PROFILE')
-@click.option('--region', help="AWS Region", envvar='AWS_REGION')
+@click.command(name="set-parameter", help="Add/Update SSM Parameters")
+@click.option("-e", "--env-file", type=click.Path(), required=True)
+@click.option("-n", "--name", required=False)
+@click.option("-d", "--description", help="SSM Parameter Description", required=False)
+@click.option("-v", "--value", help="SSM Parameter Value", required=False)
+@click.option(
+    "-t",
+    "--type",
+    required=False,
+    type=click.Choice(["String", "SecureString"], case_sensitive=True),
+)
+@click.option("-k", "--kms")
+@click.option("--profile", help="AWS Profile", envvar="AWS_PROFILE")
+@click.option("--region", help="AWS Region", envvar="AWS_REGION")
 @click.pass_context
 @catch_exceptions
 def set_parameter(
     ctx: Context,
     env_file: str,
-    name: str,
+    name: Optional[str],
     description: Optional[str],
-    type: str,
+    value: Optional[str],
+    type: Optional[str],
     kms: Optional[str],
     profile: Optional[str],
-    region: Optional[str]
+    region: Optional[str],
 ):
     """
-        Add/Update SSM Parameters CLI Commmand `aws-secrets set-parameter --help`
+    Add/Update SSM Parameters CLI Commmand `aws-secrets set-parameter --help`
 
-        Args:
-            ctx (`Context`): click context object
-            env_file (`str`): configuration YAML file
-            name (`str`): SSM parameter name
-            description (`str`, optional): SSM parameter description
-            type (`str`): SSM parameter type
-            kms (`str`, optional): SSM parameter KMS Arn or Id
-            profile (`str`, optional): AWS Profile
-            region (`str`, optional): AWS Region
+    Args:
+        ctx (`Context`): click context object
+        env_file (`str`): configuration YAML file
+        name (`str`, optional): SSM parameter name
+        description (`str`, optional): SSM parameter description
+        type (`str`, optional): SSM parameter type
+        kms (`str`, optional): SSM parameter KMS Arn or Id
+        profile (`str`, optional): AWS Profile
+        region (`str`, optional): AWS Region
     """
-    ctx.obj['config_file'] = env_file
+    ctx.obj["config_file"] = env_file
     session.aws_profile = profile
     session.aws_region = region
     config = ConfigReader(env_file)
-    provider = config.get_provider('parameters')
+    provider = config.get_provider("parameters")
 
-    context_data = {
-        'name': name,
-        'type': type,
-    }
+    context_data = {"name": name}
+    if not name:
+        context_data["name"] = prompt(
+            HTML(
+                "<question><question_mark>?</question_mark> What is the name of the SSM parameter? </question>"
+            ),
+            style=DEFAULT_PROMPT_SYLE,
+        )
+
+    parameter = provider.find(context_data["name"])
 
     if description:
-        context_data['description'] = description
+        context_data["description"] = description
+    else:
+        context_data["description"] = prompt(
+            HTML(
+                "<question><question_mark>?</question_mark> What is the description of the parameter? </question>"
+            ),
+            default=parameter.description if parameter else "",
+            style=DEFAULT_PROMPT_SYLE,
+        )
+
+    if type:
+        context_data["type"] = type
+    else:
+        context_data["type"] = radiolist_prompt(
+            title=HTML(
+                "<question><question_mark>?</question_mark> What is the type of the parameter? </question>"
+            ),
+            values=[
+                ("String", "String"),
+                ("SecureString", "SecureString"),
+            ],
+            default=parameter.type if parameter else "SecureString",
+        )
 
     if kms:
-        context_data['kms'] = kms
+        context_data["kms"] = kms
 
-    click.echo("Enter/Paste your secret. Ctrl-D or Ctrl-Z ( windows ) to save it.")
-    contents = []
-    while True:
-        try:
-            line = input()
-        except EOFError:
-            break
-        contents.append(line)
+    if not value:
+        initial_value = ""
+        if parameter is not None:
+            initial_value = parameter.decrypt(format=True)
 
-    context_data['value'] = '\n'.join(contents)
+        click.echo(
+            f"Press {get_multiline_input_save_keyboard()} or [Esc] followed by [Enter] to accept input."
+        )
+        input_value = prompt("", multiline=True, default=initial_value)
+        if input_value.strip() == "":
+            return
+
+        context_data["value"] = input_value
+    else:
+        context_data["value"] = value
 
-    parameter = provider.find(name)
     if parameter is None:
         parameter = provider.add(context_data)
     else:

aws_secrets/cli/set_secret.py

@@ -1,74 +1,108 @@
 import logging
+import textwrap
 from typing import Optional
 
 import click
+from click.core import Context
+from prompt_toolkit import HTML, prompt
+
 from aws_secrets.config.config_reader import ConfigReader
 from aws_secrets.helpers.catch_exceptions import catch_exceptions
 from aws_secrets.miscellaneous import session
-from click.core import Context
+from aws_secrets.miscellaneous.prompt import (
+    DEFAULT_PROMPT_SYLE, get_multiline_input_save_keyboard)
 
 logger = logging.getLogger(__name__)
 
 
-@click.command(name='set-secret', help='Add/Update AWS Secrets Manager secrets')
-@click.option('-e', '--env-file', type=click.Path(), required=True)
-@click.option('-n', '--name', required=True)
-@click.option('-d', '--description', help='Secret Description', required=False)
-@click.option('-k', '--kms')
-@click.option('--profile', help="AWS Profile", envvar='AWS_PROFILE')
-@click.option('--region', help="AWS Region", envvar='AWS_REGION')
+@click.command(name="set-secret", help="Add/Update AWS Secrets Manager secrets")
+@click.option("-e", "--env-file", type=click.Path(), required=True)
+@click.option("-n", "--name", required=False)
+@click.option("-d", "--description", help="Secret Description", required=False)
+@click.option("-v", "--value", help="Secret Value", required=False)
+@click.option("-k", "--kms")
+@click.option("--profile", help="AWS Profile", envvar="AWS_PROFILE")
+@click.option("--region", help="AWS Region", envvar="AWS_REGION")
 @click.pass_context
 @catch_exceptions
 def set_secret(
     ctx: Context,
     env_file: str,
-    name: str,
+    name: Optional[str],
     description: Optional[str],
+    value: Optional[str],
     kms: Optional[str],
     profile: Optional[str],
-    region: Optional[str]
+    region: Optional[str],
 ):
     """
-        Add/Update AWS Secrets Manager secrets CLI Commmand `aws-secrets set-secret --help`
-
-        Args:
-            ctx (`Context`): click context object
-            env_file (`str`): configuration YAML file
-            name (`str`): SSM parameter name
-            description (`str`, optional): SSM parameter description
-            kms (`str`, optional): SSM parameter KMS Arn or Id
-            profile (`str`, optional): AWS Profile
-            region (`str`, optional): AWS Region
+    Add/Update AWS Secrets Manager secrets CLI Commmand `aws-secrets set-secret --help`
+
+    Args:
+        ctx (`Context`): click context object
+        env_file (`str`): configuration YAML file
+        name (`str`, optional): SSM parameter name
+        description (`str`, optional): SSM parameter description
+        kms (`str`, optional): SSM parameter KMS Arn or Id
+        profile (`str`, optional): AWS Profile
+        region (`str`, optional): AWS Region
     """
-    ctx.obj['config_file'] = env_file
+    ctx.obj["config_file"] = env_file
     session.aws_profile = profile
     session.aws_region = region
 
     config = ConfigReader(env_file)
-    provider = config.get_provider('secrets')
+    provider = config.get_provider("secrets")
 
-    context_data = {
-        'name': name
-    }
+    context_data = {"name": name}
+    if not name:
+        context_data["name"] = prompt(
+            HTML(
+                "<question><question_mark>?</question_mark> What is the name of the secret? </question>"
+            ),
+            style=DEFAULT_PROMPT_SYLE,
+        )
+
+    secret = provider.find(context_data["name"])
 
     if description:
-        context_data['description'] = description
+        context_data["description"] = description
+    else:
+        context_data["description"] = prompt(
+            HTML(
+                "<question><question_mark>?</question_mark> What is the description of the secret? </question>"
+            ),
+            default=secret.description if secret else "",
+            style=DEFAULT_PROMPT_SYLE,
+        )
 
     if kms:
-        context_data['kms'] = kms
+        context_data["kms"] = kms
 
-    click.echo("Enter/Paste your secret. Ctrl-D or Ctrl-Z ( windows ) to save it.")
-    contents = []
-    while True:
-        try:
-            line = input()
-        except EOFError:
-            break
-        contents.append(line)
+    if not value:
+        initial_value = ""
+        if secret is not None:
+            initial_value = secret.decrypt(format=True)
 
-    context_data['value'] = '\n'.join(contents)
+        input_value = prompt(
+            HTML(
+                textwrap.dedent(f"""\
+                    <question><question_mark>?</question_mark> What is the value of the secret?
+                    Press {get_multiline_input_save_keyboard()} or [Esc] followed by [Enter] to accept input.
+                    </question>
+                    """)
+            ),
+            multiline=True,
+            default=initial_value,
+            style=DEFAULT_PROMPT_SYLE,
+        )
+        if input_value.strip() == "":
+            return
+
+        context_data["value"] = input_value
+    else:
+        context_data["value"] = value
 
-    secret = provider.find(name)
     if secret is None:
         secret = provider.add(context_data)
     else: