New type of credentials: MQTT

Andrii Shvaika
1 parent 6e492c78
Showing 8 changed files with 201 additions and 33 deletions
application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
common/data/src/main/java/org/thingsboard/server/common/data/device/credentials/BasicMqttCredentials.java
common/message/src/main/java/org/thingsboard/server/common/msg/EncryptionUtil.java
common/queue/src/main/proto/queue.proto
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
dao/src/main/java/org/thingsboard/server/dao/device/DeviceCredentialsServiceImpl.java
--- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
View file @a3326b4
+++ b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
View file @a3326b4
@@ -23,12 +23,14 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.protobuf.ByteString;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.TenantProfile;
+import org.thingsboard.server.common.data.device.credentials.BasicMqttCredentials;
 import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.DeviceProfileId;
@@ -36,6 +38,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
 import org.thingsboard.server.common.data.security.DeviceCredentialsType;
+import org.thingsboard.server.common.msg.EncryptionUtil;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgDataType;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
@@ -46,6 +49,7 @@ import org.thingsboard.server.dao.device.DeviceService;
 import org.thingsboard.server.dao.relation.RelationService;
 import org.thingsboard.server.dao.tenant.TenantProfileService;
 import org.thingsboard.server.dao.tenant.TenantService;
+import org.thingsboard.server.dao.util.mapping.JacksonUtil;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.DeviceInfoProto;
 import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg;
