SSM to Vault Migrator

migrator.py

import boto3
import hvac
import sys
import pysos
import time
import os
import argparse

def fetch_from_ssm(path):
    session = boto3.Session(profile_name='shared-services')
    client = session.client('ssm')
    envs = pysos.Dict('envs') 
    secrets = pysos.Dict('secrets')

    print("Initializing paginator for SSM...")
    p = client.get_paginator('describe_parameters')
    
    try:
        print("Fetching data from SSM path:", path)
        paginator = p.paginate().build_full_result()
    except Exception as e:
        print("Error fetching parameters from SSM:", str(e))
        exit(0)

    print(f"Total parameters retrieved: {len(paginator['Parameters'])}")

    data_to_migrate = []
    for page in paginator['Parameters']:
        if page['Name'].startswith(path):
            try:
                response = client.get_parameter(Name=page['Name'])
                value = response['Parameter']['Value']
                type = response['Parameter']['Type']
                if type == 'String':
                    envs[page['Name']] = value 
                elif type == 'SecureString':
                    secrets[page['Name']] = value
                else:
                    print(f"type: {type} is not supported")
                    exit(0)
                data_to_migrate.append(page['Name'])
            except Exception as e:
                print(f"Error fetching parameter {page['Name']}: {str(e)}")

    print(f"Parameters to migrate: {len(data_to_migrate)}")
    return envs, secrets, data_to_migrate

def migrate_to_vault(kv, prefix, envs, secrets):
    VAULT_TOKEN = os.environ['VAULT_TOKEN']
    VAULT_ADDR = os.environ['VAULT_ADDR']
  
    client = hvac.Client(url=VAULT_ADDR, token=VAULT_TOKEN, verify=False)
    
    for key, val in envs.items():
        new_key = key.replace(args.path, "").lstrip("/")
        path = f"{prefix}/{new_key}"
        client.secrets.kv.v2.create_or_update_secret(
            mount_point=kv,
            path=path,
            secret=dict(value=val),
        )
        time.sleep(0.01)
        
    for key, val in secrets.items():
        new_key = key.replace(args.path, "").lstrip("/")
        path = f"{prefix}/{new_key}"
        client.secrets.kv.v2.create_or_update_secret(
            mount_point=kv,
            path=path,
            secret=dict(value=val),
        )
        time.sleep(0.01)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Migrate SSM data to Vault")
    parser.add_argument("path", help="SSM path to migrate (e.g. /develop/miestructura)")
    parser.add_argument("kv", help="Vault KV where the data should be migrated")
    parser.add_argument("--prefix", help="Optional prefix for KV structure in Vault", default="")
    parser.add_argument("--apply", help="Apply the migration after the preview", action="store_true")

    args = parser.parse_args()

    envs, secrets, data_to_migrate = fetch_from_ssm(args.path)

    if not data_to_migrate:
        print("No data found for migration.")
        sys.exit(0)

    print("\nResumen de la migración:")
    for item in data_to_migrate:
        new_item = item.replace(args.path, "").lstrip("/")
        print(f"- {args.prefix}/{new_item}")

    if args.apply:
        migrate_to_vault(args.kv, args.prefix, envs, secrets)
        print("Migración completada.")
    else:
        print("\nEjecuta con el flag --apply para realizar la migración.")

requirements.txt

boto3==1.26.4
botocore==1.29.4
chardet==5.0.0
jmespath==1.0.1
pysos==1.2.7
python-dateutil==2.8.2
s3transfer==0.6.0
six==1.16.0
urllib3==1.26.12