Jan_31_2024-Dev

core/messaging/CMakeLists.txt

@@ -0,0 +1,12 @@
+include_directories(include)
+
+add_subdirectory(messaging_comp)
+add_subdirectory(connection)
+add_subdirectory(messaging_buffer_comp)
+add_library(messaging messaging.cc)
+target_link_libraries(
+    messaging
+    "messaging_comp"
+    "connection"
+    "messaging_buffer_comp"
+)

core/messaging/connection/CMakeLists.txt

@@ -0,0 +1,2 @@
+add_library(connection connection_comp.cc connection.cc)
+add_subdirectory(connection_ut)

core/messaging/connection/connection.cc

@@ -0,0 +1,694 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#include "time_print.h"
+#include "connection.h"
+
+#include <arpa/inet.h>
+#include <netdb.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <fstream>
+#include <sstream>
+#include <string>
+
+#include "config.h"
+#include "http_request.h"
+#include "maybe_res.h"
+#include "messaging.h"
+#include "response_parser.h"
+#include "scope_exit.h"
+#include "smart_bio.h"
+
+using namespace std;
+using namespace smartBIO;
+
+USE_DEBUG_FLAG(D_CONNECTION);
+
+static const HTTPResponse sending_timeout(HTTPStatusCode::HTTP_UNKNOWN, "Failed to send all data in time");
+static const HTTPResponse receving_timeout(HTTPStatusCode::HTTP_UNKNOWN, "Failed to receive all data in time");
+static const HTTPResponse parsing_error(HTTPStatusCode::HTTP_UNKNOWN, "Failed to parse the HTTP response");
+
+const string &
+MessageConnectionKey::getHostName() const
+{
+    return host_name;
+}
+
+uint16_t
+MessageConnectionKey::getPort() const
+{
+    return port;
+}
+
+const MessageCategory &
+MessageConnectionKey::getCategory() const
+{
+    return category;
+}
+
+bool
+MessageConnectionKey::operator<(const MessageConnectionKey &other) const
+{
+    if (host_name != other.host_name) return host_name < other.host_name;
+    if (port != other.port) return port < other.port;
+    return category < other.category;
+}
+
+enum class ConnectionFlags
+{
+    UNSECURE,
+    ONE_TIME,
+    IGNORE_SSL_VALIDATION,
+    PROXY,
+
+    COUNT
+};
+
+class Connection::Impl
+{
+public:
+    Impl(const MessageConnectionKey &_key, const MessageMetadata &metadata) : key(_key)
+    {
+        auto metadata_flags = metadata.getConnectionFlags();
+        if (metadata_flags.isSet(MessageConnectionConfig::UNSECURE_CONN)) flags.setFlag(ConnectionFlags::UNSECURE);
+        if (metadata_flags.isSet(MessageConnectionConfig::ONE_TIME_CONN)) flags.setFlag(ConnectionFlags::ONE_TIME);
+        if (metadata_flags.isSet(MessageConnectionConfig::IGNORE_SSL_VALIDATION)) {
+            flags.setFlag(ConnectionFlags::IGNORE_SSL_VALIDATION);
+        }
+    }
+
+    void
+    setProxySettings(const MessageProxySettings &_settings)
+    {
+        flags.setFlag(ConnectionFlags::PROXY);
+        settings = _settings;
+    }
+
+    void
+    setConnectMessage(const string &connect_msg)
+    {
+        connect_message = connect_msg;
+    }
+
+    void
+    setExternalCertificate(const string &_certificate)
+    {
+        certificate = _certificate;
+    }
+
+    const MessageProxySettings &
+    getProxySettings() const
+    {
+        return settings;
+    }
+
+    const string &
+    getExternalCertificate() const
+    {
+        return certificate;
+    }
+
+    const MessageConnectionKey &
+    getConnKey() const
+    {
+        return key;
+    }
+
+    bool
+    isOverProxy() const
+    {
+        return flags.isSet(ConnectionFlags::PROXY);
+    }
+
+    bool
+    isUnsecure() const
+    {
+        return flags.isSet(ConnectionFlags::UNSECURE);
+    }
+
+    bool
+    isSuspended()
+    {
+        if (active.ok()) return false;
+
+        I_TimeGet *i_time = Singleton::Consume<I_TimeGet>::by<Messaging>();
+        auto curr_time = chrono::duration_cast<chrono::seconds>(i_time->getMonotonicTime());
+
+        if (active.getErr() > curr_time) {
+            dbgTrace(D_MESSAGING) << "Connection is suspended for another " << (active.getErr() - curr_time);
+            return true;
+        }
+
+        if (establishConnection().ok()) {
+            dbgDebug(D_MESSAGING) << "Reestablish connection";
+            return true;
+        }
+
+        dbgWarning(D_MESSAGING) << "Reestablish connection failed";
+        active = genError(curr_time + chrono::seconds(300));
+        return false;
+    }
+
+    Maybe<void>
+    establishConnection()
+    {
+        dbgFlow(D_CONNECTION) << "Establishing a new connection";
+        auto set_socket = setSocket();
+        if (!set_socket.ok()) {
+            dbgWarning(D_CONNECTION) << "Failed to set socket: " << set_socket.getErr();
+            return set_socket;
+        }
+
+        auto connect = connectToHost();
+        if (!connect.ok()) {
+            dbgWarning(D_CONNECTION) << "Failed to connect to host: " << connect.getErr();
+            return connect;
+        }
+
+        if (flags.isSet(ConnectionFlags::PROXY)) {
+            dbgDebug(D_CONNECTION) << "Sending a CONNECT request: " << connect_message;
+            auto res = sendAndReceiveData(connect_message, true);
+            if (!res.ok()) {
+                string connect_error = res.getErr().getBody();
+                dbgWarning(D_CONNECTION) << "Failed to connect to proxy: " << connect_error;
+                return genError(connect_error);
+            }
+
+            if (!isUnsecure()) {
+                auto encrypt_res = encryptProxyConnection();
+                if (!encrypt_res.ok()) {
+                    return genError("Failed to encrypt the socket after the CONNECT request" + encrypt_res.getErr());
+                }
+            }
+        }
+
+        dbgDebug(D_CONNECTION)
+            << "Successfully connected to "
+            << key.getHostName()
+            << ':'
+            << key.getPort()
+            << (isOverProxy() ? ", Over proxy: " + settings.getProxyHost() + ":" + to_string(key.getPort()) : "");
+        active = Maybe<void, chrono::seconds>();
+        return Maybe<void>();
+    }
+
+    Maybe<HTTPResponse, HTTPResponse>
+    sendRequest(const string &request)
+    {
+        dbgFlow(D_CONNECTION)
+            << "Send request to "
+            << key.getHostName()
+            << ':'
+            << key.getPort()
+            << ":\n"
+            << printOut(request);
+
+        auto result = sendAndReceiveData(request, false);
+        if (!result.ok()) {
+            establishConnection();
+            result = sendAndReceiveData(request, false);
+        }
+
+        if (!result.ok()) {
+            ++failed_attempts;
+            if (failed_attempts > 10) {
+                I_TimeGet *i_time = Singleton::Consume<I_TimeGet>::by<Messaging>();
+                auto curr_time = chrono::duration_cast<chrono::seconds>(i_time->getMonotonicTime());
+                active = genError(curr_time + chrono::seconds(300));
+            }
+            return result.passErr();
+        }
+
+        failed_attempts = 0;
+        return result;
+    }
+
+private:
+    Maybe<void>
+    setSSLContext()
+    {
+        dbgFlow(D_CONNECTION) << "Setting SSL context";
+        if (isUnsecure()) {
+            dbgTrace(D_CONNECTION) << "Connection is unsecure. Skipping SSL context creation";
+            return Maybe<void>();
+        }
+        ssl_ctx = BioUniquePtr<SSL_CTX>(SSL_CTX_new(TLS_client_method()));
+        if (!ssl_ctx.get()) return genError("Failed to initialize SSL context");
+        if (shouldIgnoreSslValidation()) {
+            dbgTrace(D_CONNECTION) << "Ignoring SSL validation";
+            return Maybe<void>();
+        }
+
+        SSL_CTX_set_verify(ssl_ctx.get(), SSL_VERIFY_PEER, nullptr);
+
+        auto defualt_cert_path = getFilesystemPathConfig() + "/certs/fog.pem";
+        auto cert_path = getConfigurationWithDefault(defualt_cert_path, "message", "Certificate chain file path");
+        const char *cert = cert_path.c_str();
+
+        auto details_ssl_dir = Singleton::Consume<I_AgentDetails>::by<Messaging>()->getOpenSSLDir();
+        auto openssl_dir = details_ssl_dir.ok() ? *details_ssl_dir : "/usr/lib/ssl/certs/";
+        auto configured_ssl_dir = getConfigurationWithDefault(openssl_dir, "message", "Trusted CA directory");
+        const char *ca_dir = configured_ssl_dir.empty() ? nullptr : configured_ssl_dir.c_str();
+
+        if (SSL_CTX_load_verify_locations(ssl_ctx.get(), cert, ca_dir) != 1) {
+            return genError("Failed to load certificate locations");
+        }
+
+        dbgDebug(D_CONNECTION) << "SSL context set successfully. Certificate: " << cert << ", CA dir: " << ca_dir;
+        return Maybe<void>();
+    }
+
+    Maybe<void>
+    setSocket()
+    {
+        dbgFlow(D_CONNECTION) << "Setting socket";
+        if (isUnsecure()) {
+            bio = BioUniquePtr<BIO>(BIO_new(BIO_s_connect()));
+            if (!bio.get()) return genError("Failed to create new BIO connection");
+            return Maybe<void>();
+        }
+
+        auto build_ssl = setSSLContext();
+        if (!build_ssl.ok()) return build_ssl;
+
+        if (isOverProxy()) {
+            bio = BioUniquePtr<BIO>(BIO_new(BIO_s_connect()));
+            if (!bio.get()) return genError("Failed to create new BIO connection");
+            return Maybe<void>();
+        }
+
+        bio = BioUniquePtr<BIO>(BIO_new_ssl_connect(ssl_ctx.get()));
+        BIO_get_ssl(bio.get(), &ssl_socket);
+        if (!ssl_socket) return genError("Failed to locate SSL pointer");
+
+        SSL_set_mode(ssl_socket, SSL_MODE_AUTO_RETRY);
+        SSL_set_hostflags(ssl_socket, X509_CHECK_FLAG_NO_PARTIAL_WILDCARDS);
+
+        auto host = key.getHostName().c_str();
+        if (!SSL_set1_host(ssl_socket, host)) {
+            return genError("Failed to set host name verification. Host: " + string(host));
+        }
+        if (!SSL_set_tlsext_host_name(ssl_socket, host)) {
+            return genError("Failed to set TLS host name extension. Host: " + string(host));
+        }
+
+        return Maybe<void>();
+    }
+
+    static chrono::microseconds
+    getConnectionTimeout()
+    {
+        I_Environment *env = Singleton::Consume<I_Environment>::by<Messaging>();
+
+        auto executable = env->get<string>("Service Name");
+        auto service_name = getProfileAgentSetting<string>("agent.config.message.connectionTimeoutServiceName");
+
+        if (executable.ok() && service_name.ok() && *executable == *service_name) {
+            auto service_timeout = getProfileAgentSetting<uint>("agent.config.message.connectionTimeout");
+            if (service_timeout.ok()) return chrono::microseconds(*service_timeout);
+        }
+
+        auto env_timeout = env->get<uint>("Connection timeout Override");
+        if (env_timeout.ok()) return chrono::microseconds(*env_timeout);
+
+        return chrono::microseconds(getConfigurationWithDefault<uint>(10000000, "message", "Connection timeout"));
+    }
+
+    bool
+    shouldIgnoreSslValidation() const
+    {
+        if (flags.isSet(ConnectionFlags::UNSECURE)) return true;
+        if (flags.isSet(ConnectionFlags::IGNORE_SSL_VALIDATION)) return true;
+        return getProfileAgentSettingWithDefault<bool>(false, "agent.config.message.ignoreSslValidation");
+    }
+
+    bool
+    isBioSocketReady() const
+    {
+        if (!bio.get()) return false;
+
+        auto fd = BIO_get_fd(bio.get(), nullptr);
+        fd_set rfds;
+        FD_ZERO(&rfds);
+        FD_SET(fd, &rfds);
+        struct timeval tv = { 0, 0 };
+
+        return select(fd + 1, nullptr, &rfds, nullptr, &tv) == 1;
+    }
+
+// LCOV_EXCL_START Reason: No ssl ut
+    Maybe<void>
+    verifyCertPinning(const BioUniquePtr<X509> &cert) const
+    {
+        BioUniquePtr<BIO> outbio = BioUniquePtr<BIO>(BIO_new(BIO_s_mem()));
+        if (!outbio.get()) return genError("Failed to allocate new BIO");
+        BioUniquePtr<EVP_PKEY> pkey = BioUniquePtr<EVP_PKEY>(X509_get_pubkey(cert.get()));
+        if (!pkey.get()) return genError("Error getting public key from certificate");
+
+        if (!PEM_write_bio_PUBKEY(outbio.get(), pkey.get())) return genError("Error writing key in PEM format");
+
+        char *buf;
+        auto len = BIO_get_mem_data(outbio.get(), &buf);
+        string recieved_public_key(buf, len);
+        dbgTrace(D_CONNECTION) << "Received public key " << recieved_public_key;
+
+        auto defualt_key_path = getFilesystemPathConfig() + "/certs/public-key.pem";
+        auto key_path = getConfigurationWithDefault(defualt_key_path, "message", "Public key path");
+        dbgTrace(D_CONNECTION) << "Load public key path. Path: " << key_path;
+
+        ifstream pinned_public_file(key_path);
+        if (!pinned_public_file.is_open()) return genError("Failed to open pinned public key file");
+
+        stringstream pinned_key;
+        pinned_key << pinned_public_file.rdbuf();
+        dbgTrace(D_CONNECTION) << "Saved public key: " << pinned_key.str();
+
+        if (recieved_public_key != pinned_key.str()) return genError("Received and pinned keys don't match");
+
+        return Maybe<void>();
+    }
+
+    Maybe<void>
+    verifyCert()
+    {
+        dbgFlow(D_CONNECTION) << "Verifying certificate";
+
+        if (shouldIgnoreSslValidation()) {
+            dbgTrace(D_CONNECTION) << "Ignoring SSL validation";
+            return Maybe<void>();
+        }
+
+        BioUniquePtr<X509> cert = BioUniquePtr<X509>(SSL_get_peer_certificate(ssl_socket));
+        if (!cert.get() || cert.get() == nullptr) return genError("Server did not provide a cert during handshake");
+
+        if (SSL_get_verify_result(ssl_socket) != X509_V_OK) {
+            string error = ERR_error_string(ERR_get_error(), nullptr);
+            return genError("Failed to verify server certificate. OpenSSL error: " + error);
+        }
+
+        if (!getConfigurationWithDefault<bool>(false, "message", "Verify SSL pinning")) return Maybe<void>();
+
+        return verifyCertPinning(cert);
+    }
+
+    Maybe<void>
+    performHandshakeAndVerifyCert(I_TimeGet *i_time, I_MainLoop *i_mainloop)
+    {
+        dbgFlow(D_CONNECTION) << "Performing SSL handshake";
+        auto handshake_timeout = getConfigurationWithDefault<uint>(500000, "message", "Connection handshake timeout");
+        auto handshake_end_time = i_time->getMonotonicTime() + chrono::microseconds(handshake_timeout);
+
+        while (i_time->getMonotonicTime() < handshake_end_time) {
+            if (!isBioSocketReady()) {
+                dbgTrace(D_CONNECTION) << "Socket is not ready for use.";
+                i_mainloop->yield(true);
+                continue;
+            }
+
+            if (BIO_do_handshake(bio.get()) > 0) return verifyCert();
+            if (!BIO_should_retry(bio.get())) {
+                string error = ERR_error_string(ERR_get_error(), nullptr);
+                return genError("Failed to obtain a successful SSL handshake. OpenSSL error: " + error);
+            }
+        }
+
+        return genError("SSL handshake reached timed out");
+    }
+// LCOV_EXCL_STOP
+
+    Maybe<void>
+    connectToHost()
+    {
+        string full_address;
+        if (isOverProxy()) {
+            full_address = settings.getProxyHost() + ":" + to_string(settings.getProxyPort());
+        } else {
+            full_address = key.getHostName() + ":" + to_string(key.getPort());
+        }
+
+        dbgFlow(D_CONNECTION) << "Connecting to " << full_address;
+        BIO_set_conn_hostname(bio.get(), full_address.c_str());
+        BIO_set_nbio(bio.get(), 1);
+
+        I_MainLoop *i_mainloop = Singleton::Consume<I_MainLoop>::by<Messaging>();
+        I_TimeGet *i_time = Singleton::Consume<I_TimeGet>::by<Messaging>();
+        auto conn_end_time = i_time->getMonotonicTime() + getConnectionTimeout();
+
+        uint attempts_count = 0;
+        while (i_time->getMonotonicTime() < conn_end_time) {
+            attempts_count++;
+            if (BIO_do_connect(bio.get()) > 0) {
+                if (isUnsecure() || isOverProxy()) return Maybe<void>();
+                return performHandshakeAndVerifyCert(i_time, i_mainloop);
+            }
+
+            if (!BIO_should_retry(bio.get())) {
+                string bio_error = ERR_error_string(ERR_get_error(), nullptr);
+                return genError("Failed to connect (BIO won't retry!): " + bio_error);
+            }
+
+            if ((attempts_count % 10) == 0) i_mainloop->yield(true);
+        }
+
+        return genError("Failed to establish new connection to " + full_address + " after reaching timeout.");
+    }
+
+    Maybe<uint, HTTPResponse>
+    sendData(const string &request, size_t data_left_to_send) const
+    {
+        if (!isBioSocketReady()) return 0;
+
+        dbgTrace(D_MESSAGING) << "Sending request: " << printOut(request);
+        size_t offset = request.length() - data_left_to_send;
+        auto curr_data_to_send = request.c_str() + offset;
+        int data_sent_len = BIO_write(bio.get(), curr_data_to_send, data_left_to_send);
+
+        if (data_sent_len >= 0) {
+            dbgTrace(D_CONNECTION) << "Sent " << data_sent_len << " bytes, out of: " << data_left_to_send << " bytes.";
+            return data_sent_len;
+        }
+
+        if (BIO_should_retry(bio.get())) {
+            dbgTrace(D_CONNECTION) << "Failed to send data - retrying";
+            return 0;
+        }
+
+        char error_buf[256];
+        ERR_error_string(ERR_get_error(), error_buf);
+        string error = "Failed to write data into BIO socket. Error: " + string(error_buf);
+        dbgWarning(D_CONNECTION) << error;
+        return genError(HTTPResponse(HTTPStatusCode::HTTP_UNKNOWN, error));
+    }
+
+    Maybe<string, HTTPResponse>
+    receiveData() const
+    {
+        if (!isBioSocketReady()) return string();
+
+        char buffer[1000];
+        int receive_len = BIO_read(bio.get(), buffer, sizeof(buffer) - 1);
+
+        if (receive_len > 0) {
+            dbgTrace(D_CONNECTION) << "Received " << receive_len << " bytes";
+            return string(buffer, receive_len);
+        }
+
+        if (BIO_should_retry(bio.get())) return string();
+
+        char error_buf[256];
+        ERR_error_string(ERR_get_error(), error_buf);
+        string error = receive_len == 0 ?
+            "Connection closed by peer" :
+            "Failed to read data from BIO socket. Error: " + string(error_buf);
+        dbgWarning(D_CONNECTION) << error;
+        return genError(HTTPResponse(HTTPStatusCode::HTTP_UNKNOWN, error));
+    }
+
+// LCOV_EXCL_START Reason: Fix in a future commit
+    Maybe<void>
+    encryptProxyConnection()
+    {
+        dbgFlow(D_CONNECTION) << "Encrypting BIO socket";
+        if (ssl_ctx.get() == nullptr) return genError("SSL context does not exist");
+
+        BioUniquePtr<BIO> s_bio = BioUniquePtr<BIO>(BIO_new_ssl(ssl_ctx.get(), 1));
+        if (s_bio.get() == nullptr) return genError("Failed to create encrypted BIO socket");
+
+        bio = BioUniquePtr<BIO>(BIO_push(s_bio.release(), bio.release()));
+        BIO_get_ssl(bio.get(), &ssl_socket);
+        if (!ssl_socket) return genError("Failed to locate SSL pointer");
+
+        SSL_set_hostflags(ssl_socket, X509_CHECK_FLAG_NO_PARTIAL_WILDCARDS);
+        auto host = key.getHostName().c_str();
+        if (!SSL_set1_host(ssl_socket, host)) {
+            return genError("Failed to set host name verification. Host: " + string(host));
+        }
+
+        I_MainLoop *i_mainloop = Singleton::Consume<I_MainLoop>::by<Messaging>();
+        I_TimeGet *i_time = Singleton::Consume<I_TimeGet>::by<Messaging>();
+        return performHandshakeAndVerifyCert(i_time, i_mainloop);
+    }
+// LCOV_EXCL_STOP
+
+    Maybe<HTTPResponse, HTTPResponse>
+    sendAndReceiveData(const string &request, bool is_connect)
+    {
+        I_MainLoop *i_mainloop = Singleton::Consume<I_MainLoop>::by<Messaging>();
+        while (lock) {
+            i_mainloop->yield(true);
+        }
+        lock = true;
+        auto unlock = make_scope_exit([&] () { lock = false; });
+
+        I_TimeGet *i_time = Singleton::Consume<I_TimeGet>::by<Messaging>();
+        auto sending_end_time = i_time->getMonotonicTime() + getConnectionTimeout();
+        size_t data_left_to_send = request.length();
+
+        while (data_left_to_send > 0) {
+            if (i_time->getMonotonicTime() > sending_end_time) return genError(sending_timeout);
+            auto send_size = sendData(request, data_left_to_send);
+            if (!send_size.ok()) return send_size.passErr();
+            data_left_to_send -= *send_size;
+            i_mainloop->yield(*send_size == 0); // We want to force waiting if we failed to send the data
+        }
+
+        auto receiving_end_time = i_time->getMonotonicTime() + getConnectionTimeout();
+        HTTPResponseParser http_parser;
+        dbgTrace(D_CONNECTION) << "Sent the message, now waiting for response";
+        while (!http_parser.hasReachedError()) {
+            if (i_time->getMonotonicTime() > receiving_end_time) return genError(receving_timeout);
+            auto receieved = receiveData();
+            if (!receieved.ok()) return receieved.passErr();
+            auto response = http_parser.parseData(*receieved, is_connect);
+            if (response.ok()) {
+                dbgTrace(D_MESSAGING) << printOut(response.unpack().toString());
+                return response.unpack();
+            }
+            i_mainloop->yield(receieved.unpack().empty());
+        }
+        return genError(parsing_error);
+    }
+
+    static string
+    printOut(const string &data)
+    {
+        string type = getConfigurationWithDefault<string>("chopped", "message", "Data printout type");
+        uint length = getConfigurationWithDefault<uint>(10, "message", "Data printout length");
+        if (type == "full") return data;
+        if (type == "size") return to_string(data.size()) + " bytes";
+        if (type == "none") return "";
+        string chopped_str = data.substr(0, length) + (data.size() > length ? " ..." : "");
+        if (type == "chopped") return chopped_str;
+
+
+        dbgWarning(D_CONNECTION) << "Unknown data printout option '" << type << "' - going with 'chopped' instead.";
+        return chopped_str;
+    }
+
+    MessageConnectionKey key;
+    Flags<ConnectionFlags> flags;
+
+    MessageProxySettings settings;
+    string connect_message;
+    string certificate;
+
+    smartBIO::BioUniquePtr<BIO> bio = nullptr;
+    smartBIO::BioUniquePtr<SSL_CTX> ssl_ctx = nullptr;
+    SSL *ssl_socket = nullptr;
+
+    Maybe<void, chrono::seconds> active = genError<chrono::seconds>(0);
+    uint failed_attempts = 0;
+
+    bool lock = false;
+};
+
+Connection::Connection(const MessageConnectionKey &key, const MessageMetadata &metadata)
+        :
+    pimpl(make_shared<Connection::Impl>(key, metadata))
+{}
+
+Connection::~Connection()
+{}
+
+Maybe<void>
+Connection::setProxySettings(const MessageProxySettings &settings)
+{
+    pimpl->setProxySettings(settings);
+
+    map<string, string> headers;
+    auto i_encrypt = Singleton::Consume<I_Encryptor>::by<Messaging>();
+    if (!settings.getProxyAuth().empty()) {
+        headers["Proxy-Authorization"] = i_encrypt->base64Encode(settings.getProxyAuth());
+    }
+
+    auto req = HTTPRequest::prepareRequest(*this, HTTPMethod::CONNECT, "", headers, "");
+    if (!req.ok()) return genError("Failed to create connect request. Error: " + req.getErr());
+
+    pimpl->setConnectMessage(req.unpack().toString());
+    return Maybe<void>();
+}
+
+void
+Connection::setExternalCertificate(const string &certificate)
+{
+    pimpl->setExternalCertificate(certificate);
+}
+
+const MessageProxySettings &
+Connection::getProxySettings() const
+{
+    return pimpl->getProxySettings();
+}
+
+const string &
+Connection::getExternalCertificate() const
+{
+    return pimpl->getExternalCertificate();
+}
+
+const MessageConnectionKey &
+Connection::getConnKey() const
+{
+    return pimpl->getConnKey();
+}
+
+bool
+Connection::isOverProxy() const
+{
+    return pimpl->isOverProxy();
+}
+
+bool
+Connection::isUnsecure() const
+{
+    return pimpl->isUnsecure();
+}
+
+bool
+Connection::isSuspended()
+{
+    return pimpl->isSuspended();
+}
+
+Maybe<void>
+Connection::establishConnection()
+{
+    return pimpl->establishConnection();
+}
+
+Maybe<HTTPResponse, HTTPResponse>
+Connection::sendRequest(const string &request)
+{
+    return pimpl->sendRequest(request);
+}

