# # 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.") 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_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) # 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"]) # ✅ Retrieve Fernet key fernet_key = secret_dict.get("Fernet_Key") if not fernet_key: # print("❌ Fernet_Key missing in secret.") return None cipher = Fernet(fernet_key) # ✅ 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 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, ) # 👇 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.")