@@ -91,6 +95,7 @@ public class DefaultTransportApiService implements TransportApiService {
     private final TbClusterService tbClusterService;
     private final DataDecodingEncodingService dataDecodingEncodingService;
+
     private final ConcurrentMap<String, ReentrantLock> deviceCreationLocks = new ConcurrentHashMap<>();
     public DefaultTransportApiService(DeviceProfileService deviceProfileService, TenantService tenantService,
@@ -117,6 +122,10 @@ public class DefaultTransportApiService implements TransportApiService {
             ValidateDeviceTokenRequestMsg msg = transportApiRequestMsg.getValidateTokenRequestMsg();
             return Futures.transform(validateCredentials(msg.getToken(), DeviceCredentialsType.ACCESS_TOKEN),
                     value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
+        } else if (transportApiRequestMsg.hasValidateBasicMqttCredRequestMsg()) {
+            TransportProtos.ValidateBasicMqttCredRequestMsg msg = transportApiRequestMsg.getValidateBasicMqttCredRequestMsg();
+            return Futures.transform(validateCredentials(msg),
+                    value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
         } else if (transportApiRequestMsg.hasValidateX509CertRequestMsg()) {
             ValidateDeviceX509CertRequestMsg msg = transportApiRequestMsg.getValidateX509CertRequestMsg();
             return Futures.transform(validateCredentials(msg.getHash(), DeviceCredentialsType.X509_CERTIFICATE),
@@ -130,7 +139,6 @@ public class DefaultTransportApiService implements TransportApiService {
         } else if (transportApiRequestMsg.hasGetDeviceProfileRequestMsg()) {
             return Futures.transform(handle(transportApiRequestMsg.getGetDeviceProfileRequestMsg()),
                     value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
-
         }
         return Futures.transform(getEmptyTransportApiResponseFuture(),
                 value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
@@ -146,6 +154,62 @@ public class DefaultTransportApiService implements TransportApiService {
         }
     }
+    private ListenableFuture<TransportApiResponseMsg> validateCredentials(TransportProtos.ValidateBasicMqttCredRequestMsg mqtt) {
+        DeviceCredentials credentials = deviceCredentialsService.findDeviceCredentialsByCredentialsId(mqtt.getUserName());
+        if (credentials != null) {
+            if (credentials.getCredentialsType() == DeviceCredentialsType.ACCESS_TOKEN) {
+                return getDeviceInfo(credentials.getDeviceId(), credentials);
+            } else if (credentials.getCredentialsType() == DeviceCredentialsType.MQTT_BASIC) {
+                if (!checkMqttCredentials(mqtt, credentials)) {
+                    credentials = null;
+                }
+            }
+        }
+        if (credentials == null) {
+            credentials = checkMqttCredentials(mqtt, EncryptionUtil.getSha3Hash("|", mqtt.getClientId(), mqtt.getUserName()));
+            if (credentials == null) {
+                credentials = checkMqttCredentials(mqtt, EncryptionUtil.getSha3Hash(mqtt.getClientId()));
+            }
+        }
+        if (credentials != null) {
+            return getDeviceInfo(credentials.getDeviceId(), credentials);
+        } else {
+            return getEmptyTransportApiResponseFuture();
+        }
+    }
+
+    private DeviceCredentials checkMqttCredentials(TransportProtos.ValidateBasicMqttCredRequestMsg clientCred, String credId) {
+        DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByCredentialsId(credId);
+        if (deviceCredentials != null && deviceCredentials.getCredentialsType() == DeviceCredentialsType.MQTT_BASIC) {
+            if (!checkMqttCredentials(clientCred, deviceCredentials)) {
+                return null;
+            } else {
+                return deviceCredentials;
+            }
+        }
+        return null;
+    }
+
+    private boolean checkMqttCredentials(TransportProtos.ValidateBasicMqttCredRequestMsg clientCred, DeviceCredentials deviceCredentials) {
+        BasicMqttCredentials dbCred = JacksonUtil.fromString(deviceCredentials.getCredentialsValue(), BasicMqttCredentials.class);
+        if (!StringUtils.isEmpty(dbCred.getClientId()) && !dbCred.getClientId().equals(clientCred.getClientId())) {
+            return false;
+        }
+        if (!StringUtils.isEmpty(dbCred.getUserName()) && !dbCred.getUserName().equals(clientCred.getUserName())) {
+            return false;
+        }
+        if (!StringUtils.isEmpty(dbCred.getPassword())) {
+            if (StringUtils.isEmpty(clientCred.getPassword())) {
+                return false;
+            } else {
+                if (!dbCred.getPassword().equals(clientCred.getPassword())) {
+                    return false;
+                }
+            }
+        }
+        return true;
+    }
+
     private ListenableFuture<TransportApiResponseMsg> handle(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) {
         DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB()));
         ListenableFuture<Device> gatewayFuture = deviceService.findDeviceByIdAsync(TenantId.SYS_TENANT_ID, gatewayId);
@@ -237,7 +301,7 @@ public class DefaultTransportApiService implements TransportApiService {
                     builder.setCredentialsBody(credentials.getCredentialsValue());
                 }
                 return TransportApiResponseMsg.newBuilder()
-                        .setValidateTokenResponseMsg(builder.build()).build();
+                        .setValidateCredResponseMsg(builder.build()).build();
             } catch (JsonProcessingException e) {
                 log.warn("[{}] Failed to lookup device by id", deviceId, e);
                 return getEmptyTransportApiResponse();
@@ -265,6 +329,6 @@ public class DefaultTransportApiService implements TransportApiService {
     private TransportApiResponseMsg getEmptyTransportApiResponse() {
         return TransportApiResponseMsg.newBuilder()
-                .setValidateTokenResponseMsg(ValidateDeviceCredentialsResponseMsg.getDefaultInstance()).build();
+                .setValidateCredResponseMsg(ValidateDeviceCredentialsResponseMsg.getDefaultInstance()).build();
     }
 }
--- a/common/data/src/main/java/org/thingsboard/server/common/data/device/credentials/BasicMqttCredentials.java 0 → 100644
View file @a3326b4
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/credentials/BasicMqttCredentials.java 0 → 100644
View file @a3326b4
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * 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.
+ */
+package org.thingsboard.server.common.data.device.credentials;
+
+import lombok.Data;
+
+@Data
+public class BasicMqttCredentials {
+
+    private String clientId;
+    private String userName;
+    private String password;
+
+}
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/EncryptionUtil.java
View file @a3326b4
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/EncryptionUtil.java
View file @a3326b4
@@ -18,6 +18,7 @@ package org.thingsboard.server.common.msg;
 import lombok.extern.slf4j.Slf4j;
 import org.bouncycastle.crypto.digests.SHA3Digest;
 import org.bouncycastle.pqc.math.linearalgebra.ByteUtils;
+
 /**
  * @author Valerii Sosliuk
  */
@@ -30,8 +31,8 @@ public class EncryptionUtil {
     public static String trimNewLines(String input) {
         return input.replaceAll("-----BEGIN CERTIFICATE-----", "")
                 .replaceAll("-----END CERTIFICATE-----", "")
-                .replaceAll("\n","")
-                .replaceAll("\r","");
+                .replaceAll("\n", "")
+                .replaceAll("\r", "");
     }
     public static String getSha3Hash(String data) {
@@ -45,4 +46,20 @@ public class EncryptionUtil {
         String sha3Hash = ByteUtils.toHexString(hashedBytes);
         return sha3Hash;
     }
+
+    public static String getSha3Hash(String delim, String... tokens) {
+        StringBuilder sb = new StringBuilder();
+        boolean first = true;
+        for (String token : tokens) {
+            if (token != null && !token.isEmpty()) {
+                if (first) {
+                    first = false;
+                } else {
+                    sb.append(delim);
+                }
+                sb.append(token);
+            }
+        }
+        return getSha3Hash(sb.toString());
+    }
 }
--- a/common/queue/src/main/proto/queue.proto
View file @a3326b4
+++ b/common/queue/src/main/proto/queue.proto
View file @a3326b4
@@ -147,6 +147,12 @@ message ValidateDeviceX509CertRequestMsg {
   string hash = 1;
 }
+message ValidateBasicMqttCredRequestMsg {
+  string clientId = 1;
+  string userName = 2;
+  string password = 3;
+}
+
 message ValidateDeviceCredentialsResponseMsg {
   DeviceInfoProto deviceInfo = 1;
   string credentialsBody = 2;
@@ -429,11 +435,12 @@ message TransportApiRequestMsg {
   GetOrCreateDeviceFromGatewayRequestMsg getOrCreateDeviceRequestMsg = 3;
   GetTenantRoutingInfoRequestMsg getTenantRoutingInfoRequestMsg = 4;
   GetDeviceProfileRequestMsg getDeviceProfileRequestMsg = 5;
+  ValidateBasicMqttCredRequestMsg validateBasicMqttCredRequestMsg = 6;
 }
 /* Response from ThingsBoard Core Service to Transport Service */
 message TransportApiResponseMsg {
-  ValidateDeviceCredentialsResponseMsg validateTokenResponseMsg = 1;
+  ValidateDeviceCredentialsResponseMsg validateCredResponseMsg = 1;
   GetOrCreateDeviceFromGatewayResponseMsg getOrCreateDeviceResponseMsg = 2;
   GetTenantRoutingInfoResponseMsg getTenantRoutingInfoResponseMsg = 4;
   GetDeviceProfileResponseMsg getDeviceProfileResponseMsg = 5;
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View file @a3326b4
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View file @a3326b4
@@ -66,6 +66,7 @@ import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.security.cert.X509Certificate;
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
@@ -365,25 +366,27 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     private void processAuthTokenConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
         String userName = msg.payload().userName();
         log.info("[{}] Processing connect msg for client with user name: {}!", sessionId, userName);
-        if (StringUtils.isEmpty(userName)) {
-            ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD));
-            ctx.close();
-        } else {
-            transportService.process(DeviceTransportType.MQTT, ValidateDeviceTokenRequestMsg.newBuilder().setToken(userName).build(),
-                    new TransportServiceCallback<ValidateDeviceCredentialsResponse>() {
-                        @Override
-                        public void onSuccess(ValidateDeviceCredentialsResponse msg) {
-                            onValidateDeviceResponse(msg, ctx);
-                        }
-
-                        @Override
-                        public void onError(Throwable e) {
-                            log.trace("[{}] Failed to process credentials: {}", address, userName, e);
-                            ctx.writeAndFlush(createMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE));
-                            ctx.close();
-                        }
-                    });
+        TransportProtos.ValidateBasicMqttCredRequestMsg.Builder request = TransportProtos.ValidateBasicMqttCredRequestMsg.newBuilder()
+                .setClientId(msg.payload().clientIdentifier())
+                .setUserName(userName);
+        String password = msg.payload().password();
+        if (password != null) {
+            request.setPassword(password);
         }
+        transportService.process(DeviceTransportType.MQTT, request.build(),
+                new TransportServiceCallback<ValidateDeviceCredentialsResponse>() {
+                    @Override
+                    public void onSuccess(ValidateDeviceCredentialsResponse msg) {
+                        onValidateDeviceResponse(msg, ctx);
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+                        log.trace("[{}] Failed to process credentials: {}", address, userName, e);
+                        ctx.writeAndFlush(createMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE));
+                        ctx.close();
+                    }
+                });
     }
     private void processX509CertConnect(ChannelHandlerContext ctx, X509Certificate cert) {
--- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java
View file @a3326b4
+++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java
View file @a3326b4
@@ -23,7 +23,6 @@ import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsRes
 import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayResponseMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.GetTenantRoutingInfoRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.GetTenantRoutingInfoResponseMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg;
@@ -35,7 +34,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToRPCMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionInfoProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg;
-import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceCredentialsResponseMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ValidateBasicMqttCredRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
@@ -49,6 +48,9 @@ public interface TransportService {
     void process(DeviceTransportType transportType, ValidateDeviceTokenRequestMsg msg,
                  TransportServiceCallback<ValidateDeviceCredentialsResponse> callback);
+    void process(DeviceTransportType transportType, ValidateBasicMqttCredRequestMsg msg,
+                 TransportServiceCallback<ValidateDeviceCredentialsResponse> callback);
+
     void process(DeviceTransportType transportType, ValidateDeviceX509CertRequestMsg msg,
                  TransportServiceCallback<ValidateDeviceCredentialsResponse> callback);
--- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View file @a3326b4
+++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View file @a3326b4
@@ -252,9 +252,20 @@ public class DefaultTransportService implements TransportService {
     }
     @Override
-    public void process(DeviceTransportType transportType, TransportProtos.ValidateDeviceTokenRequestMsg msg, TransportServiceCallback<ValidateDeviceCredentialsResponse> callback) {
+    public void process(DeviceTransportType transportType, TransportProtos.ValidateDeviceTokenRequestMsg msg,
+                        TransportServiceCallback<ValidateDeviceCredentialsResponse> callback) {
         log.trace("Processing msg: {}", msg);
-        TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setValidateTokenRequestMsg(msg).build());
+        TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(),
+                TransportApiRequestMsg.newBuilder().setValidateTokenRequestMsg(msg).build());
+        doProcess(transportType, protoMsg, callback);
+    }
+
+    @Override
+    public void process(DeviceTransportType transportType, TransportProtos.ValidateBasicMqttCredRequestMsg msg,
+                        TransportServiceCallback<ValidateDeviceCredentialsResponse> callback) {
+        log.trace("Processing msg: {}", msg);
+        TbProtoQueueMsg<TransportApiRequestMsg> protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(),
+                TransportApiRequestMsg.newBuilder().setValidateBasicMqttCredRequestMsg(msg).build());
         doProcess(transportType, protoMsg, callback);
     }
@@ -265,9 +276,10 @@ public class DefaultTransportService implements TransportService {
         doProcess(transportType, protoMsg, callback);
     }
-    private void doProcess(DeviceTransportType transportType, TbProtoQueueMsg<TransportApiRequestMsg> protoMsg, TransportServiceCallback<ValidateDeviceCredentialsResponse> callback) {
+    private void doProcess(DeviceTransportType transportType, TbProtoQueueMsg<TransportApiRequestMsg> protoMsg,
+                           TransportServiceCallback<ValidateDeviceCredentialsResponse> callback) {
         ListenableFuture<ValidateDeviceCredentialsResponse> response = Futures.transform(transportApiRequestTemplate.send(protoMsg), tmp -> {
-            TransportProtos.ValidateDeviceCredentialsResponseMsg msg = tmp.getValue().getValidateTokenResponseMsg();
+            TransportProtos.ValidateDeviceCredentialsResponseMsg msg = tmp.getValue().getValidateCredResponseMsg();
             ValidateDeviceCredentialsResponse.ValidateDeviceCredentialsResponseBuilder result = ValidateDeviceCredentialsResponse.builder();
             if (msg.hasDeviceInfo()) {
                 result.credentials(msg.getCredentialsBody());
--- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceCredentialsServiceImpl.java
View file @a3326b4
+++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceCredentialsServiceImpl.java
View file @a3326b4
@@ -21,18 +21,20 @@ import org.hibernate.exception.ConstraintViolationException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cache.annotation.CacheEvict;
 import org.springframework.cache.annotation.Cacheable;
+import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.device.credentials.BasicMqttCredentials;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
-import org.thingsboard.server.common.data.security.DeviceCredentialsType;
 import org.thingsboard.server.common.msg.EncryptionUtil;
 import org.thingsboard.server.dao.entity.AbstractEntityService;
 import org.thingsboard.server.dao.exception.DataValidationException;
 import org.thingsboard.server.dao.service.DataValidator;
+import org.thingsboard.server.dao.util.mapping.JacksonUtil;
 import static org.thingsboard.server.common.data.CacheConstants.DEVICE_CREDENTIALS_CACHE;
 import static org.thingsboard.server.dao.service.Validator.validateId;
@@ -75,8 +77,16 @@ public class DeviceCredentialsServiceImpl extends AbstractEntityService implemen
     }
     private DeviceCredentials saveOrUpdate(TenantId tenantId, DeviceCredentials deviceCredentials) {
-        if (deviceCredentials.getCredentialsType() == DeviceCredentialsType.X509_CERTIFICATE) {
-            formatCertData(deviceCredentials);
+        if(deviceCredentials.getCredentialsType() == null){
+            throw new DataValidationException("Device credentials type should be specified");
+        }
+        switch (deviceCredentials.getCredentialsType()) {
+            case X509_CERTIFICATE:
+                formatCertData(deviceCredentials);
+                break;
+            case MQTT_BASIC:
+                formatSimpleMqttCredentials(deviceCredentials);
+                break;
         }
         log.trace("Executing updateDeviceCredentials [{}]", deviceCredentials);
         credentialsValidator.validate(deviceCredentials, id -> tenantId);
@@ -93,6 +103,32 @@ public class DeviceCredentialsServiceImpl extends AbstractEntityService implemen
         }
     }
+    private void formatSimpleMqttCredentials(DeviceCredentials deviceCredentials) {
+        BasicMqttCredentials mqttCredentials;
+        try {
+            mqttCredentials = JacksonUtil.fromString(deviceCredentials.getCredentialsValue(), BasicMqttCredentials.class);
+            if (mqttCredentials == null) {
+                throw new IllegalArgumentException();
+            }
+        } catch (IllegalArgumentException e) {
+            throw new DataValidationException("Invalid credentials body for simple mqtt credentials!");
+        }
+        if (StringUtils.isEmpty(mqttCredentials.getClientId()) && StringUtils.isEmpty(mqttCredentials.getUserName())) {
+            throw new DataValidationException("Both mqtt client id and user name are empty!");
+        }
+        if (StringUtils.isEmpty(mqttCredentials.getClientId())) {
+            deviceCredentials.setCredentialsId(mqttCredentials.getUserName());
+        } else if (StringUtils.isEmpty(mqttCredentials.getUserName())) {
+            deviceCredentials.setCredentialsId(EncryptionUtil.getSha3Hash(mqttCredentials.getClientId()));
+        } else {
+            deviceCredentials.setCredentialsId(EncryptionUtil.getSha3Hash("|", mqttCredentials.getClientId(), mqttCredentials.getUserName()));
+        }
+        if (!StringUtils.isEmpty(mqttCredentials.getPassword())) {
+            mqttCredentials.setPassword(mqttCredentials.getPassword());
+        }
+        deviceCredentials.setCredentialsValue(JacksonUtil.toString(mqttCredentials));
+    }
+
     private void formatCertData(DeviceCredentials deviceCredentials) {
         String cert = EncryptionUtil.trimNewLines(deviceCredentials.getCredentialsValue());
         String sha3Hash = EncryptionUtil.getSha3Hash(cert);