Skip to content

Commit 47f4a4a

Browse files
github-actions[bot]claude
github-actions[bot]
and
claude
committed
test(mqtt): add MQTT integration test suite (#1188)
New file: integrationTests/mqtt.test.ts New fixture: integrationTests/components/fixtures/mqtt-full/ Covers the full auth + ACL stack from fleet investigation: - Connect/disconnect with valid HS256 and RS256 JWT credentials - Invalid credentials and mismatched clientId rejection - Group-based ACL via @harperdb/acl-connect (dogPublisher/dogSubscriber) - QoS 0, 1, and 2 publish/subscribe semantics - %u topic substitution: user-topics/<username>/# enforced in resources.js - anonymousSubscriber:true — anonymous clients can subscribe to public/# - MQTT-backed REST table: PUT via REST triggers MQTT subscriber on Pet table - Schema-less retained message table (Sensor, attributes:[]) via schema.graphql - Durable session (cleanSession:false): queued messages delivered on reconnect - MQTT broker accessible via WebSocket transport - SYS_CON monitoring: $SYS/monitor/con/connects and $SYS/drops events - High-throughput: 1K rapid connect/disconnect with no SYS_CON data loss Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent 88c94e6 commit 47f4a4a

6 files changed

Lines changed: 942 additions & 0 deletions

File tree

‎integrationTests/components/fixtures/mqtt-full/config.yaml‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
rest: true
2+
graphqlSchema:
3+
files: schema.graphql
4+
jsResource:
5+
files: resources.js
6+
'@harperdb/acl-connect':
7+
package: '@harperdb/acl-connect'
8+
files: connect.json

‎integrationTests/components/fixtures/mqtt-full/connect.json‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
{
2+
"options": {
3+
"clientId": "clientID",
4+
"userName": "username",
5+
"authorizations": "authGroups"
6+
},
7+
"acls": [
8+
{
9+
"topicFilter": "dog/#",
10+
"publishers": ["dogPublisher"],
11+
"subscribers": ["dogSubscriber"],
12+
"anonymousSubscriber": false
13+
},
14+
{
15+
"topicFilter": "public/#",
16+
"publishers": [],
17+
"subscribers": [],
18+
"anonymousSubscriber": true
19+
},
20+
{
21+
"topicFilter": "$SYS/#",
22+
"publishers": [],
23+
"subscribers": ["sysMonitor"],
24+
"anonymousSubscriber": false
25+
},
26+
{
27+
"topicFilter": "user-topics/#",
28+
"publishers": ["userPub"],
29+
"subscribers": ["userSub"],
30+
"anonymousSubscriber": false
31+
}
32+
]
33+
}

‎integrationTests/components/fixtures/mqtt-full/package.json‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
{
2+
"name": "mqtt-full",
3+
"version": "1.0.0",
4+
"type": "module",
5+
"main": "resources.js",
6+
"license": "Apache-2.0",
7+
"dependencies": {
8+
"@harperdb/acl-connect": "1.0.10",
9+
"jsonwebtoken": "^9.0.2"
10+
}
11+
}

‎integrationTests/components/fixtures/mqtt-full/resources.js‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
import jwt from 'jsonwebtoken';
2+
import SETTINGS from './connect.json' with { type: 'json' };
3+
4+
class User {
5+
constructor(username, clientID, authGroups) {
6+
this.active = true;
7+
this.username = username;
8+
this.client_id = clientID;
9+
this.authGroups = authGroups;
10+
this.role = { role: authGroups, permission: { super_user: false } };
11+
}
12+
}
13+
14+
// Override server.getUser to accept JWT tokens as the MQTT password.
15+
// jwt.decode() reads claims without verifying the signature, which means both
16+
// HS256 and RS256 (and any other algorithm) tokens are accepted — the test
17+
// exercises that RS256 claims are extracted correctly.
18+
const hdbGetUser = server.getUser;
19+
server.getUser = async function (username, password) {
20+
if (password?.length > 100 && password.split('.').length === 3) {
21+
try {
22+
const decoded = jwt.decode(password);
23+
if (decoded) {
24+
return new User(
25+
decoded[SETTINGS.options.userName] ?? username,
26+
decoded[SETTINGS.options.clientId],
27+
decoded[SETTINGS.options.authorizations]
28+
);
29+
}
30+
} catch (e) {
31+
const msg = `Error decoding token: ${e.message}. For username: ${username}`;
32+
throw new Error(msg);
33+
}
34+
}
35+
const user = await hdbGetUser(username, password);
36+
user.client_id = username;
37+
return user;
38+
};
39+
40+
// Validate that the MQTT clientId matches the clientID claim in the JWT.
41+
// Anonymous connections (no user) are allowed as long as they do not specify
42+
// a clientId and use a clean session.
43+
server.mqtt.authorizeClient = (connection_message, user) => {
44+
if (!user) {
45+
if (connection_message.clientId) throw new Error('Cannot specify a client id for anonymous connections');
46+
if (!connection_message.clean) throw new Error('Anonymous connections must use clean sessions');
47+
} else if (connection_message.clientId !== user.client_id && !user.role?.permission?.super_user) {
48+
throw new Error('Invalid client id: must match the clientID claim in the JWT token');
49+
}
50+
};
51+
52+
// Implement %u topic substitution for user-topics/# subscriptions and publishes.
53+
// When a client subscribes or publishes to user-topics/<segment>/..., the first
54+
// segment after the prefix must equal the authenticated username. This mirrors
55+
// the Mosquitto/production %u pattern that prevents cross-user topic access
56+
// (Ubisoft prod, v4.3.34).
57+
server.mqtt.events.on('connected', (session) => {
58+
const USER_TOPICS_PREFIX = 'user-topics/';
59+
60+
const origAddSubscription = session.addSubscription.bind(session);
61+
session.addSubscription = async (subscription, needsAck, filter) => {
62+
const { topic } = subscription;
63+
if (topic.startsWith(USER_TOPICS_PREFIX)) {
64+
const rest = topic.slice(USER_TOPICS_PREFIX.length);
65+
const userSeg = rest.split('/')[0];
66+
if (userSeg && userSeg !== '#' && userSeg !== '+') {
67+
if (userSeg !== session.user?.username) {
68+
const err = Object.assign(
69+
new Error('%u substitution: topic user segment must match the connected username'),
70+
{
71+
statusCode: 403,
72+
}
73+
);
74+
throw err;
75+
}
76+
}
77+
}
78+
return origAddSubscription(subscription, needsAck, filter);
79+
};
80+
81+
const origPublish = session.publish.bind(session);
82+
session.publish = async (message, data) => {
83+
const { topic } = message;
84+
if (topic.startsWith(USER_TOPICS_PREFIX)) {
85+
const rest = topic.slice(USER_TOPICS_PREFIX.length);
86+
const userSeg = rest.split('/')[0];
87+
if (userSeg && userSeg !== '#' && userSeg !== '+') {
88+
if (userSeg !== session.user?.username) {
89+
const err = Object.assign(
90+
new Error('%u substitution: topic user segment must match the connected username'),
91+
{ statusCode: 403 }
92+
);
93+
throw err;
94+
}
95+
}
96+
}
97+
return origPublish(message, data);
98+
};
99+
});

‎integrationTests/components/fixtures/mqtt-full/schema.graphql‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
type Pet @table @export {
2+
id: String @primaryKey
3+
name: String
4+
}
5+
6+
type Sensor @table @export {
7+
id: String @primaryKey
8+
}

0 commit comments

Comments
 (0)