Working auth with plugin

Wiebe Cazemier
1 parent 543d4451
Showing 9 changed files with 76 additions and 37 deletions
authplugin.cpp
logger.cpp
logger.h
main.cpp
mainapp.cpp
mqttpacket.cpp
mqttpacket.h
threaddata.cpp
threaddata.h
@@ -12,7 +12,7 @@ void mosquitto_log_printf(int level, const char *fmt, ...)
     Logger *logger = Logger::getInstance();
     va_list valist;
     va_start(valist, fmt);
-    logger->logf(level, fmt);
+    logger->logf(level, fmt, valist);
     va_end(valist);
 }
@@ -40,7 +40,15 @@ Logger *Logger::getInstance()
 void Logger::logf(int level, const char *str, ...)
 {
-    if (level > curLogLevel)
+    va_list valist;
+    va_start(valist, str);
+    this->logf(level, str, valist);
+    va_end(valist);
+}
+
+void Logger::logf(int level, const char *str, va_list valist)
+{
+    if (level > curLogLevel) // TODO: wrong: bitmap based
         return;
     std::lock_guard<std::mutex> locker(logMutex);
@@ -55,8 +63,6 @@ void Logger::logf(int level, const char *str, ...)
     const std::string s = oss.str();
     const char *logfmtstring = s.c_str();
-    va_list valist;
-    va_start(valist, str);
     if (this->file)
     {
         vfprintf(this->file, logfmtstring, valist);
@@ -75,5 +81,4 @@ void Logger::logf(int level, const char *str, ...)
         fflush(stdout);
 #endif
     }
-    va_end(valist);
 }
@@ -25,6 +25,7 @@ class Logger
 public:
     static Logger *getInstance();
+    void logf(int level, const char *str, va_list args);
     void logf(int level, const char *str, ...);
 };
@@ -65,6 +65,9 @@ int main()
 {
     try
     {
+        Logger *logger = Logger::getInstance();
+        logger->logf(LOG_NOTICE, "Starting FlashMQ");
+
         mainApp = MainApp::getMainApp();
         check<std::runtime_error>(register_signal_handers());
         mainApp->start();
@@ -17,6 +17,21 @@ void do_thread_work(ThreadData *threadData)
     std::vector<MqttPacket> packetQueueIn;
     time_t lastKeepAliveCheck = time(NULL);
+    Logger *logger = Logger::getInstance();
+
+    try
+    {
+        logger->logf(LOG_NOTICE, "Thread %d doing auth init.", threadData->threadnr);
+        threadData->initAuthPlugin();
+    }
+    catch(std::exception &ex)
+    {
+        logger->logf(LOG_ERR, "Error initializing auth back-end: %s", ex.what());
+        threadData->running = false;
+        MainApp *instance = MainApp::getMainApp();
+        instance->quit();
+    }
+
     while (threadData->running)
     {
         int fdcount = epoll_wait(epoll_fd, events, MAX_EVENTS, 100);
@@ -83,7 +98,7 @@ void do_thread_work(ThreadData *threadData)
         {
             try
             {
-                packet.handle(threadData->getSubscriptionStore());
+                packet.handle();
             }
             catch (std::exception &ex)
             {
@@ -209,17 +224,16 @@ void MainApp::start()
         }
     }
+    for(std::shared_ptr<ThreadData> &thread : threads)
+    {
+        thread->quit();
+    }
+
     close(listen_fd);
 }
 void MainApp::quit()
 {
     std::cout << "Quitting FlashMQ" << std::endl;
-
     running = false;
-
-    for(std::shared_ptr<ThreadData> &thread : threads)
-    {
-        thread->quit();
-    }
 }
@@ -98,7 +98,7 @@ MqttPacket::MqttPacket(const Publish &amp;publish) :
     calculateRemainingLength();
 }
-void MqttPacket::handle(std::shared_ptr<SubscriptionStore> &subscriptionStore)
+void MqttPacket::handle()
 {
     if (packetType != PacketType::CONNECT)
     {
@@ -113,9 +113,9 @@ void MqttPacket::handle(std::shared_ptr&lt;SubscriptionStore&gt; &amp;subscriptionStore)
     else if (packetType == PacketType::PINGREQ)
         sender->writePingResp();
     else if (packetType == PacketType::SUBSCRIBE)
-        handleSubscribe(subscriptionStore);
+        handleSubscribe();
     else if (packetType == PacketType::PUBLISH)
-        handlePublish(subscriptionStore);
+        handlePublish();
 }
 void MqttPacket::handleConnect()
@@ -146,7 +146,7 @@ void MqttPacket::handleConnect()
             MqttPacket response(connAck);
             sender->setReadyForDisconnect();
             sender->writeMqttPacket(response);
-            std::cout << "Rejecting because of invalid protocol version: " << sender->repr() << std::endl;
+            logger->logf(LOG_ERR, "Rejecting because of invalid protocol version: %s", sender->repr().c_str());
             return;
         }
@@ -200,7 +200,7 @@ void MqttPacket::handleConnect()
             MqttPacket response(connAck);
             sender->setReadyForDisconnect();
             sender->writeMqttPacket(response);
-            std::cout << "Client ID, username or passwords has invalid UTF8: " << sender->repr() << std::endl;
+            logger->logf(LOG_ERR, "Client ID, username or passwords has invalid UTF8: ", client_id.c_str());
             return;
         }
