Export and Import SSM Parameters To/From CSV files

example.csv

get_parameterstore_to_csv.py

import argparse
import re
import sys

import boto3
from botocore.exceptions import ClientError

import csv


class ParameterExporter(object):
    def __init__(self):
        self.target_file = None
        self.source_profile = None
        self.source_region = None
        self.source_ssm = None
        self.source_sts = None

    @staticmethod
    def connect_to(profile, region):
        kwargs = {}
        if profile is not None:
            kwargs["profile_name"] = profile
        if region is not None:
            kwargs["region_name"] = region
        return boto3.Session(**kwargs)

    def connect_to_source(self, profile, region):
        self.source_ssm = self.connect_to(profile, region).client("ssm")
        self.source_sts = self.connect_to(profile, region).client("sts")

    def load_source_parameters(self, arg, recursive, one_level):
        result = {}
        paginator = self.source_ssm.get_paginator("describe_parameters")
        kwargs = {}
        if recursive or one_level:
            option = "Recursive" if recursive else "OneLevel"
            kwargs["ParameterFilters"] = [
                {"Key": "Path", "Option": option, "Values": [arg]}
            ]
        else:
            kwargs["ParameterFilters"] = [
                {"Key": "Name", "Option": "Equals", "Values": [arg]}
            ]

        for page in paginator.paginate(**kwargs):
            for parameter in page["Parameters"]:
                result[parameter["Name"]] = parameter

        if len(result) == 0:
            sys.stderr.write("ERROR: {} not found.\n".format(arg))
            sys.exit(1)
        return result

    def export(
        self,
        args,
        recursive,
        one_level
    ):
        export_parameters = []

        with open(self.target_file, 'w', newline='') as file:
            fieldnames = ['Name', 'Type','KeyId','LastModifiedDate','LastModifiedUser','Description','Value','AllowedPattern','Tier','Version','Labels','Policies','DataType']
            writer = csv.DictWriter(file, fieldnames=fieldnames)
            writer.writeheader()
            for arg in args:
                parameters = self.load_source_parameters(arg, recursive, one_level)
                for name in parameters:
                    value = self.source_ssm.get_parameter(Name=name, WithDecryption=True)
                    parameter = parameters[name]
                    parameter["Value"] = value["Parameter"]["Value"]

                    if "LastModifiedDate" in parameter:
                        del parameter["LastModifiedDate"]
                    if "LastModifiedUser" in parameter:
                        del parameter["LastModifiedUser"]
                    if "Version" in parameter:
                        del parameter["Version"]
                    if "Policies" in parameter:
                        if not parameter["Policies"]:
                            # an empty policies list causes an exception
                            del parameter["Policies"]

                    writer.writerow(parameter)

    def main(self):
        parser = argparse.ArgumentParser(description="copy parameter store ")
        parser.add_argument(
            "--one-level",
            "-1",
            dest="one_level",
            action="store_true",
            help="one-level copy",
        )
        parser.add_argument(
            "--recursive",
            "-r",
            dest="recursive",
            action="store_true",
            help="recursive copy",
        )
        parser.add_argument(
            "--source-region",
            dest="source_region",
            help="to get the parameters from ",
            metavar="AWS::Region",
        )
        parser.add_argument(
            "--source-profile",
            dest="source_profile",
            help="to obtain the parameters from",
            metavar="NAME",
        )
        parser.add_argument(
            "--file",
            dest="target_file",
            help="The file path to export to",
        )
        parser.add_argument(
            "parameters", metavar="PARAMETER", type=str, nargs="+", help="source path"
        )
        options = parser.parse_args()

        try:
            self.connect_to_source(options.source_profile, options.source_region)
            self.target_file = options.target_file
            self.export(
                options.parameters,
                options.recursive,
                options.one_level
            )
        except ClientError as e:
            sys.stderr.write("ERROR: {}\n".format(e))
            sys.exit(1)


def main():
    cp = ParameterExporter()
    cp.main()

if __name__ == "__main__":
    main()

set_parameterstore_from_csv.py

import argparse
import re
import sys

import boto3
from botocore.exceptions import ClientError

import csv


