タグ: aws

こんにちは。
気づくと既に 2023 年も 1/4 が経過しており、時間の進む速度感覚がおかしくなってきた今日この頃でございます。

さて、とある事情から SES receipt rule で受信したメールを加工して Slack に通知させる必要がでてきました。
具体的には最終的に下記構成のようなパイプラインを構築しようと考えております。

・メール受信時にメール加工用 Lambda を起動し、データを S3 に格納
・特定の From のメールが S3 に格納されたことをトリガーに通知用 Lambda を起動
・別システムのデータベースから情報を引っ張ってきて Slack へ通知

つきましては、この構成の一部を Terraform で管理してみようと思います。

構成図

今回は SES receipt rule → SNS → Lambda → S3 → EventBridge → Lambda までを一括でデプロイできる HCL を作成します。

HCL

作成したコード一式は こちら になります。

.
├── README.md
├── example-slack-lambda.tf.example
├── modules
│   ├── notifier
│   │   ├── eventbridge.tf
│   │   ├── example
│   │   │   ├── basic
│   │   │   │   ├── lambda
│   │   │   │   │   ├── bin
│   │   │   │   │   └── source
│   │   │   │   │       └── basic.py
│   │   │   │   ├── lambda.tf
│   │   │   │   └── variables.tf
│   │   │   └── slack
│   │   │       ├── lambda
│   │   │       │   ├── bin
│   │   │       │   └── source
│   │   │       │       └── slack.py
│   │   │       ├── lambda.tf
│   │   │       └── variables.tf
│   │   └── variables.tf
│   └── receiver
│       ├── lambda
│       │   ├── bin
│       │   └── source
│       │       └── receiver.py
│       ├── lambda.tf
│       ├── s3-processed.tf
│       ├── s3-receiver.tf
│       ├── ses-receipt-rule.tf
│       ├── sns.tf
│       └── variables.tf
├── notifier.tf.example
└── receiver.tf.example

modules/ 配下に各種リソース定義を配置し、上位からモジュールを読み込む想定です。
※ .example はモジュール読み込み用のテンプレートです。

以降、要所をツラツラと記載させて頂きます。

Lambda ( メール加工 )

SES で受信したメールを加工する Lambda を用意します。
処理の詳細はメールの From、受信日、件名、本文、表示名を抜き出し、S3 に格納する、とします。
S3 に格納する情報は、後に繋ぎやすいように、本文のみのオブジェクトとし、その他の情報はオブジェクトを格納するディレクトリ名として保持させます。

・modules/receiver/lambda/source/receiver.py

import boto3
import json
import logging
import email
import random
import string
import os
import re

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):

    logger.info("Event: " + str(event))
    
    message = json.loads(event['Records'][0]['Sns']['Message'])
    m_from = message['mail']['commonHeaders']['from'][0]
    date = message['mail']['commonHeaders']['date']
    subject = message['mail']['commonHeaders']['subject']
    content = message['content']
    email_obj = email.message_from_string(content)

    body = perth_mail_body(email_obj)
    logger.info("Body: " + str(body))

    fname = extract_mail_address(m_from).replace("/","[slash]") + "/" \
        + subject.replace("/","[slash]") \
        + "/" + date.replace("/","[slash]") \
        + "/" + extract_display_name(m_from) + "[dn]" + randomstr(20) + ".txt"
    logger.info("Fname: " + str(fname))

    res = put2s3(body, fname)
    logger.info("Response: " + str(res))

def perth_mail_body(email_obj):
    """
    Retrieve the body part of the mail.

    Parameters
    ----------
    email_obj : object structure

    Returns
    -------
    body : string
        body part of the mail
    """

    body = ""
    for part in email_obj.walk():
        logger.info("maintype: " + part.get_content_maintype())
        if part.get_content_maintype() == 'multipart':
            continue

        attach_fname = part.get_filename()

    if not attach_fname:
        charset = str(part.get_content_charset())
        if charset:
            body += part.get_payload(decode=True).decode(charset, errors="replace")
        else:
            body += part.get_payload(decode=True)
    else:
        logger.info("There is Attach File")
        body += "Error: Attachments are not supported -> " + str(part.get_payload(decode=True))

    return body

def put2s3(body, fname):
    """
    Upload files to S3.

    Parameters
    ----------
    body : string
        File Contents
    fname : string
        File Name ( Path )

    Returns
    -------
    res : dict
        API Return Values
    """

    s3 = boto3.client("s3")

    try:
        res = s3.put_object(
            Bucket=os.environ['s3BucketName'],
            ACL='private',
            Body=body,
            Key=fname,
            ContentType='text/plain'
            )
        logger.info("Success: %s has been written.", fname)
        logger.info("Success: %s", res)
    except Exception as e:
        logger.error("Error: %s", e)
    return res

def randomstr(n):
    """
    Generate a random string.

    Parameters
    ----------
    n : int
        length of a character string
    
    Returns
    -------
        : string
        random string
    """
    return ''.join(random.choices(string.ascii_letters + string.digits, k=n))

def extract_mail_address(m_from):
    """
    Extracting email addresses from a string.

    Parameters
    ----------
    m_from : string
        String containing an email address
    
    Returns
    -------
        : list
        email addresses 
    """
    pattern = r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+'
    return re.findall(pattern, m_from)[0]

def extract_display_name(m_from):
    """    
    Extracting display name from a Mail From.

    Parameters
    ----------
    m_from : string
        String containing an email address
    
    Returns
    -------
        : string
        display name
    """
    delimiter = ' <'

    if delimiter in m_from:
        idx = m_from.find(delimiter)
        logger.info("There is Display Name")
        return m_from[:idx]
    else:
        logger.info("There is no Display Name")
        return "no_display_name"

