Open in app

Sign in

Write

Sign in

GCP Cloud FunctionsとBoxのWebhookの使用

Yuko Taniguchi
Box Developer Japan Blog
Published in
22 min readApr 11, 2024

--

Image by freepik

Google Cloud Platform (GCP) は、クラウドコンピューティングソリューションにとって頼りになる存在です。強力なツールやグローバルインフラストラクチャが備わっているため、アプリ開発、機械学習、インフラストラクチャ管理が簡単になります。

BoxのWebhookは、Box Platformを使用する開発者にとって強力なツールです。本来、Webhookを使用すると、Box内で特定のイベント (ファイルのアップロード、コメント、フォルダへの変更など) が発生するたびにほぼリアルタイムでアプリケーションに通知することができます。

Webhookを使用することで、Boxとの応答性が高くよりインタラクティブな統合を作成し、全体的なユーザーエクスペリエンスを向上させることができます。

この記事では、GCP Cloud FunctionsとBoxのWebhookを併用する方法について説明します。すべてのコードは、こちらのGitHubリポジトリで入手できます。

それでは、始めましょう。

BoxのWebhook

BoxのWebhookを使用すると、Boxコンテンツ内で発生するイベントを追跡し、このようなイベントが発生するたびに、指定したURLに通知を配信することができます。

ファイルのアップロードでもコメントの作成でも、ファイルやフォルダなどの特定のオブジェクトにWebhookを設定し、FILE.UPLOADやCOMMENT.CREATEなどのトリガーを定義できます。

トリガーされると、Box Platformはすぐに、指定したURLにイベントデータペイロードを送信するため、アプリケーションではシームレスな統合とほぼリアルタイムの更新が実現します。

使用可能なトリガーの一覧については、Boxのガイドを参照してください。

GCP Cloud Functionsの作成

この演習では、Pythonの次世代SDKを使用します。

Boxの無料のDeveloperアカウント、アプリケーションとEnterpriseへのアクセスが構成されたカスタムアプリケーション、クライアント資格情報許可による承認が必要になります。シンプルにするため、開発者コンソールですべてのアプリケーションスコープを有効にしましょう。

CCGによる承認
アプリ + Enterpriseアクセス

構成を保存したら、必ず承認を受けるためにアプリケーションを送信し、管理コンソールでアプリを承認してください。

承認を受けるためにアプリを送信
アプリを承認

GCP側では、GCPのクイックスタートガイドに従ってください。

自分の情報を取得する

まず、Boxの「自分の情報を取得する」関数を作成します。「Hello World」と同様に、この関数は、CCGアプリケーションの現在のユーザーを出力するだけです。

次のコードを考えてみましょう。

import functions_framework
from utils.box_client_ccg import (
    ConfigCCG,
    get_ccg_enterprise_client,
)
from box_sdk_gen import BoxAPIError@functions_framework.http

def whoami(request):
    client = get_ccg_enterprise_client(ConfigCCG())
    try:
        me = client.users.get_user_me()
        return me.to_dict()
    except BoxAPIError as e:
        return (e.response_info.body, e.response_info.status_code)

これは、次のコードを使用して展開できます。

gcloud functions deploy whoami \
--gen2 \
--runtime=python312 \
--source=. \
--entry-point=whoami \
--trigger-http \
--allow-unauthenticated

以下が返されます。

state: ACTIVE
url: https://us-east4-box-webhooks-gcp.cloudfunctions.net/whoami

ここで、そのエンドポイントにcurlを実行します。

curl --location 'https://us-east4-box-webhooks-gcp.cloudfunctions.net/whoami'

アプリケーションに関連付けられたサービスアカウントを表すユーザーの詳細が表示されます。

{
    ...
    "id": "32541914011",
    "language": "en",
    "login": "AutomationUser_2225086_SmUBjW8wcv@boxdevedition.com",
    "max_upload_size": 2147483648,
    "name": "GCP CCG",
    "space_amount": 10737418240,
    "space_used": 0,
    "status": "active",
    "timezone": "America/Los_Angeles",
    "type": "user"
}

