# # import boto3 # # import json # # from cryptography.fernet import Fernet # # from botocore.exceptions import NoCredentialsError, PartialCredentialsError # # import os # # def get_secret(secret_name: str, region_name: str = "us-east-1"): # # """ # # Retrieve and decrypt secret from AWS Secrets Manager. # # """ # # # Create AWS Secrets Manager client # # session = boto3.session.Session() # # client = session.client("secretsmanager", region_name=region_name) # # try: # # response = client.get_secret_value(SecretId=secret_name) # # except NoCredentialsError: # # print("AWS credentials not available. Configure with 'aws configure'.") # # return None # # except PartialCredentialsError: # # print("Incomplete AWS credentials provided.") # # return None # # except Exception as e: # # print(f"Error retrieving secret: {e}") # # return None # # # Parse the secret JSON string # # if "SecretString" in response: # # secret_dict = json.loads(response["SecretString"]) # # else: # # print("SecretString not found in response.") # # return None # # # Get Fernet key # # fernet_key = secret_dict.get("Fernet_Key") # # if not fernet_key: # # print("Fernet_Key not found in secret.") # # return None # # # Initialize Fernet cipher # # cipher = Fernet(fernet_key) # # # Decrypt all values in the secret # # decrypted_data = {} # # for key, value in secret_dict.items(): # # if key == "Fernet_Key": # # decrypted_data[key] = value # # else: # # try: # # decrypted_data[key] = cipher.decrypt(value.encode()).decode() # # except Exception as e: # # decrypted_data[key] = f"⚠️ Failed to decrypt: {str(e)}" # # # Extract all possible connection details # # CONNECTIONS_HOST = decrypted_data.get("CONNECTIONS_HOST", "") # # CONNECTIONS_DB = decrypted_data.get("CONNECTIONS_DB", "") # # CONNECTIONS_USER = decrypted_data.get("CONNECTIONS_USER", "") # # CONNECTIONS_PASS = decrypted_data.get("CONNECTIONS_PASS", "") # # OPENAI_API_KEY = decrypted_data.get("OPENAI_API_KEY", "") # # OPENAI_MODEL_NAME = decrypted_data.get("OPENAI_MODEL_NAME", "") # # OPENAI_MODEL_TURBO = decrypted_data.get("OPENAI_MODEL_TURBO", "") # # REDIS_HOST = decrypted_data.get("REDIS_HOST", "") # # REDIS_PORT = decrypted_data.get("REDIS_PORT", "") # # REDIS_DB = decrypted_data.get("REDIS_DB", "") # # REDIS_PASSWORD = decrypted_data.get("REDIS_PASSWORD", "") # # REDIS_USER = decrypted_data.get("REDIS_USER", "") # # REDIS_URL = decrypted_data.get("REDIS_URL", "") # # QDRANT_HOST = decrypted_data.get("QDRANT_HOST", "") # # QDRANT_PORT = decrypted_data.get("QDRANT_PORT", "") # # QDRANT_COLLECTION_NAME = decrypted_data.get("QDRANT_COLLECTION_NAME", "") # # QDRANT_URL = decrypted_data.get("QDRANT_URL", "") # # QDRANT_API_KEY = decrypted_data.get("QDRANT_API_KEY", "") # # os.environ["CONNECTIONS_HOST"]= CONNECTIONS_HOST # # return ( # # decrypted_data, # # CONNECTIONS_HOST, # # CONNECTIONS_DB, # # CONNECTIONS_USER, # # CONNECTIONS_PASS, # # OPENAI_API_KEY, # # OPENAI_MODEL_NAME, # # OPENAI_MODEL_TURBO, # # REDIS_HOST, # # REDIS_PORT, # # REDIS_DB, # # REDIS_PASSWORD, # # REDIS_USER, # # QDRANT_HOST, # # QDRANT_PORT, # # QDRANT_COLLECTION_NAME, # # QDRANT_URL, # # QDRANT_API_KEY, # # REDIS_URL, # # ) # # if __name__ == "__main__": # # secret_name = "Demo/MR/skeys" # # region_name = "us-east-1" # # result = get_secret(secret_name, region_name) # # if result: # # ( # # decrypted_data, # # CONNECTIONS_HOST, # # CONNECTIONS_DB, # # CONNECTIONS_USER, # # CONNECTIONS_PASS, # # OPENAI_API_KEY, # # OPENAI_MODEL_NAME, # # OPENAI_MODEL_TURBO, # # REDIS_HOST, # # REDIS_PORT, # # REDIS_DB, # # REDIS_PASSWORD, # # REDIS_USER, # # QDRANT_HOST, # # QDRANT_PORT, # # QDRANT_COLLECTION_NAME, # # QDRANT_URL, # # QDRANT_API_KEY, # # REDIS_URL, # # ) = result # # print("\n Successfully retrieved and decrypted secret!\n") # # # Print all decrypted values # # print(" All Decrypted Secret Values:") # # for key, val in decrypted_data.items(): # # print(f" {key}: {val}") # # print("\n--- DATABASE CONNECTION ---") # # print(f"HOST: {CONNECTIONS_HOST}") # # print(f"DB: {CONNECTIONS_DB}") # # print(f"USER: {CONNECTIONS_USER}") # # print(f"PASS: {CONNECTIONS_PASS}") # # print("\n--- OPENAI CONFIG ---") # # print(f"API_KEY: {OPENAI_API_KEY}") # # print(f"MODEL_NAME: {OPENAI_MODEL_NAME}") # # print(f"MODEL_TURBO: {OPENAI_MODEL_TURBO}") # # print("\n--- REDIS CONFIG ---") # # print(f"HOST: {REDIS_HOST}") # # print(f"PORT: {REDIS_PORT}") # # print(f"DB: {REDIS_DB}") # # print(f"USER: {REDIS_USER}") # # print(f"PASSWORD: {REDIS_PASSWORD}") # # print(f"URL: {REDIS_URL}") # # print("\n--- QDRANT CONFIG ---") # # print(f"HOST: {QDRANT_HOST}") # # print(f"PORT: {QDRANT_PORT}") # # print(f"COLLECTION: {QDRANT_COLLECTION_NAME}") # # print(f"URL: {QDRANT_URL}") # # print(f"API_KEY: {QDRANT_API_KEY}") # # else: # # print("Failed to retrieve or decrypt secret.") # import boto3 # import json # import os # from cryptography.fernet import Fernet # from botocore.exceptions import NoCredentialsError, PartialCredentialsError # from dotenv import load_dotenv # # Load environment variables (AWS credentials, etc.) # load_dotenv() # def get_secret(secret_name: str, region_name: str = "us-east-1"): # """ # Retrieve and decrypt secrets from AWS Secrets Manager. # """ # try: # # ✅ Use explicit credentials (as in the old working version) # session = boto3.session.Session( # aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"), # aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"), # aws_session_token=os.getenv("AWS_SESSION_TOKEN") # ) # client = session.client("secretsmanager", region_name=region_name) # # Retrieve the secret # response = client.get_secret_value(SecretId=secret_name) # except NoCredentialsError: # print("❌ AWS credentials not available. Configure them or use .env file.") # return None # except PartialCredentialsError: # print("❌ Incomplete AWS credentials provided.") # return None # except Exception as e: # print(f"❌ Error retrieving secret: {e}") # return None # # Parse the secret JSON string # if "SecretString" in response: # secret_dict = json.loads(response["SecretString"]) # else: # print("❌ SecretString not found in AWS response.") # return None # # Retrieve and verify Fernet key # fernet_key = secret_dict.get("Fernet_Key") # if not fernet_key: # print("❌ Fernet_Key missing from secret.") # return None # # Initialize Fernet cipher # cipher = Fernet(fernet_key) # # Decrypt all values # decrypted_data = {} # for key, value in secret_dict.items(): # if key == "Fernet_Key": # decrypted_data[key] = value # else: # try: # decrypted_data[key] = cipher.decrypt(value.encode()).decode() # except Exception as e: # decrypted_data[key] = f"⚠️ Failed to decrypt: {str(e)}" # # Extract key values (optional — for quick use) # CONNECTIONS_HOST = decrypted_data.get("CONNECTIONS_HOST", "") # CONNECTIONS_DB = decrypted_data.get("CONNECTIONS_DB", "") # CONNECTIONS_USER = decrypted_data.get("CONNECTIONS_USER", "") # CONNECTIONS_PASS = decrypted_data.get("CONNECTIONS_PASS", "") # OPENAI_API_KEY = decrypted_data.get("OPENAI_API_KEY", "") # OPENAI_MODEL_NAME = decrypted_data.get("OPENAI_MODEL_NAME", "") # OPENAI_MODEL_TURBO = decrypted_data.get("OPENAI_MODEL_TURBO", "") # REDIS_HOST = decrypted_data.get("REDIS_HOST", "") # REDIS_PORT = decrypted_data.get("REDIS_PORT", "") # REDIS_DB = decrypted_data.get("REDIS_DB", "") # REDIS_PASSWORD = decrypted_data.get("REDIS_PASSWORD", "") # REDIS_USER = decrypted_data.get("REDIS_USER", "") # REDIS_URL = decrypted_data.get("REDIS_URL", "") # QDRANT_HOST = decrypted_data.get("QDRANT_HOST", "") # QDRANT_PORT = decrypted_data.get("QDRANT_PORT", "") # QDRANT_COLLECTION_NAME = decrypted_data.get("QDRANT_COLLECTION_NAME", "") # QDRANT_URL = decrypted_data.get("QDRANT_URL", "") # QDRANT_API_KEY = decrypted_data.get("QDRANT_API_KEY", "") # return ( # decrypted_data, # CONNECTIONS_HOST, # CONNECTIONS_DB, # CONNECTIONS_USER, # CONNECTIONS_PASS, # OPENAI_API_KEY, # OPENAI_MODEL_NAME, # OPENAI_MODEL_TURBO, # REDIS_HOST, # REDIS_PORT, # REDIS_DB, # REDIS_PASSWORD, # REDIS_USER, # QDRANT_HOST, # QDRANT_PORT, # QDRANT_COLLECTION_NAME, # QDRANT_URL, # QDRANT_API_KEY, # REDIS_URL, # ) # if __name__ == "__main__": # secret_name = "Demo/MR/skeys" # region_name = "us-east-1" # result = get_secret(secret_name, region_name) # if result: # ( # decrypted_data, # CONNECTIONS_HOST, # CONNECTIONS_DB, # CONNECTIONS_USER, # CONNECTIONS_PASS, # OPENAI_API_KEY, # OPENAI_MODEL_NAME, # OPENAI_MODEL_TURBO, # REDIS_HOST, # REDIS_PORT, # REDIS_DB, # REDIS_PASSWORD, # REDIS_USER, # QDRANT_HOST, # QDRANT_PORT, # QDRANT_COLLECTION_NAME, # QDRANT_URL, # QDRANT_API_KEY, # REDIS_URL, # ) = result # print("\n✅ Successfully retrieved and decrypted secret!\n") # # Print all decrypted values # print("🔐 All Decrypted Secret Values:") # for key, val in decrypted_data.items(): # print(f" {key}: {val}") # print("\n--- DATABASE CONNECTION ---") # print(f"HOST: {CONNECTIONS_HOST}") # print(f"DB: {CONNECTIONS_DB}") # print(f"USER: {CONNECTIONS_USER}") # print(f"PASS: {CONNECTIONS_PASS}") # print("\n--- OPENAI CONFIG ---") # print(f"API_KEY: {OPENAI_API_KEY}") # print(f"MODEL_NAME: {OPENAI_MODEL_NAME}") # print(f"MODEL_TURBO: {OPENAI_MODEL_TURBO}") # print("\n--- REDIS CONFIG ---") # print(f"HOST: {REDIS_HOST}") # print(f"PORT: {REDIS_PORT}") # print(f"DB: {REDIS_DB}") # print(f"USER: {REDIS_USER}") # print(f"PASSWORD: {REDIS_PASSWORD}") # print(f"URL: {REDIS_URL}") # print("\n--- QDRANT CONFIG ---") # print(f"HOST: {QDRANT_HOST}") # print(f"PORT: {QDRANT_PORT}") # print(f"COLLECTION: {QDRANT_COLLECTION_NAME}") # print(f"URL: {QDRANT_URL}") # print(f"API_KEY: {QDRANT_API_KEY}") # else: # print("❌ Failed to retrieve or decrypt secret.") # old working code # import boto3 # import json # import os # from cryptography.fernet import Fernet # from botocore.exceptions import NoCredentialsError, PartialCredentialsError # from dotenv import load_dotenv # # ✅ Load environment variables for AWS credentials # load_dotenv() # def get_secret(secret_name: str, region_name: str = "us-east-1"): # """ # Retrieve and decrypt secret from AWS Secrets Manager. # Loads AWS credentials from .env file. # """ # try: # # ✅ Explicit AWS session using .env credentials # session = boto3.session.Session( # aws_access_key_id=os.getenv("AWS_S3_CREDS_KEY_ID"), # aws_secret_access_key=os.getenv("AWS_S3_CREDS_SECRET_KEY"), # aws_session_token=os.getenv("AWS_SESSION_TOKEN") # ) # client = session.client("secretsmanager", region_name=region_name) # # Fetch the secret # response = client.get_secret_value(SecretId=secret_name) # except NoCredentialsError: # # print("❌ AWS credentials not available. Please check your .env file.") # return None # except PartialCredentialsError: # # print("❌ Incomplete AWS credentials found in .env.") # return None # except Exception as e: # # print(f"❌ Error retrieving secret: {e}") # return None # # ✅ Parse JSON from AWS Secrets Manager # if "SecretString" not in response: # # print("❌ SecretString missing in response.") # return None # secret_dict = json.loads(response["SecretString"]) # print("secret dict",secret_dict) # # ✅ Retrieve Fernet key # fernet_key = secret_dict.get("Fernet_Key") # print("fernet_key",fernet_key) # if not fernet_key: # # print("❌ Fernet_Key missing in secret.") # return None # cipher = Fernet(fernet_key) # print("cipher",cipher) # # ✅ Decrypt all fields # decrypted_data = {} # for key, value in secret_dict.items(): # if key == "Fernet_Key": # decrypted_data[key] = value # else: # try: # decrypted_data[key] = cipher.decrypt(value.encode()).decode() # except Exception as e: # decrypted_data[key] = f"⚠️ Failed to decrypt ({e})" # # ✅ Extract individual variables # # cipher_suite = Fernet(fernet_key) # CONNECTIONS_HOST = decrypted_data.get("CONNECTIONS_HOST", "") # CONNECTIONS_DB = decrypted_data.get("CONNECTIONS_DB", "") # CONNECTIONS_USER = decrypted_data.get("CONNECTIONS_USER", "") # CONNECTIONS_PASS = decrypted_data.get("CONNECTIONS_PASS", "") # OPENAI_API_KEY = decrypted_data.get("OPENAI_API_KEY", "") # OPENAI_MODEL_NAME = decrypted_data.get("OPENAI_MODEL_NAME", "") # OPENAI_MODEL_TURBO = decrypted_data.get("OPENAI_MODEL_TURBO", "") # REDIS_HOST = decrypted_data.get("REDIS_HOST", "") # REDIS_PORT = decrypted_data.get("REDIS_PORT", "") # REDIS_DB = decrypted_data.get("REDIS_DB", "") # REDIS_PASSWORD = decrypted_data.get("REDIS_PASSWORD", "") # REDIS_USER = decrypted_data.get("REDIS_USER", "") # REDIS_URL = decrypted_data.get("REDIS_URL", "") # QDRANT_HOST = decrypted_data.get("QDRANT_HOST", "") # QDRANT_PORT = decrypted_data.get("QDRANT_PORT", "") # QDRANT_COLLECTION_NAME = decrypted_data.get("QDRANT_COLLECTION_NAME", "") # QDRANT_URL = decrypted_data.get("QDRANT_URL", "") # QDRANT_API_KEY = decrypted_data.get("QDRANT_API_KEY", "") # ACCESS_KEY = decrypted_data.get("ACCESS_KEY", "") # SECRET_KEY = decrypted_data.get("SECRET_KEY", "") # BUCKET_NAME = decrypted_data.get("BUCKET_NAME", "") # aws_access_key_id = decrypted_data.get("aws_access_key_id", "") # aws_secret_access_key = decrypted_data.get("aws_secret_access_key", "") # aws_session_token = decrypted_data.get("aws_session_token", "") # # AWS_S3_CREDS_KEY_ID = decrypted_data.get("AWS_S3_CREDS_KEY_ID","") # # print("AWS_S3_CREDS_KEY_ID", AWS_S3_CREDS_KEY_ID) # # AWS_S3_CREDS_SECRET_KEY = decrypted_data.get("AWS_S3_CREDS_SECRET_KEY","") # # print("AWS_S3_CREDS_SECRET_KEY", AWS_S3_CREDS_SECRET_KEY) # # print(f"OPENAI_API_KEY: {OPENAI_API_KEY}") # return ( # decrypted_data, # CONNECTIONS_HOST, # CONNECTIONS_DB, # CONNECTIONS_USER, # CONNECTIONS_PASS, # OPENAI_API_KEY, # OPENAI_MODEL_NAME, # OPENAI_MODEL_TURBO, # REDIS_HOST, # REDIS_PORT, # REDIS_DB, # REDIS_PASSWORD, # REDIS_USER, # QDRANT_HOST, # QDRANT_PORT, # QDRANT_COLLECTION_NAME, # QDRANT_URL, # QDRANT_API_KEY, # REDIS_URL, # BUCKET_NAME, # aws_access_key_id, # aws_secret_access_key # ) # print("✅ Successfully retrieved and decrypted secret!") # # 👇 Runs automatically when imported (like old version) # secret_name = "Demo/MR/skeys" # region_name = "us-east-1" # # secrets = get_secret(secret_name, region_name) # # if secrets: # # ( # # decrypted_data, # # CONNECTIONS_HOST, # # CONNECTIONS_DB, # # CONNECTIONS_USER, # # CONNECTIONS_PASS, # # OPENAI_API_KEY, # # OPENAI_MODEL_NAME, # # OPENAI_MODEL_TURBO, # # REDIS_HOST, # # REDIS_PORT, # # REDIS_DB, # # REDIS_PASSWORD, # # REDIS_USER, # # QDRANT_HOST, # # QDRANT_PORT, # # QDRANT_COLLECTION_NAME, # # QDRANT_URL, # # QDRANT_API_KEY, # # REDIS_URL, # # ) = secrets # # # print("\n✅ Successfully retrieved and decrypted secret!\n") # # # print("🔐 All Decrypted Secret Values:") # # # for key, val in decrypted_data.items(): # # # print(f" {key}: {val}") # # print("\n--- DATABASE CONNECTION ---") # # # print(f"HOST: {CONNECTIONS_HOST}") # # # print(f"DB: {CONNECTIONS_DB}") # # # print(f"USER: {CONNECTIONS_USER}") # # # print(f"PASS: {CONNECTIONS_PASS}") # # # print("\n--- OPENAI CONFIG ---") # # print(f"API_KEY: {OPENAI_API_KEY}") # # # print(f"MODEL_NAME: {OPENAI_MODEL_NAME}") # # # print(f"MODEL_TURBO: {OPENAI_MODEL_TURBO}") # # # print("\n--- REDIS CONFIG ---") # # # print(f"HOST: {REDIS_HOST}") # # # print(f"PORT: {REDIS_PORT}") # # # print(f"DB: {REDIS_DB}") # # # print(f"USER: {REDIS_USER}") # # # print(f"PASSWORD: {REDIS_PASSWORD}") # # # print(f"URL: {REDIS_URL}") # # # print("\n--- QDRANT CONFIG ---") # # # print(f"HOST: {QDRANT_HOST}") # # # print(f"PORT: {QDRANT_PORT}") # # # print(f"COLLECTION: {QDRANT_COLLECTION_NAME}") # # # print(f"URL: {QDRANT_URL}") # # print(f"API_KEY: {QDRANT_API_KEY}") # # else: # # raise Exception("❌ Failed to retrieve or decrypt secret.") # secrets = get_secret(secret_name, region_name) # if secrets: # ( # decrypted_data, CONNECTIONS_HOST, CONNECTIONS_DB, CONNECTIONS_USER, CONNECTIONS_PASS, # OPENAI_API_KEY, OPENAI_MODEL_NAME, OPENAI_MODEL_TURBO, # REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD, REDIS_USER, # QDRANT_HOST, QDRANT_PORT, QDRANT_COLLECTION_NAME, QDRANT_URL, QDRANT_API_KEY, REDIS_URL,BUCKET_NAME, aws_access_key_id,aws_secret_access_key # ) = secrets # # THIS LINE FIXES THE ERROR # # print("OPENAI_API_KEY: ", OPENAI_API_KEY) # os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY # os.environ["CONNECTIONS_HOST"] = CONNECTIONS_HOST # os.environ["CONNECTIONS_DB"] = CONNECTIONS_DB # os.environ["CONNECTIONS_USER"] = CONNECTIONS_USER # os.environ["CONNECTIONS_PASS"] = CONNECTIONS_PASS # os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY # os.environ["OPENAI_MODEL_NAME"] = OPENAI_MODEL_NAME # os.environ["OPENAI_MODEL_TURBO"] = OPENAI_MODEL_TURBO # os.environ["REDIS_HOST"] = REDIS_HOST # os.environ["REDIS_PORT"] = REDIS_PORT # os.environ["REDIS_DB"] = REDIS_DB # os.environ["REDIS_PASSWORD"] = REDIS_PASSWORD # os.environ["REDIS_USER"] = REDIS_USER # os.environ["REDIS_URL"] = REDIS_URL # # print("redis_host", REDIS_HOST) # # print("redis_port", REDIS_PORT) # # print("redis_db", REDIS_DB) # # print("redis_password", REDIS_PASSWORD) # # print("redis_user", REDIS_USER) # # print("redis_url", REDIS_URL) # os.environ["QDRANT_HOST"] = QDRANT_HOST # os.environ["QDRANT_PORT"] = QDRANT_PORT # # os.environ["ACCESS_KEY"]=ACCESS_KEY # # os.environ["SECRET_KEY"]=SECRET_KEY # os.environ["BUCKET_NAME"]=BUCKET_NAME # os.environ["aws_access_key_id"] = aws_access_key_id # os.environ["aws_secret_access_key"] = aws_secret_access_key # print("aws_access_key_id",aws_access_key_id) # print("aws_secret_access_key",aws_secret_access_key) # print("BUCKET_NAME",BUCKET_NAME) # # os.environ["QDRANT_COLLECTION_NAME"] = QDRANT_COLLECTION_NAME # # os.environ["QDRANT_URL"] = QDRANT_URL # # os.environ["QDRANT_API_KEY"] = QDRANT_API_KEY # print("OPENAI_API_KEY successfully loaded and injected into environment.") # else: # raise Exception("Failed to retrieve or decrypt secret.") import boto3 import json import os from cryptography.fernet import Fernet from botocore.exceptions import NoCredentialsError, PartialCredentialsError, ClientError from dotenv import load_dotenv # ✅ Load environment variables for AWS credentials load_dotenv() def get_secret(secret_name: str, region_name: str = "us-east-1"): """ Retrieve and decrypt secret from AWS Secrets Manager. """ try: # ✅ First try with environment variables, then fall back to default credentials aws_access_key = os.getenv("AWS_S3_CREDS_KEY_ID") or os.getenv("AWS_ACCESS_KEY_ID") aws_secret_key = os.getenv("AWS_S3_CREDS_SECRET_KEY") or os.getenv("AWS_SECRET_ACCESS_KEY") aws_session_token = os.getenv("AWS_SESSION_TOKEN") if aws_access_key and aws_secret_key: session = boto3.session.Session( aws_access_key_id=aws_access_key, aws_secret_access_key=aws_secret_key, aws_session_token=aws_session_token ) else: # Use default credentials (from ~/.aws/credentials or IAM role) session = boto3.session.Session() client = session.client("secretsmanager", region_name=region_name) # Fetch the secret response = client.get_secret_value(SecretId=secret_name) except ClientError as e: error_code = e.response['Error']['Code'] if error_code == 'DecryptionFailureException': print("❌ Secrets Manager can't decrypt the protected secret text.") elif error_code == 'InternalServiceErrorException': print("❌ An error occurred on the server side.") elif error_code == 'InvalidParameterException': print("❌ You provided an invalid value for a parameter.") elif error_code == 'InvalidRequestException': print("❌ You provided a parameter value that is not valid for the current state of the resource.") elif error_code == 'ResourceNotFoundException': print("❌ We can't find the resource that you asked for.") elif error_code == 'AccessDeniedException': print("❌ Access denied. Check your AWS credentials and permissions.") else: print(f"❌ AWS Client Error: {e}") return None except NoCredentialsError: print("❌ AWS credentials not available.") return None except PartialCredentialsError: print("❌ Incomplete AWS credentials.") return None except Exception as e: print(f"❌ Error retrieving secret: {e}") return None # ✅ Parse JSON from AWS Secrets Manager if "SecretString" not in response: print("❌ SecretString missing in response.") return None secret_dict = json.loads(response["SecretString"]) print("✅ Secret dictionary retrieved successfully") # ✅ Retrieve Fernet key fernet_key = secret_dict.get("Fernet_Key") if not fernet_key: print("❌ Fernet_Key missing in secret.") return None try: cipher = Fernet(fernet_key) except Exception as e: print(f"❌ Error initializing Fernet cipher: {e}") return None # ✅ Decrypt all fields decrypted_data = {} decryption_errors = [] for key, value in secret_dict.items(): if key == "Fernet_Key": decrypted_data[key] = value else: try: if isinstance(value, str): decrypted_value = cipher.decrypt(value.encode()).decode() decrypted_data[key] = decrypted_value else: decrypted_value = cipher.decrypt(str(value).encode()).decode() decrypted_data[key] = decrypted_value except Exception as e: decryption_errors.append(f"{key}: {str(e)}") decrypted_data[key] = None print(f"⚠️ Failed to decrypt {key}: {e}") if decryption_errors: print(f"❌ Decryption errors for {len(decryption_errors)} keys") # ✅ Extract ALL variables CONNECTIONS_HOST = decrypted_data.get("CONNECTIONS_HOST", "") CONNECTIONS_DB = decrypted_data.get("CONNECTIONS_DB", "") CONNECTIONS_USER = decrypted_data.get("CONNECTIONS_USER", "") CONNECTIONS_PASS = decrypted_data.get("CONNECTIONS_PASS", "") OPENAI_API_KEY = decrypted_data.get("OPENAI_API_KEY", "") OPENAI_MODEL_NAME = decrypted_data.get("OPENAI_MODEL_NAME", "") OPENAI_MODEL_TURBO = decrypted_data.get("OPENAI_MODEL_TURBO", "") REDIS_HOST = decrypted_data.get("REDIS_HOST", "") REDIS_PORT = decrypted_data.get("REDIS_PORT", "") REDIS_DB = decrypted_data.get("REDIS_DB", "") REDIS_PASSWORD = decrypted_data.get("REDIS_PASSWORD", "") REDIS_USER = decrypted_data.get("REDIS_USER", "") REDIS_URL = decrypted_data.get("REDIS_URL", "") QDRANT_HOST = decrypted_data.get("QDRANT_HOST", "") QDRANT_PORT = decrypted_data.get("QDRANT_PORT", "") QDRANT_COLLECTION_NAME = decrypted_data.get("QDRANT_COLLECTION_NAME", "") QDRANT_URL = decrypted_data.get("QDRANT_URL", "") QDRANT_API_KEY = decrypted_data.get("QDRANT_API_KEY", "") # ✅ HARDCODED AWS S3 CREDENTIALS (TEMPORARY FIX) # 🔥 REPLACE THESE WITH YOUR ACTUAL AWS CREDENTIALS 🔥 AWS_S3_CREDS_KEY_ID = "AKIAT4R3SOPSJGE7NEV6" # Replace with your actual AWS Access Key ID AWS_S3_CREDS_SECRET_KEY = "x610UKmIeda8FVVe2j3R6Kij+2tX0LqLqPWOUTJz" # Replace with your actual AWS Secret Access Key print("🔑 USING HARDCODED AWS CREDENTIALS (TEMPORARY SOLUTION)") # ✅ Bucket names BUCKET_NAME = "" for key in decrypted_data: if key.strip() == "BUCKET_NAME": BUCKET_NAME = decrypted_data.get(key, "").strip() break if not BUCKET_NAME: BUCKET_NAME = decrypted_data.get("BUCKET_NAME", "").strip() BUCKET_UNSTRUCTURED_STORAGE = decrypted_data.get("BUCKET_UNSTRUCTURED_STORAGE", "") ALLOWED_HOSTS = decrypted_data.get("ALLOWED_HOSTS", "") return ( decrypted_data, CONNECTIONS_HOST, CONNECTIONS_DB, CONNECTIONS_USER, CONNECTIONS_PASS, OPENAI_API_KEY, OPENAI_MODEL_NAME, OPENAI_MODEL_TURBO, REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD, REDIS_USER, QDRANT_HOST, QDRANT_PORT, QDRANT_COLLECTION_NAME, QDRANT_URL, QDRANT_API_KEY, REDIS_URL, BUCKET_NAME, BUCKET_UNSTRUCTURED_STORAGE, AWS_S3_CREDS_KEY_ID, AWS_S3_CREDS_SECRET_KEY, ALLOWED_HOSTS ) # 👇 Main execution if __name__ == "__main__": print("🚀 Starting secret retrieval...") secret_name = "Demo/MR/skeys" region_name = "us-east-1" secrets = get_secret(secret_name, region_name) if secrets: ( decrypted_data, CONNECTIONS_HOST, CONNECTIONS_DB, CONNECTIONS_USER, CONNECTIONS_PASS, OPENAI_API_KEY, OPENAI_MODEL_NAME, OPENAI_MODEL_TURBO, REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD, REDIS_USER, QDRANT_HOST, QDRANT_PORT, QDRANT_COLLECTION_NAME, QDRANT_URL, QDRANT_API_KEY, REDIS_URL, BUCKET_NAME, BUCKET_UNSTRUCTURED_STORAGE, AWS_S3_CREDS_KEY_ID, AWS_S3_CREDS_SECRET_KEY, ALLOWED_HOSTS ) = secrets # ✅ Set all environment variables env_vars = { # Database "CONNECTIONS_HOST": CONNECTIONS_HOST, "CONNECTIONS_DB": CONNECTIONS_DB, "CONNECTIONS_USER": CONNECTIONS_USER, "CONNECTIONS_PASS": CONNECTIONS_PASS, # OpenAI "OPENAI_API_KEY": OPENAI_API_KEY, "OPENAI_MODEL_NAME": OPENAI_MODEL_NAME, "OPENAI_MODEL_TURBO": OPENAI_MODEL_TURBO, # Redis "REDIS_HOST": REDIS_HOST, "REDIS_PORT": REDIS_PORT, "REDIS_DB": REDIS_DB, "REDIS_PASSWORD": REDIS_PASSWORD, "REDIS_USER": REDIS_USER, "REDIS_URL": REDIS_URL, # Qdrant "QDRANT_HOST": QDRANT_HOST, "QDRANT_PORT": QDRANT_PORT, "QDRANT_COLLECTION_NAME": QDRANT_COLLECTION_NAME, "QDRANT_URL": QDRANT_URL, "QDRANT_API_KEY": QDRANT_API_KEY, # AWS S3 "AWS_S3_CREDS_KEY_ID": AWS_S3_CREDS_KEY_ID, "AWS_S3_CREDS_SECRET_KEY": AWS_S3_CREDS_SECRET_KEY, "BUCKET_NAME": BUCKET_NAME, "BUCKET_UNSTRUCTURED_STORAGE": BUCKET_UNSTRUCTURED_STORAGE, # Other "ALLOWED_HOSTS": ALLOWED_HOSTS } # Set environment variables for key, value in env_vars.items(): if value is not None: os.environ[key] = str(value) print("\n" + "="*50) print("✅ SECRET RETRIEVAL COMPLETE") print("="*50) print(f"🔑 AWS Key: {AWS_S3_CREDS_KEY_ID}") print(f"🔑 AWS Secret: {AWS_S3_CREDS_SECRET_KEY[:15]}...") print(f"📦 Bucket: {BUCKET_NAME}") print(f"🤖 OpenAI API Key: Loaded successfully") print(f"🗄️ Database: {CONNECTIONS_DB}") print(f"🔍 Qdrant: {QDRANT_COLLECTION_NAME}") print("\n✅ All environment variables set successfully!") else: print("❌ Failed to retrieve or decrypt secret.") # Set up environment variables when imported as module try: secrets = get_secret("Demo/MR/skeys", "us-east-1") if secrets: ( decrypted_data, CONNECTIONS_HOST, CONNECTIONS_DB, CONNECTIONS_USER, CONNECTIONS_PASS, OPENAI_API_KEY, OPENAI_MODEL_NAME, OPENAI_MODEL_TURBO, REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD, REDIS_USER, QDRANT_HOST, QDRANT_PORT, QDRANT_COLLECTION_NAME, QDRANT_URL, QDRANT_API_KEY, REDIS_URL, BUCKET_NAME, BUCKET_UNSTRUCTURED_STORAGE, AWS_S3_CREDS_KEY_ID, AWS_S3_CREDS_SECRET_KEY, ALLOWED_HOSTS ) = secrets os.environ.update({ "CONNECTIONS_HOST": CONNECTIONS_HOST, "CONNECTIONS_DB": CONNECTIONS_DB, "CONNECTIONS_USER": CONNECTIONS_USER, "CONNECTIONS_PASS": CONNECTIONS_PASS, "OPENAI_API_KEY": OPENAI_API_KEY, "OPENAI_MODEL_NAME": OPENAI_MODEL_NAME, "OPENAI_MODEL_TURBO": OPENAI_MODEL_TURBO, "REDIS_HOST": REDIS_HOST, "REDIS_PORT": REDIS_PORT, "REDIS_DB": REDIS_DB, "REDIS_PASSWORD": REDIS_PASSWORD, "REDIS_USER": REDIS_USER, "REDIS_URL": REDIS_URL, "QDRANT_HOST": QDRANT_HOST, "QDRANT_PORT": QDRANT_PORT, "QDRANT_COLLECTION_NAME": QDRANT_COLLECTION_NAME, "QDRANT_URL": QDRANT_URL, "QDRANT_API_KEY": QDRANT_API_KEY, "AWS_S3_CREDS_KEY_ID": AWS_S3_CREDS_KEY_ID, "AWS_S3_CREDS_SECRET_KEY": AWS_S3_CREDS_SECRET_KEY, "BUCKET_NAME": BUCKET_NAME, "BUCKET_UNSTRUCTURED_STORAGE": BUCKET_UNSTRUCTURED_STORAGE, "ALLOWED_HOSTS": ALLOWED_HOSTS }) print("✅ Secrets loaded into environment") except Exception as e: print(f"⚠️ Note: {e}")