Skip to content

Implementing Webhook Endpoint

To get started with Prosa Webhook API, you need to implement your own HTTP endpoint and deploy them as a publicly accessible HTTPS URL.

Example Implementation

Here is a code snippet to help you get started quickly.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
from dataclasses import dataclass
from hashlib import sha256
from time import time
from typing import List

from fastapi import FastAPI, Body, Header
from starlette.requests import Request
from starlette.responses import Response


@dataclass
class Signatures:
    timestamp: int
    v1: List[str]

    @staticmethod
    def from_string(string: str) -> "Signatures":
        s = Signatures(0, [])
        for line in string.split(","):
            key, value = line.split("=")
            if key == "t":
                s.timestamp = int(value)
            elif key == "v1":
                s.v1.append(value)
        return s


def sign_payload(secret: str, payload: str) -> str:
    m = sha256()
    m.update(secret.encode("utf-8"))
    m.update(b".")
    m.update(payload.encode("utf-8"))
    return m.digest().hex()


def verify(secret: str, payload: str, signatures: Signatures) -> bool:
    signature = sign_payload(secret, payload)

    current_time = time()
    if current_time > signatures.timestamp + 300:  # Allow a 5 minutes tolerance
        return False

    return signature in signatures.v1


app = FastAPI(title="Example Webhook Callback")


_SECRET_KEY = "..."


@app.post("/prosa-webhook")
async def prosa_webhook(
    request: Request,
    body: dict = Body(...),
    prosa_signature: str = Header(..., alias="X-Prosa-Signature"),
    prosa_event_id: str = Header(..., alias="X-Prosa-Event-UUID"),
    prosa_event_type: str = Header(..., alias="X-Prosa-Event"),
):
    signatures = Signatures.from_string(prosa_signature)
    payload = (await request.body()).decode("utf-8")

    verified = verify(_SECRET_KEY, payload, signatures)

    if not verified:
        # Reject unverified events
        return Response(status_code=400)

    event = {
        "event_id": prosa_event_id,
        "event_type": prosa_event_type,
        "body": body
    }
    # Do something with the webhook event

    return Response()

Parsing Prosa Webhook Events

All webhook events are invoked using HTTPS with the following scheme:

Example

Header Value
X-Prosa-Event stt.jobs.completed
X-Prosa-Event-UUID 063c928c-0b07-7a03-8000-d2823fa70ca3
X-Prosa-Signature t=1674132950,v1=2535fbe6ead0c82c46597b04bb5ba96c49bdaca6714219753a83f9ef405afb47
Content-Type application/json

See events for complete list of Prosa webhook events.

Verifying Signatures

It is recommended to use webhook signature to verify events coming from Prosa servers.

Prosa uses HMAC with SHA-256.

It is possible to have multiple signatures as it can happen when you roll your secret.

Warning

Verification requires the raw payload of the requests. Any modification on the payload will cause the verification to fail.

Here is an example snippet for signature verification:

Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from dataclasses import dataclass
from hashlib import sha256
from time import time
from typing import List


@dataclass
class Signatures:
    timestamp: int
    v1: List[str]

    @staticmethod
    def from_string(string: str) -> "Signatures":
        s = Signatures(0, [])
        for line in string.split(","):
            key, value = line.split("=")
            if key == "t":
                s.timestamp = int(value)
            elif key == "v1":
                s.v1.append(value)
        return s


def sign_payload(secret: str, payload: str) -> str:
    m = sha256()
    m.update(secret.encode("utf-8"))
    m.update(b".")
    m.update(payload.encode("utf-8"))
    return m.digest().hex()


def verify(secret: str, payload: str, signatures: Signatures) -> bool:
    signature = sign_payload(secret, payload)

    current_time = time()
    if current_time > signatures.timestamp + 300:  # Allow a 5 minutes tolerance
        return False

    return signature in signatures.v1

Note

Ensure that your server uses Network Time Protocol (NTP) to ensure that your server's clock is accurate.