MQTT Device Authentication¶

TL;DR — the credential the ThingsBoard UI shows does NOT work for MQTT. When you create a device in the UI it generates an ACCESS_TOKEN. But this platform authenticates MQTT (and the primary HTTP ingest) with a short-lived device JWT. To connect a device over MQTT you must mint a JWT and use it as the MQTT username. This diverges from stock ThingsBoard on purpose (JWTs are signed and short-lived, so a leaked credential expires in minutes).

Why the UI access token is not the MQTT credential¶

rmqtt (the MQTT broker) runs the rmqtt-auth-jwt plugin. It:

reads the JWT from the MQTT username (from = "username"),
verifies the ES256 signature against flow-core's device public key (/api/noauth/device-jwt-public.pem; the broker mounts the same key),
requires claims iss = thingsflow-device, aud = thingsflow-mqtt, and a valid exp,
rejects anonymous connections (allow_anonymous = false).

The ACCESS_TOKEN the UI generates is stored in device_credentials, but it is not accepted by the MQTT broker and is not an ingest credential at all: the classic POST /api/v1/{accessToken}/telemetry path returns 503 here. It survives for GET /api/v1/{accessToken}/attributes and as a TB-compatible identifier.

Credential	Generated by	Used for
Device JWT (ES256, short TTL)	provisioning / `POST /api/device/{id}/jwt` / refresh	MQTT (username) and HTTP `POST /api/v1/telemetry` (Bearer)
ACCESS_TOKEN (20-char)	UI device create; provisioning (returned alongside the JWT)	TB-compatible identifier; `GET /api/v1/{accessToken}/attributes`. Not valid for telemetry ingest.

Onboarding from the ThingsBoard UI (recommended)¶

The TB-native self-provisioning flow works end to end:

Device profile → "Device provisioning" tab: strategy Allow to create new devices, set a provision device key + secret, save. (The UI stores the secret nested in profileData.provisionConfiguration; flow-core reads it there. Reopening the profile shows the saved strategy/key — the secret is masked, only its bcrypt hash is stored.)
The device self-provisions: POST /api/v1/provision with {deviceName, provisionDeviceKey, provisionDeviceSecret} → returns {status:"SUCCESS", credentialsType:"ACCESS_TOKEN", credentialsValue, deviceId, tenantId, deviceJwt}. deviceJwt is the credential that actually authenticates ingest.
Device → "Check connectivity" in the UI shows ready-to-run curl (HTTPS) and mosquitto_pub (MQTT-over-TLS) commands for that device, already carrying a freshly minted device JWT, the pinned MQTT Client ID and a ts field. Copy-paste and it works.

Onboarding a fleet (CSV)¶

Devices → Import accepts a CSV with NAME, TYPE, LABEL, ACCESS_TOKEN, DESCRIPTION, IS_GATEWAY columns (update: true re-imports as updates). Attribute / timeseries column types are not imported and are reported as a warning rather than dropped silently. Imported devices are registered; each still obtains its ingest credential (a device JWT) via provisioning or POST /api/device/{id}/jwt.

The Add device wizard (/api/device-with-credentials) also works, including setting a custom access token at creation time.

How to get a device JWT¶

For a device you already created in the UI¶

CORE=https://<your-thingsflow-host>
TOKEN=$(curl -s -X POST "$CORE/api/auth/login" \
  -H 'Content-Type: application/json' \
  -d '{"username":"tenant@thingsboard.org","password":"<your password>"}' \
  | python3 -c 'import json,sys; print(json.load(sys.stdin)["token"])')
curl -s -X POST "$CORE/api/device/<deviceId>/jwt" \
  -H "X-Authorization: Bearer $TOKEN"

The tenant-authenticated endpoint POST /api/device/{id}/jwt returns { "token": "<device JWT>", "mqttId": "...", ... } — the mqttId is the MQTT Client ID to use below.

For a brand-new device (auto-provisioning)¶

POST /api/v1/provision with a provisionDeviceKey + provisionDeviceSecret returns both an ACCESS_TOKEN and a device JWT (token). See the provisioning section of Device SDK for a scripted client.

Connecting an MQTT client¶

