diff --git a/.skills/compose-ui/strings-index.txt b/.skills/compose-ui/strings-index.txt index f4325fa808..b87add6771 100644 --- a/.skills/compose-ui/strings-index.txt +++ b/.skills/compose-ui/strings-index.txt @@ -1057,6 +1057,10 @@ mpwrd_os mqtt mqtt_config mqtt_enabled +mqtt_error_connection_lost +mqtt_error_credentials_rejected +mqtt_error_proxy_failed +mqtt_error_rejected mqtt_probe_dns_failure mqtt_probe_other_failure mqtt_probe_rejected diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MqttManagerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MqttManagerImpl.kt index 68e6b9fa43..520128dfb5 100644 --- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MqttManagerImpl.kt +++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MqttManagerImpl.kt @@ -31,15 +31,25 @@ import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.stateIn import org.koin.core.annotation.Named import org.koin.core.annotation.Single +import org.meshtastic.core.common.util.safeCatchingAll import org.meshtastic.core.model.MqttConnectionState import org.meshtastic.core.model.MqttProbeStatus import org.meshtastic.core.network.repository.MQTTRepository +import org.meshtastic.core.network.repository.MQTT_KEEPALIVE_SECONDS +import org.meshtastic.core.network.repository.isCredentialRejection import org.meshtastic.core.network.repository.mqttTlsConfig import org.meshtastic.core.network.repository.resolveEndpoint import org.meshtastic.core.repository.MqttManager import org.meshtastic.core.repository.NodeRepository import org.meshtastic.core.repository.PacketHandler import org.meshtastic.core.repository.ServiceStateWriter +import org.meshtastic.core.resources.Res +import org.meshtastic.core.resources.getStringSuspend +import org.meshtastic.core.resources.mqtt_error_connection_lost +import org.meshtastic.core.resources.mqtt_error_credentials_rejected +import org.meshtastic.core.resources.mqtt_error_proxy_failed +import org.meshtastic.core.resources.mqtt_error_rejected +import org.meshtastic.core.resources.unknown import org.meshtastic.mqtt.ConnectionState import org.meshtastic.mqtt.MqttClient import org.meshtastic.mqtt.MqttException @@ -80,12 +90,26 @@ class MqttManagerImpl( .onEach { message -> packetHandler.sendToRadio(ToRadio(mqttClientProxyMessage = message)) } .catch { throwable -> _proxyActive.value = false + // safeCatchingAll swallows the Skiko ExceptionInInitializerError that + // compose-resources raises on headless JVM tests; production resolves the + // localized string and the error is still surfaced either way. val message = - when (throwable) { - is MqttException.ConnectionRejected -> "MQTT: connection rejected (check credentials)" - is MqttException.ConnectionLost -> "MQTT: connection lost" - else -> "MQTT proxy failed: ${throwable.message}" + safeCatchingAll { + when { + throwable is MqttException.ConnectionRejected && + throwable.isCredentialRejection() -> + getStringSuspend(Res.string.mqtt_error_credentials_rejected) + + throwable is MqttException.ConnectionRejected -> + getStringSuspend(Res.string.mqtt_error_rejected, throwable.detail()) + + throwable is MqttException.ConnectionLost -> + getStringSuspend(Res.string.mqtt_error_connection_lost) + + else -> getStringSuspend(Res.string.mqtt_error_proxy_failed, throwable.detail()) + } } + .getOrDefault("") serviceStateWriter.setErrorMessage(text = message, severity = Severity.Warn) } .launchIn(scope) @@ -144,6 +168,9 @@ class MqttManagerImpl( // including its scoped private-CA trust hook — otherwise a probe would fail where a connect succeeds. val tls = mqttTlsConfig() transportFactory = TcpTransportFactory(tls) + WebSocketTransportFactory(tls) + // Mirror the live client's keepalive too: the library default is 0 (no keepalive), + // which some brokers reject — misleadingly, as CLIENT_IDENTIFIER_NOT_VALID. + keepAliveSeconds = MQTT_KEEPALIVE_SECONDS // Per-connection random suffix: myId identifies the node (and is null → // "unknown" before the node record loads), so two probes can collide on one // client-id and evict each other (SESSION_TAKEN_OVER). See MQTTRepositoryImpl. @@ -188,3 +215,6 @@ class MqttManagerImpl( is ProbeResult.Other -> MqttProbeStatus.Other(message = cause.message) } } + +/** Failure detail for a user-facing message; a throwable without a message still needs a placeholder to substitute. */ +private suspend fun Throwable.detail(): String = message ?: getStringSuspend(Res.string.unknown) 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 83c01c6b1e..fa76e3a70e 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 @@ -52,6 +52,7 @@ import org.meshtastic.mqtt.MqttException import org.meshtastic.mqtt.MqttLogLevel import org.meshtastic.mqtt.MqttMessage import org.meshtastic.mqtt.QoS +import org.meshtastic.mqtt.ReasonCode import org.meshtastic.mqtt.packet.Subscription import org.meshtastic.mqtt.plus import org.meshtastic.mqtt.transport.tcp.TcpTransportFactory @@ -199,17 +200,17 @@ class MQTTRepositoryImpl( } Logger.i { "MQTT connected and subscribed" } } + val failure = result.exceptionOrNull() when { result.isSuccess -> return@launch - result.exceptionOrNull() is MqttException.ConnectionRejected -> { - Logger.e(result.exceptionOrNull()) { "MQTT connection rejected (unrecoverable), stopping" } - close(result.exceptionOrNull()!!) + failure is MqttException.ConnectionRejected && failure.isCredentialRejection() -> { + Logger.e(failure) { "MQTT connection rejected (unrecoverable), stopping" } + close(failure) return@launch } else -> { - val failure = result.exceptionOrNull() // Broker- and network-side failures are what this retry loop exists to absorb — an // unreachable host, a TLS problem, a dropped connection, or a broker that violates the // MQTT 5 spec (e.g. the topic-alias limit). None are defects in this app, and reporting @@ -320,6 +321,24 @@ class MQTTRepositoryImpl( } } +/** + * `true` only for CONNACK reason codes where retrying can never help — the broker examined our credentials or client + * identity and refused them. The library also wraps transport-level connect failures (timeout, TLS, socket EOF) as + * [MqttException.ConnectionRejected] with [ReasonCode.UNSPECIFIED_ERROR]; those are transient and must stay in the + * retry loop, not permanently stop the proxy with a "check credentials" dialog. Public (not internal) so + * `MqttManagerImpl` in `:core:data` can phrase its user-facing error from the same classification. + */ +fun MqttException.ConnectionRejected.isCredentialRejection(): Boolean = when (reasonCode) { + ReasonCode.BAD_USER_NAME_OR_PASSWORD, + ReasonCode.NOT_AUTHORIZED, + ReasonCode.BAD_AUTHENTICATION_METHOD, + ReasonCode.CLIENT_IDENTIFIER_NOT_VALID, + ReasonCode.BANNED, + -> true + + else -> false +} + /** * `true` when an MQTT connect/subscribe failure is one the retry loop is designed to absorb, rather than a defect worth * reporting to Crashlytics/Datadog. @@ -387,8 +406,9 @@ private fun defaultMqttClientFactory(setup: MqttClientSetup): MqttClientSession transportFactory = TcpTransportFactory(tls) + WebSocketTransportFactory(tls) keepAliveSeconds = MQTT_KEEPALIVE_SECONDS autoReconnect = true - username = setup.mqttConfig?.username - setup.mqttConfig?.password?.let { password(it) } + val (user, pass) = effectiveCredentials(setup.mqttConfig) + username = user + pass?.let { password(it) } logger = KermitMqttLogger() // WARN for production: the library emits endpoint addresses and topic strings at // INFO level. WARN messages (reconnect, timeout, retry) contain no PII and are @@ -397,7 +417,9 @@ private fun defaultMqttClientFactory(setup: MqttClientSetup): MqttClientSession }, ) -private const val MQTT_KEEPALIVE_SECONDS = 30 +// Public (not internal/private) so MqttManagerImpl's probe in :core:data can mirror the live client — +// some brokers reject a keepalive-0 CONNECT (misleadingly, as CLIENT_IDENTIFIER_NOT_VALID). +const val MQTT_KEEPALIVE_SECONDS = 30 private const val MQTT_PORT_PLAIN = 1883 private const val MQTT_PORT_TLS = 8883 @@ -426,6 +448,26 @@ fun resolveEndpoint(rawAddress: String, tlsEnabled: Boolean): MqttEndpoint = if private const val DEFAULT_PUBLIC_SERVER = "mqtt.meshtastic.org" +// The public broker's well-known credentials, same values as the firmware's Default.h. +private const val DEFAULT_MQTT_USERNAME = "meshdev" +private const val DEFAULT_MQTT_PASSWORD = "large4cats" + +/** + * Mirrors the firmware's `PubSubConfig` rule: an empty `address` means "the public broker with its well-known + * credentials", substituting username and password together with the server — the stored username/password are ignored + * in that case, whatever they contain. Substituting only the address (as this repository does for the endpoint) while + * passing the stored empty credentials through makes the proxy connect anonymously, which the public broker rejects + * with BAD_USER_NAME_OR_PASSWORD — surfaced to the user as a bogus "check credentials" error on configs the firmware + * itself connects with happily. Empty-address configs occur in the wild: lockdown-enabled firmware hands + * unauthenticated clients a zeroed MQTTConfig. + */ +internal fun effectiveCredentials(config: ModuleConfig.MQTTConfig?): Pair = + if (config?.address.isNullOrEmpty()) { + DEFAULT_MQTT_USERNAME to DEFAULT_MQTT_PASSWORD + } else { + config?.username to config?.password + } + fun effectiveTlsEnabled(address: String, tlsEnabled: Boolean): Boolean = tlsEnabled || extractHost(address).equals(DEFAULT_PUBLIC_SERVER, ignoreCase = true) 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 a3b3b53a29..50caf893f9 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 @@ -37,6 +37,7 @@ import kotlinx.coroutines.test.runTest import kotlinx.serialization.json.Json import okio.ByteString.Companion.toByteString import org.meshtastic.core.common.BuildConfigProvider +import org.meshtastic.core.common.util.safeCatching import org.meshtastic.core.di.CoroutineDispatchers import org.meshtastic.core.model.MqttJsonPayload import org.meshtastic.core.testing.FakeNodeRepository @@ -184,6 +185,40 @@ class MQTTRepositoryImplTest { // endregion + // region effectiveCredentials — firmware-parity credential defaulting. + + @Test + fun `empty address substitutes the public broker's well-known credentials`() { + // Mirrors firmware PubSubConfig: lockdown-redacted (zeroed) configs must not connect anonymously. + val creds = effectiveCredentials(ModuleConfig.MQTTConfig(address = "", username = "", password = "")) + assertEquals("meshdev" to "large4cats", creds) + } + + @Test + fun `null config substitutes the public broker's well-known credentials`() { + assertEquals("meshdev" to "large4cats", effectiveCredentials(null)) + } + + @Test + fun `empty address ignores stored credentials entirely - firmware parity`() { + val creds = effectiveCredentials(ModuleConfig.MQTTConfig(address = "", username = "custom", password = "pw")) + assertEquals("meshdev" to "large4cats", creds) + } + + @Test + fun `explicit address uses the stored credentials as-is`() { + val config = ModuleConfig.MQTTConfig(address = "broker.example.com", username = "user", password = "pass") + assertEquals("user" to "pass", effectiveCredentials(config)) + } + + @Test + fun `explicit default server address uses the stored credentials as-is - firmware parity`() { + val config = ModuleConfig.MQTTConfig(address = "mqtt.meshtastic.org", username = "user", password = "pass") + assertEquals("user" to "pass", effectiveCredentials(config)) + } + + // endregion + // region extractHost — address canonicalization tests. @Test @@ -337,6 +372,74 @@ class MQTTRepositoryImplTest { runCurrent() } + @Test + fun `transport failure wrapped as ConnectionRejected retries instead of stopping the proxy`() = runTest { + // The MQTT library wraps ANY connect failure — timeout, TLS, socket EOF — as + // ConnectionRejected with UNSPECIFIED_ERROR. Treating those as unrecoverable stopped + // the proxy permanently and showed users a bogus "check credentials" dialog. + val harness = createHarness() + harness.client.failConnectWith( + MqttException.ConnectionRejected(ReasonCode.UNSPECIFIED_ERROR, "Connection failed: Connection timed out"), + ) + + val collector = startProxyCollection(harness.repository) + runCurrent() + assertEquals(1, harness.client.connectCalls.size) + + advanceTimeBy(1_000) + runCurrent() + assertEquals(2, harness.client.connectCalls.size) + assertEquals(1, harness.client.subscribeCalls.size) + + collector.cancelAndJoin() + runCurrent() + } + + @Test + fun `credential rejection stops the proxy permanently`() = runTest { + val harness = createHarness() + harness.client.failConnectWith( + MqttException.ConnectionRejected(ReasonCode.BAD_USER_NAME_OR_PASSWORD, "Connection refused"), + ) + + val outcome = backgroundScope.async { safeCatching { harness.repository.proxyMessageFlow.collect {} } } + runCurrent() + advanceTimeBy(60_000) + runCurrent() + + assertEquals(1, harness.client.connectCalls.size) + assertIs(outcome.await().exceptionOrNull()) + } + + @Test + fun `only credential and identity reason codes classify as credential rejections`() { + val fatal = + listOf( + ReasonCode.BAD_USER_NAME_OR_PASSWORD, + ReasonCode.NOT_AUTHORIZED, + ReasonCode.BAD_AUTHENTICATION_METHOD, + ReasonCode.CLIENT_IDENTIFIER_NOT_VALID, + ReasonCode.BANNED, + ) + val transient = + listOf( + ReasonCode.UNSPECIFIED_ERROR, + ReasonCode.SERVER_UNAVAILABLE, + ReasonCode.SERVER_BUSY, + ReasonCode.CONNECTION_RATE_EXCEEDED, + ) + + fatal.forEach { code -> + assertTrue(MqttException.ConnectionRejected(code, "x").isCredentialRejection(), "expected fatal: $code") + } + transient.forEach { code -> + assertFalse( + MqttException.ConnectionRejected(code, "x").isCredentialRejection(), + "expected transient: $code", + ) + } + } + @Test fun `subscription failures trigger reconnect retry`() = runTest { val harness = createHarness() diff --git a/core/resources/src/commonMain/composeResources/values/strings.xml b/core/resources/src/commonMain/composeResources/values/strings.xml index 6d90f3b4e6..4be92ff63b 100644 --- a/core/resources/src/commonMain/composeResources/values/strings.xml +++ b/core/resources/src/commonMain/composeResources/values/strings.xml @@ -1090,6 +1090,10 @@ MQTT MQTT Config MQTT enabled + MQTT: connection lost + MQTT: connection rejected (check credentials) + MQTT proxy failed: %1$s + MQTT: connection rejected: %1$s Host not found Connection failed Broker rejected: %1$s