Created
October 26, 2023 15:01
-
-
Save jefrnc/01e5383c9c66610db13b6a842d355cb4 to your computer and use it in GitHub Desktop.
SSM to Vault Migrator
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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.") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment