sendSESEmail.py

#!/usr/bin/env python

"""Send emails through AWS SES.

Uses the AWS SES call send-mail.
Has the following parameters:

Body contents: (at least one of the following has to be specified.)
- text -- the plain text version of the email.
- html -- the HTML-formatted version of the email.

- subject -- the subject of the email.  This is required.
- efrom -- the from email.  This is required.
The efrom requires verification by AWS SES (either by email address or by domain.)
- replyto -- the reply to email.  (submitform.pl & submitsecure.pl)

Email destinations: (at least one of the following has to be specified.)
Destinations are space-separated like a@comp.dom b@comp.dom ...
- to -- a list of to recipients.  (Who the email is actually addressed to.)
- bcc -- a list of bcc recipients. (List isn't visible in email.)
- cc -- the cc for the email.  (Visible list in email.)

Amazon requires the from email to be verified or the domain verified.
"""

import sys
import boto3
import os
import json
import argparse
from botocore.exceptions import ClientError
import textwrap
import logging

format=("[%(levelname)-5s] %(asctime)-15s ({}|%(funcName)s|%(lineno)d) "
        ":: %(message)s").format(__file__)
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
logging_handler = logging.StreamHandler()
logging_handler.setFormatter(logging.Formatter(format))
LOGGER.addHandler(logging_handler)

# Global settings
USAGE_HELP = ("Usage: sendSESEmail.py [-h] [--to TO [TO ...]]"
              "[--bcc BCC [BCC ...]] [--cc CC [CC ...]] --efrom EFROM"
              "[--replyto REPLYTO [REPLYTO ...]] --subject SUBJECT"
              "[--plaintext PLAINTEXT] [--htmltext HTMLTEXT]"
              "[--quiet] [--context CONTEXT]")

# Use us-west-2 as a default region
# This should work for local development, alpha stack(us-east-2) and prod
AWS_REGION = os.getenv('AWS_REGION', 'us-west-2')


class CustomArgumentParser(argparse.ArgumentParser):
    """Custom ArgumentParser class

    Extends:
        argparse.ArgumentParser
    """

    def error(self, message):
        """Suppressing default error method

        Return custom error message on ArgumentParser.error.

        Arguments:
            message -- error message

        Raises:
            BaseException
        """
        raise BaseException(message)


def getParser():
    """Prepare argument parser.

    Returns:
        parser -- ArgumentParser object
    """
    parser = CustomArgumentParser (
        description = "Send emails through AWS SES.",
        formatter_class = argparse.RawDescriptionHelpFormatter,
        epilog = textwrap.dedent('''\
            additional information:
                One of --to, --bcc, or --cc has to be defined.
                One of --plaintext or --htmltext has to be defined.
                --efrom and --subject are required.
            '''))

    parser.add_argument ("--to", nargs = "+",
                        help = "Space-separated list of to email addresses",
                        required = False)

    parser.add_argument ("--bcc", nargs = "+",
                        help = "Space-separated list of bcc email addresses",
                        required = False)

    parser.add_argument ("--cc", nargs = "+",
                         help = "Space-separated list of cc email addresses",
                         required = False)

    parser.add_argument ("--efrom",
                        help = "From email address",
                        required = True)

    parser.add_argument ("--replyto", nargs = "+",
                        help = "Reply to email address",
                        required = False)

    parser.add_argument ("--subject",
                         help = "Subject of the email",
                         required = True)

    parser.add_argument ("--plaintext",
                        help = "Email body in plaintext",
                        required = False)

    parser.add_argument ("--htmltext",
                        help = "Html body in HTML",
                        required = False)

    parser.add_argument("--quiet", action = "store_true",
                        help = "Suppress success message",
                        required = False)

    parser.add_argument("--context",
                        help = ("Json string for logging application"
                                "context for the email messages being sent."),
                        required = False)


    return parser


def main(argv):
    """Main method to execute operations on email verification.

    Arguments:
        argv {list} -- list of script arguments

    Raises:
        SystemExit
        BaseException
    """
    return_code = 0
    response = {}
    mainReturn = {
        'status': '000',
        'error': [],
        'response': response,
        'context': {}
    }

    # Get argument parser
    parser = getParser()

    try:
        args = parser.parse_args(argv)
        mainReturn["context"] = args.context

        # Validation
        if (args.to == None and args.bcc == None and args.cc == None):
            raise BaseException ("Destination has to be specified. One of"
                                 " the --to, --bcc or --cc options need"
                                 " to be provided.")

        if (args.plaintext == None and args.htmltext == None):
            raise BaseException ("Email body has to be specified")

    except (BaseException) as e:
        return_code = 102
        evalue = str(e)
        if (evalue != "0"): # If it is empty, it is a help message
            mainReturn["status"] = return_code
            mainReturn["error"].append(USAGE_HELP)
            mainReturn["error"].append(evalue)
            LOGGER.error(json.dumps(mainReturn))

    except (Exception) as e:
        return_code = 103
        mainReturn["status"] = return_code
        mainReturn["error"] = e
        LOGGER.error(json.dumps(mainReturn))

    # exit if validation error
    if return_code > 0:
        return return_code

    try:
        # Create boto3 client object
        client = boto3.client('ses', region_name=AWS_REGION)

        # Construct destination
        destination = {}
        if (args.to != None):
            destination["ToAddresses"] = args.to
        if (args.bcc != None):
            destination["BccAddresses"] = args.bcc
        if (args.cc != None):
            destination["CcAddresses"] = args.cc

        # Construct message
        charset = "UTF-8"
        message = {}
        message["Subject"] = {}
        message["Subject"]["Data"] = args.subject
        message["Subject"]["Charset"] = charset
        if (args.plaintext != None):
            message["Body"] = {}
            message["Body"]["Text"] = {}
            message["Body"]["Text"]["Data"] = args.plaintext
            message["Body"]["Text"]["Charset"] = charset

        if (args.htmltext != None):
            if "Body" not in message:
                message["Body"] = {}

            message["Body"]["Html"] = {}
            message["Body"]["Html"]["Data"] = args.htmltext
            message["Body"]["Html"]["Charset"] = charset

        if (args.replyto != None):
            # Call AWS SES
            response = client.send_email (
                Source = args.efrom,
                Destination = destination,
                Message = message,
                ReplyToAddresses = args.replyto,
                ConfigurationSetName='ses-detail-logging',
            )
        else:
            # Call AWS SES
            response = client.send_email (
                Source = args.efrom,
                Destination = destination,
                Message = message,
                ConfigurationSetName='ses-detail-logging',
            )

    except ClientError as e:
        return_code = 902
        mainReturn["status"] = return_code
        mainReturn["error"] = str(e)

    except Exception as e:
        return_code = 903
        mainReturn["status"] = return_code
        mainReturn['error'] = str(e)

    mainReturn['response'] = response

    if return_code > 0:
        LOGGER.error(json.dumps(mainReturn))
    else:
        LOGGER.info(json.dumps(mainReturn))

    return return_code


def run():
    """Script entrypoint"""
    sys.exit(main(sys.argv[1:]))


if __name__ == "__main__":
    run()