Merge pull request #1468 from openWB/feature-partner-id

Feature partner id

Author: benderl

packages/helpermodules/setdata.py

@@ -273,31 +273,30 @@ def _validate_collection_value(self, msg: mqtt.MQTTMessage, data_type, ranges=No
             Broker-Nachricht
         data_type: float, int
             Datentyp, den die Liste enthalten soll
-        min_value: int/float
-            Minimalwert, den die Elemente in der Liste nicht unterschreiten dürfen
-        max_value= int/float
-            Maximalwert, den die Elemente in der Liste nicht überschreiten dürfen
-        collection = list/dict
-            Angabe, ob und welche Kollektion erwartet wird
+        ranges: tuple, optional
+            (min_value, max_value), die die Minimal- und Maximalwerte angeben
+        collection: type, optional
+            Angabe, ob und welche Kollektion erwartet wird (list oder dict)
         """
         try:
             valid = False
             value = decode_payload(msg.payload)
-            if isinstance(value, list):
-                for item in value:
-                    if not self._validate_min_max_value(item, msg, data_type, ranges):
-                        break
-                else:
-                    valid = True
-            elif isinstance(value, dict):
-                for item in value.values():
-                    if not self._validate_min_max_value(item, msg, data_type, ranges):
-                        break
-                else:
-                    valid = True
-            else:
-                log.error("Payload ungültig: Topic "+str(msg.topic)+", Payload " +
-                          str(value)+" sollte eine Kollektion vom Typ "+str(collection)+" sein.")
+            if collection is not None and isinstance(value, collection):
+                if isinstance(value, list):
+                    if ranges is not None:
+                        valid = all(self._validate_min_max_value(item, msg, data_type, ranges) for item in value)
+                    else:
+                        valid = all(isinstance(item, data_type) for item in value)
+                elif isinstance(value, dict):
+                    if ranges is not None:
+                        valid = all(
+                            self._validate_min_max_value(item, msg, data_type, ranges) for item in value.values())
+                    else:
+                        valid = all(isinstance(item, data_type) for item in value.values())
+            if not valid:
+                log.error(f"Payload ungültig: Topic '{msg.topic}', Payload '{value}' "
+                          f"sollte eine Kollektion vom Typ {collection} sein "
+                          f"und nur Elemente vom Typ {data_type} enthalten.")
             return valid
         except Exception:
             log.exception(f"Fehler im setdata-Modul: Topic {msg.topic}, Value: {msg.payload}")
@@ -1010,6 +1009,8 @@ def process_system_topic(self, msg: mqtt.MQTTMessage):
                 self._validate_value(msg, str)
             elif "openWB/set/system/mqtt/bridge/" in msg.topic:
                 self._validate_value(msg, "json")
+            elif "openWB/set/system/mqtt/valid_partner_ids" == msg.topic:
+                self._validate_value(msg, str, collection=list)
             elif "configurable" in msg.topic:
                 self._validate_value(msg, None)
             elif "device" in msg.topic:

packages/helpermodules/subdata.py

@@ -135,6 +135,7 @@ def on_connect(self, client: mqtt.Client, userdata, flags: dict, rc: int):
             # MQTT Bridge Topics vor "openWB/system/+" abonnieren, damit sie auch vor
             # "openWB/system/subdata_initialized" empfangen werden!
             ("openWB/system/mqtt/bridge/+", 2),
+            ("openWB/system/mqtt/+", 2),
             # Nicht mit hash # abonnieren, damit nicht die Komponenten vor den Devices empfangen werden!
             ("openWB/system/+", 2),
             ("openWB/system/backup_cloud/#", 2),
@@ -763,6 +764,10 @@ def process_system_topic(self, client: mqtt.Client, var: dict, msg: mqtt.MQTTMes
                                            MessageType.SUCCESS if result.returncode == 0 else MessageType.ERROR)
                 else:
                     log.debug("skipping mqtt bridge message on startup")
+            elif "mqtt" and "valid_partner_ids" in msg.topic:
+                # duplicate topic for remote support service
+                log.error(f"received valid partner ids: {decode_payload(msg.payload)}")
+                Pub().pub("openWB-remote/valid_partner_ids", decode_payload(msg.payload))
             # will be moved to separate handler!
             elif "GetRemoteSupport" in msg.topic:
                 log.warning("deprecated topic for remote support received!")

packages/helpermodules/update_config.py

@@ -392,6 +392,7 @@ class UpdateConfig:
         "^openWB/system/lastlivevaluesJson$",
         "^openWB/system/messages/[0-9]+$",
         "^openWB/system/mqtt/bridge/[0-9]+$",
+        "^openWB/system/mqtt/valid_partner_ids$",
         "^openWB/system/release_train$",
         "^openWB/system/time$",
         "^openWB/system/update_in_progress$",
@@ -474,6 +475,7 @@ class UpdateConfig:
         ("openWB/system/debug_level", 30),
         ("openWB/system/device/module_update_completed", True),
         ("openWB/system/ip_address", "unknown"),
+        ("openWB/system/mqtt/valid_partner_ids", []),
         ("openWB/system/release_train", "master"),
     )
     invalid_topic = (

runs/remoteSupport/remoteSupport.py

@@ -1,21 +1,30 @@
 #!/usr/bin/env python3
 import logging
 import re
+import json
 from subprocess import Popen
 from pathlib import Path
+from time import sleep
+from typing import Optional
 import paho.mqtt.client as mqtt
 import platform
 
+API_VERSION = "1"
 BASE_PATH = Path(__file__).resolve().parents[2]
 RAMDISK_PATH = BASE_PATH / "ramdisk"
 RUNS_PATH = BASE_PATH / "runs"
 BASE_TOPIC = "openWB-remote/"
+API_TOPIC = BASE_TOPIC + "api_version"
+STATE_TOPIC = BASE_TOPIC + "connection_state"
 REMOTE_SUPPORT_TOPIC = BASE_TOPIC + "support"
 REMOTE_PARTNER_TOPIC = BASE_TOPIC + "partner"
+REMOTE_PARTNER_IDS_TOPIC = BASE_TOPIC + "valid_partner_ids"
 CLOUD_TOPIC = BASE_TOPIC + "cloud"
+
 support_tunnel: Popen = None
 partner_tunnel: Popen = None
 cloud_tunnel: Popen = None
+valid_partner_ids: list[str] = []
 logging.basicConfig(
     filename=str(RAMDISK_PATH / "remote_support.log"),
     level=logging.DEBUG, format='%(asctime)s: %(message)s'
@@ -32,10 +41,43 @@ def get_serial():
         return "0000000000000000"
 
 
+def publish_as_json(client: mqtt.Client, topic: str, str_payload: str, qos: int = 0, retain: bool = False,
+                    properties: Optional[mqtt.Properties] = None) -> mqtt.MQTTMessageInfo:
+    return client.publish(topic, json.dumps(str_payload), qos, retain, properties)
+
+
+def get_lt_executable() -> Optional[Path]:
+    machine = platform.machine()
+    bits, linkage = platform.architecture()
+    lt_executable = f"lt-{machine}_{linkage}"
+
+    log.debug("System Info:")
+    log.debug(f"Architecture: ({(bits, linkage)})")
+    log.debug(f"Machine: {machine}")
+    log.debug(f"Node: {platform.node()}")
+    log.debug(f"Platform: {platform.platform()}")
+    log.debug(f"System: {platform.system()}")
+    log.debug(f"Release: {platform.release()}")
+    log.debug(f"using binary: '{lt_executable}'")
+
+    lt_path = RUNS_PATH / lt_executable
+    if not lt_path.is_file():
+        log.error(f"file '{lt_executable}' does not exist!")
+        return None
+    return lt_path
+
+
 def on_connect(client: mqtt.Client, userdata, flags: dict, rc: int):
     """connect to broker and subscribe to set topics"""
     log.info("Connected")
-    client.subscribe(BASE_TOPIC + "#", 2)
+    client.subscribe([
+        (REMOTE_SUPPORT_TOPIC, 2),
+        (CLOUD_TOPIC, 2),
+        (REMOTE_PARTNER_TOPIC, 2),
+        (REMOTE_PARTNER_IDS_TOPIC, 2)
+    ])
+    publish_as_json(client, API_TOPIC, API_VERSION, qos=2, retain=True)
+    publish_as_json(client, STATE_TOPIC, "online", qos=2, retain=True)
 
 
 def on_message(client: mqtt.Client, userdata, msg: mqtt.MQTTMessage):
@@ -56,6 +98,8 @@ def is_tunnel_closed(tunnel: Popen) -> bool:
     global support_tunnel
     global partner_tunnel
     global cloud_tunnel
+    global valid_partner_ids
+    clear_topic = False
     payload = msg.payload.decode("utf-8")
     if len(payload) > 0:
         log.debug("Topic: %s, Message: %s", msg.topic, payload)
@@ -81,6 +125,9 @@ def is_tunnel_closed(tunnel: Popen) -> bool:
                     log.info(f"tunnel running with pid {support_tunnel.pid}")
             else:
                 log.info("unknown message: " + payload)
+            clear_topic = True
+        elif msg.topic == REMOTE_PARTNER_IDS_TOPIC:
+            valid_partner_ids = json.loads(payload)
         elif msg.topic == REMOTE_PARTNER_TOPIC:
             if payload == 'stop':
                 if partner_tunnel is None:
@@ -90,22 +137,39 @@ def is_tunnel_closed(tunnel: Popen) -> bool:
                     partner_tunnel.terminate()
                     partner_tunnel.wait(timeout=3)
                     partner_tunnel = None
-            elif re.match(r'^([^;]+)(?:;([1-9][0-9]+)(?:;([a-zA-Z0-9]+))?)?$', payload):
+            elif re.match(r'^([^;]+)(?:;((?:cnode)?[0-9]+)(?:;([\wäöüÄÖÜ-]+))?)?$', payload):
                 if is_tunnel_closed(partner_tunnel):
                     splitted = payload.split(";")
                     if len(splitted) != 3:
                         log.error("invalid number of settings received!")
                     else:
                         token = splitted[0]
-                        port = splitted[1]
-                        user = splitted[2]
-                        log.info("start partner support")
-                        partner_tunnel = Popen(["sshpass", "-p", token, "ssh", "-N", "-tt", "-o",
-                                                "StrictHostKeyChecking=no", "-o", "ServerAliveInterval 60", "-R",
-                                                f"{port}:localhost:80", f"{user}@partner.openwb.de"])
-                        log.info(f"tunnel running with pid {partner_tunnel.pid}")
+                        port_or_node = splitted[1]
+                        user = splitted[2]  # not used in v0, partner-id in v1
+                        if port_or_node.isdecimal():
+                            # v0
+                            log.info("start partner support")
+                            partner_tunnel = Popen(["sshpass", "-p", token, "ssh", "-N", "-tt", "-o",
+                                                    "StrictHostKeyChecking=no", "-o", "ServerAliveInterval 60", "-R",
+                                                    f"{port_or_node}:localhost:80", f"{user}@partner.openwb.de"])
+                            log.info(f"tunnel running with pid {partner_tunnel.pid}")
+                        else:
+                            # v1
+                            if lt_executable is None:
+                                log.error("start partner tunnel requested but lt executable not found!")
+                            else:
+                                if user in valid_partner_ids:
+                                    log.info("start partner support v1")
+                                    if lt_executable is not None:
+                                        partner_tunnel = Popen([f"{lt_executable}", "-h",
+                                                                "https://" + port_or_node + ".openwb.de/",
+                                                                "-p", "80", "-s", token])
+                                        log.info(f"tunnel running with pid {partner_tunnel.pid}")
+                                else:
+                                    log.error(f"invalid partner-id: {user}")
             else:
                 log.info("unknown message: " + payload)
+            clear_topic = True
         elif msg.topic == CLOUD_TOPIC:
             if payload == 'stop':
                 if cloud_tunnel is None:
@@ -125,37 +189,44 @@ def is_tunnel_closed(tunnel: Popen) -> bool:
                         cloud_node = splitted[1]
                         user = splitted[2]
 
-                        machine = platform.machine()
-                        bits, linkage = platform.architecture()
-                        lt_executable = f"lt-{machine}_{linkage}"
-
-                        log.debug("System Info:")
-                        log.debug(f"Architecture: ({(bits, linkage)})")
-                        log.debug(f"Machine: {machine}")
-                        log.debug(f"Node: {platform.node()}")
-                        log.debug(f"Platform: {platform.platform()}")
-                        log.debug(f"System: {platform.system()}")
-                        log.debug(f"Release: {platform.release()}")
-                        log.debug(f"using binary: '{lt_executable}'")
-
-                        log.info(f"start cloud tunnel '{token[:4]}...{token[-4:]}' on '{cloud_node}'")
-                        try:
-                            cloud_tunnel = Popen([f"{RUNS_PATH}/{lt_executable}", "-h",
+                        if lt_executable is None:
+                            log.error("start cloud tunnel requested but lt executable not found!")
+                        else:
+                            log.info(f"start cloud tunnel '{token[:4]}...{token[-4:]}' on '{cloud_node}'")
+                            cloud_tunnel = Popen([f"{lt_executable}", "-h",
                                                   "https://" + cloud_node + ".openwb.de/", "-p", "80", "-s", token])
                             log.info(f"cloud tunnel running with pid {cloud_tunnel.pid}")
-                        except FileNotFoundError:
-                            log.exception(f"executable '{lt_executable}' does not exist!")
             else:
                 log.info("unknown message: " + payload)
+            clear_topic = True
         # clear topic
-        client.publish(msg.topic, "", qos=2, retain=True)
+        if clear_topic and msg.retain:
+            client.publish(msg.topic, "", qos=2, retain=True)
 
 
+lt_executable = get_lt_executable()
 mqtt_broker_host = "localhost"
 client = mqtt.Client("openWB-remote-" + get_serial())
 client.on_connect = on_connect
 client.on_message = on_message
+client.will_set(STATE_TOPIC, json.dumps("offline"), qos=2, retain=True)
 
+log.debug("connecting to broker")
 client.connect(mqtt_broker_host, 1883)
-client.loop_forever()
-client.disconnect()
+log.debug("starting loop")
+client.loop_start()
+try:
+    while True:
+        sleep(1)
+except (Exception, KeyboardInterrupt) as e:
+    log.debug(e)
+    log.debug("terminated")
+finally:
+    log.debug("publishing state 'offline'")
+    publish_as_json(client, STATE_TOPIC, "offline", qos=2, retain=True)
+    sleep(0.5)
+    log.debug("stopping loop")
+    client.loop_stop()
+    client.disconnect()
+    log.debug("disconnected")
+log.debug("exit")

runs/save_mqtt.php

@@ -178,6 +178,8 @@ function cleanAndExit($message)
 		<<<EOS
 		topic openWB/system/time out 2 "" {$configuration->remote->prefix}
 		topic openWB-remote/support both 2 "" {$configuration->remote->prefix}
+		topic openWB-remote/api_version out 2 "" {$configuration->remote->prefix}
+		topic openWB-remote/connection_state out 2 "" {$configuration->remote->prefix}
 		
 		EOS
 	);
@@ -199,6 +201,7 @@ function cleanAndExit($message)
 			$configFile,
 			<<<EOS
 			topic openWB-remote/partner both 2 "" {$configuration->remote->prefix}
+			topic openWB-remote/valid_partner_ids out 2 "" {$configuration->remote->prefix}
 			
 			EOS
 		);