Endpoint: <your-mqtt-host>:8883 (TLS). The host is deployment-specific: it is what the operator set in flowCore.deviceConnectivity.mqttHost, and the per-device connection snippets from GET /api/device-connectivity/{id} already contain it.
TLS: required on any publicly exposed edge. The chart's TLS listener is opt-in (rmqttEdge.tls with a cert-manager Certificate; the reference deployment uses a Let's Encrypt cert, so clients verify against the public CA with no custom CA). Port 8883 is the IANA secure-mqtt port; connecting plain there fails the TLS handshake. The chart default without rmqttEdge.tls is a plaintext 1883 listener on an internal ClusterIP — acceptable only inside a trusted network.
Username: the device JWT. Password: empty.
Client ID (required): must equal the JWT's clientid/mqttId claim (the hyphen-stripped device id). rmqtt rejects a connection whose Client ID differs (validate_claims.clientid), and the ACL binds the topic to it (%c), so a device can publish only to its own thingsflow/devices/<mqttId>/... — never another device's topic. The /jwt response above includes the mqttId to use.
Publish topic: thingsflow/devices/<mqttId>/telemetry (or /attributes), where <mqttId> is your own Client ID.
ts is optional but recommended (epoch milliseconds). A payload without one is stamped with server receipt time, matching stock ThingsBoard, and the thingsflow_materializer_ts_stamped counter records how often that happens. Send your own ts when the device can: it is the difference between the time the reading was taken and the time it arrived, which diverge whenever a device buffers through a connectivity gap.

Set INGEST_REQUIRE_TS=true on the materializer to reject ts-less payloads instead. That restores strict determinism — a redelivered message keeps its original ts_ns and upserts in place rather than writing a second row — at the cost of dropping every reading from a device that does not send a timestamp.

mosquitto_pub -h <your-mqtt-host> -p 8883 --capath /etc/ssl/certs \
  -u "$DEVICE_JWT" -t "thingsflow/devices/<mqttId>/telemetry" \
  -m "{\"ts\":$(date +%s000),\"temperature\":21.5}"

Verified end-to-end over TLS: handshake (LE cert verified) → JWT auth (CONNACK 0) → rmqtt → NATS → materializer → GreptimeDB. device_id is bound to the JWT's deviceId claim, so the topic/payload cannot impersonate another device.

The written device_id is bound to the JWT's deviceId claim (a device can only write as itself — see the anti-spoofing note in Security Architecture), so the topic/payload cannot be used to impersonate another device.

Token lifetime & refresh¶

Device JWTs are short-lived bearer credentials (DEVICE_JWT_TTL_SECONDS; the chart default is 86400s, and hardened overlays shorten it — the reference deployment runs 900s / 15 minutes). A long-running client must refresh before expiry:

POST /api/v1/devices/me/jwt/refresh      # authenticated with the current JWT

Long-running clients should refresh at ~2/3 of the TTL. Treat the JWT as a rotating credential, not a permanent secret; the short TTL is the revocation window (there is no per-token denylist on the ingest path).

Rotating the device JWT signing key¶

flow-core signs device JWTs with an ES256 private key; the broker verifies them with the matching public key, which it reads once, at process start. rmqtt's plugin config-reload API answers true for rmqtt-auth-jwt but does not re-read the key file (verified against rmqtt 0.20.0), so a rotation only takes effect when the broker process restarts.

The chart handles that restart. helm upgrade rolls rmqtt automatically when the signing key changes, and its init container waits for the expected key before starting the broker. Rotation is therefore:

# 1. new key, and bump the key id so the wait below is exact
helm upgrade thingsflow k8s/helm/thingsflow -f <values> \
  --set flowCore.deviceJwt.privateKeyPemBase64="$(base64 -w0 new-key.pem)" \
  --set flowCore.deviceJwt.keyId="thingsflow-device-es256-2"

Nothing else is required — no manual broker restart. Verified end to end: after the upgrade the key flow-core serves and the key on the broker's disk hash identically, and a device connects with a freshly issued JWT.

Rotating without stranding devices¶

A one-shot swap has a gap. A device still holding a token signed by the retired key cannot connect (the broker already has the new public key) and cannot refresh — POST /api/v1/devices/me/jwt/refresh authenticates with the very token being rejected. Recovering it needs an out-of-band re-issue, which for a meter in the field means a site visit.

flow-core verifies against the active key plus every key in flowCore.deviceJwt.additionalPublicJwksBase64. Signing always uses the active key; the additional set is verify-only. That asymmetry is what makes a staged rotation possible:

