# import boto3 # import json # from cryptography.fernet import Fernet # from botocore.exceptions import NoCredentialsError, PartialCredentialsError # def get_secret(secret_name, region_name): # """ # Retrieve secret value from AWS Secrets Manager. # Args: # secret_name (str): The name of the secret to retrieve. # region_name (str): AWS region where the secret is stored. # Returns: # dict: Decrypted secret values. # """ # # Create a 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("Credentials not available") # return None # except PartialCredentialsError: # print("Incomplete credentials provided") # return None # except Exception as e: # print(f"Error retrieving secret: {e}") # return None # # Depending on whether the secret is a string or binary, one of these fields will be populated # if 'SecretString' in response: # secret_dict = json.loads(response['SecretString']) # # print("secret_dict", secret_dict) # fernet_key = secret_dict.get('Fernet_Key', None) # # print("fernet_key", fernet_key) # else: # print("Failed to retrieve secret.") # return None # if not fernet_key: # print("Fernet key not found in secret.") # return None # # Create a Fernet cipher object # cipher_suite = Fernet(fernet_key) # CONNECTIONS_HOST = cipher_suite.decrypt(secret_dict["CONNECTIONS_HOST"].encode()).decode() # CONNECTIONS_DB = cipher_suite.decrypt(secret_dict["CONNECTIONS_DB"].encode()).decode() # CONNECTIONS_USER = cipher_suite.decrypt(secret_dict["CONNECTIONS_USER"].encode()).decode() # CONNECTIONS_PASS = cipher_suite.decrypt(secret_dict["CONNECTIONS_PASS"].encode()).decode() # ARANGO_URL = cipher_suite.decrypt(secret_dict["ARANGO_URL"].encode()).decode() # ARANGO_USERNAME = cipher_suite.decrypt(secret_dict["ARANGO_USERNAME"].encode()).decode() # ARANGO_PASSWORD = cipher_suite.decrypt(secret_dict["ARANGO_PASSWORD"].encode()).decode() # ARANGO_DB = cipher_suite.decrypt(secret_dict["ARANGO_DB"].encode()).decode() # JAVA_URL = cipher_suite.decrypt(secret_dict["JAVA_URL"].encode()).decode() # SERVER_PORT = cipher_suite.decrypt(secret_dict["SERVER_PORT"].encode()).decode() # PYTHON_URL = cipher_suite.decrypt(secret_dict["PYTHON_URL"].encode()).decode() # AWS_S3_CREDS_KEY_ID = cipher_suite.decrypt(secret_dict["AWS_S3_CREDS_KEY_ID"].encode()).decode() # AWS_S3_CREDS_SECRET_KEY = cipher_suite.decrypt(secret_dict["AWS_S3_CREDS_SECRET_KEY"].encode()).decode() # NOTIFICATION_ENDPOINT = cipher_suite.decrypt(secret_dict["NOTIFICATION_ENDPOINT"].encode()).decode() # SUPPORT_EMAIL = cipher_suite.decrypt(secret_dict["SUPPORT_EMAIL"].encode()).decode() # SUPPORT_EMAIL_PASS = cipher_suite.decrypt(secret_dict["SUPPORT_EMAIL_PASS"].encode()).decode() # MAIL_SERVER_SMTP = cipher_suite.decrypt(secret_dict["MAIL_SERVER_SMTP"].encode()).decode() # MAIL_SERVER_PORT = cipher_suite.decrypt(secret_dict["MAIL_SERVER_PORT"].encode()).decode() # FOLDER_ETL_S3_BUCKET_NAME = "" # FOLDER_UNSTRUCTURED_STORAGE = "" # BUCKET_ETL_S3_BUCKET_NAME = cipher_suite.decrypt(secret_dict["BUCKET_ETL_S3_BUCKET_NAME"].encode()).decode() # BUCKET_UNSTRUCTURED_STORAGE = cipher_suite.decrypt(secret_dict["BUCKET_UNSTRUCTURED_STORAGE"].encode()).decode() # BUCKET_PERFORMANCE_DRIVERS = "" # APP_URL = cipher_suite.decrypt(secret_dict["APP_URL"].encode()).decode() # ALLOWED_HOSTS = cipher_suite.decrypt(secret_dict["ALLOWED_HOSTS"].encode()).decode() + ',http://192.168.0.110:4521/' # return CONNECTIONS_HOST, \ # CONNECTIONS_DB, CONNECTIONS_USER, CONNECTIONS_PASS, ARANGO_URL, ARANGO_USERNAME, ARANGO_PASSWORD, \ # ARANGO_DB, JAVA_URL, SERVER_PORT, PYTHON_URL, AWS_S3_CREDS_KEY_ID, AWS_S3_CREDS_SECRET_KEY, NOTIFICATION_ENDPOINT, SUPPORT_EMAIL, \ # SUPPORT_EMAIL_PASS, MAIL_SERVER_SMTP, MAIL_SERVER_PORT, BUCKET_ETL_S3_BUCKET_NAME, \ # BUCKET_UNSTRUCTURED_STORAGE, APP_URL, ALLOWED_HOSTS, FOLDER_ETL_S3_BUCKET_NAME, FOLDER_UNSTRUCTURED_STORAGE, BUCKET_PERFORMANCE_DRIVERS # secret_name = "Demo/Ingen/skeys" # region_name = "us-east-1" # CONNECTIONS_HOST, CONNECTIONS_DB, CONNECTIONS_USER, CONNECTIONS_PASS, ARANGO_URL, ARANGO_USERNAME, ARANGO_PASSWORD, \ # ARANGO_DB, JAVA_URL, SERVER_PORT, PYTHON_URL, AWS_S3_CREDS_KEY_ID, AWS_S3_CREDS_SECRET_KEY, NOTIFICATION_ENDPOINT, SUPPORT_EMAIL, \ # SUPPORT_EMAIL_PASS, MAIL_SERVER_SMTP, MAIL_SERVER_PORT, BUCKET_ETL_S3_BUCKET_NAME, \ # BUCKET_UNSTRUCTURED_STORAGE, APP_URL, ALLOWED_HOSTS, FOLDER_ETL_S3_BUCKET_NAME, FOLDER_UNSTRUCTURED_STORAGE, \ # BUCKET_PERFORMANCE_DRIVERS = get_secret(secret_name, region_name) # # Got it! You have AWS credentials configured locally (via `aws configure`), and you want to make sure your `get_secret` function uses those credentials to fetch the secret from AWS Secrets Manager. # # The great news is: **boto3 automatically picks up credentials from your AWS CLI configuration (`~/.aws/credentials`)** by default. So if you’ve already run `aws configure` and set your credentials, your existing code will use those credentials without any extra change. # # ### What you *don’t* need to do: # # * You don’t need to manually pass credentials in the function unless you want to override. # # * You don’t need to add `aws_access_key_id` or `aws_secret_access_key` manually unless you want to hardcode or dynamically pass them. # # ### How it works currently: # # ```python # # session = boto3.session.Session() # # client = session.client('secretsmanager', region_name=region_name) # # ``` # # * This uses the default credential provider chain, which looks for credentials in environment variables, AWS CLI config files, EC2 instance roles, etc. # # * Since you have credentials configured via `aws configure`, boto3 will find and use those credentials automatically. # # --- # # ### Optional: If you want to explicitly use the credentials from your `.env` or environment variables # # If you want to explicitly use credentials stored in environment variables (e.g., loaded from `.env`), you can modify your function like this: # # ```python # # import os # # def get_secret(secret_name, region_name): # # 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") # optional, if using temporary credentials # # ) # # client = session.client('secretsmanager', region_name=region_name) # # # rest of your function # # ``` # # This assumes you have loaded your `.env` variables (using `python-dotenv` or similar) or have them set in your environment. # # --- # # ### Summary: # # * If you have run `aws configure` and have credentials locally, **no change is needed**. Your current code uses those credentials. # # * If you want to use environment variables or `.env` values for AWS keys explicitly, pass them to `Session()` as shown above. # # --- # # If you want, I can help you with a quick snippet to load `.env` variables too! Just ask. import boto3 import json import os from cryptography.fernet import Fernet from botocore.exceptions import NoCredentialsError, PartialCredentialsError from dotenv import load_dotenv load_dotenv() def get_secret(secret_name, region_name): try: 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) response = client.get_secret_value(SecretId=secret_name) if 'SecretString' not in response: print("SecretString not found.") return None secret_dict = json.loads(response['SecretString']) fernet_key = secret_dict.get('Fernet_Key') if not fernet_key: print("Fernet key missing.") return None cipher_suite = Fernet(fernet_key.encode()) def decrypt(key): return cipher_suite.decrypt(secret_dict[key].encode()).decode() CONNECTIONS_HOST = decrypt("CONNECTIONS_HOST") CONNECTIONS_DB = decrypt("CONNECTIONS_DB") CONNECTIONS_USER = decrypt("CONNECTIONS_USER") CONNECTIONS_PASS = decrypt("CONNECTIONS_PASS") ARANGO_URL = decrypt("ARANGO_URL") ARANGO_USERNAME = decrypt("ARANGO_USERNAME") ARANGO_PASSWORD = decrypt("ARANGO_PASSWORD") ARANGO_DB = decrypt("ARANGO_DB") JAVA_URL = decrypt("JAVA_URL") SERVER_PORT = decrypt("SERVER_PORT") PYTHON_URL = decrypt("PYTHON_URL") AWS_S3_CREDS_KEY_ID = decrypt("AWS_S3_CREDS_KEY_ID") AWS_S3_CREDS_SECRET_KEY = decrypt("AWS_S3_CREDS_SECRET_KEY") NOTIFICATION_ENDPOINT = decrypt("NOTIFICATION_ENDPOINT") SUPPORT_EMAIL = decrypt("SUPPORT_EMAIL") SUPPORT_EMAIL_PASS = decrypt("SUPPORT_EMAIL_PASS") MAIL_SERVER_SMTP = decrypt("MAIL_SERVER_SMTP") MAIL_SERVER_PORT = decrypt("MAIL_SERVER_PORT") BUCKET_ETL_S3_BUCKET_NAME = decrypt("BUCKET_ETL_S3_BUCKET_NAME") BUCKET_UNSTRUCTURED_STORAGE = decrypt("BUCKET_UNSTRUCTURED_STORAGE") APP_URL = decrypt("APP_URL") ALLOWED_HOSTS = decrypt("ALLOWED_HOSTS") + ',http://192.168.0.110:4521/' # Static/empty strings FOLDER_ETL_S3_BUCKET_NAME = "" FOLDER_UNSTRUCTURED_STORAGE = "" BUCKET_PERFORMANCE_DRIVERS = "" return CONNECTIONS_HOST, CONNECTIONS_DB, CONNECTIONS_USER, CONNECTIONS_PASS, ARANGO_URL, ARANGO_USERNAME, \ ARANGO_PASSWORD, ARANGO_DB, JAVA_URL, SERVER_PORT, PYTHON_URL, AWS_S3_CREDS_KEY_ID, AWS_S3_CREDS_SECRET_KEY, \ NOTIFICATION_ENDPOINT, SUPPORT_EMAIL, SUPPORT_EMAIL_PASS, MAIL_SERVER_SMTP, MAIL_SERVER_PORT, \ BUCKET_ETL_S3_BUCKET_NAME, BUCKET_UNSTRUCTURED_STORAGE, APP_URL, ALLOWED_HOSTS, \ FOLDER_ETL_S3_BUCKET_NAME, FOLDER_UNSTRUCTURED_STORAGE, BUCKET_PERFORMANCE_DRIVERS except Exception as e: print(f"Error retrieving secrets: {e}") return None # 👇 Move this outside __main__ so it runs on import secret_name = "Demo/Ingen/skeys" region_name = "us-east-1" secrets = get_secret(secret_name, region_name) if secrets: ( CONNECTIONS_HOST, CONNECTIONS_DB, CONNECTIONS_USER, CONNECTIONS_PASS, ARANGO_URL, ARANGO_USERNAME, ARANGO_PASSWORD, ARANGO_DB, JAVA_URL, SERVER_PORT, PYTHON_URL, AWS_S3_CREDS_KEY_ID, AWS_S3_CREDS_SECRET_KEY, NOTIFICATION_ENDPOINT, SUPPORT_EMAIL, SUPPORT_EMAIL_PASS, MAIL_SERVER_SMTP, MAIL_SERVER_PORT, BUCKET_ETL_S3_BUCKET_NAME, BUCKET_UNSTRUCTURED_STORAGE, APP_URL, ALLOWED_HOSTS, FOLDER_ETL_S3_BUCKET_NAME, FOLDER_UNSTRUCTURED_STORAGE, BUCKET_PERFORMANCE_DRIVERS ) = secrets print("Secrets successfully loaded and decrypted.") else: raise Exception("❌ Failed to load secrets. Check AWS credentials or secret structure.")