ignore PKI for testmsg as well

Author: andyshinn

bridger/cogs/testmsg.py

@@ -2,6 +2,7 @@
 import re
 from datetime import datetime
 from functools import partial
+from typing import Optional
 
 import aiomqtt
 from aiocache import SimpleMemoryCache
@@ -17,6 +18,7 @@
 from bridger.influx.interfaces import InfluxReader
 from bridger.log import logger
 from bridger.mqtt import PBPacketProcessor
+from bridger.utils import should_ignore_pki_message
 
 MQTT_TEST_CHANNEL_MESHTASTIC = os.getenv("MQTT_TEST_CHANNEL", "+")
 MQTT_TEST_CHANNEL_DISCORD = int(os.getenv("MQTT_TEST_CHANNEL_ID", 1253788609316913265))
@@ -43,7 +45,7 @@ async def on_ready(self):
         logger.info(f"TestMsg cog is ready and channel is: {self.discord_channel}")
 
     @staticmethod
-    def format_node_name(node_id: int, node_info: dict = None) -> str:
+    def format_node_name(node_id: int, node_info: Optional[dict] = None) -> str:
         """Format a consistent node name based on available info"""
         if not node_info:
             return f"**{node_id}**"
@@ -71,14 +73,17 @@ def create_embed(self, service_envelope: ServiceEnvelope):
 
         embed = Embed(color=color)
         # Try to get node info for the gateway hex ID
+        gateway_id = None
+        node_info = None
+
         try:
             gateway_id = int(gateway.strip("!"), 16)
             node_info = self.influx_reader.get_node_info(gateway_id)
         except (ValueError, TypeError) as e:
             logger.error(f"Failed to parse gateway ID '{gateway}': {e}")
-            node_info = None
 
-        gateway_name = self.format_node_name(gateway_id if "gateway_id" in locals() else gateway, node_info)
+        node_id = gateway_id if gateway_id is not None else int(gateway, 16)
+        gateway_name = self.format_node_name(node_id, node_info)
         embed.description = f"Heard by {gateway_name} - `{gateway}` at {formatted_time}"
         embed.add_field(name="SNR", value=snr, inline=True)
         embed.add_field(name="RSSI", value=rssi, inline=True)
