aws_secrets.pyโข5.14 kB
"""
AWS Secrets Manager utilities for retrieving database credentials and API keys.
"""
import json
import os
from typing import Dict, Any, Optional
import boto3
from botocore.exceptions import ClientError
def get_secret_value(secret_id: str, region: str = None) -> Optional[Dict[str, Any]]:
"""
Retrieve a secret value from AWS Secrets Manager.
Args:
secret_id: The ARN or name of the secret
region: AWS region (defaults to environment variable or eu-west-2)
Returns:
Dictionary containing the secret value, or None if not found
"""
if not region:
region = os.getenv('AWS_REGION', 'eu-west-2')
try:
# Create secrets manager client
secrets_client = boto3.client('secretsmanager', region_name=region)
# Get the secret value
response = secrets_client.get_secret_value(SecretId=secret_id)
# Parse the JSON secret string
return json.loads(response['SecretString'])
except ClientError as e:
import sys
print(f"Error retrieving secret {secret_id}: {e}", file=sys.stderr)
return None
except json.JSONDecodeError as e:
import sys
print(f"Error parsing secret JSON for {secret_id}: {e}", file=sys.stderr)
return None
except Exception as e:
import sys
print(f"Unexpected error retrieving secret {secret_id}: {e}", file=sys.stderr)
return None
def get_database_credentials() -> Optional[str]:
"""
Get PostgreSQL database connection string from AWS Secrets Manager.
Returns:
PostgreSQL connection string, or None if not available
"""
secret_arn = os.getenv('DATABASE_SECRET_ARN')
if not secret_arn:
import sys
print("DATABASE_SECRET_ARN environment variable not set", file=sys.stderr)
return None
credentials = get_secret_value(secret_arn)
if not credentials:
return None
# Build PostgreSQL connection string
return f"postgresql://{credentials['username']}:{credentials['password']}@{credentials['host']}:{credentials['port']}/{credentials['dbname']}"
def get_openai_api_key() -> Optional[str]:
"""
Get OpenAI API key from AWS Secrets Manager.
Returns:
OpenAI API key, or None if not available
"""
secret_arn = os.getenv('OPENAI_SECRET_ARN')
if not secret_arn:
import sys
print("OPENAI_SECRET_ARN environment variable not set", file=sys.stderr)
return None
credentials = get_secret_value(secret_arn)
if not credentials:
return None
return credentials.get('api_key')
def get_x402_private_key() -> Optional[str]:
"""
Get X402 private key from AWS Secrets Manager.
Returns:
X402 private key, or None if not available
"""
secret_arn = os.getenv('X402_SECRET_ARN')
if not secret_arn:
import sys
print("X402_SECRET_ARN environment variable not set", file=sys.stderr)
return None
credentials = get_secret_value(secret_arn)
if not credentials:
return None
return credentials.get('private_key')
def setup_environment_from_secrets():
"""
Set up environment variables from AWS Secrets Manager.
This allows existing code to work without modification.
"""
import sys
# Use stderr for logging to avoid MCP JSON parsing issues
def log(message):
print(message, file=sys.stderr)
log("๐ Retrieving secrets from AWS Secrets Manager...")
# Get database URL
db_url = get_database_credentials()
if db_url:
os.environ['DATABASE_URL'] = db_url
log("โ
Set DATABASE_URL from AWS Secrets Manager")
else:
log("โ Failed to get DATABASE_URL from AWS Secrets Manager")
# Get OpenAI API key
openai_key = get_openai_api_key()
if openai_key and openai_key.strip() and not openai_key.strip().startswith('your_'):
os.environ['OPENAI_API_KEY'] = openai_key
log("โ
Set OPENAI_API_KEY from AWS Secrets Manager")
else:
log("โ OpenAI API key not set or contains placeholder value")
log(" Update with: aws secretsmanager update-secret --secret-id 'timelooker/openai/api-key' --secret-string '{\"api_key\":\"sk-your-actual-key\"}' --region eu-west-2")
# Get X402 private key
x402_key = get_x402_private_key()
if x402_key and x402_key.strip() and not x402_key.strip().startswith('your_'):
os.environ['PRIVATE_KEY'] = x402_key
log("โ
Set PRIVATE_KEY from AWS Secrets Manager")
else:
log("โ X402 private key not set or contains placeholder value")
log(" Update with: aws secretsmanager update-secret --secret-id 'timelooker/x402/private-key' --secret-string '{\"private_key\":\"0x...\"}' --region eu-west-2")
def is_cloud_mode() -> bool:
"""
Check if we're running in cloud mode (AWS deployment).
Returns:
True if cloud mode is enabled, False otherwise
"""
return os.getenv('DEPLOY_TO_CLOUD', 'false').lower() == 'true'