Adds support for clientId and authentication config in MQTT Consumer

databridge.camel-paho/src/main/java/org/eclipse/digitaltwin/basyx/databridge/paho/configuration/MqttConsumerConfiguration.java

@@ -25,20 +25,32 @@
 package org.eclipse.digitaltwin.basyx.databridge.paho.configuration;
 
 import org.eclipse.digitaltwin.basyx.databridge.core.configuration.entity.DataSourceConfiguration;
+import org.eclipse.digitaltwin.basyx.databridge.core.security.BasicCredential;
 
 /**
  * An implementation of MQTT consumer configuration
  * @author haque
  *
  */
 public class MqttConsumerConfiguration extends DataSourceConfiguration {
+
+	private static final String MQTT_PROTOCOL_NAME = "paho:";
+	private static final String BROKER_URL_PARAM_NAME = "brokerUrl";
+	private static final String BROKER_URL_PREFIX = "tcp://";
+	private static final String USERNAME_PARAMETER = "userName";
+	private static final String PASSWORD_PARAMETER = "password";
+
 	private String topic;
+	private String clientId;
+	private BasicCredential credentials;
 
 	public MqttConsumerConfiguration() {}
 
-	public MqttConsumerConfiguration(String uniqueId, String serverUrl, int serverPort, String topic) {
+	public MqttConsumerConfiguration(String uniqueId, String serverUrl, int serverPort, String clientId, String topic, BasicCredential credentials) {
 		super(uniqueId, serverUrl, serverPort);
 		this.topic = topic;
+		this.clientId = clientId;
+		this.credentials = credentials;
 	}
 
 	public String getTopic() {
@@ -49,8 +61,56 @@ public void setTopic(String topic) {
 		this.topic = topic;
 	}
 
+	public BasicCredential getCredentials() {
+		return credentials;
+	}
+
 	public String getConnectionURI() {
-		return "paho:" + getTopic() + "?brokerUrl=tcp://" 
-				+ getServerUrl() + ":" + getServerPort();
+
+		StringBuilder connectionUriBuilder = new StringBuilder();
+		connectionUriBuilder.append(MQTT_PROTOCOL_NAME);
+		connectionUriBuilder.append(topic);
+		connectionUriBuilder.append("?");
+		connectionUriBuilder.append(getMqttBrokerUrl(getServerUrl(), getServerPort()));
+
+		addCredentialsIfConfigured(connectionUriBuilder);
+
+		return connectionUriBuilder.toString();
+	}
+
+	private void addCredentialsIfConfigured(StringBuilder connectionUriBuilder) {
+
+		if (credentials == null || credentials.getUserName().isBlank())
+			return;
+
+		connectionUriBuilder.append("&" + USERNAME_PARAMETER + "=");
+		connectionUriBuilder.append(credentials.getUserName());
+		connectionUriBuilder.append("&" + PASSWORD_PARAMETER + "=");
+		connectionUriBuilder.append(credentials.getPassword());
 	}
+
+	private String getMqttBrokerUrl(String serverUrl, int port) {
+
+		StringBuilder mqttBrokerUrlBuilder = new StringBuilder();
+		mqttBrokerUrlBuilder.append(BROKER_URL_PARAM_NAME);
+		mqttBrokerUrlBuilder.append("=");
+		mqttBrokerUrlBuilder.append(BROKER_URL_PREFIX);
+		mqttBrokerUrlBuilder.append(serverUrl);
+		mqttBrokerUrlBuilder.append(":");
+		mqttBrokerUrlBuilder.append(port);
+
+		addClientIdIfConfigured(mqttBrokerUrlBuilder);
+
+		return mqttBrokerUrlBuilder.toString();
+	}
+
+	private void addClientIdIfConfigured(StringBuilder mqttBrokerUrlBuilder) {
+
+		if (clientId == null || clientId.isBlank())
+			return;
+
+		mqttBrokerUrlBuilder.append("&clientId=");
+		mqttBrokerUrlBuilder.append(clientId);
+	}
+
 }

databridge.core/src/main/java/org/eclipse/digitaltwin/basyx/databridge/core/security/BasicCredential.java

@@ -0,0 +1,52 @@
+/*******************************************************************************
+ * Copyright (C) 2024 the Eclipse BaSyx Authors
+ * 
+ * Permission is hereby granted, free of charge, to any person obtaining
+ * a copy of this software and associated documentation files (the
+ * "Software"), to deal in the Software without restriction, including
+ * without limitation the rights to use, copy, modify, merge, publish,
+ * distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to
+ * the following conditions:
+ * 
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ * 
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+ * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+ * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ * 
+ * SPDX-License-Identifier: MIT
+ ******************************************************************************/
+
+package org.eclipse.digitaltwin.basyx.databridge.core.security;
+
+/**
+ * Basic credentials comprising of username and password
+ * 
+ * @author danish
+ */
+public class BasicCredential {
+
+	private String userName;
+	private String password;
+
+	public BasicCredential(String userName, String password) {
+		super();
+		this.userName = userName;
+		this.password = password;
+	}
+
+	public String getUserName() {
+		return userName;
+	}
+
+	public String getPassword() {
+		return password;
+	}
+
+}

databridge.examples/databridge.examples.mqtt-jsonata-aas/src/main/resources/config/password.conf

@@ -0,0 +1,7 @@
+#*********************************************
+# Each line define a user login in the format
+#   <username>:sha256(<password>)
+#*********************************************
+#NB this password is sha256(passwd)
+
+test:ecd71870d1963316a97e3ac3408c9835ad8cf0f3c1bc703527c30265534f75ae

databridge.examples/databridge.examples.mqtt-jsonata-aas/src/main/resources/mqttconsumer.json

@@ -3,18 +3,31 @@
 		"uniqueId": "property1",
 		"serverUrl": "localhost",
 		"serverPort": 1884,
-		"topic": "Properties"
+		"topic": "Properties",
+		"clientId": "mqttx_c18f5405",
+		"credentials": {
+			"userName": "test",
+			"password": "test123"
+		}
 	},
 	{
 		"uniqueId": "property2",
 		"serverUrl": "localhost",
 		"serverPort": 1884,
-		"topic": "PropertyB"
+		"topic": "PropertyB",
+		"credentials": {
+			"userName": "test",
+			"password": "test123"
+		}
 	},
 	{
 		"uniqueId": "property3",
 		"serverUrl": "localhost",
 		"serverPort": 1884,
-		"topic": "PropertyC"
+		"topic": "PropertyC",
+		"credentials": {
+			"userName": "test",
+			"password": "test123"
+		}
 	}
 ]