@@ -212,19 +212,29 @@ void MqttPacket::handleConnect()
             MqttPacket response(connAck);
             sender->setReadyForDisconnect();
             sender->writeMqttPacket(response);
-            std::cout << "ClientID has + or # in the id: " << sender->repr() << std::endl;
+            logger->logf(LOG_ERR, "ClientID '%s' has + or # in the id:", client_id.c_str());
             return;
         }
         sender->setClientProperties(client_id, username, true, keep_alive);
         sender->setWill(will_topic, will_payload, will_retain, will_qos);
-        sender->setAuthenticated(true);
-        std::cout << "Connect: " << sender->repr() << std::endl;
-
-        ConnAck connAck(ConnAckReturnCodes::Accepted);
-        MqttPacket response(connAck);
-        sender->writeMqttPacket(response);
+        if (sender->getThreadData()->authPlugin.unPwdCheck(username, password) == AuthResult::success)
+        {
+            sender->setAuthenticated(true);
+            ConnAck connAck(ConnAckReturnCodes::Accepted);
+            MqttPacket response(connAck);
+            sender->writeMqttPacket(response);
+            logger->logf(LOG_NOTICE, "User %s logged in successfully", username.c_str());
+        }
+        else
+        {
+            ConnAck connDeny(ConnAckReturnCodes::NotAuthorized);
+            MqttPacket response(connDeny);
+            sender->setReadyForDisconnect();
+            sender->writeMqttPacket(response);
+            logger->logf(LOG_NOTICE, "User %s access denied", username.c_str());
+        }
     }
     else
     {
@@ -232,7 +242,7 @@ void MqttPacket::handleConnect()
     }
 }
