From ca7c8bd4faad84b7f50499c2f1ad10403afa0de4 Mon Sep 17 00:00:00 2001 From: john30 Date: Sun, 27 Aug 2017 09:40:43 +0200 Subject: [PATCH] fix for potential access to no longer existing Message instances in DataHandler, validate MQTT topic during argument parsing, extended DataHandler registration to allow registering multiple instances at once, only create MqttHandler if active --- src/ebusd/datahandler.cpp | 11 ++--- src/ebusd/datahandler.h | 4 +- src/ebusd/main.cpp | 2 + src/ebusd/mqtthandler.cpp | 98 +++++++++++++++++++++++---------------- src/ebusd/mqtthandler.h | 12 ++--- src/lib/ebus/message.cpp | 2 + src/lib/ebus/message.h | 14 ++++++ 7 files changed, 86 insertions(+), 57 deletions(-) diff --git a/src/ebusd/datahandler.cpp b/src/ebusd/datahandler.cpp index 8da17654..0cae5981 100644 --- a/src/ebusd/datahandler.cpp +++ b/src/ebusd/datahandler.cpp @@ -32,10 +32,10 @@ static const struct argp_child g_last_argp_child = {NULL, 0, NULL, 0}; /** the list of @a argp_child structures. */ static struct argp_child g_argp_children[ -#ifdef HAVE_MQTT 1 -#endif +#ifdef HAVE_MQTT +1 +#endif ]; const struct argp_child* datahandler_getargs() { @@ -54,10 +54,7 @@ bool datahandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap list* handlers) { bool success = true; #ifdef HAVE_MQTT - DataHandler* handler = mqtthandler_register(userInfo, busHandler, messages); - if (handler) { - handlers->push_back(handler); - } else { + if (!mqtthandler_register(userInfo, busHandler, messages, handlers)) { success = false; } #endif @@ -66,7 +63,7 @@ bool datahandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap void DataSink::notifyUpdate(Message* message) { if (message && message->hasLevel(m_levels)) { - m_updatedMessages[message]++; + m_updatedMessages[message->getKey()]++; } } diff --git a/src/ebusd/datahandler.h b/src/ebusd/datahandler.h index b43ccd8f..1f7965c0 100644 --- a/src/ebusd/datahandler.h +++ b/src/ebusd/datahandler.h @@ -164,8 +164,8 @@ class DataSink : virtual public DataHandler { /** the allowed access levels. */ string m_levels; - /** a map of updated @p Message instances. */ - map m_updatedMessages; + /** a map of updated @p Message keys. */ + map m_updatedMessages; }; diff --git a/src/ebusd/main.cpp b/src/ebusd/main.cpp index 078ded3c..ffec85da 100644 --- a/src/ebusd/main.cpp +++ b/src/ebusd/main.cpp @@ -867,6 +867,7 @@ void executeInstructions(MessageMap* messages, bool verbose) { result_t loadConfigFiles(MessageMap* messages, bool verbose, bool denyRecursive) { logInfo(lf_main, "loading configuration files from %s", opt.configPath); + messages->lock(); messages->clear(); s_globalTemplates.clear(); for (auto& it : s_templatesByPath) { @@ -886,6 +887,7 @@ result_t loadConfigFiles(MessageMap* messages, bool verbose, bool denyRecursive) logError(lf_main, "error reading config files: %s, last error: %s", getResultCode(result), errorDescription.c_str()); } + messages->unlock(); return RESULT_OK; } diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index db9daa21..03a834eb 100644 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -66,7 +66,10 @@ static const char* g_host = "localhost"; //!< host name of MQTT broker [localho static uint16_t g_port = 0; //!< optional port of MQTT broker, 0 to disable [0] static const char* g_username = NULL; //!< optional user name for MQTT broker (no default) static const char* g_password = NULL; //!< optional password for MQTT broker (no default) -static const char* g_topic = PACKAGE; //!< MQTT topic to use (prefix if without wildcards) [ebusd] +/** the MQTT topic string parts. */ +static vector g_topicStrs; +/** the MQTT topic field parts. */ +static vector g_topicFields; static bool g_retain = false; //!< whether to retail all topics static OutputFormat g_publishFormat = 0; //!< the OutputFormat for publishing messages @@ -78,10 +81,11 @@ static const char* g_keyfile = NULL; //!< client key file for TLS static const char* g_keypass = NULL; //!< client key file password for TLS #endif +bool parseTopic(const string& topic, vector* strs, vector* fields); /** * The MQTT argument parsing function. - * @param key the key from @a mqtt_argp_options. + * @param key the key from @a g_mqtt_argp_options. * @param arg the option argument, or NULL. * @param state the parsing state. */ @@ -126,7 +130,9 @@ static error_t mqtt_parse_opt(int key, char *arg, struct argp_state *state) { argp_error(state, "invalid mqtttopic"); return EINVAL; } - g_topic = arg; + if (!parseTopic(arg, &g_topicStrs, &g_topicFields)) { + argp_error(state, "malformed mqtttopic"); + } break; case O_RETA: // --mqttretain @@ -186,12 +192,17 @@ static error_t mqtt_parse_opt(int key, char *arg, struct argp_state *state) { static const struct argp g_mqtt_argp = { g_mqtt_argp_options, mqtt_parse_opt, NULL, NULL, NULL, NULL, NULL }; static const struct argp_child g_mqtt_argp_child = {&g_mqtt_argp, 0, "", 1}; + const struct argp_child* mqtthandler_getargs() { return &g_mqtt_argp_child; } -DataHandler* mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages) { - return new MqttHandler(userInfo, busHandler, messages); +bool mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages, + list* handlers) { + if (g_port > 0) { + handlers->push_back(new MqttHandler(userInfo, busHandler, messages)); + } + return true; } /** the known topic field names. */ @@ -216,6 +227,8 @@ bool parseTopic(const string& topic, vector* strs, vector* field size_t lastpos = 0; size_t end = topic.length(); vector columns; + strs->clear(); + fields->clear(); for (size_t pos=topic.find('%', lastpos); pos != string::npos; ) { size_t idx = knownFieldCount; size_t len = 0; @@ -281,37 +294,29 @@ void on_connect( MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages) : DataSink(userInfo, "mqtt"), DataSource(busHandler), Thread(), m_messages(messages), m_connected(false), m_lastUpdateCheckResult(".") { - bool enabled = g_port != 0; m_publishByField = false; m_mosquitto = NULL; - if (enabled && !parseTopic(g_topic, &m_topicStrs, &m_topicFields)) { - logOtherError("mqtt", "malformed topic %s", g_topic); - return; - } - if (!enabled) { - return; - } int major = -1; mosquitto_lib_version(&major, NULL, NULL); if (major != LIBMOSQUITTO_MAJOR) { logOtherError("mqtt", "invalid mosquitto version %d instead of %d", major, LIBMOSQUITTO_MAJOR); return; } - if (m_topicFields.empty()) { - if (m_topicStrs.empty()) { - m_topicStrs.push_back(""); + if (g_topicFields.empty()) { + if (g_topicStrs.empty()) { + g_topicStrs.push_back(""); } else { - string str = m_topicStrs[0]; + string str = g_topicStrs[0]; if (str.empty() || str[str.length()-1] != '/') { - m_topicStrs[0] = str+"/"; + g_topicStrs[0] = str+"/"; } } - m_topicFields.push_back("circuit"); - m_topicStrs.push_back("/"); - m_topicFields.push_back("name"); + g_topicFields.push_back("circuit"); + g_topicStrs.push_back("/"); + g_topicFields.push_back("name"); } else { - for (size_t i = 0; i < m_topicFields.size(); i++) { - if (m_topicFields[i] == "field") { + for (size_t i = 0; i < g_topicFields.size(); i++) { + if (g_topicFields[i] == "field") { m_publishByField = true; break; } @@ -431,16 +436,16 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) { size_t last = 0; string circuit, name; size_t idx; - for (idx = 0; idx < m_topicStrs.size()+1; idx++) { + for (idx = 0; idx < g_topicStrs.size()+1; idx++) { string field; string chk; - if (idx < m_topicStrs.size()) { - chk = m_topicStrs[idx]; + if (idx < g_topicStrs.size()) { + chk = g_topicStrs[idx]; pos = remain.find(chk, last); if (pos == string::npos) { return; } - } else if (idx-1 < m_topicFields.size()) { + } else if (idx-1 < g_topicFields.size()) { pos = remain.size(); } else if (last < remain.size()) { return; @@ -457,7 +462,7 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) { if (field.empty()) { return; } - string fieldName = m_topicFields[idx-1]; + string fieldName = g_topicFields[idx-1]; if (fieldName == "circuit") { circuit = field; } else if (fieldName == "name") { @@ -547,15 +552,28 @@ void MqttHandler::run() { publishTopic(uptimeTopic, updates.str()); time(&lastTaskRun); } - if (m_connected && !m_updatedMessages.empty()) { - for (const auto it : m_updatedMessages) { - updates.str(""); - updates.clear(); - updates << dec; - publishMessage(it.first, &updates); // TODO avoid using the message while reload command is executed + if (!m_updatedMessages.empty()) { + if (m_connected) { + m_messages->lock(); + for(auto it = m_updatedMessages.begin(); it != m_updatedMessages.end(); ) { + const vector* messages = m_messages->getByKey(it->first); + if (messages) { + updates.str(""); + updates.clear(); + updates << dec; + for (auto message : *messages) { + if (message->getLastChangeTime() > 0 && message->isAvailable()) { + publishMessage(message, &updates); + } + } + } + it = m_updatedMessages.erase(it); + } + m_messages->unlock(); + } else { + m_updatedMessages.clear(); } } - m_updatedMessages.clear(); } } @@ -586,16 +604,16 @@ void MqttHandler::handleTraffic() { string MqttHandler::getTopic(const Message* message, const string& suffix, const string& fieldName) { ostringstream ret; - for (size_t i = 0; i < m_topicStrs.size(); i++) { - ret << m_topicStrs[i]; + for (size_t i = 0; i < g_topicStrs.size(); i++) { + ret << g_topicStrs[i]; if (!message) { break; } - if (i < m_topicFields.size()) { - if (m_topicFields[i] == "field") { + if (i < g_topicFields.size()) { + if (g_topicFields[i] == "field") { ret << fieldName; } else { - message->dumpField(m_topicFields[i], false, &ret); + message->dumpField(g_topicFields[i], false, &ret); } } } diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h index cc2c1f34..e9afca2d 100644 --- a/src/ebusd/mqtthandler.h +++ b/src/ebusd/mqtthandler.h @@ -48,9 +48,11 @@ const struct argp_child* mqtthandler_getargs(); * @param userInfo the @a UserInfo instance. * @param busHandler the @a BusHandler instance. * @param messages the @a MessageMap instance. - * @return the create @a DataHandler, or NULL on error. + * @param handlers the @a list to which new @a DataHandler instances shall be added. + * @return true if registration was successful. */ -DataHandler* mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages); +bool mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages, + list* handlers); /** * The main class supporting MQTT data handling. @@ -121,12 +123,6 @@ class MqttHandler : public DataSink, public DataSource, public Thread { /** the @a MessageMap instance. */ MessageMap* m_messages; - /** the MQTT topic string parts. */ - vector m_topicStrs; - - /** the MQTT topic field parts. */ - vector m_topicFields; - /** the global topic prefix. */ string m_globalTopic; diff --git a/src/lib/ebus/message.cpp b/src/lib/ebus/message.cpp index 62d8c69f..cdcfa02b 100644 --- a/src/lib/ebus/message.cpp +++ b/src/lib/ebus/message.cpp @@ -2559,8 +2559,10 @@ void MessageMap::invalidateCache(Message* message) { void MessageMap::addPollMessage(bool toFront, Message* message) { if (message != NULL && message->getPollPriority() > 0) { + lock(); message->m_lastPollTime = toFront ? 0 : m_pollMessages.size(); m_pollMessages.push(message); + unlock(); } } diff --git a/src/lib/ebus/message.h b/src/lib/ebus/message.h index 7969dc78..7fca7540 100644 --- a/src/lib/ebus/message.h +++ b/src/lib/ebus/message.h @@ -26,6 +26,7 @@ #include #include #include +#include #include "lib/ebus/data.h" #include "lib/ebus/result.h" #include "lib/ebus/symbol.h" @@ -1432,6 +1433,16 @@ class MessageMap : public MappedFileReader { */ bool decodeCircuit(const string& circuit, OutputFormat outputFormat, ostringstream* output) const; + /** + * Lock this instance against simultaneous modifying access. + */ + void lock() { m_mutex.lock(); } + + /** + * Unlock this instance against simultaneous modifying access. + */ + void unlock() { m_mutex.unlock(); } + /** * Removes all @a Message instances. */ @@ -1545,6 +1556,9 @@ class MessageMap : public MappedFileReader { /** additional attributes by circuit name. */ map m_circuitData; + + /** a @a mutex for locking out changes. */ + mutex m_mutex; }; } // namespace ebusd