Cpp agent mqtt sink

agent_lib/CMakeLists.txt

@@ -131,6 +131,7 @@ set(AGENT_SOURCES
 
 # src/mqtt HEADER_FILE_ONLY
 
+	"${CMAKE_CURRENT_SOURCE_DIR}/../src/mqtt/mqtt_authorization.hpp"
         "${CMAKE_CURRENT_SOURCE_DIR}/../src/mqtt/mqtt_client.hpp"
         "${CMAKE_CURRENT_SOURCE_DIR}/../src/mqtt/mqtt_server.hpp"
         "${CMAKE_CURRENT_SOURCE_DIR}/../src/mqtt/mqtt_client_impl.hpp"

src/mqtt/mqtt_authorization.hpp

@@ -0,0 +1,108 @@
+//
+// Copyright Copyright 2009-2022, AMT – The Association For Manufacturing Technology (“AMT”)
+// All rights reserved.
+//
+//    Licensed under the Apache License, Version 2.0 (the "License");
+//    you may not use this file except in compliance with the License.
+//    You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+//    Unless required by applicable law or agreed to in writing, software
+//    distributed under the License is distributed on an "AS IS" BASIS,
+//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//    See the License for the specific language governing permissions and
+//    limitations under the License.
+//
+
+#pragma once
+
+#include "source/adapter/adapter.hpp"
+#include "source/adapter/adapter_pipeline.hpp"
+
+using namespace std;
+using namespace mtconnect;
+using namespace mtconnect::configuration;
+
+namespace mtconnect {
+  namespace mqtt_client {
+
+    class MqttTopicPermission
+    {
+      enum AuthorizationType
+      {
+        Allow,
+        Deny
+      };
+
+      enum TopicMode
+      {
+        Subscribe,
+        Publish,
+        Both
+      };
+
+      void MqttTopicPermission(const std::string &topic, const std::string& clientId) 
+      {
+        m_topic = topic;
+        m_clientId = clientId;
+        m_type = AuthorizationType::Allow;
+        m_mode = TopicMode::Subscribe;
+      }
+
+      void MqttTopicPermission(const std::string& topic, const std::string& clientId,
+          AuthorizationType type)
+      {
+        m_topic = topic;
+        m_clientId = clientId;
+        m_type = type;
+        m_mode = TopicMode::Subscribe;
+      }
+
+      void MqttTopicPermission(const std::string& topic, const std::string& clientId,
+          AuthorizationType type, TopicMode mode)
+      {
+        m_topic = topic;
+        m_clientId = clientId;
+        m_type = type;
+        m_mode = mode;
+      }
+
+    protected:
+      TopicMode m_mode;
+      AuthorizationType m_type;
+      std::string m_clientId;
+      std::string m_topic;
+    };
+
+    class MqttAuthorization : public std::enable_shared_from_this<MqttAuthorization>
+    {
+    public:
+      MqttAuthorization(const ConfigOptions &options) : m_options(options)
+      {
+        m_clientId = GetOption<std::string>(options, configuration::MqttClientId);
+        m_username = GetOption<std::string>(options, configuration::MqttUserName);
+        m_password = GetOption<std::string>(options, configuration::MqttPassword);
+      }
+
+      virtual ~MqttAuthorization() = default;
+
+      bool checkCredentials() 
+      {          
+        if (!m_username && !m_password)
+        {
+          LOG(error) << "MQTT USERNAME_OR_PASSWORD are Not Available";
+          return false;
+        }
+
+        return true;          
+      }
+
+    protected:
+      std::optional<std::string> m_username;
+      std::optional<std::string> m_password;
+      std::string m_clientId;
+      ConfigOptions m_options;
+    };
+  }  // namespace MqttAuthorization
+}  // namespace mtconnect

src/mqtt/mqtt_client_impl.hpp

@@ -134,7 +134,7 @@ namespace mtconnect {
             else
             {
               LOG(debug) << "No connect handler, setting connected";
-              m_connected = true;
+              m_connected = true;              
             }
           }
           else
@@ -224,9 +224,9 @@ namespace mtconnect {
         }
 
         LOG(debug) << "Subscribing to topic: " << topic;
-        m_clientId = derived().getClient()->acquire_unique_packet_id();
+        m_packetId = derived().getClient()->acquire_unique_packet_id();
         derived().getClient()->async_subscribe(
-            m_clientId, topic.c_str(), mqtt::qos::at_least_once, [topic](mqtt::error_code ec) {
+            m_packetId, topic.c_str(), mqtt::qos::at_least_once, [topic](mqtt::error_code ec) {
               if (ec)
               {
                 LOG(error) << "Subscribe failed: " << topic << ": " << ec.message();
@@ -251,9 +251,9 @@ namespace mtconnect {
           return false;
         }
 
-        m_clientId = derived().getClient()->acquire_unique_packet_id();
+        m_packetId = derived().getClient()->acquire_unique_packet_id();
         derived().getClient()->async_publish(
-            m_clientId, topic, payload, mqtt::qos::at_least_once | mqtt::retain::yes,
+            m_packetId, topic, payload, mqtt::qos::at_least_once | mqtt::retain::yes,
             [topic](mqtt::error_code ec) {
               if (ec)
               {
@@ -332,7 +332,7 @@ namespace mtconnect {
 
       unsigned int m_port {1883};
 
-      std::uint16_t m_clientId {0};
+      std::uint16_t m_packetId {0};
 
       std::optional<std::string> m_username;
 
@@ -425,16 +425,7 @@ namespace mtconnect {
           if (cacert)
           {
             m_client->get_ssl_context().load_verify_file(*cacert);
-          }
-          auto private_key = GetOption<string>(m_options, configuration::MqttPrivateKey);
-          auto cert = GetOption<string>(m_options, configuration::MqttCert);
-          if (private_key && cert)
-          {
-            m_client->get_ssl_context().set_verify_mode(boost::asio::ssl::verify_peer);
-            m_client->get_ssl_context().use_certificate_chain_file(*cert);
-            m_client->get_ssl_context().use_private_key_file(*private_key,
-                                                             boost::asio::ssl::context::pem);
-          }
+          }         
         }
 
         return m_client;

src/mqtt/mqtt_server_impl.hpp

@@ -300,18 +300,39 @@ namespace mtconnect {
           boost::asio::ssl::context ctx(boost::asio::ssl::context::tlsv12);
           ctx.set_options(boost::asio::ssl::context::default_workarounds |
                           boost::asio::ssl::context::single_dh_use);
-          auto serverPrivateKey = GetOption<string>(m_options, configuration::TlsPrivateKey);
-          auto serverCert = GetOption<string>(m_options, configuration::TlsCertificateChain);
-          ctx.use_certificate_chain_file(*serverCert);
-          // ctx.use_tmp_dh_file(*GetOption<string>(m_options, configuration::TlsDHKey));
-          ctx.use_private_key_file(*serverPrivateKey, boost::asio::ssl::context::pem);
 
-          if (HasOption(m_options, configuration::TlsCertificatePassword))
+          if (HasOption(m_options, configuration::TlsCertificateChain) &&
+              HasOption(m_options, configuration::TlsPrivateKey) && 
+              HasOption(m_options, configuration::TlsDHKey))
           {
-            ctx.set_password_callback(
-                [this](size_t, boost::asio::ssl::context_base::password_purpose) -> string {
-                  return *GetOption<string>(m_options, configuration::TlsCertificatePassword);
-                });
+            LOG(info) << "Server: Initializing TLS support";
+            if (HasOption(m_options, configuration::TlsCertificatePassword))
+            {
+              ctx.set_password_callback(
+                  [this](size_t, boost::asio::ssl::context_base::password_purpose) -> string {
+                    return *GetOption<string>(m_options, configuration::TlsCertificatePassword);
+                  });
+            }
+
+            auto serverPrivateKey = GetOption<string>(m_options, configuration::TlsPrivateKey);
+            auto serverCert = GetOption<string>(m_options, configuration::TlsCertificateChain);
+            auto serverDHKey = GetOption<string>(m_options, configuration::TlsDHKey);
+
+            ctx.use_certificate_chain_file(*serverCert);
+            ctx.use_private_key_file(*serverPrivateKey, boost::asio::ssl::context::pem);
+            ctx.use_tmp_dh_file(*serverDHKey);
+
+            if (IsOptionSet(m_options, configuration::TlsVerifyClientCertificate))
+            {
+              LOG(info) << "Server: Will only accept client connections with valid certificates";
+
+              ctx.set_verify_mode(boost::asio::ssl::verify_peer);
+              if (HasOption(m_options, configuration::TlsClientCAs))
+              {
+                LOG(info) << "Server: Adding Client Certificates.";
+                ctx.load_verify_file(*GetOption<string>(m_options, configuration::TlsClientCAs));
+              }
+            }
           }
           m_server.emplace(boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), m_port),
                            std::move(ctx), m_ioContext);
@@ -346,24 +367,15 @@ namespace mtconnect {
           boost::asio::ssl::context ctx(boost::asio::ssl::context::tlsv12);
           ctx.set_options(boost::asio::ssl::context::default_workarounds |
                           boost::asio::ssl::context::single_dh_use);
-          auto serverPrivateKey = GetOption<string>(m_options, configuration::TlsPrivateKey);
-          auto serverCert = GetOption<string>(m_options, configuration::TlsCertificateChain);
-          ctx.use_certificate_chain_file(*serverCert);
-          // ctx.use_tmp_dh_file(*GetOption<string>(m_options, configuration::TlsDHKey));
-          ctx.use_private_key_file(*serverPrivateKey, boost::asio::ssl::context::pem);
 
-          /*if (IsOptionSet(m_options, configuration::TlsVerifyClientCertificate))
+          if (HasOption(m_options, configuration::TlsCertificateChain) &&
+              HasOption(m_options, configuration::TlsPrivateKey))
           {
-            LOG(info) << "Server: Will only accept client connections with valid certificates";
-
-            ctx.set_verify_mode(boost::asio::ssl::verify_peer |
-                                boost::asio::ssl::verify_fail_if_no_peer_cert);
-            if (HasOption(m_options, configuration::MqttCaCert))
-            {
-              LOG(info) << "Server: Adding Client Certificates.";
-              ctx.load_verify_file(*GetOption<string>(m_options, configuration::MqttCaCert));
-            }
-          }*/
+            auto serverPrivateKey = GetOption<string>(m_options, configuration::TlsPrivateKey);
+            auto serverCert = GetOption<string>(m_options, configuration::TlsCertificateChain);
+            ctx.use_certificate_chain_file(*serverCert);
+            ctx.use_private_key_file(*serverPrivateKey, boost::asio::ssl::context::pem);
+          }
 
           m_server.emplace(boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), m_port),
                            std::move(ctx), m_ioContext);