claude-force: Skill for Claude Code

Secrets Management Production

Production-grade secrets management for cryptocurrency trading systems.

AWS Secrets Manager Integration

import boto3
import json
from typing import Dict, Optional
from functools import lru_cache
from datetime import datetime, timedelta

class SecretsManager:
    """
    AWS Secrets Manager integration for production secrets
    NEVER use python-dotenv or environment variables in production!
    """

    def __init__(
        self,
        region_name: str = 'us-east-1',
        cache_ttl_seconds: int = 300  # 5 minutes
    ):
        self.client = boto3.client('secretsmanager', region_name=region_name)
        self.cache_ttl = timedelta(seconds=cache_ttl_seconds)
        self._cache = {}
        self._cache_timestamps = {}

    async def get_secret(self, secret_name: str, force_refresh: bool = False) -> Dict:
        """
        Retrieve secret from AWS Secrets Manager with local caching

        Args:
            secret_name: Name of secret in AWS Secrets Manager
            force_refresh: Bypass cache and fetch fresh secret

        Returns:
            Secret data as dictionary
        """
        # Check cache first (unless force refresh)
        if not force_refresh and secret_name in self._cache:
            cache_age = datetime.utcnow() - self._cache_timestamps[secret_name]

            if cache_age < self.cache_ttl:
                logger.debug(f"Using cached secret: {secret_name}")
                return self._cache[secret_name]

        # Fetch from AWS
        try:
            logger.info(f"Fetching secret from AWS: {secret_name}")

            response = self.client.get_secret_value(SecretId=secret_name)

            # Parse secret
            if 'SecretString' in response:
                secret_data = json.loads(response['SecretString'])
            else:
                secret_data = json.loads(response['SecretBinary'].decode('utf-8'))

            # Update cache
            self._cache[secret_name] = secret_data
            self._cache_timestamps[secret_name] = datetime.utcnow()

            return secret_data

        except self.client.exceptions.ResourceNotFoundException:
            logger.error(f"Secret not found: {secret_name}")
            raise
        except Exception as e:
            logger.error(f"Failed to retrieve secret {secret_name}: {e}")
            raise

    async def get_exchange_credentials(self, exchange_id: str) -> Dict:
        """
        Retrieve exchange API credentials

        Secret format in AWS Secrets Manager:
        {
            "api_key": "...",
            "secret": "...",
            "passphrase": "...",  // Optional (for some exchanges)
            "subaccount": "..."   // Optional
        }
        """
        secret_name = f"trading-bot/{exchange_id}/credentials"
        return await self.get_secret(secret_name)

    async def get_telegram_token(self) -> str:
        """Retrieve Telegram bot token"""
        secret = await self.get_secret("trading-bot/telegram/token")
        return secret['token']

    async def get_database_credentials(self) -> Dict:
        """Retrieve database credentials"""
        return await self.get_secret("trading-bot/database/credentials")

    async def rotate_exchange_credentials(
        self,
        exchange_id: str,
        new_api_key: str,
        new_secret: str,
        new_passphrase: Optional[str] = None
    ):
        """
        Rotate exchange API credentials

        Steps:
        1. Create new credentials on exchange
        2. Update AWS Secrets Manager
        3. Wait for cache expiry or force refresh
        4. Revoke old credentials on exchange
        """
        secret_name = f"trading-bot/{exchange_id}/credentials"

        # Prepare new secret
        new_secret_data = {
            "api_key": new_api_key,
            "secret": new_secret,
        }

        if new_passphrase:
            new_secret_data["passphrase"] = new_passphrase

        # Update secret in AWS
        try:
            self.client.update_secret(
                SecretId=secret_name,
                SecretString=json.dumps(new_secret_data)
            )

            logger.info(f"Rotated credentials for {exchange_id}")

            # Force cache refresh
            await self.get_secret(secret_name, force_refresh=True)

        except Exception as e:
            logger.error(f"Failed to rotate credentials for {exchange_id}: {e}")
            raise