Merge pull request #214 from openappsec/Dec_29_2024-Dev

Dec 29 2024 dev

attachments/nginx/nginx_attachment_util/nginx_attachment_util.cc

@@ -173,6 +173,12 @@ getReqBodySizeTrigger()
     return conf_data.getNumericalValue("body_size_trigger");
 }
 
+unsigned int
+getRemoveResServerHeader()
+{
+    return conf_data.getNumericalValue("remove_server_header");
+}
+
 int
 isIPAddress(c_str ip_str)
 {

attachments/nginx/nginx_attachment_util/nginx_attachment_util_ut/nginx_attachment_util_ut.cc

@@ -66,7 +66,8 @@ TEST_F(HttpAttachmentUtilTest, GetValidAttachmentConfiguration)
             "\"static_resources_path\": \"" + static_resources_path + "\",\n"
             "\"min_retries_for_verdict\": 1,\n"
             "\"max_retries_for_verdict\": 3,\n"
-            "\"body_size_trigger\": 777\n"
+            "\"body_size_trigger\": 777,\n"
+            "\"remove_server_header\": 1\n"
         "}\n";
     ofstream valid_configuration_file(attachment_configuration_file_name);
     valid_configuration_file << valid_configuration;
@@ -95,6 +96,7 @@ TEST_F(HttpAttachmentUtilTest, GetValidAttachmentConfiguration)
     EXPECT_EQ(getReqBodySizeTrigger(), 777u);
     EXPECT_EQ(getWaitingForVerdictThreadTimeout(), 75u);
     EXPECT_EQ(getInspectionMode(), ngx_http_inspection_mode::BLOCKING_THREAD);
+    EXPECT_EQ(getRemoveResServerHeader(), 1u);
 
     EXPECT_EQ(isDebugContext("1.2.3.4", "5.6.7.8", 80, "GET", "test", "/abc"), 1);
     EXPECT_EQ(isDebugContext("1.2.3.9", "5.6.7.8", 80, "GET", "test", "/abc"), 0);

components/attachment-intakers/nginx_attachment/nginx_attachment_config.cc

@@ -203,6 +203,13 @@ HttpAttachmentConfig::setFailOpenTimeout()
         "NGINX wait thread timeout msec"
     ));
 
+    conf_data.setNumericalValue("remove_server_header", getAttachmentConf<uint>(
+        0,
+        "agent.removeServerHeader.nginxModule",
+        "HTTP manager",
+        "Response server header removal"
+    ));
+
     uint inspection_mode = getAttachmentConf<uint>(
         static_cast<uint>(ngx_http_inspection_mode_e::NON_BLOCKING_THREAD),
         "agent.inspectionMode.nginxModule",

components/http_manager/http_manager.cc

@@ -18,7 +18,9 @@
 #include <sys/stat.h>
 #include <climits>
 #include <unordered_map>
+#include <unordered_set>
 #include <boost/range/iterator_range.hpp>
+#include <boost/algorithm/string.hpp>
 #include <fstream>
 #include <algorithm>
 
@@ -28,6 +30,7 @@
 #include "http_manager_opaque.h"
 #include "log_generator.h"
 #include "http_inspection_events.h"
+#include "agent_core_utilities.h"
 
 USE_DEBUG_FLAG(D_HTTP_MANAGER);
 
@@ -66,6 +69,22 @@ public:
         i_transaction_table = Singleton::Consume<I_Table>::by<HttpManager>();
 
         Singleton::Consume<I_Logging>::by<HttpManager>()->addGeneralModifier(compressAppSecLogs);
+        
+        const char* ignored_headers_env = getenv("SAAS_IGNORED_UPSTREAM_HEADERS");
+        if (ignored_headers_env) {
+            string ignored_headers_str = ignored_headers_env;
+            ignored_headers_str = NGEN::Strings::removeTrailingWhitespaces(ignored_headers_str);
+
+            if (!ignored_headers_str.empty()) {
+                dbgInfo(D_HTTP_MANAGER)
+                    << "Ignoring SAAS_IGNORED_UPSTREAM_HEADERS environment variable: "
+                    << ignored_headers_str;
+
+                vector<string> ignored_headers_vec;
+                boost::split(ignored_headers_vec, ignored_headers_str, boost::is_any_of(";"));
+                for (const string &header : ignored_headers_vec) ignored_headers.insert(header);
+            }
+        }
     }
 
     FilterVerdict
@@ -90,6 +109,14 @@ public:
             return FilterVerdict(default_verdict);
         }
 
+        if (is_request && ignored_headers.find(static_cast<string>(event.getKey())) != ignored_headers.end()) {
+            dbgTrace(D_HTTP_MANAGER)
+                << "Ignoring header key - "
+                << static_cast<string>(event.getKey())
+                << " - as it is in the ignored headers list";
+            return FilterVerdict(ngx_http_cp_verdict_e::TRAFFIC_VERDICT_INSPECT);
+        }
+
         ScopedContext ctx;
         ctx.registerValue(app_sec_marker_key, i_transaction_table->keyToString(), EnvKeyAttr::LogSection::MARKER);
 
@@ -394,6 +421,7 @@ private:
     I_Table *i_transaction_table;
     static const ngx_http_cp_verdict_e default_verdict;
     static const string app_sec_marker_key;
+    unordered_set<string> ignored_headers;
 };
 
 const ngx_http_cp_verdict_e HttpManager::Impl::default_verdict(ngx_http_cp_verdict_e::TRAFFIC_VERDICT_DROP);

components/include/rate_limit.h

@@ -7,15 +7,21 @@
 #include "singleton.h"
 #include "i_mainloop.h"
 #include "i_environment.h"
+#include "i_geo_location.h"
 #include "i_generic_rulebase.h"
+#include "i_shell_cmd.h"
+#include "i_env_details.h"
 
 class RateLimit
     :
     public Component,
     Singleton::Consume<I_MainLoop>,
     Singleton::Consume<I_TimeGet>,
+    Singleton::Consume<I_GeoLocation>,
     Singleton::Consume<I_Environment>,
-    Singleton::Consume<I_GenericRulebase>
+    Singleton::Consume<I_GenericRulebase>,
+    Singleton::Consume<I_ShellCmd>,
+    Singleton::Consume<I_EnvDetails>
 {
 public:
     RateLimit();

components/include/waap.h

@@ -33,6 +33,7 @@ class I_WaapAssetStatesManager;
 class I_Messaging;
 class I_AgentDetails;
 class I_Encryptor;
+class I_WaapModelResultLogger;
 
 const std::string WAAP_APPLICATION_NAME = "waap application";
 
@@ -50,7 +51,8 @@ class WaapComponent
     Singleton::Consume<I_AgentDetails>,
     Singleton::Consume<I_Messaging>,
     Singleton::Consume<I_Encryptor>,
-    Singleton::Consume<I_Environment>
+    Singleton::Consume<I_Environment>,
+    Singleton::Consume<I_WaapModelResultLogger>
 {
 public:
     WaapComponent();

components/security_apps/http_geo_filter/http_geo_filter.cc

@@ -88,9 +88,17 @@ public:
             dbgWarning(D_GEO_FILTER) << "failed to get source ip from env";
             return EventVerdict(default_action);
         }
-
         auto source_ip = convertIpAddrToString(maybe_source_ip.unpack());
-        ip_set.insert(source_ip);
+
+        // saas profile setting
+        bool ignore_source_ip =
+            getProfileAgentSettingWithDefault<bool>(false, "agent.geoProtaction.ignoreSourceIP");
+        if (ignore_source_ip){
+            dbgDebug(D_GEO_FILTER) << "Geo protection ignoring source ip: " << source_ip;
+        } else {
+            ip_set.insert(convertIpAddrToString(maybe_source_ip.unpack()));
+        }
+
 
         ngx_http_cp_verdict_e exception_verdict = getExceptionVerdict(ip_set);
         if (exception_verdict != ngx_http_cp_verdict_e::TRAFFIC_VERDICT_IRRELEVANT) {
@@ -343,7 +351,7 @@ private:
 
             auto asset_location = i_geo_location->lookupLocation(maybe_source_ip.unpack());
             if (!asset_location.ok()) {
-                dbgWarning(D_GEO_FILTER) << "Lookup location failed for source: " <<
+                dbgDebug(D_GEO_FILTER) << "Lookup location failed for source: " <<
                 source <<
                 ", Error: " <<
                 asset_location.getErr();

components/security_apps/ips/include/ips_signatures.h

@@ -336,9 +336,16 @@ public:
         return metadata.getYear();
     }
 
+    bool
+    isOk() const
+    {
+        return is_loaded;
+    }
+
 private:
     IPSSignatureMetaData metadata;
     std::shared_ptr<BaseSignature> rule;
+    bool is_loaded;
 };
 
 /// \class SignatureAndAction

components/security_apps/ips/ips_signatures.cc

@@ -219,10 +219,16 @@ IPSSignatureMetaData::getYear() const
 void
 CompleteSignature::load(cereal::JSONInputArchive &ar)
 {
+    try {
         ar(cereal::make_nvp("protectionMetadata", metadata));
         RuleDetection rule_detection(metadata.getName());
         ar(cereal::make_nvp("detectionRules", rule_detection));
         rule = rule_detection.getRule();
+        is_loaded = true;
+    } catch (cereal::Exception &e) {
+        is_loaded = false;
+        dbgWarning(D_IPS) << "Failed to load signature: " << e.what();
+    }
 }
 
 MatchType
@@ -367,7 +373,16 @@ SignatureAndAction::matchSilent(const Buffer &sample) const
     if (method.ok()) log << LogField("httpMethod", method.unpack());
 
     auto path  = env->get<Buffer>("HTTP_PATH_DECODED");
-    if (path.ok()) log << LogField("httpUriPath", getSubString(path, 1536), LogFieldOption::XORANDB64);
+    if (path.ok()) {
+        log << LogField("httpUriPath", getSubString(path, 1536), LogFieldOption::XORANDB64);
+    } else {
+        auto transaction_path = env->get<string>(HttpTransactionData::uri_path_decoded);
+        if (transaction_path.ok()) {
+            auto uri_path = transaction_path.unpack();
+            auto question_mark = uri_path.find('?');
+            log << LogField("httpUriPath", uri_path.substr(0, question_mark), LogFieldOption::XORANDB64);
+        }
+    }
 
     auto req_header = ips_state.getTransactionData(IPSCommonTypes::requests_header_for_log);
     if (req_header.ok()) log << LogField("httpRequestHeaders", getSubString(req_header), LogFieldOption::XORANDB64);
@@ -485,13 +500,30 @@ SignatureAndAction::isMatchedPrevent(const Buffer &context_buffer, const set<PMP
     auto method = env->get<string>(HttpTransactionData::method_ctx);
     if (method.ok()) log << LogField("httpMethod", method.unpack());
     uint max_size = getConfigurationWithDefault<uint>(1536, "IPS", "Max Field Size");
+
+    if (trigger.isWebLogFieldActive(url_path)) {
         auto path  = env->get<Buffer>("HTTP_PATH_DECODED");
-    if (path.ok() && trigger.isWebLogFieldActive(url_path)) {
+        if (path.ok()) {
             log << LogField("httpUriPath", getSubString(path, max_size), LogFieldOption::XORANDB64);
+        } else {
+            auto transaction_path = env->get<string>(HttpTransactionData::uri_path_decoded);
+            if (transaction_path.ok()) {
+                auto uri_path = transaction_path.unpack();
+                auto question_mark = uri_path.find('?');
+                log << LogField("httpUriPath", uri_path.substr(0, question_mark), LogFieldOption::XORANDB64);
             }
+        }
+    }
+    if (trigger.isWebLogFieldActive(url_query)) {
         auto query = env->get<Buffer>("HTTP_QUERY_DECODED");
-    if (query.ok() && trigger.isWebLogFieldActive(url_query)) {
+        if (query.ok()) {
             log << LogField("httpUriQuery", getSubString(query, max_size), LogFieldOption::XORANDB64);
+        } else {
+            auto transaction_query = env->get<string>(HttpTransactionData::uri_query_decoded);
+            if (transaction_query.ok()) {
+                log << LogField("httpUriQuery", transaction_query.unpack());
+            }
+        }
     }
 
     auto res_code = env->get<Buffer>("HTTP_RESPONSE_CODE");
@@ -533,9 +565,11 @@ IPSSignaturesResource::load(cereal::JSONInputArchive &ar)
 
     all_signatures.reserve(sigs.size());
     for (auto &sig : sigs) {
+        if (sig.isOk()) {
             all_signatures.emplace_back(make_shared<CompleteSignature>(move(sig)));
         }
     }
+}
 
 class CompleteSignatureWrapper
 {

components/security_apps/ips/ips_ut/signatures_ut.cc

@@ -104,6 +104,12 @@ public:
             cereal::JSONInputArchive ar(ss);
             high_medium_confidance_signatures.load(ar);
         }
+        {
+            stringstream ss;
+            ss << "[" << signature_performance_high << ", " << signature_broken << "]";
+            cereal::JSONInputArchive ar(ss);
+            single_broken_signature.load(ar);
+        }
     }
 
     ~SignatureTest()
@@ -250,6 +256,7 @@ public:
     IPSSignaturesResource performance_signatures1;
     IPSSignaturesResource performance_signatures2;
     IPSSignaturesResource performance_signatures3;
+    IPSSignaturesResource single_broken_signature;
     NiceMock<MockTable> table;
     MockAgg mock_agg;
 
@@ -483,6 +490,26 @@ private:
                 "\"context\": [\"HTTP_REQUEST_BODY\", \"HTTP_RESPONSE_BODY\"]"
             "}"
         "}";
+
+    string signature_broken =
+    "{"
+        "\"protectionMetadata\": {"
+            "\"protectionName\": \"BrokenTest\","
+            "\"maintrainId\": \"101\","
+            "\"severity\": \"Medium High\","
+            "\"confidenceLevel\": \"Low\","
+            "\"performanceImpact\": \"High\","
+            "\"lastUpdate\": \"20210420\","
+            "\"tags\": [],"
+            "\"cveList\": []"
+        "},"
+        "\"detectionRules\": {"
+            "\"type\": \"simple\","
+            "\"SSM\": \"\","
+            "\"keywosrds\": \"data: \\\"www\\\";\","
+            "\"context\": [\"HTTP_REQUEST_BODY\", \"HTTP_RESPONSE_BODY\"]"
+        "}"
+    "}";
 };
 
 TEST_F(SignatureTest, basic_load_of_signatures)
@@ -665,3 +692,14 @@ TEST_F(SignatureTest, high_confidance_signatures_matching)
     expectLog("\"protectionId\": \"Test4\"", "\"matchedSignatureConfidence\": \"Medium\"");
     EXPECT_FALSE(checkData("mmm"));
 }
