はじめに

この記事は、ミロゴス Advent Calendar 2023 17日目の記事です。

弊社では、S3にCSVを格納し開発外の部署で取り扱うといったケースが多く存在します。
SlackからS3のCSVを直接取りに行けたら便利だな、という発想からSlack ワークフロー → Chatbot → Lambda → S3といった流れで、CSVを取得する仕組みをCDKで実装しました。S3からのファイルの取得には署名付きURLを使用します。
この方法であればSlackさえ使えれば、誰でも簡単にS3上のファイルを取得することが可能です。

下記のような簡単な構成で作成できます。
なお、今回は既存のS3からファイルを取得するためCDKでバケットの作成はしません。

環境

AWS CDKは2.95.1を使用

cdk --version
2.95.1 (build ae455d8)

実装

CDKの実装

S3は既存のバケットが利用し、それ以外のChatbot、Lambda、SNSのリソースをCDKで作成します。

完成系

それぞれのリソース毎に解説していきます。

main_stack.py

import aws_cdk.aws_iam as iam
from aws_cdk import Stack
from aws_cdk import aws_chatbot as chatbot
from aws_cdk import aws_lambda as _lambda
from aws_cdk import aws_lambda_python_alpha as python_lambda
from aws_cdk import aws_sns as sns
from constructs import Construct


class MainStack(Stack):
    def __init__(
        self,
        scope: Construct,
    ) -> None:
        super().__init__(scope, "sample-main-stack")

        self.scope = scope

        topic = self._create_sns_topic()
        func = self._create_python_lambda(topic=topic)

        self.create_chatbot_slack_configuration(topic=topic, func=func)

    def _add_lambda_role(self, role: iam.Role, topic: sns.Topic) -> None:
        role.add_to_policy(
            iam.PolicyStatement(
                actions=["SNS:Publish"],
                resources=[
                    topic.topic_arn,
                ],
                effect=iam.Effect.ALLOW,
            ),
        )

    def _create_python_lambda(
        self,
        topic: sns.Topic,
    ) -> python_lambda.PythonFunction:
        func = python_lambda.PythonFunction(
            self,
            "function",
            entry="src/sample_function",
            runtime=_lambda.Runtime.PYTHON_3_11,
            index="main.py",
            handler="lambda_handler",
            function_name="sample-function",
            environment={
                "TOPIC_ARN": topic.topic_arn,
            },
        )
        self._add_lambda_role(func.role, topic=topic)
        return func

    def _create_sns_topic(self) -> sns.Topic:
        return sns.Topic(self, "sns-topic")

    def _add_chatbot_role(
        self,
        role: iam.Role,
        func: python_lambda.PythonFunction,
    ) -> iam.Role:
        role.add_to_policy(
            iam.PolicyStatement(
                actions=["lambda:InvokeFunction"],
                resources=[
                    func.function_arn,
                ],
                effect=iam.Effect.ALLOW,
            ),
        )
        return role

    def create_chatbot_slack_configuration(
        self,
        topic: sns.Topic,
        func: python_lambda.PythonFunction,
    ) -> chatbot.SlackChannelConfiguration:
        config = chatbot.SlackChannelConfiguration(
            self,
            "chatbot-slack-configuration",
            slack_channel_configuration_name="sample-chatbot-slack-configuration",
            slack_workspace_id="XXXXXXXXXXX",  # SlackのワークスペースID
            slack_channel_id="XXXXXXXXXXX",  # SlackのチャンネルID
            guardrail_policies=[
                iam.ManagedPolicy(
                    self,
                    "sample-chatbot-invoke-function-guardrail-policy",
                    statements=[
                        iam.PolicyStatement(
                            actions=["lambda:InvokeFunction"],
                            resources=[func.function_arn],
                            effect=iam.Effect.ALLOW,
                        ),
                    ],
                ),
            ],
            notification_topics=[topic],
        )

        self._add_chatbot_role(role=config.role, func=func)

        return config

SNSトピックの作成

SNSトピックを作成します。
後に記載するChatbotのSlack連携に必要となります。
参考：[CFn]AWS Chatbotを利用してSlackに通知を行う
CDKリファレンス：Topic

main_stack.py

    def _create_sns_topic(self) -> sns.Topic:
        return sns.Topic(self, "sns-topic")

Lambdaの作成

S3の署名付きURLを発行するLambdaを作成します。
LambdaのRoleには先ほど作成したトピックに対するSNS:Publishの権限を付与しています。
LambdaからChatbot経由でSlackへメッセージを送る際に、先ほどのトピックが必要となります。

参考：署名付き URL を使用したオブジェクトの共有
CDKリファレンス：PythonFunction

main_stack.py

    def _add_lambda_role(self, role: iam.Role, topic: sns.Topic) -> None:
        role.add_to_policy(
            iam.PolicyStatement(
                actions=["SNS:Publish"],
                resources=[
                    topic.topic_arn,
                ],
                effect=iam.Effect.ALLOW,
            ),
        )

    def _create_python_lambda(
        self,
        topic: sns.Topic,
    ) -> python_lambda.PythonFunction:
        func = python_lambda.PythonFunction(
            self,
            "function",
            entry="src/sample_function",
            runtime=_lambda.Runtime.PYTHON_3_11,
            index="main.py",
            handler="lambda_handler",
            function_name="sample-function",
            environment={
                "TOPIC_ARN": topic.topic_arn,
            },
        )
        self._add_lambda_role(func.role, topic=topic)
        return func

Chatbotの作成

