feat/target_site_id

hivemind_bus_client/message.py

@@ -26,7 +26,7 @@ class HiveMessageType(str, Enum):
 
 class HiveMessage:
     def __init__(self, msg_type, payload=None, node=None, source_peer=None,
-                 route=None, target_peers=None, meta=None):
+                 route=None, target_peers=None, meta=None, target_site_id=None):
         #  except for the hivemind node classes receiving the message and
         #  creating the object nothing should be able to change these values
         #  node classes might change them a runtime by the private attribute
@@ -50,12 +50,17 @@ def __init__(self, msg_type, payload=None, node=None, source_peer=None,
             payload = json.loads(payload)
         self._payload = payload or {}
 
+        self._site_id = target_site_id
         self._node = node  # node semi-unique identifier
         self._source_peer = source_peer  # peer_id
         self._route = route or []  # where did this message come from
         self._targets = target_peers or []  # where will it be sent
         self._meta = meta or {}
 
+    @property
+    def target_site_id(self):
+        return self._site_id
+
     @property
     def msg_type(self):
         return self._msg_type
@@ -104,6 +109,7 @@ def as_dict(self):
                 "payload": pload,
                 "route": self.route,
                 "node": self.node_id,
+                "target_site_id": self.target_site_id,
                 "source_peer": self.source_peer}
 
     @property
@@ -120,19 +126,22 @@ def deserialize(payload):
 
         if "msg_type" in payload:
             try:
-                return HiveMessage(payload["msg_type"], payload["payload"])
+                return HiveMessage(payload["msg_type"], payload["payload"],
+                                   target_site_id=payload.get("target_site_id"))
             except:
                 pass  # not a hivemind message
 
         if "type" in payload:
             try:
                 # NOTE: technically could also be SHARED_BUS or THIRDPRTY
                 return HiveMessage(HiveMessageType.BUS,
-                                   Message.deserialize(payload))
+                                   Message.deserialize(payload),
+                                   target_site_id=payload.get("target_site_id"))
             except:
                 pass  # not a mycroft message
 
-        return HiveMessage(HiveMessageType.THIRDPRTY, payload)
+        return HiveMessage(HiveMessageType.THIRDPRTY, payload,
+                           target_site_id=payload.get("target_site_id"))
 
     def __getitem__(self, item):
         return self._payload.get(item)

hivemind_bus_client/protocol.py

@@ -36,8 +36,7 @@ def handle_send(self, message: Message):
         payload = message.data.get("payload")
         msg_type = message.data["msg_type"]
 
-        hmessage = HiveMessage(msg_type,
-                               payload=payload)
+        hmessage = HiveMessage(msg_type, payload=payload)
 
         if msg_type == HiveMessageType.BROADCAST:
             # only masters can broadcast, ignore silently
@@ -216,6 +215,16 @@ def handle_bus(self, message: HiveMessage):
 
     def handle_broadcast(self, message: HiveMessage):
         LOG.info(f"BROADCAST: {message.payload}")
+
+        # if the message targets our site_id, send it to internal bus
+        site = message.target_site_id
+        if site and site == self.site_id:
+            pload = message.payload
+            # broadcast messages always come from a trusted source
+            # only masters can emit them
+            if isinstance(pload, MycroftMessage):
+                self.handle_bus(message)
+
         # if this device is also a hivemind server
         # forward to HiveMindListenerInternalProtocol
         data = message.serialize()
@@ -224,6 +233,17 @@ def handle_broadcast(self, message: HiveMessage):
 
     def handle_propagate(self, message: HiveMessage):
         LOG.info(f"PROPAGATE: {message.payload}")
+
+        # if the message targets our site_id, send it to internal bus
+        site = message.target_site_id
+        if site and site == self.site_id:
+           # might originate from untrusted
+           # satellite anywhere in the hive
+           # do not inject by default
+           pload = message.payload
+           #if isinstance(pload, MycroftMessage):
+           #    self.handle_bus(message)
+
         # if this device is also a hivemind server
         # forward to HiveMindListenerInternalProtocol
         data = message.serialize()