pyvault-agent 0.2.0

Python implementation of Vault Agent with client-side caching and automatic authentication

PyVault Agent

A Python implementation of Vault Agent providing client-side caching and automatic authentication for HashiCorp Vault. PyVault Agent brings the core functionality of HashiCorp's Vault Agent directly into your Python applications as a library, eliminating the need for external processes while providing the same benefits of credential caching and automatic token management.

Why PyVault Agent?

Traditional HashiCorp Vault Agent runs as a separate daemon process that manages authentication and caching. PyVault Agent provides the same functionality as an embedded Python library, offering several advantages:

• Simplified deployment: No need to manage separate agent processes
• Direct integration: Native Python API for your applications
• Reduced complexity: Single process architecture eliminates IPC overhead
• Fine-grained control: Programmatic access to cache management and configuration
• Development friendly: Easy to use in development environments and testing

Key Problems Solved

1. Credential Management: Automatically handles multiple authentication methods (AppRole, UserPass, Kubernetes) and token renewal
2. Performance Optimization: Caches secrets to reduce Vault API calls and improve response times
3. Connection Pool Issues: Manages database connection pools with expiring credentials
4. Token Expiration: Seamlessly re-authenticates when tokens expire
5. Thread Safety: Safe for use in multi-threaded applications

Features

• Client-side caching: Reduces API calls to Vault with configurable TTL
• Automatic re-authentication: Seamlessly handles token expiration
• Multiple authentication methods: AppRole, UserPass, and Kubernetes service account authentication
• Flexible auth mount points: Support for custom auth method mount paths
• KV secrets engine support: Read-only secret access with version support (v1 & v2)
• Database secrets engine: Manage dynamic database credentials
• Identity delegation tokens: Cached token generation for workload identity delegation
• Custom endpoint access: Generic POST method for any Vault endpoint
• Connection pool management: Automatic credential refresh for database pools
• Thread-safe caching: Safe for concurrent usage

Installation

pip install pyvault-agent

Quick Start

Authentication Methods

PyVault Agent supports three authentication methods. Choose the one that fits your deployment:

AppRole Authentication

import os
from vault_agent import VaultAgentClient

client = VaultAgentClient.with_approle(
    url=os.getenv("VAULT_ADDR"),
    role_id=os.getenv("VAULT_ROLE_ID"),
    secret_id=os.getenv("VAULT_SECRET_ID"),
    cache_ttl=300,  # Cache for 5 minutes
    max_cache_size=1000
)

UserPass Authentication

client = VaultAgentClient.with_userpass(
    url=os.getenv("VAULT_ADDR"),
    username=os.getenv("VAULT_USERNAME"),
    password=os.getenv("VAULT_PASSWORD"),
    cache_ttl=300,
    max_cache_size=1000
)

Kubernetes Authentication

# Auto-detects JWT from /var/run/secrets/kubernetes.io/serviceaccount/token
client = VaultAgentClient.with_kubernetes(
    url=os.getenv("VAULT_ADDR"),
    role="my-app-role",
    cache_ttl=300,
    max_cache_size=1000
)

# Or provide explicit JWT token
client = VaultAgentClient.with_kubernetes(
    url=os.getenv("VAULT_ADDR"),
    role="my-app-role",
    jwt=os.getenv("VAULT_K8S_JWT"),
    cache_ttl=300,
    max_cache_size=1000
)

Basic Usage

# Once authenticated (using any method above)

# Mount secrets engines dynamically
kv = client.mount(path="secret", type="kv")
database = client.mount(path="database", type="database")

# KV Secrets - Read application configuration
try:
    config = kv.read("myapp/config")
    api_key = config["api_key"]
    db_password = config["db_password"]
    print("Configuration loaded from Vault")
except Exception as e:
    print(f"Failed to load config: {e}")

# Database Credentials - Get dynamic database credentials
try:
    creds = database.read("myapp-db-role")
    print(f"Database user: {creds['username']}")

    # Create connection string
    conn_str = database.get_connection_string(
        role="myapp-db-role",
        template="postgresql://{username}:{password}@{host}:{port}/{database}",
        host="db.example.com",
        port=5432,
        database="myapp"
    )
except Exception as e:
    print(f"Failed to get database credentials: {e}")

# Cache Management - Monitor cache performance
stats = client.get_cache_stats()
print(f"Cache efficiency: {stats['hits']}/{stats['hits'] + stats['misses']} hits")

Advanced Usage

Configuration Options

All authentication methods support the same configuration options:

# AppRole with all options
client = VaultAgentClient.with_approle(
    url="https://vault.example.com",
    role_id="role-id",
    secret_id="secret-id",
    auth_mount_point="approle",  # Auth mount path (default: "approle")
    cache_ttl=300,               # Default cache TTL in seconds
    max_cache_size=1000,         # Maximum number of cached entries
    namespace="team-a",          # Vault namespace (Enterprise)
    verify=True                  # SSL certificate verification
)

