Tutorial: Step-by-Step Secret Migration from Vault to Key Vault¶

#!/usr/bin/env python3
"""Generate Vault secrets inventory report."""
import hvac
import json
import os
from datetime import datetime

client = hvac.Client(
    url=os.getenv('VAULT_ADDR'),
    token=os.getenv('VAULT_TOKEN')
)

def inventory_mount(mount_point, path=''):
    """Recursively inventory secrets in a KV v2 mount."""
    inventory = []
    try:
        response = client.secrets.kv.v2.list_secrets(
            path=path, mount_point=mount_point
        )
        keys = response['data']['keys']
    except Exception:
        return inventory

    for key in keys:
        full_path = f"{path}{key}"
        if key.endswith('/'):
            inventory.extend(inventory_mount(mount_point, full_path))
        else:
            try:
                metadata = client.secrets.kv.v2.read_secret_metadata(
                    path=full_path, mount_point=mount_point
                )
                meta = metadata['data']
                inventory.append({
                    'mount': mount_point,
                    'path': full_path,
                    'versions': meta['current_version'],
                    'created': meta['created_time'],
                    'updated': meta['updated_time'],
                    'custom_metadata': meta.get('custom_metadata', {}),
                })
            except Exception as e:
                inventory.append({
                    'mount': mount_point,
                    'path': full_path,
                    'error': str(e)
                })
    return inventory

# Inventory all mounts
all_inventory = []
mounts_response = client.sys.list_mounted_secrets_engines()
for mount, config in mounts_response['data'].items():
    if config['type'] == 'kv':
        print(f"Inventorying mount: {mount}")
        mount_name = mount.rstrip('/')
        all_inventory.extend(inventory_mount(mount_name))

# Save report
report = {
    'generated_at': datetime.utcnow().isoformat(),
    'total_secrets': len(all_inventory),
    'secrets': all_inventory
}

with open('vault-inventory.json', 'w') as f:
    json.dump(report, f, indent=2)

print(f"\nInventory complete: {len(all_inventory)} secrets found")
print("Report saved to vault-inventory.json")

#!/usr/bin/env python3
"""Export secrets from HashiCorp Vault KV v2 to JSON file."""
import hvac
import json
import os
import hashlib
from datetime import datetime

client = hvac.Client(
    url=os.getenv('VAULT_ADDR'),
    token=os.getenv('VAULT_TOKEN')
)

assert client.is_authenticated(), "Vault authentication failed"

def export_secrets(mount_point='secret', path=''):
    """Recursively export all secrets with their values."""
    secrets = {}
    try:
        response = client.secrets.kv.v2.list_secrets(
            path=path, mount_point=mount_point
        )
        keys = response['data']['keys']
    except Exception:
        return secrets

    for key in keys:
        full_path = f"{path}{key}"
        if key.endswith('/'):
            secrets.update(export_secrets(mount_point, full_path))
        else:
            try:
                secret_response = client.secrets.kv.v2.read_secret_version(
                    path=full_path, mount_point=mount_point
                )
                data = secret_response['data']['data']
                metadata = secret_response['data']['metadata']

                # Hash values for verification (never log actual values)
                hashed = {k: hashlib.sha256(str(v).encode()).hexdigest()[:12]
                         for k, v in data.items()}

                secrets[full_path] = {
                    'data': data,
                    'metadata': {
                        'version': metadata['version'],
                        'created_time': metadata['created_time'],
                        'custom_metadata': metadata.get('custom_metadata', {}),
                    },
                    'value_hashes': hashed,
                }
                print(f"  Exported: {full_path} ({len(data)} keys)")

            except Exception as e:
                print(f"  ERROR: {full_path}: {e}")

    return secrets

print("Starting Vault secret export...")
print(f"Vault address: {os.getenv('VAULT_ADDR')}")

all_secrets = {}
mounts = client.sys.list_mounted_secrets_engines()
for mount, config in mounts['data'].items():
    if config['type'] == 'kv':
        mount_name = mount.rstrip('/')
        print(f"\nExporting mount: {mount_name}/")
        mount_secrets = export_secrets(mount_name)
        for path, data in mount_secrets.items():
            all_secrets[f"{mount_name}/{path}"] = data

export = {
    'exported_at': datetime.utcnow().isoformat(),
    'vault_addr': os.getenv('VAULT_ADDR'),
    'total_secrets': len(all_secrets),
    'secrets': all_secrets,
}

output_file = 'vault-export.json'
with open(output_file, 'w') as f:
    json.dump(export, f, indent=2)

print(f"\nExport complete: {len(all_secrets)} secrets written to {output_file}")
print("WARNING: This file contains plaintext secrets. Encrypt or delete after import.")

#!/usr/bin/env python3
"""Import secrets from Vault export to Azure Key Vault."""
import json
import re
import os
from datetime import datetime, timedelta, timezone
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient

# Configuration
VAULT_NAME = os.getenv('KEY_VAULT_NAME', 'kv-migrated-prod')
EXPORT_FILE = os.getenv('EXPORT_FILE', 'vault-export.json')
DEFAULT_EXPIRY_DAYS = int(os.getenv('DEFAULT_EXPIRY_DAYS', '365'))

