diff --git a/repeater/data_acquisition/mqtt_handler.py b/repeater/data_acquisition/mqtt_handler.py index 726559e..a7c033d 100644 --- a/repeater/data_acquisition/mqtt_handler.py +++ b/repeater/data_acquisition/mqtt_handler.py @@ -40,6 +40,26 @@ def b64url(x: bytes) -> str: return base64.urlsafe_b64encode(x).rstrip(b"=").decode() +def _payload_packet_type(payload: dict) -> Optional[int]: + """MeshCore payload-type code of a published payload, or None. + + ``type`` is the message class the MC2MQTT schema puts on the wire - the + literal string "PACKET" - not the packet's type. The code lives in + ``packet_type``, serialised as a decimal string by PacketRecord, and must + be read as an int to match ``disallowed_packet_types`` (which resolves + configured names through PAYLOAD_TYPES). Status and diagnostic payloads + have no packet type and are never filtered. + """ + value = payload.get("packet_type") + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + logger.debug(f"Unreadable packet_type in payload: {value!r}") + return None + + def _summarize_payload_for_log(payload: Any, message: Optional[str] = None) -> str: """Return a compact single-line payload summary for debug logging.""" if message is None: @@ -1048,13 +1068,13 @@ class MeshCoreToMqttPusher: f"Publishing topic='{subtopic}', {_summarize_payload_for_log(payload, message)}" ) - packet_type = payload.get("type") + packet_type = _payload_packet_type(payload) results = [] with self._lock: for conn in self.connections: if conn.enabled and conn.is_connected(): - if packet_type in conn.disallowed_types: + if packet_type is not None and packet_type in conn.disallowed_types: _trace(f"Skipped publishing packet type 0x{packet_type:02X} (disallowed)") continue result = conn.publish(subtopic, message, retain=retain, qos=qos) diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index c16384b..b88d9d3 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -297,6 +297,12 @@ class StorageCollector: Only fast, per-packet work runs here. The aggregate stats broadcast is driven separately by _stats_broadcast_loop so the writer thread is not held by the multi-second get_packet_stats(24h) query. + + ``skip_mqtt`` withholds a packet the caller judged invalid (a malformed + advert, an empty payload, an over-long path) from the brokers only. It + is still stored and still reaches Glass and the dashboard, which are + the surfaces an operator debugs their own RF from; what it must not do + is feed a network-wide observer a packet this node could not parse. """ self._publish_to_glass(packet_record, "packet") @@ -306,6 +312,13 @@ class StorageCollector: except Exception as e: logger.debug(f"WebSocket broadcast failed: {e}") + if skip_mqtt and packet_record.get("drop_reason"): + logger.debug( + "Skipping mqtt publish for invalid packet: %s", + packet_record.get("drop_reason"), + ) + return + self._publish_packet_to_mqtt(packet_record) # The heavy 24h SQL aggregate rides one beat in six (30s at the 5s diff --git a/tests/test_mqtt_publish_filters.py b/tests/test_mqtt_publish_filters.py new file mode 100644 index 0000000..0122be3 --- /dev/null +++ b/tests/test_mqtt_publish_filters.py @@ -0,0 +1,216 @@ +"""The two MQTT publish filters, which both used to pass everything through. + +``disallowed_packet_types`` compared the payload's ``type`` field - the +literal string "PACKET" that the MC2MQTT schema puts on every packet message - +against integer payload-type codes, so no configured type was ever blocked. +``skip_mqtt_if_invalid`` was threaded from the engine down to the publish call +and then never read, so packets this node could not parse were published to +every broker anyway. +""" + +import json +from unittest.mock import MagicMock, patch + +from repeater.data_acquisition.mqtt_handler import MeshCoreToMqttPusher +from repeater.data_acquisition.storage_collector import StorageCollector +from repeater.data_acquisition.storage_utils import PacketRecord + +ADVERT = 4 +TXT_MSG = 2 + + +class _FakeIdentity: + def __init__(self, public_key_hex: str): + self._pk = bytes.fromhex(public_key_hex) + + def get_public_key(self) -> bytes: + return self._pk + + +def _make_config(disallowed=None) -> dict: + broker = { + "name": "test-broker", + "enabled": True, + "host": "broker.example", + "port": 1883, + "transport": "tcp", + "format": "letsmesh", + "use_jwt_auth": False, + "tls": {"enabled": False, "insecure": False}, + } + if disallowed is not None: + broker["disallowed_packet_types"] = disallowed + return { + "repeater": {"node_name": "test-node"}, + "radio_type": "sx1262", + "radio": { + "frequency": 869618000, + "bandwidth": 62500, + "spreading_factor": 8, + "coding_rate": 8, + "preamble_length": 32, + }, + "duty_cycle": {"max_airtime_per_minute": 3600}, + "mqtt_brokers": {"iata_code": "LAX", "status_interval": 0, "brokers": [broker]}, + } + + +def _capturing_pusher(config: dict): + pusher = MeshCoreToMqttPusher(local_identity=_FakeIdentity("AB" * 32), config=config) + conn = pusher.connections[0] + captured = [] + conn._running = True + conn.client = MagicMock() + conn.client.publish = lambda topic, payload, retain=False, qos=0: captured.append( + {"topic": topic, "payload": payload} + ) + return pusher, captured + + +def _packet_payload(payload_type: int) -> dict: + record = PacketRecord.from_packet_record( + { + "timestamp": 1700000000.0, + "type": payload_type, + "route": 1, + "rssi": -90, + "snr": 7.5, + "score": 0.5, + "payload_length": 32, + "packet_hash": "DEADBEEF" + "00" * 4, + "raw_packet": bytes(range(40)).hex(), + "airtime_ms": 123.0, + }, + origin="test-node", + origin_id="AB" * 32, + ) + assert record is not None + return record.to_dict() + + +# -------------------------------------------------------------------- +# disallowed_packet_types +# -------------------------------------------------------------------- +def test_disallowed_packet_type_is_not_published(): + pusher, captured = _capturing_pusher(_make_config(disallowed=["ADVERT"])) + + pusher.publish_packet(_packet_payload(ADVERT)) + + assert captured == [] + + +def test_allowed_packet_type_still_publishes(): + pusher, captured = _capturing_pusher(_make_config(disallowed=["ADVERT"])) + + pusher.publish_packet(_packet_payload(TXT_MSG)) + + assert len(captured) == 1 + assert json.loads(captured[0]["payload"])["packet_type"] == str(TXT_MSG) + + +def test_no_disallowed_list_publishes_every_type(): + pusher, captured = _capturing_pusher(_make_config()) + + pusher.publish_packet(_packet_payload(ADVERT)) + pusher.publish_packet(_packet_payload(TXT_MSG)) + + assert len(captured) == 2 + + +def test_status_is_never_filtered_by_the_packet_type_list(): + """Status carries no packet type; a filtered broker must still get it.""" + pusher, captured = _capturing_pusher(_make_config(disallowed=["ADVERT"])) + + pusher.publish_status(state="online") + + assert len(captured) == 1 + assert json.loads(captured[0]["payload"])["status"] == "online" + + +def test_unreadable_packet_type_is_published_rather_than_dropped(): + """A payload we cannot classify is not evidence it was disallowed.""" + pusher, captured = _capturing_pusher(_make_config(disallowed=["ADVERT"])) + payload = _packet_payload(ADVERT) + payload["packet_type"] = "not-a-number" + + pusher.publish("packets", payload) + + assert len(captured) == 1 + + +# -------------------------------------------------------------------- +# skip_mqtt_if_invalid +# -------------------------------------------------------------------- +def _make_collector(tmp_path) -> StorageCollector: + config = {"storage": {"storage_dir": str(tmp_path)}, "repeater": {"node_name": "test-node"}} + with ( + patch("repeater.data_acquisition.storage_collector.SQLiteHandler"), + patch("repeater.data_acquisition.storage_collector.RRDToolHandler"), + patch("repeater.data_acquisition.hardware_stats.HardwareStatsCollector"), + ): + collector = StorageCollector(config=config) + + collector._stats_stop_event.set() + if collector._stats_thread is not None: + collector._stats_thread.join(timeout=1) + collector.mqtt_handler = MagicMock() + collector.mqtt_handler.public_key = "AB" * 32 + collector._publish_to_glass = MagicMock() + collector.websocket_broadcast_packet = MagicMock() + collector.websocket_available = True + return collector + + +def _record(**overrides) -> dict: + record = { + "timestamp": 1700000000.0, + "type": ADVERT, + "route": 1, + "rssi": -90, + "snr": 7.5, + "score": 0.5, + "payload_length": 32, + "packet_hash": "DEADBEEF" + "00" * 4, + "raw_packet": bytes(range(40)).hex(), + "airtime_ms": 123.0, + } + record.update(overrides) + return record + + +def test_invalid_packet_is_withheld_from_brokers(tmp_path): + collector = _make_collector(tmp_path) + + collector._publish_packet_sync(_record(drop_reason="Invalid advert"), skip_mqtt=True) + + collector.mqtt_handler.publish_packet.assert_not_called() + + +def test_invalid_packet_still_reaches_glass_and_the_dashboard(tmp_path): + """The operator's own surfaces are how they debug their RF; only the + network-wide observer feed is withheld.""" + collector = _make_collector(tmp_path) + + collector._publish_packet_sync(_record(drop_reason="Invalid advert"), skip_mqtt=True) + + collector._publish_to_glass.assert_called_once() + collector.websocket_broadcast_packet.assert_called_once() + + +def test_operational_drops_are_still_published(tmp_path): + """A duplicate or a policy drop is a valid packet this node chose not to + repeat, which is exactly what an observer wants to see.""" + collector = _make_collector(tmp_path) + + collector._publish_packet_sync(_record(drop_reason="Duplicate"), skip_mqtt=False) + + collector.mqtt_handler.publish_packet.assert_called_once() + + +def test_valid_packet_is_published_even_when_the_flag_is_set(tmp_path): + """The flag only withholds packets that actually carry a drop reason.""" + collector = _make_collector(tmp_path) + + collector._publish_packet_sync(_record(), skip_mqtt=True) + + collector.mqtt_handler.publish_packet.assert_called_once()