"""Lambda get-db-info module.""" import boto3 import sentry_sdk from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration import config from common.logic import database # noqa from common.utils.logger import log as logger # noqa if config.SENTRY_DSN: sentry_sdk.init( dsn=config.SENTRY_DSN, environment=config.ENVIRONMENT, integrations=[AwsLambdaIntegration(timeout_warning=True)] ) def assume_source_account_role(account_id, role): """ Assume IAM role in source account. Args: account_id (str): AWS account ID Returns: dict: dict of AWS credentials """ client = boto3.client('sts') assume_role_response = client.assume_role( RoleArn=f'arn:aws:iam::{account_id}:role/{role}', RoleSessionName=config.SERVICE_NAME, ExternalId=config.EXTERNAL_ID, DurationSeconds=3600, ) credentials = assume_role_response['Credentials'] return credentials def handler(event, _context): """Lambda entry point.""" try: credentials = assume_source_account_role( event['account_id'], event['role'] ) rds_client = boto3.client( 'rds', region_name=config.AWS_DEFAULT_REGION, aws_access_key_id=credentials['AccessKeyId'], aws_secret_access_key=credentials['SecretAccessKey'], aws_session_token=credentials['SessionToken'] ) kms_client = boto3.client( 'kms', region_name=config.AWS_DEFAULT_REGION, aws_access_key_id=credentials['AccessKeyId'], aws_secret_access_key=credentials['SecretAccessKey'], aws_session_token=credentials['SessionToken'] ) db_id, db_type = database.find_database_identifier( event['db_name'], rds_client) db_config = database.get_config(db_id, db_type, rds_client, kms_client) return { 'db_identifier': db_id, 'db_type': db_type, 'config': db_config } except Exception as error: logger.exception(str(error)) raise error