MQTT real-time messaging

Real-time pub/sub over AWS IoT Core, MQTT on WebSocket. Browsers connect directly to the broker with temporary AWS credentials; there is no proxy in the data path.

Browser → SigV4-signed WebSocket → AWS IoT Core (managed broker)
Server  → POST /mqtt/publish     → Lambda → IoT Core → all subscribers

Enable once with openkbs mqtt enable (or "mqtt": true in openkbs.json).

The one rule

A token is only as private as the scope you mint it with. Pass subscribe, publish and userId on every token you hand to a browser. AWS IoT Core then enforces that list at the broker — no client-side trust involved.

Omitting the scope mints a legacy project-wide token: its holder can read and write every channel of every user in the project. That form exists only so apps written before scoping keep working. Never mint it for an end user.

Unpredictable channel names are not a substitute. An unscoped token can wildcard-subscribe across the whole project prefix and enumerate every channel without guessing a single name.

Minting a token (server-side)

const projectId = process.env.OPENKBS_PROJECT_ID;
const apiKey = process.env.OPENKBS_API_KEY;

const res = await fetch(`https://project.openkbs.com/projects/${projectId}/mqtt/token`, {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'Authorization': `Bearer ${apiKey}`,
  },
  body: JSON.stringify({
    userId: String(userId),                    // pins the client id to this user
    subscribe: [`user:${userId}`, 'broadcast'], // exactly what this user may read
    publish: [],                                // clients rarely need to publish
    presence: false,                            // opt in only if you use presence
  }),
});
const mqttData = await res.json();

Response:

{
  iotEndpoint, region, topicPrefix, clientIdPrefix,
  presence,                  // whether presence was granted
  scope: { subscribe, publish, presence },   // null on a legacy unscoped token
  credentials: { accessKeyId, secretAccessKey, sessionToken }
}

Credentials last 15 minutes. Mint per user, per session — never cache one token and serve it to several users.

Scope parameters

Channel names accept A-Za-z0-9:._-/ and one optional trailing * for a subtree grant (user:123:*). MQTT wildcards (#, +) and a bare * are rejected — they would silently widen the grant.

Scope budget

The STS session policy is capped at 2048 characters: roughly 8 channels, or 2 with presence. Over the limit the API returns a 400 naming the overflow. For many per-user channels collapse them into one subtree grant:

subscribe: [`user:${userId}:*`]   // one ARN covers every subchannel

Client SDK (browser)

<script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
<script src="https://openkbs.com/sdk/mqtt.js"></script>

const realtime = new MQTT.Realtime({
  credentials: mqttData.credentials,
  iotEndpoint: mqttData.iotEndpoint,
  region: mqttData.region,
  topicPrefix: mqttData.topicPrefix,
  clientIdPrefix: mqttData.clientIdPrefix,
  presence: mqttData.presence,   // required — see "presence" gotcha below
  clientId: `user-${userId}-${Date.now()}`,   // your own unique suffix
  debug: false,
});

realtime.connection.on('connected', () => console.log('online'));
realtime.connection.on('disconnected', () => console.log('offline'));

const channel = realtime.channels.get('posts');
channel.subscribe((msg) => console.log(msg.name, msg.data));    // all messages
channel.subscribe('new_post', (msg) => console.log(msg.data));  // one event
channel.publish('greeting', { text: 'Hello!' });                // if granted

// Presence — only when the token granted it
channel.presence.enter({ name: 'Alice' });
channel.presence.subscribe((members) => console.log(members));
channel.presence.subscribe('enter', (m) => console.log(m, 'joined'));
channel.presence.leave();

realtime.close();

Subscribe only to channels the token granted. See the failure modes below for why that matters more than it looks.

Publishing from the server (metered)

await fetch(`https://project.openkbs.com/projects/${projectId}/mqtt/publish`, {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'Authorization': `Bearer ${apiKey}`,
  },
  body: JSON.stringify({ channel: 'posts', event: 'new_post', data: { id: 1 } }),
});

Prefer this over client publish for anything that matters. A client granted publish on a shared channel can forge messages that look like they came from any other user — the broker authorises the topic, not the sender's identity.

Failure modes

All of these are silent: the connection dies between browser and broker, so nothing appears in your function logs. Diagnose in DevTools → Network → the mqtt?X-Amz-... socket → Messages. A 4-byte reply beginning 20 02 00 is the CONNACK; its last byte is the reason code.

20 02 00 02 — identifier rejected

The client id exceeds IoT Core's 128-byte limit. The SDK sends clientIdPrefix + clientId; a scoped prefix is 49 bytes, so your own suffix must stay under ~79. user-${uuid}-${Date.now()} is 55 and fine; concatenating two UUIDs is not. Projects whose users.id is a UUID hit this first.

20 02 00 05 — not authorised

The client id does not match the prefix the token was minted with. Always use mqttData.clientIdPrefix verbatim; never hardcode it or carry one over from an earlier token.

Connection dies the instant it opens

presence: mqttData.presence was not passed into MQTT.Realtime. The SDK then arms a presence Last-Will on a topic a scoped token does not cover, and the broker drops the connection at CONNECT.

The symptom is distinctive and worth recognising: the user can still send but receives nothing, because sending is plain HTTPS while receiving needs the dead socket. Combined with auto-reconnect it looks like a permissions bug and produces no error anywhere.

Everything stops after one bad subscribe

An unauthorised subscribe does not fail on its own — IoT Core closes the whole connection, taking every other subscription with it. The SDK then reconnects and repeats, forever. Subscribe strictly within the granted scope; scope is on the token response if you need to check at runtime.

A vendored SDK freezes for a year

openkbs site deploy caches an unhashed .js for an hour and a file with a build hash in its name (mqtt-a1b2c3d4.js) for a year, immutable (see Site Cache-Control). A vendored copy under a fixed filename therefore lags up to an hour behind each deploy; put a version in the filename (openkbs-mqtt.v3.js) when an update must land immediately — or just load it from https://openkbs.com/sdk/mqtt.js.

Test with a real user id

Client-id length, channel-name validity and policy size all scale with your actual id format. A token minted for userId: '123' proves nothing about a project whose users are UUIDs.

Security summary

• Credentials reach only IoT Core — never S3, Lambda, or any other AWS service.
• Cross-project access is impossible: topic ARNs are pinned to the project.
• Within a project, isolation is exactly what you put in subscribe/publish.
• Revocation is not instant. Credentials expire in 15 minutes, but an established connection keeps its subscriptions until it drops.
• Presence exposes the clientIds of everyone on the granted channels. Do not put anything private in a clientId.

Billing

Server-side publish is billed probabilistically: 5 credits per 10,000 messages. Client-side publish and presence traffic are free.

CLI

openkbs mqtt info
openkbs mqtt enable
openkbs mqtt disable
openkbs mqtt token -u <userId> -s user:123,broadcast   # --publish, --presence
openkbs mqtt publish <channel> -d '<json>'

openkbs mqtt token warns when it returns an unscoped token.

Related

• Elastic Workers
• Site Cache-Control
• CLI reference