🐧

Slackコマンドからのリクエストの認証をLambdaで実装する(Python)

2022/01/17に公開

AWS

はじめに

こんにちは、クラスメソッド AWS事業本部の筧です。
皆さんは Slack のスラッシュコマンドをよく使いますか？

先日投稿した Lambda から Amazon Connect を使って自動音声で架電する仕組みを、Slack スラッシュコマンド（以降、Slack コマンド）で呼び出ししたくなったので今回実装しました。Slack コマンドで躓きやすい箇所として、リクエスト認証が挙げられるかと思います。今回はリクエスト認証の処理をご紹介します。

やってみた

Slack の設定

以下のブログを参考にして、Slack App の作成と Slash Commands の設定をしてください。
App Credentials の Signing Secret は Lambda の環境変数 SLACK_API_SIGNING_SECRET に後ほど利用します。

Lambda

Slack コマンドの仕様上、3秒以内に処理を完了し Slack コマンドへレスポンスを返却する必要があります。そのため今回は以下のような構成にします。

request.py
- メイン処理をキックする
- Slack コマンドのリクエストに即座にレスポンスを返す
call.py
- リクエスト認証して、成功したら架電するメイン処理

src/handlers/request.py

import json
import logging
import os
from typing import Dict, List, Optional
from urllib import parse

import boto3
from mypy_boto3_ssm import SSMClient
from mypy_boto3_stepfunctions.client import SFNClient

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

TOKYO = "ap-northeast-1"


def fetch_ssm_param(param_name: str) -> Optional[str]:
    ssm: SSMClient = boto3.client("ssm", region_name=TOKYO)
    resp = ssm.get_parameter(Name=param_name)["Parameter"]["Value"]
    return resp


def handler(event, context) -> None:
    """
    request エントリーポイント
    """
    logger.info(f"event: {event}")

    sfn: SFNClient = boto3.client("stepfunctions", region_name=TOKYO)
    sfn.start_execution(
        stateMachineArn=fetch_ssm_param(os.environ["STATEMACHINE_ARN"]),
        input=json.dumps(event),
    )

    logger.info(f'invoke {os.environ["STATEMACHINE_ARN"]}')
    return {
        "statusCode": 200,
        "body": json.dumps(
            {"text": "Now, calling command. Please, stay a few seconds."}
        ),
    }

Slack コマンドを実行したチャンネルに、「Now, calling command. Please, stay a few seconds.」というレスポンスを即座に返します。
STATEMACHINE_ARN はメイン処理のステーマシン ARN です。SSM のパラメータストアに格納して、fetch_ssm_param から呼び出しています。fetch_ssm_param については以下のブログを参照ください。

src/handlers/call.py

import datetime
import hashlib
import hmac
import json
import logging
import os
from typing import Dict, Optional
from urllib import parse

import boto3
import requests
import src.use_cases.call
from mypy_boto3_secretsmanager.client import SecretsManagerClient

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

TOKYO = "ap-northeast-1"


def verify(headers: Dict, body: Dict, slack_key: str) -> bool:
    """
    slackからのmessageかを判別する
    see: https://api.slack.com/authentication/verifying-requests-from-slack
    """
    try:
        request_ts = int(headers["X-Slack-Request-Timestamp"])
        now_ts = int(datetime.datetime.now().timestamp())
        if abs(request_ts - now_ts) > (60 * 5):
            return False

        signature = headers["X-Slack-Signature"]
        message = f"v0:{headers['X-Slack-Request-Timestamp']}:{body}"
        message_hmac = hmac.new(
            bytes(slack_key, "UTF-8"), bytes(message, "UTF-8"), hashlib.sha256
        )
        expected = f"v0={message_hmac.hexdigest()}"
    except Exception:
        return False
    else:
        return hmac.compare_digest(expected, signature)


def handler(event, context) -> Dict:
    """
    call エントリーポイント
    """
    logger.info(f"event: {event}")

    # Get slack Signing Secret
    secret_name = os.environ["SLACK_API_SIGNING_SECRET"]
    secretsmanager: SecretsManagerClient = boto3.client(
        "secretsmanager", region_name=TOKYO
    )
    resp = secretsmanager.get_secret_value(SecretId=secret_name)
    secret = json.loads(resp["SecretString"])

    # Slack認証失敗したらエラーメッセージをSlack投稿して終了
    if not verify(event["headers"], event["body"], secret["key"]):
        logger.info("don't invoke")

        request_body = parse.parse_qs(event["body"])
        response_url = request_body["response_url"][0]
        payload = {
            "text": f"Slackコマンドの認証に失敗しました",
            "color": "warning",
            "response_type": "in_channel",
        }
        requests.post(response_url, data=json.dumps(payload))
        logger.info("post error message to slack")
        return {}

    # 架電処理実行
    logger.info("invoke")
    resp = src.use_cases.call.exec(event)
    return resp

request.py の STATEMACHINE_ARNに定義した、ステートマシンの最初に定義している Lambda です。
verify が Slack コマンドのリクエストを認証する関数です。True の場合のみ、メイン処理（今回は、src.use_cases.call.exec）を実行します。False の場合は、「Slackコマンドの認証に失敗しました」という旨を Slack コマンドを実行したチャンネルに返して終了します。
SLACK_API_SIGNING_SECRET には Slack 側の設定にあった App Credentials の Signing Secret です。AWS Secrets Manager に格納して呼び出しています。呼び出す処理については参考までに以下のブログをご紹介します。

Step Functions

参考までにステートマシンの定義をご紹介します。Fn::GetAtt: [call, Arn] では、src/handlers/call.py を参照しています。

includes/state-machines.yml

stateMachines:
  CallCommandFunc:
    name: CallCommandFunc-${self:provider.stage}
    definition:
      StartAt: Call
      States:
        Call:
          Type: Task
          InputPath: "$"
          Resource:
            Fn::GetAtt: [call, Arn]
          ResultPath: "$"
          Next: Done
        Done:
          Type: Pass
          End: true

あとがき

Slack コマンドでリクエスト認証が必要になった時に当該コードを使い回せるかと思います。細かい処理内容の説明は随時追記します。コードの全体像もツールが完成したら公開予定です。

それではまた！

はじめに

やってみた

Slack の設定

Lambda

Step Functions

あとがき

Discussion