+
+TEST_F(SignatureTest, broken_signature)
+{
+    load(single_broken_signature, "Low or above", "Low");
+    EXPECT_FALSE(checkData("ggg"));
+
+    expectLog("\"matchedSignaturePerformance\": \"High\"");
+    EXPECT_TRUE(checkData("fff"));
+
+    EXPECT_FALSE(checkData("www"));
+}

components/security_apps/local_policy_mgmt_gen/CMakeLists.txt

@@ -22,4 +22,5 @@ add_library(local_policy_mgmt_gen
     access_control_practice.cc
     configmaps.cc
     reverse_proxy_section.cc
+    policy_activation_data.cc
 )

components/security_apps/local_policy_mgmt_gen/include/k8s_policy_utils.h

@@ -48,7 +48,7 @@ public:
     void init();
 
     std::tuple<std::map<std::string, AppsecLinuxPolicy>, std::map<std::string, V1beta2AppsecLinuxPolicy>>
-    createAppsecPoliciesFromIngresses();
+    createAppsecPolicies();
     void getClusterId() const;
 
 private:
@@ -101,12 +101,18 @@ private:
     ) const;
 
     template<class T, class K>
-    void createPolicy(
+    void createPolicyFromIngress(
         T &appsec_policy,
         std::map<std::string, T> &policies,
         std::map<AnnotationKeys, std::string> &annotations_values,
         const SingleIngressData &item) const;
 
+    template<class T, class K>
+    void createPolicyFromActivation(
+        T &appsec_policy,
+        std::map<std::string, T> &policies,
+        const EnabledPolicy &policy) const;
+
     std::tuple<Maybe<AppsecLinuxPolicy>, Maybe<V1beta2AppsecLinuxPolicy>> createAppsecPolicyK8s(
         const std::string &policy_name,
         const std::string &ingress_mode

components/security_apps/local_policy_mgmt_gen/include/policy_activation_data.h

@@ -0,0 +1,90 @@
+// 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 __POLICY_ACTIVATION_DATA_H__
+#define __POLICY_ACTIVATION_DATA_H__
+
+#include <vector>
+#include <map>
+
+#include "config.h"
+#include "debug.h"
+#include "rest.h"
+#include "cereal/archives/json.hpp"
+#include <cereal/types/map.hpp>
+#include "customized_cereal_map.h"
+
+#include "local_policy_common.h"
+
+class PolicyActivationMetadata
+{
+public:
+    void load(cereal::JSONInputArchive &archive_in);
+
+private:
+    std::string name;
+};
+
+class EnabledPolicy
+{
+public:
+    void load(cereal::JSONInputArchive &archive_in);
+
+    const std::string & getName() const;
+    const std::vector<std::string> & getHosts() const;
+
+private:
+    std::string name;
+    std::string mode;
+    std::vector<std::string> hosts;
+};
+
+class PolicyActivationSpec
+{
+public:
+    void load(cereal::JSONInputArchive &archive_in);
+
+    const std::vector<EnabledPolicy> & getPolicies() const;
+
+private:
+    std::string appsec_class_name;
+    std::vector<EnabledPolicy> policies;
+};
+
+class SinglePolicyActivationData
+{
+public:
+    void load(cereal::JSONInputArchive &archive_in);
+
+    const PolicyActivationSpec & getSpec() const;
+
+private:
+    std::string api_version;
+    std::string kind;
+    PolicyActivationMetadata metadata;
+    PolicyActivationSpec spec;
+};
+
+class PolicyActivationData : public ClientRest
+{
+public:
+    bool loadJson(const std::string &json);
+
+    const std::vector<SinglePolicyActivationData> & getItems() const;
+
+private:
+    std::string api_version;
+    std::vector<SinglePolicyActivationData> items;
+};
+
+#endif // __POLICY_ACTIVATION_DATA_H__

components/security_apps/local_policy_mgmt_gen/include/policy_maker_utils.h

@@ -32,6 +32,7 @@
 #include "i_messaging.h"
 #include "appsec_practice_section.h"
 #include "ingress_data.h"
+#include "policy_activation_data.h"
 #include "settings_section.h"
 #include "triggers_section.h"
 #include "local_policy_common.h"

components/security_apps/local_policy_mgmt_gen/k8s_policy_utils.cc

@@ -577,7 +577,7 @@ K8sPolicyUtils::createAppsecPolicyK8s(const string &policy_name, const string &i
 
 template<class T, class K>
 void
-K8sPolicyUtils::createPolicy(
+K8sPolicyUtils::createPolicyFromIngress(
     T &appsec_policy,
     map<std::string, T> &policies,
     map<AnnotationKeys, string> &annotations_values,
@@ -615,10 +615,35 @@ K8sPolicyUtils::createPolicy(
     }
 }
 
-std::tuple<map<string, AppsecLinuxPolicy>, map<string, V1beta2AppsecLinuxPolicy>>
+template<class T, class K>
-K8sPolicyUtils::createAppsecPoliciesFromIngresses()
+void
+K8sPolicyUtils::createPolicyFromActivation(
+    T &appsec_policy,
+    map<std::string, T> &policies,
+    const EnabledPolicy &policy) const
 {
-    dbgFlow(D_LOCAL_POLICY) << "Getting all policy object from Ingresses";
+    if (policies.find(policy.getName()) == policies.end()) {
+        policies[policy.getName()] = appsec_policy;
+    }
+    auto default_mode = appsec_policy.getAppsecPolicySpec().getDefaultRule().getMode();
+
+    for (const string &host : policy.getHosts()) {
+        if (!appsec_policy.getAppsecPolicySpec().isAssetHostExist(host)) {
+            dbgTrace(D_LOCAL_POLICY)
+                << "Inserting Host data to the specific asset set:"
+                << "URL: '"
+                << host
+                << "'";
+            K ingress_rule = K(host, default_mode);
+            policies[policy.getName()].addSpecificRule(ingress_rule);
+        }
+    }
+}
+
+std::tuple<map<string, AppsecLinuxPolicy>, map<string, V1beta2AppsecLinuxPolicy>>
+K8sPolicyUtils::createAppsecPolicies()
+{
+    dbgFlow(D_LOCAL_POLICY) << "Getting all policy object from Ingresses and PolicyActivation";
     map<string, AppsecLinuxPolicy> v1bet1_policies;
     map<string, V1beta2AppsecLinuxPolicy> v1bet2_policies;
     auto maybe_ingress = getObjectFromCluster<IngressData>("/apis/networking.k8s.io/v1/ingresses");
@@ -628,7 +653,7 @@ K8sPolicyUtils::createAppsecPoliciesFromIngresses()
         dbgWarning(D_LOCAL_POLICY)
             << "Failed to retrieve K8S Ingress configurations. Error: "
             << maybe_ingress.getErr();
-        return make_tuple(v1bet1_policies, v1bet2_policies);
+        maybe_ingress = IngressData{};
     }
 
 
@@ -658,19 +683,50 @@ K8sPolicyUtils::createAppsecPoliciesFromIngresses()
 
         if (!std::get<0>(maybe_appsec_policy).ok()) {
             auto appsec_policy=std::get<1>(maybe_appsec_policy).unpack();
-            createPolicy<V1beta2AppsecLinuxPolicy, NewParsedRule>(
+            createPolicyFromIngress<V1beta2AppsecLinuxPolicy, NewParsedRule>(
                 appsec_policy,
                 v1bet2_policies,
                 annotations_values,
                 item);
         } else {
             auto appsec_policy=std::get<0>(maybe_appsec_policy).unpack();
-            createPolicy<AppsecLinuxPolicy, ParsedRule>(
+            createPolicyFromIngress<AppsecLinuxPolicy, ParsedRule>(
                 appsec_policy,
                 v1bet1_policies,
                 annotations_values,
                 item);
         }
     }
+
+    auto maybe_policy_activation =
+        getObjectFromCluster<PolicyActivationData>("/apis/openappsec.io/v1beta2/policyactivations");
+
+    if (!maybe_policy_activation.ok()) {
+        dbgWarning(D_LOCAL_POLICY)
+            << "Failed to retrieve K8S PolicyActivation configurations. Error: "
+            << maybe_policy_activation.getErr();
+        return make_tuple(v1bet1_policies, v1bet2_policies);
+    }
+
+    PolicyActivationData policy_activation = maybe_policy_activation.unpack();
+    for (const SinglePolicyActivationData &item : policy_activation.getItems()) {
+        for (const auto &policy : item.getSpec().getPolicies()) {
+            auto maybe_appsec_policy = createAppsecPolicyK8s(policy.getName(), "");
+
+            if (!std::get<1>(maybe_appsec_policy).ok()) {
+                dbgWarning(D_LOCAL_POLICY)
+                    << "Failed to create appsec policy. v1beta2 Error: "
+                    << std::get<1>(maybe_appsec_policy).getErr();
+                continue;
+            } else {
+                auto appsec_policy=std::get<1>(maybe_appsec_policy).unpack();
+                createPolicyFromActivation<V1beta2AppsecLinuxPolicy, NewParsedRule>(
+                    appsec_policy,
+                    v1bet2_policies,
+                    policy);
+            }
+        }
+    }
+
     return make_tuple(v1bet1_policies, v1bet2_policies);
 }

components/security_apps/local_policy_mgmt_gen/local_policy_mgmt_gen.cc

@@ -36,6 +36,7 @@
 #include "customized_cereal_map.h"
 #include "include/appsec_practice_section.h"
 #include "include/ingress_data.h"
+#include "include/policy_activation_data.h"
 #include "include/settings_section.h"
 #include "include/triggers_section.h"
 #include "include/local_policy_common.h"
@@ -85,7 +86,7 @@ public:
         K8sPolicyUtils k8s_policy_utils;
         k8s_policy_utils.init();
 
-        auto appsec_policies = k8s_policy_utils.createAppsecPoliciesFromIngresses();
+        auto appsec_policies = k8s_policy_utils.createAppsecPolicies();
         if (!std::get<0>(appsec_policies).empty()) {
             return policy_maker_utils.proccesMultipleAppsecPolicies<AppsecLinuxPolicy, ParsedRule>(
                 std::get<0>(appsec_policies),

components/security_apps/local_policy_mgmt_gen/policy_activation_data.cc

@@ -0,0 +1,116 @@
+// 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 "policy_activation_data.h"
+#include "customized_cereal_map.h"
+
+using namespace std;
+
+USE_DEBUG_FLAG(D_LOCAL_POLICY);
+
+static const set<string> valid_modes = {
+    "prevent-learn",
+    "detect-learn",
+    "prevent",
+    "detect",
+    "inactive"
+};
+
+void
+PolicyActivationMetadata::load(cereal::JSONInputArchive &archive_in)
+{
+    dbgTrace(D_LOCAL_POLICY) << "PolicyActivationMetadata load";
+    parseAppsecJSONKey<string>("name", name, archive_in);
+}
+
+void
+EnabledPolicy::load(cereal::JSONInputArchive &archive_in)
+{
+    dbgTrace(D_LOCAL_POLICY) << "Loading policyActivation enabled policy";
+    parseMandatoryAppsecJSONKey<vector<string>>("hosts", hosts, archive_in);
+    parseAppsecJSONKey<string>("name", name, archive_in);
+    parseAppsecJSONKey<string>("mode", mode, archive_in, "detect");
+    if (valid_modes.count(mode) == 0) {
+        dbgWarning(D_LOCAL_POLICY) << "AppSec policy activation mode invalid: " << mode;
+        mode = "detect";
+    }
+}
+
+const string &
+EnabledPolicy::getName() const
+{
+    return name;
+}
+
+const vector<string> &
+EnabledPolicy::getHosts() const
+{
+    return hosts;
+}
+
+void
+PolicyActivationSpec::load(cereal::JSONInputArchive &archive_in)
+{
+    dbgTrace(D_LOCAL_POLICY) << "PolicyActivationSpec load";
+    parseAppsecJSONKey<string>("appsecClassName", appsec_class_name, archive_in);
+    parseMandatoryAppsecJSONKey<vector<EnabledPolicy>>("enabledPolicies", policies, archive_in);
+}
+
+const vector<EnabledPolicy> &
+PolicyActivationSpec::getPolicies() const
+{
+    return policies;
+}
+
+void
+SinglePolicyActivationData::load(cereal::JSONInputArchive &archive_in)
+{
+    dbgTrace(D_LOCAL_POLICY) << "Loading single policy activation data";
+    parseAppsecJSONKey<string>("apiVersion", api_version, archive_in);
+    parseAppsecJSONKey<string>("kind", kind, archive_in);
+    parseAppsecJSONKey<PolicyActivationMetadata>("metadata", metadata, archive_in);
+    parseAppsecJSONKey<PolicyActivationSpec>("spec", spec, archive_in);
+}
+
+const PolicyActivationSpec &
+SinglePolicyActivationData::getSpec() const
+{
+    return spec;
+}
+
+bool
+PolicyActivationData::loadJson(const string &json)
+{
+    string modified_json = json;
+    modified_json.pop_back();
+    stringstream in;
+    in.str(modified_json);
+    dbgTrace(D_LOCAL_POLICY) << "Loading policy activations data";
+    try {
+        cereal::JSONInputArchive in_ar(in);
+        in_ar(
+            cereal::make_nvp("apiVersion", api_version),
+            cereal::make_nvp("items", items)
+        );
+    } catch (cereal::Exception &e) {
+        dbgError(D_LOCAL_POLICY) << "Failed to load policy activations data JSON. Error: " << e.what();
+        return false;
+    }
+    return true;
+}
+
+const vector<SinglePolicyActivationData> &
+PolicyActivationData::getItems() const
+{
+    return items;
+}

components/security_apps/orchestration/CMakeLists.txt

@@ -14,7 +14,6 @@ add_subdirectory(details_resolver)
 add_subdirectory(health_check)
 add_subdirectory(health_check_manager)
 add_subdirectory(updates_process_reporter)
-add_subdirectory(env_details)
 add_subdirectory(external_sdk_server)
 
 #add_subdirectory(orchestration_ut)

components/security_apps/orchestration/details_resolver/details_resolver_handlers/details_resolver_impl.h

@@ -29,7 +29,7 @@
 // shell command execution output as its input
 
 #ifdef SHELL_PRE_CMD
-#if defined(gaia) || defined(smb)
+#if defined(gaia) || defined(smb) || defined(smb_thx_v3) || defined(smb_sve_v2) || defined(smb_mrv_v1)
 SHELL_PRE_CMD("read sdwan data",
     "(cpsdwan get_data > /tmp/cpsdwan_getdata_orch.json~) "
     "&& (mv /tmp/cpsdwan_getdata_orch.json~ /tmp/cpsdwan_getdata_orch.json)")
@@ -40,7 +40,7 @@ SHELL_PRE_CMD("gunzip local.cfg", "gunzip -c $FWDIR/state/local/FW1/local.cfg.gz
 #endif
 
 #ifdef SHELL_CMD_HANDLER
-#if defined(gaia) || defined(smb)
+#if defined(gaia) || defined(smb) || defined(smb_thx_v3) || defined(smb_sve_v2) || defined(smb_mrv_v1)
 SHELL_CMD_HANDLER("cpProductIntegrationMgmtObjectType", "cpprod_util CPPROD_IsMgmtMachine", getMgmtObjType)
 SHELL_CMD_HANDLER(
     "cpProductIntegrationMgmtObjectUid",
@@ -51,6 +51,14 @@ SHELL_CMD_HANDLER("prerequisitesForHorizonTelemetry",
     "FS_PATH=<FILESYSTEM-PREFIX>; [ -f ${FS_PATH}/cp-nano-horizon-telemetry-prerequisites.log ] "
     "&& head -1 ${FS_PATH}/cp-nano-horizon-telemetry-prerequisites.log || echo ''",
     checkIsInstallHorizonTelemetrySucceeded)
+SHELL_CMD_HANDLER(
+    "IS_AIOPS_RUNNING",
+    "FS_PATH=<FILESYSTEM-PREFIX>; "
+    "PID=$(ps auxf | grep -v grep | grep -E ${FS_PATH}.*cp-nano-horizon-telemetry | awk -F' ' '{printf $2}'); "
+    "[ -z \"${PID}\" ] && echo 'false' || echo 'true'",
+    getIsAiopsRunning)
+#endif
+#if defined(gaia)
 SHELL_CMD_HANDLER("GLOBAL_QUID", "[ -d /opt/CPquid ] "
     "&& python3 /opt/CPquid/Quid_Api.py -i /opt/CPotelcol/quid_api/get_global_id.json | jq -r .message || echo ''",
     getQUID)
@@ -76,12 +84,21 @@ SHELL_CMD_HANDLER("MGMT_QUID", "[ -d /opt/CPquid ] "
 SHELL_CMD_HANDLER("AIOPS_AGENT_ROLE", "[ -d /opt/CPOtlpAgent/custom_scripts ] "
     "&& ENV_NO_FORMAT=1 /opt/CPOtlpAgent/custom_scripts/agent_role.sh",
     getOtlpAgentGaiaOsRole)
-SHELL_CMD_HANDLER(
+#endif
-    "IS_AIOPS_RUNNING",
+#if defined(smb) || defined(smb_thx_v3) || defined(smb_sve_v2) || defined(smb_mrv_v1)
-    "FS_PATH=<FILESYSTEM-PREFIX>; "
+SHELL_CMD_HANDLER("GLOBAL_QUID",
-    "PID=$(ps auxf | grep -v grep | grep -E ${FS_PATH}.*cp-nano-horizon-telemetry | awk -F' ' '{printf $2}'); "
+    "cat $FWDIR/database/myown.C "
-    "[ -z \"{PID}\" ] && echo 'false' || echo 'true'",
+    "| awk -F'[()]' '/:name/ { found=1; next } found && /:uuid/ { uid=tolower($2); print uid; exit }'",
-    getIsAiopsRunning)
+    getQUID)
+SHELL_CMD_HANDLER("QUID",
+    "cat $FWDIR/database/myown.C "
+    "| awk -F'[()]' '/:name/ { found=1; next } found && /:uuid/ { uid=tolower($2); print uid; exit }'",
+    getQUID)
+SHELL_CMD_HANDLER("SMO_QUID", "echo ''", getQUID)
+SHELL_CMD_HANDLER("MGMT_QUID", "echo ''", getQUID)
+SHELL_CMD_HANDLER("AIOPS_AGENT_ROLE", "echo 'SMB'", getOtlpAgentGaiaOsRole)
+#endif
+#if defined(gaia) || defined(smb) || defined(smb_thx_v3) || defined(smb_sve_v2) || defined(smb_mrv_v1)
 SHELL_CMD_HANDLER("hasSDWan", "[ -f $FWDIR/bin/sdwan_steering ] && echo '1' || echo '0'", checkHasSDWan)
 SHELL_CMD_HANDLER(
     "canUpdateSDWanData",
@@ -194,7 +211,7 @@ SHELL_CMD_HANDLER(
 )
 #endif //gaia
 
-#if defined(smb)
+#if defined(smb) || defined(smb_thx_v3) || defined(smb_sve_v2) || defined(smb_mrv_v1)
 SHELL_CMD_HANDLER(
     "cpProductIntegrationMgmtParentObjectName",
     "jq -r .cluster_name /tmp/cpsdwan_getdata_orch.json",
@@ -252,7 +269,6 @@ SHELL_CMD_HANDLER(
 
 SHELL_CMD_OUTPUT("kernel_version", "uname -r")
 SHELL_CMD_OUTPUT("helloWorld", "cat /tmp/agentHelloWorld 2>/dev/null")
-SHELL_CMD_OUTPUT("report_timestamp", "date -u +\%s")
 #endif // SHELL_CMD_OUTPUT
 
 
@@ -282,7 +298,7 @@ FILE_CONTENT_HANDLER("AppSecModelVersion", "<FILESYSTEM-PREFIX>/conf/waap/waap.d
 #endif // FILE_CONTENT_HANDLER
 
 #ifdef SHELL_POST_CMD
-#if defined(smb)
+#if defined(smb) || defined(smb_thx_v3) || defined(smb_sve_v2) || defined(smb_mrv_v1)
 SHELL_POST_CMD("remove local.cfg", "rm -rf /tmp/local.cfg")
 #endif  //smb
 #endif

components/security_apps/orchestration/orchestration_comp.cc

@@ -1630,7 +1630,7 @@ private:
 
         string server_name = getAttribute("registered-server", "registered_server");
         auto server = TagAndEnumManagement::convertStringToTag(server_name);
-        if (server_name == "'SWAG'") server = Tags::WEB_SERVER_SWAG;
+        if (server_name == "'SWAG'" || server_name == "'SWAG Server'") server = Tags::WEB_SERVER_SWAG;
         if (server.ok()) tags.insert(*server);
 
         if (getAttribute("no-setting", "CROWDSEC_ENABLED") == "true") tags.insert(Tags::CROWDSEC);
@@ -2055,7 +2055,6 @@ private:
     OrchestrationPolicy policy;
     UpdatesProcessReporter updates_process_reporter_listener;
     HybridModeMetric hybrid_mode_metric;
-    EnvDetails env_details;
     chrono::minutes upgrade_delay_time;
 
     string filesystem_prefix = "";

components/security_apps/orchestration/orchestration_ut/orchestration_ut.cc

@@ -28,6 +28,7 @@ std::ostream & operator<<(std::ostream &os, const Package &) { return os; }
 #include "health_check_status/health_check_status.h"
 #include "updates_process_event.h"
 #include "declarative_policy_utils.h"
+#include "mock/mock_env_details.h"
 
 using namespace testing;
 using namespace std;
@@ -324,6 +325,7 @@ public:
     StrictMock<MockOrchestrationTools> mock_orchestration_tools;
     StrictMock<MockDownloader> mock_downloader;
     StrictMock<MockShellCmd> mock_shell_cmd;
+    StrictMock<EnvDetailsMocker> mock_env_details;
     StrictMock<MockMessaging> mock_message;
     StrictMock<MockRestApi> rest;
     StrictMock<MockServiceController> mock_service_controller;
@@ -583,6 +585,8 @@ TEST_F(OrchestrationTest, check_sending_registration_data)
     env.init();
     init();
 
+    EXPECT_CALL(mock_env_details, getEnvType()).WillRepeatedly(Return(EnvType::LINUX));
+
     EXPECT_CALL(mock_service_controller, updateServiceConfiguration(_, _, _, _, _, _))
         .WillOnce(Return(Maybe<void>()));
     EXPECT_CALL(mock_orchestration_tools, calculateChecksum(_, _)).WillRepeatedly(Return(string()));

components/security_apps/rate_limit/rate_limit.cc

@@ -246,6 +246,27 @@ public:
         return matched_rule;
     }
 
+    void
+    fetchReplicaCount()
+    {
+        string curl_cmd =
+            "curl -H \"Authorization: Bearer " + kubernetes_token + "\" "
+            "https://kubernetes.default.svc.cluster.local/apis/apps/v1/namespaces/" + kubernetes_namespace +
+            "/deployments/${AGENT_DEPLOYMENT_NAME} -k  -s | jq .status.replicas";
+        auto maybe_replicas = i_shell_cmd->getExecOutput(curl_cmd);
+        if (maybe_replicas.ok()) {
+            try {
+                replicas = std::stoi(maybe_replicas.unpack());
+            } catch (const std::exception &e) {
+                dbgWarning(D_RATE_LIMIT) << "error while converting replicas: " << e.what();
+            }
+        }
+        if (replicas == 0) {
+            dbgWarning(D_RATE_LIMIT) << "replicas is set to 0, setting replicas to 1";
+            replicas = 1;
+        }
+    }
+
     EventVerdict
     respond(const HttpRequestHeaderEvent &event) override
     {
@@ -271,10 +292,72 @@ public:
         dbgDebug(D_RATE_LIMIT) << "source identifier value: " << source_identifier;
 
         auto maybe_source_ip = env->get<IPAddr>(HttpTransactionData::client_ip_ctx);
+        set<string> ip_set;
         string source_ip = "";
-        if (maybe_source_ip.ok()) source_ip = ipAddrToStr(maybe_source_ip.unpack());
+        if (maybe_source_ip.ok()) {
+            source_ip = ipAddrToStr(maybe_source_ip.unpack());
 
-        unordered_map<string, set<string>> condition_map = createConditionMap(uri, source_ip, source_identifier);
+            if (getProfileAgentSettingWithDefault<bool>(false, "agent.rateLimit.ignoreSourceIP")) {
+                dbgDebug(D_RATE_LIMIT) << "Rate limit ignoring source ip: " << source_ip;
+            } else {
+                ip_set.insert(source_ip);
+            }
+        }
+
+        auto maybe_xff = env->get<string>(HttpTransactionData::xff_vals_ctx);
+        if (!maybe_xff.ok()) {
+            dbgTrace(D_RATE_LIMIT) << "Rate limit failed to get xff vals from env";
+        } else {
+            auto ips = split(maybe_xff.unpack(), ',');
+            ip_set.insert(ips.begin(), ips.end());
+        }
+
+        EnumArray<I_GeoLocation::GeoLocationField, string> geo_location_data;
+        set<string> country_codes;
+        set<string> country_names;
+        for (const string& source : ip_set) {
+            Maybe<IPAddr> maybe_source_ip = IPAddr::createIPAddr(source);
+            if (!maybe_source_ip.ok()){
+                dbgWarning(D_RATE_LIMIT)
+                    << "Rate limit failed to create ip address from source: "
+                    << source
+                    << ", Error: "
+                    << maybe_source_ip.getErr();
+                continue;
+            }
+            auto asset_location =
+                Singleton::Consume<I_GeoLocation>::by<RateLimit>()->lookupLocation(maybe_source_ip.unpack());
+            if (!asset_location.ok()) {
+                dbgWarning(D_RATE_LIMIT)
+                    << "Rate limit lookup location failed for source: "
+                    << source_ip
+                    << ", Error: "
+                    << asset_location.getErr();
+                continue;
+            }
+            geo_location_data = asset_location.unpack();
+            auto code = geo_location_data[I_GeoLocation::GeoLocationField::COUNTRY_CODE];
+            auto name =  geo_location_data[I_GeoLocation::GeoLocationField::COUNTRY_NAME];
+            country_codes.insert(code);
+            country_names.insert(name);
+            dbgTrace(D_RATE_LIMIT)
+                << "Rate limit found "
+                << "country code: "
+                << code
+                << ", country name: "
+                << name
+                << ", source ip address: "
+                << source;
+        }
+
+
+        unordered_map<string, set<string>> condition_map = createConditionMap(
+            uri,
+            source_ip,
+            source_identifier,
+            country_codes,
+            country_names
+        );
         if (shouldApplyException(condition_map)) {
             dbgDebug(D_RATE_LIMIT) << "found accept exception, not enforcing rate limit on this URI: " << uri;
             return ACCEPT;
@@ -293,11 +376,6 @@ public:
             return ACCEPT;
         }
 
-        auto replicas = getenv("REPLICA_COUNT") ? std::stoi(getenv("REPLICA_COUNT")) : 1;
-        if (replicas == 0) {
-            dbgWarning(D_RATE_LIMIT) << "REPLICA_COUNT environment variable is set to 0, setting REPLICA_COUNT to 1";
-            replicas = 1;
-        }
         burst = static_cast<float>(rule.getRateLimit()) / replicas;
         limit = static_cast<float>(calcRuleLimit(rule)) / replicas;
 
@@ -476,10 +554,18 @@ public:
     }
 
     unordered_map<string, set<string>>
-    createConditionMap(const string &uri, const string &source_ip, const string &source_identifier)
+    createConditionMap(
+        const string &uri,
+        const string &source_ip,
+        const string &source_identifier,
+        const set<string> &country_codes,
+        const set<string> &country_names
+    )
     {
         unordered_map<string, set<string>> condition_map;
         if (!source_ip.empty()) condition_map["sourceIP"].insert(source_ip);
+        if (!country_codes.empty()) condition_map["countryCode"].insert(country_codes.begin(), country_codes.end());
+        if (!country_names.empty()) condition_map["countryName"].insert(country_names.begin(), country_names.end());
         condition_map["sourceIdentifier"].insert(source_identifier);
         condition_map["url"].insert(uri);
 
@@ -616,6 +702,21 @@ public:
             "Initialize rate limit component",
             false
         );
+
+        i_shell_cmd = Singleton::Consume<I_ShellCmd>::by<RateLimit>();
+        i_env_details = Singleton::Consume<I_EnvDetails>::by<RateLimit>();
+        env_type = i_env_details->getEnvType();
+        if (env_type == EnvType::K8S) {
+            kubernetes_token = i_env_details->getToken();
+            kubernetes_namespace = i_env_details->getNameSpace();
+            fetchReplicaCount();
+            Singleton::Consume<I_MainLoop>::by<RateLimit>()->addRecurringRoutine(
+                I_MainLoop::RoutineType::Offline,
+                chrono::seconds(120),
+                [this]() { fetchReplicaCount(); },
+                "Fetch current replica count from the Kubernetes cluster"
+            );
+        }
     }
 
     void
@@ -624,6 +725,9 @@ public:
         disconnectRedis();
     }
 
+    I_ShellCmd *i_shell_cmd = nullptr;
+    I_EnvDetails* i_env_details = nullptr;
+
 private:
     static constexpr auto DROP = ngx_http_cp_verdict_e::TRAFFIC_VERDICT_DROP;
     static constexpr auto ACCEPT = ngx_http_cp_verdict_e::TRAFFIC_VERDICT_ACCEPT;
@@ -634,6 +738,10 @@ private:
     int burst;
     float limit;
     redisContext* redis = nullptr;
+    int replicas = 1;
+    EnvType env_type;
+    string kubernetes_namespace = "";
+    string kubernetes_token = "";
 };
 
 RateLimit::RateLimit() : Component("RateLimit"), pimpl(make_unique<Impl>()) {}

components/security_apps/waap/waap_clib/Serializator.cc

@@ -194,6 +194,10 @@ void SerializeToFileBase::saveData()
         dbgWarning(D_WAAP_CONFIDENCE_CALCULATOR) << "Failed to gzip data";
     } else {
         ss.str(string((const char *)res.output, res.num_output_bytes));
+        // free the memory allocated by compressData
+        if (res.output) free(res.output);
+        res.output = nullptr;
+        res.num_output_bytes = 0;
     }
     if (res.output) free(res.output);
     res.output = nullptr;

components/security_apps/waap/waap_clib/WaapScanner.cc

@@ -112,20 +112,6 @@ double Waap::Scanner::getScoreData(Waf2ScanResult& res, const std::string &poolN
     }
     double res_score = getScoreFromPool(res, newKeywords, poolName);
 
-    std::string other_pool_name = Waap::Scores::getOtherScorePoolName();
-    Waap::Scores::ModelLoggingSettings modelLoggingSettings = Waap::Scores::getModelLoggingSettings();
-
-    if (applyLearning && poolName != other_pool_name &&
-        modelLoggingSettings.logLevel != Waap::Scores::ModelLogLevel::OFF) {
-        double other_score = getScoreFromPool(res, newKeywords, other_pool_name);
-        dbgDebug(D_WAAP_SCANNER) << "Comparing score from pool " << poolName << ": " << res_score
-                                << ", vs. pool " << other_pool_name << ": " << other_score
-                                << ", score difference: " << res_score - other_score
-                                << ", sample: " << res.unescaped_line;
-        res.other_model_score = other_score;
-    } else {
-        res.other_model_score = res_score;
-    }
     return res_score;
 }
 

core/CMakeLists.txt

@@ -31,6 +31,7 @@ add_subdirectory(tenant_manager)
 add_subdirectory(compression)
 add_subdirectory(attachments)
 add_subdirectory(report_messaging)
+add_subdirectory(env_details)
 
 add_library(ngen_core SHARED ".")
 target_link_libraries(
@@ -39,7 +40,7 @@ target_link_libraries(
     "table;debug_is;shell_cmd;metric;tenant_manager;messaging;encryptor;time_proxy;singleton;mainloop;environment;logging;report;rest"
     "compression_utils;-lz;config;intelligence_is_v2;event_is;memory_consumption;connkey"
     "instance_awareness;socket_is;agent_details;agent_details_reporter;buffers;cpu;agent_core_utilities"
-    "report_messaging"
+    "report_messaging;env_details"
     -Wl,-no-whole-archive
 )
 

core/attachments/http_configuration/http_configuration.cc

@@ -111,7 +111,8 @@ HttpAttachmentConfiguration::save(cereal::JSONOutputArchive &archive) const
         cereal::make_nvp("keep_alive_interval_msec", getNumericalValue("keep_alive_interval_msec")),
         cereal::make_nvp("min_retries_for_verdict", getNumericalValue("min_retries_for_verdict")),
         cereal::make_nvp("max_retries_for_verdict", getNumericalValue("max_retries_for_verdict")),
-        cereal::make_nvp("body_size_trigger", getNumericalValue("body_size_trigger"))
+        cereal::make_nvp("body_size_trigger", getNumericalValue("body_size_trigger")),
+        cereal::make_nvp("remove_server_header", getNumericalValue("remove_server_header"))
     );
 }
 
@@ -167,6 +168,7 @@ HttpAttachmentConfiguration::load(cereal::JSONInputArchive &archive)
     loadNumericalValue(archive, "min_retries_for_verdict", 3);
     loadNumericalValue(archive, "max_retries_for_verdict", 15);
     loadNumericalValue(archive, "body_size_trigger", 200000);
+    loadNumericalValue(archive, "remove_server_header", 0);
 }
 
 bool

core/attachments/http_configuration/http_configuration_ut/http_configuration_ut.cc

@@ -63,7 +63,8 @@ TEST_F(HttpAttachmentUtilTest, GetValidAttachmentConfiguration)
             "\"waiting_for_verdict_thread_timeout_msec\": 60,\n"
             "\"req_header_thread_timeout_msec\": 10,\n"
             "\"ip_ranges\": " + createIPRangesString(ip_ranges) + ",\n"
-            "\"static_resources_path\": \"" + static_resources_path + "\""
+            "\"static_resources_path\": \"" + static_resources_path + "\",\n"
+            "\"remove_server_header\": 0"
         "}\n";
     ofstream valid_configuration_file(attachment_configuration_file_name);
     valid_configuration_file << valid_configuration;
@@ -89,6 +90,7 @@ TEST_F(HttpAttachmentUtilTest, GetValidAttachmentConfiguration)
     EXPECT_EQ(conf_data_out.getNumericalValue("res_body_thread_timeout_msec"), 80u);
     EXPECT_EQ(conf_data_out.getNumericalValue("waiting_for_verdict_thread_timeout_msec"), 60u);
     EXPECT_EQ(conf_data_out.getNumericalValue("nginx_inspection_mode"), 1u);
+    EXPECT_EQ(conf_data_out.getNumericalValue("remove_server_header"), 0u);
 }
 
 TEST_F(HttpAttachmentUtilTest, GetMalformedAttachmentConfiguration)

core/env_details/env_details.cc

@@ -15,18 +15,32 @@
 
 #include "config.h"
 #include "debug.h"
-#include "orchestration_tools.h"
+
+#include <sys/stat.h>
 
 using namespace std;
 
 USE_DEBUG_FLAG(D_LOCAL_POLICY);
 
 static const string k8s_service_account = "/var/run/secrets/kubernetes.io/serviceaccount";
-// LCOV_EXCL_START Reason: can't use on the pipline environment
+
-EnvDetails::EnvDetails() : env_type(EnvType::LINUX)
+static bool
+checkExistence(const string &path, bool is_dir)
 {
-    auto tools = Singleton::Consume<I_OrchestrationTools>::from<OrchestrationTools>();
+    try {
-    if (tools->doesFileExist("/.dockerenv")) env_type = EnvType::DOCKER;
+        struct stat info;
+        if (stat(path.c_str(), &info) != 0) return false;
+        int flag = is_dir ? S_IFDIR : S_IFREG;
+        return info.st_mode & flag;
+    } catch (exception &e) {
+        return false;
+    }
+}
+
+// LCOV_EXCL_START Reason: can't use on the pipline environment
+EnvDetails::EnvDetails() : Component("EnvDetails")
+{
+    if (doesFileExist("/.dockerenv")) env_type = EnvType::DOCKER;
     token = retrieveToken();
     agent_namespace = retrieveNamespace();
     if (!token.empty()) {
@@ -82,4 +96,10 @@ EnvDetails::readFileContent(const string &file_path)
     }
 }
 
+bool
+EnvDetails::doesFileExist(const string &file_path) const
+{
+    return checkExistence(file_path, false);
+}
+
 // LCOV_EXCL_STOP

core/include/attachments/nginx_attachment_util.h

@@ -57,6 +57,7 @@ unsigned int getResBodyThreadTimeout();
 unsigned int getMinRetriesForVerdict();
 unsigned int getMaxRetriesForVerdict();
 unsigned int getReqBodySizeTrigger();
+unsigned int getRemoveResServerHeader();
 
 unsigned int getWaitingForVerdictThreadTimeout();
 

core/include/services_sdk/interfaces/messaging/messaging_metadata.h

@@ -228,6 +228,18 @@ public:
         return sni_host_name;
     }
 
+    void
+    setDnHostName(const std::string &_dn_host_name)
+    {
+        dn_host_name = _dn_host_name;
+    }
+
+    Maybe<std::string>
+    getDnHostName() const
+    {
+        return dn_host_name;
+    }
+
     void
     setRateLimitBlock(uint block_time)
     {
@@ -273,6 +285,7 @@ public:
 private:
     std::string host_name = "";
     Maybe<std::string> sni_host_name = genError("SNI host name not set");
+    Maybe<std::string> dn_host_name = genError("DN host name not set");
     std::string ca_path = "";
     std::string client_cert_path = "";
     std::string client_key_path = "";

core/include/services_sdk/resources/component_is/components_list_impl.h

@@ -47,6 +47,7 @@
 #include "buffer.h"
 #include "intelligence_comp_v2.h"
 #include "messaging.h"
+#include "env_details.h"
 
 USE_DEBUG_FLAG(D_COMP_IS);
 
@@ -233,6 +234,7 @@ class ComponentListCore
         MemoryCalculator,
         TenantManager,
         GenericRulebase,
+        EnvDetails,
         Components...
     >
 {

core/include/services_sdk/resources/env_details.h

@@ -21,8 +21,12 @@
 #include "i_env_details.h"
 #include "singleton.h"
 #include "debug.h"
+#include "component.h"
 
-class EnvDetails : Singleton::Provide<I_EnvDetails>::SelfInterface
+class EnvDetails
+        :
+    public Component,
+    Singleton::Provide<I_EnvDetails>::SelfInterface
 {
 public:
     EnvDetails();
@@ -35,10 +39,11 @@ private:
     std::string retrieveToken();
     std::string retrieveNamespace();
     std::string readFileContent(const std::string &file_path);
+    bool doesFileExist(const std::string &file_path) const;
 
     std::string token;
     std::string agent_namespace;
-    EnvType env_type;
+    EnvType env_type = EnvType::LINUX;
 };
 
 #endif // __ENV_DETAILS_H__

core/include/services_sdk/utilities/socket_is.h

@@ -18,12 +18,14 @@
 #include "i_mainloop.h"
 #include "singleton.h"
 #include "component.h"
+#include "i_time_get.h"
 
 class SocketIS
         :
     public Component,
     Singleton::Provide<I_Socket>,
-    Singleton::Consume<I_MainLoop>
+    Singleton::Consume<I_MainLoop>,
+    Singleton::Consume<I_TimeGet>
 {
 public:
     SocketIS();

core/logging/CMakeLists.txt

@@ -1,3 +1,3 @@
-add_library(logging logging.cc log_generator.cc debug_stream.cc file_stream.cc fog_stream.cc syslog_stream.cc cef_stream.cc k8s_svc_stream.cc)
+add_library(logging logging.cc log_generator.cc debug_stream.cc file_stream.cc fog_stream.cc syslog_stream.cc cef_stream.cc k8s_svc_stream.cc log_connector.cc)
 
 add_subdirectory(logging_ut)

core/logging/cef_stream.cc

@@ -24,22 +24,21 @@ USE_DEBUG_FLAG(D_REPORT);
 static string lookup_cmd = "nslookup ";
 static string line_selection_cmd = "| grep Address | sed -n 2p";
 static string parsing_cmd = "| cut -f2 -d' ' | tr -d '\n'";
+static string CEF_NAME = "CEF";
 
 CefStream::CefStream(const string &_address, int _port, I_Socket::SocketType _protocol)
         :
-    i_socket(Singleton::Consume<I_Socket>::by<LoggingComp>()),
+    LogStreamConnector(_address, _port, _protocol, CEF_NAME)
-    address(_address),
-    port(_port),
-    protocol(_protocol)
 {
-    connect();
+    init();
-    if (!socket.ok()) {
+    socket = genError("Not set yet");
-        dbgWarning(D_REPORT) << "Failed to connect to the CEF server";
-    }
 }
 
 CefStream::~CefStream()
 {
+    sendAllLogs();
+    if (mainloop != nullptr && mainloop->doesRoutineExist(connecting_routine)) mainloop->stop(connecting_routine);
+
     if (socket.ok()) {
         i_socket->closeSocket(const_cast<int &>(*socket));
         socket = genError("Closed socket");
@@ -49,50 +48,60 @@ CefStream::~CefStream()
 void
 CefStream::sendLog(const Report &log)
 {
-    if (!socket.ok()) {
-        connect();
-        if (!socket.ok()) {
-            dbgWarning(D_REPORT) << "Failed to connect to the CEF server, log will not be sent.";
-            return;
-        }
-    }
-    dbgTrace(D_REPORT) << "Connected to socket.";
     string cef_report = log.getCef();
     if (protocol == I_Socket::SocketType::TCP) {
         cef_report = to_string(cef_report.length()) + " " + cef_report;
     }
     vector<char> data(cef_report.begin(), cef_report.end());
-    for (size_t tries = 0; tries < 3; tries++) {
+    sendLogWithQueue(data);
-        if (i_socket->writeData(socket.unpack(), data)) {
-            dbgTrace(D_REPORT) << "log was sent to CEF server";
-            return;
-        } else {
-            dbgWarning(D_REPORT) << "Failed to send log to CEF server";
-        }
 }
+
+void
+CefStream::init()  {
+    updateSettings();
+    maintainConnection();
+
+    auto ceflog_retry_interval = getProfileAgentSettingWithDefault<uint>(
+        RETRY_CONNECT_INTERVAL,
+        "agent.config.log.cefServer.connect_retry_interval");
+    dbgTrace(D_REPORT) << "retry interval: " << ceflog_retry_interval;
+    chrono::seconds connect_retry_interval = chrono::seconds(ceflog_retry_interval);
+    connecting_routine = mainloop->addRecurringRoutine(
+        I_MainLoop::RoutineType::Offline,
+        connect_retry_interval,
+        [this] ()
+        {
+            dbgTrace(D_REPORT) << CEF_CONNECT_NAME;
+            maintainConnection();
+        },
+        CEF_CONNECT_NAME
+    );
 }
 
 void
 CefStream::connect()
 {
-    auto cef_address = getProfileAgentSettingWithDefault<string>(address, "agent.config.log.cefServer.IP");
+    dbgDebug(D_REPORT)
-    auto cef_port = getProfileAgentSettingWithDefault<uint>(port, "agent.config.log.cefServer.port");
+        << "Connecting to CEF server"
-
+        << " Address: "
-    if (cef_address.empty()) {
+        << address
+        << " Port: "
+        << port;
+    if (address.empty()) {
         dbgWarning(D_REPORT) << "Cannot connect to CEF server, IP/Domain is not configured.";
         return;
     }
 
     struct in_addr addr;
-    if (inet_pton(AF_INET, cef_address.data(), &addr) != 1) {
+    if (inet_pton(AF_INET, address.data(), &addr) != 1) {
         I_ShellCmd *shell_cmd = Singleton::Consume<I_ShellCmd>::by<LoggingComp>();
-        string host_cmd = lookup_cmd + cef_address + line_selection_cmd + parsing_cmd;
+        string host_cmd = lookup_cmd + address + line_selection_cmd + parsing_cmd;
         Maybe<string> res = shell_cmd->getExecOutput(host_cmd, 500);
         if (!res.ok()) {
             dbgWarning(D_REPORT)
                 << "Failed to execute domain lookup command. "
                 << "CEF Domain: "
-                << cef_address
+                << address
                 << "Error: "
                 << res.getErr();
             return;
@@ -102,7 +111,7 @@ CefStream::connect()
             dbgWarning(D_REPORT)
                 << "Got en empty ip address from lookup command. "
                 << "CEF Domain: "
-                << cef_address
+                << address
                 << "Got bad ip address: "
                 << res.unpack();
             return;
@@ -113,19 +122,47 @@ CefStream::connect()
             dbgWarning(D_REPORT)
                 << "Got a faulty ip address from lookup command. "
                 << "CEF Domain: "
-                << cef_address
+                << address
                 << "Got bad ip address: "
                 << res.unpack();
             return;
         }
         
-        cef_address = res.unpack();
+        address = res.unpack();
     }
 
     socket = i_socket->genSocket(
         protocol,
         false,
         false,
-        cef_address + ":" + to_string(cef_port)
+        address + ":" + to_string(port)
     );
 }
+
+void
+CefStream::updateSettings()
+{
+    max_logs_per_send = getProfileAgentSettingWithDefault<int>(
+        NUMBER_OF_LOGS_PER_SEND,
+        "agent.config.log.cefServer.MaxLogsPerSend"
+    );
+    if (max_logs_per_send < 0) {
+        max_logs_per_send = NUMBER_OF_LOGS_PER_SEND;
+    }
+    address = getProfileAgentSettingWithDefault<string>(address, "agent.config.log.cefServer.IP");
+    port = getProfileAgentSettingWithDefault<uint>(port, "agent.config.log.cefServer.port");
+    max_data_in_queue = getProfileAgentSettingWithDefault<uint>(
+        MAX_LOG_QUEUE,
+        "agent.config.log.cefServer.MaxDataInQueue"
+    );
+    dbgTrace(D_REPORT)
+        << "CEF server settings updated. "
+        << "Address: "
+        << address
+        << " Port: "
+        << port
+        << " Max logs per send: "
+        << max_logs_per_send
+        << " Max data in queue: "
+        << max_data_in_queue;
+}

core/logging/log_connector.cc

@@ -0,0 +1,131 @@
+// 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 "log_streams.h"
+
+void
+LogStreamConnector::maintainConnection()
+{
+    dbgTrace(D_REPORT)
+        << "Check if the connection is alive:"
+        << (socket.ok() ? " socket ok" : " socket not ok")
+        << (did_write_fail_in_this_window ? " previous write failed" : " previous write succeeded");
+    if (!socket.ok() || did_write_fail_in_this_window) {
+        dbgTrace(D_REPORT)
+            << (socket.ok() ? "" : "The current socket is not ok, trying to connect.");
+        connect();
+        did_write_fail_in_this_window = false;
+        if (!socket.ok()) {
+            dbgWarning(D_REPORT) << "Failed to connect to the server, logs will not be sent";
+            return;
+        }
+    }
+}
+
+void
+LogStreamConnector::addLogToQueue(const std::vector<char> &data)
+{
+    if (logs_in_queue.size() < max_data_in_queue) {
+        dbgTrace(D_REPORT)
+            << "Adding log to queue, Amount of logs in queue: "
+            << logs_in_queue.size();
+        logs_in_queue.push_back(data);
+    } else {
+        dbgWarning(D_REPORT) << "Queue is full, dropping log";
+    }
+}
+
+void
+LogStreamConnector::writeFail()
+{
+    if (!socket.ok()) {
+        dbgTrace(D_REPORT) << "Socket is not ok, stopping the connect after write failure";
+        return;
+    }
+    dbgTrace(D_REPORT) << (did_write_fail_in_this_window ? "Previous write failed" : "Previous write succeeded");
+    if (!did_write_fail_in_this_window) {
+        dbgTrace(D_REPORT)
+            << "First time in window that write failed, trying to reconnect to server";
+        connect();
+    }
+    did_write_fail_in_this_window = true;
+}
+
+bool
+LogStreamConnector::basicWriteLog(const std::vector<char> &data)
+{
+    for (size_t tries = 0; tries < 3; tries++) {
+        if (socket.ok() && i_socket->writeData(socket.unpack(), data)) {
+            dbgTrace(D_REPORT) << "log was sent to server";
+            return true;
+        } else {
+            dbgTrace(D_REPORT) << "Failed to send log to server";
+            writeFail();
+        }
+    }
+    return false;
+}
+
+void
+LogStreamConnector::sendLogWithQueue(const std::vector<char> &data)
+{
+    if (!socket.ok()) {
+        dbgTrace(D_REPORT)
+            << "Socket not ok. Size of logs in queue: "
+            << logs_in_queue.size()
+            << ". Adding logs to the queue until the connection is established.";
+        addLogToQueue(data);
+        return;
+    }
+
+    if (logs_in_queue.empty() && basicWriteLog(data)) return;
+
+    addLogToQueue(data);
+
+    int write_iterations = 0;
+    
+    while (write_iterations < max_logs_per_send && !logs_in_queue.empty()) {
+        dbgTrace(D_REPORT)
+            << " Iteration: "
+            << write_iterations
+            << " to try and write a log from queue to server"
+            << log_name;
+        int i = 0;
+        bool write_success = false;
+        while (
+            socket.ok() &&
+            (i < 3) &&
+            !(write_success = i_socket->writeData(socket.unpack(), logs_in_queue.front()))) {
+            i++;
+        }
+        if (write_success) {
+            dbgTrace(D_REPORT) << "log was written to " << log_name << " server";
+            logs_in_queue.erase(logs_in_queue.begin());
+            write_iterations++;
+        } else {
+            dbgTrace(D_REPORT) << "Failed to send log to " << log_name << " server";
+            writeFail();
+            return;
+        }
+    }
+}
+
+void
+LogStreamConnector::sendAllLogs()
+{
+    dbgTrace(D_REPORT) << "Sending all logs from queue to server";
+    for(auto &log : logs_in_queue) {
+        basicWriteLog(log);
+    }
+    logs_in_queue.clear();
+}

core/logging/log_streams.h

@@ -23,6 +23,13 @@
 #include "logging_metric.h"
 #include "i_logging.h"
 #include "i_socket_is.h"
+#include "logging_comp.h"
+
+static const int RETRY_CONNECT_INTERVAL = 120;
+static const std::string SYSLOG_CONNECT_NAME = "connecting to Syslog server";
+static const std::string CEF_CONNECT_NAME = "connecting to CEF server";
+static const int NUMBER_OF_LOGS_PER_SEND = 15;
+static size_t MAX_LOG_QUEUE = 1000;
 
 USE_DEBUG_FLAG(D_REPORT);
 
@@ -93,43 +100,77 @@ private:
     I_Messaging *i_msg = nullptr;
 };
 
-class SyslogStream : public Stream
+class LogStreamConnector : public Stream
+{
+public:
+    LogStreamConnector(
+        const std::string &_address,
+        int _port,
+        I_Socket::SocketType _protocol,
+        const std::string &_log_name) :
+        mainloop(Singleton::Consume<I_MainLoop>::by<LoggingComp>()),
+        i_socket(Singleton::Consume<I_Socket>::by<LoggingComp>()),
+        address(_address),
+        port(_port),
+        protocol(_protocol),
+        logs_in_queue(),
+        log_name(_log_name) {}
+    virtual ~LogStreamConnector() {}
+
+protected:
+    virtual void connect() = 0;
+    virtual void updateSettings() = 0;
+
+    void maintainConnection();
+    void addLogToQueue(const std::vector<char> &data);
+    void writeFail();
+    bool basicWriteLog(const std::vector<char> &data);
+    void sendLogWithQueue(const std::vector<char> &data);
+    void sendAllLogs();
+
+    I_MainLoop *mainloop = nullptr;
+    I_Socket *i_socket = nullptr;
+    std::string address;
+    int port;
+    I_Socket::SocketType protocol = I_Socket::SocketType::UDP;
+    Maybe<I_Socket::socketFd> socket = genError("Not set yet");
+    bool did_write_fail_in_this_window = false;
+    std::vector<std::vector<char>> logs_in_queue;
+    I_MainLoop::RoutineID connecting_routine = -1;
+    int max_logs_per_send = NUMBER_OF_LOGS_PER_SEND;
+    std::string log_name;
+    uint max_data_in_queue = MAX_LOG_QUEUE;
+};
+
+class SyslogStream : public LogStreamConnector
 {
 public:
     SyslogStream(const std::string &_address, int _port, I_Socket::SocketType protocol);
     ~SyslogStream();
-
     void sendLog(const Report &log) override;
 
-private:
+protected:
-    void sendLog(const std::vector<char> &data);
+    void connect() override;
-    void connect();
+    void updateSettings() override;
 
-    I_Socket *i_socket = nullptr;
+private:
-    I_MainLoop *mainloop = nullptr;
+    void init();
-    std::string address;
+    void sendLog(const std::vector<char> &data);
-    int port;
-    I_Socket::SocketType protocol = I_Socket::SocketType::UDP;
     I_MainLoop::RoutineID log_send_routine = -1;
-    Maybe<I_Socket::socketFd> socket = genError("Not set yet");
 };
 
-class CefStream : public Stream
+class CefStream : public LogStreamConnector
 {
 public:
     CefStream(const std::string &_address, int _port, I_Socket::SocketType _protocol);
     ~CefStream();
-
     void sendLog(const Report &log) override;
 
+protected:
+    void connect() override;
+    void updateSettings() override;
 private:
-    void connect();
+    void init();
-
-    I_Socket *i_socket = nullptr;
-    std::string address;
-    int port;
-    I_Socket::SocketType protocol = I_Socket::SocketType::UDP;
-    Maybe<I_Socket::socketFd> socket = genError("Not set yet");
 };
 
 #endif // __LOG_STREAMS_H__

core/logging/logging.cc

@@ -171,6 +171,7 @@ public:
         } else {
             LogEventLogsSent(true).notify();
             for (auto &iter : streams) {
+                dbgTrace(D_REPORT) << "Sending log to stream: " << TagAndEnumManagement::convertToString(iter.first);
                 if (log.isStreamActive(iter.first)) iter.second->sendLog(log);
             }
         }

core/logging/logging_ut/logging_ut.cc

@@ -24,6 +24,7 @@
 #include "metric/all_metric_event.h"
 #include "mock/mock_shell_cmd.h"
 #include "version.h"
+#include "../log_streams.h"
 
 using namespace testing;
 using namespace std;
@@ -104,6 +105,7 @@ public:
 class LogTest : public testing::Test
 {
 public:
+    const ::Report CreateReport(ReportIS::Tags &tag1, ReportIS::Tags &tag2);
     LogTest()
             :
         agent_details(),
@@ -132,6 +134,16 @@ public:
             DoAll(SaveArg<1>(&sysog_routine), Return(0))
         );
 
+        EXPECT_CALL(
+                mock_mainloop,
+                addRecurringRoutine(_, _, _, "connecting to Syslog server", _)
+            ).WillRepeatedly(DoAll(SaveArg<2>(&connect_syslog_routine), Return(2)));
+
+        EXPECT_CALL(
+                mock_mainloop,
+                addRecurringRoutine(_, _, _, "connecting to CEF server", _)
+            ).WillRepeatedly(DoAll(SaveArg<2>(&connect_cef_routine), Return(3)));
+
         EXPECT_CALL(mock_socket_is, writeData(1, _)).WillRepeatedly(
             WithArg<1>(
                 Invoke(
@@ -291,6 +303,8 @@ public:
     ConfigComponent           config;
     vector<string>            capture_syslog_cef_data;
     I_MainLoop::Routine       sysog_routine = nullptr;
+    I_MainLoop::Routine       connect_syslog_routine = nullptr;
+    I_MainLoop::Routine       connect_cef_routine = nullptr;
     StrictMock<MockShellCmd>  mock_shell_cmd;
     bool                      is_domain;
 
@@ -1469,98 +1483,148 @@ TEST_F(LogTest, BulkModification)
     EXPECT_EQ(local_body, str1);
 }
 
-TEST_F(LogTest, ObfuscationTest)
+const ::Report
+LogTest::CreateReport(Tags &tag1, Tags &tag2) {
+    LogField origin("String", "Another string");
+
+    const ::Report report(
+        "String=\"Another string\"",
+        chrono::microseconds(90000),
+        Type::EVENT,
+        Level::LOG,
+        LogLevel::INFO,
+        Audience::INTERNAL,
+        AudienceTeam::AGENT_CORE,
+        Severity::INFO,
+        Priority::LOW,
+        chrono::seconds(3600),
+        origin,
+        tag1,
+        tag2,
+        Notification::POLICY_UPDATE,
+        IssuingEngine::AGENT_CORE
+    );
+    return report;
+}
+
+TEST_F(LogTest, ObfuscationCefSysLogTest)
 {
     loadFakeConfiguration(false);
     Tags tag1 = Tags::POLICY_INSTALLATION;
     Tags tag2 = Tags::ACCESS_CONTROL;
+    std::string address = "172.28.1.6";
+    int port = 514;
+    I_Socket::SocketType protocol = I_Socket::SocketType::TCP;
+    // for cef
+    CefStream cef_stream(address, port, protocol);
+    ASSERT_NE(connect_cef_routine, nullptr);
+    connect_cef_routine();
+    cef_stream.sendLog(CreateReport(tag1, tag2));
+    EXPECT_EQ(capture_syslog_cef_data.size(), 1u);
+    // for syslog activate send log
+    SyslogStream syslog_stream(address, port, protocol);
 
-    static const string expected_obfuscated_log(
+    // connection to socket before send log
-        "{\n"
+    ASSERT_NE(connect_syslog_routine, nullptr);
-        "    \"log\": {\n"
+    connect_syslog_routine();
-        "        \"eventTime\": \"0:0:0\",\n"
-        "        \"eventName\": \"Install policy\",\n"
-        "        \"eventSeverity\": \"Info\",\n"
-        "        \"eventPriority\": \"Low\",\n"
-        "        \"eventType\": \"Event Driven\",\n"
-        "        \"eventLevel\": \"Log\",\n"
-        "        \"eventLogLevel\": \"info\",\n"
-        "        \"eventAudience\": \"Internal\",\n"
-        "        \"eventAudienceTeam\": \"\",\n"
-        "        \"eventFrequency\": 0,\n"
-        "        \"eventTags\": [\n"
-        "            \"Access Control\",\n"
-        "            \"Policy Installation\"\n"
-        "        ],\n"
-        "        \"eventSource\": {\n"
-        "            \"agentId\": \"Unknown\",\n"
-        "            \"eventTraceId\": \"\",\n"
-        "            \"eventSpanId\": \"\",\n"
-        "            \"issuingEngineVersion\": \"\",\n"
-        "            \"serviceName\": \"Unnamed Nano Service\"\n"
-        "        },\n"
-        "        \"eventData\": {\n"
-        "            \"logIndex\": 1,\n"
-        "            \"String\": \"{XORANDB64}:mocked field\"\n"
-        "        }\n"
-        "    }\n"
-        "}"
-    );
-    StrictMock<MockEncryptor> mock_encrypt;
-    EXPECT_CALL(mock_encrypt, base64Encode(_)).WillOnce(Return("mocked field"));
     
-    static const string expected_clear_log(
+    syslog_stream.sendLog(CreateReport(tag1, tag2)); // send log in routine sysog_routine
-        "{\n"
-        "    \"eventTime\": \"0:0:0\",\n"
-        "    \"eventName\": \"Install policy\",\n"
-        "    \"eventSeverity\": \"Info\",\n"
-        "    \"eventPriority\": \"Low\",\n"
-        "    \"eventType\": \"Event Driven\",\n"
-        "    \"eventLevel\": \"Log\",\n"
-        "    \"eventLogLevel\": \"info\",\n"
-        "    \"eventAudience\": \"Internal\",\n"
-        "    \"eventAudienceTeam\": \"\",\n"
-        "    \"eventFrequency\": 0,\n"
-        "    \"eventTags\": [\n"
-        "        \"Access Control\",\n"
-        "        \"Policy Installation\"\n"
-        "    ],\n"
-        "    \"eventSource\": {\n"
-        "        \"agentId\": \"Unknown\",\n"
-        "        \"eventTraceId\": \"\",\n"
-        "        \"eventSpanId\": \"\",\n"
-        "        \"issuingEngineVersion\": \"\",\n"
-        "        \"serviceName\": \"Unnamed Nano Service\"\n"
-        "    },\n"
-        "    \"eventData\": {\n"
-        "        \"logIndex\": 1,\n"
-        "        \"String\": \"Another string\"\n"
-        "    }\n"
-        "}"
-    );
-
-    {
-        LogGen log(
-            "Install policy",
-            Audience::INTERNAL,
-            Severity::INFO,
-            Priority::LOW,
-            tag1,
-            tag2,
-            Enreachments::BEAUTIFY_OUTPUT
-        );
-        log << LogField("String", "Another string", LogFieldOption::XORANDB64);
-        EXPECT_EQ(toJson(log), expected_clear_log);
-    }
-
-    EXPECT_THAT(getMessages(), HasSubstr(expected_clear_log));
-    EXPECT_THAT(readLogFile(), HasSubstr(expected_clear_log));
-    EXPECT_EQ(getBodyFogMessage(), expected_obfuscated_log);
     ASSERT_NE(sysog_routine, nullptr);
     sysog_routine();
-    EXPECT_EQ(capture_syslog_cef_data.size(), 2u);
+
+    EXPECT_EQ(capture_syslog_cef_data.size(), 2u); // 1 for CEF 1 for Syslog
     for (const string &str : capture_syslog_cef_data) {
-        EXPECT_THAT(str, AnyOf(HasSubstr("String='Another string'"), HasSubstr("String=\"Another string\"")));
+        EXPECT_THAT(str, AnyOf(
+            HasSubstr("String='Another string'"),
+            HasSubstr(R"(String="Another string")"),
+            HasSubstr("String=\"Another string\"")));
+    }
+}
+
+TEST_F(LogTest, SysLogWriteFailTest)
+{
+    loadFakeConfiguration(false);
+    capture_syslog_cef_data.clear();
+    Tags tag1 = Tags::POLICY_INSTALLATION;
+    Tags tag2 = Tags::ACCESS_CONTROL;
+
+    // for syslog activate send log
+    std::string address = "172.28.1.6";
+    int port = 514;
+    I_Socket::SocketType protocol = I_Socket::SocketType::TCP;
+    SyslogStream syslog_stream(address, port, protocol);
+
+    ASSERT_NE(connect_syslog_routine, nullptr);
+    connect_syslog_routine();
+
+    EXPECT_CALL(mock_socket_is, writeData(1, _))
+        .WillOnce(Return(false))
+        .WillOnce(Return(false))
+        .WillOnce(Return(false))
+        .WillRepeatedly(
+            WithArg<1>(
+                Invoke(
+                    [this](const vector<char> &data)
+                    {
+                        capture_syslog_cef_data.emplace_back(data.begin(), data.end());
+                        return true;
+                    }
+                )
+            )
+        );
+
+    syslog_stream.sendLog(CreateReport(tag1, tag2));
+    ASSERT_NE(sysog_routine, nullptr);
+    EXPECT_EQ(capture_syslog_cef_data.size(), 0u); //before write
+    sysog_routine();
+    EXPECT_EQ(capture_syslog_cef_data.size(), 1u);
+    for (const string &str : capture_syslog_cef_data) {
+        EXPECT_THAT(str, AnyOf(
+            HasSubstr("String='Another string'"),
+            HasSubstr(R"(String="Another string")"),
+            HasSubstr("String=\"Another string\"")));
+    }
+}
+
+TEST_F(LogTest, CefWriteFailTest)
+{
+    loadFakeConfiguration(false);
+    capture_syslog_cef_data.clear();
+    Tags tag1 = Tags::POLICY_INSTALLATION;
+    Tags tag2 = Tags::ACCESS_CONTROL;
+
+    // for syslog activate send log
+    std::string address = "172.28.1.6";
+    int port = 514;
+    I_Socket::SocketType protocol = I_Socket::SocketType::TCP;
+    CefStream cef_stream(address, port, protocol);
+
+    ASSERT_NE(connect_cef_routine, nullptr);
+    connect_cef_routine();
+
+    EXPECT_CALL(mock_socket_is, writeData(1, _))
+        .WillOnce(Return(false))
+        .WillOnce(Return(false))
+        .WillOnce(Return(false))
+        .WillRepeatedly(
+            WithArg<1>(
+                Invoke(
+                    [this](const vector<char> &data)
+                    {
+                        capture_syslog_cef_data.emplace_back(data.begin(), data.end());
+                        return true;
+                    }
+                )
+            )
+        );
+    EXPECT_EQ(capture_syslog_cef_data.size(), 0u); //before write
+    cef_stream.sendLog(CreateReport(tag1, tag2));
+    EXPECT_EQ(capture_syslog_cef_data.size(), 1u);
+    for (const string &str : capture_syslog_cef_data) {
+        EXPECT_THAT(str, AnyOf(
+            HasSubstr("String='Another string'"),
+            HasSubstr(R"(String="Another string")"),
+            HasSubstr("String=\"Another string\"")));
     }
 }
 

core/logging/syslog_stream.cc

@@ -23,24 +23,22 @@ USE_DEBUG_FLAG(D_REPORT);
 static string lookup_cmd = "nslookup ";
 static string line_selection_cmd = "| grep Address | sed -n 2p";
 static string parsing_cmd = "| cut -f2 -d' ' | tr -d '\n'";
+static string SYSLOG_NAME = "Syslog";
 
 SyslogStream::SyslogStream(const string &_address, int _port, I_Socket::SocketType _protocol)
         :
-    i_socket(Singleton::Consume<I_Socket>::by<LoggingComp>()),
+    LogStreamConnector(_address, _port, _protocol, SYSLOG_NAME)
-    mainloop(Singleton::Consume<I_MainLoop>::by<LoggingComp>()),
-    address(_address),
-    port(_port),
-    protocol(_protocol)
 {
-    connect();
+    socket = genError("Not set yet");
-    if (!socket.ok()) {
+    init();
-        dbgWarning(D_REPORT) << "Failed to connect to the syslog server";
-    }
 }
 
 SyslogStream::~SyslogStream()
 {
+    sendAllLogs();
     if (mainloop != nullptr && mainloop->doesRoutineExist(log_send_routine)) mainloop->stop(log_send_routine);
+    if (mainloop != nullptr && mainloop->doesRoutineExist(connecting_routine)) mainloop->stop(connecting_routine);
+
     if (socket.ok()) {
         i_socket->closeSocket(const_cast<int &>(*socket));
         socket = genError("Closed socket");
@@ -55,7 +53,7 @@ SyslogStream::sendLog(const Report &log)
         syslog_report = to_string(syslog_report.length()) + " " + syslog_report;
     }
     vector<char> data(syslog_report.begin(), syslog_report.end());
-    mainloop->addOneTimeRoutine(
+    log_send_routine = mainloop->addOneTimeRoutine(
         I_MainLoop::RoutineType::Offline,
         [this, data] () { sendLog(data); },
         "Logging Syslog stream messaging"
@@ -65,45 +63,57 @@ SyslogStream::sendLog(const Report &log)
 void
 SyslogStream::sendLog(const vector<char> &data)
 {
-    for (int tries = 0; tries < 3; ++tries) {
+    dbgTrace(D_REPORT) << "Sending Syslog log." << " Max logs per send: " << max_logs_per_send;
-        if (!socket.ok()) {
+    sendLogWithQueue(data);
-            connect();
-            if (!socket.ok()) {
-                dbgWarning(D_REPORT) << "Failed to connect to the syslog server, Log will not be sent.";
-                return;
-            }
-            dbgTrace(D_REPORT) << "Successfully connect to the syslog server";
 }
 
-        if (i_socket->writeData(socket.unpack(), data)) {
+
-            dbgTrace(D_REPORT) << "log was sent to syslog server";
+void
-            return;
+SyslogStream::init() {
-        }
+    updateSettings();
-    }
+    maintainConnection();
-    dbgWarning(D_REPORT) << "Failed to send log to syslog server";
+
+    auto syslog_retry_interval = getProfileAgentSettingWithDefault<uint>(
+        RETRY_CONNECT_INTERVAL,
+        "agent.config.log.syslogServer.connect_retry_interval");
+    chrono::seconds connect_retry_interval = chrono::seconds(syslog_retry_interval);
+    connecting_routine = mainloop->addRecurringRoutine(
+        I_MainLoop::RoutineType::Offline,
+        connect_retry_interval,
+        [this] ()
+        {
+            dbgTrace(D_REPORT) << SYSLOG_CONNECT_NAME;
+            maintainConnection();
+        },
+        SYSLOG_CONNECT_NAME
+    );
 }
 
 void
 SyslogStream::connect()
 {
-    auto syslog_address = getProfileAgentSettingWithDefault<string>(address, "agent.config.log.syslogServer.IP");
+    dbgDebug(D_REPORT)
-    auto syslog_port = getProfileAgentSettingWithDefault<uint>(port, "agent.config.log.syslogServer.port");
+        << "Connecting to Syslog server"
+        << " Address: "
+        << address
+        << " Port: "
+        << port;
 
-    if (syslog_address.empty()) {
+    if (address.empty()) {
         dbgWarning(D_REPORT) << "Cannot connect to Syslog server, Address IP/Domain not configured.";
         return;
     }
 
     struct in_addr addr;
-    if (inet_pton(AF_INET, syslog_address.data(), &addr) != 1) {
+    if (inet_pton(AF_INET, address.data(), &addr) != 1) {
         I_ShellCmd *shell_cmd = Singleton::Consume<I_ShellCmd>::by<LoggingComp>();
-        string host_cmd = lookup_cmd + syslog_address + line_selection_cmd + parsing_cmd;
+        string host_cmd = lookup_cmd + address + line_selection_cmd + parsing_cmd;
         Maybe<string> res = shell_cmd->getExecOutput(host_cmd, 500);
         if (!res.ok()) {
             dbgWarning(D_REPORT)
                 << "Failed to execute domain lookup command. "
                 << "SYSLOG Domain: "
-                << syslog_address
+                << address
                 << "Error: "
                 << res.getErr();
             return;
@@ -113,7 +123,7 @@ SyslogStream::connect()
             dbgWarning(D_REPORT)
                 << "Got en empty ip address from lookup command. "
                 << "SYSLOG Domain: "
-                << syslog_address
+                << address
                 << "Got bad ip address: "
                 << res.unpack();
             return;
@@ -124,19 +134,46 @@ SyslogStream::connect()
             dbgWarning(D_REPORT)
                 << "Got a faulty ip address from lookup command. "
                 << "SYSLOG Domain: "
-                << syslog_address
+                << address
                 << "Got bad ip address: "
                 << res.unpack();
             return;
         }
 
-        syslog_address = res.unpack();
+        address = res.unpack();
     }
 
     socket = i_socket->genSocket(
         protocol,
         false,
         false,
-        syslog_address + ":" + to_string(syslog_port)
+        address + ":" + to_string(port)
     );
 }
+
+void
+SyslogStream::updateSettings()
+{
+    max_logs_per_send = getProfileAgentSettingWithDefault<int>(
+        NUMBER_OF_LOGS_PER_SEND,
+        "agent.config.log.syslogServer.MaxLogsPerSend"
+    );
+    if (max_logs_per_send < 0) {
+        max_logs_per_send = NUMBER_OF_LOGS_PER_SEND;
+    }
+    address = getProfileAgentSettingWithDefault<string>(address, "agent.config.log.syslogServer.IP");
+    port = getProfileAgentSettingWithDefault<uint>(port, "agent.config.log.syslogServer.port");
+    max_data_in_queue =
+        getProfileAgentSettingWithDefault<uint>(MAX_LOG_QUEUE, "agent.config.log.syslogServer.MaxLogQueue");
+
+    dbgTrace(D_REPORT)
+        << "Syslog server settings updated. "
+        << "Address: "
+        << address
+        << " Port: "
+        << port
+        << " Max logs per send: "
+        << max_logs_per_send
+        << " Max data in queue: "
+        << max_data_in_queue;
+}

core/messaging/connection/connection.cc

@@ -98,6 +98,8 @@ public:
         }
 
         sni_hostname = metadata.getSniHostName();
+        dn_host_name = metadata.getDnHostName();
+
     }
 
     void
@@ -328,17 +330,24 @@ private:
         SSL_set_hostflags(ssl_socket, X509_CHECK_FLAG_NO_PARTIAL_WILDCARDS);
 
         auto host = key.getHostName().c_str();
-        if (!SSL_set1_host(ssl_socket, host)) {
+        auto dn = host;
-            return genError("Failed to set host name verification. Host: " + string(host));
+        auto sni = host;
+
+        if (dn_host_name.ok()) {
+            dn = dn_host_name->c_str();
         }
 
+        dbgDebug(D_CONNECTION) << "Setting host DN: " << dn;
+        if (!SSL_set1_host(ssl_socket, dn)) {
+            return genError("Failed to set host name verification. Host: " + string(dn));
+        }
         if (sni_hostname.ok()) {
-            host = sni_hostname->c_str();
+            sni = sni_hostname->c_str();
         }
 
-        dbgDebug(D_CONNECTION) << "Setting TLS host name extension. Host: " << host;
+        dbgDebug(D_CONNECTION) << "Setting TLS host name extension. Host: " << sni;
-        if (!SSL_set_tlsext_host_name(ssl_socket, host)) {
+        if (!SSL_set_tlsext_host_name(ssl_socket, sni)) {
-            return genError("Failed to set TLS host name extension. Host: " + string(host));
+            return genError("Failed to set TLS host name extension. Host: " + string(sni));
         }
 
         return Maybe<void>();
@@ -698,6 +707,8 @@ private:
     bool should_close_connection = false;
     bool is_dual_auth = false;
     Maybe<string> sni_hostname = genError<string>("Uninitialized");
+    Maybe<string> dn_host_name = genError<string>("Uninitialized");
+
 };
 
 Connection::Connection(const MessageConnectionKey &key, const MessageMetadata &metadata)
@@ -716,7 +727,7 @@ Connection::setProxySettings(const MessageProxySettings &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());
+        headers["Proxy-Authorization"] = "Basic " + i_encrypt->base64Encode(settings.getProxyAuth());
     }
 
     auto req = HTTPRequest::prepareRequest(*this, HTTPMethod::CONNECT, "", headers, "");

core/messaging/messaging_buffer_comp/messaging_buffer_comp.cc

@@ -30,11 +30,15 @@ using namespace std;
 USE_DEBUG_FLAG(D_MESSAGING_BUFFER);
 
 #ifndef smb
-static constexpr uint buffer_max_size_MB = 100;
+static constexpr uint buffer_max_size_MB = 300;
 #else
 static constexpr uint buffer_max_size_MB = 3;
 #endif
 
+static const uint reservation_default_size = 32;
+static const uint memory_messages_max_size_default = 1024;
+static const uint additional_buffer_size_default = 128;
+
 static bool
 checkExistence(const string &path)
 {
@@ -74,7 +78,7 @@ private:
 
     void handleInMemoryMessages();
 
-    void writeToDisk(const BufferedMessage &message);
+    bool writeToDisk(const BufferedMessage &message);
 
     static Maybe<uint32_t> seekStartOfMessage(FILE *file);
     static bool readBytes(FILE *file, uint size_to_read, char *output_bytes);
@@ -94,6 +98,9 @@ private:
     string buffer_output;
     string buffer_root_path;
     uint max_size_on_disk_MB = 0;
+    uint memory_messages_max_size = 0;
+    uint additional_buffer_size = 0;
+    uint memory_messages_reserve_size = reservation_default_size;
     uint curr_no_retries = 0;
     I_ShellCmd *shell_cmd = nullptr;
     I_Encryptor *encryptor = nullptr;
@@ -105,6 +112,16 @@ void
 MessagingBufferComponent::Impl::init()
 {
     max_size_on_disk_MB = getProfileAgentSettingWithDefault<uint>(buffer_max_size_MB, "eventBuffer.maxSizeOnDiskInMB");
+    memory_messages_max_size =
+        getProfileAgentSettingWithDefault<uint>(
+            memory_messages_max_size_default,
+            "eventBuffer.maxMemoryMessagesToStore"
+        );
+    additional_buffer_size =
+        getProfileAgentSettingWithDefault<uint>(
+            additional_buffer_size_default,
+            "eventBuffer.additionalBufferSize"
+        );
     shell_cmd = Singleton::Consume<I_ShellCmd>::by<Messaging>();
     encryptor = Singleton::Consume<I_Encryptor>::by<Messaging>();
     mainloop = Singleton::Consume<I_MainLoop>::by<Messaging>();
@@ -121,7 +138,7 @@ MessagingBufferComponent::Impl::init()
     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);
+    memory_messages.reserve(memory_messages_reserve_size);
 
     uint tmo = getConfigurationWithDefault<uint>(5, "message", "Send event retry in sec");
     mainloop->addRecurringRoutine(
@@ -138,6 +155,26 @@ MessagingBufferComponent::Impl::init()
         "Handling in-memory messages",
         false
     );
+
+    registerConfigLoadCb(
+        [this]() {
+            memory_messages_max_size =
+                getProfileAgentSettingWithDefault<uint>(
+                    1000,
+                    "eventBuffer.maxMemoryMessagesToStore"
+                );
+            max_size_on_disk_MB =
+                getProfileAgentSettingWithDefault<uint>(
+                    buffer_max_size_MB,
+                    "eventBuffer.maxSizeOnDiskInMB"
+                );
+            additional_buffer_size =
+                getProfileAgentSettingWithDefault<uint>(
+                    100,
+                    "eventBuffer.additionalBufferSize"
+                );
+        }
+    );
 }
 
 void
@@ -152,8 +189,10 @@ MessagingBufferComponent::Impl::pushNewBufferedMessage(
 {
     dbgTrace(D_MESSAGING_BUFFER) << "Pushing new message to buffer";
 
-    if (!force_immediate_writing) {
+    if (!force_immediate_writing && memory_messages.size() < memory_messages_max_size + additional_buffer_size) {
-        dbgDebug(D_MESSAGING_BUFFER) << "Holding message temporarily in memory";
+        dbgTrace(D_MESSAGING_BUFFER)
+            << "Holding message temporarily in memory. Memory messages size: "
+            << memory_messages.size();
         memory_messages.emplace_back(body, method, uri, category, message_metadata);
         return;
     }
@@ -328,30 +367,80 @@ void
 MessagingBufferComponent::Impl::handleInMemoryMessages()
 {
     auto messages = move(memory_messages);
-    memory_messages.reserve(32);
+    uint failed_messages = 0;
+    dbgDebug(D_MESSAGING_BUFFER) << "Handling " << to_string(messages.size()) <<" new in-memory messages";
 
-    for (const auto &message : messages) {
+    memory_messages.reserve(memory_messages_reserve_size);
-        if (sendMessage(message) != HTTPStatusCode::HTTP_OK) {
+    auto it = messages.begin();
-            if (message.getMessageMetadata().shouldBufferMessage()) writeToDisk(message);
+    for (; it != messages.end() && memory_messages.size() < memory_messages_max_size; ++it) {
+        if (sendMessage(*it) != HTTPStatusCode::HTTP_OK) {
+            if (it->getMessageMetadata().shouldBufferMessage()) {
+                if (!writeToDisk(*it)) ++failed_messages;
             }
+        }
+        dbgTrace(D_MESSAGING_BUFFER)
+            << "Processed "
+            << (it - messages.begin() + 1)
+            << " messages out of "
+            << messages.size();
         mainloop->yield();
     }
+    if (it == messages.end()) {
+        memory_messages_reserve_size = reservation_default_size;
+        if (failed_messages > 0) {
+            dbgDebug(D_MESSAGING_BUFFER)
+                << "Failed to handle "
+                << to_string(failed_messages)
+                << " messages out of "
+                << to_string(messages.size());
+        }
+        return;
     }
 
-void
+    memory_messages_reserve_size =
+        min(memory_messages_reserve_size * 2, memory_messages_max_size + additional_buffer_size);
+
+    dbgDebug(D_MESSAGING_BUFFER) << "Heap buffer is full. Storing messages to disk";
+    auto it2 = messages.end() - 1;
+    do {
+        if (it2->getMessageMetadata().shouldBufferMessage() && !writeToDisk(*it2)) {
+            failed_messages += it2 - it + 1;
+            break;
+        }
+    } while(it2-- != it);
+
+    if (failed_messages > 0) {
+        dbgDebug(D_MESSAGING_BUFFER)
+            << "Failed to handle "
+            << to_string(failed_messages)
+            << " messages out of "
+            << to_string(messages.size());
+    }
+}
+
+bool
 MessagingBufferComponent::Impl::writeToDisk(const BufferedMessage &message)
 {
+    static uint full_buffer_failed_messages = 0;
     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();
+        full_buffer_failed_messages++;
-        return;
+        if (full_buffer_failed_messages % 10 == 0) {
+            dbgWarning(D_MESSAGING_BUFFER)
+                << "Buffer is full. "
+                << full_buffer_failed_messages
+                << " messages will not be written to disk";
         }
+        dbgDebug(D_MESSAGING_BUFFER) << "Buffer is full. Message will not be written to disk: " << message.getURI();
+        return false;
+    }
+    full_buffer_failed_messages = 0;
 
     ofstream file(buffer_input, ios::app);
     if (!file.is_open()) {
         dbgWarning(D_MESSAGING_BUFFER) << "Failed to open file for writing. File: " << buffer_input;
-        return;
+        return false;
     }
 
     uint32_t size = serialized_message.size();
@@ -359,6 +448,8 @@ MessagingBufferComponent::Impl::writeToDisk(const BufferedMessage &message)
     file.write(reinterpret_cast<char *>(&size), sizeof(size));
     char type = 0;
     file.write(&type, 1);
+
+    return true;
 }
 
 Maybe<uint32_t>
@@ -538,7 +629,7 @@ MessagingBufferComponent::Impl::canWriteToDisk(size_t message_size) const
         return true;
     }
 
-    dbgWarning(D_MESSAGING_BUFFER)
+    dbgDebug(D_MESSAGING_BUFFER)
         << "Buffer size is full. Directry size: "
         << *maybe_directory_size
         << ", Message size: "

core/messaging/messaging_buffer_comp/messaging_buffer_comp_ut/messaging_buffer_comp_ut.cc

@@ -282,3 +282,51 @@ TEST_F(TestMessagingBuffer, testRoutinInMemory)
     msg = buffer_provider->peekMessage();
     ASSERT_FALSE(msg.ok());
 }
+
+TEST_F(TestMessagingBuffer, testRoutinInMemoryOverflow)
+{
+    string config_json =
+        "{"
+        "   \"agentSettings\": [\n"
+        "   {\n"
+        "       \"id\": \"123\",\n"
+        "       \"key\": \"eventBuffer.maxMemoryMessagesToStore\",\n"
+        "       \"value\": \"5\"\n"
+        "   },\n"
+        "   {\n"
+        "       \"id\": \"123\",\n"
+        "       \"key\": \"eventBuffer.additionalBufferSize\",\n"
+        "       \"value\": \"1\"\n"
+        "   }]\n"
+        "}";
+
+    istringstream ss(config_json);
+    Singleton::Consume<Config::I_Config>::from(config)->loadConfiguration(ss);
+
+    MessageCategory category = MessageCategory::GENERIC;
+    MessageMetadata message_metadata = MessageMetadata();
+    message_metadata.setShouldBufferMessage(true);
+    HTTPMethod method = HTTPMethod::POST;
+    HTTPResponse res(HTTPStatusCode::HTTP_OK, "");
+
+    for (int i = 0; i < 6; i++) {
+        string body = "body" + to_string(i);
+        buffer_provider->pushNewBufferedMessage(body, method,  "/" + to_string(i), category, message_metadata, false);
+        EXPECT_CALL(mock_messaging, sendSyncMessage(method,  "/" + to_string(i), body, _, _)).WillOnce(Return(res));
+    }
+
+    for (int i = 0; i < 2; i++) {
+        string body = "body" + to_string(i);
+        buffer_provider->pushNewBufferedMessage(body, method,  "/" + to_string(i), category, message_metadata, false);
+    }
+
+    memory_routine();
+
+    for (int i = 0; i < 2; i++) {
+        auto msg = buffer_provider->peekMessage();
+        ASSERT_TRUE(msg.ok());
+        buffer_provider->popMessage();
+    }
+    auto msg = buffer_provider->peekMessage();
+    ASSERT_FALSE(msg.ok());
+}

core/socket_is/socket_is.cc

@@ -26,9 +26,16 @@
 #include <sys/stat.h>
 #include <errno.h>
 #include "debug.h"
+#include "config.h"
+#include "singleton.h"
+#include "i_time_get.h"
 
 static const uint udp_max_packet_size = 1024 * 64;
 static const AlertInfo alert(AlertTeam::CORE, "socket i/s");
+static const uint CONNECT_TIMEOUT_MICROSECOUNDS(10000000); //10 seconds
+static const uint CHECK_CONNECTION_INTERVAL_MICROSECONDS(250000);  //0.25 seconds
+static const std::chrono::microseconds CHRONO_CHECK_CONNECTION_INTERVAL =
+    std::chrono::microseconds(CHECK_CONNECTION_INTERVAL_MICROSECONDS);
 
 USE_DEBUG_FLAG(D_SOCKET);
 
@@ -235,9 +242,117 @@ private:
     }
 };
 
+int setNonBlocking(int socket) {
+    dbgTrace(D_SOCKET) << "Setting socket to non-blocking mode";
+    int flags = fcntl(socket, F_GETFL, 0);
+    if (flags == -1) return -1;
+    return fcntl(socket, F_SETFL, flags | O_NONBLOCK);
+}
+
+int setBlocking(int socket) {
+    dbgTrace(D_SOCKET) << "Setting socket to blocking mode";
+    int flags = fcntl(socket, F_GETFL, 0);
+    if (flags == -1) return -1;
+    return fcntl(socket, F_SETFL, flags & ~O_NONBLOCK);
+}
+
 class TCPSocket : public SocketInternal
 {
 public:
+    static Maybe<unique_ptr<TCPSocket>>
+    connectAsClient(unique_ptr<TCPSocket> &tcp_socket)
+    {
+        if (setNonBlocking(tcp_socket->getSocket()) < 0) {
+            dbgTrace(D_SOCKET) << "Failed to set the socket to non-blocking mode";
+            return genError("Failed to set the socket to non-blocking mode");
+        }
+
+        chrono::microseconds time_before_connect =
+            Singleton::Consume<I_TimeGet>::by<SocketIS>()->getWalltime();
+
+
+        if (connect(
+                tcp_socket->getSocket(),
+                reinterpret_cast<struct sockaddr *>(&tcp_socket->server),
+                sizeof(struct sockaddr_in)
+            ) >= 0
+        ) {
+            dbgTrace(D_SOCKET) << "Successfully connected to socket";
+            if (setBlocking(tcp_socket->getSocket()) < 0) {
+                dbgWarning(D_SOCKET) << "Failed to set the socket to blocking mode";
+                close(tcp_socket->getSocket());
+                return genError("Failed to set the socket to blocking mode");
+            }
+            return move(tcp_socket);
+        }
+
+        if (setBlocking(tcp_socket->getSocket()) < 0) {
+            dbgWarning(D_SOCKET) << "Failed to set the socket to blocking mode";
+            close(tcp_socket->getSocket());
+            return genError("Failed to set the socket to blocking mode");
+        }
+
+        auto connection_timeout_to_server = getProfileAgentSettingWithDefault<uint>(
+            CONNECT_TIMEOUT_MICROSECOUNDS,
+            "agent.config.log.TCP.connectTimeout");
+
+        dbgTrace(D_SOCKET)
+            << "Waiting for the socket connection to be established"
+            << " with a timeout of "
+            << connection_timeout_to_server
+            << " microseconds and each iteration in this timeout is "
+            << CHECK_CONNECTION_INTERVAL_MICROSECONDS
+            << " microseconds";
+
+        int ready_fds = 0; // parameters for select
+        int err;
+        socklen_t len;
+        fd_set writefds;
+        struct timeval timeout;
+        timeout.tv_sec = 0;
+        timeout.tv_usec = 0;
+
+        while (
+            Singleton::Consume<I_TimeGet>::by<SocketIS>()->getWalltime() - time_before_connect
+            < chrono::microseconds(connection_timeout_to_server))
+        {
+            dbgTrace(D_SOCKET) << "Iterating to check the connection status";
+            Singleton::Consume<I_MainLoop>::by<SocketIS>()->yield(CHRONO_CHECK_CONNECTION_INTERVAL);
+
+            FD_ZERO(&writefds);
+            FD_SET(tcp_socket->getSocket(), &writefds);
+
+            ready_fds = select(tcp_socket->getSocket() + 1, NULL, &writefds, NULL, &timeout);
+
+            if (ready_fds > 0) {
+                len = sizeof(err);
+                if (getsockopt(tcp_socket->getSocket(), SOL_SOCKET, SO_ERROR, &err, &len) >= 0) {
+                    if (err == 0) {
+                        dbgTrace(D_SOCKET) << "Connected to socket";
+                        return move(tcp_socket);
+                    }
+                }
+            }
+        }
+
+        if (ready_fds > 0) {
+            // there is at least one file descriptor ready for IO operation
+            if (getsockopt(tcp_socket->getSocket(), SOL_SOCKET, SO_ERROR, &err, &len) < 0) {
+                dbgWarning(D_SOCKET) << "Failed to get socket options";
+                close(tcp_socket->getSocket());
+                return genError("Failed to get socket options");
+            }
+            if (err != 0) {
+                dbgWarning(D_SOCKET) << "Failed to connect socket. Error number: " << err;
+                close(tcp_socket->getSocket());
+                return genError("Failed to connect socket");
+            }
+        }
+        dbgWarning(D_SOCKET) << "No file descriptor is ready for IO operation";
+        close(tcp_socket->getSocket());
+        return genError("Failed to connect socket");
+    }
+
     static Maybe<unique_ptr<TCPSocket>>
     connectSock(bool _is_blocking, bool _is_server, const string &_address)
     {
@@ -262,15 +377,7 @@ public:
         tcp_socket->server.sin_port = htons(port);
 
         if (!tcp_socket->isServerSock()) {
-            if (connect(
+            return tcp_socket->connectAsClient(tcp_socket);
-                    tcp_socket->getSocket(),
-                    reinterpret_cast<struct sockaddr *>(&tcp_socket->server),
-                    sizeof(struct sockaddr_in)
-                ) == -1
-            ) {
-                return genError("Failed to connect socket");
-            }
-            return move(tcp_socket);
         }
 
         static const int on = 1;
@@ -638,7 +745,10 @@ SocketIS::Impl::genSocket(
         socketTypeName = "UNIXDG";
     } else if (type == SocketType::TCP) {
         Maybe<unique_ptr<SocketInternal>> tcp_sock = TCPSocket::connectSock(is_blocking, is_server, address);
-        if (!tcp_sock.ok()) return tcp_sock.passErr();
+        if (!tcp_sock.ok()) {
+            dbgWarning(D_SOCKET) << "Failed to initialize TCP socket. Error: " << tcp_sock.getErr();
+            return tcp_sock.passErr();
+        }
         new_sock = tcp_sock.unpackMove();
         socketTypeName = "TCP";
     } else if (type == SocketType::UDP) {

nodes/orchestration/package/orchestration_package.sh

@@ -791,7 +791,7 @@ upgrade_conf_if_needed()
         fi
     fi
 
-    cp_exec "${FILESYSTEM_PATH}/${SCRIPTS_PATH}/${CP_NANO_DEBUG} --default --service orchestration"
+    CP_ENV_FILESYSTEM=${FILESYSTEM_PATH} CP_ENV_LOG_FILE=${LOG_FILE_PATH} cp_exec "${FILESYSTEM_PATH}/${SCRIPTS_PATH}/${CP_NANO_DEBUG} --default --service orchestration"
 
     if [ ! -f "${FILESYSTEM_PATH}/${ENV_DETAILS_FILE}" ]; then
         cp_print "Creating env details file" ${FORCE_STDOUT}
@@ -1145,7 +1145,7 @@ install_orchestration()
     cp_copy configuration/cp-nano-orchestration-debug-conf.json ${FILESYSTEM_PATH}/${ORCHESTRATION_DEBUG_CONF_FILE}
     cp_exec "chmod 600 ${FILESYSTEM_PATH}/${ORCHESTRATION_DEBUG_CONF_FILE}"
 
-    cp_exec "${FILESYSTEM_PATH}/${SCRIPTS_PATH}/${CP_NANO_DEBUG} --default --service orchestration"
+    CP_ENV_FILESYSTEM=${FILESYSTEM_PATH} CP_ENV_LOG_FILE=${LOG_FILE_PATH} cp_exec "${FILESYSTEM_PATH}/${SCRIPTS_PATH}/${CP_NANO_DEBUG} --default --service orchestration"
 
     set_conf_additional_flags