core/messaging/connection/connection_comp.cc

@@ -0,0 +1,157 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#include "connection_comp.h"
+
+#include <arpa/inet.h>
+#include <netdb.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <algorithm>
+#include <fstream>
+#include <string>
+
+#include "connection.h"
+#include "maybe_res.h"
+#include "messaging.h"
+#include "smart_bio.h"
+
+using namespace std;
+
+USE_DEBUG_FLAG(D_CONNECTION);
+
+class ConnectionComponent::Impl : Singleton::Provide<I_MessagingConnection>::From<ConnectionComponent>
+{
+public:
+    void
+    init()
+    {
+        SSL_library_init();
+        SSL_load_error_strings();
+        OpenSSL_add_all_algorithms();
+    }
+
+    Maybe<Connection>
+    establishConnection(const MessageMetadata &metadata, MessageCategory category) override
+    {
+        if (metadata.isProxySet()) return establishNewProxyConnection(metadata, category);
+        return establishNewConnection(metadata, category);
+    }
+
+    Maybe<Connection>
+    getPersistentConnection(const string &host_name, uint16_t port, MessageCategory category) override
+    {
+        auto conn = persistent_connections.find(MessageConnectionKey(host_name, port, category));
+        if (conn == persistent_connections.end()) return genError("No persistent connection found");
+        return conn->second;
+    }
+
+    Maybe<Connection>
+    getFogConnectionByCategory(MessageCategory category) override
+    {
+        auto maybe_fog_domain = Singleton::Consume<I_AgentDetails>::by<Messaging>()->getFogDomain();
+        if (!maybe_fog_domain.ok()) {
+            return genError("Failed to retrieve FOG domain " + maybe_fog_domain.getErr());
+        }
+        auto maybe_fog_port = Singleton::Consume<I_AgentDetails>::by<Messaging>()->getFogPort();
+        if (!maybe_fog_port.ok()) {
+            return genError("Failed to retrieve FOG port " + maybe_fog_port.getErr());
+        }
+
+        return getPersistentConnection(*maybe_fog_domain, *maybe_fog_port, category);
+    }
+
+    Maybe<HTTPResponse, HTTPResponse>
+    sendRequest(Connection &connection, HTTPRequest request) override
+    {
+        return connection.sendRequest(request.toString());
+    }
+
+private:
+    Maybe<Connection>
+    establishNewConnection(const MessageMetadata &metadata, MessageCategory category)
+    {
+        dbgFlow(D_CONNECTION)
+            << "Establishing a new connection. Host: "
+            << metadata.getHostName()
+            << ", port: "
+            << metadata.getPort();
+        MessageConnectionKey conn_key(metadata.getHostName(), metadata.getPort(), category);
+        Connection conn(conn_key, metadata);
+
+        const auto &external_certificate = metadata.getExternalCertificate();
+        if (!external_certificate.empty()) conn.setExternalCertificate(external_certificate);
+
+        auto connected = conn.establishConnection();
+
+        if (!connected.ok()) {
+            string connection_err = "Failed to establish connection. Error: " + connected.getErr();
+            dbgWarning(D_CONNECTION) << connection_err;
+            return genError(connection_err);
+        }
+
+        dbgTrace(D_CONNECTION) << "Connection establish succssesfuly";
+        persistent_connections.emplace(conn_key, conn);
+        return conn;
+    }
+
+    Maybe<Connection>
+    establishNewProxyConnection(const MessageMetadata &metadata, MessageCategory category)
+    {
+        dbgTrace(D_CONNECTION)
+            << "Establishing a new connection over proxy. Host: "
+            << metadata.getHostName()
+            << ", port: "
+            << metadata.getPort()
+            << ", proxy host: "
+            << metadata.getProxySettings().getProxyHost()
+            << ", proxy port: "
+            << metadata.getProxySettings().getProxyPort();
+
+        const auto &proxy = metadata.getProxySettings();
+        MessageConnectionKey conn_key(metadata.getHostName(), metadata.getPort(), category);
+        Connection conn(conn_key, metadata);
+
+        auto is_proxy = conn.setProxySettings(proxy);
+        if (!is_proxy.ok()) {
+            string proxy_err = "Failed to set proxy settings. Error: " + is_proxy.getErr();
+            return genError(proxy_err);
+        }
+
+        auto is_connected = conn.establishConnection();
+        if (!is_connected.ok()) {
+            string connection_err = "Failed to establish connection over proxy. Error: " + is_connected.getErr();
+            dbgWarning(D_CONNECTION) << connection_err;
+            return genError(connection_err);
+        }
+
+        dbgTrace(D_CONNECTION) << "Connection over proxy established succssesfuly";
+
+        persistent_connections.emplace(conn_key, conn);
+        return conn;
+    }
+
+    std::map<MessageConnectionKey, Connection> persistent_connections;
+};
+
+ConnectionComponent::ConnectionComponent() : pimpl(make_unique<Impl>())
+{}
+
+ConnectionComponent::~ConnectionComponent()
+{}
+
+void
+ConnectionComponent::init()
+{
+    pimpl->init();
+}

core/messaging/connection/connection_ut/CMakeLists.txt

@@ -0,0 +1,5 @@
+add_unit_test(
+    connection_comp_ut
+    "connection_comp_ut.cc"
+    "connection;ssl;connkey;singleton;boost_context;rest;metric;event_is;-lboost_regex;agent_details;-lcrypto;messaging;"
+)

core/messaging/connection/connection_ut/connection_comp_ut.cc