これにより、GCP Cloud Functionsの関数whoamiに適切な資格情報があり、Box APIを使用できることを確認できます。

Webhookを作成する

APIを使用してWebhookを作成できます。folder_idとURLエンドポイントを受け取り、自動的にアプリケーションにWebhookを作成するメソッドを作成します。この例で必要なのはFILE.UPLOADEDイベントのみです。

from box_sdk_gen.managers.webhooks import (
    Webhook,
    CreateWebhookTarget,
    CreateWebhookTargetTypeField,
    CreateWebhookTriggers,
)

@functions_framework.http
def init(request):
    # only allow POST
    if request.method != "POST":
        return "Method not allowed", 405
    # get the request data
    request_json = request.get_json()
    if request_json and "folder_id" in request_json:
        folder_id = request_json["folder_id"]
    else:
        return "Bad request, folder_id is mandatory", 400
    if request_json and "url" in request_json:
        url = request_json["url"]
    else:
        return "Bad request, url is mandatory", 400
    # folder id cannot be empty, 0 or None
    if not folder_id or folder_id == 0 or folder_id == "0" or folder_id == "":
        return (
            "Bad request, folder_id is a string, and cannot be 0, or empty",
            400,
        )
    # get the client
    client = get_ccg_enterprise_client(ConfigCCG())
    # make sure folder exists
    try:
        folder = client.folders.get_folder_by_id(folder_id)
    except BoxAPIError as e:
        return (e.response_info.body, e.response_info.status_code)
    # create the webhook
    try:
        webhook: Webhook = client.webhooks.create_webhook(
            target=CreateWebhookTarget(
                id=folder.id, type=CreateWebhookTargetTypeField.FOLDER.value
            ),
            address=url,
            triggers=[CreateWebhookTriggers.FILE_UPLOADED],
        )
    except BoxAPIError as e:
        return (e.response_info.body, e.response_info.status_code)
    return webhook.to_dict()

この関数をGCPに展開すると、次のように表示されます。

state: ACTIVE
url: https://us-east4-box-webhooks-gcp.cloudfunctions.net/init

これで、Webhookを作成できます。folder_idが253757099719の「CCS 24」という名前のフォルダを使用しているとします。

curl --location 'https://us-east4-box-webhooks-gcp.cloudfunctions.net/init' \
--header 'Content-Type: application/json' \
--data '{
    "folder_id":"253757099719",
    "url":"https://us-east4-box-webhooks-gcp.cloudfunctions.net/echo"
}'

以下が返されます。

{
    "address": "https://us-east4-box-webhooks-gcp.cloudfunctions.net/echo",
    "created_at": "2024-03-15T09:07:00-07:00",
    "created_by": {
        "id": "32541914011",
        "login": "AutomationUser_2225086_SmUBjW8wcv@boxdevedition.com",
        "name": "GCP CCG",
        "type": "user"
    },
    "id": "2546627740",
    "target": {
        "id": "253757099719",
        "type": "folder"
    },
    "triggers": [
        "FILE.UPLOADED"
    ],
    "type": "webhook"
}

次のWebhookが開発者コンソールに表示されるようになりました。

開発者コンソールのWebhookの詳細

Webhookペイロードをログに記録する

デバッグや検索の目的でWebhookペイロードをログに記録すると役立つことがよくあります。

ペイロードをログに記録するメソッドを作成します。

@functions_framework.http
def echo(request):
    # only allow POST
    if request.method != "POST":
        return "Method not allowed", 405
    # get the request headers
    # request_headers = request.headers
    box_timestamp = request.headers.get("box-delivery-timestamp")
    signature1 = request.headers.get("box-signature-primary")
    signature2 = request.headers.get("box-signature-secondary")
    box_headers = {
        "box-delivery-timestamp": box_timestamp,
        "box-signature-primary": signature1,
        "box-signature-secondary": signature2,
    }
    print(box_headers)
    # get the request data
    request_json = request.get_json()
    print(request_json)
    return ("OK", 200)

これで、ターゲットフォルダにファイルをアップロードすると、GCPのログにWebhookペイロードが表示されます。

