Webhooks

Note, all requests must be authenticated with Bearer Authentication. Make sure your request has a header containing your Nextmv Cloud API key, as such:

Key: Authorization
Value: Bearer <YOUR-API-KEY>

Bash

Authorization: Bearer <YOUR-API-KEY>

Export your API key as an environment variable:

Bash

export NEXTMV_API_KEY="<YOUR-API-KEY>"

Create a webhook

The following snipped uses curl to create a webhook. Please change the id, endpoint_url, and description to your desired values. Also note that at the time of writing, only the run.status event type is supported.

Bash
Output (JSON)

curl -X 'POST' \
  'https://api.cloud.nextmv.io/v1/webhooks' \
  -H 'accept: */*' \
  -H "Authorization: Bearer $NEXTMV_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{
  "id": "SOME_ID",
  "event_types": [
    "run.status"
  ],
  "endpoint_url": "A_REACHABLE_URL",
  "description": "SOME DESCRIPTION"
}'

Getting the webhook secret

After creating a webhook, you can retrieve the secret for the webhook. Note that we are using a placeholder for the id in the URL. The secret is used to recompute the HMAC signature of the request to verify its authenticity.

Bash
Output (JSON)

curl -X 'GET' \
  'https://api.cloud.nextmv.io/v1/webhooks/SOME_ID/secret' \
  -H 'accept: application/json' \
  -H "Authorization: Bearer $NEXTMV_API_KEY"

Writing a webhook handler

We will send POST requests to the endpoint_url with a JSON payload. In the header of the request, we will include the Nextmv-Signature that you can use to check the authenticity of the request. It has the following format:

Text

Nextmv-Signature: t=UNIX_TIME_IN_MILLISECONDS,v1=HMAC_SHA256_ENCODED_SIGNATURE

In order to check the authenticity of the request, you need to recompute the HMAC signature using the secret, the payload (the request body), and the timestamp t from the header.

See the following example in Python:

Python

def check_signature(payload: bytes, t: int, signature: str, secret: str):
    """
    Recompute the signature of the payload using the secret and the timestamp. Compare the
    recomputed signature with the signature provided in the header.
    """
    mac = hmac.new(secret, digestmod=hashlib.sha256)
    mac.update(str(t).encode())
    mac.update(b".")
    mac.update(payload)
    recomputed_signature = mac.hexdigest()
    return recomputed_signature == signature

The following code snippet shows a basic self-contained FastAPI web server that listens for POST requests on the /webhookhandler endpoint. It checks the signature and returns a 200 status code if the signature is valid. Simply place the code in a file named main.py and run it with uvicorn (see below).

Python

import hashlib
import hmac
import time

from fastapi import FastAPI, Request, Response

SECRET = "GET_YOUR_SECRET_FROM_A_SECURE_LOCATION"

app = FastAPI()


def check_signature(payload: bytes, t: int, signature: str, secret: str):
    """
    Recompute the signature of the payload using the secret and the timestamp. Compare the
    recomputed signature with the signature provided in the header.
    """
    mac = hmac.new(secret, digestmod=hashlib.sha256)
    mac.update(str(t).encode())
    mac.update(b".")
    mac.update(payload)
    recomputed_signature = mac.hexdigest()
    return recomputed_signature == signature


@app.post("/webhookhandler")
async def create_item(request: Request):
    signature = request.headers.get("nextmv-signature")

    # Extract timestamp and signature from header
    signature_time, signature_string = 0, ""
    try:
        t, sig = signature.split(",")
        now = time.time() * 1000
        signature_time = int(t.split("=")[1])
        signature_string = sig.split("=")[1]
        # Check if the timestamp is not older than 5 minutes and not in the
        # future. This means to avoid replay attacks.
        if not (now - 300000 < signature_time < now + 5000):
            print(
                f"Invalid Time Value: {now - 300000} < {signature_time} < {now + 5000}"
            )
            return Response(
                status_code=401,
                headers={"content-type": "text/plain"},
                content=bytes("Unauthorized: Invalid Time Value", "utf-8"),
            )
    except Exception:
        return Response(
            status_code=401,
            headers={"content-type": "text/plain"},
            content=bytes("Unauthorized: Invalid Signature Format", "utf-8"),
        )

    # Make sure secret is set
    if not SECRET:
        return Response(
            status_code=500,
            headers={"content-type": "text/plain"},
            content=bytes("Internal Server Error", "utf-8"),
        )

    # Check signature
    body = await request.body()
    if not check_signature(body, signature_time, signature_string, SECRET.encode()):
        return Response(
            status_code=401,
            headers={"content-type": "text/plain"},
            content=bytes("Unauthorized: Invalid signature", "utf-8"),
        )

    # Process the webhook
    return Response(
        status_code=200,
        headers={"content-type": "text/plain"},
        content=bytes("Webhook received!", "utf-8"),
    )

The required dependencies can be installed with:

Bash

pip install fastapi uvicorn

You can run the server with:

Bash

uvicorn main:app --reload --host 0.0.0.0

Once you have a webhook set up and the handler is ready, you can use the trigger function to send a request to the webhook:

Bash

curl -X 'POST' \
  'https://api.cloud.nextmv.io/v1/events/trigger' \
  -H 'accept: */*' \
  -H "Authorization: Bearer $NEXTMV_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{
  "event_type": "run.status"
}'

Currently there is only one event type supported: run.status. The request body of a triggered event of this type will look like this:

JSON

{
  "event_type": "run.status",
  "api_version": "2024-03-04",
  "data": {
    "run_id": "test-run-id",
    "status": "succeeded",
    "status_v2": "succeeded",
    "created_at": "2024-03-21T13:19:34Z",
    "duration": 1000,
    "input_size": 1000,
    "output_size": 1000,
    "error": null,
    "application_id": "app_id",
    "application_instance_id": "app_instance_id",
    "application_version_id": "app_version_id"
  }
}

For further endpoints to interact with webhooks, see the OpenAPI specification.