def get_secret()

in Templates/kerberosSideCar/krb_side_car.py [0:0]

• 39 lines of code
• 4 McCabe index (conditional complexity)

def get_secret(region_name_arg, secret_arn_arg):
    """
    Get secret from AWS Secrets Manager using IAM role,
    see  https://boto3.amazonaws.com/v1/documentation/api/latest/guide/secrets
    -manager.html
    :param region_name_arg: Region name of current container
    :type region_name_arg: basestring such as "us-west-1"
    :param secret_arn_arg: Secret ARN for AWS Secrets Manager secret
    :type secret_arn_arg: basestring such as "arn:aws:secretsmanager:us-west-1...
    :return: Secrets in string or None if there is an error
    :rtype: basestring or none
    """

    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name_arg,
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=secret_arn_arg
        )
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceNotFoundException':
            print("The requested secret " + secret_arn_arg + " was not found",
                  flush=True)
        elif e.response['Error']['Code'] == 'InvalidRequestException':
            print("The request was invalid due to:", e, flush=True)
        elif e.response['Error']['Code'] == 'InvalidParameterException':
            print("The request had invalid params:", e, flush=True)
        elif e.response['Error']['Code'] == 'DecryptionFailure':
            print(
                "The requested secret can't be decrypted using the provided KMS "
                "key:",
                e, flush=True)
        elif e.response['Error']['Code'] == 'InternalServiceError':
            print("An error occurred on service side:", e, flush=True)
    else:
        # Secrets Manager decrypts the secret value using the associated KMS CMK
        # Depending on whether the secret was a string or binary, only one of
        # these fields will be populated
        if 'SecretString' in get_secret_value_response:
            text_secret_data = get_secret_value_response['SecretString']
            secret = text_secret_data
        else:
            binary_secret_data = get_secret_value_response['SecretBinary']
            secret = binary_secret_data

    secret_string = json.loads(secret)
    username = secret_string[USERNAME_KEY]
    password = secret_string[PASSWORD_KEY]

    if username is None or password is None:
        """
        If Secrets Manager is not properly configured, the program will exit
        """
        print("ERROR* Secret not available from Secrets Manager", flush=True)
        sys.exit(1)

    return username, password