Dynamic Secrets Migration: Vault Database Engine to Azure Managed Identity¶

import hvac
import pyodbc

# Get dynamic credentials from Vault
vault_client = hvac.Client(url='https://vault:8200')
vault_client.auth.approle.login(role_id='xxx', secret_id='yyy')
creds = vault_client.secrets.database.generate_credentials('webapp-sql-role')

username = creds['data']['username']
password = creds['data']['password']

conn = pyodbc.connect(
    f"DRIVER={{ODBC Driver 18 for SQL Server}};"
    f"SERVER=sql-prod.database.windows.net;"
    f"DATABASE=appdb;"
    f"UID={username};"
    f"PWD={password};"
)

from azure.identity import DefaultAzureCredential
import pyodbc
import struct

credential = DefaultAzureCredential()
token = credential.get_token("https://database.windows.net/.default")

# Pack token for pyodbc
token_bytes = token.token.encode("UTF-16-LE")
token_struct = struct.pack(f'<I{len(token_bytes)}s', len(token_bytes), token_bytes)
SQL_COPT_SS_ACCESS_TOKEN = 1256

conn = pyodbc.connect(
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=sql-prod.database.windows.net;"
    "DATABASE=appdb;",
    attrs_before={SQL_COPT_SS_ACCESS_TOKEN: token_struct}
)

# No username. No password. No Vault. No rotation.

from azure.identity import DefaultAzureCredential
from sqlalchemy import create_engine, event

credential = DefaultAzureCredential()

engine = create_engine(
    "mssql+pyodbc://sql-prod.database.windows.net/appdb"
    "?driver=ODBC+Driver+18+for+SQL+Server"
    "&Encrypt=yes"
    "&TrustServerCertificate=no"
)

@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
    token = credential.get_token("https://database.windows.net/.default")
    token_bytes = token.token.encode("UTF-16-LE")
    token_struct = struct.pack(f'<I{len(token_bytes)}s', len(token_bytes), token_bytes)
    cparams["attrs_before"] = {1256: token_struct}
    # Remove uid/pwd if present
    cparams.pop("user", None)
    cparams.pop("password", None)

import hvac
import psycopg2

vault_client = hvac.Client(url='https://vault:8200')
vault_client.auth.kubernetes.login(role='webapp', jwt=service_account_token)
creds = vault_client.secrets.database.generate_credentials('webapp-pg-role')

conn = psycopg2.connect(
    host='pg-prod.postgres.database.azure.com',
    database='appdb',
    user=creds['data']['username'],
    password=creds['data']['password'],
    sslmode='require'
)

from azure.identity import DefaultAzureCredential
import psycopg2

credential = DefaultAzureCredential()
token = credential.get_token("https://ossrdbms-aad.database.windows.net/.default")

conn = psycopg2.connect(
    host='pg-prod.postgres.database.azure.com',
    database='appdb',
    user='mi-webapp-prod',  # Managed identity name
    password=token.token,   # OAuth2 access token as password
    sslmode='require'
)

# No Vault. Token auto-refreshes via DefaultAzureCredential.

from azure.identity.aio import DefaultAzureCredential
import asyncpg

async def get_connection():
    credential = DefaultAzureCredential()
    token = await credential.get_token("https://ossrdbms-aad.database.windows.net/.default")

    conn = await asyncpg.connect(
        host='pg-prod.postgres.database.azure.com',
        database='appdb',
        user='mi-webapp-prod',
        password=token.token,
        ssl='require'
    )
    return conn

import hvac
from azure.cosmos import CosmosClient

vault_client = hvac.Client(url='https://vault:8200')
vault_client.auth.approle.login(role_id='xxx', secret_id='yyy')
secret = vault_client.secrets.kv.v2.read_secret_version(path='cosmos/prod')

client = CosmosClient(
    url='https://cosmos-prod.documents.azure.com:443/',
    credential=secret['data']['data']['primary_key']
)

from azure.identity import DefaultAzureCredential
from azure.cosmos import CosmosClient

credential = DefaultAzureCredential()

client = CosmosClient(
    url='https://cosmos-prod.documents.azure.com:443/',
    credential=credential  # Managed identity - no key needed
)

database = client.get_database_client('appdb')
container = database.get_container_client('items')

# No primary key. No Vault. Entra RBAC controls access.

# Azure Function for automated database password rotation
import azure.functions as func
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
import secrets
import psycopg2

def main(event: func.EventGridEvent):
    data = event.get_json()
    secret_name = data['ObjectName']

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

    # Generate new password
    new_password = secrets.token_urlsafe(32)

    # Update password on the database
    current_secret = kv_client.get_secret(secret_name)
    current_config = json.loads(current_secret.value)

    conn = psycopg2.connect(
        host=current_config['host'],
        user=current_config['admin_user'],
        password=current_config['admin_password'],
        database='postgres'
    )
    with conn.cursor() as cur:
        cur.execute(f"ALTER USER {current_config['app_user']} PASSWORD %s", (new_password,))
    conn.commit()
    conn.close()

    # Update Key Vault secret
    current_config['password'] = new_password
    kv_client.set_secret(
        name=secret_name,
        value=json.dumps(current_config),
        content_type="application/json",
        expires_on=datetime.now(timezone.utc) + timedelta(days=30),
        tags={"rotation-type": "database-password"}
    )

import os
from azure.identity import DefaultAzureCredential

def get_db_connection():
    credential_source = os.getenv("DB_CREDENTIAL_SOURCE", "managed-identity")

    if credential_source == "managed-identity":
        credential = DefaultAzureCredential()
        token = credential.get_token("https://database.windows.net/.default")
        # Connect with token...
    elif credential_source == "vault":
        # Legacy Vault path
        vault_client = hvac.Client(url=os.getenv("VAULT_ADDR"))
        # Connect with Vault credentials...