Secrets Management: Store API Keys Safely

Problem

Hardcoding API keys, database credentials, and sensitive tokens directly in source code is a critical security vulnerability. This exposes secrets to:

• Version control exposure: Secrets committed to Git repositories are permanently visible in history
• Accidental leaks: Developers sharing code snippets or screenshots inadvertently reveal credentials
• Unauthorized access: Anyone with repository access gains production credentials
• Compliance violations: GDPR, HIPAA, and SOC 2 require secure credential management
• Lateral movement: Compromised keys enable attackers to access multiple systems

Traditional approaches like environment variables in .env files still risk exposure through misconfiguration or accidental commits.

Solution

Implement a dedicated secrets management system using vault services that:

1. Centralize storage: Single source of truth for all credentials
2. Encrypt at rest: Secrets encrypted with strong encryption algorithms
3. Control access: Fine-grained permissions and audit trails
4. Rotate automatically: Periodic credential rotation without downtime
5. Audit logging: Track who accessed what secrets and when
6. Dynamic secrets: Generate temporary credentials with limited lifespans

Popular Vault Solutions

Code

1. HashiCorp Vault Implementation

import hvac
import os
from typing import Dict, Any

class VaultSecretsManager:
    """Secure secrets management using HashiCorp Vault"""

    def __init__(self, vault_addr: str, vault_token: str):
        """
        Initialize Vault client

        Args:
            vault_addr: Vault server address (e.g., 'http://localhost:8200')
            vault_token: Authentication token
        """
        self.client = hvac.Client(url=vault_addr, token=vault_token)

    def store_secret(self, path: str, secret_data: Dict[str, Any]) -> None:
        """Store a secret in Vault"""
        try:
            self.client.secrets.kv.v2.create_or_update_secret(
                path=path,
                secret=secret_data
            )
            print(f"✓ Secret stored at {path}")
        except Exception as e:
            print(f"✗ Error storing secret: {e}")
            raise

    def retrieve_secret(self, path: str) -> Dict[str, Any]:
        """Retrieve a secret from Vault"""
        try:
            response = self.client.secrets.kv.v2.read_secret_version(path=path)
            return response['data']['data']
        except Exception as e:
            print(f"✗ Error retrieving secret: {e}")
            raise

    def delete_secret(self, path: str) -> None:
        """Delete a secret from Vault"""
        try:
            self.client.secrets.kv.v2.delete_secret_version(path=path)
            print(f"✓ Secret deleted from {path}")
        except Exception as e:
            print(f"✗ Error deleting secret: {e}")
            raise

    def rotate_secret(self, path: str, new_secret: Dict[str, Any]) -> None:
        """Rotate a secret (update with new values)"""
        self.store_secret(path, new_secret)
        print(f"✓ Secret rotated at {path}")

# Usage
if __name__ == "__main__":
    vault = VaultSecretsManager(
        vault_addr=os.getenv('VAULT_ADDR', 'http://localhost:8200'),
        vault_token=os.getenv('VAULT_TOKEN')
    )

    # Store API credentials
    vault.store_secret('secret/api/stripe', {
        'api_key': 'sk_live_xxxxx',
        'webhook_secret': 'whsec_xxxxx'
    })

    # Retrieve credentials
    stripe_creds = vault.retrieve_secret('secret/api/stripe')
    print(f"API Key: {stripe_creds['api_key']}")

2. AWS Secrets Manager Implementation

import boto3
import json
from typing import Dict, Any

class AWSSecretsManager:
    """Secrets management using AWS Secrets Manager"""

    def __init__(self, region_name: str = 'us-east-1'):
        """Initialize AWS Secrets Manager client"""
        self.client = boto3.client('secretsmanager', region_name=region_name)

    def create_secret(self, name: str, secret_value: Dict[str, Any]) -> str:
        """Create a new secret"""
        try:
            response = self.client.create_secret(
                Name=name,
                Description=f'Secret for {name}',
                SecretString=json.dumps(secret_value),
                Tags=[
                    {'Key': 'Environment', 'Value': 'production'},
                    {'Key': 'ManagedBy', 'Value': 'application'}
                ]
            )
            print(f"✓ Secret created: {response['ARN']}")
            return response['ARN']
        except self.client.exceptions.ResourceExistsException:
            print(f"✗ Secret {name} already exists")
            raise

    def get_secret(self, name: str) -> Dict[str, Any]:
        """Retrieve a secret"""
        try:
            response = self.client.get_secret_value(SecretId=name)
            return json.loads(response['SecretString'])
        except self.client.exceptions.ResourceNotFoundException:
            print(f"✗ Secret {name} not found")
            raise

    def update_secret(self, name: str, secret_value: Dict[str, Any]) -> None:
        """Update an existing secret"""
        try:
            self.client.update_secret(
                SecretId=name,
                SecretString=json.dumps(secret_value)
            )
            print(f"✓ Secret updated: {name}")
        except Exception as e:
            print(f"✗ Error updating secret: {e}")
            raise

    def delete_secret(self, name: str, recovery_days: int = 7) -> None:
        """Delete a secret with recovery window"""
        try:
            self.client.delete_secret(
                SecretId=name,
                RecoveryWindowInDays=recovery_days
            )
            print(f"✓ Secret scheduled for deletion: {name}")
        except Exception as e:
            print(f"✗ Error deleting secret: {e}")
            raise

    def rotate_secret(self, name: str, new_value: Dict[str, Any]) -> None:
        """Rotate secret with automatic versioning"""
        self.update_secret(name, new_value)
        print(f"✓ Secret rotated with new version")

