1
import hvac
2
import os
3
from typing import Dict, Any
4

5
class VaultClient:
6
    def __init__(self, vault_addr: str = "http://localhost:8200", token: str = None):
7
        self.client = hvac.Client(url=vault_addr, token=token or os.getenv('VAULT_TOKEN'))
8

9
    def get_secret(self, path: str, field: str = None) -> Dict[str, Any]:
10
        """Fetch a secret from Vault"""
11
        response = self.client.secrets.kv.v2.read_secret_version(path=path)
12
        data = response['data']['data']
13

14
        if field:
15
            return data.get(field)
16
        return data
17

18
    def get_database_creds(self, environment: str) -> Dict[str, str]:
19
        """Get database credentials for specific environment"""
20
        path = f"secret/{environment}/database"
21
        return self.get_secret(path)
22

23
    def get_api_keys(self, environment: str) -> Dict[str, str]:
24
        """Get API keys for specific environment"""
25
        path = f"secret/{environment}/app-api"
26
        return self.get_secret(path)
27

28
# Usage
29
if __name__ == "__main__":
30
    vault = VaultClient()
31

32
    # Get dev database credentials
33
    db_creds = vault.get_database_creds("dev")
34
    print(f"Database: {db_creds['host']}")
35
    print(f"User: {db_creds['username']}")
36

37
    # Get specific field
38
    api_key = vault.get_secret("secret/prod/app-api", field="api_key")
39
    print(f"API Key: {api_key}")

1
import * as VaultClient from 'node-vault';
2

3
class SecretManager {
4
  private vault: ReturnType<typeof VaultClient>;
5

6
  constructor(address: string = 'http://localhost:8200', token: string) {
7
    this.vault = VaultClient({
8
      apiVersion: 'v1',
9
      endpoint: address,
10
      token: token || process.env.VAULT_TOKEN,
11
    });
12
  }
13

14
  async getSecret(path: string, field?: string): Promise<any> {
15
    try {
16
      const response = await this.vault.read(path);
17
      const data = response.data.data;
18

19
      if (field) {
20
        return data[field];
21
      }
22
      return data;
23
    } catch (error) {
24
      console.error(`Failed to retrieve secret from ${path}:`, error);
25
      throw error;
26
    }
27
  }
28

29
  async getDatabaseConfig(environment: string): Promise<any> {
30
    return this.getSecret(`secret/${environment}/database`);
31
  }
32

33
  async getApiCredentials(environment: string): Promise<any> {
34
    return this.getSecret(`secret/${environment}/app-api`);
35
  }
36
}
37

38
// Usage
39
(async () => {
40
  const manager = new SecretManager('http://localhost:8200', 'root');
41

42
  const dbConfig = await manager.getDatabaseConfig('dev');
43
  console.log(`Connecting to: ${dbConfig.host}`);
44

45
  const apiKey = await manager.getApiCredentials('prod');
46
  console.log(`API Key: ${apiKey.api_key}`);
47
})();

1
import hvac
2
import psycopg2
3
from typing import Generator
4
import contextlib
5

6
class DynamicDatabaseAccess:
7
    def __init__(self, vault_addr: str, vault_token: str):
8
        self.vault = hvac.Client(url=vault_addr, token=vault_token)
9
        self.db_host = os.getenv('DB_HOST', 'db.prod.internal')
10
        self.db_port = os.getenv('DB_PORT', '5432')
11
        self.db_name = os.getenv('DB_NAME', 'appdb')
12

13
    def get_dynamic_credentials(self, role: str) -> dict:
14
        """Fetch temporary database credentials from Vault"""
15
        response = self.vault.secrets.database.read_dynamic_credentials(role)
16
        return response['data']
17

18
    @contextlib.contextmanager
19
    def get_connection(self, role: str = 'readonly') -> Generator:
20
        """Get a database connection with dynamic credentials"""
21
        creds = self.get_dynamic_credentials(role)
22

23
        conn = psycopg2.connect(
24
            host=self.db_host,
25
            port=self.db_port,
26
            database=self.db_name,
27
            user=creds['username'],
28
            password=creds['password']
29
        )
30

31
        try:
32
            yield conn
33
        finally:
34
            conn.close()
35

36
# Usage
37
db_access = DynamicDatabaseAccess('http://localhost:8200', 'root')
38

39
with db_access.get_connection(role='readonly') as conn:
40
    cursor = conn.cursor()
41
    cursor.execute("SELECT COUNT(*) FROM users")
42
    print(f"Total users: {cursor.fetchone()[0]}")

1
# Read-only access to dev/staging secrets
2
path "secret/data/dev/*" {
3
  capabilities = ["read", "list"]
4
}
5

6
path "secret/data/staging/*" {
7
  capabilities = ["read", "list"]
8
}
9

10
# Deny access to production
11
path "secret/data/prod/*" {
12
  capabilities = ["deny"]
13
}
14

15
# Allow token self-renewal
16
path "auth/token/renew-self" {
17
  capabilities = ["update"]
18
}

1
# Full access to production secrets
2
path "secret/data/prod/*" {
3
  capabilities = ["create", "read", "update", "delete", "list"]
4
}
5

6
# Read-only access to staging
7
path "secret/data/staging/*" {
8
  capabilities = ["read", "list"]
9
}
10

11
# Database credential access
12
path "database/creds/prod/*" {
13
  capabilities = ["read"]
14
}
15

16
# Audit logging
17
path "sys/audit" {
18
  capabilities = ["read"]
19
}

1
apiVersion: apps/v1
2
kind: Deployment
3
metadata:
4
  name: app
5
spec:
6
  replicas: 2
7
  selector:
8
    matchLabels:
9
      app: myapp
10
  template:
11
    metadata:
12
      labels:
13
        app: myapp
14
    spec:
15
      serviceAccountName: app
16
      containers:
17
        - name: myapp
18
          image: myapp:latest
19
          env:
20
            - name: VAULT_ADDR
21
              value: 'http://vault.vault.svc.cluster.local:8200'
22
            - name: VAULT_SKIP_VERIFY
23
              value: 'false'
24
          volumeMounts:
25
            - name: vault-token
26
              mountPath: /vault/secrets
27
      volumes:
28
        - name: vault-token
29
          projected:
30
            sources:
31
              - serviceAccountToken:
32
                  audience: vault
33
                  expirationSeconds: 3600
34
                  path: token