databridge.examples/databridge.examples.mqtt-jsonata-aas/src/test/java/org/eclipse/digitaltwin/basyx/databridge/examples/mqttjsonataaas/test/TestAASUpdater.java

@@ -26,7 +26,6 @@
 
 import static org.junit.Assert.assertEquals;
 import java.io.IOException;
-
 import org.eclipse.basyx.aas.manager.ConnectedAssetAdministrationShellManager;
 import org.eclipse.basyx.aas.metamodel.connected.ConnectedAssetAdministrationShell;
 import org.eclipse.basyx.aas.metamodel.map.descriptor.CustomId;
@@ -45,6 +44,7 @@
 import org.eclipse.digitaltwin.basyx.databridge.jsonata.configuration.factory.JsonataDefaultConfigurationFactory;
 import org.eclipse.digitaltwin.basyx.databridge.paho.configuration.factory.MqttDefaultConfigurationFactory;
 import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
@@ -73,6 +73,9 @@ public class TestAASUpdater {
 
 	protected static IIdentifier deviceAASPlainId = new CustomId("TestUpdatedDeviceAAS");
 	protected static IIdentifier deviceAASIriId = new CustomId("https://example.com/ids/aas/7053_6021_1032_9066");
+	private static final String USERNAME = "test";
+	private static final String PASSWORD = "test123";
+
 	private static BaSyxContextConfiguration aasContextConfig;
 
 	@BeforeClass
@@ -166,7 +169,10 @@ private void publishNewDatapoint(String topic) throws MqttException, MqttSecurit
 
 		String json = "{\"Account\":{\"Account Name\":\"Firefly\",\"Order\":[{\"OrderID\":\"order103\",\"Product\":[{\"Product Name\":\"Bowler Hat\",\"ProductID\":858383,\"SKU\":\"0406654608\",\"Description\":{\"Colour\":\"Purple\",\"Width\":300,\"Height\":200,\"Depth\":210,\"Weight\":0.75},\"Price\":34.45,\"Quantity\":2},{\"Product Name\":\"Trilby hat\",\"ProductID\":858236,\"SKU\":\"0406634348\",\"Description\":{\"Colour\":\"Orange\",\"Width\":300,\"Height\":200,\"Depth\":210,\"Weight\":0.6},\"Price\":21.67,\"Quantity\":1}]},{\"OrderID\":\"order104\",\"Product\":[{\"Product Name\":\"Bowler Hat\",\"ProductID\":858383,\"SKU\":\"040657863\",\"Description\":{\"Colour\":\"Purple\",\"Width\":300,\"Height\":200,\"Depth\":210,\"Weight\":0.75},\"Price\":34.45,\"Quantity\":4},{\"ProductID\":345664,\"SKU\":\"0406654603\",\"Product Name\":\"Cloak\",\"Description\":{\"Colour\":\"Black\",\"Width\":30,\"Height\":20,\"Depth\":210,\"Weight\":2},\"Price\":107.99,\"Quantity\":1}]}]}}";
 		MqttClient mqttClient = new MqttClient("tcp://localhost:1884", "testClient", new MemoryPersistence());
-		mqttClient.connect();
+
+		MqttConnectOptions connOpts = setUpMqttConnectionOptions(USERNAME, PASSWORD);
+		mqttClient.connect(connOpts);
+
 		mqttClient.publish(topic, new MqttMessage(json.getBytes()));
 		mqttClient.disconnect();
 		mqttClient.close();
@@ -176,6 +182,8 @@ private static void configureAndStartMqttBroker() throws IOException {
 		mqttBroker = new Server();
 		IResourceLoader classpathLoader = new ClasspathResourceLoader();
 		final IConfig classPathConfig = new ResourceLoaderConfig(classpathLoader);
+		classPathConfig.setProperty("password_file", "config/password.conf");
+
 		mqttBroker.startServer(classPathConfig);
 	}
 
@@ -192,4 +200,12 @@ private ConnectedAssetAdministrationShell getAAS(IIdentifier identifier) {
 		return aas;
 	}
 
+	private static MqttConnectOptions setUpMqttConnectionOptions(String username, String password) {
+		MqttConnectOptions connectOptions = new MqttConnectOptions();
+		connectOptions.setUserName(username);
+		connectOptions.setPassword(password.toCharArray());
+
+		return connectOptions;
+	}
+
 }