diff --git a/poetry.lock b/poetry.lock index e354833..30a8d20 100644 --- a/poetry.lock +++ b/poetry.lock @@ -307,6 +307,17 @@ botocore = ">=1.33.2,<2.0a.0" [package.extras] crt = ["botocore[crt] (>=1.33.2,<2.0a.0)"] +[[package]] +name = "shellingham" +version = "1.5.4" +description = "Tool to Detect Surrounding Shell" +optional = false +python-versions = ">=3.7" +files = [ + {file = "shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686"}, + {file = "shellingham-1.5.4.tar.gz", hash = "sha256:8dbca0739d487e5bd35ab3ca4b36e11c4078f3a234bfce294b0a0291363404de"}, +] + [[package]] name = "six" version = "1.16.0" @@ -370,4 +381,4 @@ zstd = ["zstandard (>=0.18.0)"] [metadata] lock-version = "2.0" python-versions = "^3.12" -content-hash = "0b6e75cdeffab9ca7e4872fcd6770afc7295cd0e8cfdeab9fd1972e0336722cb" +content-hash = "e1b4fa843328b0c8f23ac0ea869f40808ae2912e99ba30bde8ca0d79b32d17ab" diff --git a/pyproject.toml b/pyproject.toml index 5bbeb13..61c9372 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ python = "^3.12" typer = "^0.9.0" boto3 = "^1.33.5" rich = "^13.7.0" +shellingham = "^1.5.4" [tool.poetry.group.dev.dependencies] mypy = "^1.7.1" diff --git a/ssm.py b/ssm.py index 808334c..6818b16 100644 --- a/ssm.py +++ b/ssm.py @@ -1,48 +1,126 @@ +from __future__ import annotations + +import os import pathlib +import sys import typing import boto3 +import botocore.client +import botocore.session import rich.console import rich.table import typer -from botocore.session import ProfileNotFound app = typer.Typer() stdout = rich.console.Console() -stderr = rich.console.Console(stderr=True, style="bold red") +stderr = rich.console.Console(stderr=True) +aws_profiles = boto3.session.Session().available_profiles + + +class SSMPath(pathlib.PurePosixPath): + def __init__(self, *segments: str | os.PathLike): + super().__init__("/", *segments) + + @staticmethod + def parse(value: str) -> SSMPath: + return SSMPath(value) + + +ProfileArg = typing.Annotated[str, typer.Argument(autocompletion=aws_profiles)] + + +def get_client(profile: str) -> botocore.client.BaseClient: + try: + session = boto3.Session(profile_name=profile) + return session.client("ssm") + except botocore.session.ProfileNotFound: + stderr.print(f"Invalid profile '{profile}'", style="bold red") + sys.exit(1) @app.command() def profiles() -> None: - for profile in boto3.session.Session().available_profiles: + """List available AWS profiles.""" + for profile in aws_profiles: print(profile) @app.command() -def list(profile: str, path: str, recursive: bool = True) -> None: - try: - session = boto3.Session(profile_name=profile) - except ProfileNotFound: - stderr.print(f"Invalid profile '{profile}'") - return - client = session.client("ssm") +def list( + profile: ProfileArg, + path: pathlib.Path, + recursive: bool = True, +) -> None: + """List parameters and their values at a requested PATH.""" + root = SSMPath(path) + client = get_client(profile) console = rich.console.Console() - console.print("SSM Parameters") - table = rich.table.Table("Name", "Value", "Type") + table = rich.table.Table( + "Name", + "Value", + "Description", + "Type", + title=f"{root} ({profile})", + ) results = client.get_paginator("get_parameters_by_path").paginate( - Path=str(pathlib.PurePosixPath("/") / path), + Path=str(root), Recursive=recursive, WithDecryption=True, ) for parameter in ( result for chunk in results for result in chunk.get("Parameters") ): - if parameter.get("Name"): + if name := parameter.get("Name"): table.add_row( - parameter.get("Name"), parameter.get("Value"), parameter.get("Type") + str(pathlib.Path(name).relative_to(root)), + parameter.get("Value"), + parameter.get("Description"), + parameter.get("Type"), ) console.print(table) +@app.command() +def set( + profile: ProfileArg, + path: str, + value: str, + secure: bool = False, + overwrite: bool = False, + description: typing.Annotated[str, typer.Option] = "", +) -> None: + """Set a parameter at PATH to VALUE. + + If --secure is used, it will be stored as a SecureString. + """ + client = get_client(profile) + try: + client.put_parameter( + Name=str(SSMPath(path)), + Value=value, + Type="SecureString" if secure else "String", + Overwrite=overwrite, + Description=description, + ) + except client.exceptions.ParameterAlreadyExists: + stderr.print( + "Parameter already exists; use --overwrite to replace it.", + style="bold red", + ) + sys.exit(1) + + +@app.command() +def unset(profile: ProfileArg, path: str) -> None: + """Remove a parameter at PATH.""" + client = get_client(profile) + try: + client.delete_parameter(Name=str(SSMPath(path))) + except client.exceptions.ParameterNotFound: + stderr.print("Parameter not found", style="yellow") + sys.exit(1) + + if __name__ == "__main__": app()