Tutorial: Managed Identity for Zero Stored Secrets¶

import hvac
import pyodbc

# Authenticate to Vault
vault = hvac.Client(url='https://vault:8200')
vault.auth.approle.login(role_id='xxx', secret_id='yyy')

# Get dynamic SQL credentials
creds = vault.secrets.database.generate_credentials('webapp-sql-role')
username = creds['data']['username']
password = creds['data']['password']

# Connect with Vault-generated credentials
conn_str = (
    f"DRIVER={{ODBC Driver 18 for SQL Server}};"
    f"SERVER=sql-prod.database.windows.net;"
    f"DATABASE=appdb;"
    f"UID={username};"
    f"PWD={password};"
    f"Encrypt=yes;"
)
conn = pyodbc.connect(conn_str)

from azure.identity import DefaultAzureCredential
import pyodbc
import struct

# Get token using managed identity (no secrets anywhere)
credential = DefaultAzureCredential(
    managed_identity_client_id="<MI_CLIENT_ID>"  # Optional: specify which MI
)
token = credential.get_token("https://database.windows.net/.default")

# Build connection with token
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;"
    "Encrypt=yes;",
    attrs_before={SQL_COPT_SS_ACCESS_TOKEN: token_struct}
)

# Execute a query to verify
cursor = conn.cursor()
cursor.execute("SELECT CURRENT_USER AS connected_as")
row = cursor.fetchone()
print(f"Connected as: {row.connected_as}")  # Should print: mi-webapp-sql

#!/usr/bin/env python3
"""Test managed identity SQL connection."""
from azure.identity import DefaultAzureCredential
import pyodbc
import struct

SERVER = "sql-prod.database.windows.net"
DATABASE = "appdb"

print(f"Testing connection to {SERVER}/{DATABASE}...")

try:
    credential = DefaultAzureCredential()
    token = credential.get_token("https://database.windows.net/.default")
    print(f"Token acquired (expires: {token.expires_on})")

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

    conn = pyodbc.connect(
        f"DRIVER={{ODBC Driver 18 for SQL Server}};"
        f"SERVER={SERVER};"
        f"DATABASE={DATABASE};"
        f"Encrypt=yes;",
        attrs_before={1256: token_struct}
    )

    cursor = conn.cursor()
    cursor.execute("SELECT CURRENT_USER AS username, @@VERSION AS version")
    row = cursor.fetchone()
    print(f"Connected as: {row.username}")
    print(f"SQL Server: {row.version[:50]}...")
    print("SUCCESS: Managed identity connection works.")

    conn.close()
except Exception as e:
    print(f"FAILED: {e}")

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",
    port=5432,
    database="appdb",
    user="mi-webapp-sql",     # Managed identity name
    password=token.token,     # Token as password
    sslmode="require"
)

cursor = conn.cursor()
cursor.execute("SELECT current_user, version()")
row = cursor.fetchone()
print(f"Connected as: {row[0]}")
print(f"PostgreSQL: {row[1][:50]}...")

from azure.identity import DefaultAzureCredential
import psycopg2
import time

class ManagedIdentityPgConnection:
    """PostgreSQL connection with automatic token refresh."""

    def __init__(self, host, database, identity_name, mi_client_id=None):
        self.host = host
        self.database = database
        self.identity_name = identity_name
        self.credential = DefaultAzureCredential(
            managed_identity_client_id=mi_client_id
        )
        self._conn = None
        self._token_expires = 0

    def get_connection(self):
        """Get or refresh the database connection."""
        now = time.time()
        if self._conn is None or now >= self._token_expires - 300:
            # Token expired or will expire in 5 minutes -- refresh
            if self._conn:
                self._conn.close()

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

            self._conn = psycopg2.connect(
                host=self.host,
                port=5432,
                database=self.database,
                user=self.identity_name,
                password=token.token,
                sslmode="require"
            )
        return self._conn

# Usage
pg = ManagedIdentityPgConnection(
    host="pg-prod.postgres.database.azure.com",
    database="appdb",
    identity_name="mi-webapp-sql"
)
conn = pg.get_connection()

import hvac
from azure.cosmos import CosmosClient

vault = hvac.Client(url='https://vault:8200')
vault.auth.approle.login(role_id='xxx', secret_id='yyy')
secret = vault.secrets.kv.v2.read_secret_version(path='cosmos/primary-key')
primary_key = secret['data']['data']['key']

client = CosmosClient(
    url='https://cosmos-prod.documents.azure.com:443/',
    credential=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')

# Query data
items = list(container.query_items(
    query="SELECT * FROM c WHERE c.category = @cat",
    parameters=[{"name": "@cat", "value": "electronics"}],
    enable_cross_partition_query=True
))
print(f"Found {len(items)} items")

from azure.identity import DefaultAzureCredential

# This single line works everywhere:
# - Local dev: uses az login / Visual Studio / IntelliJ credentials
# - App Service: uses system-assigned or user-assigned managed identity
# - AKS: uses workload identity (OIDC federation)
# - Azure VM: uses managed identity via IMDS
credential = DefaultAzureCredential()