【Flask】でHTTP(S) Callback / Webhook を構築する

Flask で HTTP Callback をたてる。

• 特定のイベントに対してリアルタイムで通知を受け取るwebhookを構築する
• アプリ側でステータスが変更された際に jsonフォーマットで送られてくる notification 受け取る
• 認証はBear tokenを利用

環境

• macOS Sonoma 14.6.1
• Python 3.12.3

スニペットとファイル

Flask app

• https_auth_webhook.py

PEMs

• cert.pem
• key.pem
  ※httpsの設定をするための鍵

シーケンス

jsonデータ (サンプル)

1. cert.pemとkey.pemを作成する

• nodesをつけて、パスフレーズなしで秘密鍵を保存
• 開発環境や自動化されたスクリプトなどで手動のパスフレーズ入力を避けたい場合に便利

openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem -days 365 -nodes

2. https_auth_webhook.pyを作成

import json
import os

from flask import Flask, jsonify, request

app = Flask(__name__)

VALID_TOKEN = "auth_token" #本番では安全性の高いものを使うこと！


def authenticate(request):
    """Bearer Auth Token"""
    auth_header = request.headers.get("Authorization")
    print(auth_header)
    if auth_header and auth_header.startswith("Bearer "):
        token = auth_header.split(" ")[1]  
        if token == VALID_TOKEN:
            return True
    return False


@app.route("/webhook", methods=["POST"])
def webhook():

    if not authenticate(request):
        return jsonify({"status": "fail", "message": "Unauthorized"}), 401

    if request.is_json:
        data = request.get_json()

        # JSONデータをログに出力
        print(f"Received JSON: {data}")

        # 同じディレクトリにjson logファイルを置く
        base_dir = os.path.dirname(__file__)
        webhook_log = os.path.join(base_dir, "webhook_log.json")

        with open(webhook_log, "a") as logfile:
            json.dump(data, logfile)  # JSONデータをファイルに書き込む
            logfile.write("\n")  # 次のログが新しい行に書かれるように改行を追加

        return jsonify({"status": "success", "message": "Webhook received"}), 200
    else:
        return jsonify({"status": "fail", "message": "Invalid content type"}), 400


if __name__ == "__main__":
    app.run(
        ssl_context=("cert.pem", "key.pem"),    #httpsで起動させる
        host="0.0.0.0",
        port=8888,
        debug=True,
    )

3. スニペットを起動する

% ./https_auth_webhook.py
 * Serving Flask app 'https_auth_webhook'
 * Debug mode: on
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on all addresses (0.0.0.0)
 * Running on https://127.0.0.1:8888
 * Running on https://172.24.5.98:8888
Press CTRL+C to quit
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 817-440-211

4. curl で notification を post する

% curl -k -X POST https://127.0.0.1:8888/webhook \
-H "Content-Type: application/json" \
-H "Authorization: Bearer auth_token" \
-d '{
   "id": "string",
   "source": "string",
   "type": "org.camaraproject.qod.v0.qos-status-changed",
   "specversion": "1.0",
   "datacontenttype": "application/json",
   "time": "2024-04-30T13:05:29.189Z",
   "event": {
      "eventType": "QOS_STATUS_CHANGED",
      "eventTime": "string",
      "eventDetail": {
         "sessionId": "string",
         "qosStatus": "AVAILABLE",
         "statusInfo": "DURATION_EXPIRED"
      }
   },
   "data": {
      "sessionId": "string",
      "qosStatus": "AVAILABLE",
      "statusInfo": "DURATION_EXPIRED"
   }
}'
{
  "message": "Webhook received",
  "status": "success"
}

Press CTRL+C to quit
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 817-440-211

Bearer auth_token
Received JSON: {'id': 'string', 'source': 'string', 'type': 'org.camaraproject.qod.v0.qos-status-changed', 'specversion': '1.0', 'datacontenttype': 'application/json', 'time': '2024-04-30T13:05:29.189Z', 'event': {'eventType': 'QOS_STATUS_CHANGED', 'eventTime': 'string', 'eventDetail': {'sessionId': 'string', 'qosStatus': 'AVAILABLE', 'statusInfo': 'DURATION_EXPIRED'}}, 'data': {'sessionId': 'string', 'qosStatus': 'AVAILABLE', 'statusInfo': 'DURATION_EXPIRED'}}
127.0.0.1 - - [05/Oct/2024 23:52:03] "POST /webhook HTTP/1.1" 200 -

ポイント

• webhookは認証情報も送るのでHTTPではなく、HTTPSで暗号化
• Flaskの開発用環境サーバなので、本番環境で使うのはNG