-void MqttPacket::handleSubscribe(std::shared_ptr<SubscriptionStore> &subscriptionStore)
+void MqttPacket::handleSubscribe()
 {
     uint16_t packet_id = readTwoBytesToUInt16();
@@ -245,7 +255,7 @@ void MqttPacket::handleSubscribe(std::shared_ptr&lt;SubscriptionStore&gt; &amp;subscriptio
         if (qos > 0)
             throw NotImplementedException("QoS not implemented");
         std::cout << sender->repr() << " Subscribed to " << topic << std::endl;
-        subscriptionStore->addSubscription(sender, topic);
+        sender->getThreadData()->getSubscriptionStore()->addSubscription(sender, topic);
         subs_reponse_codes.push_back(qos);
     }
@@ -254,7 +264,7 @@ void MqttPacket::handleSubscribe(std::shared_ptr&lt;SubscriptionStore&gt; &amp;subscriptio
     sender->writeMqttPacket(response);
 }
-void MqttPacket::handlePublish(std::shared_ptr<SubscriptionStore> &subscriptionStore)
+void MqttPacket::handlePublish()
 {
     uint16_t variable_header_length = readTwoBytesToUInt16();
@@ -287,7 +297,7 @@ void MqttPacket::handlePublish(std::shared_ptr&lt;SubscriptionStore&gt; &amp;subscriptionS
         size_t payload_length = remainingAfterPos();
         std::string payload(readBytes(payload_length), payload_length);
-        subscriptionStore->setRetainedMessage(topic, payload, qos);
+        sender->getThreadData()->getSubscriptionStore()->setRetainedMessage(topic, payload, qos);
     }
     // Set dup flag to 0, because that must not be propagated [MQTT-3.3.1-3].
@@ -295,7 +305,7 @@ void MqttPacket::handlePublish(std::shared_ptr&lt;SubscriptionStore&gt; &amp;subscriptionS
     bites[0] &= 0b11110110;
     // For the existing clients, we can just write the same packet back out, with our small alterations.
-    subscriptionStore->queuePacketAtSubscribers(topic, *this, sender);
+    sender->getThreadData()->getSubscriptionStore()->queuePacketAtSubscribers(topic, *this, sender);
 }
 void MqttPacket::calculateRemainingLength()
@@ -13,6 +13,7 @@
 #include "types.h"
 #include "subscriptionstore.h"
 #include "cirbuf.h"
+#include "logger.h"
 struct RemainingLength
 {
@@ -31,6 +32,7 @@ class MqttPacket
     char first_byte = 0;
     size_t pos = 0;
     ProtocolVersion protocolVersion = ProtocolVersion::None;
+    Logger *logger = Logger::getInstance();
     char *readBytes(size_t length);
     char readByte();
@@ -50,11 +52,11 @@ public:
     MqttPacket(const SubAck &subAck);
     MqttPacket(const Publish &publish);
-    void handle(std::shared_ptr<SubscriptionStore> &subscriptionStore);
+    void handle();
     void handleConnect();
-    void handleSubscribe(std::shared_ptr<SubscriptionStore> &subscriptionStore);
+    void handleSubscribe();
     void handlePing();
-    void handlePublish(std::shared_ptr<SubscriptionStore> &subscriptionStore);
+    void handlePublish();
     size_t getSizeIncludingNonPresentHeader() const;
     const std::vector<char> &getBites() const { return bites; }
@@ -11,10 +11,6 @@ ThreadData::ThreadData(int threadnr, std::shared_ptr&lt;SubscriptionStore&gt; &amp;subscri
     logger = Logger::getInstance();
     epollfd = check<std::runtime_error>(epoll_create(999));
-
-    authPlugin.loadPlugin(confFileParser.getAuthPluginPath());
-    authPlugin.init();
-    authPlugin.securityInit(false);
 }
 void ThreadData::moveThreadHere(std::thread &&thread)
@@ -100,6 +96,13 @@ bool ThreadData::doKeepAliveCheck()
     return true;
 }
+void ThreadData::initAuthPlugin()
+{
+    authPlugin.loadPlugin(confFileParser.getAuthPluginPath());
+    authPlugin.init();
+    authPlugin.securityInit(false);
+}
+
 void ThreadData::reload()
 {
     try
@@ -26,10 +26,10 @@ class ThreadData
     std::mutex clients_by_fd_mutex;
     std::shared_ptr<SubscriptionStore> subscriptionStore;
     ConfigFileParser &confFileParser;
-    AuthPlugin authPlugin;
     Logger *logger;
 public:
+    AuthPlugin authPlugin;
     bool running = true;
     std::thread thread;
     int threadnr = 0;
@@ -48,6 +48,7 @@ public:
     std::shared_ptr<SubscriptionStore> &getSubscriptionStore();
     bool doKeepAliveCheck();
+    void initAuthPlugin();
     void reload();
 };