External Secret Management Systems

External Secret Management Systems

Enterprise secret management often requires dedicated systems like HashiCorp Vault, AWS Secrets Manager, or Azure Key Vault. These systems provide centralized secret storage, fine-grained access control, audit logging, and secret rotation capabilities. Integration with containers requires careful architecture to maintain security while enabling seamless secret access.

HashiCorp Vault provides comprehensive secret management with dynamic secret generation. Vault's App Role authentication enables containers to authenticate without pre-shared secrets. Dynamic database credentials provide time-limited access with automatic revocation. Encryption as a Service protects data without managing encryption keys. Transit backend enables application-level encryption without exposing keys.

# Example: Container application with Vault integration
import os
import sys
import json
import time
import hvac
from functools import wraps
from datetime import datetime, timedelta

class VaultSecretManager:
    def __init__(self, vault_addr, role_id=None, secret_id=None):
        self.vault_addr = vault_addr
        self.client = hvac.Client(url=vault_addr)
        
        # Authenticate using AppRole
        if role_id and secret_id:
            self.authenticate_app_role(role_id, secret_id)
        else:
            # Use Kubernetes auth if running in k8s
            self.authenticate_kubernetes()
        
        self.secrets_cache = {}
        self.lease_renewal_threads = {}
        
    def authenticate_app_role(self, role_id, secret_id):
        """Authenticate using AppRole method"""
        response = self.client.auth.approle.login(
            role_id=role_id,
            secret_id=secret_id
        )
        self.client.token = response['auth']['client_token']
        self.start_token_renewal(response['auth'])
        
    def authenticate_kubernetes(self):
        """Authenticate using Kubernetes service account"""
        jwt_path = '/var/run/secrets/kubernetes.io/serviceaccount/token'
        role = os.environ.get('VAULT_ROLE', 'webapp')
        
        with open(jwt_path, 'r') as f:
            jwt = f.read()
            
        response = self.client.auth.kubernetes.login(
            role=role,
            jwt=jwt
        )
        self.client.token = response['auth']['client_token']
        self.start_token_renewal(response['auth'])
    
    def get_secret(self, path, key=None, cache_duration=300):
        """Retrieve secret with caching"""
        cache_key = f"{path}:{key}" if key else path
        
        # Check cache
        if cache_key in self.secrets_cache:
            cached = self.secrets_cache[cache_key]
            if cached['expires'] > datetime.now():
                return cached['value']
        
        # Fetch from Vault
        try:
            response = self.client.secrets.kv.v2.read_secret_version(
                path=path
            )
            data = response['data']['data']
            
            # Cache the result
            self.secrets_cache[cache_key] = {
                'value': data.get(key) if key else data,
                'expires': datetime.now() + timedelta(seconds=cache_duration)
            }
            
            # Handle renewable secrets
            if 'lease_id' in response:
                self.start_lease_renewal(response['lease_id'], response['lease_duration'])
            
            return data.get(key) if key else data
            
        except Exception as e:
            print(f"Error fetching secret: {e}", file=sys.stderr)
            raise
    
    def get_dynamic_database_credentials(self, database_role):
        """Get dynamic database credentials"""
        try:
            response = self.client.secrets.database.generate_credentials(
                name=database_role
            )
            
            creds = {
                'username': response['data']['username'],
                'password': response['data']['password'],
                'lease_id': response['lease_id'],
                'lease_duration': response['lease_duration']
            }
            
            # Start automatic renewal
            self.start_lease_renewal(response['lease_id'], response['lease_duration'])
            
            return creds
            
        except Exception as e:
            print(f"Error getting database credentials: {e}", file=sys.stderr)
            raise
    
    def encrypt_data(self, plaintext, key_name='transit-key'):
        """Encrypt data using Vault's transit backend"""
        import base64
        
        encoded = base64.b64encode(plaintext.encode()).decode()
        response = self.client.secrets.transit.encrypt_data(
            name=key_name,
            plaintext=encoded
        )
        
        return response['data']['ciphertext']
    
    def decrypt_data(self, ciphertext, key_name='transit-key'):
        """Decrypt data using Vault's transit backend"""
        import base64
        
        response = self.client.secrets.transit.decrypt_data(
            name=key_name,
            ciphertext=ciphertext
        )
        
        return base64.b64decode(response['data']['plaintext']).decode()
    
    def start_lease_renewal(self, lease_id, lease_duration):
        """Start background lease renewal"""
        import threading
        
        def renew_lease():
            while True:
                try:
                    # Renew at 2/3 of lease duration
                    time.sleep(lease_duration * 2 / 3)
                    self.client.sys.renew_lease(lease_id)
                except Exception as e:
                    print(f"Lease renewal failed: {e}", file=sys.stderr)
                    break
        
        thread = threading.Thread(target=renew_lease, daemon=True)
        thread.start()
        self.lease_renewal_threads[lease_id] = thread

# Usage example
vault = VaultSecretManager(
    vault_addr=os.environ.get('VAULT_ADDR', 'https://vault:8200'),
    role_id=os.environ.get('VAULT_ROLE_ID'),
    secret_id=os.environ.get('VAULT_SECRET_ID')
)

# Get static secrets
api_key = vault.get_secret('webapp/api-keys', 'payment-gateway')

# Get dynamic database credentials
db_creds = vault.get_dynamic_database_credentials('webapp-db')
connection_string = f"postgresql://{db_creds['username']}:{db_creds['password']}@postgres:5432/webapp"

# Encrypt sensitive data
encrypted = vault.encrypt_data('sensitive user data')

Cloud provider secret management services offer native integration with container platforms. AWS Secrets Manager integrates with ECS and EKS through IAM roles. Azure Key Vault uses managed identities for authentication. Google Secret Manager leverages workload identity. These integrations simplify secret access while maintaining security through cloud-native identity systems.