こんにちは、エンジニアのクロ(@kro96_xr)です。
今回はタイトルの構成で署名付きURLを生成する仕組みを作ってみました。
はじめに
ことの発端は「GithubActionsでビルドした成果物を社内メンバーがダウンロードできるようにしたい」という要望です。
素直に実装するのであれば、AWS Cognitoを使ってクライアントアプリケーションを実装することになりそうです。
しかし、
- 極力工数はかけずに作れるのであればその方が望ましい
- 厳密なチェックも不要
という要件もあったので、Slackのinteractive messagesを使って署名付きURLを都度発行する方法を検証してみました。
これは、Slackのワークスペースに含まれているユーザーを社内メンバーとみなして一定時間アクセスを許可するという考え方です。
さらにダウンロード出来るユーザーを絞りたければプライベートチャンネルを送信先にしてもいいでしょう。
超雑な構成図は以下のようになります。
なお、今回の記事にはGithubActions上でのファイルアップロード処理については含まれておりませんのでご了承ください。
実装内容
GithubActions、Slack、Lambdaの設定や実装についてそれぞれ説明します。
Slack
まずはSlackアプリを作成し、送信先のワークスペースとチャンネルの設定を行います。
設定方法については以下のリンクが参考になります。
Slack が提供する GitHub Action "slack-send" を使って GitHub から Slack に通知する - Qiita
作成が完了すると、アプリのIDとWebhookのURLを取得することができます。
- Basic InformationからApp IDを確認
- Install AppからWebhook URLを確認
後々ボタンを押した時のコールバック先を設定する必要がありますが、現状はOFFのままで構いません。
これでSlackの設定は一旦終了です。
Lambda
次にLambdaの実装を行います。今回はPythonを使っています。
やっていることは以下の通りです。
- Slackから送信されたPayloadのパース
- SlackアプリIDのチェック
- S3のオブジェクトキー名の取得と署名付きURLの発行
- Slackへのメッセージ送信
先に全体を載せておきます。
import datetime import pytz import boto3 import logging import urllib.parse import json import http.client from botocore.exceptions import NoCredentialsError # 定数 APP_ID = 'YOUR_APP_ID' BUCKET_NAME = 'YOUR_BUCKET_NAME' EXPIRATION_SECONDS = 300 # 5分 # ログ設定 logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): # S3クライアントの設定 s3_client = boto3.client('s3') # event["body"]をJSON形式にパースする処理 # "payload={body}"という文字列から"payload="を削除 payload = event["body"].split('=', 1)[-1] # URLデコード url_decoded_payload = urllib.parse.unquote(payload) # JSONパース json_payload = json.loads(url_decoded_payload) # SlackのAppIDで制限 app_id = json_payload['api_app_id'] if app_id != APP_ID: err = "Unauthorized AppID" return { 'statusCode': 403, 'body': err } # JSONから返信用URLとオブジェクトキー(value)を取得 object_key = "" slack_url = urllib.parse.urlparse(json_payload['response_url']) if 'actions' in json_payload: for action in json_payload['actions']: value = action.get('value') object_key = value else: # アクションが存在しない場合はエラー err = "No Action found" send_error_message(slack_url, err) return { 'statusCode': 400, 'body': err } try: # 署名付きURLを取得 presigned_url = s3_client.generate_presigned_url('get_object', Params={'Bucket': BUCKET_NAME, 'Key': object_key}, ExpiresIn=EXPIRATION_SECONDS) # 権限エラーの場合 except NoCredentialsError as e: err = "No AWS credentials found" send_error_message(slack_url, err) logging.error(f'Error occurred: {e}') # レスポンス return { 'statusCode': 400, 'body': err } # その他のエラーの場合 except Exception as e: err = "An unexpected error occurred" send_error_message(slack_url, err) # エラーメッセージをログに出力 logging.error(f'Error occurred: {e}') # レスポンス return { 'statusCode': 500, 'body': err } # 有効期限取得 parsed_url = urllib.parse.urlparse(presigned_url) parsed_query = urllib.parse.parse_qs(parsed_url.query) expires_str = "" expires = parsed_query.get('Expires') if expires: expires = expires[0] # parse_qsが配列を返すため expires_float = float(expires) # unixtimeをfloatに変換 expires_datetime = datetime.datetime.utcfromtimestamp(expires_float) # datetimeに変換 jst = pytz.timezone('Asia/Tokyo') expires_datetime_jst = expires_datetime.replace(tzinfo=datetime.timezone.utc).astimezone(jst) expires_str = expires_datetime_jst.strftime('%Y-%m-%d %H:%M:%S %Z%z') # 文字列に再変換 # Slackへのメッセージ送信 resp = send_presigned_url(slack_url, presigned_url, expires_str) # レスポンス if resp.status != 200: err = f"Slack API call failed: {resp.status}" return { 'statusCode': 500, 'body': err } return { 'statusCode': 200, 'body': "Message sent to Slack" } # Slackへの署名付きURL送信 def send_presigned_url(response_url, presigned_url, expires_str): message = f""" Here is your signed URL: {presigned_url} Expires: {expires_str} """ slack_msg = { "replace_original": False, "text": message, } resp = send_message(response_url, slack_msg) return resp # Slackへのエラーメッセージ送信 def send_error_message(response_url, error): slack_msg = { "replace_original": False, "text": "Error: " + error, } resp = send_message(response_url, slack_msg) return resp # Slackへのメッセージ送信 def send_message(response_url, message): headers = { 'Content-Type': 'application/json', } conn = http.client.HTTPSConnection(response_url.netloc) conn.request("POST", response_url.path, body=json.dumps(message), headers=headers) resp = conn.getresponse() return resp
次に、個別にポイントだけ見ていきます。
- Slackから送信されたPayloadのパース
# event["body"]をJSON形式にパースする処理 # "payload=AAA"という文字列から"payload="を削除 payload = event["body"].split('=', 1)[-1] # URLデコード url_decoded_payload = urllib.parse.unquote(payload) # JSONパース json_payload = json.loads(url_decoded_payload)
この処理はSlackから送られたPayloadを取得する処理です。リファレンスにある通り、payloadをjsonにパースしてやります。
と言いつつリファレンスの記載がわかりづらかったので、実際にはログを出力して地道に確認しながらやりました。
We mentioned above that there were a few different types of interaction payloads your app might receive. They'll be sent to your specified Request URL in an HTTP POST request in the form application/x-www-form-urlencoded. For more information, refer to Using the Slack Web API: Basics. The body of the request will contain a payload parameter; your app should parse this payload parameter as JSON.
Handling user interaction in your Slack apps | Slack
- SlackアプリIDのチェック
# SlackのAppIDで制限 app_id = json_payload['api_app_id'] if app_id != APP_ID: err = "Unauthorized AppID" return { 'statusCode': 403, 'body': err }
この処理はアプリ作成時に取得したIDをチェックしています。
これにより自社のワークスペース以外からのアクセスを防いでいます。このコードではIDをコード内に埋め込んでいますが、実際には環境変数に設定した方がいいかと思います。
- S3のオブジェクトキー名の取得と署名付きURLの発行
# JSONから返信用URLとオブジェクトキー(value)を取得 object_key = "" slack_url = urllib.parse.urlparse(json_payload['response_url']) # 返信用URL if 'actions' in json_payload: # actionsは実際には1つしかない for action in json_payload['actions']: value = action.get('value') object_key = value else: # アクションが存在しない場合はエラー # 中略 try: # 署名付きURLを取得 presigned_url = s3_client.generate_presigned_url('get_object', Params={'Bucket': BUCKET_NAME, 'Key': object_key}, ExpiresIn=EXPIRATION_SECONDS)
この処理はオブジェクトキーを取得し、署名付きURLの発行を行う処理です。
後述しますが、Slackのボタンにvalueを割り当てることができるため、そこにオブジェクトキーをセットする想定です。
- Slackへのメッセージ送信
# Slackへの署名付きURL送信 def send_presigned_url(response_url, presigned_url, expires_str): message = f""" Here is your signed URL: {presigned_url} Expires: {expires_str} """ slack_msg = { "replace_original": False, "text": message, } resp = send_message(response_url, slack_msg) return resp # Slackへのエラーメッセージ送信 # 略 # Slackへのメッセージ送信 def send_message(response_url, message): headers = { 'Content-Type': 'application/json', } conn = http.client.HTTPSConnection(response_url.netloc) conn.request("POST", response_url.path, body=json.dumps(message), headers=headers) resp = conn.getresponse() return resp
この処理でSlackにメッセージを送っています。
replace_originをFalseにしないと元のメッセージを上書きしてしまいます。
以上でLambdaの実装は完了ですが、Lambdaをデプロイ後にSlackアプリの設定が必要です。
アプリの設定画面からInteractivity & ShortcutsをONにし、RequestURLにLambdaのエンドポイントを設定すればOKです。
Github Actions
最後にGithubActionsからSlackにメッセージを送信する部分の実装です。
まずはじめに、リポジトリの設定からRepository secretsを作成し、WebhookのURLを設定しておきます。
そして、GithubActions用のyamlを書いていきます。
name: Slack Notification on: push: branches: - main jobs: slack_notification: runs-on: ubuntu-latest steps: - name: Checkout Repository uses: actions/checkout@v2 - name: Send GitHub Action trigger data to Slack workflow id: slack uses: slackapi/slack-github-action@v1.24.0 with: payload: | { "text": "ビルド完了", "blocks": [ { "type": "section", "text": { "type": "plain_text", "text": "ビルド完了", "emoji": true } }, { "type": "actions", "elements": [ { "type": "button", "text": { "type": "plain_text", "text": "Download", "emoji": false }, "value": "OBJECT_KEY", "action_id": "action_id" } ] } ] } env: SLACK_WEBHOOK_TYPE: INCOMING_WEBHOOK SLACK_WEBHOOK_URL: ${{ secrets.SLACK_INCOMING_WEBHOOK_URL }}
このように実装することで、mainブランチにマージされたタイミングでSlackにボタンが送信できます。
あとはビルド成果物をS3にアップロードする処理を追加し、アップロード先のオブジェクトキーをvalueに設定してやれば一連の仕組みは完成です。
なお、ボタンを増やしたい場合やメッセージを整えたい場合はBlockKitを使うと良さそうです。
以下が参考になるかと思います。
Slack で UI を構築するためのフレームワーク「Block Kit」の基本を確認してみた | DevelopersIO
おわりに
以上、極力手間をかけずに社内向けのファイルダウンロード機能を実装してみました。
何かの参考になれば幸いです。
