diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt index 89e214a50..95d0c7261 100644 --- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt +++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt @@ -40,6 +40,7 @@ import org.koin.core.annotation.Single import org.meshtastic.core.common.util.safeCatching import org.meshtastic.core.di.CoroutineDispatchers import org.meshtastic.core.model.MqttJsonPayload +import org.meshtastic.core.model.util.decodeOrNull import org.meshtastic.core.model.util.subscribeList import org.meshtastic.core.repository.NodeRepository import org.meshtastic.core.repository.RadioConfigRepository @@ -56,6 +57,7 @@ import org.meshtastic.mqtt.transport.tcp.TcpTransportFactory import org.meshtastic.mqtt.transport.ws.WebSocketTransportFactory import org.meshtastic.proto.ModuleConfig import org.meshtastic.proto.MqttClientProxyMessage +import org.meshtastic.proto.ServiceEnvelope import kotlin.concurrent.Volatile import kotlin.uuid.Uuid @@ -162,7 +164,13 @@ class MQTTRepositoryImpl( ) } } - add(Subscription("$rootTopic${DEFAULT_TOPIC_LEVEL}PKI/+", maxQos = QoS.AT_LEAST_ONCE, noLocal = true)) + add( + Subscription( + "$rootTopic$DEFAULT_TOPIC_LEVEL$PKI_CHANNEL_ID/+", + maxQos = QoS.AT_LEAST_ONCE, + noLocal = true, + ), + ) } // Collect from the SharedFlow before connecting to avoid missing retained messages @@ -248,6 +256,20 @@ class MQTTRepositoryImpl( Logger.e(e) { "Failed to parse MQTT JSON: ${e.message}" } } } else { + // Drop provably-undeliverable downlink packets before spending BLE bandwidth on + // them. In client-proxy mode the public broker floods payload-less packet-header + // stubs the node can only decrypt-fail and discard; stopping them here saves BLE + // airtime, a decrypt attempt, and a node-side warning per packet. + if (isUndeliverableDownlink(payload, nodeRepository.myId.value)) { + Logger.d { + if (buildConfigProvider.isDebug) { + "MQTT downlink dropped (no payload): $topic" + } else { + "MQTT downlink dropped (no payload)" + } + } + return + } trySend(MqttClientProxyMessage(topic = topic, data_ = payload.toByteString(), retained = msg.retain)) } } @@ -391,3 +413,35 @@ internal fun extractHost(address: String): String { // Remove path (if any), then remove port return afterScheme.substringBefore("/").substringBefore(":") } + +private const val PKI_CHANNEL_ID = "PKI" + +/** + * Returns true when a downlink [ServiceEnvelope] is provably un-usable by the node and should be dropped before + * forwarding over BLE (MQTT client-proxy mode). + * + * Fails open — returns false (forward) for anything it cannot positively prove undeliverable: unparseable bytes, + * traffic on the `PKI` channel (public-key-encrypted direct messages the node accepts without first decrypting), our + * own echoed-back packets (used as implicit ACKs), a packet-less envelope, or any packet that actually carries a + * payload. The only drop case is Tier 1: a [MeshPacket] with neither `decoded` nor `encrypted` set — no legitimate + * Meshtastic packet is payload-less. + * + * Extracted as an internal top-level function so [MQTTRepositoryImplTest] can exercise every branch without spinning up + * the full repository. + * + * @param payload the raw MQTT payload bytes (a serialized ServiceEnvelope on the binary topic). + * @param myId this device's node id in `!xxxxxxxx` hex form, or null before the local node loads. + */ +internal fun isUndeliverableDownlink(payload: ByteArray, myId: String?): Boolean { + // Decode without a logger: this is fail-open and, on a public broker, may see arbitrary bytes on the binary + // topic. Passing Logger would emit an error log per unparseable downlink — expensive noise for an expected case. + val envelope = ServiceEnvelope.ADAPTER.decodeOrNull(payload) ?: return false // fail open on garbage + // Guards — never drop: PKI-channel traffic (accepted without decryption) or our own echoed-back + // packets (used as implicit ACKs). + val isGuarded = envelope.channel_id == PKI_CHANNEL_ID || (myId != null && envelope.gateway_id == myId) + // Tier 1: a packet with neither `decoded` nor `encrypted` carries nothing the node can use. + // A packet-less envelope (packet == null) has nothing to prove, so it is treated as deliverable. + val packet = envelope.packet + val hasNoPayload = packet != null && packet.decoded == null && packet.encrypted == null + return hasNoPayload && !isGuarded +} diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt index ce6ebe3f8..a3b3b53a2 100644 --- a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt +++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt @@ -51,12 +51,16 @@ import org.meshtastic.mqtt.ReasonCode import org.meshtastic.mqtt.packet.Subscription import org.meshtastic.proto.ChannelSet import org.meshtastic.proto.ChannelSettings +import org.meshtastic.proto.Data import org.meshtastic.proto.LocalModuleConfig +import org.meshtastic.proto.MeshPacket import org.meshtastic.proto.ModuleConfig +import org.meshtastic.proto.ServiceEnvelope import kotlin.test.BeforeTest import kotlin.test.Test import kotlin.test.assertContentEquals import kotlin.test.assertEquals +import kotlin.test.assertFalse import kotlin.test.assertIs import kotlin.test.assertNull import kotlin.test.assertTrue @@ -427,6 +431,89 @@ class MQTTRepositoryImplTest { // endregion + // region isUndeliverableDownlink — Tier 1 drop filter for MQTT client-proxy downlink packets. + + private fun envelopeBytes( + channelId: String = "LongFast", + gatewayId: String = "!aabbccdd", + packet: MeshPacket? = MeshPacket(), + ): ByteArray = + ServiceEnvelope.ADAPTER.encode(ServiceEnvelope(packet = packet, channel_id = channelId, gateway_id = gatewayId)) + + @Test + fun `payload-less packet is undeliverable`() { + // The observed LongFast flood: a packet with neither decoded nor encrypted set. + val bytes = envelopeBytes(packet = MeshPacket(from = 1, to = 2)) + assertTrue(isUndeliverableDownlink(bytes, myId = "!12345678")) + } + + @Test + fun `decoded packet is deliverable`() { + val bytes = envelopeBytes(packet = MeshPacket(decoded = Data())) + assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678")) + } + + @Test + fun `encrypted packet is deliverable`() { + val bytes = envelopeBytes(packet = MeshPacket(encrypted = byteArrayOf(1, 2, 3).toByteString())) + assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678")) + } + + @Test + fun `PKI payload-less packet is deliverable - guard`() { + val bytes = envelopeBytes(channelId = "PKI", packet = MeshPacket(from = 1)) + assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678")) + } + + @Test + fun `own echo payload-less packet is deliverable - guard`() { + val bytes = envelopeBytes(gatewayId = "!12345678", packet = MeshPacket(from = 1)) + assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678")) + } + + @Test + fun `a different gateway's payload-less packet is undeliverable`() { + val bytes = envelopeBytes(gatewayId = "!deadbeef", packet = MeshPacket(from = 1)) + assertTrue(isUndeliverableDownlink(bytes, myId = "!12345678")) + } + + @Test + fun `null myId does not suppress dropping`() { + val bytes = envelopeBytes(gatewayId = "!deadbeef", packet = MeshPacket(from = 1)) + assertTrue(isUndeliverableDownlink(bytes, myId = null)) + } + + @Test + fun `packet-less envelope is deliverable - fail open`() { + val bytes = envelopeBytes(packet = null) + assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678")) + } + + @Test + fun `garbage bytes are deliverable - fail open`() { + // Non-protobuf bytes must never be dropped. + val bytes = byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0xFF.toByte(), 0x42) + assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678")) + } + + @Test + fun `payload-less downlink stubs are dropped before forwarding`() = runTest { + val harness = createHarness() + val stub = envelopeBytes(packet = MeshPacket(from = 1, to = 2)) // no payload → dropped + val real = envelopeBytes(packet = MeshPacket(decoded = Data())) // has payload → forwarded + + val nextMessage = backgroundScope.async { harness.repository.proxyMessageFlow.first() } + runCurrent() + harness.client.emitMessage(MqttMessage(topic = "msh/2/e/alpha/node", payload = stub, retain = false)) + harness.client.emitMessage(MqttMessage(topic = "msh/2/e/alpha/node", payload = real, retain = false)) + + // first() returns the first *forwarded* message; the stub was dropped, so it must be `real`. + val proxyMessage = nextMessage.await() + assertContentEquals(real, proxyMessage.data_?.toByteArray()) + } + + // endregion + private fun TestScope.createHarness( radioConfigRepository: FakeRadioConfigRepository = defaultRadioConfigRepository(), nodeRepository: FakeNodeRepository = FakeNodeRepository().apply { setMyId("!12345678") },