Node MQTT credentials

Nodes authenticate to EMQX with the same MySQL authenticator as streaming agents. There is no EMQX account API. The Nodes row is the account.

The MQTT username is the node id. The MQTT password is the canonical pairing key: uppercase, with dashes and spaces removed. EMQX compares that password to Nodes.SecretHash (bcrypt).

Authentication

Replace the MySQL authentication query with:

SELECT password_hash FROM (
  SELECT SecretHash AS password_hash
  FROM StreamingAgents
  WHERE Id = ${username} AND Enabled = 1
  UNION ALL
  SELECT SecretHash AS password_hash
  FROM Nodes
  WHERE Id = ${username} AND SecretHash IS NOT NULL
) creds
LIMIT 1;

Authorization

One statement. A streaming agent may publish dataset/{datasetId} and metadata/dataset/{datasetId} when that dataset is linked and allowed to stream. A node may subscribe to dataset/{datasetId} for each streamed dataset whose primary node is this node and that already has a ma-DMP, when Nodes.Id is the MQTT username and SecretHash is set. RelationType and DatasetType are stored as {id}-{name}-{type}.

SELECT
  'publish' AS action,
  'allow' AS permission,
  CONCAT('dataset/', sad.DatasetId) AS topic
FROM StreamingAgentDatasets AS sad
JOIN StreamingAgents AS sa
  ON sad.StreamingAgentId = sa.Id
WHERE sa.Id = ${username}
  AND sa.Enabled = 1
  AND sad.AllowedToStream = 1
UNION ALL
SELECT
  'publish' AS action,
  'allow' AS permission,
  CONCAT('metadata/dataset/', sad.DatasetId) AS topic
FROM StreamingAgentDatasets AS sad
JOIN StreamingAgents AS sa
  ON sad.StreamingAgentId = sa.Id
WHERE sa.Id = ${username}
  AND sa.Enabled = 1
  AND sad.AllowedToStream = 1
UNION ALL
SELECT
  'subscribe' AS action,
  'allow' AS permission,
  CONCAT('dataset/', dn.DatasetId) AS topic
FROM DatasetNodes AS dn
JOIN Datasets AS d ON d.Id = dn.DatasetId
JOIN Nodes AS n ON n.Id = dn.NodeId
WHERE n.Id = ${username}
  AND n.SecretHash IS NOT NULL
  AND dn.RelationType = '1-Primary-DatasetNodeRelationType'
  AND d.DatasetType = '2-StreamedDataset-DatasetType'
  AND EXISTS (
    SELECT 1 FROM MaDmps AS m WHERE m.DatasetId = d.Id
  )

Cutover

After every node is using its pairing key, revoke the shared MQTT JWT. Engines no longer send that token. An already-connected session stays up until the engine process exits, so regenerate a pairing key only when someone can paste the new key on that machine.

Restart engines once after replacing the subscribe rule. A session that already subscribed to dataset/+ keeps that wildcard until the engine process reconnects.