# UserPass with custom mount point
client = VaultAgentClient.with_userpass(
    url="https://vault.example.com",
    username="myuser",
    password="mypassword",
    auth_mount_point="userpass-ldap",  # Custom auth mount path
    cache_ttl=300,
    max_cache_size=1000
)

# Kubernetes with custom mount point
client = VaultAgentClient.with_kubernetes(
    url="https://vault.example.com",
    role="my-app-role",
    auth_mount_point="kubernetes-prod",  # Custom auth mount path
    cache_ttl=300,
    max_cache_size=1000
)

Working with Different Secret Engines

# Mount engines with custom paths
kv = client.mount(path="secret", type="kv")
kv_custom = client.mount(path="kv-v2", type="kv")
database = client.mount(path="database", type="database")
db_custom = client.mount(path="db", type="database")

# KV v1 and v2 secrets
config = kv.read("app/config")

# KV v2 secrets with versioning
old_config = kv.read("app/config", version=1)

# Database dynamic credentials
creds = database.read("postgres-readonly")

# Database static credentials
static_creds = database.get_static_credentials("app-service-account")

# Skip cache for fresh credentials
fresh_creds = database.read("postgres-readonly", skip_cache=True)

Custom Vault Endpoints

For endpoints not covered by the built-in secrets engines, use the generic post() method:

# POST to any Vault endpoint
response = client.post("my-custom-engine/data/path", data={"key": "value"})

Identity Delegation Tokens

For workload identity delegation, use get_delegation_token() which provides automatic caching based on the JWT subject and Vault entity:

# Get delegation token with caching
response = client.get_delegation_token(
    role="customer-service",
    subject_token=subject_jwt,  # JWT token of the subject entity
    mount_point="identity-delegation",  # Optional, defaults to "identity-delegation"
)

# Access the delegated token
delegated_token = response["auth"]["client_token"]

# Force fresh token (skip cache)
response = client.get_delegation_token(
    role="customer-service",
    subject_token=subject_jwt,
    skip_cache=True,
)

The cache key is derived from the Vault entity ID and JWT subject claim, with TTL based on the token's lease duration.

Database Connection Pools

One of the key challenges with dynamic database credentials is managing connection pools when credentials expire. PyVault Agent provides a DatabaseConnectionManager that automatically handles credential refresh and pool recreation:

from vault_agent import VaultAgentClient
from vault_agent.database_pool import DatabaseConnectionManager
import psycopg2.pool

# Create client with any auth method
client = VaultAgentClient.with_approle(...)

# Mount database secrets engine
database = client.mount(path="database", type="database")

# Managed connection pool with auto-refresh
with DatabaseConnectionManager(
    database_secrets=database,
    role="postgres-role",
    pool_class=psycopg2.pool.SimpleConnectionPool,
    pool_kwargs={
        "minconn": 1,
        "maxconn": 10,
        "host": "db.example.com",
        "database": "myapp"
    },
    refresh_buffer=0.8,  # Refresh at 80% of credential TTL
    validation_query="SELECT 1",  # Query to validate connections
) as manager:

    # Get connections that are automatically managed
    with manager.get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users")
        results = cursor.fetchall()

Background Refresh

For high-performance applications, use BackgroundRefreshManager to refresh credentials proactively:

from vault_agent.database_pool import BackgroundRefreshManager

# Mount database secrets engine
database = client.mount(path="database", type="database")

with BackgroundRefreshManager(
    database_secrets=database,
    role="postgres-role",
    pool_class=psycopg2.pool.ThreadedConnectionPool,
    pool_kwargs={"minconn": 2, "maxconn": 10, "host": "db.example.com"},
    check_interval=30,  # Check every 30 seconds
) as manager:
    # Credentials refresh in background, zero-latency for requests
    with manager.get_connection() as conn:
        # Your database operations
        pass

Error Handling and Resilience

from vault_agent.utils import SecretNotFoundError, AuthenticationError

# Mount secrets engine
kv = client.mount(path="secret", type="kv")

try:
    # Attempt to read secret with automatic retry
    config = kv.read("app/config")
except SecretNotFoundError:
    print("Secret not found - using defaults")
    config = {"api_key": "default"}
except AuthenticationError:
    print("Failed to authenticate with Vault")
    # Handle authentication failure
except Exception as e:
    print(f"Unexpected error: {e}")
    # Handle other errors

# Cache management
if client.get_cache_stats()["size"] > 500:
    client.clear_cache()  # Clear cache if getting too large

Integration with Existing Applications

Django Integration

# settings.py
import os
from vault_agent import VaultAgentClient

vault_client = VaultAgentClient.with_approle(
    url=os.getenv("VAULT_ADDR"),
    role_id=os.getenv("VAULT_ROLE_ID"),
    secret_id=os.getenv("VAULT_SECRET_ID"),
)

