Page MenuHomeDevCentral

vault-migrate.py

Authored By
dereckson
Jan 28 2024, 19:30
Size
1 KB
Referenced Files
None
Subscribers
None

vault-migrate.py

#!/usr/bin/env python3
import os
import sys
import hvac
CA_CERT = "/usr/local/share/certs/nasqueron-vault-ca.crt"
# -------------------------------------------------------------
# Vault migration
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
class VaultMigration:
def __init__(self, mount_point="secret", prefix = ""):
self.mount_point = mount_point
self.prefix = prefix
vault_url = os.getenv("VAULT_ADDR", "http://localhost:8200")
vault_token = os.getenv("VAULT_TOKEN", "")
self.client = hvac.Client(url=vault_url, token=vault_token, verify=CA_CERT)
def migrate(self, orig, dest):
response = self.client.secrets.kv.v2.read_secret_version(
mount_point=self.mount_point,
path=f"{self.prefix}{orig}",
)
if response is None or "data" not in response:
print(f"Error: Unable to read secret at path '{self.prefix}{orig}'", file=sys.stderr)
return
secret = response["data"]
if secret["metadata"]["custom_metadata"]:
print(f"{self.prefix}{dest}")
def migrate_all(mount_point, prefix, orig_list, dest_list):
migration = VaultMigration(mount_point, prefix)
for orig, dest in zip(orig_list, dest_list):
migration.migrate(orig, dest)
# -------------------------------------------------------------
# Application entry point
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
def run(mount_point, prefix, orig_file, dest_file):
orig_list = [line.strip() for line in open(orig_file)]
dest_list = [line.strip() for line in open(dest_file)]
migrate_all(mount_point, prefix, orig_list, dest_list)
if __name__ == "__main__":
run("ops", "secrets/", "from", "to")

File Metadata

Mime Type
text/plain; charset=utf-8
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1639359
Default Alt Text
vault-migrate.py (1 KB)

Event Timeline