malepati/custom_template_working

1import boto32import json3import os4from cryptography.fernet import Fernet5from botocore.exceptions import NoCredentialsError, PartialCredentialsError, ClientError6 7def get_secret(secret_name: str, region_name: str = "us-east-1"):8    """9    Retrieve and decrypt secret from AWS Secrets Manager.10    """11    try:12        # ✅ HARDCODED AWS CREDENTIALS (AS REQUESTED BY USER)13        aws_access_key = "AKIAT4R3SOPSJGE7NEV6"14        aws_secret_key = "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"15        16        session = boto3.session.Session(17            aws_access_key_id=aws_access_key,18            aws_secret_access_key=aws_secret_key,19            region_name=region_name20        )21        22        client = session.client("secretsmanager", region_name=region_name)23 24        # Fetch the secret25        response = client.get_secret_value(SecretId=secret_name)26 27    except ClientError as e:28        print(f"❌ AWS Client Error: {e}")29        return None30    except Exception as e:31        print(f"❌ Error retrieving secret: {e}")32        return None33 34    # Parse JSON from AWS Secrets Manager35    if "SecretString" not in response:36        print("❌ SecretString missing in response.")37        return None38 39    secret_dict = json.loads(response["SecretString"])40    41    # Retrieve Fernet key42    fernet_key = secret_dict.get("Fernet_Key")43    if not fernet_key:44        print("❌ Fernet_Key missing in secret.")45        return None46 47    try:48        cipher = Fernet(fernet_key)49    except Exception as e:50        print(f"❌ Error initializing Fernet cipher: {e}")51        return None52 53    # Decrypt all fields54    decrypted_data = {}55    56    for key, value in secret_dict.items():57        if key == "Fernet_Key":58            decrypted_data[key] = value59        else:60            try:61                # Handle both string and non-string values safely62                val_to_decrypt = str(value) if not isinstance(value, str) else value63                decrypted_value = cipher.decrypt(val_to_decrypt.encode()).decode()64                decrypted_data[key] = decrypted_value65            except Exception as e:66                print(f"⚠️ Failed to decrypt {key}: {e}")67                decrypted_data[key] = None68 69    return decrypted_data70 71 72def load_secrets_to_env():73    """74    Fetches secrets and injects them into os.environ.75    Should be called at application startup.76    """77    print("🚀 Loading secrets from AWS Secrets Manager...")78    secret_name = "Demo/MR/skeys"79    region_name = "us-east-1"80 81    secrets = get_secret(secret_name, region_name)82 83    if secrets:84        # List of keys to inject into environment85        keys_to_inject = [86            "OPENAI_API_KEY", "OPENAI_MODEL_NAME", "OPENAI_MODEL_TURBO",87            "CONNECTIONS_HOST", "CONNECTIONS_DB", "CONNECTIONS_USER", "CONNECTIONS_PASS",88            "REDIS_HOST", "REDIS_PORT", "REDIS_DB", "REDIS_PASSWORD", "REDIS_USER", "REDIS_URL",89            "QDRANT_HOST", "QDRANT_PORT", "QDRANT_COLLECTION_NAME", "QDRANT_URL", "QDRANT_API_KEY",90            "BUCKET_NAME", "BUCKET_UNSTRUCTURED_STORAGE", "GEMINI_API_KEY", "QDRANT_HOST_AWS",91            "ALLOWED_HOSTS"92        ]93 94        for key in keys_to_inject:95            val = secrets.get(key)96            if val:97                os.environ[key] = str(val)98                # Mask key in logs99                masked_val = f"{val[:5]}...{val[-4:]}" if val and len(val) > 10 else "***"100                print(f"✅ Loaded {key}: {masked_val}")101            # Special case for AWS S3 Creds if they exist in secret102            if key == "aws_access_key_id" and secrets.get(key):103                 os.environ["AWS_ACCESS_KEY_ID"] = secrets.get(key)104            if key == "aws_secret_access_key" and secrets.get(key):105                 os.environ["AWS_SECRET_ACCESS_KEY"] = secrets.get(key)106 107        # Set specific defaults as requested by user ("there only you set dalle-3")108        os.environ["IMAGE_PROVIDER"] = "dall-e-3" 109        print("✅ Set IMAGE_PROVIDER to dall-e-3 (default)")110        111        print("✅ Secrets loaded successfully.")112    else:113        print("❌ Failed to load secrets. Application may not function correctly.")114