Step	`privateKeyPemBase64` / `keyId`	`additionalPublicJwksBase64`	Effect
1	old	new public key	Both verify. Nothing signs with the new key yet.
2	new	old public key	New tokens are issued; devices holding old ones still verify, so they can refresh normally.
3	new	(empty)	Old key retired. Run this only after one full token TTL has passed.

Each step is a helm upgrade. Step 2 restarts the broker (it verifies against a single key and reads it at start), so devices reconnect — but they reconnect successfully, because their existing token still verifies long enough to refresh. Step 3 is the one that must wait: run it before DEVICE_JWT_TTL_SECONDS has elapsed and you strand whatever has not refreshed yet.

A token naming a key id that is neither active nor in the additional set is rejected outright — it does not fall back to the active key, which would make a stale or attacker-chosen kid indistinguishable from a valid token.

Bump keyId when you rotate. The init container waits for that id to appear in /api/noauth/device-jwks before taking the key. Without a new id it falls back to requiring the served key to be stable across two reads, which is weaker: flow-core and rmqtt roll concurrently, so the broker can otherwise fetch the retired key from a flow-core replica that has not been replaced yet.

Why this is worth care. A broker left on the retired key hides its own breakage: MQTT authenticates only at CONNECT, so sessions that are already open keep publishing and telemetry keeps flowing, while every device that reconnects is refused. The platform looks healthy and the fleet drains away slowly.

If the key is left unset, flow-core generates an ephemeral one per process and every restart has the same effect. It refuses to start that way in FLOW_ENV=production and warns loudly elsewhere.

Server-to-device RPC¶

RPC is opt-in (flowCore.rpc.enabled, off by default). For an energy fleet a command can mean opening or closing a breaker, so the ability to push one is a capability an operator turns on deliberately, not a default posture.

Topics — platform-native, not ThingsBoard's v1/devices/me/rpc/.... The broker ACL binds every device topic to %c (the connection's Client ID, pinned to the JWT's clientid claim); a literal me segment cannot be bound that way, so TB's spelling would cost the broker-level anti-spoofing this platform relies on.

Direction	Topic
Device subscribes	`thingsflow/devices/<mqttId>/rpc/request/+`
Device publishes reply	`thingsflow/devices/<mqttId>/rpc/response/<requestId>`

The request payload is {"id": "<requestId>", "method": "...", "params": {...}}. <requestId> is also the last topic segment, so a device can reply without parsing the body. A device may subscribe only to its own request topic and publish only on its own response topic — the ACL rejects anything else.

Calling it (ThingsBoard's contract, unchanged):

curl -X POST "$BASE/api/rpc/twoway/$DEVICE_ID" \
  -H "X-Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
  -d '{"method":"getStatus","params":{},"timeout":5000}'

/api/rpc/oneway/{id} returns an empty 200 as soon as the broker accepts the command. /api/rpc/twoway/{id} returns the device's reply verbatim, or:

Status	Meaning
`504`	Device is not connected, or did not answer within the timeout
`503`	Reply transport (NATS) is down — one-way RPC still works
`501`	RPC is disabled (`flowCore.rpc.enabled=false`)
`403`	The device belongs to another tenant

timeout is capped at 60s. persistent is accepted and ignored: there is no store-and-forward queue here, so a command to an offline device fails rather than being delivered later.

Handling it on the device (Python SDK):

client.connect_mqtt(host, port=8883, tls=True)

def handle(method, params):
    if method == "getStatus":
        return {"uptime_s": 1234}
    raise ValueError(f"unknown method {method}")

client.on_rpc(handle)

A handler that raises replies {"error": "..."} rather than staying silent — an exception would otherwise be indistinguishable from an offline device, leaving the operator to wait out the timeout with no explanation.

Two independent identity checks guard the reply path: the broker ACL binds the response topic to the connection's Client ID, and flow-core independently re-checks the device JWT claim that rmqtt forwards alongside the message. A reply whose claimed identity disagrees with its topic is dropped, not delivered to the waiter.

Status¶

Both edges are verified end to end: HTTP with a Bearer device JWT, and MQTT-over-TLS — TLS handshake, JWT auth, broker ACL, and a row landing in GreptimeDB.