> For the complete documentation index, see [llms.txt](https://docs.developer.disruptive-technologies.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.developer.disruptive-technologies.com/data-connectors/receiving-events.md). # Receiving Events A few things to consider when receiving events forwarded by a Data Connector. ## Request Contents Events are delivered to the receiving endpoint as HTTPS POST requests. Both the headers and body of the incoming request contain information of interest that can be extracted and, depending on the [configuration](/data-connectors/advanced-configurations.md), used to verify the content and origin of the request. {% hint style="info" %} **Exploring the request contents** You can use [webhook.site](https://webhook.site) to explore the contents of the POST requests. This site will generate a URL that you can point a Data Connector to, and it will display all the relevant details about each request. Note that using this service will make the events publicly available through the URL generated by that site. We recommend exploring this with emulators in a separate project in DT Studio. {% endhint %} ### Header Every forwarded event includes the following header. * **DT-Asymmetric-Signature**\ Includes a [JSON Web Token](https://jwt.io/) (JWT) signed by DT with an ECDSA private key (`ES256`). It is verified using DT's public key, published at a JWKS endpoint. No shared secret is involved. This is the recommended way to verify events. If a [signature secret](/data-connectors/advanced-configurations.md#signing-events) is set in the Data Connector configuration, the following header will be included. * **X-Dt-Signature**\ Includes a JWT signed with your Signature Secret using `HS256`. Supported for existing integrations. Both tokens carry a checksum of the same request body. Verifying either one is sufficient; if both are present, prefer `DT-Asymmetric-Signature`. Note that depending on the framework used to receive the Data Connector events, the header name casing may differ. Some services will force header names to be lower-cased, like `x-dt-signature`. Read more about validating the request signature in the [Verifying Signed Events](#verifying-signed-events) section below. ### Body The request body contains three fields, `event`, `labels`, and `metadata`. The following snippet shows an example request body of a `touch` event for a `humidity` sensor forwarded by a Data Connector. ```javascript { "event": { "eventId": "", "targetName": "projects//devices/", "eventType": "touch", "data": { "touch": { "updateTime": "2021-05-28T08:34:06.225872Z" } }, "timestamp": "2021-05-28T08:34:06.225872Z" }, "labels": { "room-number": "99" }, "metadata": { "deviceId": "", "projectId": "", "deviceType": "humidity", "productNumber": "102081" } } ```
FieldTypeDescription
FieldTypeDescription
eventstructContains event data. See the Event documentation where the structure is explained in detail for each event type.
labelsstructDevice label key- and value pairs included by the Data Connector. See the Advanced Configuration page for details about including labels.
metadatastructContains metadata about the device that is the source of the event. See the section below for more details.
#### Event Metadata Each event has a `metadata` field that includes details about the device that is the source of the event. The event metadata has the following structure. | Field | Type | Description | | --------------- | -------- | -------------------------------------------------------------------------------------------- | | `deviceId` | `string` | The identifier of the device that published the event. | | `projectId` | `string` | The identifier of the project the device is in. | | `deviceType` | `string` | The [device type](/concepts/devices.md#device-types) that published the event. | | `productNumber` | `string` | The [product number](/concepts/devices.md#structure) of the device that published the event. | {% hint style="info" %} **Note** The structure of the `metadata` field might change in the future if event types are added that are not published by devices. Make sure to first check `event.eventType` to make sure it is a known device event before processing the `metadata` field. See the code sample below for an example of how to do this. {% endhint %} The event metadata makes it possible to check which [device type](/concepts/devices.md#device-types) has published the event, even for event types like `touch` or `networkStatus` which are published by many types of devices. This makes it possible to add new devices to a database without having to first look up the device using the REST API. The metadata also provides a more convenient way to get the `deviceId` and `projectId` of the device that published the event, without having to parse the `event.targetName` field. ## Implementing Your Endpoint In order to receive events from a Data Connector, your server needs to be set up to do the following: * Start listening for incoming requests on the URL specified in the Data Connector's Endpoint URL. * Read the HTTP headers and body, and process the event. * Reply with a **2XX** status code in a timely manner. Any response code outside the 2xx range will be considered a failed delivery and will be [retried](#retry-policy). If your endpoint fails to reply with status codes in the 2xx range consistently for an extended time period, the Data Connector will eventually be [automatically disabled](/data-connectors/advanced-configurations.md#auto-disabled-data-connectors). This server can be written in any programming language. Example implementations in a selection of languages can be found on the [Example Integrations](/data-connectors/example-integrations.md) page, as well as in the [Verifying Signed Events](#verifying-signed-events) section on this page. Regardless of which language is used to implement the server, there are a few things to keep in mind: #### Server Configuration The server needs to be set up to listen for HTTPS POST requests on the URL specified in the Data Connector's Endpoint URL. This endpoint needs to be publicly available and not require any authorization. You can verify that the event originates from DT by following the steps in the Verifying Signed Events section. The server also needs to be configured with a valid SSL certificate that is issued from one of the root certificates from [Mozilla's CA Certificate Program](https://wiki.mozilla.org/CA). This will be the case for most certificates (e.g. certificates issued by Let's Encrypt, DigiCert, GlobalSign, etc). Self-signed certificates are not supported. {% hint style="info" %} **SSL Certificate Verification** You can verify that you have a valid SSL cert by running `curl -v {YOUR_ENDPOINT}` in your terminal and look for the string "SSL certificate verify ok". To verify that your SSL cert is issued by one of the root certs in Mozilla's CA Certificate Program, you can spin up an Alpine docker container locally, and check your SSL cert from that container. Alpine uses the root certs in Mozilla's CA Certificate Program to validate SSL certificates. Run the following commands in your terminal (assuming Docker is installed): 1. `docker run --rm -ti alpine:latest sh` 2. `apk --update add ca-certificates curl` 3. `curl -v {YOUR_ENDPOINT}` 4. Make sure the "SSL certificate verify ok" string is present in the output 5. Use ctrl+d to quit the container To help diagnose any SSL issues, you can use the [SSL Server Test](https://www.ssllabs.com/ssltest/analyze.html) from SSL Labs. {% endhint %} #### Handle Incoming Event Each event will be delivered to your endpoint separately as individual requests. When processing a request, you should do the following steps: 1. (Recommended) Verify the integrity of the event, and that it originates from DT (see [Verifying Signed Events](#verifying-signed-events) below). 2. Do something with the event. You might for example put in on a separate queue for later processing (Google Pubsub, Amazon SQS, Azure ServiceBus, etc), write it to a database, or do some other processing. 3. Respond with a status code in the 2xx range once you've accepted/processed the event. These steps should all be done in a timely manner, and respond within 10 seconds. If your endpoint takes more than 10 seconds to respond, you run the risk that it will time out and be retried according to the [Retry Policy](#retry-policy). If you think there's a possibility that your processing might take more than 10 seconds, prefer to write the event to a separate queue or a database and do the processing at a later point in time. ## Handling Duplicates Every event received by DT Cloud is put in a [dedicated, per-Data Connector queue](/data-connectors/introduction-to-data-connector.md#at-least-once-guarantee). Messages are removed from this queue once acknowledged, or if the message is older than 12 hours. A side effect of this delivery guarantee is that, under certain conditions, **you may receive duplicates** of the same event. While rare, deduplication should be performed on the receiving end by checking event IDs. {% hint style="success" %} **Best Practice** Use the included **eventId** field to check for duplicated events. {% endhint %} ## Retry policy Any time a Data Connector does not receive a successful response (**HTTP status code 2xx**), the event will be retried. If an event has not been successfully acknowledged after 12 hours, it will be discarded. The retry interval is calculated as an exponential backoff policy, given by $$ t\_0\cdot2^{n-1}, $$ where $$t0$$ is the initial interval of 8 seconds and $$n$$ the attempt number. The interval will not exceed 1 hour. For very slow endpoints, the minimum retry interval will be $$4x$$ the response time. The following table shows the retry interval after a given number of delivery attempts:
AttemptRetry Interval [s]
AttemptRetry Interval [s]
18
216
332
......
92048
103600
113600
## Verifying Signed Events A signed Data Connector event lets you verify two things before you process it: that the request actually originated from DT Cloud, and that the body has not been modified in transit. In outline, both signing methods work the same way. A [JWT](https://jwt.io/) in an HTTP header contains a checksum of the request body. What differs is how the token is signed and verified. | | Asymmetric (recommended) | Symmetric | | ------------------------------------------------------ | ------------------------------------------------ | ------------------------------------------ | | Availability | Included on every event, no configuration needed | Only when a Signature Secret is configured | | Header | `DT-Asymmetric-Signature` | `X-Dt-Signature` | | Algorithm | `ES256` (ECDSA) | `HS256` (HMAC) | | Verified with | DT's public key, fetched from a JWKS endpoint | Your Signature Secret | | Shared secret to store and rotate | No | Yes | | Identifies the sending data-connector and organization | Yes, via the `sub` and `organization_id` claims | No | {% hint style="success" %} **No setup required** Every Data Connector signs every event asymmetrically. You do not need to configure anything in DT Studio or on the Data Connector resource. The `DT-Asymmetric-Signature` header is already there, and all that is left is to verify it at your endpoint. {% endhint %} DT signs each event with a private key that never leaves DT Cloud, and you verify it with the corresponding public key. Because the verifying key is public, there is no secret for your endpoint to store, no secret to leak, and no secret to rotate. We recommend asymmetric verification for all integrations. {% hint style="info" %} **Note** A Signature Secret remains available as an opt-in configuration, and existing integrations using one continue to work unchanged. Configuring a secret does not disable asymmetric signing. Events will carry both headers, and verifying either one is sufficient. See Symmetric verification with a Signature Secret below {% endhint %} ### **Asymmetric Verification** DT publishes its signing keys through an OpenID Connect discovery document: ``` https://identity.disruptive-technologies.com/data-connector/.well-known/openid-configuration ``` The `jwks_uri` field in that document points to the JSON Web Key Set containing the public keys. Most languages have an OIDC or JWKS library that handles discovery, key lookup, and caching for you. We recommend using a library rather than implementing the verification by hand. {% hint style="info" %} **Cache the key set** Fetching the discovery document and JWKS on every request adds latency to each event and may get you rate limited. Use a library that caches the key set, or cache it yourself. Do not hardcode a public key. Signing keys are rotated, and your verification will break when they are. Always resolve the key by the `kid` in the token header. {% endhint %} The token contains the following claims. | Claim | Type | Description | | ----------------- | -------- | ----------------------------------------------------------------------------------------------------------------------------- | | `iss` | `string` | The issuer, always `https://identity.disruptive-technologies.com/data-connector`. | | `sub` | `string` | The resource name of the Data Connector that sent the event, e.g. `projects//dataconnectors/`. | | `iat` | `number` | Unix timestamp for when the token was issued. | | `exp` | `number` | Unix timestamp for when the token expires. | | `organization_id` | `string` | The identifier of the DT organization the sending Data Connector belongs to. | | `checksum_sha256` | `string` | Hex-encoded SHA256 checksum of the request body. | The steps to verify a received request are as follows. 1. Extract the signed JWT from the **DT-Asymmetric-Signature** header of the received request. 2. Look up the signing key for the JWT token's `kid` header field in DT's JWKS, and verify the token's signature. Also verify the `iss`, `exp`, and `iat` claims. Data Connector events are always signed with `ES256`. Configure your verifier to accept that algorithm and no other. 3. Verify that the event came from a sender you expect, using either the `sub` or the `organization_id` claim. See the note below on choosing between them. 4. Calculate a **SHA256** checksum over the raw request body. 5. Compare that checksum with the `checksum_sha256` claim in the JWT. If the signature is valid and the checksums are identical, the event originated from a Data Connector in your organization and has not been tampered with. {% hint style="info" %} **Choosing between `sub` and `organization_id`** `sub` tells you which Data Connector sent the event. `organization_id` tells you which organization it belongs to. Check whichever fits your setup: * **One endpoint, one Data Connector.** Compare `sub` to that connector's resource name. This is the strictest check. * **Your own code creates the Data Connectors.** If you create them through the REST API, save each resource name and check `sub` against that list. Your code creates them, so the list stays up to date. Events from any other connector are rejected, including ones added by hand in DT Studio. * **You don't control which connectors send events.** Compare `organization_id` to your organization ID. Checking `sub` is stricter, but a list of connector names breaks when someone adds a connector you don't know about. Check at least one. A valid signature and checksum only prove the event came from DT, not that it came from a connector you set up. {% endhint %} {% hint style="warning" %} **Checksum the raw body, not a re-serialized copy** The checksum is calculated over the exact bytes DT sent. If your framework parses the JSON body for you, do not re-serialize the parsed object to compute the checksum — key ordering, whitespace, and number formatting may differ, and verification will fail. Read the raw body (for example `request.get_data()` in Flask, `express.raw()` in Express, or `io.ReadAll(r.Body)` in Go) and checksum that. {% endhint %} {% tabs %} {% tab title="Python 3.12" %} ```py # This Python script is built on Flask, docs are available here: # https://flask.palletsprojects.com/en/2.0.x/quickstart/ import os import hashlib from typing import Any import requests import jwt # pip install pyjwt==2.7.0 from flask import Flask, request # pip install Flask==2.3.2 app = Flask(__name__) # Read environment variable. # Refuse to start without an organization ID, rather than silently rejecting # every event. DT_ORGANIZATION_ID = os.environ["DT_ORGANIZATION_ID"] OIDC_SERVER = "https://identity.disruptive-technologies.com/data-connector" # The OpenID configuration and the signing keys are fetched once at startup # rather than per event. PyJWKClient caches the key set and refetches it when it # sees an unknown key ID, so key rotation is handled without a restart. oidc_config: dict[str, Any] = requests.get( f"{OIDC_SERVER}/.well-known/openid-configuration" ).json() jwks_client = jwt.PyJWKClient(oidc_config["jwks_uri"], cache_jwk_set=True, lifespan=360) @app.route("/", methods=["POST"]) def data_connector_endpoint() -> tuple[str, int]: # Extract the body as a bytestring and the signed JWT. # We'll use these values to verify the request. payload = request.get_data() token = request.headers.get("dt-asymmetric-signature") if token is None: return ("Missing dt-asymmetric-signature header.", 401) # Verify request origin and content integrity. if not verify_request(payload, token): return ("Could not verify request.", 401) # We now know the request came from DT Cloud, and the integrity # of the body has been verified. We can now handle the event safely. # Any error raised while handling the event is answered with a 500 so that # the event is retried rather than silently dropped. try: handle_event(request.get_json()) except Exception as e: print(f"Failed to handle event: {e}") return ("Failed to handle event.", 500) # Respond with a 200 status code to ack the event. Any status codes # that are outside the 2xx range will nack the event, meaning it will # be retried later. return ("OK", 200) def verify_request(body: bytes, token: str) -> bool: """ Verifies that the request originated from DT, and that the body hasn't been modified since it was sent. This is done by verifying the JWT signature against DT's public key, checking that the token was issued for the expected organization, and comparing the checksum claim against a checksum of the request body. Every failure path returns False rather than raising, so that a malformed request produces a 400 instead of an unhandled exception and a 500. """ # Decode the JWT and verify the signature using DT's public key, # found in the jwks_uri from the OpenID configuration. The signing # algorithm is pinned to ES256 rather than read from the discovery # document, which advertises additional algorithms used by other # DT integrations. try: signing_key = jwks_client.get_signing_key_from_jwt(token) claims = jwt.decode( token, signing_key.key, algorithms=["ES256"], issuer=oidc_config["issuer"], options={ "verify_signature": True, "verify_iss": True, "verify_exp": True, "verify_iat": True, }, ) except Exception as e: print(f"Failed to verify token: {e}") return False # Verify that the event was sent from the expected organization. if claims.get("organization_id") != DT_ORGANIZATION_ID: print(f"Organization mismatch: {claims.get('organization_id')} != {DT_ORGANIZATION_ID}") return False # Verify the request body checksum. m = hashlib.sha256() m.update(body) checksum = m.digest().hex() if claims.get("checksum_sha256") != checksum: print(f"Checksum mismatch: {claims.get('checksum_sha256')} != {checksum}") return False return True def handle_event(body: dict[str, Any]) -> None: """ Processes the event itself. For this example, we will just decode a touch event, and print out the timestamp, device ID, and the device type. Other event types, such as the networkStatus and batteryStatus events a device sends on its own schedule, fall through and are acked without being handled. """ # First, check if the event type is one of the event # types we're expecting. # As an example, we'll check for touch events here. if body["event"]["eventType"] == "touch": # Now that we know this is a device event, we can # check for the device type and device identifier # in the event metadata. device_type = body["metadata"]["deviceType"] device_id = body["metadata"]["deviceId"] timestamp = body["event"]["data"]["touch"]["updateTime"] print( "Got touch event at {} from {} sensor with id {}".format( timestamp, device_type, device_id, ) ) ``` {% endtab %} {% tab title="Node.js 20" %} ```javascript const crypto = require('crypto') const jwt = require('jsonwebtoken') // npm install jsonwebtoken@9 const { Issuer } = require('openid-client') // npm install openid-client@3 const jwksClient = require('jwks-rsa') // npm install jwks-rsa@2 const express = require('express') // npm install express@4 // Read environment variable const dtOrganizationId = process.env.DT_ORGANIZATION_ID const dtOidcIssuer = "https://identity.disruptive-technologies.com/data-connector" // The discovery document and the key set are fetched once at startup, not per // event. The jwks client caches the keys and refetches them when it sees an // unknown key ID, so key rotation is handled without a restart. let keys const initKeyClient = async () => { const issuer = await Issuer.discover(dtOidcIssuer) keys = jwksClient({ jwksUri: issuer.jwks_uri, cache: true, }) } // dataConnectorEndpoint receives, validates, and returns a response // for the forwarded event. const dataConnectorEndpoint = async (req, res) => { // req.body is a Buffer holding the raw request bytes, see express.raw() // below. It is only a Buffer if the Content-Type matched, so check before // checksumming it. if (!Buffer.isBuffer(req.body)) { console.log('Expected a JSON body') res.sendStatus(400) return } // Validate request origin and content integrity. const token = req.headers['dt-asymmetric-signature'] const verified = await verifyRequest(req.body, token) if (verified === false) { res.sendStatus(401) return } // We now know the request came from DT Cloud, and the integrity // of the body has been verified. We can now handle the event safely. // Any error thrown while handling the event is answered with a 500 so that // the event is retried rather than silently dropped. try { handleEvent(JSON.parse(req.body.toString())) } catch (err) { console.log(err) res.sendStatus(500) return } // Respond with a 200 status code to ack the event. Any status codes // that are outside the 2xx range will nack the event, meaning it will // be retried later. res.sendStatus(200) } // Verifies that the request originated from DT, and that the body // hasn't been modified since it was sent. This is done by verifying the // JWT signature against DT's public key, checking that the token was // issued for the expected organization, and comparing the checksum claim // against a checksum of the raw request body. // // Every failure returns false rather than throwing: an unhandled rejection in // an async express handler terminates the process, so one malformed request // would otherwise take the receiver down. const verifyRequest = async (payload, token) => { try { if (typeof token !== 'string' || token === '') { console.log('Missing dt-asymmetric-signature header') return false } // Decode the JWT header and look up the public key by its key ID. // jwt.decode returns null for a token it cannot parse. const decoded = jwt.decode(token, { complete: true }) if (decoded === null || !decoded.header.kid) { console.log('Malformed token, or token without a key ID') return false } const key = await keys.getSigningKey(decoded.header.kid) const signingKey = key.getPublicKey() // Verify the signature, the signing algorithm, and the issuer. The // algorithm is pinned to ES256 rather than read from the discovery // document, which advertises additional algorithms used by other DT // integrations. const claims = jwt.verify(token, signingKey, { algorithms: ["ES256"], issuer: dtOidcIssuer, }) // Verify that the event was sent from the expected organization. if (claims.organization_id !== dtOrganizationId) { console.log('Organization Mismatch') return false } // Verify the request body checksum. const hash = crypto.createHash("sha256") const checksum = hash.update(payload).digest("hex") if (checksum !== claims.checksum_sha256) { console.log('Checksum Mismatch') return false } return true } catch (err) { console.log(`Could not verify request: ${err.message}`) return false } } // handleEvent processes the event itself. For this example, // we will just decode a touch event, and print out the timestamp, // device ID, and the device type. Other event types, such as the // networkStatus and batteryStatus events a device sends on its own // schedule, fall through and are acked without being handled. const handleEvent = (payload) => { // First, check if the event type is one of the event // types we're expecting. // As an example, we'll check for touch events here. switch (payload.event.eventType) { case 'touch': // Now that we know this is a device event, we can // check for the device type and device identifier // in the event metadata. const deviceType = payload.metadata.deviceType const deviceId = payload.metadata.deviceId const timestamp = payload.event.data.touch.updateTime console.log(`Received touch event at ${timestamp} from ${deviceType} sensor with id ${deviceId}`) break default: break } } // Sets up a bare-bones server that listens on port 8080, and routes // all requests to the path "/" to the `dataConnectorEndpoint` function. // express.raw() is used instead of express.json() so that the checksum // can be calculated over the exact bytes that DT sent. const app = express() app.use(express.raw({ type: 'application/json' })) app.post('/', dataConnectorEndpoint) // Fetch the signing keys before accepting events, so that the first event // isn't rejected while discovery is still in flight. initKeyClient() .then(() => app.listen(8080, () => console.log('Listening on :8080 ...'))) .catch((err) => { console.error(`Failed to fetch DT signing keys: ${err.message}`) process.exit(1) }) ``` {% endtab %} {% tab title="Go 1.26" %} ```go package main import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "io" "net/http" "os" // go get github.com/coreos/go-oidc/v3/oidc oidc "github.com/coreos/go-oidc/v3/oidc" ) // Read environment variable var dtOrganizationID = os.Getenv("DT_ORGANIZATION_ID") const dtOIDCIssuer = "https://identity.disruptive-technologies.com/data-connector" // The verifier is built once at startup, not per event. It holds a remote key // set that caches DT's public keys and refetches them when it sees an unknown // key ID, so key rotation is handled without a restart. var verifier *oidc.IDTokenVerifier func newVerifier(ctx context.Context) (*oidc.IDTokenVerifier, error) { // Set up an OIDC provider with the correct issuer. This fetches the // discovery document, so it needs the identity server to be reachable. provider, err := oidc.NewProvider(ctx, dtOIDCIssuer) if err != nil { return nil, err } // Create a verifier with ES256 as the supported signing algorithm. // SkipClientIDCheck is set to true, as we have not supplied a client // ID (this is not an OIDC login flow). return provider.Verifier( &oidc.Config{ SupportedSigningAlgs: []string{"ES256"}, SkipClientIDCheck: true, }, ), nil } // DataConnectorEndpoint receives, validates, and returns a response // for the forwarded event. func DataConnectorEndpoint(w http.ResponseWriter, r *http.Request) { // Extract the body and the signed JWT. // We'll use these values to verify the request. tokenString := r.Header.Get("DT-Asymmetric-Signature") bodyBytes, err := io.ReadAll(r.Body) if err != nil { fmt.Println(err) http.Error(w, "could not read request body", http.StatusBadRequest) return } // Validate request origin and content integrity. The reason is logged but // not returned to the caller, so that verification details aren't echoed // back to whoever sent the request. if err := verifyRequest(r.Context(), bodyBytes, tokenString); err != nil { fmt.Printf("could not verify request: %v\n", err) http.Error(w, "could not verify request", http.StatusUnauthorized) return } // We now know the request came from DT Cloud, and the integrity // of the body has been verified. We can now handle the event safely. // A failure to handle the event is answered with a 500 so that the event // is retried rather than silently dropped. if err := handleEvent(bodyBytes); err != nil { fmt.Printf("could not handle event: %v\n", err) http.Error(w, "could not handle event", http.StatusInternalServerError) return } // Respond with a 200 status code to ack the event. Any status codes // that are outside the 2xx range will nack the event, meaning it // will be retried later. w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("OK")) } // verifyRequest verifies that the request originated from your organization // in DT and that the body hasn't been modified since it was sent. This is // done by verifying the JWT against DT's public key, checking the organization_id // claim against the expected organization, and comparing the checksum claim // against a checksum of the request body. func verifyRequest(ctx context.Context, bodyBytes []byte, tokenString string) error { // Verify the JWT. This checks the signature of the token against the // public key from the data-connector discovery endpoint. token, err := verifier.Verify(ctx, tokenString) if err != nil { return err } // Parse the claims from the token. Claims are the payload of the JWT // and cannot be altered without invalidating the signature. claims := &Claims{} if err := token.Claims(claims); err != nil { return err } // Verify that the event was sent from the expected organization. if claims.OrganizationID != dtOrganizationID { return fmt.Errorf("organization mismatch") } // Verify the request body checksum. sha256Bytes := sha256.Sum256(bodyBytes) sha256String := hex.EncodeToString(sha256Bytes[:]) if sha256String != claims.ChecksumSHA256 { return fmt.Errorf("checksum mismatch") } return nil } type Claims struct { // JWT standard claims Issuer string `json:"iss"` Subject string `json:"sub"` Expiry int64 `json:"exp"` IssuedAt int64 `json:"iat"` // DT claims OrganizationID string `json:"organization_id"` ChecksumSHA256 string `json:"checksum_sha256"` } // handleEvent processes the event itself. For this example, // we will just decode a touch event, and print out the timestamp, // device ID, and the device type. Other event types, such as the // networkStatus and batteryStatus events a device sends on its own // schedule, fall through and are acked without being handled. func handleEvent(payload []byte) error { // The structure of the events we'll receive from a Data Connector. type Event struct { Event struct { EventId string `json:"eventId"` EventType string `json:"eventType"` Data json.RawMessage `json:"data"` Timestamp string `json:"timestamp"` } `json:"event"` Labels map[string]string `json:"labels"` Metadata map[string]string `json:"metadata"` } // The structure of the `Event.Data` field for a touch event. // We'll be using touch events for this example. type TouchData struct { Touch struct { Timestamp string `json:"updateTime"` } `json:"touch"` } // Decode the event var event Event if err := json.Unmarshal(payload, &event); err != nil { return err } // First, check if the event type is one of the event // types we're expecting. // As an example, we'll check for touch events here. switch event.Event.EventType { case "touch": // Now that we know this is a touch event, we can decode // the `Event.Data` field. var touchData TouchData if err := json.Unmarshal(event.Event.Data, &touchData); err != nil { return err } // Also, since we now know this is a device event, we can // check for the device type and device identifier // in the event metadata. deviceType := event.Metadata["deviceType"] deviceId := event.Metadata["deviceId"] timestamp := touchData.Touch.Timestamp fmt.Printf("Received touch event at %s from %s sensor with id %s\n", timestamp, deviceType, deviceId, ) } return nil } func main() { // Fetch DT's signing keys before accepting events, so that the first // event isn't rejected while discovery is still in flight. v, err := newVerifier(context.Background()) if err != nil { fmt.Printf("Failed to fetch DT signing keys: %v\n", err) os.Exit(1) } verifier = v // Sets up a bare-bones server that listens on port 8080, and // routes all requests to the path "/" to the // `DataConnectorEndpoint` function. http.HandleFunc("/", DataConnectorEndpoint) fmt.Println("Started listening on localhost:8080 ...") if err := http.ListenAndServe(":8080", nil); err != nil { fmt.Printf("Closed with error: %v\n", err) } else { fmt.Println("Closed server successfully") } } ``` {% endtab %} {% endtabs %} ### Symmetric Verification with Signature Secret {% hint style="info" %} **Note** New integrations should prefer asymmetric verification instead. This method is documented for existing integrations. {% endhint %} The following steps sum up the process of verifying the received request at the receiving endpoint. 1. Extract the signed JWT from the HTTP header **X-Dt-Signature** of the received request. 2. Verify the JWT's `HMAC` signature with the **signature secret.** 3. Calculate a **SHA256** checksum over the entire request body. 4. Compare the body checksum with the `checksum_sha256` field contained in the JWT (there's a `checksum` field as well that uses SHA1 which is less secure than SHA256, and is kept only for backward compatibility). 5. If these checksums are identical, you can be certain that the event has not been tampered with and originated from your Data Connector. The following snippet from our [Google Cloud Function example integration](/data-connectors/example-integrations/google-cloud-functions.md) implements this verification process. {% tabs %} {% tab title="Python 3.12" %} ```python # This Python script is built on Flask, docs are available here: # https://flask.palletsprojects.com/en/2.0.x/quickstart/ import os import hashlib from typing import Any import jwt # pip install pyjwt==2.7.0 from flask import Flask, request # pip install Flask==2.3.2 app = Flask(__name__) # Read environment variable. # Refuse to start without a secret, rather than failing on every event. An # unset secret must never be allowed to act as the HMAC key. SIGNATURE_SECRET = os.environ['DT_SIGNATURE_SECRET'] @app.route('/', methods=["POST"]) def data_connector_endpoint() -> tuple[str, int]: # Extract the body as a bytestring and the signed JWT. # We'll use these values to verify the request. payload = request.get_data() token = request.headers.get('x-dt-signature') if token is None: return ('Missing x-dt-signature header.', 400) # Verify request origin and content integrity. if not verify_request(payload, token): return ('Could not verify request.', 401) # We now know the request came from DT Cloud, and the integrity # of the body has been verified. We can now handle the event safely. # Any error raised while handling the event is answered with a 500 so that # the event is retried rather than silently dropped. try: handle_event(request.get_json()) except Exception as e: print(f"Failed to handle event: {e}") return ('Failed to handle event.', 500) # Respond with a 200 status code to ack the event. Any status codes # that are outside the 2xx range will nack the event, meaning it will # be retried later. return ('OK', 200) def verify_request(body: bytes, token: str) -> bool: """ Verifies that the request originated from DT, and that the body hasn't been modified since it was sent. This is done by verifying that the checksum field of the JWT token matches the checksum of the request body, and that the JWT is signed with the signature secret. Every failure path returns False rather than raising, so that a malformed request produces a 400 instead of an unhandled exception and a 500. """ # Decode the JWT, and verify that it was signed using the # signature secret. Also verifies that the algorithm used was HS256. try: claims = jwt.decode(token, SIGNATURE_SECRET, algorithms=["HS256"]) except Exception as err: print(err) return False # Verify the request body checksum. m = hashlib.sha256() m.update(body) checksum = m.digest().hex() if claims.get("checksum_sha256") != checksum: print('Checksum Mismatch') return False return True def handle_event(body: dict[str, Any]) -> None: """ Processes the event itself. For this example, we will just decode a touch event, and print out the timestamp, device ID, and the device type. Other event types, such as the networkStatus and batteryStatus events a device sends on its own schedule, fall through and are acked without being handled. """ # First, check if the event type is one of the event # types we're expecting. # As an example, we'll check for touch events here. if body['event']['eventType'] == 'touch': # Now that we know this is a device event, we can # check for the device type and device identifier # in the event metadata. device_type = body['metadata']['deviceType'] device_id = body['metadata']['deviceId'] timestamp = body['event']['data']['touch']['updateTime'] print("Got touch event at {} from {} sensor with id {}".format( timestamp, device_type, device_id, )) ``` {% endtab %} {% tab title="Node.js 20" %} ```javascript const crypto = require('crypto') const express = require('express') // npm install express@4 const jwt = require('jsonwebtoken') // npm install jsonwebtoken@9 // Read environment variable const signatureSecret = process.env.DT_SIGNATURE_SECRET // dataConnectorEndpoint receives, validates, and returns a response // for the forwarded event. const dataConnectorEndpoint = (req, res) => { // req.body is a Buffer holding the raw request bytes, see express.raw() // below. It is only a Buffer if the Content-Type matched, so check before // checksumming it. if (!Buffer.isBuffer(req.body)) { console.log('Expected a JSON body') res.sendStatus(400) return } // Validate request origin and content integrity. const token = req.headers['x-dt-signature'] if (verifyRequest(req.body, token) === false) { res.sendStatus(401) return } // We now know the request came from DT Cloud, and the integrity // of the body has been verified. We can now handle the event safely. // Any error thrown while handling the event is answered with a 500 so that // the event is retried rather than silently dropped. try { handleEvent(JSON.parse(req.body.toString())) } catch (err) { console.log(err) res.sendStatus(500) return } // Respond with a 200 status code to ack the event. Any status codes // that are outside the 2xx range will nack the event, meaning it will // be retried later. res.sendStatus(200) } // Verifies that the request originated from DT, and that the body // hasn't been modified since it was sent. This is done by verifying // that the checksum field of the JWT token matches the checksum of the // raw request body, and that the JWT is signed with the signature secret. const verifyRequest = (payload, token) => { if (typeof token !== 'string' || token === '') { console.log('Missing x-dt-signature header') return false } // Decode the JWT, and verify that it was signed using the // signature secret. The permitted algorithms are given as an array, so // that the allow-list is matched exactly rather than by substring. let decoded try { decoded = jwt.verify(token, signatureSecret, { algorithms: ["HS256"] }) } catch (err) { console.log(err.message) return false } // Verify the request body checksum. const hash = crypto.createHash("sha256") const checksum = hash.update(payload).digest("hex") if (checksum !== decoded.checksum_sha256) { console.log('Checksum Mismatch') return false } return true } // handleEvent processes the event itself. For this example, // we will just decode a touch event, and print out the timestamp, // device ID, and the device type. Other event types, such as the // networkStatus and batteryStatus events a device sends on its own // schedule, fall through and are acked without being handled. const handleEvent = (payload) => { // First, check if the event type is one of the event // types we're expecting. // As an example, we'll check for touch events here. switch (payload.event.eventType) { case 'touch': // Now that we know this is a device event, we can // check for the device type and device identifier // in the event metadata. const deviceType = payload.metadata.deviceType const deviceId = payload.metadata.deviceId const timestamp = payload.event.data.touch.updateTime console.log(`Received touch event at ${timestamp} from ${deviceType} sensor with id ${deviceId}`) break default: break } } // Refuse to start without a secret. An unset secret must never be allowed to // act as the HMAC key. if (!signatureSecret) { console.error('DT_SIGNATURE_SECRET is not set') process.exit(1) } // Sets up a bare-bones server that listens on port 8080, and routes // all requests to the path "/" to the `dataConnectorEndpoint` function. // express.raw() is used instead of express.json() so that the checksum // can be calculated over the exact bytes that DT sent. const app = express() app.use(express.raw({ type: 'application/json' })) app.post('/', dataConnectorEndpoint) app.listen(8080, () => console.log('Listening on :8080 ...')) ``` {% endtab %} {% tab title="Go 1.26" %} ```go package main import ( "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "io" "net/http" "os" // go get github.com/golang-jwt/jwt/v5@v5.0.0 jwt "github.com/golang-jwt/jwt/v5" ) // Read environment variable var signatureSecret = os.Getenv("DT_SIGNATURE_SECRET") // DataConnectorEndpoint receives, validates, and returns a response // for the forwarded event. func DataConnectorEndpoint(w http.ResponseWriter, r *http.Request) { // Extract the body and the signed JWT. // We'll use these values to verify the request. tokenString := r.Header.Get("x-dt-signature") bodyBytes, err := io.ReadAll(r.Body) if err != nil { fmt.Println(err) http.Error(w, "could not read request body", http.StatusBadRequest) return } // Validate request origin and content integrity. The reason is logged but // not returned to the caller, so that verification details aren't echoed // back to whoever sent the request. if err := verifyRequest(bodyBytes, tokenString); err != nil { fmt.Printf("could not verify request: %v\n", err) http.Error(w, "could not verify request", http.StatusUnauthorized) return } // We now know the request came from DT Cloud, and the integrity // of the body has been verified. We can now handle the event safely. // A failure to handle the event is answered with a 500 so that the event // is retried rather than silently dropped. if err := handleEvent(bodyBytes); err != nil { fmt.Printf("could not handle event: %v\n", err) http.Error(w, "could not handle event", http.StatusInternalServerError) return } // Respond with a 200 status code to ack the event. Any status codes // that are outside the 2xx range will nack the event, meaning it // will be retried later. w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("OK")) } // verifyRequest verifies that the request originated from DT, and that // the body hasn't been modified since it was sent. This is done by // verifying that the checksum field of the JWT token matches the checksum // of the request body, and that the JWT is signed with the signature secret. func verifyRequest(bodyBytes []byte, tokenString string) error { // Decode the JWT, and verify that it was signed using the // signature secret. Also verifies the algorithm used to sign the JWT. token, err := jwt.Parse( tokenString, func(token *jwt.Token) (interface{}, error) { // Return our signature secret to verify that it was used to // sign the JWT. return []byte(signatureSecret), nil }, jwt.WithValidMethods([]string{"HS256"}), ) if err != nil { return err } // Read the checksum claim. The type assertion is checked, so an // unexpected token shape returns an error instead of panicking. claims, ok := token.Claims.(jwt.MapClaims) if !ok { return fmt.Errorf("unexpected claims type %T", token.Claims) } claimedChecksum, ok := claims["checksum_sha256"].(string) if !ok { return fmt.Errorf("token has no checksum_sha256 claim") } // Verify the request body checksum. sha256Bytes := sha256.Sum256(bodyBytes) sha256String := hex.EncodeToString(sha256Bytes[:]) if sha256String != claimedChecksum { return fmt.Errorf("checksum mismatch") } return nil } // handleEvent processes the event itself. For this example, // we will just decode a touch event, and print out the timestamp, // device ID, and the device type. Other event types, such as the // networkStatus and batteryStatus events a device sends on its own // schedule, fall through and are acked without being handled. func handleEvent(payload []byte) error { // The structure of the events we'll receive from a Data Connector. type Event struct { Event struct { EventId string `json:"eventId"` EventType string `json:"eventType"` Data json.RawMessage `json:"data"` Timestamp string `json:"timestamp"` } `json:"event"` Labels map[string]string `json:"labels"` Metadata map[string]string `json:"metadata"` } // The structure of the `Event.Data` field for a touch event. // We'll be using touch events for this example. type TouchData struct { Touch struct { Timestamp string `json:"updateTime"` } `json:"touch"` } // Decode the event var event Event if err := json.Unmarshal(payload, &event); err != nil { return err } // First, check if the event type is one of the event // types we're expecting. // As an example, we'll check for touch events here. switch event.Event.EventType { case "touch": // Now that we know this is a touch event, we can decode // the `Event.Data` field. var touchData TouchData if err := json.Unmarshal(event.Event.Data, &touchData); err != nil { return err } // Also, since we now know this is a device event, we can // check for the device type and device identifier // in the event metadata. deviceType := event.Metadata["deviceType"] deviceId := event.Metadata["deviceId"] timestamp := touchData.Touch.Timestamp fmt.Printf("Received touch event at %s from %s sensor with id %s\n", timestamp, deviceType, deviceId, ) } return nil } func main() { // Refuse to start without a secret. An empty secret would otherwise be // used as the HMAC key, and anyone could sign a token that verifies. if signatureSecret == "" { fmt.Println("DT_SIGNATURE_SECRET is not set") os.Exit(1) } // Sets up a bare-bones server that listens on port 8080, and // routes all requests to the path "/" to the // `DataConnectorEndpoint` function. http.HandleFunc("/", DataConnectorEndpoint) fmt.Println("Started listening on localhost:8080 ...") if err := http.ListenAndServe(":8080", nil); err != nil { fmt.Printf("Closed with error: %v\n", err) } else { fmt.Println("Closed server successfully") } } ``` {% endtab %} {% endtabs %}