GCPのログエクスプローラ

さらに、ペイロードのサンプルを以下に示します。

{
    "type": "webhook_event",
    "id": "b8fbc64a-5642-4331-8e03-d63c6a8d2ae2",
    "created_at": "2024-03-15T10: 13: 04-07: 00",
    "trigger": "FILE.UPLOADED",
    "webhook": {
        "id": "2546627740",
        "type": "webhook"
    },
...
    "source": {
        "id": "1472426524585",
        "type": "file",
        "sha1": "7d351bfab81a6866ada215d1541cdd48aaaad0cd",
        "name": "upload-test.png",
        "size": 423045,
...
        "parent": {
            "type": "folder",
            "id": "253757099719",
            "sequence_id": "0",
            "etag": "0",
            "name": "CCS 24"
        },
        "item_status": "active"
    },
    "additional_info": []
}

Webhookの呼び出しを保護する

Box Webhookのリクエストには定義済みの認証ヘッダーがないため、認証なしでGCP Cloud Functionsの関数を実行しています。

ただし、それが信頼できるものかどうかを検証する必要があります。アプリケーションの [Webhook] タブで [署名キーを管理] ボタンを選択し、2つのキーを生成します。

[Webhook] の [署名キーを管理]

さまざまなプロパティを確認する必要があります。

  • タイムスタンプは妥当な期間内に含まれている必要があります。
  • アプリは適切な署名バージョン (現時点では1つのみ) を実装する必要があります。
  • アプリは適切なアルゴリズムバージョン (現時点では1つのみ) を実装する必要があります。
  • アプリでは、両方の署名キーを使用して、ヘッダーで送信されたダイジェスト署名のいずれかが、本文とタイムスタンプのダイジェストと一致しているかどうかを確認する必要があります。

たとえば、タイムスタンプの確認では、タイムゾーンに関して工夫が必要です。

def validate_timestamp(timestamp_header) -> bool:
    # the timestamp must be within 10 minutes
    date = dateutil.parser.parse(timestamp_header).astimezone(pytz.utc)
    now = datetime.datetime.now(pytz.utc)
    deltaMinutes = datetime.timedelta(minutes=10)
    expiry_date = now + deltaMinutes
    is_expired = date >= expiry_date
    if is_expired:
        print(
            f"Webhook is expired: Timestamp: {date} : Now: {now} : Expiry: {expiry_date}"
        )
    return is_expired

ペイロードへの署名は次のようになります。

def sign_payload(key: str, payload: bytes, timestamp: str) -> Optional[str]:
    if key is None:
        return None
    encoded_signature_key = key.encode("utf-8")
    encoded_delivery_time_stamp = timestamp.encode("utf-8")
    new_hmac = hmac.new(encoded_signature_key, digestmod=hashlib.sha256)
    new_hmac.update(payload + encoded_delivery_time_stamp)
    signature = base64.b64encode(new_hmac.digest()).decode()
    return signature

すべての確認事項をまとめると、次のようになります。

def validate_webhook_signature(
    key_a,
    key_b,
    timestamp_header,
    signature1,
    signature2,
    payload,
    box_signature_version,
    box_signature_algorithm,
) -> bool:
    is_expired = validate_timestamp(timestamp_header)
    is_correct_version = validate_signature_version(box_signature_version)
    is_correct_algorithm = validate_signature_algorithm(
        box_signature_algorithm
    )
    digest_key_a = sign_payload(key_a, payload, timestamp_header)
    digest_key_b = sign_payload(key_b, payload, timestamp_header)
    is_signature1_valid = digest_key_a == signature1
    is_signature2_valid = digest_key_b == signature2
    if not is_signature1_valid:
        print(f"Invalid signature 1: {digest_key_a} != {signature1}")
    if not is_signature2_valid:
        print(f"Invalid signature 2: {digest_key_b} != {signature2}")
    return (
        not is_expired
        and is_correct_version
        and is_correct_algorithm
        and (is_signature1_valid or is_signature2_valid)
    )