# Mount database secrets engine and get credentials
database = vault_client.mount(path="database", type="database")
db_creds = database.read("django-db-role")

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'myapp',
        'USER': db_creds['username'],
        'PASSWORD': db_creds['password'],
        'HOST': 'db.example.com',
        'PORT': '5432',
    }
}

Flask Integration

from flask import Flask
from vault_agent import VaultAgentClient
import os

app = Flask(__name__)

# Initialize Vault client (works in Kubernetes pod)
vault_client = VaultAgentClient.with_kubernetes(
    url=os.getenv("VAULT_ADDR"),
    role="flask-app-role",
)

@app.before_first_request
def setup():
    # Mount KV secrets engine and load configuration from Vault
    kv = vault_client.mount(path="secret", type="kv")
    config = kv.read("flask/config")
    app.config.update(config)

Development and Testing

Running Tests

# Install development dependencies
pip install -e ".[dev]"

# Run unit tests
pytest tests/

# Run functional tests (requires Vault dev server)
vault server -dev -dev-root-token-id="root"
VAULT_ADDR=http://127.0.0.1:8200 VAULT_TOKEN=root pytest tests/functional/

# Run with coverage
pytest --cov=vault_agent tests/

Development Setup

# Clone the repository
git clone https://github.com/your-username/pyvault-agent.git
cd pyvault-agent

# Install in development mode
pip install -e ".[dev]"

# Run linting and formatting
black vault_agent/ tests/
ruff vault_agent/ tests/
mypy vault_agent/

Running Examples

# Set environment variables (for AppRole)
export VAULT_ADDR="http://127.0.0.1:8200"
export VAULT_ROLE_ID="your-role-id"
export VAULT_SECRET_ID="your-secret-id"

# Run basic example (AppRole)
python example.py

# Run multi-auth example (demonstrates all auth methods)
python example_multi_auth.py

# Run connection pool example
python example_pool.py

Performance Considerations

Cache Tuning

• TTL: Set appropriate cache TTL based on secret sensitivity and change frequency
• Size: Limit cache size to prevent memory growth in long-running applications
• Hit Rate: Monitor cache hit rates to optimize TTL settings

# Monitor cache performance
stats = client.get_cache_stats()
hit_rate = stats['hits'] / (stats['hits'] + stats['misses'])
print(f"Cache hit rate: {hit_rate:.2%}")

# Adjust TTL based on performance needs
if hit_rate < 0.8:  # Less than 80% hit rate
    client.set_cache_ttl(600)  # Increase TTL

Connection Pool Best Practices

• Buffer: Set refresh_buffer to 0.7-0.8 to refresh before expiry
• Validation: Use connection validation to catch stale connections
• Pool Size: Size pools appropriately for your application load
• Monitoring: Monitor credential refresh frequency

Security Considerations

1. Secure Storage: Store role_id and secret_id securely (environment variables, not in code)
2. Network Security: Use HTTPS for Vault connections in production
3. Credential Rotation: Regularly rotate AppRole credentials
4. Audit Logging: Enable Vault audit logging to track secret access
5. Least Privilege: Configure Vault policies with minimal required permissions

Troubleshooting

Common Issues

Authentication Failures

# Check Vault connectivity
try:
    client = VaultAgentClient.with_approle(
        url="https://vault.example.com",
        role_id="...",
        secret_id="..."
    )
except AuthenticationError as e:
    print(f"Auth failed: {e}")
    # Check credentials and auth method configuration

Cache Issues

# Clear cache if data seems stale
client.clear_cache()

# Check cache statistics
stats = client.get_cache_stats()
print(f"Cache size: {stats['size']}")

Connection Pool Problems

# Force credential refresh
manager.refresh_now()

# Check credential expiry
print(f"Credentials expire at: {manager.credentials_expire_at}")

Debug Logging

import logging
logging.basicConfig(level=logging.DEBUG)

# This will show cache hits/misses and authentication events
client = VaultAgentClient.with_approle(...)

Roadmap

Current Version (0.2.0)

• Multiple authentication methods: AppRole, UserPass, Kubernetes
• Flexible auth mount points for custom configurations
• Automatic re-authentication on token expiry
• KV secrets engine (v1 & v2) read-only access with caching
• Database secrets engine read-only access with caching
• Connection pool management
• Thread-safe operations
• Generic POST method for custom Vault endpoints
• Identity delegation token support with caching

Planned Features

• Token renewal: Proactive token refresh before expiry
• Lease renewal: Automatic secret lease renewal
• Additional auth methods: JWT, AWS IAM, Azure, GCP
• More secret engines: PKI, Transit, SSH
• Metrics integration: Prometheus metrics export
• Configuration files: YAML/TOML configuration support
• Async support: AsyncIO-compatible client

Contributing

Contributions are welcome! Please read our contributing guidelines and submit pull requests to our GitHub repository.

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.