@@ -0,0 +1,298 @@
+#include "connection_comp.h"
+
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <arpa/inet.h>
+#include <pthread.h>
+#include <thread>
+#include <fcntl.h>
+#include <poll.h>
+
+#include "agent_core_utilities.h"
+#include "agent_details.h"
+#include "config.h"
+#include "config_component.h"
+#include "cptest.h"
+#include "environment.h"
+#include "mainloop.h"
+#include "connection.h"
+#include "mocks/mock_messaging_buffer.h"
+#include "mock/mock_agent_details.h"
+#include "mock/mock_mainloop.h"
+#include "mock/mock_time_get.h"
+#include "mock/mock_encryptor.h"
+#include "rest.h"
+#include "rest_server.h"
+
+using namespace std;
+using namespace testing;
+
+USE_DEBUG_FLAG(D_CONNECTION);
+
+class DummySocket : Singleton::Consume<I_MainLoop>
+{
+public:
+    ~DummySocket()
+    {
+        if (server_fd != -1) close(server_fd);
+        if (connection_fd != -1) close(connection_fd);
+    }
+
+    void
+    init()
+    {
+        server_fd = socket(AF_INET, SOCK_STREAM, 0);
+        dbgAssert(server_fd >= 0) << "Failed to open a socket";
+        int socket_enable = 1;
+        setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &socket_enable, sizeof(int));
+
+        struct sockaddr_in addr;
+        bzero(&addr, sizeof(addr));
+
+        addr.sin_family = AF_INET;
+        addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+        addr.sin_port = htons(8080);
+        bind(server_fd, (struct sockaddr *)&addr, sizeof(struct sockaddr_in));
+        listen(server_fd, 100);
+    }
+
+    void
+    acceptSocket()
+    {
+        if (connection_fd == -1) connection_fd = accept(server_fd, nullptr, nullptr);
+    }
+
+    string
+    readFromSocket()
+    {
+        acceptSocket();
+
+        string res;
+        char buffer[1024];
+        while (int bytesRead = readRaw(buffer, sizeof(buffer))) {
+            res += string(buffer, bytesRead);
+        }
+        return res;
+    }
+
+    void
+    writeToSocket(const string &msg)
+    {
+        acceptSocket();
+        EXPECT_EQ(write(connection_fd, msg.data(), msg.size()), msg.size());
+    }
+
+private:
+    int
+    readRaw(char *buf, uint len)
+    {
+        struct pollfd s_poll;
+        s_poll.fd = connection_fd;
+        s_poll.events = POLLIN;
+        s_poll.revents = 0;
+
+        if (poll(&s_poll, 1, 0) <= 0 || (s_poll.revents & POLLIN) == 0) return 0;
+
+        return read(connection_fd, buf, len);
+    }
+
+    int server_fd = -1;
+    int connection_fd = -1;
+};
+
+static ostream &
+operator<<(ostream &os, const BufferedMessage &)
+{
+    return os;
+}
+
+class TestConnectionComp : public testing::Test
+{
+public:
+    TestConnectionComp()
+    {
+        Debug::setUnitTestFlag(D_CONNECTION, Debug::DebugLevel::TRACE);
+        connection_comp.init();
+        i_conn = Singleton::Consume<I_MessagingConnection>::from(connection_comp);
+        setAgentDetails();
+        dummy_socket.init();
+    }
+
+    void
+    setAgentDetails()
+    {
+        EXPECT_CALL(mock_agent_details, getFogDomain()).WillRepeatedly(Return(string(fog_addr)));
+        EXPECT_CALL(mock_agent_details, getFogPort()).WillRepeatedly(Return(fog_port));
+        EXPECT_CALL(mock_agent_details, getOpenSSLDir()).WillRepeatedly(Return(string("/usr/lib/ssl/certs/")));
+        EXPECT_CALL(mock_agent_details, getAccessToken()).WillRepeatedly(Return(string("accesstoken")));
+    }
+
+    const string fog_addr = "127.0.0.1";
+    int fog_port = 8080;
+    CPTestTempfile agent_details_file;
+    ConnectionComponent connection_comp;
+    I_MessagingConnection *i_conn;
+    ::Environment env;
+    ConfigComponent config;
+    NiceMock<MockMessagingBuffer> mock_messaging_buffer;
+    NiceMock<MockAgentDetails> mock_agent_details;
+    NiceMock<MockTimeGet> mock_timer;
+    NiceMock<MockMainLoop> mock_mainloop;
+    StrictMock<MockEncryptor> mock_encryptor;
+    DummySocket dummy_socket;
+};
+
+TEST_F(TestConnectionComp, testSetAndGetFogConnection)
+{
+    Flags<MessageConnectionConfig> conn_flags;
+    conn_flags.setFlag(MessageConnectionConfig::UNSECURE_CONN);
+    MessageMetadata conn_metadata(fog_addr, fog_port, conn_flags);
+    auto maybe_connection = i_conn->establishConnection(conn_metadata, MessageCategory::GENERIC);
+    ASSERT_TRUE(maybe_connection.ok());
+
+    auto maybe_get_connection = i_conn->getFogConnectionByCategory(MessageCategory::GENERIC);
+    ASSERT_TRUE(maybe_get_connection.ok());
+}
+
+TEST_F(TestConnectionComp, testSetAndGetConnection)
+{
+    Flags<MessageConnectionConfig> conn_flags;
+    conn_flags.setFlag(MessageConnectionConfig::UNSECURE_CONN);
+    MessageMetadata conn_metadata("127.0.0.1", 8080, conn_flags);
+    auto maybe_connection = i_conn->establishConnection(conn_metadata, MessageCategory::LOG);
+    ASSERT_TRUE(maybe_connection.ok());
+
+    auto maybe_get_connection = i_conn->getPersistentConnection("127.0.0.1", 8080, MessageCategory::LOG);
+    ASSERT_TRUE(maybe_get_connection.ok());
+    auto get_conn = maybe_get_connection.unpack();
+    EXPECT_EQ(get_conn.getConnKey().getHostName(), "127.0.0.1");
+    EXPECT_EQ(get_conn.getConnKey().getPort(), 8080);
+    EXPECT_EQ(get_conn.getConnKey().getCategory(), MessageCategory::LOG);
+}
+
+TEST_F(TestConnectionComp, testEstablishNewConnection)
+{
+    Flags<MessageConnectionConfig> conn_flags;
+    conn_flags.setFlag(MessageConnectionConfig::UNSECURE_CONN);
+    conn_flags.setFlag(MessageConnectionConfig::ONE_TIME_CONN);
+    MessageMetadata conn_metadata("127.0.0.1", 8080, conn_flags);
+    conn_metadata.setExternalCertificate("external cert");
+
+    auto maybe_connection = i_conn->establishConnection(conn_metadata, MessageCategory::LOG);
+    cout << "[OREN] read: " <<  endl;
+    ASSERT_TRUE(maybe_connection.ok());
+    auto get_conn = maybe_connection.unpack();
+    EXPECT_EQ(get_conn.getConnKey().getHostName(), "127.0.0.1");
+}
+
+TEST_F(TestConnectionComp, testSendRequest)
+{
+    Flags<MessageConnectionConfig> conn_flags;
+    conn_flags.setFlag(MessageConnectionConfig::UNSECURE_CONN);
+    MessageMetadata conn_metadata("127.0.0.1", 8080, conn_flags);
+
+    auto maybe_connection = i_conn->establishConnection(conn_metadata, MessageCategory::LOG);
+    ASSERT_TRUE(maybe_connection.ok());
+    auto conn = maybe_connection.unpack();
+
+    auto req = HTTPRequest::prepareRequest(conn, HTTPMethod::POST, "/test", conn_metadata.getHeaders(), "test-body");
+    ASSERT_TRUE(req.ok());
+
+    ON_CALL(mock_mainloop, yield(false))
+        .WillByDefault(
+            InvokeWithoutArgs(
+                [&] ()
+                {
+                    dummy_socket.acceptSocket();
+                    dummy_socket.writeToSocket("HTTP/1.1 200 OK\r\nContent-Length: 7\r\n\r\nmy-test");
+                }
+            )
+        );
+
+    EXPECT_CALL(mock_timer, getMonotonicTime())
+        .WillRepeatedly(Invoke([] () {static int j = 0; return chrono::microseconds(++j * 1000 * 1000);}));
+    auto maybe_response = i_conn->sendRequest(conn, *req);
+    ASSERT_TRUE(maybe_response.ok());
+    EXPECT_EQ((*maybe_response).getBody(), "my-test");
+
+    string expected_msg =
+        "POST /test HTTP/1.1\r\n"
+        "Accept-Encoding: identity\r\n"
+        "Authorization: Bearer accesstoken\r\n"
+        "Connection: keep-alive\r\n"
+        "Content-Length: 9\r\n"
+        "Content-type: application/json\r\n"
+        "Host: 127.0.0.1\r\n"
+        "\r\n"
+        "test-body";
+    EXPECT_EQ(dummy_socket.readFromSocket(), expected_msg);
+}
+
+TEST_F(TestConnectionComp, testSendRequestReplyChunked)
+{
+    Flags<MessageConnectionConfig> conn_flags;
+    conn_flags.setFlag(MessageConnectionConfig::UNSECURE_CONN);
+    MessageMetadata conn_metadata("127.0.0.1", 8080, conn_flags);
+
+    auto maybe_connection = i_conn->establishConnection(conn_metadata, MessageCategory::LOG);
+    ASSERT_TRUE(maybe_connection.ok());
+    auto conn = maybe_connection.unpack();
+
+    auto req = HTTPRequest::prepareRequest(conn, HTTPMethod::POST, "/test", conn_metadata.getHeaders(), "test-body");
+    ASSERT_TRUE(req.ok());
+
+    ON_CALL(mock_mainloop, yield(false))
+        .WillByDefault(
+            InvokeWithoutArgs(
+                [&] ()
+                {
+                    dummy_socket.acceptSocket();
+                    string msg =
+                        "HTTP/1.1 200 OK\r\n"
+                        "Transfer-Encoding: chunked\r\n"
+                        "\r\n"
+                        "3\r\n"
+                        "my-\r\n"
+                        "4\r\n"
+                        "test\r\n"
+                        "0\r\n"
+                        "\r\n";
+                    dummy_socket.writeToSocket(msg);
+                }
+            )
+        );
+
+    EXPECT_CALL(mock_timer, getMonotonicTime())
+        .WillRepeatedly(Invoke([] () {static int j = 0; return chrono::microseconds(++j * 1000 * 1000);}));
+    auto maybe_response = i_conn->sendRequest(conn, *req);
+    ASSERT_TRUE(maybe_response.ok());
+    EXPECT_EQ((*maybe_response).getHTTPStatusCode(), HTTPStatusCode::HTTP_OK);
+    EXPECT_EQ((*maybe_response).getBody(), "my-test");
+    EXPECT_EQ((*maybe_response).toString(), "[Status-code]: 200 - HTTP_OK, [Body]: my-test");
+}
+
+TEST_F(TestConnectionComp, testEstablishNewProxyConnection)
+{
+    Flags<MessageConnectionConfig> conn_flags;
+    conn_flags.setFlag(MessageConnectionConfig::UNSECURE_CONN);
+    MessageMetadata conn_metadata("1.1.1.1", 9000, conn_flags);
+
+    MessageProxySettings proxy_settings("127.0.0.1", "oren", 8080);
+    conn_metadata.setProxySettings(proxy_settings);
+
+    //ON_CALL(mock_encryptor, base64Encode("oren")).WillByDefault(Return("encoded_oren"));
+    EXPECT_CALL(mock_encryptor, base64Encode("oren")).WillRepeatedly(Return("encoded_oren"));
+
+    ON_CALL(mock_mainloop, yield(false))
+        .WillByDefault(
+            InvokeWithoutArgs(
+                [&] ()
+                {
+                    dummy_socket.acceptSocket();
+                    dummy_socket.writeToSocket("HTTP/1.1 200 OK\r\n\r\n");
+                }
+            )
+        );
+
+    auto maybe_connection = i_conn->establishConnection(conn_metadata, MessageCategory::LOG);
+}

core/messaging/include/buffered_message.h

@@ -0,0 +1,63 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __BUFFERED_MESSAGE_H__
+#define __BUFFERED_MESSAGE_H__
+
+#include <iostream>
+#include <sstream>
+#include <string>
+
+#include "i_messaging.h"
+
+#include "cereal/archives/json.hpp"
+#include "cereal/types/string.hpp"
+#include "maybe_res.h"
+
+class BufferedMessage
+{
+public:
+    BufferedMessage() {}
+
+    BufferedMessage(
+        std::string _body,
+        HTTPMethod _method,
+        std::string _uri,
+        MessageCategory _category,
+        MessageMetadata _message_metadata
+    ) :
+        body(_body), method(_method), uri(_uri), category(_category), message_metadata(_message_metadata)
+    {}
+
+    void save(cereal::JSONOutputArchive &out_ar) const;
+    void load(cereal::JSONInputArchive &archive_in);
+
+    std::string toString() const;
+    bool operator==(const BufferedMessage &other) const;
+
+    const std::string &getBody() const;
+    const std::string &getURI() const;
+    HTTPMethod getMethod() const;
+    MessageCategory getCategory() const;
+    const MessageMetadata &getMessageMetadata() const;
+
+private:
+    std::string body;
+    HTTPMethod method;
+    std::string uri;
+    MessageCategory category;
+    MessageMetadata message_metadata;
+    uint16_t retries_number = 0;
+};
+
+#endif // __BUFFERED_MESSAGE_H__

core/messaging/include/connection.h

@@ -0,0 +1,75 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __CONNECTION_H__
+#define __CONNECTION_H__
+
+#include <map>
+#include <memory>
+#include <string>
+
+#include "i_agent_details.h"
+#include "i_environment.h"
+#include "i_mainloop.h"
+#include "i_time_get.h"
+#include "messaging/http_response.h"
+#include "messaging/messaging_metadata.h"
+
+#include "maybe_res.h"
+
+class MessageConnectionKey
+{
+public:
+    MessageConnectionKey() {}
+
+    MessageConnectionKey(const std::string &_host_name, uint16_t _port, MessageCategory _category) :
+        host_name(_host_name), port(_port), category(_category)
+    {}
+
+    const std::string & getHostName() const;
+    uint16_t getPort() const;
+    const MessageCategory & getCategory() const;
+
+    bool operator<(const MessageConnectionKey &other) const;
+
+private:
+    std::string host_name;
+    uint16_t port;
+    MessageCategory category;
+};
+
+class Connection
+{
+public:
+    Connection(const MessageConnectionKey &conn_key, const MessageMetadata &metadata);
+    ~Connection();
+
+    Maybe<void> setProxySettings(const MessageProxySettings &settings);
+    void setExternalCertificate(const std::string &certificate);
+    const MessageProxySettings & getProxySettings() const;
+    const std::string & getExternalCertificate() const;
+
+    const MessageConnectionKey &getConnKey() const;
+    bool isOverProxy() const;
+    bool isUnsecure() const;
+    bool isSuspended();
+
+    Maybe<void> establishConnection();
+    Maybe<HTTPResponse, HTTPResponse> sendRequest(const std::string &request);
+
+private:
+    class Impl;
+    std::shared_ptr<Impl> pimpl;
+};
+
+#endif // __CONNECTION_H__

core/messaging/include/connection_comp.h

@@ -0,0 +1,42 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __CONNECTION_COMPONENT_H__
+#define __CONNECTION_COMPONENT_H__
+
+#include <map>
+#include <string>
+
+#include "i_agent_details.h"
+#include "i_encryptor.h"
+#include "i_environment.h"
+#include "i_mainloop.h"
+#include "i_messaging.h"
+#include "i_time_get.h"
+#include "interfaces/i_messaging_connection.h"
+
+#include "maybe_res.h"
+
+class ConnectionComponent : Singleton::Provide<I_MessagingConnection>
+{
+public:
+    ConnectionComponent();
+    ~ConnectionComponent();
+    void init();
+
+private:
+    class Impl;
+    std::unique_ptr<Impl> pimpl;
+};
+
+#endif // __CONNECTION_COMPONENT_H__

core/messaging/include/http_request.h

@@ -0,0 +1,68 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __HTTP_REQUEST_H__
+#define __HTTP_REQUEST_H__
+
+#include <map>
+#include <string>
+
+#include "connection.h"
+#include "i_agent_details.h"
+#include "i_encryptor.h"
+#include "i_messaging.h"
+
+class HTTPRequest
+{
+public:
+    static Maybe<HTTPRequest> prepareRequest(
+        const Connection &conn,
+        HTTPMethod method,
+        const std::string &uri,
+        const std::map<std::string, std::string> &headers,
+        const std::string &body
+    );
+
+    Maybe<void> setConnectionHeaders(const Connection &conn, bool is_access_token_needed);
+
+    bool
+    isConnect() const
+    {
+        return method == HTTPMethod::CONNECT;
+    }
+
+    std::string toString() const;
+
+private:
+    HTTPRequest(
+        HTTPMethod _method,
+        const std::string &_uri,
+        const std::map<std::string, std::string> &_headers,
+        const std::string &_body
+    ) :
+        body(_body), uri(_uri), headers(_headers), method(_method)
+    {}
+
+    bool setConnectionHeaders(const Connection &conn);
+    void insertHeader(const std::string &header_key, const std::string &header_val);
+    Maybe<void> addAccessToken(const Connection &conn, bool is_registration);
+    Maybe<void> addProxyAuthorization(const Connection &conn);
+
+    std::string body;
+    std::string uri;
+    std::string method_statement;
+    std::map<std::string, std::string> headers;
+    HTTPMethod method;
+};
+
+#endif // __HTTP_REQUEST_H__

core/messaging/include/http_request_event.h

@@ -0,0 +1,124 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __HTTP_REQUEST_EVENT_H__
+#define __HTTP_REQUEST_EVENT_H__
+
+#include <string>
+
+#include "cereal/archives/json.hpp"
+
+// LCOV_EXCL_START
+
+class HTTPRequestSignature
+{
+public:
+    HTTPRequestSignature() = default;
+    HTTPRequestSignature(const std::string &_method, const std::string &_url, const std::string &_tag)
+            :
+        method(_method),
+        url(_url),
+        tag(_tag)
+    {
+    }
+
+    bool
+    operator<(const HTTPRequestSignature &other) const
+    {
+        return getSignature() < other.getSignature();
+    }
+
+    std::string getSignature() const { return method + url + tag; }
+
+    const std::string & getMethod()  const { return method;  }
+    const std::string & getURL()     const { return url;     }
+    const std::string & getTag()     const { return tag;     }
+
+    template<class Archive>
+    void load(Archive &ar)
+    {
+        try {
+            ar(cereal::make_nvp("tag", tag));
+        } catch(...) {
+            tag = "buffered messages";
+            ar.setNextName(nullptr);
+        }
+        ar(method, url);
+    }
+
+    template<class Archive>
+    void save(Archive &ar) const
+    {
+        ar(cereal::make_nvp("tag", tag));
+        ar(method, url);
+    }
+
+private:
+    std::string method;
+    std::string url;
+    std::string tag;
+};
+
+class HTTPRequestEvent : public HTTPRequestSignature
+{
+public:
+    HTTPRequestEvent() = default;
+    HTTPRequestEvent(
+        const std::string &_method,
+        const std::string &_url,
+        const std::string &_headers,
+        const std::string &&_body,
+        const std::string &_tag = "buffered messages")
+            :
+        HTTPRequestSignature(_method, _url, _tag),
+        headers(_headers),
+        body(std::move(_body))
+    {
+    }
+
+    HTTPRequestEvent(
+        const HTTPRequestSignature &&sig,
+        const std::string &_headers,
+        const std::string &&_body)
+            :
+        HTTPRequestSignature(std::move(sig)),
+        headers(_headers),
+        body(std::move(_body))
+    {
+    }
+
+    template<class Archive>
+    void load(Archive &ar)
+    {
+        HTTPRequestSignature::load(ar);
+        ar(headers, body);
+    }
+
+    template<class Archive>
+    void save(Archive &ar) const
+    {
+        HTTPRequestSignature::save(ar);
+        ar(headers, body);
+    }
+
+    const std::string & getHeaders() const { return headers; }
+    const std::string & getBody()    const { return body;    }
+
+private:
+    std::string headers;
+    std::string body;
+};
+
+// LCOV_EXCL_STOP
+
+#endif // __HTTP_REQUEST_EVENT_H__

core/messaging/include/interfaces/i_messaging_buffer.h

@@ -0,0 +1,45 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __I_MESSAGING_BUFFER_H__
+#define __I_MESSAGING_BUFFER_H__
+
+#include <iostream>
+#include <sstream>
+#include <string>
+
+#include "i_messaging.h"
+
+#include "../buffered_message.h"
+#include "cereal/archives/json.hpp"
+#include "cereal/types/string.hpp"
+#include "maybe_res.h"
+
+class I_MessageBuffer
+{
+public:
+    virtual void pushNewBufferedMessage(
+        const std::string &body,
+        HTTPMethod method,
+        const std::string &uri,
+        MessageCategory category,
+        MessageMetadata message_metadata,
+        bool force_immediate_writing
+    ) = 0;
+
+    virtual Maybe<BufferedMessage> peekMessage() = 0;
+    virtual void popMessage() = 0;
+    virtual void cleanBuffer() = 0;
+};
+
+#endif // __I_MESSAGING_BUFFER_H__

core/messaging/include/interfaces/i_messaging_connection.h

