Added Remote shell

thingsboard_gateway/gateway/tb_gateway_service.py

@@ -39,6 +39,7 @@ from thingsboard_gateway.gateway.tb_logger import TBLoggerHandler
 from thingsboard_gateway.storage.memory_event_storage import MemoryEventStorage
 from thingsboard_gateway.storage.file_event_storage import FileEventStorage
 from thingsboard_gateway.gateway.tb_gateway_remote_configurator import RemoteConfigurator
+from thingsboard_gateway.gateway.tb_remote_shell import RemoteShell
 
 
 
@@ -115,8 +116,10 @@ class TBGatewayService:
             "update": self.__rpc_update,
             "version": self.__rpc_version,
         }
+        self.__remote_shell = RemoteShell(platform=self.__updater.get_platform(), release=self.__updater.get_release())
+        self.__rpc_remote_shell_command_in_progress = None
         self.__sheduled_rpc_calls = []
-        self.__self_rpc_sheduled_methods_functions = {
+        self.__rpc_sheduled_methods_functions = {
             "restart": {"function": execv, "arguments": (executable, [executable.split(pathsep)[-1]] + argv)},
             "reboot": {"function": system, "arguments": ("reboot 0",)},
             }
@@ -470,13 +473,15 @@ class TBGatewayService:
                                 if self.available_connectors[connector_name]._connector_type == module:
                                     log.debug("Sending command RPC %s to connector %s", content["method"], connector_name)
                                     result = self.available_connectors[connector_name].server_side_rpc_handler(content)
-                        elif module == 'gateway':
+                        elif module == 'gateway' or module in self.__remote_shell.shell_commands:
                             result = self.__rpc_gateway_processing(request_id, content)
                         else:
                             log.error("Connector \"%s\" not found", module)
                             result = {"error": "%s - connector not found in available connectors." % module, "code": 404}
                         if result is None:
                             self.send_rpc_reply(None, request_id, success_sent=False)
+                        elif "qos" in result:
+                            self.send_rpc_reply(None, request_id, dumps({k: v for k, v in result.items() if k != "qos"}), quality_of_service=result["qos"])
                         else:
                             self.send_rpc_reply(None, request_id, dumps(result))
                 except Exception as e:
@@ -487,24 +492,29 @@ class TBGatewayService:
 
     def __rpc_gateway_processing(self, request_id, content):
         log.info("Received RPC request to the gateway, id: %s, method: %s", str(request_id), content["method"])
-        arguments = content.get('params')
+        arguments = content.get('params', {})
+        if content.get("timeout") is not None:
+            arguments.update({"timeout": content["timeout"]})
         method_to_call = content["method"].replace("gateway_", "")
         result = None
-        if isinstance(arguments, list):
-            result = self.__gateway_rpc_methods[method_to_call](*arguments)
-        elif method_to_call in self.__self_rpc_sheduled_methods_functions:
+        method_function = self.__remote_shell.shell_commands.get(method_to_call, self.__gateway_rpc_methods.get(method_to_call))
+        if method_function is None and method_to_call in self.__rpc_sheduled_methods_functions:
             seconds_to_restart = arguments*1000 if arguments and arguments != '{}' else 0
-            self.__sheduled_rpc_calls.append([time()*1000 + seconds_to_restart, self.__self_rpc_sheduled_methods_functions[method_to_call]])
-            log.info("Gateway %s sheduled in %i seconds", method_to_call, seconds_to_restart/1000)
+            self.__sheduled_rpc_calls.append([time() * 1000 + seconds_to_restart, self.__rpc_sheduled_methods_functions[method_to_call]])
+            log.info("Gateway %s scheduled in %i seconds", method_to_call, seconds_to_restart/1000)
             result = {"success": True}
-        elif arguments is not None:
-            result = self.__gateway_rpc_methods[method_to_call](arguments)
+        elif method_function is None:
+            return {"error": "Method not found", "code": 404}
+        elif isinstance(arguments, list):
+            result = method_function(*arguments)
+        elif arguments:
+            result = method_function(arguments)
         else:
-            result = self.__gateway_rpc_methods[method_to_call]()
-        log.debug(result)
+            result = method_function()
         return result
 
-    def __rpc_ping(self, *args):
+    @staticmethod
+    def __rpc_ping(*args):
         return {"code": 200, "resp": "pong"}
 
     def __rpc_devices(self, *args):
@@ -643,8 +653,6 @@ class TBGatewayService:
                     if self.available_connectors.get(devices[device_name]):
                         self.__connected_devices[device_name] = {
                             "connector": self.available_connectors[devices[device_name]]}
-                    else:
-                        log.info("Pair device %s - connector %s from persistent device storage - not found.", device_name, devices[device_name])
                 except Exception as e:
                     log.exception(e)
                     continue

thingsboard_gateway/gateway/tb_remote_shell.py

@@ -0,0 +1,114 @@
+#      Copyright 2020. ThingsBoard
+#  #
+#      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.
+#
+
+from os import getcwd, chdir
+from subprocess import Popen, PIPE, STDOUT, TimeoutExpired
+from logging import getLogger
+from queue import Queue
+
+log = getLogger("service")
+
+
+class RemoteShell:
+    def __init__(self, platform, release):
+        self.__session_active = False
+        self.__platform = platform
+        self.__release = release
+        self.shell_commands = {
+            "getTermInfo": self.get_term_info,
+            "sendCommand": self.send_command,
+            "getCommandStatus": self.get_command_status,
+            "terminateCommand": self.terminate_command,
+        }
+        self.command_in_progress = None
+        self.__previous_stdout = b""
+        self.__previous_stderr = b""
+
+    def get_term_info(self, *args):
+        return {"platform": self.__platform, "release": self.__release, "cwd": str(getcwd())}
+
+    def send_command(self, *args):
+        result = {"ok": False, "qos": 0}
+        log.debug("Received command to shell with args: %r", args)
+        command = args[0]['command']
+        cwd = args[0].get('cwd')
+        if cwd is not None and str(getcwd()) != cwd:
+            chdir(cwd)
+        if command.split():
+            if self.command_in_progress is not None:
+                log.debug("Received a new command: \"%s\", during old command is running, terminating old command...", command)
+                old_command = self.command_in_progress.args
+                self.terminate_command()
+                log.debug("Old command: \"%s\" terminated.", old_command)
+            if command.split()[0] in ["quit", "exit"]:
+                self.command_in_progress = None
+            elif command.split()[0] == "cd":
+                chdir(command.split()[1])
+                self.command_in_progress = "cd"
+            else:
+                log.debug("Run command in remote shell: %s", command)
+                self.command_in_progress = Popen(command, shell=True, stdout=PIPE, stdin=PIPE, stderr=STDOUT, universal_newlines=True)
+            result.update({"ok": True})
+        return result
+
+    def get_command_status(self, *args):
+        result = {"data": [{"stdout": "",
+                            "stderr": ""}],
+                  "cwd": str(getcwd()),
+                  "done": True,
+                  "qos": 0,
+                  }
+        done = False
+        if self.command_in_progress == "cd":
+            done = True
+        elif self.command_in_progress is not None:
+            stdout_value = b""
+            stderr_value = b""
+            done = True if self.command_in_progress.poll() is not None else False
+            try:
+                stdout_value, stderr_value = self.command_in_progress.communicate(timeout=.1)
+            except TimeoutExpired as e:
+                log.debug("Process is run")
+                stdout_value = b"" if e.stdout is None else e.stdout.replace(self.__previous_stdout, b"")
+                stderr_value = b"" if e.stderr is None else e.stderr.replace(self.__previous_stderr, b"")
+                stdout_value = stdout_value[:-1] if len(stdout_value)>0 and stdout_value[-1] == b"\n" else stdout_value
+                stderr_value = stderr_value[:-1] if len(stderr_value)>0 and stderr_value[-1] == b"\n" else stderr_value
+            self.__previous_stderr = self.__previous_stderr + stderr_value
+            self.__previous_stdout = self.__previous_stdout + stdout_value
+            str_stdout = str(stdout_value, "UTF-8") if isinstance(stdout_value, bytes) else stdout_value
+            str_stderr = str(stderr_value, "UTF-8") if isinstance(stderr_value, bytes) else stderr_value
+            result.update({"data": [{"stdout": str_stdout,
+                                     "stderr": str_stderr}]
+                           })
+        result.update({"done": done})
+        if done:
+            self.command_in_progress = None
+
+        return result
+
+    def terminate_command(self, *args):
+        result = {"ok": False}
+        if self.command_in_progress is not None:
+            try:
+                self.command_in_progress.terminate()
+                self.__previous_stderr = b""
+                self.__previous_stdout = b""
+                result.update({"ok": True})
+            except Exception as e:
+                log.exception(e)
+                result["error"] = str(e)
+        else:
+            result["error"] = "Process for termination not found."
+        return result

thingsboard_gateway/gateway/tb_updater.py

@@ -14,7 +14,7 @@
 
 from requests import post
 from uuid import uuid1
-from platform import platform
+from platform import platform, system, release
 from logging import getLogger
 from pkg_resources import get_distribution
 from threading import Thread
@@ -35,7 +35,8 @@ class TBUpdater(Thread):
         self.__version = {"current_version": get_distribution('thingsboard_gateway').version,
                           "latest_version": get_distribution('thingsboard_gateway').version}
         self.__instance_id = str(uuid1())
-        self.__platform = "deb"
+        self.__platform = system()
+        self.__release = release()
         self.__os_version = platform()
         self.__previous_check = 0
         self.__check_period = 3600.0
@@ -58,6 +59,12 @@ class TBUpdater(Thread):
     def get_version(self):
         return self.__version
 
+    def get_platform(self):
+        return self.__platform
+
+    def get_release(self):
+        return self.__release
+
     def check_for_new_version(self):
         log.debug("Checking for new version")
         request_args = self.form_request_params()