pyvault-agent 0.2.0
Python implementation of Vault Agent with client-side caching and automatic authentication
License: Apache-2.0
Requires Python: >=3.12

# 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

```bash
pip install pyvault-agent
```

## Quick Start

### Authentication Methods

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

#### AppRole Authentication

```python
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

```python
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

```python
# 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

```python
# 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:

```python
# 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

```python
# 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:

```python
# 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:

```python
# 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:

```python
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:

```python
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

```python
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

```python
# 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

```python
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

```bash
# 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

```bash
# 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

```bash
# 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

```python
# 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**
```python
# 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**
```python
# 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**
```python
# Force credential refresh
manager.refresh_now()

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

### Debug Logging

```python
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)
- [x] Multiple authentication methods: AppRole, UserPass, Kubernetes
- [x] Flexible auth mount points for custom configurations
- [x] Automatic re-authentication on token expiry
- [x] KV secrets engine (v1 & v2) read-only access with caching
- [x] Database secrets engine read-only access with caching
- [x] Connection pool management
- [x] Thread-safe operations
- [x] Generic POST method for custom Vault endpoints
- [x] 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](LICENSE) file for details.