@@ -0,0 +1,42 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __I_MESSAGING_CONNECTION_H__
+#define __I_MESSAGING_CONNECTION_H__
+
+#include <map>
+#include <string>
+
+#include "http_request.h"
+#include "i_messaging.h"
+
+#include "connection.h"
+#include "maybe_res.h"
+
+class I_MessagingConnection
+{
+public:
+    virtual Maybe<Connection> establishConnection(const MessageMetadata &metadata, MessageCategory category) = 0;
+
+    virtual Maybe<Connection> getPersistentConnection(
+        const std::string &host_name, uint16_t port, MessageCategory category
+    ) = 0;
+
+    virtual Maybe<Connection> getFogConnectionByCategory(MessageCategory category) = 0;
+    virtual Maybe<HTTPResponse, HTTPResponse> sendRequest(Connection &connection, HTTPRequest request) = 0;
+
+protected:
+    virtual ~I_MessagingConnection() {}
+};
+
+#endif // __I_MESSAGING_CONNECTION_H__

core/messaging/include/messaging_buffer.h

@@ -0,0 +1,37 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __MESSAGNIG_BUFFER_H__
+#define __MESSAGNIG_BUFFER_H__
+
+#include <map>
+#include <string>
+
+#include "i_environment.h"
+#include "interfaces/i_messaging_buffer.h"
+#include "singleton.h"
+
+class MessagingBufferComponent : Singleton::Provide<I_MessageBuffer>
+{
+public:
+    MessagingBufferComponent();
+    ~MessagingBufferComponent();
+
+    void init();
+
+private:
+    class Impl;
+    std::unique_ptr<Impl> pimpl;
+};
+
+#endif // __MESSAGNIG_BUFFER_H__

core/messaging/include/messaging_comp.h

@@ -0,0 +1,101 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __MESSAGNIG_COMP_H__
+#define __MESSAGNIG_COMP_H__
+
+#include "messaging.h"
+
+#include <fstream>
+#include <map>
+#include <ostream>
+#include <sstream>
+#include <string>
+
+#include "cache.h"
+#include "connection.h"
+#include "connection_comp.h"
+#include "flags.h"
+#include "interfaces/i_messaging_buffer.h"
+#include "interfaces/i_messaging_connection.h"
+#include "maybe_res.h"
+#include "messaging_buffer.h"
+#include "singleton.h"
+
+class MessagingComp
+{
+public:
+    void init();
+
+    Maybe<HTTPResponse, HTTPResponse> sendSyncMessage(
+        HTTPMethod method,
+        const std::string &uri,
+        const std::string &body,
+        MessageCategory category,
+        const MessageMetadata &message_metadata
+    );
+
+    void sendAsyncMessage(
+        HTTPMethod method,
+        const std::string &uri,
+        const std::string &body,
+        MessageCategory category,
+        const MessageMetadata &message_metadata
+    );
+
+    Maybe<HTTPStatusCode, HTTPResponse> downloadFile(
+        HTTPMethod method,
+        const std::string &uri,
+        const std::string &download_file_path,
+        MessageCategory category = MessageCategory::GENERIC,
+        const MessageMetadata &message_metadata = MessageMetadata()
+    );
+
+    Maybe<HTTPStatusCode, HTTPResponse> uploadFile(
+        const std::string &uri,
+        const std::string &upload_file_path,
+        MessageCategory category,
+        const MessageMetadata &message_metadata
+    );
+
+    bool setFogConnection(const std::string &host, uint16_t port, bool is_secure, MessageCategory category);
+    bool setFogConnection(MessageCategory category);
+
+private:
+    Maybe<Connection> getConnection(MessageCategory category, const MessageMetadata &message_metadata);
+    Maybe<Connection> getPersistentConnection(const MessageMetadata &metadata, MessageCategory category) const;
+
+    Maybe<HTTPResponse, HTTPResponse> sendMessage(
+        HTTPMethod method,
+        const std::string &uri,
+        const std::string &body,
+        MessageCategory category,
+        const MessageMetadata &message_metadata
+    );
+
+    Maybe<HTTPResponse, HTTPResponse> suspendMessage(
+        const std::string &body,
+        HTTPMethod method,
+        const std::string &uri,
+        MessageCategory category,
+        const MessageMetadata &message_metadata
+    ) const;
+
+    I_MessagingConnection *i_conn;
+    I_MessageBuffer *i_messaging_buffer;
+    I_AgentDetails *agent_details;
+    bool should_buffer_failed_messages;
+    TemporaryCache<std::string, HTTPResponse> fog_get_requests_cache;
+};
+
+#endif //__MESSAGNIG_COMP_H__

core/messaging/include/mocks/mock_messaging_buffer.h

@@ -0,0 +1,25 @@
+#ifndef __MOCK_MESSAGING_BUFFER_H__
+#define __MOCK_MESSAGING_BUFFER_H__
+
+#include "cptest.h"
+#include "interfaces/i_messaging_buffer.h"
+
+// LCOV_EXCL_START Reason: No need to test mocks
+
+class MockMessagingBuffer : public Singleton::Provide<I_MessageBuffer>::From<MockProvider<I_MessageBuffer>>
+{
+public:
+    using string = std::string;
+    MOCK_METHOD6(
+        pushNewBufferedMessage,
+        void(const std::string &, HTTPMethod, const std::string &, MessageCategory, MessageMetadata, bool)
+    );
+
+    MOCK_METHOD0(peekMessage, Maybe<BufferedMessage>());
+    MOCK_METHOD0(popMessage, void());
+    MOCK_METHOD0(cleanBuffer, void());
+};
+
+// LCOV_EXCL_STOP
+
+#endif // __MOCK_MESSAGING_BUFFER_H__

core/messaging/include/mocks/mock_messaging_connection.h

@@ -0,0 +1,38 @@
+#ifndef __MOCK_MESSAGING_CONNECTION_H__
+#define __MOCK_MESSAGING_CONNECTION_H__
+
+#include "cptest.h"
+#include "interfaces/i_messaging_connection.h"
+
+// LCOV_EXCL_START Reason: No need to test mocks
+
+class MockMessagingConnection :
+    public Singleton::Provide<I_MessagingConnection>::From<MockProvider<I_MessagingConnection>>
+{
+public:
+    using string = std::string;
+
+    MOCK_METHOD2(establishConnection, Maybe<Connection>(const MessageMetadata &, MessageCategory));
+
+    MOCK_METHOD3(
+        establishNewConnection,
+        Maybe<Connection>(MessageConnectionKey, Flags<MessageConnectionConfig>, const string &)
+    );
+
+    MOCK_METHOD2(establishNewProxyConnection, Maybe<Connection>(Flags<MessageConnectionConfig>, MessageProxySettings));
+
+    Maybe<HTTPResponse, HTTPResponse>
+    sendRequest(Connection &conn, HTTPRequest req)
+    {
+        return mockSendRequest(conn, req, false);
+    }
+
+    MOCK_METHOD3(mockSendRequest, Maybe<HTTPResponse, HTTPResponse>(Connection &, HTTPRequest, bool));
+
+    MOCK_METHOD3(getPersistentConnection, Maybe<Connection>(const string &, uint16_t, MessageCategory));
+    MOCK_METHOD1(getFogConnectionByCategory, Maybe<Connection>(MessageCategory));
+};
+
+// LCOV_EXCL_STOP
+
+#endif // __MOCK_MESSAGING_CONNECTION_H__

core/messaging/include/response_parser.h

@@ -0,0 +1,34 @@
+#ifndef __RESPONSE_PARSER_H__
+#define __RESPONSE_PARSER_H__
+
+#include "i_messaging.h"
+
+class HTTPResponseParser
+{
+public:
+    Maybe<HTTPResponse> parseData(const std::string &data, bool is_connect);
+
+    bool
+    hasReachedError() const
+    {
+        return error;
+    }
+
+private:
+    bool parseStatusLine();
+    bool handleHeaders();
+    bool handleBody(bool is_connect);
+
+    Maybe<std::string> getHeaderVal(const std::string &header_key);
+
+    bool getChunkedResponse();
+    bool isLegalChunkedResponse(const std::string &res);
+
+    Maybe<HTTPStatusCode> status_code = genError("Not received");
+    Maybe<std::map<std::string, std::string>> headers = genError("Not received");
+    std::string body;
+    std::string raw_response;
+    bool error = false;
+};
+
+#endif // __RESPONSE_PARSER_H__

core/messaging/include/smart_bio.h

@@ -0,0 +1,75 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#ifndef __SMART_BIO_H__
+#define __SMART_BIO_H__
+
+#include "openssl/bio.h"
+#include "openssl/err.h"
+#include "openssl/ssl.h"
+#include "openssl/x509v3.h"
+
+namespace smartBIO
+{
+
+template <class T>
+struct Destrctor;
+
+template <>
+struct Destrctor<BIO>
+{
+    void
+    operator()(BIO *pointer) const
+    {
+        if (pointer != nullptr) BIO_free_all(pointer);
+    }
+};
+
+// LCOV_EXCL_START Reason: No ssl ut
+template <>
+struct Destrctor<SSL_CTX>
+{
+    void
+    operator()(SSL_CTX *pointer) const
+    {
+        if (pointer != nullptr) SSL_CTX_free(pointer);
+    }
+};
+
+template <>
+struct Destrctor<X509>
+{
+    void
+    operator()(X509 *pointer) const
+    {
+        if (pointer != nullptr) X509_free(pointer);
+    }
+};
+
+template <>
+struct Destrctor<EVP_PKEY>
+{
+    void
+    operator()(EVP_PKEY *pointer) const
+    {
+        if (pointer != nullptr) EVP_PKEY_free(pointer);
+    }
+};
+// LCOV_EXCL_STOP
+
+template <class OpenSSLType>
+using BioUniquePtr = std::unique_ptr<OpenSSLType, smartBIO::Destrctor<OpenSSLType>>;
+
+} // namespace smartBIO
+
+#endif // __SMART_BIO_H__

core/messaging/messaging.cc

@@ -0,0 +1,132 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#include "messaging.h"
+
+#include <memory>
+
+#include "connection_comp.h"
+#include "interfaces/i_messaging_connection.h"
+#include "messaging_buffer.h"
+#include "messaging_comp.h"
+
+using namespace std;
+
+USE_DEBUG_FLAG(D_MESSAGING);
+
+// LCOV_EXCL_START Reason: This wrapper for the other components, all logic is tested there.
+
+class Messaging::Impl : Singleton::Provide<I_Messaging>::From<Messaging>
+{
+public:
+    void
+    init()
+    {
+        messaging_comp.init();
+        connection_comp.init();
+        messaging_buffer_comp.init();
+    }
+
+    Maybe<HTTPResponse, HTTPResponse>
+    sendSyncMessage(
+        HTTPMethod method,
+        const std::string &uri,
+        const std::string &body,
+        MessageCategory category,
+        MessageMetadata message_metadata
+    ) override
+    {
+        return messaging_comp.sendSyncMessage(method, uri, body, category, message_metadata);
+    }
+
+    void
+    sendAsyncMessage(
+        const HTTPMethod method,
+        const std::string &uri,
+        const std::string &body,
+        const MessageCategory category,
+        MessageMetadata message_metadata
+    ) override
+    {
+        return messaging_comp.sendAsyncMessage(method, uri, body, category, message_metadata);
+    }
+
+    Maybe<HTTPStatusCode, HTTPResponse>
+    downloadFile(
+        const HTTPMethod method,
+        const std::string &uri,
+        const std::string &download_file_path,
+        const MessageCategory category,
+        MessageMetadata message_metadata
+    ) override
+    {
+        return messaging_comp.downloadFile(method, uri, download_file_path, category, message_metadata);
+    }
+
+    Maybe<HTTPStatusCode, HTTPResponse>
+    uploadFile(
+        const std::string &uri,
+        const std::string &upload_file_path,
+        const MessageCategory category,
+        MessageMetadata message_metadata
+    ) override
+    {
+        return messaging_comp.uploadFile(uri, upload_file_path, category, message_metadata);
+    }
+
+    bool
+    setFogConnection(const string &host, uint16_t port, bool is_secure, MessageCategory category) override
+    {
+        return messaging_comp.setFogConnection(host, port, is_secure, category);
+    }
+
+    bool
+    setFogConnection(MessageCategory category = MessageCategory::GENERIC) override
+    {
+        return messaging_comp.setFogConnection(category);
+    }
+
+private:
+    MessagingComp messaging_comp;
+    ConnectionComponent connection_comp;
+    MessagingBufferComponent messaging_buffer_comp;
+};
+
+Messaging::Messaging() : Component("Messaging"), pimpl(make_unique<Impl>())
+{}
+
+Messaging::~Messaging()
+{}
+
+void
+Messaging::init()
+{
+    pimpl->init();
+}
+
+void
+Messaging::preload()
+{
+    registerExpectedConfiguration<int>("message", "Cache timeout");
+    registerExpectedConfiguration<uint>("message", "Connection timeout");
+    registerExpectedConfiguration<uint>("message", "Connection handshake timeout");
+    registerExpectedConfiguration<bool>("message", "Verify SSL pinning");
+    registerExpectedConfiguration<bool>("message", "Buffer Failed Requests");
+    registerExpectedConfiguration<string>("message", "Certificate chain file path");
+    registerExpectedConfiguration<string>("message", "Trusted CA directory");
+    registerExpectedConfiguration<string>("message", "Public key path");
+    registerExpectedConfiguration<string>("message", "Data printout type");
+    registerExpectedConfiguration<uint>("message", "Data printout length");
+}
+
+// LCOV_EXCL_STOP

core/messaging/messaging_buffer_comp/CMakeLists.txt

@@ -0,0 +1,2 @@
+add_library(messaging_buffer_comp messaging_buffer_comp.cc buffered_message.cc)
+add_subdirectory(messaging_buffer_comp_ut)

core/messaging/messaging_buffer_comp/buffered_message.cc

@@ -0,0 +1,128 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#include "buffered_message.h"
+#include "customized_cereal_map.h"
+
+using namespace std;
+
+static const std::map<std::string, MessageCategory> string_to_category = {
+    {"generic",       MessageCategory::GENERIC     },
+    { "log",          MessageCategory::LOG         },
+    { "debug",        MessageCategory::DEBUG       },
+    { "metric",       MessageCategory::METRIC      },
+    { "intelligence", MessageCategory::INTELLIGENCE}
+};
+
+static const std::map<MessageCategory, std::string> category_to_string = {
+    {MessageCategory::GENERIC,       "generic"     },
+    { MessageCategory::LOG,          "log"         },
+    { MessageCategory::DEBUG,        "debug"       },
+    { MessageCategory::METRIC,       "metric"      },
+    { MessageCategory::INTELLIGENCE, "intelligence"}
+};
+
+static const std::map<std::string, HTTPMethod> string_to_method = {
+    {"get",      HTTPMethod::GET    },
+    { "post",    HTTPMethod::POST   },
+    { "patch",   HTTPMethod::PATCH  },
+    { "connect", HTTPMethod::CONNECT},
+    { "put",     HTTPMethod::PUT    }
+};
+
+static const std::map<HTTPMethod, std::string> method_to_string = {
+    {HTTPMethod::GET,      "get"    },
+    { HTTPMethod::POST,    "post"   },
+    { HTTPMethod::PATCH,   "patch"  },
+    { HTTPMethod::CONNECT, "connect"},
+    { HTTPMethod::PUT,     "put"    }
+};
+
+void
+BufferedMessage::save(cereal::JSONOutputArchive &out_ar) const
+{
+    string category_str = category_to_string.find(category)->second;
+    string method_str = method_to_string.find(method)->second;
+
+    out_ar(
+        cereal::make_nvp("body", body),
+        cereal::make_nvp("uri", uri),
+        cereal::make_nvp("method", method_str),
+        cereal::make_nvp("category", category_str),
+        cereal::make_nvp("message_metadata", message_metadata)
+    );
+}
+
+void
+BufferedMessage::load(cereal::JSONInputArchive &archive_in)
+{
+    string method_str;
+    string category_str;
+    archive_in(
+        cereal::make_nvp("body", body),
+        cereal::make_nvp("uri", uri),
+        cereal::make_nvp("method", method_str),
+        cereal::make_nvp("category", category_str),
+        cereal::make_nvp("message_metadata", message_metadata)
+    );
+    method = string_to_method.find(method_str)->second;
+    category = string_to_category.find(category_str)->second;
+}
+
+string
+BufferedMessage::toString() const
+{
+    stringstream ss;
+    {
+        cereal::JSONOutputArchive ar(ss);
+        save(ar);
+    }
+
+    return ss.str();
+}
+
+bool
+BufferedMessage::operator==(const BufferedMessage &other) const
+{
+    return body == other.body && uri == other.uri;
+}
+
+const string &
+BufferedMessage::getBody() const
+{
+    return body;
+}
+
+const string &
+BufferedMessage::getURI() const
+{
+    return uri;
+}
+
+HTTPMethod
+BufferedMessage::getMethod() const
+{
+    return method;
+}
+
+MessageCategory
+BufferedMessage::getCategory() const
+{
+    return category;
+}
+
+const MessageMetadata &
+BufferedMessage::getMessageMetadata() const
+{
+    return message_metadata;
+}

core/messaging/messaging_buffer_comp/messaging_buffer_comp.cc

