Working with GCP Cloud Functions and Box webhooks

Image by freepik

Google Cloud Platform (GCP) is your go-to for cloud computing solutions. With powerful tools and global infrastructure, it simplifies app development, machine learning, and infrastructure management.

Box webhooks are a powerful tool for developers working with the Box Platform. Essentially, they allow your application to be notified in near real-time whenever certain events occur within Box, such as file uploads, comments, or changes to folders.

By using webhooks, you can create more responsive and interactive integrations with Box, enhancing the overall user experience.

In this article we’ll explore how GCP Cloud Functions can work together with Box webhooks. All code is available in this GitHub repo.

Let’s get started!

Box webhooks

Box webhooks provide a way to track events happening within your Box content, delivering notifications to a designated URL whenever these events occur.

Whether it’s a file upload or a comment creation, you can set up webhooks on specific objects like files or folders, and define triggers such as FILE.UPLOAD or COMMENT.CREATE.

Once triggered, the Box Platform promptly sends the event data payload to your specified URL, ensuring seamless integration and near real-time updates for your applications.

For a comprehensive list of available triggers, refer to our guide.

Creating the GCP Cloud Functions

In this exercise we are using the Python Next Generation SDK.

You will need a Box free developer account, and a custom application configured with application and enterprise access, and client credential grants authorization. To simplify things, just enable all application scopes in your developer console.

CCG authorization

Application and enterprise access

Remember to submit the application for authorization once you save the configurations, and authorize the app in the administrator console.

Submitting the app for authorization

Authorizing the app

On the GCP side, follow their quick start guide.

Who am I

We’ll start by creating a Box “Who am I” function. Similar to a “Hello Word,” this function will simply output the current user of the CCG application.

Consider this code:

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)

We can deploy it using:

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

Returning:

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

Now if we curl that endpoint:

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

We get the user details representing the service account associated with the application:

{
    ...
    "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"
}

This proves that the GCP cloud function whoami has the correct credentials and is able to use the Box API.

Creating a webhook

We can create webhooks using the API. Let’s create a method to receive a folder_id , and a URL endpoint and automatically create the webhook for our application. In this example we are only interested in the FILE.UPLOADED event.

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()

After deploying this function to GCP, we get:

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

Now we can create the webhook. I’m using a folder named “CCS 24” which has the folder_id of 253757099719:

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"
}'

Returning:

{
    "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"
}

This webhook is now visible in the developer console:

Webhook details in the developer console

Logging the webhook payload

It is often useful to log the webhook payload for debug and discovery purposes.

Let’s create a method to log the payload:

@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)

Now if we upload a file into our target folder, we can see the webhook payload in the GCP log:

GCP log explorer

And here is a sample of the payload:

{
    "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": []
}

Securing the webhook call

Box webhooks requests do not have any defined authentication headers, this is why we are running our GCP cloud function without authentication.

However we must verify their authenticity. In the webhooks tab of your application, select the manage signature keys button, and generate 2 keys:

Webhook manage signature keys

Various different properties should be checked:

• The timestamp should be within a reasonable time frame
• The app should implement the correct signature version (at present only one)
• The app should implement the correct algorithm version (at present only one)
• The app should check if any of digest signatures sent in the headers match the digest of the body and time stamp with both signature keys

For example, validating the timestamp, has some quirks with timezones:

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

Signing the payload:

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

Putting all the checks together:

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)
    )

The reason for having 2 different keys and accepting only one as valid is so that you can rotate the keys without downtime.

Creating a Box task from a file upload

We can now finally do something useful with this webhook.

How about creating a task for a human user to go and check the uploaded files into that specific folder?

Creating a task for a user is a 2 step process:

• Create the task for the file
• Assign the task to a user

For example:

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

Next we implement this logic in our cloud function.

The Box webhook GCP cloud function

We now have all the parts to implement this webhook. The complete function looks like this:

@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)

To see this in action all we need to do is upload a file into the specified folder, and a few seconds later we see a task automatically assigned to my user:

The folder with the uploaded file and the task assigned.

We can even upload multiple files:

Multiple files uploaded, multiple tasks created

Which will trigger the webhook multiple times and create multiple tasks.

Final thoughts

Integrating GCP Cloud Functions with Box Webhooks enables near real-time event-driven applications. This combination offers scalability, responsiveness, and security.

Developers benefit from instant notifications, seamless scalability, and simplified development.

By adhering to security best practices, they can ensure data integrity and user trust.

Overall, this integration empowers developers to create dynamic and secure cloud-based applications efficiently.

Thoughts? Comments? Feedback?

Drop us a line on our community forum.