Adolfo Delorenzo
9170198c6c
- Implement encrypted secrets storage with AES-128 encryption - Add secret rotation capabilities with scheduling - Implement comprehensive audit logging for all secret operations - Create centralized configuration management system - Add CLI tool for interactive secret management - Integrate secrets with Flask configuration - Support environment-specific configurations - Add integrity verification for stored secrets - Implement secure key derivation with PBKDF2 Features: - Encrypted storage in .secrets.json - Master key protection with file permissions - Automatic secret rotation scheduling - Audit trail for compliance - Migration from environment variables - Flask CLI integration - Validation and sanitization Security improvements: - No more hardcoded secrets in configuration - Encrypted storage at rest - Secure key management - Access control via authentication - Comprehensive audit logging - Integrity verification CLI commands: - manage_secrets.py init - Initialize secrets - manage_secrets.py set/get/delete - Manage secrets - manage_secrets.py rotate - Rotate secrets - manage_secrets.py audit - View audit logs - manage_secrets.py verify - Check integrity 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
411 lines
15 KiB
Python
411 lines
15 KiB
Python
# Secrets management system for secure configuration
|
|
import os
|
|
import json
|
|
import base64
|
|
import logging
|
|
from typing import Any, Dict, Optional, List
|
|
from datetime import datetime, timedelta
|
|
from cryptography.fernet import Fernet
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
from functools import lru_cache
|
|
from threading import Lock
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class SecretsManager:
|
|
"""
|
|
Secure secrets management with encryption, rotation, and audit logging
|
|
"""
|
|
def __init__(self, config_file: str = None):
|
|
self.config_file = config_file or os.environ.get('SECRETS_CONFIG', '.secrets.json')
|
|
self.lock = Lock()
|
|
self._secrets_cache = {}
|
|
self._encryption_key = None
|
|
self._master_key = None
|
|
self._audit_log = []
|
|
self._rotation_schedule = {}
|
|
self._validators = {}
|
|
|
|
# Initialize encryption
|
|
self._init_encryption()
|
|
|
|
# Load secrets
|
|
self._load_secrets()
|
|
|
|
def _init_encryption(self):
|
|
"""Initialize encryption key from environment or generate new one"""
|
|
# Try to get master key from environment
|
|
master_key = os.environ.get('MASTER_KEY')
|
|
|
|
if not master_key:
|
|
# Try to load from secure file
|
|
key_file = os.environ.get('MASTER_KEY_FILE', '.master_key')
|
|
if os.path.exists(key_file):
|
|
try:
|
|
with open(key_file, 'rb') as f:
|
|
master_key = f.read().decode('utf-8').strip()
|
|
except Exception as e:
|
|
logger.error(f"Failed to load master key from file: {e}")
|
|
|
|
if not master_key:
|
|
# Generate new master key
|
|
logger.warning("No master key found. Generating new one.")
|
|
master_key = Fernet.generate_key().decode('utf-8')
|
|
|
|
# Save to secure file (should be protected by OS permissions)
|
|
key_file = os.environ.get('MASTER_KEY_FILE', '.master_key')
|
|
try:
|
|
with open(key_file, 'wb') as f:
|
|
f.write(master_key.encode('utf-8'))
|
|
os.chmod(key_file, 0o600) # Owner read/write only
|
|
logger.info(f"Master key saved to {key_file}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to save master key: {e}")
|
|
|
|
self._master_key = master_key.encode('utf-8')
|
|
|
|
# Derive encryption key from master key
|
|
kdf = PBKDF2HMAC(
|
|
algorithm=hashes.SHA256(),
|
|
length=32,
|
|
salt=b'talk2me-secrets-salt', # In production, use random salt
|
|
iterations=100000,
|
|
)
|
|
key = base64.urlsafe_b64encode(kdf.derive(self._master_key))
|
|
self._encryption_key = Fernet(key)
|
|
|
|
def _load_secrets(self):
|
|
"""Load encrypted secrets from file"""
|
|
if not os.path.exists(self.config_file):
|
|
logger.info(f"No secrets file found at {self.config_file}")
|
|
return
|
|
|
|
try:
|
|
with open(self.config_file, 'r') as f:
|
|
data = json.load(f)
|
|
|
|
# Decrypt secrets
|
|
for key, value in data.get('secrets', {}).items():
|
|
if isinstance(value, dict) and 'encrypted' in value:
|
|
try:
|
|
decrypted = self._decrypt(value['encrypted'])
|
|
self._secrets_cache[key] = {
|
|
'value': decrypted,
|
|
'created': value.get('created'),
|
|
'rotated': value.get('rotated'),
|
|
'metadata': value.get('metadata', {})
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Failed to decrypt secret {key}: {e}")
|
|
else:
|
|
# Plain text (for migration)
|
|
self._secrets_cache[key] = {
|
|
'value': value,
|
|
'created': datetime.now().isoformat(),
|
|
'rotated': None,
|
|
'metadata': {}
|
|
}
|
|
|
|
# Load rotation schedule
|
|
self._rotation_schedule = data.get('rotation_schedule', {})
|
|
|
|
# Load audit log
|
|
self._audit_log = data.get('audit_log', [])
|
|
|
|
logger.info(f"Loaded {len(self._secrets_cache)} secrets")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load secrets: {e}")
|
|
|
|
def _save_secrets(self):
|
|
"""Save encrypted secrets to file"""
|
|
with self.lock:
|
|
data = {
|
|
'secrets': {},
|
|
'rotation_schedule': self._rotation_schedule,
|
|
'audit_log': self._audit_log[-1000:] # Keep last 1000 entries
|
|
}
|
|
|
|
# Encrypt secrets
|
|
for key, secret_data in self._secrets_cache.items():
|
|
data['secrets'][key] = {
|
|
'encrypted': self._encrypt(secret_data['value']),
|
|
'created': secret_data.get('created'),
|
|
'rotated': secret_data.get('rotated'),
|
|
'metadata': secret_data.get('metadata', {})
|
|
}
|
|
|
|
# Save to file
|
|
try:
|
|
# Write to temporary file first
|
|
temp_file = f"{self.config_file}.tmp"
|
|
with open(temp_file, 'w') as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
# Set secure permissions
|
|
os.chmod(temp_file, 0o600) # Owner read/write only
|
|
|
|
# Atomic rename
|
|
os.rename(temp_file, self.config_file)
|
|
|
|
logger.info(f"Saved {len(self._secrets_cache)} secrets")
|
|
except Exception as e:
|
|
logger.error(f"Failed to save secrets: {e}")
|
|
raise
|
|
|
|
def _encrypt(self, value: str) -> str:
|
|
"""Encrypt a value"""
|
|
if not isinstance(value, str):
|
|
value = str(value)
|
|
return self._encryption_key.encrypt(value.encode('utf-8')).decode('utf-8')
|
|
|
|
def _decrypt(self, encrypted_value: str) -> str:
|
|
"""Decrypt a value"""
|
|
return self._encryption_key.decrypt(encrypted_value.encode('utf-8')).decode('utf-8')
|
|
|
|
def _audit(self, action: str, key: str, user: str = None, details: dict = None):
|
|
"""Add entry to audit log"""
|
|
entry = {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'action': action,
|
|
'key': key,
|
|
'user': user or 'system',
|
|
'details': details or {}
|
|
}
|
|
self._audit_log.append(entry)
|
|
logger.info(f"Audit: {action} on {key} by {user or 'system'}")
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
"""Get a secret value"""
|
|
# Try cache first
|
|
if key in self._secrets_cache:
|
|
self._audit('access', key)
|
|
return self._secrets_cache[key]['value']
|
|
|
|
# Try environment variable
|
|
env_key = f"SECRET_{key.upper()}"
|
|
env_value = os.environ.get(env_key)
|
|
if env_value:
|
|
self._audit('access', key, details={'source': 'environment'})
|
|
return env_value
|
|
|
|
# Try regular environment variable
|
|
env_value = os.environ.get(key)
|
|
if env_value:
|
|
self._audit('access', key, details={'source': 'environment'})
|
|
return env_value
|
|
|
|
self._audit('access_failed', key)
|
|
return default
|
|
|
|
def set(self, key: str, value: str, metadata: dict = None, user: str = None):
|
|
"""Set a secret value"""
|
|
with self.lock:
|
|
old_value = self._secrets_cache.get(key, {}).get('value')
|
|
|
|
self._secrets_cache[key] = {
|
|
'value': value,
|
|
'created': self._secrets_cache.get(key, {}).get('created', datetime.now().isoformat()),
|
|
'rotated': datetime.now().isoformat() if old_value else None,
|
|
'metadata': metadata or {}
|
|
}
|
|
|
|
self._audit('set' if not old_value else 'update', key, user)
|
|
self._save_secrets()
|
|
|
|
def delete(self, key: str, user: str = None):
|
|
"""Delete a secret"""
|
|
with self.lock:
|
|
if key in self._secrets_cache:
|
|
del self._secrets_cache[key]
|
|
self._audit('delete', key, user)
|
|
self._save_secrets()
|
|
return True
|
|
return False
|
|
|
|
def rotate(self, key: str, new_value: str = None, user: str = None):
|
|
"""Rotate a secret"""
|
|
with self.lock:
|
|
if key not in self._secrets_cache:
|
|
raise KeyError(f"Secret {key} not found")
|
|
|
|
old_value = self._secrets_cache[key]['value']
|
|
|
|
# Generate new value if not provided
|
|
if not new_value:
|
|
if key.endswith('_KEY') or key.endswith('_TOKEN'):
|
|
new_value = secrets.token_urlsafe(32)
|
|
elif key.endswith('_PASSWORD'):
|
|
new_value = secrets.token_urlsafe(24)
|
|
else:
|
|
raise ValueError(f"Cannot auto-generate value for {key}")
|
|
|
|
# Update secret
|
|
self._secrets_cache[key]['value'] = new_value
|
|
self._secrets_cache[key]['rotated'] = datetime.now().isoformat()
|
|
|
|
self._audit('rotate', key, user, {'generated': new_value is None})
|
|
self._save_secrets()
|
|
|
|
return old_value, new_value
|
|
|
|
def list_secrets(self) -> List[Dict[str, Any]]:
|
|
"""List all secrets (without values)"""
|
|
secrets_list = []
|
|
for key, data in self._secrets_cache.items():
|
|
secrets_list.append({
|
|
'key': key,
|
|
'created': data.get('created'),
|
|
'rotated': data.get('rotated'),
|
|
'metadata': data.get('metadata', {}),
|
|
'has_value': bool(data.get('value'))
|
|
})
|
|
return secrets_list
|
|
|
|
def add_validator(self, key: str, validator):
|
|
"""Add a validator function for a secret"""
|
|
self._validators[key] = validator
|
|
|
|
def validate(self, key: str, value: str) -> bool:
|
|
"""Validate a secret value"""
|
|
if key in self._validators:
|
|
try:
|
|
return self._validators[key](value)
|
|
except Exception as e:
|
|
logger.error(f"Validation failed for {key}: {e}")
|
|
return False
|
|
return True
|
|
|
|
def schedule_rotation(self, key: str, days: int):
|
|
"""Schedule automatic rotation for a secret"""
|
|
self._rotation_schedule[key] = {
|
|
'days': days,
|
|
'last_rotated': self._secrets_cache.get(key, {}).get('rotated', datetime.now().isoformat())
|
|
}
|
|
self._save_secrets()
|
|
|
|
def check_rotation_needed(self) -> List[str]:
|
|
"""Check which secrets need rotation"""
|
|
needs_rotation = []
|
|
now = datetime.now()
|
|
|
|
for key, schedule in self._rotation_schedule.items():
|
|
last_rotated = datetime.fromisoformat(schedule['last_rotated'])
|
|
if now - last_rotated > timedelta(days=schedule['days']):
|
|
needs_rotation.append(key)
|
|
|
|
return needs_rotation
|
|
|
|
def get_audit_log(self, key: str = None, limit: int = 100) -> List[Dict]:
|
|
"""Get audit log entries"""
|
|
logs = self._audit_log
|
|
|
|
if key:
|
|
logs = [log for log in logs if log['key'] == key]
|
|
|
|
return logs[-limit:]
|
|
|
|
def export_for_environment(self) -> Dict[str, str]:
|
|
"""Export secrets as environment variables"""
|
|
env_vars = {}
|
|
for key, data in self._secrets_cache.items():
|
|
env_key = f"SECRET_{key.upper()}"
|
|
env_vars[env_key] = data['value']
|
|
return env_vars
|
|
|
|
def verify_integrity(self) -> bool:
|
|
"""Verify integrity of secrets"""
|
|
try:
|
|
# Try to decrypt all secrets
|
|
for key, secret_data in self._secrets_cache.items():
|
|
if 'value' in secret_data:
|
|
# Re-encrypt and compare
|
|
encrypted = self._encrypt(secret_data['value'])
|
|
decrypted = self._decrypt(encrypted)
|
|
if decrypted != secret_data['value']:
|
|
logger.error(f"Integrity check failed for {key}")
|
|
return False
|
|
|
|
logger.info("Integrity check passed")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Integrity check failed: {e}")
|
|
return False
|
|
|
|
# Global instance
|
|
_secrets_manager = None
|
|
_secrets_lock = Lock()
|
|
|
|
def get_secrets_manager(config_file: str = None) -> SecretsManager:
|
|
"""Get or create global secrets manager instance"""
|
|
global _secrets_manager
|
|
|
|
with _secrets_lock:
|
|
if _secrets_manager is None:
|
|
_secrets_manager = SecretsManager(config_file)
|
|
return _secrets_manager
|
|
|
|
def get_secret(key: str, default: Any = None) -> Any:
|
|
"""Convenience function to get a secret"""
|
|
manager = get_secrets_manager()
|
|
return manager.get(key, default)
|
|
|
|
def set_secret(key: str, value: str, metadata: dict = None):
|
|
"""Convenience function to set a secret"""
|
|
manager = get_secrets_manager()
|
|
manager.set(key, value, metadata)
|
|
|
|
# Flask integration
|
|
def init_app(app):
|
|
"""Initialize secrets management for Flask app"""
|
|
manager = get_secrets_manager()
|
|
|
|
# Load secrets into app config
|
|
app.config['SECRET_KEY'] = manager.get('FLASK_SECRET_KEY') or app.config.get('SECRET_KEY')
|
|
app.config['TTS_API_KEY'] = manager.get('TTS_API_KEY') or app.config.get('TTS_API_KEY')
|
|
|
|
# Add secret manager to app
|
|
app.secrets_manager = manager
|
|
|
|
# Add CLI commands
|
|
@app.cli.command('secrets-list')
|
|
def list_secrets_cmd():
|
|
"""List all secrets"""
|
|
secrets = manager.list_secrets()
|
|
for secret in secrets:
|
|
print(f"{secret['key']}: created={secret['created']}, rotated={secret['rotated']}")
|
|
|
|
@app.cli.command('secrets-set')
|
|
def set_secret_cmd():
|
|
"""Set a secret"""
|
|
import click
|
|
key = click.prompt('Secret key')
|
|
value = click.prompt('Secret value', hide_input=True)
|
|
manager.set(key, value, user='cli')
|
|
print(f"Secret {key} set successfully")
|
|
|
|
@app.cli.command('secrets-rotate')
|
|
def rotate_secret_cmd():
|
|
"""Rotate a secret"""
|
|
import click
|
|
key = click.prompt('Secret key to rotate')
|
|
old_value, new_value = manager.rotate(key, user='cli')
|
|
print(f"Secret {key} rotated successfully")
|
|
print(f"New value: {new_value}")
|
|
|
|
@app.cli.command('secrets-check-rotation')
|
|
def check_rotation_cmd():
|
|
"""Check which secrets need rotation"""
|
|
needs_rotation = manager.check_rotation_needed()
|
|
if needs_rotation:
|
|
print("Secrets needing rotation:")
|
|
for key in needs_rotation:
|
|
print(f" - {key}")
|
|
else:
|
|
print("No secrets need rotation")
|
|
|
|
logger.info("Secrets management initialized") |