Add cleanup secrets step

Author: Jim Robinson

croudtech_bootstrap_app/bootstrap.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+from ast import For
 
 from typing import TYPE_CHECKING, Any
 if TYPE_CHECKING:
@@ -235,6 +236,15 @@ def parse_value(self, value):
             parsed_value = value
         return str(parsed_value).strip()
 
+    def cleanup_secrets(self):
+        local_secret_keys = self.local_secrets.keys()
+        remote_secret_keys = self.remote_secret_records.keys()
+        orphaned_secrets = [item for item in remote_secret_keys if item not in local_secret_keys]
+        for secret in orphaned_secrets:
+            secret_record = self.remote_secrets[secret]           
+            self.secrets_client.delete_secret(SecretId=secret_record["ARN"], ForceDeleteWithoutRecovery=True)
+            logger.info(f"Deleted orphaned secret {secret_record['ARN']}")
+
     @property
     def local_secrets(self) -> typing.Dict[str, Any]:
         if not hasattr(self, "_secrets"):
@@ -266,6 +276,13 @@ def remote_secrets(self) -> typing.Dict[str, Any]:
 
         return self._remote_secrets
 
+    @property
+    def remote_secret_records(self) -> typing.Dict[str, Any]:
+        if not hasattr(self, "_remote_secrets"):
+            self._remote_secrets = self.get_remote_secret_records()
+                
+        return self._remote_secrets
+
     @property
     def remote_values(self) -> typing.Dict[str, Any]:
         if not hasattr(self, "_remote_values"):
@@ -333,11 +350,9 @@ def fetch_secret_value(self, secret):
             return ""
         return response["SecretString"]
 
-    def get_remote_secrets(self) -> typing.Dict[str, str]:
-        paginator = self.secrets_client.get_paginator('list_secrets')
-        secrets = {}
-        response = paginator.paginate(
-            Filters=[
+    @property
+    def remote_secret_filters(self):
+        return [
                 {
                     'Key': 'tag-key',
                     'Values': [
@@ -362,7 +377,13 @@ def get_remote_secrets(self) -> typing.Dict[str, str]:
                         self.name
                     ]
                 },
-            ],
+            ]
+
+    def get_remote_secrets(self) -> typing.Dict[str, str]:
+        paginator = self.secrets_client.get_paginator('list_secrets')
+        secrets = {}
+        response = paginator.paginate(
+            Filters=self.remote_secret_filters,
         )
         for page in response:
             for secret in page["SecretList"]:
@@ -371,6 +392,19 @@ def get_remote_secrets(self) -> typing.Dict[str, str]:
 
         return secrets
 
+    def get_remote_secret_records(self):
+        paginator = self.secrets_client.get_paginator('list_secrets')
+        secrets = {}
+        response = paginator.paginate(
+            Filters=self.remote_secret_filters,
+        )
+        for page in response:
+            for secret in page["SecretList"]:
+                secret_key = os.path.split(secret["Name"])[-1]
+                secrets[secret_key] = secret
+
+        return secrets
+
     def convert_flatten(self, d, parent_key="", sep="_"):
         items = []
         if isinstance(d, dict):
@@ -481,12 +515,18 @@ def initBootstrap(self):
 
     def put_config(self, delete_first):
         self.cleanup_values()
+        self.cleanup_secrets()
         for environment_name, environment in self.environments.items():
             for app_name, app in environment.apps.items():
                 # pass
                 app.upload_to_s3()
                 app.push_secrets()
 
+    def cleanup_secrets(self):
+        for environment_name, environment in self.environments.items():
+            for app_name, app in environment.apps.items():
+                app.cleanup_secrets()
+
     @property
     def environments(self) -> typing.Dict[str, BootstrapEnvironment]:
         if not hasattr(self, '_environments'):

croudtech_bootstrap_app/cli.py

@@ -156,6 +156,30 @@ def put_config(ctx, prefix, region, delete_first, values_path):
 
     bootstrap_manager.put_config(delete_first=delete_first)
 
+@cli.command()
+@click.pass_context
+@click.option("--prefix", default="/appconfig", help="The path prefix")
+@click.option("--region", default="eu-west-2", help="The AWS region")
+@click.option(
+    "--delete-first",
+    is_flag=True,
+    default=False,
+    help="Delete the values in this path before pushing (useful for cleanup)",
+)
+@click.argument("values_path")
+def cleanup_secrets(ctx, prefix, region, delete_first, values_path):
+    bootstrap_manager = BootstrapManager(
+        prefix=prefix,
+        region=region,
+        click=click,
+        values_path=values_path,
+        bucket_name=ctx.obj["BUCKET_NAME"],
+        endpoint_url=ctx.obj["AWS_ENDPOINT_URL"],
+    )
+
+    bootstrap_manager.cleanup_secrets()
+
+
 @cli.command()
 @click.pass_context
 @click.option("--prefix", default="/appconfig", help="The path prefix")