Add subcomponent/subdevice processing

Up to now the IFF-agent can only manage one single device with a certain id. This limits cases where the device
is consisting of several subsystems. For such cases, all the subsystem data was mapped to the main device with
the respective id. With these changes, a device can now consist of several subsystems and these IDs can be added
to the device token. This PR contains everything needed to support subdevice/subcomponent processing:

* IFF-Agent accepts deviceIds in the TCP/UDP messages
* IFF-Agent utils offer additional options to add subcomponent IDs and send data for subcompoentns
* Keycloak allows now the field "subdevice_ids" in the token to add subdevice IDs
* The MQTT-Bridge permits subdevice IDs to stream data

Related Epic: IndustryFusion#514
Related User-stories: IndustryFusion#555

Signed-off-by: marcel <wagmarcel@web.de>

KafkaBridge/lib/authService/acl.js

@@ -38,6 +38,7 @@ class Acl {
  return;
  }
  const topic = req.query.topic;
+ const clientid = req.query.clientid;
  this.logger.debug('ACL request for username ' + username + ' and topic ' + topic);
  // allow all $SYS topics
  if (topic.startsWith('$SYS/')) {
@@ -51,18 +52,26 @@ class Acl {
  const splitTopic = topic.split('/');
  if (splitTopic[0] === 'spBv1.0') {
  const spBAccountId = splitTopic[1];
+ const gateway = splitTopic[3];
+ const command = splitTopic[2];
  const spBdevId = splitTopic[4];
  const spBAclKey = spBAccountId + '/' + spBdevId;
- const allowed = await this.cache.getValue(spBAclKey, 'acl');
- if (allowed === undefined || !(allowed === 'true') || spBdevId !== username) {
+ let allowed = await this.cache.getValue(spBAclKey, 'acl');
+ if (allowed === undefined && spBdevId === '' && command === 'NBIRTH') { // if it is a NBIRTH command check if gatewayid is permitted for this session
+ allowed = await this.cache.getValue(spBAccountId + '/' + gateway, 'acl');
+ if (allowed === undefined) {
+ this.logger.warn('Gateway id not permitted for this token/session. Use a token which has device_id==gateway_id.');
+ }
+ }
+ if (allowed === undefined || allowed !== clientid) {
  this.logger.info('Connection rejected for realm ' + spBAccountId + ' and device ' + spBdevId);
- res.sendStatus(400);
+ return res.status(200).json({ result: 'deny' });
  } else {
- res.status(200).json({ result: 'allow' });
+ return res.status(200).json({ result: 'allow' });
  }
  } else {
  this.logger.warn('Topic sructure not valid.');
- res.sendStatus(400);
+ return res.status(200).json({ result: 'deny' });
  }
  }
 }

KafkaBridge/lib/authService/authenticate.js

@@ -53,11 +53,22 @@ class Authenticate {
  this.cache.init();
  }
 
+ async addSubdeviceAcl (realm, clientid, decodedToken) {
+ if ('subdevice_ids' in decodedToken) {
+ const subdevices = decodedToken.subdevice_ids;
+ const parsedSubdevices = JSON.parse(subdevices);
+ for (const did of parsedSubdevices) {
+ await this.cache.setValue(realm + '/' + did, 'acl', clientid);
+ }
+ }
+ }
+
  // expects "username" and "password" as url-query-parameters
  async authenticate (req, res) {
  this.logger.debug('Auth request ' + JSON.stringify(req.query));
  const username = req.query.username;
  const token = req.query.password;
+ const clientid = req.query.clientid;
  if (username === this.config.mqtt.adminUsername) {
  if (token === this.config.mqtt.adminPassword) {
  // superuser
@@ -67,20 +78,20 @@ class Authenticate {
  } else {
  // will also kick out tokens who use the superuser name as deviceId
  this.logger.warn('Wrong Superuser password.');
- res.sendStatus(400);
+ res.status(200).json({ result: 'deny' });
  return;
  }
  }
  const decodedToken = await this.verifyAndDecodeToken(token);
  this.logger.debug('token decoded: ' + JSON.stringify(decodedToken));
  if (decodedToken === null) {
  this.logger.info('Could not decode token.');
- res.sendStatus(400);
+ res.status(200).json({ result: 'deny' });
  return;
  }
  if (!validate(decodedToken, username)) {
  this.logger.warn('Validation of token failed. Username: ' + username);
- res.sendStatus(400);
+ res.status(200).json({ result: 'deny' });
  return;
  }
  // check whether accounts contains only one element and role is device
@@ -89,15 +100,17 @@ class Authenticate {
  const realm = getRealm(decodedToken);
  if (did === null || did === undefined || realm === null || realm === undefined) {
  this.logger.warn('Validation failed: Device id or realm not valid.');
- res.sendStatus(400);
+ res.status(200).json({ result: 'deny' });
  return;
  }
  if (did === this.config.mqtt.tainted || gateway === this.config.mqtt.tainted) {
  this.logger.warn('This token is tained! Rejecting.');
- res.sendStatus(400);
+ res.status(200).json({ result: 'deny' });
  }
  // put realm/device into the list of accepted topics
- await this.cache.setValue(realm + '/' + did, 'acl', 'true');
+ await this.cache.deleteKeysWithValue('acl', clientid);
+ await this.addSubdeviceAcl(realm, clientid, decodedToken);
+ await this.cache.setValue(realm + '/' + did, 'acl', clientid);
  res.status(200).json({ result: 'allow', is_superuser: 'false' });
  }
 

KafkaBridge/lib/cache/index.js

@@ -43,5 +43,29 @@ class Cache {
  const obj = await this.redisClient.hGetAll(key);
  return obj[valueKey];
  }
+
+ async deleteKeysWithValue (valueKey, clientid) {
+ let cursor = 0;
+ const keysToDelete = [];
+
+ do {
+ const reply = await this.redisClient.scan(cursor);
+ cursor = parseInt(reply.cursor, 10);
+ const keys = reply.keys;
+
+ for (const key of keys) {
+ const value = await this.redisClient.hGet(key, valueKey);
+ if (value === clientid) {
+ keysToDelete.push(key);
+ }
+ }
+ } while (cursor !== 0);
+
+ for (const key of keysToDelete) {
+ await this.redisClient.del(key);
+ }
+
+ this.logger.info(`Deleted keys with ${valueKey}=${clientid}: ${keysToDelete.join(', ')}`);
+ }
 }
 module.exports = Cache;

KafkaBridge/mqttBridge/sparkplug_data_ingestion.js

@@ -292,6 +292,9 @@ module.exports = class SparkplugHandler {
  /* It will be checked if the ttl exist, if it exits the package need to be discarded
  */
  const subTopic = topic.split('/');
+ if (subTopic[2] !== 'DDATA') {
+ return;
+ }
  this.logger.debug('Data Submission Detected : ' + topic + ' Message: ' + JSON.stringify(message));
  if (Object.values(MESSAGE_TYPE.WITHSEQ).includes(subTopic[2])) {
  const validationResult = this.validator.validate(message, dataSchema.SPARKPLUGB);

KafkaBridge/test/lib_authServiceTest.js

@@ -240,8 +240,12 @@ describe(fileToTest, function () {
  }
  };
  const res = {
- sendStatus: function (status) {
- assert.equal(status, 400, 'Received wrong status');
+ status: function (status) {
+ assert.equal(status, 200, 'Received wrong status');
+ return this;
+ },
+ json: function (resultObj) {
+ resultObj.should.deep.equal({ result: 'deny' });
  done();
  }
  };
@@ -285,8 +289,12 @@ describe(fileToTest, function () {
  }
  };
  const res = {
- sendStatus: function (status) {
- assert.equal(status, 400, 'Received wrong status');
+ status: function (status) {
+ assert.equal(status, 200, 'Received wrong status');
+ return this;
+ },
+ json: function (resultObj) {
+ resultObj.should.deep.equal({ result: 'deny' });
  done();
  }
  };
@@ -386,7 +394,7 @@ describe(fileToTest, function () {
  getValue (subtopic, key) {
  assert.equal(aidSlashDid, subtopic, 'Wrong accountId/did subtopic');
  assert.equal(key, 'acl', 'Wrong key value');
- return 'true';
+ return 'clientid';
  }
  };
  ToTest.__set__('Cache', Cache);
@@ -403,6 +411,7 @@ describe(fileToTest, function () {
  const req = {
  query: {
  username: 'deviceId',
+ clientid: 'clientid',
  topic: 'spBv1.0/accountId/DBIRTH/eonID/deviceId'
  }
  };
@@ -447,8 +456,12 @@ describe(fileToTest, function () {
  }
  };
  const res = {
- sendStatus: function (status) {
- assert.equal(status, 400, 'Received wrong status');
+ status: function (status) {
+ assert.equal(status, 200, 'Received wrong status');
+ return this;
+ },
+ json: function (resultObj) {
+ resultObj.should.deep.equal({ result: 'deny' });
  done();
  }
  };
@@ -480,8 +493,12 @@ describe(fileToTest, function () {
  }
  };
  const res = {
- sendStatus: function (status) {
- assert.equal(status, 400, 'Received wrong status');
+ status: function (status) {
+ assert.equal(status, 200, 'Received wrong status');
+ return this;
+ },
+ json: function (resultObj) {
+ resultObj.should.deep.equal({ result: 'deny' });
  done();
  }
  };

KafkaBridge/test/lib_cacheTest.js

@@ -103,4 +103,40 @@ describe(fileToTest, function () {
  cache.getValue('key').then(result => result.should.equal('true'));
  done();
  });
+
+ it('Shall test deleteKeysWithValue', function (done) {
+ const config = {
+ cache: {
+ port: 1234,
+ host: 'redishost'
+ }
+ };
+
+ const redis = {
+ createClient: function () {
+ return {
+ on: function (evType) {
+ evType.should.equal('error');
+ },
+ scan: async function (cursor) {
+ return { cursor: '0', keys: ['key1', 'key2', 'key3'] };
+ },
+ hGet: async function (key, valueKey) {
+ if (key === 'key1' && valueKey === 'field1') return 'clientid1';
+ if (key === 'key2' && valueKey === 'field1') return 'clientid2';
+ if (key === 'key3' && valueKey === 'field1') return 'clientid1';
+ return null;
+ },
+ del: async function (key) {
+ }
+ };
+ }
+ };
+ ToTest.__set__('redis', redis);
+ const cache = new ToTest(config);
+ cache.deleteKeysWithValue('field1', 'clientid1').then(() => {
+ // Add assertions for the deletion logic if needed
+ done();
+ });
+ });
 });

Keycloak/iff-js-providers/META-INF/keycloak-scripts.json

@@ -11,6 +11,11 @@
  "name": "Device ID Mapper",
  "fileName": "deviceid-mapper.js",
  "description": "deviceId - only valid if access type is device"
+ },
+ {
+ "name": "Device SUB IDs Mapper",
+ "fileName": "subdeviceids-mapper.js",
+ "description": "subdeviceIds - only valid if access type is device"
  }
  ],
  "saml-mappers": []

Keycloak/iff-js-providers/subdeviceids-mapper.js

@@ -0,0 +1,77 @@
+/**
+ * Copyright (c) 2023 Intel Corporation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Available variables:
+ * user - the current user
+ * realm - the current realm
+ * token - the current token
+ * userSession - the current userSession
+ * keycloakSession - the current keycloakSession
+ */
+
+var onboarding_token_expiration = java.lang.System.getenv("OISP_FRONTEND_DEVICE_ACCOUNT_ENDPOINT");
+var subdeviceIdsH = keycloakSession.getContext().getRequestHeaders()
+ .getRequestHeader("X-SubDeviceIDs")[0];
+if (subdeviceIdsH !== null && subdeviceIdsH !== undefined) {
+ subdeviceIdsH = JSON.parse(subdeviceIdsH)
+}
+var inputRequest = keycloakSession.getContext().getHttpRequest();
+var params = inputRequest.getDecodedFormParameters();
+var origTokenParam = params.getFirst("orig_token");
+var grantType = params.getFirst("grant_type");
+var tokens = keycloakSession.tokens();
+var origToken = tokens.decode(origTokenParam, Java.type("org.keycloak.representations.AccessToken").class)
+
+if (typeof(onboarding_token_expiration) !== 'number') {
+ // if not otherwise configured onboardig token is valid for 5 minutes
+ onboarding_token_expiration = 300;
+}
+if (grantType === 'refresh_token' && origToken !== null) {
+ var session = userSession.getId();
+ var otherClaims = origToken.getOtherClaims();
+ var origTokenSubDeviceIds;
+ if (otherClaims !== null) {
+
+ origTokenSubDeviceIds = otherClaims.get("sub_device_ids");
+ }
+ var origTokenSession = origToken.getSessionId();
+
+ if (origTokenSubDeviceIds !== null && origTokenSubDeviceIds !== undefined) {
+ // Has origToken same session?
+ if (origTokenSession !== session) {
+ print("Warning: Rejecting subdeviceids claim due to session mismatch between refresh_token and orig_token")
+ exports = JSON.stringify([]);
+ } else {
+ exports = origTokenSubDeviceIds;
+ }
+ } else {
+ // If there is no origTokenDeviceId, there must be an X-DeviceId header AND origToken must be valid
+ if (!origToken.isExpired() && subdeviceIdsH !== null && subdeviceIdsH !== undefined) {
+ exports = subdeviceIdsH
+ } else {
+ print("Warning: Rejecting subdeviceid claim due to orig_token is expired or there is not valid X-SubDeviceIDs Header.")
+ exports = JSON.stringify([]);
+ } 
+ }
+} else if (grantType === 'password'){
+ var currentTimeInSeconds = new Date().getTime() / 1000;
+ token.exp(currentTimeInSeconds + onboarding_token_expiration);
+ exports = null
+} else if (origToken === null) {
+ print("Warning: Rejecting token due to invalid orig_token.")
+ exports = JSON.stringify([])
+}