@@ -0,0 +1,544 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#include "messaging_buffer.h"
+#include "messaging.h"
+#include "http_request_event.h"
+
+#include <dirent.h>
+#include <sys/stat.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <string>
+#include <vector>
+
+#include "config.h"
+#include "debug.h"
+
+using namespace std;
+
+USE_DEBUG_FLAG(D_MESSAGING_BUFFER);
+
+#ifndef smb
+static constexpr uint buffer_max_size_MB = 100;
+#else
+static constexpr uint buffer_max_size_MB = 3;
+#endif
+
+static bool
+checkExistence(const string &path)
+{
+    try {
+        struct stat info;
+        if (stat(path.c_str(), &info) != 0) return false;
+        return info.st_mode & S_IFREG;
+    } catch (exception &e) {
+        return false;
+    }
+}
+
+class MessagingBufferComponent::Impl : Singleton::Provide<I_MessageBuffer>::From<MessagingBufferComponent>
+{
+public:
+    void init();
+
+    void pushNewBufferedMessage(
+        const string &body,
+        HTTPMethod method,
+        const string &uri,
+        MessageCategory category,
+        MessageMetadata message_metadata,
+        bool force_immediate_writing
+    ) override;
+
+    Maybe<BufferedMessage> peekMessage() override;
+
+    void popMessage() override;
+
+    void cleanBuffer() override;
+
+private:
+    void handleBufferedMessages();
+    bool sendMessage();
+    HTTPStatusCode sendMessage(const BufferedMessage &message) const;
+
+    void handleInMemoryMessages();
+
+    void writeToDisk(const BufferedMessage &message);
+
+    static Maybe<uint32_t> seekStartOfMessage(FILE *file);
+    static bool readBytes(FILE *file, uint size_to_read, char *output_bytes);
+
+    bool canWriteToDisk(size_t message_size) const;
+    Maybe<uint> getDirectorySize() const;
+
+    // LCOV_EXCL_START - Converting old formats to new format will be added later
+    static Maybe<HTTPMethod> convertStringToHTTPMethod(const string &method_string);
+    void removeLegacyBuffer(const string &root_path, const string &exec_name);
+    void convertLegacyBuffer(const string &body_file_path);
+    Maybe<HTTPRequestEvent> serializeOldData(const string &data);
+    // LCOV_EXCL_STOP
+
+    vector<BufferedMessage> memory_messages;
+    string buffer_input;
+    string buffer_output;
+    string buffer_root_path;
+    uint max_size_on_disk_MB = 0;
+    uint curr_no_retries = 0;
+    I_ShellCmd *shell_cmd = nullptr;
+    I_Encryptor *encryptor = nullptr;
+    I_MainLoop *mainloop = nullptr;
+    I_Messaging *messaging = nullptr;
+};
+
+void
+MessagingBufferComponent::Impl::init()
+{
+    max_size_on_disk_MB = getProfileAgentSettingWithDefault<uint>(buffer_max_size_MB, "eventBuffer.maxSizeOnDiskInMB");
+    shell_cmd = Singleton::Consume<I_ShellCmd>::by<Messaging>();
+    encryptor = Singleton::Consume<I_Encryptor>::by<Messaging>();
+    mainloop = Singleton::Consume<I_MainLoop>::by<Messaging>();
+    messaging = Singleton::Consume<I_Messaging>::from<Messaging>();
+    
+    auto sub_path = getProfileAgentSettingWithDefault<string>("nano_agent/event_buffer/", "eventBuffer.baseFolder");
+    buffer_root_path = getLogFilesPathConfig() + "/" + sub_path;
+    string full_executable_name =
+        Singleton::Consume<I_Environment>::by<Messaging>()->get<string>("Executable Name").unpack();
+    string executable_name = full_executable_name.substr(full_executable_name.find_last_of("/") + 1);
+    removeLegacyBuffer(buffer_root_path, executable_name);
+    mkdir(buffer_root_path.c_str(), 0644);
+
+    auto *instance_awareness = Singleton::Consume<I_InstanceAwareness>::by<Messaging>();
+    string unique_id = instance_awareness->getInstanceID().ok() ? instance_awareness->getInstanceID().unpack() : "";
+    buffer_input = buffer_root_path + "/" + executable_name + unique_id + ".input";
+    buffer_output = buffer_root_path + "/" + executable_name + unique_id + ".output";
+    memory_messages.reserve(32);
+
+    uint tmo = getConfigurationWithDefault<uint>(5, "message", "Send event retry in sec");
+    mainloop->addRecurringRoutine(
+        I_MainLoop::RoutineType::Timer,
+        chrono::seconds(tmo),
+        [this] () { handleBufferedMessages(); },
+        "A-sync messaging routine",
+        false
+    );
+    mainloop->addRecurringRoutine(
+        I_MainLoop::RoutineType::Timer,
+        chrono::seconds(2),
+        [this] () { handleInMemoryMessages(); },
+        "Handling in-memory messages",
+        false
+    );
+}
+
+void
+MessagingBufferComponent::Impl::pushNewBufferedMessage(
+    const string &body,
+    HTTPMethod method,
+    const string &uri,
+    MessageCategory category,
+    MessageMetadata message_metadata,
+    bool force_immediate_writing
+)
+{
+    dbgTrace(D_MESSAGING_BUFFER) << "Pushing new message to buffer";
+
+    message_metadata.setShouldBufferMessage(false);
+
+    if (!force_immediate_writing) {
+        dbgDebug(D_MESSAGING_BUFFER) << "Holding message temporarily in memory";
+        memory_messages.emplace_back(body, method, uri, category, message_metadata);
+        return;
+    }
+
+    BufferedMessage buffered_message(body, method, uri, category, message_metadata);
+    writeToDisk(buffered_message);
+}
+
+Maybe<BufferedMessage>
+MessagingBufferComponent::Impl::peekMessage()
+{
+    auto move_cmd =
+        "if [ -s " + buffer_input + " ] && [ ! -s " + buffer_output + " ];"
+        "then mv " + buffer_input + " " + buffer_output + ";"
+        "fi";
+
+    shell_cmd->getExecOutput(move_cmd);
+
+    if (!checkExistence(buffer_output)) return genError(buffer_output + " does not exist");
+
+    FILE *file = fopen(buffer_output.c_str(), "rb");
+    if (file == nullptr) {
+        dbgWarning(D_MESSAGING_BUFFER) << "Failed to open file for reading. File: " << buffer_output;
+        cleanBuffer();
+        return genError("Failed to open file");
+    }
+
+    auto possition = seekStartOfMessage(file);
+    if (!possition.ok()) {
+        fclose(file);
+        dbgDebug(D_MESSAGING_BUFFER) << "Failed to find message start: " << possition.getErr();
+        cleanBuffer();
+        return possition.passErr();
+    }
+
+    string buffer;
+    buffer.resize(*possition);
+    auto read = readBytes(file, *possition, const_cast<char *>(buffer.data()));
+    fclose(file);
+    if (!read) {
+        cleanBuffer();
+        return genError("Filed to read the message");
+    }
+
+    BufferedMessage message;
+    try {
+        stringstream ss(buffer);
+        cereal::JSONInputArchive ar(ss);
+        message.load(ar);
+    } catch (const cereal::Exception &e) {
+        string err = e.what();
+        dbgError(D_MESSAGING_BUFFER) << "Parsing backlog error: " << err;
+        cleanBuffer();
+        return genError("Filed to parse the message: " + err);
+    }
+
+    return message;
+}
+
+void
+MessagingBufferComponent::Impl::popMessage()
+{
+    dbgTrace(D_MESSAGING_BUFFER) << "Popping message from buffer";
+
+    FILE *file = fopen(buffer_output.c_str(), "rb");
+    if (file == nullptr) {
+        dbgWarning(D_MESSAGING_BUFFER) << "Failed to open file for reading. File: " << buffer_input;
+        return;
+    }
+
+    auto possition = seekStartOfMessage(file);
+    auto new_size = ftell(file);
+    fclose(file);
+    if (!possition.ok()) {
+        dbgDebug(D_MESSAGING_BUFFER) << "Failed to find message start: " << possition.getErr();
+        return;
+    }
+
+    int result = truncate(buffer_output.c_str(), new_size);
+    if (result == 0) {
+        dbgTrace(D_MESSAGING_BUFFER) << "File truncated successfully.";
+    } else {
+        dbgTrace(D_MESSAGING_BUFFER) << "Error truncating the file: " << strerror(errno);
+    }
+}
+
+void
+MessagingBufferComponent::Impl::cleanBuffer()
+{
+    dbgTrace(D_MESSAGING_BUFFER) << "Cleaning buffer";
+    remove(buffer_input.c_str());
+    remove(buffer_output.c_str());
+}
+
+void
+MessagingBufferComponent::Impl::handleBufferedMessages()
+{
+    while (true) {
+        if (!sendMessage()) return;
+        mainloop->yield();
+    }
+}
+
+bool
+MessagingBufferComponent::Impl::sendMessage()
+{
+    const Maybe<BufferedMessage> &maybe_msg_to_send = peekMessage();
+    if (!maybe_msg_to_send.ok()) {
+        dbgDebug(D_MESSAGING) << "Peeking failed: " << maybe_msg_to_send.getErr();
+        return false;
+    }
+
+    auto res = sendMessage(*maybe_msg_to_send);
+
+    if (res == HTTPStatusCode::HTTP_OK) {
+        dbgDebug(D_MESSAGING) << "Successfully sent buffered message";
+        popMessage();
+        curr_no_retries = 0;
+        return true;
+    }
+
+    if (res == HTTPStatusCode::HTTP_SUSPEND) {
+        dbgDebug(D_MESSAGING) << "Suspended connection - sleeping for a while";
+        mainloop->yield(chrono::seconds(1));
+        return true;
+    }
+
+    ++curr_no_retries;
+    if (curr_no_retries >= getProfileAgentSettingWithDefault<uint>(10, "eventBuffer.maxNumOfSendigRetries")) {
+        dbgWarning(D_MESSAGING) << "Reached maximum number of retries - poping message";
+        popMessage();
+        curr_no_retries = 0;
+    }
+    return true;
+}
+
+HTTPStatusCode
+MessagingBufferComponent::Impl::sendMessage(const BufferedMessage &message) const
+{
+    auto res = messaging->sendSyncMessage(
+        message.getMethod(),
+        message.getURI(),
+        message.getBody(),
+        message.getCategory(),
+        message.getMessageMetadata()
+    );
+
+    if (res.ok()) return HTTPStatusCode::HTTP_OK;
+    if (res.getErr().getHTTPStatusCode() == HTTPStatusCode::HTTP_SUSPEND) return HTTPStatusCode::HTTP_SUSPEND;
+    return HTTPStatusCode::HTTP_UNKNOWN;
+}
+
+void
+MessagingBufferComponent::Impl::handleInMemoryMessages()
+{
+    auto messages = move(memory_messages);
+    memory_messages.reserve(32);
+
+    for (const auto &message : messages) {
+        if (sendMessage(message) != HTTPStatusCode::HTTP_OK) writeToDisk(message);
+        mainloop->yield();
+    }
+}
+
+void
+MessagingBufferComponent::Impl::writeToDisk(const BufferedMessage &message)
+{
+    auto serialized_message = message.toString();
+
+    if (!canWriteToDisk(serialized_message.size())) {
+        dbgWarning(D_MESSAGING_BUFFER) << "Buffer is full. Message will not be written to disk: " << message.getURI();
+        return;
+    }
+
+    ofstream file(buffer_input, ios::app);
+    if (!file.is_open()) {
+        dbgWarning(D_MESSAGING_BUFFER) << "Failed to open file for writing. File: " << buffer_input;
+        return;
+    }
+
+    uint32_t size = serialized_message.size();
+    file.write(serialized_message.data(), size);
+    file.write(reinterpret_cast<char *>(&size), sizeof(size));
+    char type = 0;
+    file.write(&type, 1);
+}
+
+Maybe<uint32_t>
+MessagingBufferComponent::Impl::seekStartOfMessage(FILE *file)
+{
+    int type_size = sizeof(char);
+    int lenght_size = sizeof(uint32_t);
+
+    if (fseek(file, -type_size, SEEK_END) != 0) return genError("Failed to get to type byte");
+    char type;
+    if (!readBytes(file, type_size, &type)) return genError("Failed to read type");
+    if (type != 0) return genError("Only type 0 is currently supported");
+
+    if (fseek(file, -(type_size + lenght_size), SEEK_END) != 0) return genError("Failed to get to length bytes");
+    uint32_t length;
+    if (!readBytes(file, lenght_size, reinterpret_cast<char *>(&length))) return genError("Failed to read length");
+
+    int total_offset = type_size + lenght_size + length;
+    if (ftell(file) == total_offset) {
+        if (fseek(file, 0, SEEK_SET) != 0) return genError("Failed to get to the start of the file");
+    } else {
+        if (fseek(file, -total_offset, SEEK_END) != 0) return genError("Failed to get to message start");
+    }
+
+    return length;
+}
+
+bool
+MessagingBufferComponent::Impl::readBytes(FILE *file, uint size_to_read, char *output)
+{
+    for (uint index = 0; index < size_to_read; ++index) {
+        int ch = fgetc(file);
+        if (ch == EOF) return false;
+        output[index] = static_cast<char>(ch);
+    }
+    return true;
+}
+
+Maybe<uint>
+MessagingBufferComponent::Impl::getDirectorySize() const
+{
+    DIR *dir = opendir(buffer_root_path.c_str());
+    if (dir == nullptr) {
+        return genError("Unable to open directory: " + buffer_root_path);
+    }
+
+    uint total_size = 0;
+    struct dirent *entry;
+    while ((entry = readdir(dir)) != nullptr) {
+        if (entry->d_type == DT_REG) {
+            struct stat file_info;
+            string tmp_file_path = buffer_root_path + "/" + entry->d_name;
+            if (stat(tmp_file_path.c_str(), &file_info) == 0) {
+                total_size += file_info.st_size;
+            } else {
+                return genError("Error retrieving file size. " + tmp_file_path);
+            }
+        }
+    }
+
+    closedir(dir);
+    return total_size;
+}
+
+// LCOV_EXCL_START  - Converting old formats to new format will be added later
+Maybe<HTTPMethod>
+MessagingBufferComponent::Impl::convertStringToHTTPMethod(const string &method_string)
+{
+    if (method_string == "GET") {
+        return HTTPMethod::GET;
+    } else if (method_string == "POST") {
+        return HTTPMethod::POST;
+    } else if (method_string == "PATCH") {
+        return HTTPMethod::PATCH;
+    } else if (method_string == "CONNECT") {
+        return HTTPMethod::CONNECT;
+    } else if (method_string == "PUT") {
+        return HTTPMethod::PUT;
+    } else {
+        return genError("Unknown HTTP method");
+    }
+}
+
+Maybe<HTTPRequestEvent>
+MessagingBufferComponent::Impl::serializeOldData(const string &data)
+{
+    try {
+        stringstream in;
+        in.str(data);
+        cereal::JSONInputArchive in_ar(in);
+
+        HTTPRequestEvent req;
+        req.load(in_ar);
+        return req;
+    } catch (cereal::Exception &e) {
+        return genError("JSON parsing failed: " + string(e.what()));
+    } catch (exception &e) {
+        return genError(e.what());
+    }
+}
+
+void
+MessagingBufferComponent::Impl::convertLegacyBuffer(const string &body_file_path)
+{
+    ifstream file(body_file_path);
+    if (!file.is_open()) {
+        dbgTrace(D_MESSAGING_BUFFER) << "No body file found: " << body_file_path;
+        return;
+    }
+
+    string request;
+    while (getline(file, request)) {
+        auto http_request_event = serializeOldData(encryptor->base64Decode(request));
+        if (!http_request_event.ok()) {
+            dbgWarning(D_MESSAGING_BUFFER) << "Error to serialize http_request_event: " << http_request_event.getErr();
+            continue;
+        }
+
+        auto http_method = convertStringToHTTPMethod(http_request_event.unpack().getMethod());
+        if (!http_method.ok()) {
+            dbgWarning(D_MESSAGING_BUFFER) << "Error to convert http_method: " << http_method.getErr();
+            continue;
+        }
+
+        pushNewBufferedMessage(
+            http_request_event.unpack().getBody(),
+            http_method.unpack(),
+            http_request_event.unpack().getURL(),
+            MessageCategory::GENERIC,
+            MessageMetadata(),
+            true
+        );
+    }
+}
+
+void
+MessagingBufferComponent::Impl::removeLegacyBuffer(const string &root_path, const string &executable_name)
+{
+    string file_path = root_path + "manager" + executable_name;
+    ifstream file(file_path);
+    if (!file.is_open()) {
+        dbgTrace(D_MESSAGING_BUFFER) << "No legacy MessagingBuffer buffers found: " << file_path;
+        return;
+    }
+
+    string line;
+    while (getline(file, line)) {
+        dbgTrace(D_MESSAGING_BUFFER) << "Line: " << line;
+        string body_file_path = root_path + line + executable_name;
+        convertLegacyBuffer(body_file_path);
+        if (remove(body_file_path.c_str()) == 0) {
+            dbgDebug(D_MESSAGING_BUFFER) << "File successfully removed: " << body_file_path;
+        } else {
+            dbgWarning(D_MESSAGING_BUFFER) << "Failed to remove file: " << body_file_path;
+        }
+    }
+
+    file.close();
+    if (remove(file_path.c_str()) == 0) {
+        dbgDebug(D_MESSAGING_BUFFER) << "Manager file successfully removed: " << file_path;
+    } else {
+        dbgWarning(D_MESSAGING_BUFFER) << "Failed to remove file manager: " << file_path;
+    }
+}
+// LCOV_EXCL_STOP
+
+bool
+MessagingBufferComponent::Impl::canWriteToDisk(size_t message_size) const
+{
+    dbgTrace(D_MESSAGING_BUFFER) << "Handling buffer size in disk";
+    auto maybe_directory_size = getDirectorySize();
+    if (!maybe_directory_size.ok()) {
+        dbgWarning(D_MESSAGING_BUFFER) << "Failed to get directory size. " << maybe_directory_size.getErr();
+        return false;
+    }
+    if ((*maybe_directory_size + message_size) < (max_size_on_disk_MB * 1024 * 1024)) {
+        return true;
+    }
+
+    dbgWarning(D_MESSAGING_BUFFER)
+        << "Buffer size is full. Directry size: "
+        << *maybe_directory_size
+        << ", Message size: "
+        << message_size
+        << ", Max size: "
+        << max_size_on_disk_MB * 1024 * 1024;
+    return false;
+}
+
+void
+MessagingBufferComponent::init()
+{
+    pimpl->init();
+}
+
+MessagingBufferComponent::MessagingBufferComponent() : pimpl(make_unique<Impl>())
+{}
+
+MessagingBufferComponent::~MessagingBufferComponent()
+{}