@@ -112,7 +117,7 @@ async def run_mqtt(self):
         logger.info(f"Attempting to connect to MQTT broker at {MQTT_BROKER}:{MQTT_PORT}")
         async with aiomqtt.Client(
             MQTT_BROKER,
-            MQTT_PORT,
+            int(MQTT_PORT),
             username=MQTT_USER,
             password=MQTT_PASS,
             clean_session=True,
@@ -121,9 +126,16 @@ async def run_mqtt(self):
             logger.info(f"Subscribed to {full_topic}")
             await logger.complete()
 
-            async for message in client.messages:
+            async for mqtt_message in client.messages:
+                # Ignoring PKI messages for now as we cannot decrypt them without storing keys somewhere
+                if should_ignore_pki_message(str(mqtt_message.topic)):
+                    logger.bind(topic=topic, channel=channel, *mqtt_message.properties).debug(
+                        f"Ignoring PKI message on topic {mqtt_message.topic}"
+                    )  # noqa: E501
+                    continue
+
                 try:
-                    service_envelope = ServiceEnvelope.FromString(message.payload)
+                    service_envelope = ServiceEnvelope.FromString(mqtt_message.payload)
                 except Exception:
                     logger.exception("Failed to decode MQTT message")
                     continue

bridger/mqtt.py

@@ -11,6 +11,7 @@
 from bridger.influx.interfaces import InfluxWriter
 from bridger.log import logger
 from bridger.mesh import PacketProcessorError, PBPacketProcessor
+from bridger.utils import should_ignore_pki_message
 
 
 class BridgerMQTT(Client):
@@ -45,9 +46,8 @@ def on_message(self, client, userdata, message):
         add_breadcrumb(level="info", data=breadcrumb_data, category="mqtt", message="Received message")
 
         # Ignoring PKI messages for now as we cannot decrypt them without storing keys somewhere
-        pki_topic = MQTT_TOPIC.removesuffix("/#") + "/PKI/"
-        if message.topic.startswith(pki_topic):
-            logger.bind(**breadcrumb_data).debug("Ignoring PKI message")
+        if should_ignore_pki_message(message.topic):
+            logger.bind(**breadcrumb_data).debug(f"Ignoring PKI message on topic {message.topic}")
             return
 
         try:

bridger/utils.py

@@ -0,0 +1,6 @@
+from bridger.config import MQTT_TOPIC
+
+
+def should_ignore_pki_message(topic: str) -> bool:
+    pki_topic = MQTT_TOPIC.removesuffix("/#") + "/PKI/"
+    return topic.startswith(pki_topic)

tests/test_testmsg.py

@@ -169,3 +169,38 @@ def test_integration_deduplication_prevents_true_duplicates(self, testmsg_cog):
 
         should_process_second = testmsg_cog.deduplicator.should_process(envelope2)
         assert not should_process_second
+
+
+class TestFormatNodeName:
+    def test_format_node_name_no_node_info(self):
+        result = TestMsg.format_node_name(123456789, None)
+        assert result == "**123456789**"
+
+    def test_format_node_name_empty_node_info(self):
+        result = TestMsg.format_node_name(123456789, {})
+        assert result == "**123456789**"
+
+    def test_format_node_name_with_short_and_long(self):
+        node_info = {"short_name": "ABC", "long_name": "Station ABC"}
+        result = TestMsg.format_node_name(123456789, node_info)
+        assert result == "**ABC** - Station ABC"
+
+    def test_format_node_name_with_short_only(self):
+        node_info = {"short_name": "ABC"}
+        result = TestMsg.format_node_name(123456789, node_info)
+        assert result == "**123456789**"
+
+    def test_format_node_name_with_long_only(self):
+        node_info = {"long_name": "Station ABC"}
+        result = TestMsg.format_node_name(123456789, node_info)
+        assert result == "**123456789**"
+
+    def test_format_node_name_with_empty_strings(self):
+        node_info = {"short_name": "", "long_name": ""}
+        result = TestMsg.format_node_name(123456789, node_info)
+        assert result == "**123456789**"
+
+    def test_format_node_name_with_none_values(self):
+        node_info = {"short_name": None, "long_name": None}
+        result = TestMsg.format_node_name(123456789, node_info)
+        assert result == "**123456789**"

tests/test_utils.py

@@ -0,0 +1,42 @@
+from unittest.mock import patch
+
+from bridger.utils import should_ignore_pki_message
+
+
+class TestShouldIgnorePkiMessage:
+
+    @patch("bridger.utils.MQTT_TOPIC", "msh/US/2/e/LongFast/#")
+    def test_should_ignore_pki_message_true(self):
+        pki_topic = "msh/US/2/e/LongFast/PKI/some/subtopic"
+        assert should_ignore_pki_message(pki_topic) is True
+
+    @patch("bridger.utils.MQTT_TOPIC", "msh/US/2/e/LongFast/#")
+    def test_should_ignore_pki_message_false_regular_topic(self):
+        regular_topic = "msh/US/2/e/LongFast/MQTT/some/subtopic"
+        assert should_ignore_pki_message(regular_topic) is False
+
+    @patch("bridger.utils.MQTT_TOPIC", "msh/US/2/e/LongFast/#")
+    def test_should_ignore_pki_message_false_different_path(self):
+        different_topic = "msh/US/2/e/DifferentChannel/PKI/subtopic"
+        assert should_ignore_pki_message(different_topic) is False
+
+    @patch("bridger.utils.MQTT_TOPIC", "msh/US/2/e/LongFast/#")
+    def test_should_ignore_pki_message_false_partial_match(self):
+        partial_topic = "msh/US/2/e/LongFast/PKI"
+        assert should_ignore_pki_message(partial_topic) is False
+
+    @patch("bridger.utils.MQTT_TOPIC", "simple/topic/#")
+    def test_should_ignore_pki_message_simple_topic(self):
+        pki_topic = "simple/topic/PKI/test"
+        regular_topic = "simple/topic/regular/test"
+
+        assert should_ignore_pki_message(pki_topic) is True
+        assert should_ignore_pki_message(regular_topic) is False
+
+    @patch("bridger.utils.MQTT_TOPIC", "no/wildcard")
+    def test_should_ignore_pki_message_no_wildcard(self):
+        pki_topic = "no/wildcard/PKI/test"
+        regular_topic = "no/wildcard/regular/test"
+
+        assert should_ignore_pki_message(pki_topic) is True
+        assert should_ignore_pki_message(regular_topic) is False