vault-migrate.py
ActivePublic
Actions

Authored by dereckson on Jan 28 2024, 19:30.

Tags

None

Referenced Files

	F2582512: vault-migrate.py
	Jan 28 2024, 19:30

Subscribers

None

	#!/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")

Event Timeline

dereckson created this paste.Jan 28 2024, 19:30

vault-migrate.pyActivePublicActions

Event Timeline

vault-migrate.py
ActivePublic
Actions