Secrets Migration: Vault KV to Azure Key Vault Secrets¶

import hvac
import json

client = hvac.Client(url='https://vault.example.com:8200', token='s.xxxxx')

def export_all_secrets(mount_point='secret', path=''):
    """Recursively export all secrets from a KV v2 mount."""
    secrets = {}
    try:
        list_response = client.secrets.kv.v2.list_secrets(
            path=path, mount_point=mount_point
        )
        keys = list_response['data']['keys']
    except Exception:
        return secrets

    for key in keys:
        full_path = f"{path}{key}"
        if key.endswith('/'):
            # Directory -- recurse
            secrets.update(export_all_secrets(mount_point, full_path))
        else:
            # Secret -- read
            try:
                response = client.secrets.kv.v2.read_secret_version(
                    path=full_path, mount_point=mount_point
                )
                secrets[full_path] = {
                    'data': response['data']['data'],
                    'metadata': response['data']['metadata']
                }
            except Exception as e:
                print(f"Error reading {full_path}: {e}")

    return secrets

all_secrets = export_all_secrets()
with open('vault-secrets-export.json', 'w') as f:
    json.dump(all_secrets, f, indent=2)
print(f"Exported {len(all_secrets)} secrets")

from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
import json
from datetime import datetime, timedelta, timezone

credential = DefaultAzureCredential()
vault_url = "https://kv-secrets-prod.vault.azure.net"
client = SecretClient(vault_url=vault_url, credential=credential)

with open('vault-secrets-export.json', 'r') as f:
    secrets = json.load(f)

for vault_path, secret_info in secrets.items():
    # Convert Vault path to Key Vault name
    kv_name = vault_path.replace('/', '-').strip('-')
    # Key Vault names: alphanumeric and hyphens, 1-127 chars
    kv_name = ''.join(c if c.isalnum() or c == '-' else '-' for c in kv_name)[:127]

    data = secret_info['data']

    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 (optional -- 1 year default for migrated secrets)
    expires_on = datetime.now(timezone.utc) + timedelta(days=365)

    client.set_secret(
        name=kv_name,
        value=value,
        content_type=content_type,
        expires_on=expires_on,
        tags={
            "migrated-from": "vault",
            "vault-path": vault_path,
            "migration-date": datetime.now(timezone.utc).isoformat()
        }
    )
    print(f"Imported: {vault_path} -> {kv_name}")

print(f"Migration complete: {len(secrets)} secrets imported")

import azure.functions as func
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
import json

def main(event: func.EventGridEvent):
    """Rotate a secret when near-expiry event fires."""
    data = event.get_json()
    secret_name = data['ObjectName']
    vault_uri = data['VaultName']

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

    # Read current secret to determine rotation logic
    current = client.get_secret(secret_name)
    secret_type = current.properties.tags.get('rotation-type', 'manual')

    if secret_type == 'api-key':
        new_value = rotate_api_key(current.value)
    elif secret_type == 'database-password':
        new_value = rotate_database_password(current.value)
    else:
        # Manual rotation -- send alert
        send_rotation_alert(secret_name, vault_uri)
        return

    # Set new secret version with updated expiration
    from datetime import datetime, timedelta, timezone
    client.set_secret(
        name=secret_name,
        value=new_value,
        expires_on=datetime.now(timezone.utc) + timedelta(days=90),
        tags=current.properties.tags
    )

# Python application reading from Vault
import hvac

client = hvac.Client(url='https://vault.example.com:8200')
# Auth via AppRole
client.auth.approle.login(role_id='xxx', secret_id='yyy')

secret = client.secrets.kv.v2.read_secret_version(path='app1/db-password')
db_password = secret['data']['data']['password']

# Python application reading from Key Vault
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient

credential = DefaultAzureCredential()  # Uses managed identity automatically
client = SecretClient(
    vault_url="https://kv-secrets-prod.vault.azure.net",
    credential=credential
)

secret = client.get_secret("app1-db-password")
db_password = secret.value