credential = DefaultAzureCredential()
client = SecretClient(
    vault_url=f"https://{VAULT_NAME}.vault.azure.net",
    credential=credential
)

def sanitize_name(vault_path: str) -> str:
    """Convert Vault path to valid Key Vault secret name.

    Key Vault names: alphanumeric and hyphens, 1-127 chars.
    Vault paths: alphanumeric, slashes, hyphens, underscores.
    """
    name = vault_path.replace('/', '-').replace('_', '-')
    name = re.sub(r'[^a-zA-Z0-9-]', '-', name)
    name = re.sub(r'-+', '-', name)  # Collapse multiple hyphens
    name = name.strip('-')[:127]
    return name

def import_secret(vault_path: str, secret_data: dict):
    """Import a single secret to Key Vault."""
    kv_name = sanitize_name(vault_path)
    data = secret_data['data']
    metadata = secret_data.get('metadata', {})

    # Determine value and content type
    if len(data) == 1:
        value = str(list(data.values())[0])
        content_type = 'text/plain'
    else:
        value = json.dumps(data)
        content_type = 'application/json'

    # Set expiration
    expires_on = datetime.now(timezone.utc) + timedelta(days=DEFAULT_EXPIRY_DAYS)

    # Build tags
    tags = {
        'migrated-from': 'hashicorp-vault',
        'vault-path': vault_path,
        'migration-date': datetime.now(timezone.utc).strftime('%Y-%m-%d'),
        'original-version': str(metadata.get('version', 'unknown')),
    }

    # Add custom metadata as tags
    custom_meta = metadata.get('custom_metadata', {})
    for k, v in (custom_meta or {}).items():
        tag_key = f"vault-{k}"[:256]
        tags[tag_key] = str(v)[:256]

    try:
        client.set_secret(
            name=kv_name,
            value=value,
            content_type=content_type,
            expires_on=expires_on,
            tags=tags,
        )
        return True, kv_name
    except Exception as e:
        return False, str(e)

# Load export file
with open(EXPORT_FILE, 'r') as f:
    export = json.load(f)

print(f"Key Vault: {VAULT_NAME}")
print(f"Secrets to import: {export['total_secrets']}")
print(f"Default expiry: {DEFAULT_EXPIRY_DAYS} days")
print()

# Import each secret
success_count = 0
error_count = 0
mapping = []

for vault_path, secret_data in export['secrets'].items():
    ok, result = import_secret(vault_path, secret_data)
    if ok:
        success_count += 1
        mapping.append({
            'vault_path': vault_path,
            'kv_name': result,
            'status': 'imported'
        })
        print(f"  OK: {vault_path} -> {result}")
    else:
        error_count += 1
        mapping.append({
            'vault_path': vault_path,
            'error': result,
            'status': 'failed'
        })
        print(f"  FAIL: {vault_path}: {result}")

# Save mapping report
mapping_report = {
    'imported_at': datetime.now(timezone.utc).isoformat(),
    'key_vault': VAULT_NAME,
    'total': len(mapping),
    'success': success_count,
    'errors': error_count,
    'mapping': mapping,
}
with open('migration-mapping.json', 'w') as f:
    json.dump(mapping_report, f, indent=2)

print(f"\nImport complete: {success_count} imported, {error_count} errors")
print("Mapping saved to migration-mapping.json")

# Before: reading from Vault
import hvac
client = hvac.Client(url='https://vault:8200')
client.auth.approle.login(role_id='xxx', secret_id='yyy')
secret = client.secrets.kv.v2.read_secret_version(path='webapp/db-password')
password = secret['data']['data']['password']

# After: reading from Key Vault
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient

credential = DefaultAzureCredential()
client = SecretClient(vault_url="https://kv-migrated-prod.vault.azure.net",
                      credential=credential)
password = client.get_secret("webapp-prod-db-password").value

#!/usr/bin/env python3
"""Validate migration: compare Vault secrets with Key Vault secrets."""
import json
import hashlib
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient

VAULT_NAME = 'kv-migrated-prod'
MAPPING_FILE = 'migration-mapping.json'

credential = DefaultAzureCredential()
client = SecretClient(
    vault_url=f"https://{VAULT_NAME}.vault.azure.net",
    credential=credential
)

with open(MAPPING_FILE, 'r') as f:
    mapping = json.load(f)

print("Validating migration...")
valid = 0
invalid = 0

for entry in mapping['mapping']:
    if entry['status'] != 'imported':
        continue

    kv_name = entry['kv_name']
    try:
        secret = client.get_secret(kv_name)
        assert secret.value is not None, "Secret value is None"
        assert secret.properties.expires_on is not None, "No expiration set"
        assert 'migrated-from' in (secret.properties.tags or {}), "Missing migration tag"

        valid += 1
        print(f"  VALID: {kv_name}")
    except Exception as e:
        invalid += 1
        print(f"  INVALID: {kv_name}: {e}")

print(f"\nValidation complete: {valid} valid, {invalid} invalid")