Mostly done with user properties in connect, publish and subscribe.

And one line about correlation data that was a bug.

Mostly done with user properties in connect, publish and subscribe.
And one line about correlation data that was a bug.
Wiebe Cazemier
1 parent b0def7dc
Showing 13 changed files with 122 additions and 39 deletions
authplugin.cpp
authplugin.h
flashmq_plugin.cpp
flashmq_plugin.h
mqtt5properties.cpp
mqtt5properties.h
mqttpacket.cpp
mqttpacket.h
publishcopyfactory.cpp
publishcopyfactory.h
session.cpp
types.cpp
types.h
@@ -278,7 +278,7 @@ void Authentication::securityCleanup(bool reloading)
 }
 AuthResult Authentication::aclCheck(const std::string &clientid, const std::string &username, const std::string &topic, const std::vector<std::string> &subtopics,
-                                    AclAccess access, char qos, bool retain)
+                                    AclAccess access, char qos, bool retain, const std::vector<std::pair<std::string, std::string>> *userProperties)
 {
     assert(subtopics.size() > 0);
@@ -322,7 +322,7 @@ AuthResult Authentication::aclCheck(const std::string &amp;clientid, const std::stri
         // gets disconnected.
         try
         {
-            FlashMQMessage msg(topic, subtopics, qos, retain);
+            FlashMQMessage msg(topic, subtopics, qos, retain, userProperties);
             return flashmq_auth_plugin_acl_check_v1(pluginData, access, clientid, username, msg);
         }
         catch (std::exception &ex)
@@ -335,7 +335,8 @@ AuthResult Authentication::aclCheck(const std::string &amp;clientid, const std::stri
     return AuthResult::error;
 }
-AuthResult Authentication::unPwdCheck(const std::string &username, const std::string &password)
+AuthResult Authentication::unPwdCheck(const std::string &username, const std::string &password,
+                                      const std::vector<std::pair<std::string, std::string>> *userProperties)
 {
     AuthResult firstResult = unPwdCheckFromMosquittoPasswordFile(username, password);
@@ -373,7 +374,7 @@ AuthResult Authentication::unPwdCheck(const std::string &amp;username, const std::st
         // gets disconnected.
         try
         {
-            return flashmq_auth_plugin_login_check_v1(pluginData, username, password);
+            return flashmq_auth_plugin_login_check_v1(pluginData, username, password, userProperties);
         }
         catch (std::exception &ex)
         {
@@ -65,7 +65,8 @@ typedef void(*F_flashmq_auth_plugin_deallocate_thread_memory_v1)(void *thread_da
 typedef void(*F_flashmq_auth_plugin_init_v1)(void *thread_data, std::unordered_map<std::string, std::string> &auth_opts, bool reloading);
 typedef void(*F_flashmq_auth_plugin_deinit_v1)(void *thread_data, std::unordered_map<std::string, std::string> &auth_opts, bool reloading);
 typedef AuthResult(*F_flashmq_auth_plugin_acl_check_v1)(void *thread_data, AclAccess access, const std::string &clientid, const std::string &username, const FlashMQMessage &msg);
-typedef AuthResult(*F_flashmq_auth_plugin_login_check_v1)(void *thread_data, const std::string &username, const std::string &password);
+typedef AuthResult(*F_flashmq_auth_plugin_login_check_v1)(void *thread_data, const std::string &username, const std::string &password,
+                                                          const std::vector<std::pair<std::string, std::string>> *userProperties);
 typedef void (*F_flashmq_auth_plugin_periodic_event)(void *thread_data);
 extern "C"
@@ -152,8 +153,9 @@ public:
     void securityInit(bool reloading);
     void securityCleanup(bool reloading);
     AuthResult aclCheck(const std::string &clientid, const std::string &username, const std::string &topic, const std::vector<std::string> &subtopics,
-                        AclAccess access, char qos, bool retain);
-    AuthResult unPwdCheck(const std::string &username, const std::string &password);
+                        AclAccess access, char qos, bool retain, const std::vector<std::pair<std::string, std::string>> *userProperties);
+    AuthResult unPwdCheck(const std::string &username, const std::string &password,
+                          const std::vector<std::pair<std::string, std::string>> *userProperties);
     void setQuitting();
     void loadMosquittoPasswordFile();
@@ -12,9 +12,11 @@ void flashmq_logf(int level, const char *str, ...)
     va_end(valist);
 }
-FlashMQMessage::FlashMQMessage(const std::string &topic, const std::vector<std::string> &subtopics, const char qos, const bool retain) :
+FlashMQMessage::FlashMQMessage(const std::string &topic, const std::vector<std::string> &subtopics, const char qos, const bool retain,
+                               const std::vector<std::pair<std::string, std::string>> *userProperties) :
     topic(topic),
     subtopics(subtopics),
+    userProperties(userProperties),
     qos(qos),
     retain(retain)
 {
@@ -16,6 +16,7 @@
 #include <string>
 #include <vector>
 #include <unordered_map>
+#include <memory>
 #define FLASHMQ_PLUGIN_VERSION 1
@@ -72,10 +73,12 @@ struct FlashMQMessage
 {
     const std::string &topic;
     const std::vector<std::string> &subtopics;
+    const std::vector<std::pair<std::string, std::string>> *userProperties;
     const char qos;
     const bool retain;
-    FlashMQMessage(const std::string &topic, const std::vector<std::string> &subtopics, const char qos, const bool retain);
+    FlashMQMessage(const std::string &topic, const std::vector<std::string> &subtopics, const char qos, const bool retain,
+                   const std::vector<std::pair<std::string, std::string>> *userProperties);
 };
 /**
@@ -177,7 +180,8 @@ void flashmq_auth_plugin_periodic_event(void *thread_data);
  * Note that there is a setting 'auth_plugin_serialize_auth_checks'. Use only as a last resort if your plugin is not
  * thread-safe. It will negate much of FlashMQ's multi-core model.
  */
-AuthResult flashmq_auth_plugin_login_check(void *thread_data, const std::string &username, const std::string &password);
+AuthResult flashmq_auth_plugin_login_check(void *thread_data, const std::string &username, const std::string &password,
+                                           const std::vector<std::pair<std::string, std::string>> *userProperties);
 /**
  * @brief flashmq_auth_plugin_acl_check is called on publish, deliver and subscribe.
@@ -2,6 +2,7 @@
 #include "cstring"
 #include "vector"
+#include "cassert"
 #include "exceptions.h"
@@ -38,6 +39,11 @@ void Mqtt5PropertyBuilder::clearClientSpecificBytes()
     clientSpecificBytes.clear();
 }
+std::shared_ptr<std::vector<std::pair<std::string, std::string>>> Mqtt5PropertyBuilder::getUserProperties() const
+{
+    return this->userProperties;
+}
+
 void Mqtt5PropertyBuilder::writeSessionExpiry(uint32_t val)
 {
     writeUint32(Mqtt5Properties::SessionExpiryInterval, val, genericBytes);
@@ -103,9 +109,15 @@ void Mqtt5PropertyBuilder::writeResponseTopic(const std::string &amp;str)
     writeStr(Mqtt5Properties::ResponseTopic, str);
 }
-void Mqtt5PropertyBuilder::writeUserProperty(const std::string &key, const std::string &value)
+void Mqtt5PropertyBuilder::writeUserProperty(std::string &&key, std::string &&value)
 {
     write2Str(Mqtt5Properties::UserProperty, key, value);
+
+    if (!this->userProperties)
+        this->userProperties = std::make_shared<std::vector<std::pair<std::string, std::string>>>();
+
+    std::pair<std::string, std::string> pair(std::move(key), std::move(value));
+    this->userProperties->push_back(std::move(pair));
 }
 void Mqtt5PropertyBuilder::writeCorrelationData(const std::string &correlationData)
@@ -113,6 +125,15 @@ void Mqtt5PropertyBuilder::writeCorrelationData(const std::string &amp;correlationDa
     writeStr(Mqtt5Properties::CorrelationData, correlationData);
 }
+void Mqtt5PropertyBuilder::setNewUserProperties(const std::shared_ptr<std::vector<std::pair<std::string, std::string>>> &userProperties)
+{
+    assert(!this->userProperties);
+    assert(this->genericBytes.empty());
+    assert(this->clientSpecificBytes.empty());
+
+    this->userProperties = userProperties;
+}
+
 void Mqtt5PropertyBuilder::writeUint32(Mqtt5Properties prop, const uint32_t x, std::vector<char> &target)
 {
     size_t pos = target.size();
@@ -10,6 +10,7 @@ class Mqtt5PropertyBuilder
 {
     std::vector<char> genericBytes;
     std::vector<char> clientSpecificBytes; // only relevant for publishes
+    std::shared_ptr<std::vector<std::pair<std::string, std::string>>> userProperties;
     VariableByteInt length;
     void writeUint32(Mqtt5Properties prop, const uint32_t x, std::vector<char> &target);
@@ -25,6 +26,7 @@ public:
     const std::vector<char> &getGenericBytes() const;
     const std::vector<char> &getclientSpecificBytes() const;
     void clearClientSpecificBytes();
+    std::shared_ptr<std::vector<std::pair<std::string, std::string>>> getUserProperties() const;
     void writeSessionExpiry(uint32_t val);
     void writeReceiveMax(uint16_t val);
@@ -39,8 +41,9 @@ public:
     void writePayloadFormatIndicator(uint8_t val);
     void writeMessageExpiryInterval(uint32_t val);
     void writeResponseTopic(const std::string &str);
-    void writeUserProperty(const std::string &key, const std::string &value);
+    void writeUserProperty(std::string &&key, std::string &&value);
     void writeCorrelationData(const std::string &correlationData);
+    void setNewUserProperties(const std::shared_ptr<std::vector<std::pair<std::string, std::string>>> &userProperties);
 };
 #endif // MQTT5PROPERTIES_H
@@ -132,6 +132,15 @@ MqttPacket::MqttPacket(const ProtocolVersion protocolVersion, const Publish &amp;_pu
     if (protocolVersion >= ProtocolVersion::Mqtt5)
     {
+        // Step 1: make certain properties available as objects, because FlashMQ needs access to them for internal logic.
+        if (_publish.propertyBuilder)
+        {
+            this->publishData.constructPropertyBuilder();
+            this->publishData.propertyBuilder->setNewUserProperties(_publish.propertyBuilder->getUserProperties());
+        }
+
+        // Step 2: this line will make sure the whole byte array containing all properties as flat bytes is present in the 'bites' vector,
+        // which is sent to the subscribers.
         writeProperties(_publish.propertyBuilder);
     }
@@ -368,13 +377,8 @@ void MqttPacket::handleConnect()
                     request_problem_information = !!readByte();
                     break;
                 case Mqtt5Properties::UserProperty:
-                {
-                    const uint16_t len = readTwoBytesToUInt16();
-                    readBytes(len);
-                    const uint16_t len2 = readTwoBytesToUInt16();
-                    readBytes(len2);
+                    readUserProperty();
                     break;
-                }
                 case Mqtt5Properties::AuthenticationMethod:
                 {
                     const uint16_t len = readTwoBytesToUInt16();
@@ -407,7 +411,7 @@ void MqttPacket::handleConnect()
         {
             if (protocolVersion == ProtocolVersion::Mqtt5)
             {
-                willpublish.propertyBuilder = std::make_unique<Mqtt5PropertyBuilder>();
+                willpublish.constructPropertyBuilder();
                 const size_t proplen = decodeVariableByteIntAtPos();
                 const size_t prop_end_at = pos + proplen;
@@ -450,15 +454,16 @@ void MqttPacket::handleConnect()
                     {
                         const uint16_t len = readTwoBytesToUInt16();
                         const std::string correlationData(readBytes(len), len);
-                        publishData.propertyBuilder->writeCorrelationData(correlationData);
+                        willpublish.propertyBuilder->writeCorrelationData(correlationData);
                         break;
                     }
                     case Mqtt5Properties::UserProperty:
                     {
-                        const uint16_t len = readTwoBytesToUInt16();
-                        readBytes(len);
-                        const uint16_t len2 = readTwoBytesToUInt16();
-                        readBytes(len2);
+                        const uint16_t lenKey = readTwoBytesToUInt16();
+                        std::string userPropKey(readBytes(lenKey), lenKey);
+                        const uint16_t lenVal = readTwoBytesToUInt16();
+                        std::string userPropVal(readBytes(lenVal), lenVal);
+                        willpublish.propertyBuilder->writeUserProperty(std::move(userPropKey), std::move(userPropVal));
                         break;
                     }
                     default:
@@ -554,7 +559,7 @@ void MqttPacket::handleConnect()
             sender->setDisconnectReason("Invalid username character");
             accessGranted = false;
         }
-        else if (authentication.unPwdCheck(username, password) == AuthResult::success)
+        else if (authentication.unPwdCheck(username, password, getUserProperties()) == AuthResult::success)
         {
             accessGranted = true;
         }
@@ -644,13 +649,8 @@ void MqttPacket::handleSubscribe()
                 decodeVariableByteIntAtPos();
                 break;
             case Mqtt5Properties::UserProperty:
-            {
-                const uint16_t len = readTwoBytesToUInt16();
-                readBytes(len);
-                const uint16_t len2 = readTwoBytesToUInt16();
-                readBytes(len2);
+                readUserProperty();
                 break;
-            }
             default:
                 throw ProtocolError("Invalid subscribe property.");
             }
@@ -678,7 +678,7 @@ void MqttPacket::handleSubscribe()
         std::vector<std::string> subtopics;
         splitTopic(topic, subtopics);
-        if (authentication.aclCheck(sender->getClientId(), sender->getUsername(), topic, subtopics, AclAccess::subscribe, qos, false) == AuthResult::success)
+        if (authentication.aclCheck(sender->getClientId(), sender->getUsername(), topic, subtopics, AclAccess::subscribe, qos, false, getUserProperties()) == AuthResult::success)
         {
             logger->logf(LOG_SUBSCRIBE, "Client '%s' subscribed to '%s' QoS %d", sender->repr().c_str(), topic.c_str(), qos);
             sender->getThreadData()->getSubscriptionStore()->addSubscription(sender, topic, subtopics, qos);
@@ -821,7 +821,7 @@ void MqttPacket::handlePublish()
         const size_t prop_end_at = pos + proplen;
         if (proplen > 0)
-            publishData.propertyBuilder = std::make_shared<Mqtt5PropertyBuilder>();
+            publishData.constructPropertyBuilder();
         while (pos < prop_end_at)
         {
@@ -854,10 +854,10 @@ void MqttPacket::handlePublish()
             case Mqtt5Properties::UserProperty:
             {
                 const uint16_t lenKey = readTwoBytesToUInt16();
-                const std::string userPropKey(readBytes(lenKey), lenKey);
+                std::string userPropKey(readBytes(lenKey), lenKey);
                 const uint16_t lenVal = readTwoBytesToUInt16();
-                const std::string userPropVal(readBytes(lenVal), lenVal);
-                publishData.propertyBuilder->writeUserProperty(userPropKey, userPropVal);
+                std::string userPropVal(readBytes(lenVal), lenVal);
+                publishData.propertyBuilder->writeUserProperty(std::move(userPropKey), std::move(userPropVal));
                 break;
             }
             case Mqtt5Properties::SubscriptionIdentifier:
@@ -882,7 +882,7 @@ void MqttPacket::handlePublish()
     payloadStart = pos;
     Authentication &authentication = *ThreadGlobals::getAuth();
-    if (authentication.aclCheck(sender->getClientId(), sender->getUsername(), publishData.topic, publishData.subtopics, AclAccess::write, qos, retain) == AuthResult::success)
+    if (authentication.aclCheck(sender->getClientId(), sender->getUsername(), publishData.topic, publishData.subtopics, AclAccess::write, qos, retain, getUserProperties()) == AuthResult::success)
     {
         if (retain)
         {
@@ -1176,6 +1176,26 @@ size_t MqttPacket::decodeVariableByteIntAtPos()
     return value;
 }
+void MqttPacket::readUserProperty()
+{
+    this->publishData.constructPropertyBuilder();
+
+    const uint16_t len = readTwoBytesToUInt16();
+    std::string key(readBytes(len), len);
+    const uint16_t len2 = readTwoBytesToUInt16();
+    std::string value(readBytes(len2), len2);
+
+    this->publishData.propertyBuilder->writeUserProperty(std::move(key), std::move(value));
+}
+
+const std::vector<std::pair<std::string, std::string>> *MqttPacket::getUserProperties() const
+{
+    if (this->publishData.propertyBuilder)
+        return this->publishData.propertyBuilder->getUserProperties().get();
+
+    return nullptr;
+}
+
 bool MqttPacket::getRetain() const
 {
     return (first_byte & 0b00000001);
@@ -67,11 +67,12 @@ class MqttPacket
     uint32_t readFourBytesToUint32();
     size_t remainingAfterPos();
     size_t decodeVariableByteIntAtPos();
+    void readUserProperty();
     void calculateRemainingLength();
     void pubCommonConstruct(const uint16_t packet_id, PacketType packetType, uint8_t firstByteDefaultBits = 0);
-    MqttPacket(const MqttPacket &other) = default;
+    MqttPacket(const MqttPacket &other) = delete;
 public:
     PacketType packetType = PacketType::Reserved;
     MqttPacket(CirBuf &buf, size_t packet_len, size_t fixed_header_length, std::shared_ptr<Client> &sender); // Constructor for parsing incoming packets.
@@ -123,6 +124,7 @@ public:
     void setRetain();
     const Publish &getPublishData();
     bool containsClientSpecificProperties() const;
+    const std::vector<std::pair<std::string, std::string>> *getUserProperties() const;
 };
 #endif // MQTTPACKET_H
@@ -124,3 +124,20 @@ std::shared_ptr&lt;Client&gt; PublishCopyFactory::getSender()
         return packet->getSender();
     return std::shared_ptr<Client>(0);
 }
+
+const std::vector<std::pair<std::string, std::string> > *PublishCopyFactory::getUserProperties() const
+{
+    if (packet)
+    {
+        return packet->getUserProperties();
+    }
+
+    assert(publish);
+
+    if (publish->propertyBuilder)
+    {
+        return publish->propertyBuilder->getUserProperties().get();
+    }
+
+    return nullptr;
+}
@@ -36,6 +36,8 @@ public:
     bool getRetain() const;
     Publish getNewPublish() const;
     std::shared_ptr<Client> getSender();
+    const std::vector<std::pair<std::string, std::string>> *getUserProperties() const;
+
 };
 #endif // PUBLISHCOPYFACTORY_H
@@ -160,7 +160,7 @@ void Session::writePacket(PublishCopyFactory &amp;copyFactory, const char max_qos, u
     Authentication *_auth = ThreadGlobals::getAuth();
     assert(_auth);
     Authentication &auth = *_auth;
-    if (auth.aclCheck(client_id, username, copyFactory.getTopic(), copyFactory.getSubtopics(), AclAccess::read, effectiveQos, copyFactory.getRetain()) == AuthResult::success)
+    if (auth.aclCheck(client_id, username, copyFactory.getTopic(), copyFactory.getSubtopics(), AclAccess::read, effectiveQos, copyFactory.getRetain(), copyFactory.getUserProperties()) == AuthResult::success)
     {
         std::shared_ptr<Client> c = makeSharedClient();
         if (effectiveQos == 0)
@@ -141,6 +141,14 @@ void PublishBase::setClientSpecificProperties()
         propertyBuilder->writeMessageExpiryInterval(newExpiresAfter.count());
 }
+void PublishBase::constructPropertyBuilder()
+{
+    if (this->propertyBuilder)
+        return;
+
+    this->propertyBuilder = std::make_shared<Mqtt5PropertyBuilder>();
+}
+
 Publish::Publish(const Publish &other) :
     PublishBase(other)
 {
@@ -212,6 +212,7 @@ public:
     PublishBase(const std::string &topic, const std::string &payload, char qos);
     size_t getLengthWithoutFixedHeader() const;
     void setClientSpecificProperties();
+    void constructPropertyBuilder();
 };
 class Publish : public PublishBase