video-accessibility/backend/setup_secrets.py

#!/usr/bin/env python3
"""
Setup script for creating secrets in Google Cloud Secret Manager.

This script creates all required secrets for the Accessible Video Platform.
Run this once before deploying to production.

Usage:
    python setup_secrets.py --project-id YOUR_PROJECT_ID
    python setup_secrets.py --project-id YOUR_PROJECT_ID --env prod
"""

import argparse
import asyncio
import os
import secrets
import string
from typing import Dict

from app.services.secrets_manager import secrets_manager, SecretManagerError


def generate_secure_key(length: int = 64) -> str:
    """Generate a cryptographically secure random key."""
    alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
    return ''.join(secrets.choice(alphabet) for _ in range(length))


async def create_secrets(project_id: str, environment: str = "prod") -> None:
    """Create all required secrets in Secret Manager."""
    
    # Configure project ID
    os.environ["GOOGLE_CLOUD_PROJECT"] = project_id
    
    # Define secrets to create
    secrets_config = {
        "mongodb-url": {
            "description": "MongoDB Atlas connection string",
            "example": "mongodb+srv://user:password@cluster.mongodb.net/accessible_video?retryWrites=true&w=majority",
            "generate": False
        },
        "redis-url": {
            "description": "Redis connection URL", 
            "example": "redis://redis-instance:6379/0",
            "generate": False
        },
        "jwt-secret": {
            "description": "JWT signing secret key",
            "example": None,  # Will be generated
            "generate": True
        },
        "jwt-refresh-secret": {
            "description": "JWT refresh token secret key",
            "example": None,  # Will be generated
            "generate": True
        },
        "gemini-api-key": {
            "description": "Google Gemini API key",
            "example": "AIza...your-api-key",
            "generate": False
        },
        "sendgrid-api-key": {
            "description": "SendGrid API key for email notifications",
            "example": "SG.xxx.xxx-your-sendgrid-key",
            "generate": False
        },
        "elevenlabs-api-key": {
            "description": "ElevenLabs API key for text-to-speech",
            "example": "el_xxx_your-elevenlabs-key",
            "generate": False
        },
        "sentry-dsn": {
            "description": "Sentry DSN for error tracking",
            "example": "https://xxx@xxx.ingest.sentry.io/xxx",
            "generate": False
        }
    }
    
    print(f"🔐 Setting up secrets for project: {project_id}")
    print(f"Environment: {environment}")
    print("=" * 50)
    
    labels = {
        "environment": environment,
        "service": "accessible-video-platform"
    }
    
    for secret_name, config in secrets_config.items():
        try:
            if config["generate"]:
                # Generate secure key
                secret_value = generate_secure_key()
                print(f"✅ Generated secure key for {secret_name}")
            else:
                # Prompt user for value
                print(f"\n📝 {config['description']}")
                if config["example"]:
                    print(f"Example: {config['example']}")
                
                secret_value = input(f"Enter value for {secret_name}: ").strip()
                
                if not secret_value:
                    print(f"⏭️  Skipping {secret_name} (empty value)")
                    continue
            
            # Create the secret
            await secrets_manager.create_secret(secret_name, secret_value, labels)
            print(f"✅ Created secret: {secret_name}")
            
        except SecretManagerError as e:
            if "already exists" in str(e).lower():
                print(f"ℹ️  Secret {secret_name} already exists")
            else:
                print(f"❌ Failed to create secret {secret_name}: {e}")
        except KeyboardInterrupt:
            print(f"\n❌ Setup cancelled by user")
            break
        except Exception as e:
            print(f"❌ Unexpected error creating {secret_name}: {e}")
    
    print("\n🎉 Secret setup completed!")
    print("\n📋 Next steps:")
    print("1. Verify all secrets are created in the GCP Console")
    print("2. Ensure your service accounts have secretmanager.secretAccessor role")
    print("3. Deploy your application using the Cloud Run configurations")


async def list_secrets() -> None:
    """List all existing secrets in the project."""
    try:
        # This would require additional implementation to list secrets
        print("📋 Listing existing secrets...")
        print("(Feature not implemented - check GCP Console)")
    except Exception as e:
        print(f"❌ Failed to list secrets: {e}")


async def test_secrets() -> None:
    """Test secret retrieval to ensure everything is working."""
    print("🧪 Testing secret retrieval...")
    
    test_secrets = [
        "jwt-secret",
        "jwt-refresh-secret"
    ]
    
    for secret_name in test_secrets:
        try:
            value = await secrets_manager.get_secret(secret_name)
            if value:
                print(f"✅ Successfully retrieved {secret_name}")
            else:
                print(f"❌ Empty value for {secret_name}")
        except SecretManagerError as e:
            print(f"❌ Failed to retrieve {secret_name}: {e}")


def main():
    """Main CLI entry point."""
    parser = argparse.ArgumentParser(description="Setup secrets for Accessible Video Platform")
    parser.add_argument("--project-id", required=True, help="Google Cloud Project ID")
    parser.add_argument("--env", default="prod", choices=["dev", "staging", "prod"], help="Environment")
    parser.add_argument("--list", action="store_true", help="List existing secrets")
    parser.add_argument("--test", action="store_true", help="Test secret retrieval")
    
    args = parser.parse_args()
    
    if args.list:
        asyncio.run(list_secrets())
    elif args.test:
        asyncio.run(test_secrets())
    else:
        asyncio.run(create_secrets(args.project_id, args.env))


if __name__ == "__main__":
    main()