# Secrets Management: Store API Keys Safely

# 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

| Solution | Best For | Features |
|----------|----------|----------|
| **HashiCorp Vault** | Enterprise | Self-hosted, multi-cloud, dynamic secrets |
| **AWS Secrets Manager** | AWS ecosystem | Native integration, automatic rotation |
| **Azure Key Vault** | Azure ecosystem | HSM support, compliance certifications |
| **Google Secret Manager** | GCP | Serverless, IAM integration |
| **1Password/LastPass** | Teams | User-friendly, password sharing |

## Code

### 1. HashiCorp Vault Implementation

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

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

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

```python
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**
```bash
# 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**
```bash
# 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.