core/messaging/messaging_buffer_comp/messaging_buffer_comp_ut/CMakeLists.txt

@@ -0,0 +1,7 @@
+add_unit_test(
+    messaging_buffer_comp_ut
+    "messaging_buffer_comp_ut.cc"
+    "messaging_buffer_comp;messaging_comp;metric;event_is;boost_regex;shell_cmd;time_proxy;agent_details;instance_awareness"
+)
+
+add_subdirectory(test_data)

core/messaging/messaging_buffer_comp/messaging_buffer_comp_ut/messaging_buffer_comp_ut.cc

@@ -0,0 +1,264 @@
+#include "messaging_buffer.h"
+
+#include "agent_core_utilities.h"
+#include "agent_details.h"
+#include "config.h"
+#include "config_component.h"
+#include "cptest.h"
+#include "environment.h"
+#include "agent_details.h"
+#include "instance_awareness.h"
+#include "mock/mock_mainloop.h"
+#include "mock/mock_messaging.h"
+#include "mock/mock_tenant_manager.h"
+#include "mock/mock_encryptor.h"
+#include "shell_cmd.h"
+#include "time_proxy.h"
+
+using namespace std;
+using namespace testing;
+
+string removeWhitespaces(const std::string &str);
+
+class TestMessagingBuffer : public Test
+{
+public:
+    TestMessagingBuffer()
+    {
+        env.preload();
+        Singleton::Consume<I_Environment>::from(env)->registerValue<string>("Executable Name", "tmp_test_file");
+
+        config.preload();
+        config.init();
+
+        string config_json =
+            "{"
+            "   \"agentSettings\": [\n"
+            "   {\n"
+            "       \"id\": \"123\",\n"
+            "       \"key\": \"eventBuffer.maxSizeOnDiskInMB\",\n"
+            "       \"value\": \"1\"\n"
+            "   },\n"
+            "   {\n"
+            "       \"id\": \"123\",\n"
+            "       \"key\": \"eventBuffer.baseFolder\",\n"
+            "       \"value\": \"../.." + cptestFnameInExeDir("test_data") + "\"\n"
+            "   }]\n"
+            "}";
+
+        istringstream ss(config_json);
+        Singleton::Consume<Config::I_Config>::from(config)->loadConfiguration({"--id=8"});
+        Singleton::Consume<Config::I_Config>::from(config)->loadConfiguration(ss);
+
+        EXPECT_CALL(mock_mainloop, addRecurringRoutine(_, _, _, "A-sync messaging routine", _))
+            .WillOnce(DoAll(SaveArg<2>(&async_routine), Return(0)));
+
+        EXPECT_CALL(mock_mainloop, addRecurringRoutine(_, _, _, "Handling in-memory messages", _))
+            .WillOnce(DoAll(SaveArg<2>(&memory_routine), Return(0)));
+
+        buffer_comp.init();
+        buffer_provider = Singleton::Consume<I_MessageBuffer>::from(buffer_comp);
+
+        agent_details.setFogDomain("fog_domain");
+        agent_details.setFogPort(443);
+    }
+
+    ~TestMessagingBuffer() { buffer_provider->cleanBuffer(); }
+
+    NiceMock<MockTenantManager> tenant_manager;
+    NiceMock<MockMainLoop> mock_mainloop;
+    NiceMock<MockMessaging> mock_messaging;
+    ConfigComponent config;
+    MessagingBufferComponent buffer_comp;
+    ::Environment env;
+    ShellCmd shell_cmd;
+    StrictMock<MockEncryptor> m_encryptor;
+    TimeProxyComponent time_proxy;
+    AgentDetails agent_details;
+    InstanceAwareness instance_awareness;
+    I_MessageBuffer *buffer_provider;
+    I_MainLoop::Routine async_routine;
+    I_MainLoop::Routine memory_routine;
+};
+
+TEST_F(TestMessagingBuffer, testPeekingEmptyBuffer)
+{
+    auto msg = buffer_provider->peekMessage();
+    ASSERT_FALSE(msg.ok());
+}
+
+static bool
+checkExistence(const string &path)
+{
+    try {
+        struct stat info;
+        if (stat(path.c_str(), &info) != 0) return false;
+        return info.st_mode & S_IFREG;
+    } catch (exception &e) {
+        return false;
+    }
+}
+
+TEST_F(TestMessagingBuffer, testPushOneBuffer)
+{
+    string body = "body";
+    string uri = "uri";
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata = MessageMetadata();
+    HTTPMethod method = HTTPMethod::POST;
+
+    buffer_provider->pushNewBufferedMessage(body, method, uri, category, message_metadata, true);
+
+    auto msg = buffer_provider->peekMessage();
+    ASSERT_TRUE(msg.ok());
+
+    BufferedMessage expected(body, method, uri, category, message_metadata);
+    EXPECT_EQ(*msg, expected);
+    EXPECT_TRUE(checkExistence(cptestFnameInExeDir("test_data") + "/tmp_test_file8.output"));
+}
+
+TEST_F(TestMessagingBuffer, testMultiplePushesAndPulls)
+{
+    string uri = "uri";
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata = MessageMetadata();
+    HTTPMethod method = HTTPMethod::POST;
+
+    string body1 = "body1";
+    string body2 = "body2";
+    string body3 = "body3";
+    string body4 = "body4";
+    string body5 = "body5";
+
+    BufferedMessage expected1(body1, method, uri, category, message_metadata);
+    BufferedMessage expected2(body2, method, uri, category, message_metadata);
+    BufferedMessage expected3(body3, method, uri, category, message_metadata);
+    BufferedMessage expected4(body4, method, uri, category, message_metadata);
+    BufferedMessage expected5(body5, method, uri, category, message_metadata);
+
+    buffer_provider->pushNewBufferedMessage(body1, method, uri, category, message_metadata, true);
+    buffer_provider->pushNewBufferedMessage(body2, method, uri, category, message_metadata, true);
+
+    auto msg = buffer_provider->peekMessage();
+    ASSERT_TRUE(msg.ok());
+    EXPECT_EQ(*msg, expected2);
+    buffer_provider->popMessage();
+
+    buffer_provider->pushNewBufferedMessage(body3, method, uri, category, message_metadata, true);
+    buffer_provider->pushNewBufferedMessage(body4, method, uri, category, message_metadata, true);
+
+    msg = buffer_provider->peekMessage();
+    ASSERT_TRUE(msg.ok());
+    EXPECT_EQ(*msg, expected1);
+    buffer_provider->popMessage();
+
+    msg = buffer_provider->peekMessage();
+    ASSERT_TRUE(msg.ok());
+    EXPECT_EQ(*msg, expected4);
+    buffer_provider->popMessage();
+
+
+    msg = buffer_provider->peekMessage();
+    ASSERT_TRUE(msg.ok());
+    EXPECT_EQ(*msg, expected3);
+    buffer_provider->popMessage();
+
+    buffer_provider->pushNewBufferedMessage(body5, method, uri, category, message_metadata, true);
+
+    msg = buffer_provider->peekMessage();
+    ASSERT_TRUE(msg.ok());
+    EXPECT_EQ(*msg, expected5);
+    buffer_provider->popMessage();
+
+    msg = buffer_provider->peekMessage();
+    ASSERT_FALSE(msg.ok());
+}
+
+TEST_F(TestMessagingBuffer, testPushMoreThanAllowed)
+{
+    string body_1 = "body";
+    string body_2 = string(1024 * 1024 * 1, 'a'); // 1MB
+    string body_3 = "body";
+    string uri_1 = "uri_1";
+    string uri_2 = "uri_2";
+    string uri_3 = "uri_3";
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata = MessageMetadata();
+    HTTPMethod method = HTTPMethod::POST;
+
+    BufferedMessage expected1(body_1, method, uri_1, category, message_metadata);
+    BufferedMessage expected3(body_3, method, uri_3, category, message_metadata);
+
+    buffer_provider->pushNewBufferedMessage(body_1, method, uri_1, category, message_metadata, true);
+    buffer_provider->pushNewBufferedMessage(body_2, method, uri_2, category, message_metadata, true);
+    buffer_provider->pushNewBufferedMessage(body_3, method, uri_3, category, message_metadata, true);
+
+    auto msg = buffer_provider->peekMessage();
+    ASSERT_TRUE(msg.ok());
+    EXPECT_EQ(*msg, expected3);
+    buffer_provider->popMessage();
+
+    msg = buffer_provider->peekMessage();
+    ASSERT_TRUE(msg.ok());
+    EXPECT_EQ(*msg, expected1);
+    buffer_provider->popMessage();
+
+    msg = buffer_provider->peekMessage();
+    ASSERT_FALSE(msg.ok());
+}
+
+TEST_F(TestMessagingBuffer, testRoutinePulling)
+{
+    string body_1 = "body1";
+    string body_2 = "body2";
+    string uri_1 = "uri_1";
+    string uri_2 = "uri_2";
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata = MessageMetadata();
+    HTTPMethod method = HTTPMethod::POST;
+
+    buffer_provider->pushNewBufferedMessage(body_1, method, uri_1, category, message_metadata, true);
+    buffer_provider->pushNewBufferedMessage(body_2, method, uri_2, category, message_metadata, true);
+
+    HTTPResponse res(HTTPStatusCode::HTTP_OK, "");
+
+    EXPECT_CALL(mock_messaging, sendSyncMessage(method, uri_1, body_1, _, _)).WillOnce(Return(res));
+    EXPECT_CALL(mock_messaging, sendSyncMessage(method, uri_2, body_2, _, _)).WillOnce(Return(res));
+
+    async_routine();
+}
+
+TEST_F(TestMessagingBuffer, testRoutinInMemory)
+{
+    string body_1 = "body1";
+    string body_2 = "body2";
+    string body_3 = "body3";
+    string uri_1 = "uri_1";
+    string uri_2 = "uri_2";
+    string uri_3 = "uri_3";
+
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata = MessageMetadata();
+    HTTPMethod method = HTTPMethod::POST;
+
+    buffer_provider->pushNewBufferedMessage(body_1, method, uri_1, category, message_metadata, false);
+    buffer_provider->pushNewBufferedMessage(body_2, method, uri_2, category, message_metadata, false);
+    buffer_provider->pushNewBufferedMessage(body_3, method, uri_3, category, message_metadata, false);
+
+    HTTPResponse res(HTTPStatusCode::HTTP_OK, "");
+    Maybe<HTTPResponse, HTTPResponse> err = genError(res);
+
+    EXPECT_CALL(mock_messaging, sendSyncMessage(method, uri_1, body_1, _, _)).WillOnce(Return(res));
+    EXPECT_CALL(mock_messaging, sendSyncMessage(method, uri_2, body_2, _, _)).WillOnce(Return(err));
+    EXPECT_CALL(mock_messaging, sendSyncMessage(method, uri_3, body_3, _, _)).WillOnce(Return(res));
+
+    memory_routine();
+
+    auto msg = buffer_provider->peekMessage();
+    ASSERT_TRUE(msg.ok());
+    EXPECT_EQ(*msg, BufferedMessage(body_2, method, uri_2, category, message_metadata));
+    buffer_provider->popMessage();
+
+    msg = buffer_provider->peekMessage();
+    ASSERT_FALSE(msg.ok());
+}

core/messaging/messaging_buffer_comp/messaging_buffer_comp_ut/test_data/CMakeLists.txt

@@ -0,0 +1,2 @@
+
+

core/messaging/messaging_comp/CMakeLists.txt

@@ -0,0 +1,5 @@
+add_library(messaging_comp
+    messaging_comp.cc
+    http_request.cc
+    http_response.cc)
+add_subdirectory(messaging_comp_ut)

core/messaging/messaging_comp/http_request.cc

@@ -0,0 +1,149 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#include "http_request.h"
+
+#include <sstream>
+#include <string>
+
+#include "messaging.h"
+
+using namespace std;
+
+USE_DEBUG_FLAG(D_MESSAGING);
+
+void
+HTTPRequest::insertHeader(const string &header_key, const string &header_val)
+{
+    headers[header_key] = header_val;
+}
+
+bool
+HTTPRequest::setConnectionHeaders(const Connection &conn)
+{
+    string host = conn.getConnKey().getHostName();
+    string uri_prefix = conn.isOverProxy() ? "http://" + host : "";
+
+    switch (method) {
+        case HTTPMethod::GET: {
+            method_statement = "GET " + uri_prefix + uri + " HTTP/1.1";
+            break;
+        }
+        case HTTPMethod::POST: {
+            method_statement = "POST " + uri_prefix + uri + " HTTP/1.1";
+            break;
+        }
+        case HTTPMethod::PATCH: {
+            method_statement = "PATCH " + uri_prefix + uri + " HTTP/1.1";
+            break;
+        }
+        case HTTPMethod::PUT: {
+            method_statement = "PUT " + uri_prefix + uri + " HTTP/1.1";
+            break;
+        }
+        case HTTPMethod::CONNECT: {
+            host = host + ":" + to_string(conn.getConnKey().getPort());
+            method_statement = "CONNECT " + host + " HTTP/1.1";
+            break;
+        }
+        default: {
+            return false;
+        }
+    }
+
+    insertHeader("Host", host);
+    insertHeader("Content-Length", to_string(body.size()));
+    insertHeader("Content-type", "application/json");
+    insertHeader("Accept-Encoding", "identity");
+    if (headers.find("Connection") == headers.end()) {
+        insertHeader("Connection", "keep-alive");
+    }
+    return true;
+}
+
+Maybe<HTTPRequest>
+HTTPRequest::prepareRequest(
+    const Connection &conn,
+    HTTPMethod method,
+    const string &uri,
+    const map<string, string> &headers,
+    const string &body
+)
+{
+    HTTPRequest req(method, uri, headers, body);
+
+    if (!req.setConnectionHeaders(conn)) return genError("Failed to identify the HTTP method");
+
+    string agent_registration_query = R"("authenticationMethod": "token")";
+    bool dont_add_access_token = false;
+    if (method == HTTPMethod::CONNECT || body.find(agent_registration_query) != string::npos) {
+        dont_add_access_token = true;
+        dbgTrace(D_MESSAGING) << "Request is for agent authentication";
+    }
+    auto res = req.addAccessToken(conn, dont_add_access_token);
+    if (!res.ok()) return res.passErr();
+
+    if (conn.isOverProxy()) {
+        auto res = req.addProxyAuthorization(conn);
+        if (!res.ok()) return res.passErr();
+    }
+
+    return req;
+}
+
+string
+HTTPRequest::toString() const
+{
+    stringstream res;
+    res << method_statement << "\r\n";
+    for (const auto &header : headers) {
+        res << header.first << ": " << header.second << "\r\n";
+    }
+    res << "\r\n" << body;
+    return res.str();
+}
+
+Maybe<void>
+HTTPRequest::addAccessToken(const Connection &conn, bool dont_add_access_token)
+{
+    if (headers.find("Authorization") != headers.end() || dont_add_access_token) return Maybe<void>();
+
+    if (!conn.getExternalCertificate().empty()) {
+        insertHeader("Authorization", conn.getExternalCertificate());
+        return Maybe<void>();
+    }
+
+    string access_token = Singleton::Consume<I_AgentDetails>::by<Messaging>()->getAccessToken();
+    if (access_token.empty()) return genError("Access token is missing.");
+    insertHeader("Authorization", "Bearer " + access_token);
+    return Maybe<void>();
+}
+
+Maybe<void>
+HTTPRequest::addProxyAuthorization(const Connection &conn)
+{
+    insertHeader("Accept", "*/*");
+    insertHeader("Proxy-Connection", "Keep-Alive");
+
+    if (!conn.isUnsecure()) return Maybe<void>();
+
+    MessageProxySettings proxy_settings = conn.getProxySettings();
+    if (proxy_settings.getProxyAuth().empty()) {
+        dbgTrace(D_MESSAGING) << "No proxy authentication was set";
+        return Maybe<void>();
+    }
+
+    I_Encryptor *encryptor = Singleton::Consume<I_Encryptor>::by<Messaging>();
+    insertHeader("Proxy-Authorization", "Basic " + encryptor->base64Encode(proxy_settings.getProxyAuth()));
+    return Maybe<void>();
+}

core/messaging/messaging_comp/http_response.cc

