diff --git a/src/helpers/MQTTConnectionPolicy.h b/src/helpers/MQTTConnectionPolicy.h index 0ab4be12b..ebbf49756 100644 --- a/src/helpers/MQTTConnectionPolicy.h +++ b/src/helpers/MQTTConnectionPolicy.h @@ -17,6 +17,7 @@ static const uint8_t kMaxFailuresAtMaxBackoff = 3; static const uint32_t kDefaultJwtLifetimeSecs = 86400UL; static const uint32_t kMaxJwtStaggerSecs = 300UL; static const uint32_t kMinimumValidEpoch = 1000000000UL; +static const uint32_t kJwtReconnectSafetyMarginSecs = 60UL; static const uint32_t kJwtClockThreshold = 1735689600UL; // 2025-01-01 UTC // A wall clock at or past this instant was set from a real source (NTP or an // admin); anything earlier is the firmware's unset-clock default (1715770351, @@ -157,6 +158,20 @@ static inline bool tokenNeedsRenewal(bool time_synced, uint32_t current_time, return current_time >= token_expires_at - renewal_buffer_secs; } +// A reconnect may keep its credentials only when their validity is known to +// outlast the next handshake; uncertainty refreshes them before reconnecting. +static inline bool canReuseJwtForReconnect(bool time_synced, bool has_token, + bool force_mint, + uint32_t current_time, + uint32_t token_expires_at) { + return time_synced && + has_token && + !force_mint && + token_expires_at >= kMinimumValidEpoch && + current_time < token_expires_at && + (token_expires_at - current_time) > kJwtReconnectSafetyMarginSecs; +} + static inline bool renewalAttemptAllowed(uint32_t now, uint32_t last_attempt) { return elapsedMs(now, last_attempt) >= kRenewalThrottleMs; } diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 906f525aa..5a4cc917c 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -704,6 +704,7 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg _slots[i].last_log_time = 0; _slots[i].port = 1883; _slot_reconfigure_pending[i] = false; + _slot_force_jwt_mint[i] = false; _status_publish_pending[i] = false; } @@ -1601,6 +1602,7 @@ bool MQTTBridge::ensureSlotClient(int index) { slot.client->onConnect([this, index](bool sessionPresent) { MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1); _slots[index].connected = true; + _slot_force_jwt_mint[index] = false; // NOTE: reconnect_backoff / max_backoff_failures are NOT reset here. // A CONNACK alone doesn't prove the link is healthy — a broker that // accepts and then drops within seconds would reset the ladder every @@ -1647,6 +1649,7 @@ bool MQTTBridge::ensureSlotClient(int index) { _slots[index].last_sock_errno = error.esp_transport_sock_errno; _slots[index].last_error_time = millis(); if (error.error_type == MQTT_ERROR_TYPE_CONNECTION_REFUSED) { + _slot_force_jwt_mint[index] = true; // Broker rejected the MQTT CONNECT itself — not a transport failure. // return code: 1=protocol, 2=client-id rejected, 3=server unavailable, // 4=bad username/password, 5=not authorized. Codes 3/4/5 point at a @@ -1801,6 +1804,8 @@ bool MQTTBridge::setupSlot(int index) { slot.max_backoff_failures = 0; slot.circuit_breaker_tripped = false; slot.last_reconnect_attempt = 0; + // The refusal that set this belonged to the credentials being cleared here. + _slot_force_jwt_mint[index] = false; } bool uses_jwt = (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT) || slot.audience[0] != '\0'; @@ -1990,6 +1995,8 @@ void MQTTBridge::teardownSlot(int index) { slot.last_reconnect_attempt = 0; slot.last_log_time = 0; slot.last_deferred_log_ms = 0; + // The refusal that set this belonged to the credentials being cleared here. + _slot_force_jwt_mint[index] = false; } // A stopped client needs connect(): reconnect() is a documented no-op on one, so reaching @@ -2203,6 +2210,59 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // persistent clients (Phase 1), the mbedTLS context is allocated once at // startup and the preflight is no longer necessary. + const auto prepareJwtReconnect = [&](bool force_mint, int backoff_level) { + const bool has_token = slot.auth_token && slot.auth_token[0] != '\0'; + const unsigned long expires_at = slot.token_expires_at; + const bool remaining_known = time_synced && + expires_at >= MQTTConnectionPolicy::kMinimumValidEpoch; + const unsigned long remaining_secs = current_time < expires_at + ? expires_at - current_time + : 0; + const bool force_mint_after_refusal = _slot_force_jwt_mint[index]; + force_mint = force_mint || force_mint_after_refusal; + const bool reuse_token = MQTTConnectionPolicy::canReuseJwtForReconnect( + time_synced, has_token, force_mint, static_cast(current_time), + static_cast(expires_at)); + const char* mint_reason = "none"; + if (!reuse_token) { + if (backoff_level < 0) { + mint_reason = "circuit-breaker-probe"; + } else if (force_mint_after_refusal) { + mint_reason = "connection-refused"; + } else if (!time_synced) { + mint_reason = "clock-unsynced"; + } else if (!has_token) { + mint_reason = "empty-token"; + } else if (expires_at < MQTTConnectionPolicy::kMinimumValidEpoch) { + mint_reason = "invalid-expiry"; + } else if (current_time >= expires_at) { + mint_reason = "expired"; + } else { + mint_reason = "safety-margin"; + } + } + const char* mint_result = reuse_token ? "REUSED" : "FAILED"; + if (!reuse_token && createSlotAuthToken(index)) { + slot.client->setCredentials(_jwt_username, slot.auth_token); + mint_result = "OK"; + } + char remaining_text[24]; + if (remaining_known) { + snprintf(remaining_text, sizeof(remaining_text), "%lus", remaining_secs); + } else { + strncpy(remaining_text, "unknown", sizeof(remaining_text)); + } + if (backoff_level >= 0) { + MQTT_DEBUG_PRINTLN("MQTT%d JWT reconnect backoff=%d token=%s mint_reason=%s result=%s remaining=%s", + index + 1, backoff_level, reuse_token ? "REUSE" : "MINT", mint_reason, + mint_result, remaining_text); + } else { + MQTT_DEBUG_PRINTLN("MQTT%d JWT circuit-breaker probe token=%s mint_reason=%s result=%s remaining=%s", + index + 1, reuse_token ? "REUSE" : "MINT", mint_reason, mint_result, + remaining_text); + } + }; + // Periodic probe for circuit-breaker-tripped slots (recovery from transient outages) // Attempts a single reconnect every 30 minutes to see if the server has come back if (slot.circuit_breaker_tripped && !reconnect_attempted) { @@ -2219,13 +2279,7 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns _radio ? _radio->getRadioState() : -1, (_radio && _radio->getLastRecvMillis() > 0) ? (_ms->getMillis() - _radio->getLastRecvMillis()) : 0); if (slot_uses_jwt) { - // Regenerate or refresh token, then reconnect the persistent client. - // Reaching the ladder at all means setupSlot() ran, so the client object - // and its mbedTLS context are live and no full setup is needed here. - if (createSlotAuthToken(index)) { - slot.client->setCredentials(_jwt_username, slot.auth_token); - MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker probe (fresh token)", index + 1); - } + prepareJwtReconnect(true, -1); slot.client->reconnect(); } else { slot.client->reconnect(); @@ -2259,16 +2313,7 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns reconnect_attempted = true; _last_slot_reconnect_ms = now_millis; if (slot_uses_jwt) { - // Always lightweight reconnect on the persistent client. A stale/expired - // token is handled by regenerating it in place and updating credentials - // — no teardown is needed because the client and its mbedTLS context - // persist for the bridge lifetime. - if (createSlotAuthToken(index)) { - slot.client->setCredentials(_jwt_username, slot.auth_token); - MQTT_DEBUG_PRINTLN("MQTT%d reconnect (fresh token, backoff %d)", index + 1, slot.reconnect_backoff); - } else { - MQTT_DEBUG_PRINTLN("MQTT%d reconnect (token refresh failed, backoff %d)", index + 1, slot.reconnect_backoff); - } + prepareJwtReconnect(false, slot.reconnect_backoff); slot.client->reconnect(); } else { // Non-JWT slots — lightweight reconnect on existing client. diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index cbc824827..bda752a6f 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -228,6 +228,10 @@ private: // Pending slot reconfigure: set from CLI (Core 1), processed by MQTT task (Core 0) volatile bool _slot_reconfigure_pending[RUNTIME_MQTT_SLOTS]; + // A broker refusal can invalidate an otherwise clock-valid JWT. The esp-mqtt + // callback sets this and the bridge loop consumes it; byte access is atomic. + volatile bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS]; + // Pending on-connect status publish: set from the onConnect callback (which // runs on the esp-mqtt event task, NOT this bridge task), consumed by the MQTT // task (Core 0). publishStatusToSlot() touches the shared status doc/buffer/ diff --git a/test/test_mqtt_connection_policy/test_mqtt_connection_policy.cpp b/test/test_mqtt_connection_policy/test_mqtt_connection_policy.cpp index 496a1c434..55c58793d 100644 --- a/test/test_mqtt_connection_policy/test_mqtt_connection_policy.cpp +++ b/test/test_mqtt_connection_policy/test_mqtt_connection_policy.cpp @@ -137,6 +137,22 @@ TEST(MQTTConnectionPolicy, SyncedClockRenewsInvalidExpiredOrImminentTokens) { EXPECT_TRUE(Policy::tokenNeedsRenewal(true, expires + 1U, expires, 300U)); } +TEST(MQTTConnectionPolicy, JwtReconnectReusesOnlyProvenValidCredentials) { + const uint32_t now = 1735689600U; + const uint32_t usable_expiry = now + Policy::kJwtReconnectSafetyMarginSecs + 1U; + + EXPECT_TRUE(Policy::canReuseJwtForReconnect(true, true, false, now, usable_expiry)); + EXPECT_FALSE(Policy::canReuseJwtForReconnect( + true, true, false, now, Policy::kMinimumValidEpoch - 1U)); + EXPECT_FALSE(Policy::canReuseJwtForReconnect(true, true, false, now, 0U)); + EXPECT_FALSE(Policy::canReuseJwtForReconnect( + true, true, false, now, now + Policy::kJwtReconnectSafetyMarginSecs)); + EXPECT_FALSE(Policy::canReuseJwtForReconnect(true, true, false, now, now - 1U)); + EXPECT_FALSE(Policy::canReuseJwtForReconnect(true, false, false, now, usable_expiry)); + EXPECT_FALSE(Policy::canReuseJwtForReconnect(false, true, false, now, usable_expiry)); + EXPECT_FALSE(Policy::canReuseJwtForReconnect(true, true, true, now, usable_expiry)); +} + TEST(MQTTConnectionPolicy, RenewalThrottleHasExactBoundaryAndHandlesRollover) { EXPECT_FALSE(Policy::renewalAttemptAllowed(59999U, 0U)); EXPECT_TRUE(Policy::renewalAttemptAllowed(60000U, 0U));