"""Config file for lambda function.""" import logging import os from abacus_common_logic.utils.ssh import get_private_key from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization from dotenv import load_dotenv from secrets_manager.lambda_ext import LambdaSecretsManager # Load environment variables from a .env file if present load_dotenv() APPLICATION_NAME = 'lambda-abacus-commit-mechanicals' # AWS region configuration DEFAULT_AWS_REGION = 'us-east-1' AWS_REGION = os.environ.get('AWS_REGION', DEFAULT_AWS_REGION) # Environment setup DEV_ENVIRONMENT = 'dev' QA_ENVIRONMENT = 'qa' UAT_ENVIRONMENT = 'uat' PROD_ENVIRONMENT = 'prod' ENVIRONMENT = os.environ.get('ENVIRONMENT', DEV_ENVIRONMENT) LOGGING_LEVEL = os.environ.get('LOGGING_LEVEL', logging.INFO) LOGGER_DSN = os.environ.get('LOGGER_DSN') SENTRY_DSN = None CONTRACT_BATCH_SIZE = os.environ.get('CONTRACT_BATCH_SIZE', 10000) LEDGER_BATCH_SIZE = os.environ.get('LEDGER_BATCH_SIZE', 100) # Snowflake configuration SNOWFLAKE_CONFIG = { 'role': os.environ.get('SNOWFLAKE_ROLE'), 'warehouse': os.environ.get('SNOWFLAKE_WAREHOUSE'), 'db': os.environ.get('SNOWFLAKE_DATABASE'), 'schema': os.environ.get('SNOWFLAKE_SCHEMA', ENVIRONMENT), 'user': os.environ.get('SNOWFLAKE_USER'), 'account': os.environ.get('SNOWFLAKE_ACCOUNT'), 'private_key': None } # Snowflake SSH-Keypair access SNOWFLAKE_KEY_PASSPHRASE = os.environ.get('SNOWFLAKE_KEY_PASSPHRASE') SNOWFLAKE_PRIVATE_KEY_PATH = os.environ.get('SNOWFLAKE_PRIVATE_KEY_PATH') SNOWFLAKE_PRIVATE_KEY = os.environ.get('SNOWFLAKE_PRIVATE_KEY', None) # Pull QA, UAT & Prod credentials from secrets manager if ENVIRONMENT in [QA_ENVIRONMENT, UAT_ENVIRONMENT, PROD_ENVIRONMENT]: client = LambdaSecretsManager( environment=ENVIRONMENT, service_name=APPLICATION_NAME ) SENTRY_DSN = client.get_cred('SENTRY_DSN') SNOWFLAKE_KEY_PASSPHRASE = client.get_cred('SNOWFLAKE_KEY_PASSPHRASE') SNOWFLAKE_PRIVATE_KEY = client.get_cred('SNOWFLAKE_PRIVATE_KEY') # For QA, UAT and Prod if SNOWFLAKE_PRIVATE_KEY and SNOWFLAKE_KEY_PASSPHRASE: private_key = serialization.load_pem_private_key( bytes(SNOWFLAKE_PRIVATE_KEY, 'utf8'), password=bytes(SNOWFLAKE_KEY_PASSPHRASE, 'utf8'), backend=default_backend() ) SNOWFLAKE_CONFIG['private_key'] = private_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption() ) # For local/dev env, get private key from local machine elif SNOWFLAKE_PRIVATE_KEY_PATH and SNOWFLAKE_KEY_PASSPHRASE: SNOWFLAKE_CONFIG['private_key'] = get_private_key( SNOWFLAKE_PRIVATE_KEY_PATH.replace('~', os.environ['HOME']), SNOWFLAKE_KEY_PASSPHRASE )