Lambda ( Slack 通知 )

通知用の Lambda は、特定のディレクトリ名 ( メールの From ) 配下にオブジェクトが作成されたことを EventBridge が検知しキックされる想定です。
つきましては、検知させる From 毎に EventBridge と Lambda の組を用意する可能性があったため、モジュール読み込みの際に複数のリソースの組を定義できるように工夫し、且つ、本リポジトリで管理していない Lambda も参照できる形にしています。

※ notifier_groups にリストとして　EventBridge と Lambda の組を渡します。

・notifier.tf.example

module "notifier" {
  source = "./modules/notifier"

  project     = var.project
  environment = var.environment

  notifier_groups = [
    {
      // Unique identifier
      name = "xxxx"
      // Triggering buckets
      bucket_name = "xxxx"
      // Triggering first renovation folder
      prefix = "xxxx"
      // Lambda Arn
      lambda_arn = "xxxx"
      // Lambda Function name
      lambda_function_name = "xxxx"
    },
    {
      name                 = "xxxx"
      bucket_name          = "xxxx"
      prefix               = "xxxx"
      lambda_arn           = "xxxx"
      lambda_function_name = "xxxx"
    },
    {
      name                 = "xxxx"
      bucket_name          = "xxxx"
      prefix               = "xxxx"
      lambda_arn           = "xxxx"
      lambda_function_name = "xxxx"
    }
  ]
}

Slack 通知用の Lambda は EventBridge から渡されたオブジェクトを読み込み、適宜整形して Slack の WebhookURL にリクエストを発行する流れとなります。

From の種類ごとに、異なった処理を行う Lambda 関数を用意する想定なので、1 サンプルとして基本的な Lambda 関数をリポジトリに配置しています。

・modules/notifier/example/slack/lambda/source/slack.py

import boto3
import json
import logging
import os
import re

from base64 import b64decode
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
from urllib.parse import quote

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):

    logger.info("Event: " + str(event))

    bucket_name = str(event['detail']['bucket']['name'])
    logger.info("Bucket: " + str(bucket_name))
    object_key = event['detail']['object']['key']
    object_list = event['detail']['object']['key'].split("/")

    for i in object_list:
        logger.info("Object: " + str(i))

    m_from = object_list[0]
    subject = object_list[1]
    date = object_list[2]
    content = read_s3_object(bucket_name, object_key)

    res = notify2slack(m_from, subject, date, content, object_key)

    logger.info("Response: " + str(res))


def read_s3_object(bucket_name, object_key):
    """
    Read files From S3.

    Parameters
    ----------
    bucket_name : string
        Bucket Name
    object_key : string
        File Name ( Path )

    Returns
    -------
    body : dict
        API Return Values
    """

    s3 = boto3.client("s3")

    try:
        res = s3.get_object(
            Bucket=bucket_name,
            Key=object_key
            )
        logger.info("Success: %s", res)
        body = res['Body'].read().decode('utf-8')
    except Exception as e:
        logger.error("Error: %s", e)
    return body

def notify2slack(m_from, subject, date, content, object_key):
    """
    Notify messages to slack.

    Parameters
    ----------
    m_from : string
        Mail From
    subject : string
        Mail Subject
    date : string
        Mail Date
    content : string
        Mail Content
    object_key : string
        File Name ( Path )
        
    Returns
    -------
    res : dict
        API Return Values
    """

    slack_message = {
        "channel": os.environ['channelName'],
        "icon_emoji": ":rotating_light:",
        "attachments": [
            {
                "color": "#FF0000",
                "title": "Email has been received.",
                "text": "<!here> \n *Content* \n ```%s``` \n" % (content),
                "fields": [
                    {
                        "title": "Date",
                        "value": date,
                        "short": True
                    },
                    {
                        "title": "Subject",
                        "value": subject,
                        "short": True
                    },
                    {
                        "title": "From",
                        "value": m_from,
                        "short": True
                    },
                    {
                        "title": "object_key",
                        "value": object_key,
                        "short": True
                    }
                ]

            }
        ]
    }

    req = Request(decrypt_hookurl(os.environ['HookUrl']), json.dumps(slack_message).encode('utf-8'))

    try:
        res = urlopen(req)
        res.read()
        logger.info("Message posted to %s", slack_message['channel'])
    except HTTPError as e:
        logger.error("Request failed: %d %s", e.code, e.reason)
    except URLError as e:
        logger.error("Server connection failed: %s", e.reason)
    return res


def decrypt_hookurl(hookurl):
    """
    Notify messages to slack.

    Parameters
    ----------
    hookurl : string
        WebhookURL that may be encrypted
        
    Returns
    -------
    hookurl : string
        WebhookURL

    decrypted_hookurl : string
        Decrypted WebhookURL
    """

    if  "hooks.slack.com" in hookurl:
        logger.info("HookURL is not Encrypted")
        return hookurl
    else:
        logger.info("HookURL is Encrypted")
        decrypted_hookurl = boto3.client('kms').decrypt(
            CiphertextBlob=b64decode(hookurl),
            EncryptionContext={'LambdaFunctionName': os.environ['AWS_LAMBDA_FUNCTION_NAME']}
        )['Plaintext'].decode('utf-8')
        return decrypted_hookurl

最後に

このブログを始めて、もう少しで 5 年が経過しそうです。
※ IT 業界で働きだしたのが 3 月なので、今月で丁度 5 年になります。
あくまでも自分の備忘録として存在しているサイトですので、稚拙な内容に関しては何卒ご容赦頂けると幸いでございます…。

引き続きどうぞよろしくお願いいたします。(•ᵕᴗᵕ•)