異なる2つのキーを用意していて、1つだけを有効なキーとして承認する理由は、ダウンタイムなしでキーをローテーションできるようにするためです。

ファイルのアップロードからBoxタスクを作成する

これでようやく、このWebhookを活用できます。

ユーザーが特定のフォルダにアップロードされたファイルを確認するためのタスクを作成してみましょう。

ユーザーのタスクは、以下の2つの手順からなるプロセスで作成できます。

  • ファイルのタスクを作成する
  • タスクをユーザーに割り当てる

以下に例を示します。

def create_file_task(
    client: BoxClient, file: File, user: User, message: str
) -> TaskAssignment:
    task: Task = client.tasks.create_task(
        CreateTaskItem(type=CreateTaskItemTypeField.FILE.value, id=file.id),
        action=CreateTaskAction.REVIEW.value,
        message=message,
        completion_rule=CreateTaskCompletionRule.ANY_ASSIGNEE,
    )
    task_assignment: TaskAssignment = (
        client.task_assignments.create_task_assignment(
            CreateTaskAssignmentTask(
                type=CreateTaskAssignmentTaskTypeField.TASK,
                id=task.id,
            ),
            CreateTaskAssignmentAssignTo(id=user.id),
        )
    )
    return task_assignment

次に、このロジックをCloud Functionsの関数に実装します。

BoxのWebhookを使用したGCP Cloud Functionsの関数

これで、このWebhookを実装するためのすべてのパーツが揃いました。完成した関数は次のようになります。

@functions_framework.http
def box_webhook(request):
    # only allow POST
    if request.method != "POST":
        return "Method not allowed", 405
    # get the request headers
    timestamp_header = request.headers.get("box-delivery-timestamp")
    signature1 = request.headers.get("box-signature-primary")
    signature2 = request.headers.get("box-signature-secondary")
    box_signature_version = request.headers.get("box-signature-version")
    box_signature_algorithm = request.headers.get("box-signature-algorithm")
    config = ConfigCCG()
    # validate the webhook signature
    is_valid = validate_webhook_signature(
        config.key_a,
        config.key_b,
        timestamp_header,
        signature1,
        signature2,
        request.data,
        box_signature_version,
        box_signature_algorithm,
    )
    if not is_valid:
        return ("Invalid signature", 401)
    payload = request.get_json()
    # get a box client
    client = get_ccg_enterprise_client(config)
    # file_id = (payload.get("source", {}).get("id"),)
    # user_id = (config.ccg_user_id,)
    task_assignment = create_file_task(
        client,
        file_id=payload.get("source", {}).get("id"),
        user_id=config.ccg_user_id,
        message="Please review this file",
    )
    print(task_assignment)
    return ("OK", 200)

この実際の動作を確認するには、指定されたフォルダにファイルをアップロードするだけです。数秒後、タスクが自動的にユーザーに割り当てられます。

ファイルがアップロードされたフォルダと割り当てられたタスク

複数のファイルをアップロードすることもできます。

複数のファイルのアップロードと複数のタスクの作成

この場合は、Webhookが複数回トリガーされ、複数のタスクが作成されます。

まとめ

GCP Cloud FunctionsとBox Webhookを統合することで、ほぼリアルタイムのイベント駆動型アプリケーションが実現します。この組み合わせにより、拡張性、応答性、およびセキュリティが提供されます。

開発者には、即時通知、シームレスな拡張性、開発の簡素化というメリットがあります。

セキュリティのベストプラクティスに従うことで、データの整合性とユーザーの信頼を確保できます。

この統合により、開発者は、動的で安全なクラウドベースのアプリケーションを効率的に作成できるようになります。

アイデア、コメント、フィードバックがある場合は、コミュニティフォーラム (英語のみ) にコメントをお送りください。

Box Platform
Python
ソフトウェア開発
ソフトウェアエンジニアリング
Google Cloud Platform

--

--

Yuko Taniguchi
Box Developer Japan Blog

Written by Yuko Taniguchi

13 Followers
Editor for

Help

Status

About

Careers

Press

Blog

Privacy

Terms

Text to speech

Teams