@@ -0,0 +1,286 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#include "response_parser.h"
+
+#include <algorithm>
+#include <string>
+#include <vector>
+
+using namespace std;
+
+USE_DEBUG_FLAG(D_MESSAGING);
+
+static const string CRLF = "\r\n";
+
+static const map<HTTPStatusCode, string> status_code_to_string = {
+    {HTTPStatusCode::NO_HTTP_RESPONSE,                      "0 - NO_HTTP_RESPONSE"                      },
+    { HTTPStatusCode::HTTP_OK,                              "200 - HTTP_OK"                             },
+    { HTTPStatusCode::HTTP_NO_CONTENT,                      "204 - HTTP_NO_CONTENT"                     },
+    { HTTPStatusCode::HTTP_MULTI_STATUS,                    "207 - HTTP_MULTI_STATUS"                   },
+    { HTTPStatusCode::HTTP_BAD_REQUEST,                     "400 - HTTP_BAD_REQUEST"                    },
+    { HTTPStatusCode::HTTP_UNAUTHORIZED,                    "401 - HTTP_UNAUTHORIZED"                   },
+    { HTTPStatusCode::HTTP_FORBIDDEN,                       "403 - HTTP_FORBIDDEN"                      },
+    { HTTPStatusCode::HTTP_NOT_FOUND,                       "404 - HTTP_NOT_FOUND"                      },
+    { HTTPStatusCode::HTTP_METHOD_NOT_ALLOWED,              "405 - HTTP_METHOD_NOT_ALLOWED"             },
+    { HTTPStatusCode::HTTP_PROXY_AUTHENTICATION_REQUIRED,   "407 - HTTP_PROXY_AUTHENTICATION_REQUIRED"  },
+    { HTTPStatusCode::HTTP_REQUEST_TIME_OUT,                "408 - HTTP_REQUEST_TIME_OUT"               },
+    { HTTPStatusCode::HTTP_PAYLOAD_TOO_LARGE,               "413 - HTTP_PAYLOAD_TOO_LARGE"              },
+    { HTTPStatusCode::HTTP_INTERNAL_SERVER_ERROR,           "500 - HTTP_INTERNAL_SERVER_ERROR"          },
+    { HTTPStatusCode::HTTP_NOT_IMPLEMENTED,                 "501 - HTTP_NOT_IMPLEMENTED"                },
+    { HTTPStatusCode::HTTP_BAD_GATEWAY,                     "502 - HTTP_BAD_GATEWAY"                    },
+    { HTTPStatusCode::HTTP_SERVICE_UNABAILABLE,             "503 - HTTP_SERVICE_UNABAILABLE"            },
+    { HTTPStatusCode::HTTP_GATEWAY_TIMEOUT,                 "504 - HTTP_GATEWAY_TIMEOUT"                },
+    { HTTPStatusCode::HTTP_VERSION_NOT_SUPPORTED,           "505 - HTTP_VERSION_NOT_SUPPORTED"          },
+    { HTTPStatusCode::HTTP_VARIANT_ALSO_NEGOTIATES,         "506 - HTTP_VARIANT_ALSO_NEGOTIATES"        },
+    { HTTPStatusCode::HTTP_INSUFFICIENT_STORAGE,            "507 - HTTP_INSUFFICIENT_STORAGE"           },
+    { HTTPStatusCode::HTTP_LOOP_DETECTED,                   "508 - HTTP_LOOP_DETECTED"                  },
+    { HTTPStatusCode::HTTP_NOT_EXTENDED,                    "510 - HTTP_NOT_EXTENDED"                   },
+    { HTTPStatusCode::HTTP_NETWORK_AUTHENTICATION_REQUIRED, "511 - HTTP_NETWORK_AUTHENTICATION_REQUIRED"},
+    { HTTPStatusCode::HTTP_UNKNOWN,                         "-1 - HTTP_UNKNOWN"                         },
+    { HTTPStatusCode::HTTP_SUSPEND,                         "-2 - HTTP_SUSPEND"                         }
+};
+
+static const map<int, HTTPStatusCode> num_to_status_code = {
+    {200,  HTTPStatusCode::HTTP_OK                             },
+    { 204, HTTPStatusCode::HTTP_NO_CONTENT                     },
+    { 207, HTTPStatusCode::HTTP_MULTI_STATUS                   },
+    { 400, HTTPStatusCode::HTTP_BAD_REQUEST                    },
+    { 401, HTTPStatusCode::HTTP_UNAUTHORIZED                   },
+    { 403, HTTPStatusCode::HTTP_FORBIDDEN                      },
+    { 404, HTTPStatusCode::HTTP_NOT_FOUND                      },
+    { 405, HTTPStatusCode::HTTP_METHOD_NOT_ALLOWED             },
+    { 407, HTTPStatusCode::HTTP_PROXY_AUTHENTICATION_REQUIRED  },
+    { 408, HTTPStatusCode::HTTP_REQUEST_TIME_OUT               },
+    { 413, HTTPStatusCode::HTTP_PAYLOAD_TOO_LARGE              },
+    { 500, HTTPStatusCode::HTTP_INTERNAL_SERVER_ERROR          },
+    { 501, HTTPStatusCode::HTTP_NOT_IMPLEMENTED                },
+    { 502, HTTPStatusCode::HTTP_BAD_GATEWAY                    },
+    { 503, HTTPStatusCode::HTTP_SERVICE_UNABAILABLE            },
+    { 504, HTTPStatusCode::HTTP_GATEWAY_TIMEOUT                },
+    { 505, HTTPStatusCode::HTTP_VERSION_NOT_SUPPORTED          },
+    { 506, HTTPStatusCode::HTTP_VARIANT_ALSO_NEGOTIATES        },
+    { 507, HTTPStatusCode::HTTP_INSUFFICIENT_STORAGE           },
+    { 508, HTTPStatusCode::HTTP_LOOP_DETECTED                  },
+    { 510, HTTPStatusCode::HTTP_NOT_EXTENDED                   },
+    { 511, HTTPStatusCode::HTTP_NETWORK_AUTHENTICATION_REQUIRED}
+};
+
+const string &
+HTTPResponse::getBody() const
+{
+    return body;
+}
+
+HTTPStatusCode
+HTTPResponse::getHTTPStatusCode() const
+{
+    return status_code;
+}
+
+string
+HTTPResponse::toString() const
+{
+    auto code = status_code_to_string.find(status_code);
+    dbgAssert(code != status_code_to_string.end()) << "Unknown status code " << int(status_code);
+
+    return "[Status-code]: " + code->second + ", [Body]: " + (body.empty() ? "{}" : body);
+}
+
+Maybe<HTTPResponse>
+HTTPResponseParser::parseData(const string &data, bool is_connect)
+{
+    if (data.empty()) return genError("Data is empty");
+    raw_response += data;
+
+    if (!status_code.ok()) {
+        if (!parseStatusLine()) return genError("Failed to parse the status line. Error: " + status_code.getErr());
+    }
+
+    if (!headers.ok()) {
+        if (!handleHeaders()) return genError("Failed to parse the headers. Error: " + headers.getErr());
+    }
+
+    if (!handleBody(is_connect)) return genError("Response not ready!");
+
+    return HTTPResponse(status_code.unpack(), body);
+}
+
+static string
+strip(const string &str)
+{
+    string res;
+    for (auto ch : str) {
+        if (!isspace(ch)) res += tolower(ch);
+    }
+    return res;
+}
+
+bool
+HTTPResponseParser::handleHeaders()
+{
+    stringstream ss(raw_response);
+    map<string, string> header_map;
+
+    while (true) {
+        string header;
+        getline(ss, header);
+
+        if (header.empty()) {
+            headers = genError("Headers not complete");
+            return false;
+        }
+
+        if (header == "\r") {
+            headers = header_map;
+            ss.sync();
+            raw_response = raw_response.substr(ss.tellg());
+            return true;
+        }
+
+        auto colon_index = header.find_first_of(":");
+        if (colon_index == string::npos) {
+            // The only case where not finding a colon isn't an error is if we run out of data
+            error = !ss.str().empty();
+            headers = genError(error ? "Invalid headers: " + header : "Did not reach end of headers");
+            return false;
+        }
+
+        auto header_key = header.substr(0, colon_index);
+        auto header_val = header.substr(colon_index + 2);
+        header_map[strip(header_key)] = strip(header_val);
+    }
+}
+
+Maybe<string>
+HTTPResponseParser::getHeaderVal(const string &header_key)
+{
+    auto headers_map = headers.unpack();
+    auto header = headers_map.find(header_key);
+    if (header == headers_map.end()) {
+        return genError("Header\'" + header_key + "\' not found.");
+    }
+    return header->second;
+}
+
+bool
+HTTPResponseParser::handleBody(bool is_connect)
+{
+    if (*status_code == HTTPStatusCode::HTTP_OK && is_connect) return true;
+
+    if (*status_code == HTTPStatusCode::HTTP_NO_CONTENT) return raw_response.empty();
+
+    auto content_length = getHeaderVal("content-length");
+    if (content_length.ok()) {
+        size_t body_length;
+        try {
+            body_length = stoi(content_length.unpack());
+        } catch (const exception &err) {
+            return false;
+        }
+
+        body += raw_response;
+        raw_response.clear();
+        return body.size() == body_length;
+    }
+
+    auto transfer_encoding = getHeaderVal("transfer-encoding");
+    if (transfer_encoding.ok() && *transfer_encoding == "chunked") return getChunkedResponse();
+
+    dbgError(D_MESSAGING) << "Response has neither content-lenght nor chunked encoded";
+    return false;
+}
+
+bool
+HTTPResponseParser::getChunkedResponse()
+{
+    if (!isLegalChunkedResponse(raw_response)) return false;
+
+    size_t chunk_size = 0;
+
+    for (auto line_end = raw_response.find(CRLF); line_end != string::npos; line_end = raw_response.find(CRLF)) {
+        string line = raw_response.substr(0, line_end);
+        try {
+            chunk_size = stoi(line, nullptr, 16);
+        } catch (const exception &) {
+            dbgWarning(D_MESSAGING) << "Failed to convert chunk length to a number. Line: " << line;
+            return false;
+        }
+
+        if (line_end + 2 + chunk_size > raw_response.length()) {
+            dbgWarning(D_MESSAGING) << "Invalid chunked data structure - chunk-size is bigger than chunk-data";
+            return false;
+        }
+        string chunk_body = raw_response.substr(line_end + 2, chunk_size);
+        raw_response = raw_response.substr(line_end + 2 + chunk_size);
+
+        if (raw_response.find(CRLF) != 0) {
+            dbgWarning(D_MESSAGING) << "Invalid chunked data structure - chunk-data missing final CRLF sequence";
+            return false;
+        }
+        raw_response = raw_response.substr(2);
+
+        body += chunk_body;
+    }
+
+    if (chunk_size != 0) {
+        dbgDebug(D_MESSAGING) << "Invalid chunked data structure - last-chunk of the body is not sized 0";
+        return false;
+    }
+    return true;
+}
+
+bool
+HTTPResponseParser::isLegalChunkedResponse(const string &res)
+{
+    auto end_of_data = res.find("0\r\n\r\n");
+    return end_of_data != string::npos && res.length() == end_of_data + 5;
+}
+
+bool
+HTTPResponseParser::parseStatusLine()
+{
+    auto end_of_first_line = raw_response.find(CRLF);
+    if (end_of_first_line == string::npos) {
+        status_code = genError("No Status Line was received.");
+        return false;
+    }
+
+    auto status_line = raw_response.substr(0, end_of_first_line);
+    raw_response = raw_response.substr(end_of_first_line + 2);
+
+    // Also status text can be supported at the future.
+    if (status_line.find("HTTP/1.") == string::npos) {
+        status_code = genError("Status code not found.");
+        error = true;
+        return false;
+    }
+
+    int status_num;
+    try {
+        status_num = stoi(status_line.substr(9, 3));
+    } catch (const exception &err) {
+        status_code = genError("Failed to convert status code to a number. Status code: " + status_line.substr(9, 3));
+        error = true;
+        return false;
+    }
+
+    auto status = num_to_status_code.find(status_num);
+    if (status != num_to_status_code.end()) {
+        status_code = status->second;
+    } else {
+        dbgWarning(D_MESSAGING) << "Unknown HTTP status code: " << status_num;
+        status_code = HTTPStatusCode::HTTP_UNKNOWN;
+    }
+    return true;
+}

core/messaging/messaging_comp/messaging_comp.cc