# Usage
if __name__ == "__main__":
    secrets_manager = AWSSecretsManager(region_name='us-east-1')

    # Create database credentials
    secrets_manager.create_secret('prod/database/postgres', {
        'username': 'admin',
        'password': 'SecurePassword123!',
        'host': 'db.example.com',
        'port': 5432,
        'database': 'production'
    })

    # Retrieve credentials
    db_creds = secrets_manager.get_secret('prod/database/postgres')
    print(f"Database: {db_creds['database']}@{db_creds['host']}")

3. Environment-Based Configuration with Validation

import os
from dataclasses import dataclass
from typing import Optional
from dotenv import load_dotenv

@dataclass
class SecretsConfig:
    """Configuration with secrets validation"""

    api_key: str
    database_url: str
    jwt_secret: str
    stripe_key: Optional[str] = None

    @classmethod
    def from_environment(cls) -> 'SecretsConfig':
        """Load and validate secrets from environment"""
        load_dotenv()

        required_secrets = ['API_KEY', 'DATABASE_URL', 'JWT_SECRET']
        missing = [s for s in required_secrets if not os.getenv(s)]

        if missing:
            raise ValueError(f"Missing required secrets: {missing}")

        return cls(
            api_key=os.getenv('API_KEY'),
            database_url=os.getenv('DATABASE_URL'),
            jwt_secret=os.getenv('JWT_SECRET'),
            stripe_key=os.getenv('STRIPE_KEY')
        )

    def validate(self) -> bool:
        """Validate secret format and strength"""
        if len(self.api_key) < 32:
            raise ValueError("API key too short")
        if not self.database_url.startswith(('postgresql://', 'mysql://')):
            raise ValueError("Invalid database URL format")
        if len(self.jwt_secret) < 64:
            raise ValueError("JWT secret too short")
        return True

# Usage
config = SecretsConfig.from_environment()
config.validate()

4. Secure Application Integration

from flask import Flask, request
from functools import wraps
import logging

app = Flask(__name__)
logger = logging.getLogger(__name__)

class SecureApp:
    """Flask app with secure secrets handling"""

    def __init__(self, vault_manager):
        self.vault = vault_manager
        self.secrets_cache = {}
        self.cache_ttl = 3600  # 1 hour

    def get_secret(self, secret_path: str, use_cache: bool = True):
        """Get secret with optional caching"""
        if use_cache and secret_path in self.secrets_cache:
            return self.secrets_cache[secret_path]

        secret = self.vault.retrieve_secret(secret_path)
        self.secrets_cache[secret_path] = secret
        return secret

    def require_api_key(self, f):
        """Decorator to validate API key"""
        @wraps(f)
        def decorated_function(*args, **kwargs):
            api_key = request.headers.get('X-API-Key')

            if not api_key:
                logger.warning("Missing API key in request")
                return {'error': 'Unauthorized'}, 401

            valid_key = self.get_secret('secret/api/keys')['valid_key']

            if api_key != valid_key:
                logger.warning(f"Invalid API key attempt: {api_key[:8]}...")
                return {'error': 'Unauthorized'}, 401

            return f(*args, **kwargs)
        return decorated_function

# Usage
@app.route('/api/protected', methods=['GET'])
@SecureApp(vault).require_api_key
def protected_endpoint():
    return {'data': 'sensitive information'}

Tips

1. Never Commit Secrets

# Add to .gitignore
.env
.env.local
secrets/
*.key
*.pem

2. Use Secret Rotation

• Rotate credentials every 30-90 days
• Implement automatic rotation for database passwords
• Use temporary credentials with short TTLs

3. Implement Least Privilege

• Grant minimal necessary permissions
• Use separate credentials per environment
• Restrict secret access by role/service

4. Audit and Monitor

• Log all secret access attempts
• Set up alerts for unauthorized access
• Review audit logs regularly

5. Secure Vault Access

# Use strong authentication
export VAULT_TOKEN=$(vault login -method=oidc)

# Enable TLS
vault server -config=config.hcl -tls-cert-file=cert.pem -tls-key-file=key.pem

6. Local Development

• Use separate dev credentials
• Never use production secrets locally
• Use tools like direnv for automatic environment loading

7. Backup and Disaster Recovery

• Backup encryption keys securely
• Test recovery procedures regularly
• Use multi-region replication for critical secrets

8. Compliance Considerations

• Document secret access policies
• Maintain audit trails for compliance
• Implement encryption at rest and in transit
• Use HSM (Hardware Security Module) for critical keys

---

Key Takeaway: Secrets management is not optional—it's fundamental to application security. Implement a vault solution appropriate for your infrastructure, enforce strict access controls, and maintain comprehensive audit logs.