class ParameterImporter(object):
    def __init__(self):
        self.source_file = None
        self.target_profile = None
        self.target_region = None
        self.target_ssm = None
        self.target_sts = None

    @staticmethod
    def connect_to(profile, region):
        kwargs = {}
        if profile is not None:
            kwargs["profile_name"] = profile
        if region is not None:
            kwargs["region_name"] = region
        return boto3.Session(**kwargs)

    def connect_to_target(self, profile, region):
        self.target_ssm = self.connect_to(profile, region).client("ssm")
        self.target_sts = self.connect_to(profile, region).client("sts")

    def import_params (
        self,
        overwrite,
        keep_going=False,
        key_id=None,
        clear_kms_key=False
    ):
        export_parameters = []

        with open(self.source_file, 'r') as file:
            csv_file = csv.DictReader(file)
            for parameter in csv_file:
                name = parameter["Name"]
                if self.dry_run:
                    sys.stdout.write(f"DRY-RUN: importing {name} \n")
                else:
                    try:
                        if "KeyId" in parameter and key_id is not None:
                            parameter["KeyId"] = key_id
                        if "KeyId" in parameter and clear_kms_key:
                            del parameter["KeyId"]
                        if "KeyId" in parameter:
                            if not parameter["KeyId"]:
                                del parameter["KeyId"]
                        if "LastModifiedDate" in parameter:
                            del parameter["LastModifiedDate"]
                        if "LastModifiedUser" in parameter:
                            del parameter["LastModifiedUser"]
                        if "Version" in parameter:
                            del parameter["Version"]
                        if "Policies" in parameter:
                            if not parameter["Policies"]:
                                # an empty policies list causes an exception
                                del parameter["Policies"]
                        parameter["Overwrite"] = overwrite
                        self.target_ssm.put_parameter(**parameter)
                        sys.stdout.write(f"INFO: import {name} \n")
                    except self.target_ssm.exceptions.ParameterAlreadyExists as e:
                        if not keep_going:
                            sys.stderr.write(
                                f"ERROR: failed to import {name} as it already exists: specify --overwrite or --keep-going\n"
                            )
                            exit(1)
                        else:
                            sys.stderr.write(
                                f"WARN: skipping import {name} already exists\n"
                            )
                    except ClientError as e:
                        msg = e.response["Error"]["Message"]
                        sys.stderr.write(
                            f"ERROR: failed to import {name} , {msg}\n"
                        )
                        if not keep_going:
                            exit(1)

    def main(self):
        parser = argparse.ArgumentParser(description="import parameter store ")
        group = parser.add_mutually_exclusive_group()
        group.add_argument(
            "--overwrite",
            "-f",
            dest="overwrite",
            action="store_true",
            help="existing values",
        )
        group.add_argument(
            "--keep-going",
            "-k",
            dest="keep_going",
            action="store_true",
            help="as much as possible after an error",
        )
        parser.add_argument(
            "--dry-run",
            "-N",
            dest="dry_run",
            action="store_true",
            help="only show what is to be import",
        )
        parser.add_argument(
            "--region",
            dest="target_region",
            help="to import the parameters to ",
            metavar="AWS::Region",
        )
        parser.add_argument(
            "--profile",
            dest="target_profile",
            help="to import the parameters to",
            metavar="NAME",
        )
        parser.add_argument(
            "--file",
            dest="source_file",
            help="The file path to export to",
        )
        
        key_group = parser.add_mutually_exclusive_group()
        key_group.add_argument(
            "--key-id",
            dest="key_id",
            help="to use for parameter values in the destination",
            metavar="ID",
        )
        key_group.add_argument(
            "--clear-key-id",
            "-C",
            dest="clear_key_id",
            action="store_true",
            help="clear the KMS key id associated with the parameter",
        )
        options = parser.parse_args()

        try:
            self.connect_to_target(options.target_profile, options.target_region)
            self.source_file = options.source_file
            self.dry_run = options.dry_run
            self.import_params(
                options.overwrite,
                options.keep_going,
                options.key_id,
                options.clear_key_id
            )
        except ClientError as e:
            sys.stderr.write("ERROR: {}\n".format(e))
            sys.exit(1)


def main():
    cp = ParameterImporter()
    cp.main()

if __name__ == "__main__":
    main()

This is largely based on https://github.com/binxio/aws-ssm-copy/blob/master/aws_ssm_copy/ssm_copy.py but I wanted to be able to view everything in Excel so I could update values more methodically rather than relying on renaming rules and other similar automation.

Our workflow is to develop against a QA variables when running locally during early development so values get added organically as we build out features. I found this caused some problems when we wanted to first promote the system up to higher environments as we often had a number of environment specific secrets and connection strings we wanted to update. This allows you to export the existing QA values and edit them locally to replace them with secrets and reupload them into a different AWS account.

1. Run the export command
   python get_parameterstore_to_csv.py --source-profile Account1_Profile --source-region us-east-1 --recursive /applications/qa/<our application>/ --file ourapplication.csv
2. Manipulate the CSV for new account/environment
3. Run the import command
   python set_parameterstore_from_csv.py --file ourapplication.csv --profile Account2_Profile --region us-east-1 --keep-going