Cpp agent mqtt sink

src/configuration/config_options.hpp

@@ -68,6 +68,8 @@ namespace mtconnect {
     DECLARE_CONFIGURATION(MqttPort);
     DECLARE_CONFIGURATION(MqttHost);
     DECLARE_CONFIGURATION(MqttConnectInterval);
+    DECLARE_CONFIGURATION(MqttUserName);
+    DECLARE_CONFIGURATION(MqttPassword);
 
     // Adapter Configuration
     DECLARE_CONFIGURATION(AdapterIdentity);

src/mqtt/mqtt_client_impl.hpp

@@ -15,6 +15,7 @@
 //    limitations under the License.
 //
 
+#include <boost/beast/ssl.hpp>
 #include <boost/log/trivial.hpp>
 #include <boost/uuid/name_generator_sha1.hpp>
 
@@ -50,6 +51,8 @@ namespace mtconnect {
                                         mqtt::protocol_version>;
     using mqtt_tls_client = mqtt_tls_client_ptr<boost::asio::io_context &, std::string,
                                                 std::uint16_t, mqtt::protocol_version>;
+    using mqtt_tls_client_ws = mqtt_tls_client_ws_ptr<boost::asio::io_context &, std::string, std::uint16_t, std::string,
+                               mqtt::protocol_version>;
 
     template <typename Derived>
     class MqttClientImpl : public MqttClient
@@ -60,9 +63,14 @@ namespace mtconnect {
         : MqttClient(ioContext, move(handler)),
           m_options(options),
           m_host(GetOption<std::string>(options, configuration::MqttHost).value_or("localhost")),
-          m_port(GetOption<int>(options, configuration::MqttPort).value_or(1883)),
+          m_port(GetOption<int>(options, configuration::MqttPort).value_or(1883)),          
           m_reconnectTimer(ioContext)
       {
+        if (auto userName = GetOption<std::string>(options, configuration::MqttUserName))
+          m_username = *userName;
+        if (auto pswd = GetOption<std::string>(options, configuration::MqttPassword))
+          m_password = *pswd;
+
         std::stringstream url;
         url << "mqtt://" << m_host << ':' << m_port;
         m_url = url.str();
@@ -320,6 +328,10 @@ namespace mtconnect {
 
       std::uint16_t m_clientId {0};
 
+      std::string m_username = "";
+
+      std::string m_password = "";
+
       boost::asio::steady_timer m_reconnectTimer;
     };
 
@@ -334,6 +346,10 @@ namespace mtconnect {
         if (!m_client)
         {
           m_client = mqtt::make_async_client(m_ioContext, m_host, m_port);
+          if (!m_username.empty())
+            m_client->set_user_name(m_username);
+          if (!m_password.empty())
+            m_client->set_password(m_password);
         }
 
         return m_client;
@@ -354,15 +370,22 @@ namespace mtconnect {
         if (!m_client)
         {
           m_client = mqtt::make_tls_async_client(m_ioContext, m_host, m_port);
+          if (!m_username.empty())
+            m_client->set_user_name(m_username);
+          if (!m_password.empty())
+            m_client->set_password(m_password);
+
           auto cacert = GetOption<string>(m_options, configuration::MqttCaCert);
           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);
@@ -376,5 +399,44 @@ namespace mtconnect {
       mqtt_tls_client m_client;
     };
 
+    class MqttTlsWSClient : public MqttClientImpl<MqttTlsWSClient>
+    {
+    public:
+      using base = MqttClientImpl<MqttTlsWSClient>;
+      using base::base;
+
+      auto &getClient()
+      {
+        if (!m_client)
+        {
+          m_client = mqtt::make_tls_async_client_ws(m_ioContext, m_host, m_port);
+          if (!m_username.empty())
+            m_client->set_user_name(m_username);
+          if (!m_password.empty())
+            m_client->set_password(m_password);
+
+          auto cacert = GetOption<string>(m_options, configuration::MqttCaCert);
+          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;
+      }
+
+    protected:
+      mqtt_tls_client_ws m_client;
+    };
+
   }  // namespace mqtt_client
 }  // namespace mtconnect

src/mqtt/mqtt_server_impl.hpp

@@ -288,6 +288,52 @@ namespace mtconnect {
         m_port = GetOption<int>(options, configuration::Port).value_or(8883);
       }
 
+      using base::base;
+      using server = MQTT_NS::server_tls<>;
+
+      auto &getServer() { return m_server; }
+
+      auto &createServer()
+      {
+        if (!m_server)
+        {
+          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))
+          {
+            ctx.set_password_callback(
+                [this](size_t, boost::asio::ssl::context_base::password_purpose) -> string {
+                  return *GetOption<string>(m_options, configuration::TlsCertificatePassword);
+                });
+          }
+          m_server.emplace(boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), m_port),
+                           std::move(ctx), m_ioContext);
+        }
+
+        return *m_server;
+      }
+
+    protected:
+      std::optional<server> m_server;
+    };
+
+    class MqttTlsWSServer : public MqttServerImpl<MqttTlsWSServer>
+    {
+    public:
+      using base = MqttServerImpl<MqttTlsWSServer>;
+      MqttTlsWSServer(boost::asio::io_context &ioContext, const ConfigOptions &options)
+        : base(ioContext, options)
+      {
+        m_port = GetOption<int>(options, configuration::Port).value_or(8883);
+      }
+
       using base::base;
       using server = MQTT_NS::server_tls_ws<>;
 
@@ -298,6 +344,27 @@ namespace mtconnect {
         if (!m_server)
         {
           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))
+          {
+            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));
+            }
+          }*/
+
           m_server.emplace(boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), m_port),
                            std::move(ctx), m_ioContext);
         }