@@ -0,0 +1,320 @@
+// Copyright (C) 2022 Check Point Software Technologies Ltd. All rights reserved.
+
+// Licensed under the Apache License, Version 2.0 (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.
+
+#include "messaging_comp.h"
+
+#include <fstream>
+#include <string>
+
+#include "agent_core_utilities.h"
+#include "connection_comp.h"
+#include "debug.h"
+#include "messaging_buffer.h"
+
+using namespace std;
+
+USE_DEBUG_FLAG(D_MESSAGING);
+
+void
+MessagingComp::init()
+{
+    i_conn = Singleton::Consume<I_MessagingConnection>::from<ConnectionComponent>();
+    i_messaging_buffer = Singleton::Consume<I_MessageBuffer>::from<MessagingBufferComponent>();
+    agent_details = Singleton::Consume<I_AgentDetails>::by<Messaging>();
+
+    auto i_mainloop = Singleton::Consume<I_MainLoop>::by<Messaging>();
+    auto i_time_get = Singleton::Consume<I_TimeGet>::by<Messaging>();
+    auto cache_timeout = getConfigurationWithDefault<int>(40, "message", "Cache timeout");
+    fog_get_requests_cache.startExpiration(chrono::seconds(cache_timeout), i_mainloop, i_time_get);
+
+    should_buffer_failed_messages = getConfigurationWithDefault<bool>(
+        getProfileAgentSettingWithDefault<bool>(true, "eventBuffer.bufferFailedRequests"),
+        "message",
+        "Buffer Failed Requests"
+    );
+}
+
+static bool
+isMessageToFog(const MessageMetadata message_metadata)
+{
+    return message_metadata.isToFog();
+}
+
+Maybe<Connection>
+MessagingComp::getConnection(MessageCategory category, const MessageMetadata &metadata)
+{
+    auto persistant_conn = getPersistentConnection(metadata, category);
+    if (persistant_conn.ok()) {
+        dbgTrace(D_MESSAGING) << "Found a persistant connection";
+        return persistant_conn;
+    }
+    dbgDebug(D_MESSAGING) << persistant_conn.getErr();
+
+    auto maybe_conn = i_conn->establishConnection(metadata, category);
+    if (!maybe_conn.ok()) {
+        dbgWarning(D_MESSAGING) << "Failed to establish connection: " << maybe_conn.getErr();
+    }
+    return maybe_conn;
+}
+
+Maybe<HTTPResponse, HTTPResponse>
+MessagingComp::sendMessage(
+    HTTPMethod method,
+    const string &uri,
+    const string &body,
+    MessageCategory category,
+    const MessageMetadata &message_metadata
+)
+{
+    auto maybe_conn = getConnection(category, message_metadata);
+    if (!maybe_conn.ok()) {
+        dbgWarning(D_MESSAGING) << "Failed to get connection. Error: " << maybe_conn.getErr();
+        return genError<HTTPResponse>(HTTPStatusCode::HTTP_UNKNOWN, maybe_conn.getErr());
+    }
+
+    Connection conn = maybe_conn.unpack();
+    if (conn.isSuspended()) return suspendMessage(body, method, uri, category, message_metadata);
+
+    bool is_to_fog = isMessageToFog(message_metadata);
+    auto metadata = message_metadata;
+    if (is_to_fog) {
+        if (method == HTTPMethod::GET && fog_get_requests_cache.doesKeyExists(uri)) {
+            HTTPResponse res = fog_get_requests_cache.getEntry(uri);
+            dbgTrace(D_MESSAGING) << "Response returned from Fog cache. res body: " << res.getBody();
+
+            return fog_get_requests_cache.getEntry(uri);
+        }
+
+        auto i_env = Singleton::Consume<I_Environment>::by<Messaging>();
+        metadata.insertHeader("User-Agent", "Infinity Next (a7030abf93a4c13)");
+        metadata.insertHeaders(i_env->getCurrentHeadersMap());
+    }
+
+    auto req = HTTPRequest::prepareRequest(conn, method, uri, metadata.getHeaders(), body);
+    if (!req.ok()) return genError(HTTPResponse(HTTPStatusCode::HTTP_UNKNOWN, req.getErr()));
+
+    auto response = i_conn->sendRequest(conn, *req);
+    if (!response.ok()) return response.passErr();
+
+    if (is_to_fog && method == HTTPMethod::GET) fog_get_requests_cache.emplaceEntry(uri, *response);
+    return response;
+}
+
+Maybe<HTTPResponse, HTTPResponse>
+MessagingComp::sendSyncMessage(
+    HTTPMethod method,
+    const string &uri,
+    const string &body,
+    MessageCategory category,
+    const MessageMetadata &message_metadata
+)
+{
+    Maybe<HTTPResponse, HTTPResponse> is_msg_send = sendMessage(method, uri, body, category, message_metadata);
+
+    if (is_msg_send.ok()) return *is_msg_send;
+
+    if (should_buffer_failed_messages && message_metadata.shouldBufferMessage()) {
+        dbgTrace(D_MESSAGING) << "After sending error, buffering the message";
+        i_messaging_buffer->pushNewBufferedMessage(body, method, uri, category, message_metadata, false);
+    }
+    return is_msg_send.passErr();
+}
+
+void
+MessagingComp::sendAsyncMessage(
+    HTTPMethod method,
+    const string &uri,
+    const string &body,
+    MessageCategory category,
+    const MessageMetadata &message_metadata
+)
+{
+    i_messaging_buffer->pushNewBufferedMessage(body, method, uri, category, message_metadata, false);
+}
+
+Maybe<HTTPStatusCode, HTTPResponse>
+MessagingComp::downloadFile(
+    HTTPMethod method,
+    const string &uri,
+    const string &download_file_path,
+    MessageCategory category,
+    const MessageMetadata &message_metadata
+)
+{
+    dbgTrace(D_MESSAGING) << "Send download file message";
+    string parent_directory = download_file_path.substr(0, download_file_path.find_last_of("/\\"));
+    if (!NGEN::Filesystem::exists(parent_directory)) {
+        if (!NGEN::Filesystem::makeDirRecursive(parent_directory)) {
+            string creation_err = "Failed to create the parent directory. Path: " + parent_directory;
+            dbgWarning(D_MESSAGING) << creation_err;
+            return genError(HTTPResponse(HTTPStatusCode::HTTP_UNKNOWN, creation_err));
+        }
+    }
+
+    auto response = sendSyncMessage(method, uri, "", category, message_metadata);
+    if (!response.ok()) return response.passErr();
+
+    ofstream file_stream(download_file_path);
+    if (!file_stream.is_open()) {
+        string open_err = "Failed to open the destination file. Path: " + download_file_path;
+        dbgWarning(D_MESSAGING) << open_err;
+        return genError(HTTPResponse(HTTPStatusCode::HTTP_UNKNOWN, open_err));
+    }
+    file_stream << response.unpack().getBody();
+    file_stream.close();
+
+    dbgTrace(D_MESSAGING) << "Successfully downloaded and save file to: " << download_file_path;
+    return HTTPStatusCode::HTTP_OK;
+}
+
+Maybe<HTTPStatusCode, HTTPResponse>
+MessagingComp::uploadFile(
+    const string &uri,
+    const string &upload_file_path,
+    MessageCategory category,
+    const MessageMetadata &message_metadata
+)
+{
+    dbgTrace(D_MESSAGING) << "Send upload file message";
+
+    ifstream file(upload_file_path);
+    if (!file.is_open()) {
+        string open_err = "Failed to open the file to upload. Path: " + upload_file_path;
+        dbgWarning(D_MESSAGING) << open_err;
+        return genError(HTTPResponse(HTTPStatusCode::HTTP_UNKNOWN, open_err));
+    }
+
+    stringstream buffer;
+    buffer << file.rdbuf();
+    file.close();
+
+    Maybe<HTTPResponse, HTTPResponse> response =
+        sendSyncMessage(HTTPMethod::PUT, uri, buffer.str(), category, message_metadata);
+
+    if (!response.ok()) return response.passErr();
+
+    dbgTrace(D_MESSAGING) << "Successfully upload file from: " << upload_file_path;
+    return HTTPStatusCode::HTTP_OK;
+}
+
+bool
+MessagingComp::setFogConnection(const string &host, uint16_t port, bool is_secure, MessageCategory category)
+{
+    dbgTrace(D_MESSAGING) << "Setting a fog connection to " << host << ":" << port;
+    MessageMetadata metadata(host, port, true);
+
+    I_ProxyConfiguration *proxy_configuration = Singleton::Consume<I_ProxyConfiguration>::by<Messaging>();
+    auto load_env_proxy = proxy_configuration->loadProxy();
+    if (!load_env_proxy.ok()) {
+        dbgDebug(D_MESSAGING)
+            << "Could not initialize load proxy from environment, Error: "
+            << load_env_proxy.getErr();
+    }
+
+    ProxyProtocol proxy_protocol = is_secure ? ProxyProtocol::HTTPS : ProxyProtocol::HTTP;
+    if (proxy_configuration->getProxyExists(proxy_protocol)) {
+        auto proxy_host = proxy_configuration->getProxyDomain(proxy_protocol);
+        auto proxy_port = proxy_configuration->getProxyPort(proxy_protocol);
+        auto maybe_proxy_auth = proxy_configuration->getProxyAuthentication(proxy_protocol);
+
+        if (proxy_host.ok() && proxy_port.ok()) {
+            string proxy_auth = maybe_proxy_auth.ok() ? *maybe_proxy_auth : "";
+            dbgDebug(D_MESSAGING) << "Setting proxy address: " << *proxy_host << ":" << *proxy_port;
+            MessageProxySettings proxy_settings(proxy_host.unpack(), proxy_auth, proxy_port.unpack());
+            metadata.setProxySettings(proxy_settings);
+        }
+    }
+
+    I_MessagingConnection *i_conn = Singleton::Consume<I_MessagingConnection>::from<ConnectionComponent>();
+    auto conn = i_conn->establishConnection(metadata, category);
+    if (!conn.ok()) {
+        dbgWarning(D_MESSAGING) << "Failed to establish connection to fog: " << conn.getErr();
+        return false;
+    }
+
+    dbgInfo(D_MESSAGING)
+        << "Successfully connected to the Fog: "
+        << host
+        << ":"
+        << port
+        << " via "
+        << (metadata.isProxySet() ? "proxy, using " : "")
+        << (is_secure ? "secure" : "clear")
+        << " connection";
+
+    return true;
+}
+
+bool
+MessagingComp::setFogConnection(MessageCategory category)
+{
+    I_AgentDetails *agent_details = Singleton::Consume<I_AgentDetails>::by<Messaging>();
+
+    if (agent_details->getOrchestrationMode() == OrchestrationMode::OFFLINE) {
+        dbgDebug(D_MESSAGING) << "Agent Is in offline mode and would not attempt connecting to the fog";
+        return true;
+    }
+
+    if (!agent_details->readAgentDetails()) {
+        dbgWarning(D_MESSAGING) << "Cannot establish connection to the Fog, failed to read agent details";
+        return false;
+    }
+
+    auto domain = agent_details->getFogDomain();
+    auto port = agent_details->getFogPort();
+    auto is_secure_connection = agent_details->getSSLFlag();
+
+    if (!domain.ok() || *domain == "" || !port.ok() || port == 0) {
+        dbgWarning(D_MESSAGING) << "Cannot establish connection to the Fog, failed to get host and port details";
+        return false;
+    }
+
+    return setFogConnection(*domain, *port, is_secure_connection, category);
+}
+
+Maybe<Connection>
+MessagingComp::getPersistentConnection(const MessageMetadata &metadata, MessageCategory category) const
+{
+    if (!metadata.isToFog()) {
+        auto maybe_conn = i_conn->getPersistentConnection(metadata.getHostName(), metadata.getPort(), category);
+        if (maybe_conn.ok()) return *maybe_conn;
+        return genError("Failed to get persistant connection based on host and port");
+    }
+
+    auto maybe_conn = i_conn->getFogConnectionByCategory(category);
+    if (maybe_conn.ok()) return maybe_conn;
+    return genError("Failed to get persistant connection to the fog");
+}
+
+Maybe<HTTPResponse, HTTPResponse>
+MessagingComp::suspendMessage(
+    const string &body,
+    HTTPMethod method,
+    const string &uri,
+    MessageCategory category,
+    const MessageMetadata &message_metadata
+) const
+{
+    if (message_metadata.shouldBufferMessage()) {
+        dbgWarning(D_MESSAGING) << "Buffering message due to connection suspended";
+        i_messaging_buffer->pushNewBufferedMessage(body, method, uri, category, message_metadata, false);
+        return genError<HTTPResponse>(
+            HTTPStatusCode::HTTP_SUSPEND,
+            "The connection is suspended due to consecutive message sending errors, message is buffered."
+        );
+    }
+
+    return genError<HTTPResponse>(
+        HTTPStatusCode::HTTP_SUSPEND, "The connection is suspended due to consecutive message sending errors."
+    );
+}

core/messaging/messaging_comp/messaging_comp_ut/CMakeLists.txt

@@ -0,0 +1,5 @@
+add_unit_test(
+    messaging_comp_ut
+    "messaging_comp_ut.cc"
+    "messaging_comp;connection;messaging_buffer_comp;ssl;connkey;singleton;boost_context;rest;metric;event_is;-lboost_regex;agent_details;-lcrypto;"
+)

core/messaging/messaging_comp/messaging_comp_ut/messaging_comp_ut.cc

@@ -0,0 +1,238 @@
+#include "messaging_comp.h"
+
+#include "agent_core_utilities.h"
+#include "config.h"
+#include "config_component.h"
+#include "connection.h"
+#include "cptest.h"
+#include "environment.h"
+#include "mainloop.h"
+#include "mock/mock_agent_details.h"
+#include "mock/mock_mainloop.h"
+#include "mock/mock_time_get.h"
+#include "mock/mock_proxy_configuration.h"
+#include "mocks/mock_messaging_buffer.h"
+#include "mocks/mock_messaging_connection.h"
+#include "rest.h"
+#include "rest_server.h"
+
+using namespace std;
+using namespace testing;
+
+static ostream &
+operator<<(ostream &os, const Maybe<BufferedMessage> &)
+{
+    return os;
+}
+
+static std::ostream &
+operator<<(std::ostream &os, const HTTPResponse &)
+{
+    return os;
+}
+
+static std::ostream &
+operator<<(std::ostream &os, const HTTPStatusCode &)
+{
+    return os;
+}
+
+static std::ostream &
+operator<<(std::ostream &os, const Connection &)
+{
+    return os;
+}
+
+class TestMessagingComp : public testing::Test
+{
+public:
+    TestMessagingComp()
+    {
+        EXPECT_CALL(mock_time_get, getMonotonicTime()).WillRepeatedly(Return(chrono::microseconds(0)));
+
+        ON_CALL(mock_agent_details, getFogDomain()).WillByDefault(Return(Maybe<string>(fog_addr)));
+        ON_CALL(mock_agent_details, getFogPort()).WillByDefault(Return(Maybe<uint16_t>(fog_port)));
+        messaging_comp.init();
+    }
+
+    void
+    setAgentDetails()
+    {
+        EXPECT_CALL(mock_agent_details, getFogDomain()).WillRepeatedly(Return(string(fog_addr)));
+        EXPECT_CALL(mock_agent_details, getFogPort()).WillRepeatedly(Return(fog_port));
+        EXPECT_CALL(mock_agent_details, getOpenSSLDir()).WillRepeatedly(Return(string("/usr/lib/ssl/certs/")));
+        EXPECT_CALL(mock_agent_details, getAccessToken()).WillRepeatedly(Return(string("accesstoken")));
+        EXPECT_CALL(mock_agent_details, readAgentDetails()).WillRepeatedly(Return(true));
+        EXPECT_CALL(mock_proxy_conf, loadProxy()).WillRepeatedly(Return(Maybe<void>()));
+        EXPECT_CALL(mock_proxy_conf, getProxyExists(_)).WillRepeatedly(Return(true));
+        EXPECT_CALL(mock_proxy_conf, getProxyDomain(_)).WillRepeatedly(Return(string("7.7.7.7")));
+        EXPECT_CALL(mock_proxy_conf, getProxyPort(_)).WillRepeatedly(Return(8080));
+        EXPECT_CALL(mock_proxy_conf, getProxyAuthentication(_)).WillRepeatedly(Return(string("cred")));
+    }
+
+    const string fog_addr = "1.2.3.4";
+    uint16_t fog_port = 80;
+    CPTestTempfile agent_details_file;
+    MessagingComp messaging_comp;
+    ::Environment env;
+    ConfigComponent config;
+    NiceMock<MockMessagingConnection> mock_messaging_connection;
+    NiceMock<MockMessagingBuffer> mock_messaging_buffer;
+    NiceMock<MockMainLoop> mock_mainloop;
+    NiceMock<MockTimeGet> mock_time_get;
+    NiceMock<MockAgentDetails> mock_agent_details;
+    NiceMock<MockProxyConfiguration> mock_proxy_conf;
+};
+
+TEST_F(TestMessagingComp, testInitComp)
+{
+    EXPECT_CALL(
+        mock_mainloop, addRecurringRoutine(I_MainLoop::RoutineType::Timer, _, _, "Delete expired cache entries", _)
+    )
+        .WillOnce(Return(0));
+    messaging_comp.init();
+}
+
+TEST_F(TestMessagingComp, testSendSyncMessage)
+{
+    setAgentDetails();
+    string body = "test body";
+    HTTPMethod method = HTTPMethod::POST;
+    string uri = "/test-uri";
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata;
+
+    MessageConnectionKey conn_key(fog_addr, fog_port, MessageCategory::GENERIC);
+    Connection conn(conn_key, message_metadata);
+    EXPECT_CALL(mock_messaging_connection, getFogConnectionByCategory(MessageCategory::GENERIC))
+        .WillOnce(Return(conn));
+
+    HTTPResponse res(HTTPStatusCode::HTTP_OK, "response!!");
+    EXPECT_CALL(mock_messaging_connection, mockSendRequest(_, _, _)).WillOnce(Return(res));
+    auto sending_res = messaging_comp.sendSyncMessage(method, uri, body, category, message_metadata);
+    ASSERT_TRUE(sending_res.ok());
+    HTTPResponse http_res = sending_res.unpack();
+    EXPECT_EQ(http_res.getBody(), "response!!");
+    EXPECT_EQ(http_res.getHTTPStatusCode(), HTTPStatusCode::HTTP_OK);
+}
+
+TEST_F(TestMessagingComp, testSendAsyncMessage)
+{
+    setAgentDetails();
+    string body = "test body";
+    HTTPMethod method = HTTPMethod::POST;
+    string uri = "/test-uri";
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata;
+
+    EXPECT_CALL(mock_messaging_buffer, pushNewBufferedMessage(body, method, uri, category, _, _)).Times(1);
+    messaging_comp.sendAsyncMessage(method, uri, body, category, message_metadata);
+}
+
+TEST_F(TestMessagingComp, testSendSyncMessageOnSuspendedConn)
+{
+    setAgentDetails();
+    string body = "test body";
+    HTTPMethod method = HTTPMethod::POST;
+    string uri = "/test-uri";
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata;
+
+    MessageConnectionKey conn_key(fog_addr, fog_port, MessageCategory::GENERIC);
+    Connection conn(conn_key, message_metadata);
+
+    EXPECT_CALL(mock_time_get, getMonotonicTime())
+        .WillRepeatedly(Invoke([] () { static int j = 0; return chrono::microseconds(++j * 1000 * 1000); }));
+    for (int i = 0; i < 20; i++) {
+        conn.sendRequest(".");
+    }
+    EXPECT_CALL(mock_messaging_connection, getFogConnectionByCategory(MessageCategory::GENERIC))
+        .WillOnce(Return(conn));
+
+    auto sending_res = messaging_comp.sendSyncMessage(method, uri, body, category, message_metadata);
+    ASSERT_FALSE(sending_res.ok());
+    HTTPResponse http_res = sending_res.getErr();
+    EXPECT_EQ(http_res.getBody(), "The connection is suspended due to consecutive message sending errors.");
+    EXPECT_EQ(http_res.getHTTPStatusCode(), HTTPStatusCode::HTTP_SUSPEND);
+}
+
+TEST_F(TestMessagingComp, testUploadFile)
+{
+    string path = cptestFnameInSrcDir("tests_files/file_to_send.txt");
+
+    setAgentDetails();
+    string uri = "/test-uri";
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata;
+
+    MessageConnectionKey conn_key(fog_addr, fog_port, MessageCategory::GENERIC);
+    Connection conn(conn_key, message_metadata);
+    EXPECT_CALL(mock_messaging_connection, getFogConnectionByCategory(MessageCategory::GENERIC))
+        .WillOnce(Return(conn));
+
+    HTTPResponse res(HTTPStatusCode::HTTP_OK, "");
+    EXPECT_CALL(mock_messaging_connection, mockSendRequest(_, _, _)).WillOnce(Return(res));
+    auto upload_res = messaging_comp.uploadFile(uri, path, category, message_metadata);
+    ASSERT_TRUE(upload_res.ok());
+    EXPECT_EQ(upload_res.unpack(), HTTPStatusCode::HTTP_OK);
+}
+
+TEST_F(TestMessagingComp, testDownloadFile)
+{
+    string path = cptestFnameInSrcDir("tests_files/file_to_send.txt");
+
+    setAgentDetails();
+    string uri = "/test-uri";
+    HTTPMethod method = HTTPMethod::GET;
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata;
+
+    MessageConnectionKey conn_key(fog_addr, fog_port, MessageCategory::GENERIC);
+    Connection conn(conn_key, message_metadata);
+    EXPECT_CALL(mock_messaging_connection, getFogConnectionByCategory(MessageCategory::GENERIC))
+        .WillOnce(Return(conn));
+
+    HTTPResponse res(HTTPStatusCode::HTTP_OK, "");
+    EXPECT_CALL(mock_messaging_connection, mockSendRequest(_, _, _)).WillOnce(Return(res));
+    auto upload_res = messaging_comp.downloadFile(method, uri, "/tmp/test.txt", category, message_metadata);
+    ASSERT_TRUE(upload_res.ok());
+    EXPECT_EQ(upload_res.unpack(), HTTPStatusCode::HTTP_OK);
+}
+
+bool
+operator==(const MessageProxySettings &one, const MessageProxySettings &two)
+{
+    return
+        one.getProxyHost() == two.getProxyHost() &&
+        one.getProxyAuth() == two.getProxyAuth() &&
+        one.getProxyPort() == two.getProxyPort();
+}
+
+bool
+operator==(const MessageMetadata &one, const MessageMetadata &two)
+{
+    return
+        one.getHostName() == two.getHostName() &&
+        one.getPort() == two.getPort() &&
+        one.getConnectionFlags() == two.getConnectionFlags() &&
+        one.getProxySettings() == two.getProxySettings() &&
+        one.getExternalCertificate() == two.getExternalCertificate() &&
+        one.getHeaders() == two.getHeaders() &&
+        one.shouldBufferMessage() == two.shouldBufferMessage() &&
+        one.isProxySet() == two.isProxySet();
+}
+
+TEST_F(TestMessagingComp, testSetFogConnection)
+{
+    setAgentDetails();
+
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageConnectionKey conn_key(fog_addr, fog_port, category);
+    MessageMetadata metadata(fog_addr, fog_port, true);
+    MessageProxySettings proxy_settings("7.7.7.7", "cred", 8080);
+    metadata.setProxySettings(proxy_settings);
+    Connection conn(conn_key, metadata);
+
+    EXPECT_CALL(mock_messaging_connection, establishConnection(metadata, category)).WillOnce(Return(conn));
+    EXPECT_TRUE(messaging_comp.setFogConnection(category));
+}

core/messaging/messaging_comp/messaging_comp_ut/tests_files/file_to_send.txt

@@ -0,0 +1 @@
+file to send