最後に、SlackからLambdaを呼び出す役割とLambdaからSNS経由でSlackへメッセージを送信する役割を担うChatbotを作成します。
ここでも先ほど作成したトピックを紐づける必要があり、ChatbotのRoleにはLambdaの実行権限を付与しています。
chatbot.SlackChannelConfigurationはChatbotのSlack連携設定を作成するクラスで、通知先となるSlackのワークスペースID、チャンネルIDが必要となります。

参考：Slack URL または ID を確認する
参考：SlackのチャンネルIDを調べる方法(Webブラウザとアプリで確認)
CDKリファレンス：PythonFunction

main_stack.py

    def _add_chatbot_role(
        self,
        role: iam.Role,
        func: python_lambda.PythonFunction,
    ) -> iam.Role:
        role.add_to_policy(
            iam.PolicyStatement(
                actions=["lambda:InvokeFunction"],
                resources=[
                    func.function_arn,
                ],
                effect=iam.Effect.ALLOW,
            ),
        )
        return role

    def create_chatbot_slack_configuration(
        self,
        topic: sns.Topic,
        func: python_lambda.PythonFunction,
    ) -> chatbot.SlackChannelConfiguration:
        config = chatbot.SlackChannelConfiguration(
            self,
            "chatbot-slack-configuration",
            slack_channel_configuration_name="sample-chatbot-slack-configuration",
            slack_workspace_id="XXXXXXXXXXX",  # SlackのワークスペースID
            slack_channel_id="XXXXXXXXXXX",  # SlackのチャンネルID
            guardrail_policies=[
                iam.ManagedPolicy(
                    self,
                    "sample-chatbot-invoke-function-guardrail-policy",
                    statements=[
                        iam.PolicyStatement(
                            actions=["lambda:InvokeFunction"],
                            resources=[func.function_arn],
                            effect=iam.Effect.ALLOW,
                        ),
                    ],
                ),
            ],
            notification_topics=[topic],
        )

        self._add_chatbot_role(role=config.role, func=func)

        return config

Lambdaの中身の実装

続いて、Lambdaの中身の実装になります。
処理としてはboto3を使いS3とSNSのクライアントを作成し、ファイルの取得から署名付きURLの発行とSNSへの通知を送信しています。

参考：boto3 S3 Client
参考：boto3 SNS Client

main.py

import json
import os

import boto3

TOPIC_ARN = os.environ["TOPIC_ARN"]


def lambda_handler(event: dict, context) -> None:  # noqa: ARG001
    try:
        target_bucket = event["target_bucket"]
        output_path = event["output_path"]
        print(
            {
                "target_bucket": target_bucket,
                "output_path": output_path,
            },
        )
        presigned_url = _make_presigned_url(bucket=target_bucket, path=output_path)
        print({"presigned_url": presigned_url})
        _send_message(target_bucket=target_bucket, output_path=output_path, url=presigned_url)
    except KeyError:
        print({"message": "Invalid parameters.", "event": event})
        raise
    except Exception:
        print("Something error is happened.")
        raise

    print({"Success sending presigned url to slack."})


def _make_presigned_url(bucket: str, path: str) -> str:
    s3 = boto3.client("s3")
    return s3.generate_presigned_url("get_object", Params={"Bucket": bucket, "Key": path}, ExpiresIn=3600)


def _send_message(target_bucket: str, output_path: str, url: str) -> None:
    client = boto3.client("sns")
    request_message = _convert_sns_message_event(
        target_bucket=target_bucket,
        output_path=output_path,
        url=url,
    )

    client.publish(TopicArn=TOPIC_ARN, Message=request_message)


def _convert_sns_message_event(target_bucket: str, output_path: str, url: str) -> str:
    message = f"""
対象S3バケット: *{target_bucket}*
ファイルパス: *{output_path}*

URL: *<{url}|ダウンロード>*
"""
    return json.dumps(
        {
            "version": "1.0",
            "source": "custom",
            "content": {
                "title": "【ファイルのDLリンクが出力されました:mega:】",
                "description": message,
            },
        },
    )

バケットポリシーの設定

最後に取得対象のバケットポリシーにLambdaからのアクセス許可をしてあげる必要があります。
上記で作成したLambda RoleのARNを下記のように設定します。

bucket-policy.json

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::XXXXXXXXXX:role/sample-function-role"
            },
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::csv-sample-bucket/*"
        }
    ]
}

Slackワークフロービルダーの作成

※Chatbotの呼び出しにはSlackにAWSアプリを入れる必要があります。
ここまで来ればLambdaを実行することでファイルの取得が可能になっているはずです。

ここからはSlackからLambdaを呼び出すために、Slackワークフローを用いてChatbotからLambdaを呼び出すようにします。
今回は下記のようなワークフローを作成しました。

参考：Slack ワークフロービルダーガイド

Chatbotを呼び出すコマンドは下記の通りです。

@aws lambda invoke --payload {"target_bucket": "ワークフローからの入力値", "output_path": "ワークフローからの入力値"} --function-name sample-function --region ap-northeast-1

使い方

これで全ての準備が整ったので実際にファイルを取得してみます。
上記で作成したワークフローのリンクを用意し、実行します。

バケット名とオブジェクトパスを入力しSubmitを押すとSlackのAWSアプリでコマンドが読み込まれ、実行の許可を求められます。

[Run] commandを押下すると、Lambdaが呼び出されます。
その後、ファイルの署名付きURLが発行されます。

まとめ

元々は開発部でない方向けに作成した機能ですが、わざわざAWSコンソールを開かなくてもS3からファイルを取得できるのは自分にとってもありがたいことでした。
取得対象のバケットは限られたものになりますが、使い勝手は良いと思います。
S3のファイルを頻繁に取得する方の参考になれば幸いです。