fix(mqtt): never hand back a half-initialized client or transport

esp_mqtt_client_init() succeeded without a registered event sink when
registration failed, and the WebSocket wrap returned a transport with
the unpadded SDK buffer when the padded allocation failed. Both paths
now fail cleanly instead, the IDF 4 WebSocket layout is pinned to the
validated patch level, and the connection flags esp-mqtt tasks write
become atomic.
This commit is contained in:
agessaman
2026-09-21 16:05:39 -07:00
parent 6124f3c139
commit 665ece1e12
6 changed files with 135 additions and 27 deletions
@@ -0,0 +1,17 @@
#pragma once
// Initialization is incomplete until lifecycle events have a registered sink.
// The candidate has never started, so registration failure can destroy it.
template<class Handle, class Init, class Register, class Destroy>
int initializeMqttClient(Handle& target, Init init, Register register_events,
Destroy destroy, int no_memory) {
Handle candidate = init();
if (!candidate) return no_memory;
const int result = register_events(candidate);
if (result != 0) {
destroy(candidate);
return result;
}
target = candidate;
return 0;
}
+11 -10
View File
@@ -1,3 +1,4 @@
#include "MqttClientInit.h"
#include "PsychicMqttClient.h"
#include <string.h>
@@ -386,17 +387,17 @@ esp_err_t PsychicMqttClient::applyConfig()
if (_client == nullptr)
{
// esp_mqtt_client_init() takes the whole configuration, including the
// buffer sizes it allocates once and never resizes.
_client = esp_mqtt_client_init(&_mqtt_cfg);
if (_client == nullptr)
{
ESP_LOGE(TAG, "esp_mqtt_client_init failed");
return ESP_ERR_NO_MEM;
const esp_err_t result = initializeMqttClient(_client,
[&]() { return esp_mqtt_client_init(&_mqtt_cfg); },
[&](esp_mqtt_client_handle_t client) {
return esp_mqtt_client_register_event(client, MQTT_EVENT_ANY, _onMqttEventStatic, this);
},
[](esp_mqtt_client_handle_t client) { esp_mqtt_client_destroy(client); },
ESP_ERR_NO_MEM);
if (result != ESP_OK) {
ESP_LOGE(TAG, "MQTT initialization failed: %s", esp_err_to_name(result));
return result;
}
// Register event handler only once when client is first created
// to avoid memory leak from repeated registrations
esp_mqtt_client_register_event(_client, MQTT_EVENT_ANY, _onMqttEventStatic, this);
_config_dirty = false;
return ESP_OK;
}
@@ -1,4 +1,5 @@
#pragma once
#include <atomic>
/**
* PsychicMqttClient
*
@@ -545,8 +546,8 @@ private:
esp_mqtt_client_handle_t _client = nullptr;
esp_mqtt_client_config_t _mqtt_cfg;
esp_mqtt_error_codes_t _lastError;
bool _connected = false;
bool _stopMqttClient = false;
std::atomic<bool> _connected{false};
std::atomic<bool> _stopMqttClient{false};
bool _config_dirty = true;
bool _started = false;
+13 -15
View File
@@ -62,7 +62,12 @@
#if ESP_IDF_VERSION_MAJOR == 4
#if ESP_IDF_VERSION_MINOR != 4 || ESP_IDF_VERSION_PATCH != 7
#error "Revalidate the private WebSocket layout before using another IDF 4 SDK"
#endif
#include <stdlib.h>
#include "RequiredTransportBuffer.h"
#include "sdkconfig.h"
#include "esp_transport.h"
#include "esp_transport_ws.h"
@@ -97,22 +102,15 @@ extern "C" {
esp_transport_handle_t __real_esp_transport_ws_init(esp_transport_handle_t parent_handle);
esp_transport_handle_t __wrap_esp_transport_ws_init(esp_transport_handle_t parent_handle) {
esp_transport_handle_t t = __real_esp_transport_ws_init(parent_handle);
if (t != nullptr) {
transport_ws_t* ws = (transport_ws_t*)esp_transport_get_context_data(t);
if (ws != nullptr && ws->buffer != nullptr) {
// The buffer is untouched at this point (allocated moments ago inside
// __real_esp_transport_ws_init), so a swap is safe.
char* padded = (char*)malloc(CONFIG_WS_BUFFER_SIZE + 1);
if (padded != nullptr) {
return initWithRequiredTransportBuffer<esp_transport_handle_t>(
CONFIG_WS_BUFFER_SIZE + 1, malloc,
[&]() { return __real_esp_transport_ws_init(parent_handle); },
[](esp_transport_handle_t transport, void* buffer) {
// Successful init in the pinned SDK guarantees this context and buffer.
transport_ws_t* ws = (transport_ws_t*)esp_transport_get_context_data(transport);
free(ws->buffer);
ws->buffer = padded;
}
// On alloc failure keep the original buffer: same behavior as before
// this fix, which is still strictly better than failing init here.
}
}
return t;
ws->buffer = static_cast<char*>(buffer);
}, free);
}
} // extern "C"
+18
View File
@@ -0,0 +1,18 @@
#pragma once
#include <stddef.h>
// Allocate the protection buffer first. A successful init must install it;
// neither allocation failure is allowed to return an unprotected transport.
template<class Handle, class Allocate, class Init, class Install, class Free>
Handle initWithRequiredTransportBuffer(size_t size, Allocate allocate, Init init,
Install install, Free release) {
void* buffer = allocate(size);
if (!buffer) return nullptr;
Handle transport = init();
if (!transport) {
release(buffer);
return nullptr;
}
install(transport, buffer);
return transport;
}
@@ -0,0 +1,73 @@
#include <gtest/gtest.h>
#include "helpers/RequiredTransportBuffer.h"
#include "../../lib/PsychicMqttClient/src/MqttClientInit.h"
TEST(ObserverInit, MissingPaddingPreventsSdkInitialization) {
bool initialized = false, installed = false, freed = false;
int transport;
int* result = initWithRequiredTransportBuffer<int*>(1025,
[](size_t) -> void* { return nullptr; },
[&] { initialized = true; return &transport; },
[&](int*, void*) { installed = true; },
[&](void*) { freed = true; });
EXPECT_EQ(nullptr, result);
EXPECT_FALSE(initialized);
EXPECT_FALSE(installed);
EXPECT_FALSE(freed);
}
TEST(ObserverInit, FailedSdkInitReleasesReservedPadding) {
char buffer[1025];
bool freed = false, installed = false;
int* result = initWithRequiredTransportBuffer<int*>(sizeof(buffer),
[&](size_t n) -> void* { EXPECT_EQ(sizeof(buffer), n); return buffer; },
[]() -> int* { return nullptr; },
[&](int*, void*) { installed = true; },
[&](void* p) { EXPECT_EQ(buffer, p); freed = true; });
EXPECT_EQ(nullptr, result);
EXPECT_TRUE(freed);
EXPECT_FALSE(installed);
}
TEST(ObserverInit, SuccessfulTransportOwnsProtectionBuffer) {
char buffer[1025];
int transport;
bool installed = false, freed = false;
int* result = initWithRequiredTransportBuffer<int*>(sizeof(buffer),
[&](size_t) -> void* { return buffer; },
[&] { return &transport; },
[&](int* t, void* p) { EXPECT_EQ(&transport, t); EXPECT_EQ(buffer, p); installed = true; },
[&](void*) { freed = true; });
EXPECT_EQ(&transport, result);
EXPECT_TRUE(installed);
EXPECT_FALSE(freed);
}
TEST(ObserverInit, FailedRegistrationDestroysCandidateAndCanRetry) {
int candidate;
int* client = nullptr;
int registrations = 0, destroys = 0;
auto create = [&] { return &candidate; };
auto register_events = [&](int* p) { EXPECT_EQ(&candidate, p); return ++registrations == 1 ? -7 : 0; };
auto destroy = [&](int* p) { EXPECT_EQ(&candidate, p); ++destroys; };
EXPECT_EQ(-7, initializeMqttClient(client, create, register_events, destroy, -2));
EXPECT_EQ(nullptr, client);
EXPECT_EQ(1, destroys);
EXPECT_EQ(0, initializeMqttClient(client, create, register_events, destroy, -2));
EXPECT_EQ(&candidate, client);
EXPECT_EQ(1, destroys);
}
TEST(ObserverInit, FailedClientAllocationDoesNotRegisterOrDestroy) {
int* client = nullptr;
int effects = 0;
EXPECT_EQ(-2, initializeMqttClient(client, []() -> int* { return nullptr; },
[&](int*) { ++effects; return 0; }, [&](int*) { ++effects; }, -2));
EXPECT_EQ(nullptr, client);
EXPECT_EQ(0, effects);
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}