NOISSUE - Add MQTT UserName check on register and InstanceId in Redis (#884)

* Add InstanceId and verify UserName

Signed-off-by: drasko <drasko.draskovic@gmail.com>

* Fix thingID check (compare with UserName)

Signed-off-by: Drasko DRASKOVIC <drasko.draskovic@gmail.com>

mqtt/verne/src/mfx_auth.erl

@@ -65,8 +65,14 @@ auth_on_register({_IpAddr, _Port} = Peer, {_MountPoint, _ClientId} = SubscriberI
     %% 5. return {error, whatever} -> CONNACK_AUTH is sent
 
     case identify(Password) of
-        {ok, _Id} ->
-            ok;
+        {ok, Id} ->
+            case Id of
+                UserName ->
+                    ok;
+                _ ->
+                    error_logger:info_msg("auth_on_register failed - Id: ~p, Username: ~p", [Id, UserName]),
+                    {error, invalid_credentials}
+            end;
         Other ->
             Other
     end.
@@ -142,11 +148,13 @@ auth_on_subscribe(UserName, ClientId, [{Topic, _QoS}|_] = Topics) ->
 %%% Redis ES
 publish_event(UserName, Type) ->
     Timestamp = os:system_time(second),
+    [{_, InstanceId}] = ets:lookup(mfx_cfg, instance_id),
     KeyValuePairs = [
         "mainflux.mqtt", "*",
         "thing_id", binary_to_list(UserName),
         "timestamp", integer_to_list(Timestamp),
-        "event_type", Type
+        "event_type", Type,
+        "instance", InstanceId
     ],
     mfx_redis:publish(KeyValuePairs).
 

mqtt/verne/src/mfx_auth_app.erl

@@ -26,11 +26,17 @@ start(_StartType, _StartArgs) ->
         false -> "tcp://localhost:6379";
         RedisEnv -> RedisEnv
     end,
+    InstanceId = case os:getenv("MF_MQTT_INSTANCE_ID") of
+        false -> "";
+        InstanceEnv -> InstanceEnv
+    end,
+    
 
     ets:insert(mfx_cfg, [
         {grpc_url, GrpcUrl},
         {nats_url, NatsUrl},
-        {redis_url, RedisUrl}
+        {redis_url, RedisUrl},
+        {instance_id, InstanceId}
     ]),
 
     % Also, init one ETS table for keeping the #{ClientId => Username} mapping

mqtt/verne/src/mfx_grpc.erl

@@ -45,7 +45,7 @@ handle_call({identify, Message}, _From, #state{conn = GrpcConn} = State) ->
 
           case HttpStatus of
               200 ->
-                  {reply, {ok, ThingId}, State};
+                  {reply, {ok, list_to_binary(ThingId)}, State};
               _ ->
                   {reply, {error, HttpStatus}, error}
           end;