Merge back changes (#2388)

* Add compression library to helper
* Implement sampler and processor trigger

appsec/cmake/helper.cmake

@@ -18,7 +18,7 @@ set_target_properties(helper_objects PROPERTIES
     POSITION_INDEPENDENT_CODE 1)
 target_include_directories(helper_objects PUBLIC ${HELPER_INCLUDE_DIR})
 target_compile_definitions(helper_objects PUBLIC SPDLOG_ACTIVE_LEVEL=SPDLOG_LEVEL_TRACE)
-target_link_libraries(helper_objects PUBLIC libddwaf_objects pthread spdlog cpp-base64 msgpack_c RapidJSON::rapidjson Boost::system)
+target_link_libraries(helper_objects PUBLIC libddwaf_objects pthread spdlog cpp-base64 msgpack_c RapidJSON::rapidjson Boost::system zlibstatic)
 
 add_executable(ddappsec-helper src/helper/main.cpp
 	$<TARGET_OBJECTS:helper_objects>

appsec/src/helper/client.cpp

@@ -434,6 +434,19 @@ bool client::handle_command(network::request_shutdown::request &command)
 
     auto response = std::make_shared<network::request_shutdown::response>();
     try {
+        auto sampler = service_->get_schema_sampler();
+        std::optional<sampler::scope> scope;
+        if (sampler) {
+            scope = sampler->get();
+            if (scope.has_value()) {
+                parameter context_processor = parameter::map();
+                context_processor.add(
+                    "extract-schema", parameter::as_boolean(true));
+                command.data.add(
+                    "waf.context.processor", std::move(context_processor));
+            }
+        }
+
         auto res = context_->publish(std::move(command.data));
         if (res) {
             switch (res->type) {

appsec/src/helper/compression.cpp

@@ -0,0 +1,105 @@
+// Unless explicitly stated otherwise all files in this repository are
+// dual-licensed under the Apache-2.0 License or BSD-3-Clause License.
+//
+// This product includes software developed at Datadog
+// (https://www.datadoghq.com/). Copyright 2021 Datadog, Inc.
+
+#include "compression.hpp"
+#include <cstdint>
+#include <string>
+#include <zlib.h>
+
+namespace dds {
+
+namespace {
+constexpr int64_t encoding = -0xf;
+constexpr int max_round_decompression = 100;
+// Taken from PHP approach
+// https://heap.space/xref/PHP-7.3/ext/zlib/php_zlib.h?r=8d3f8ca1#36
+size_t estimate_compressed_size(size_t in_len)
+{
+    // NOLINTBEGIN(cppcoreguidelines-avoid-magic-numbers,readability-magic-numbers)
+    return (((size_t)((double)in_len * (double)1.015)) + 10 + 8 + 4 + 1);
+    // NOLINTEND(cppcoreguidelines-avoid-magic-numbers,readability-magic-numbers)
+}
+} // namespace
+
+// The implementation of this function is based on how PHP does
+//  https://heap.space/xref/PHP-7.3/ext/zlib/zlib.c?r=9afce019#336
+std::optional<std::string> compress(const std::string &text)
+{
+    std::string ret_string;
+    z_stream strm = {};
+
+    if (text.length() == 0) {
+        return std::nullopt;
+    }
+
+    if (Z_OK == deflateInit2(&strm, -1, Z_DEFLATED, encoding, MAX_MEM_LEVEL,
+                    Z_DEFAULT_STRATEGY)) {
+        auto size = estimate_compressed_size(text.length());
+        ret_string.resize(size);
+
+        // NOLINTBEGIN(cppcoreguidelines-pro-type-reinterpret-cast)
+        strm.next_in = reinterpret_cast<const Bytef *>(text.data());
+        strm.next_out = reinterpret_cast<Bytef *>(ret_string.data());
+        // NOLINTEND(cppcoreguidelines-pro-type-reinterpret-cast)
+        strm.avail_in = text.length();
+        strm.avail_out = size;
+
+        if (Z_STREAM_END == deflate(&strm, Z_FINISH)) {
+            deflateEnd(&strm);
+            /* size buffer down to actual length */
+            ret_string.resize(strm.total_out);
+            ret_string.shrink_to_fit();
+            return ret_string;
+        }
+        deflateEnd(&strm);
+    }
+    return std::nullopt;
+}
+
+// Taken from PHP approach
+// https://heap.space/xref/PHP-7.3/ext/zlib/zlib.c?r=9afce019#422
+std::optional<std::string> uncompress(const std::string &compressed)
+{
+    int round = 0;
+    size_t used = 0;
+    size_t free;
+    size_t capacity;
+    z_stream strm = {};
+
+    if (compressed.length() < 1 || Z_OK != inflateInit2(&strm, encoding)) {
+        return std::nullopt;
+    }
+
+    // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
+    strm.next_in = reinterpret_cast<const Bytef *>(compressed.data());
+    strm.avail_in = compressed.length();
+    std::string output;
+    int status = Z_OK;
+
+    capacity = strm.avail_in;
+    output.resize(capacity);
+    while ((Z_BUF_ERROR == status || (Z_OK == status && strm.avail_in > 0)) &&
+           ++round < max_round_decompression) {
+        strm.avail_out = free = capacity - used;
+        // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
+        strm.next_out = reinterpret_cast<uint8_t *>(output.data()) + used;
+        status = inflate(&strm, Z_NO_FLUSH);
+        used += free - strm.avail_out;
+        capacity += (output.size() >> 3) + 1;
+        output.resize(capacity);
+    }
+    if (status == Z_STREAM_END) {
+        inflateEnd(&strm);
+        output.resize(used);
+        output.shrink_to_fit();
+        return output;
+    }
+    inflateEnd(&strm);
+
+    return std::nullopt;
+}
+
+} // namespace dds

appsec/src/helper/compression.hpp

@@ -0,0 +1,16 @@
+// Unless explicitly stated otherwise all files in this repository are
+// dual-licensed under the Apache-2.0 License or BSD-3-Clause License.
+//
+// This product includes software developed at Datadog
+// (https://www.datadoghq.com/). Copyright 2021 Datadog, Inc.
+#pragma once
+
+#include <optional>
+#include <string>
+
+namespace dds {
+
+std::optional<std::string> compress(const std::string &text);
+std::optional<std::string> uncompress(const std::string &compressed);
+
+} // namespace dds

appsec/src/helper/engine_settings.hpp

@@ -16,8 +16,9 @@ namespace dds {
 
 struct schema_extraction_settings {
     static constexpr double default_sample_rate = 0.1; // 10% of requests
+    static constexpr bool default_enabled = false;
 
-    bool enabled = false;
+    bool enabled = default_enabled;
     double sample_rate = default_sample_rate;
 
     MSGPACK_DEFINE_MAP(enabled, sample_rate);

appsec/src/helper/json_helper.cpp

@@ -37,6 +37,9 @@ void parameter_to_json_helper(const parameter_view &pv, T &output,
         auto sv = std::string_view(pv);
         output.SetString(sv.data(), sv.size(), alloc);
     } break;
+    case parameter_type::boolean:
+        output.SetBool(bool(pv));
+        break;
     case parameter_type::map:
         output.SetObject();
         for (const auto &v : pv) {

appsec/src/helper/network/msgpack_helpers.cpp

@@ -43,7 +43,7 @@ dds::parameter msgpack_to_param(const msgpack::object &o, unsigned depth = 0)
     case msgpack::type::STR:
         return dds::parameter::string(o.as<std::string_view>());
     case msgpack::type::BOOLEAN:
-        return dds::parameter::boolean(o.as<bool>());
+        return dds::parameter::as_boolean(o.as<bool>());
     case msgpack::type::FLOAT64:
         return dds::parameter::float64(o.as<float>());
     case msgpack::type::POSITIVE_INTEGER:

appsec/src/helper/parameter.cpp

@@ -87,7 +87,7 @@ parameter parameter::string(std::string_view str) noexcept
     return parameter{obj};
 }
 
-parameter parameter::boolean(bool value) noexcept
+parameter parameter::as_boolean(bool value) noexcept
 {
     ddwaf_object obj;
     ddwaf_object_bool(&obj, value);

appsec/src/helper/parameter.hpp

@@ -45,7 +45,7 @@ class parameter : public parameter_base {
     static parameter string(std::string_view str) noexcept;
     static parameter string(uint64_t value) noexcept;
     static parameter string(int64_t value) noexcept;
-    static parameter boolean(bool value) noexcept;
+    static parameter as_boolean(bool value) noexcept;
     static parameter float64(float value) noexcept;
     static parameter null() noexcept;
 

appsec/src/helper/parameter_base.hpp

@@ -18,7 +18,8 @@ enum parameter_type : unsigned {
     uint64 = DDWAF_OBJ_UNSIGNED,
     string = DDWAF_OBJ_STRING,
     map = DDWAF_OBJ_MAP,
-    array = DDWAF_OBJ_ARRAY
+    array = DDWAF_OBJ_ARRAY,
+    boolean = DDWAF_OBJ_BOOL
 };
 
 class parameter_base : public ddwaf_object {
@@ -80,6 +81,10 @@ class parameter_base : public ddwaf_object {
     {
         return ddwaf_object::type == DDWAF_OBJ_SIGNED;
     }
+    [[nodiscard]] bool is_boolean() const noexcept
+    {
+        return ddwaf_object::type == DDWAF_OBJ_BOOL;
+    }
     [[nodiscard]] bool is_valid() const noexcept
     {
         return ddwaf_object::type != DDWAF_OBJ_INVALID;
@@ -116,6 +121,14 @@ class parameter_base : public ddwaf_object {
         return intValue;
     }
 
+    explicit operator bool() const
+    {
+        if (!is_boolean()) {
+            throw bad_cast("parameter not a bool");
+        }
+        return boolean;
+    }
+
     [[nodiscard]] std::string debug_str() const noexcept;
 
     using length_type = decltype(nbEntries);

appsec/src/helper/sampler.hpp

@@ -0,0 +1,91 @@
+// Unless explicitly stated otherwise all files in this repository are
+// dual-licensed under the Apache-2.0 License or BSD-3-Clause License.
+//
+// This product includes software developed at Datadog
+// (https://www.datadoghq.com/). Copyright 2022 Datadog, Inc.
+
+#pragma once
+
+#include <atomic>
+#include <cmath>
+#include <cstdint>
+#include <iostream>
+#include <mutex>
+#include <optional>
+
+namespace dds {
+static const double min_rate = 0.0001;
+class sampler {
+public:
+    sampler(double sample_rate) : sample_rate_(sample_rate)
+    {
+        // NOLINTBEGIN(cppcoreguidelines-avoid-magic-numbers,readability-magic-numbers)
+        if (sample_rate_ <= 0) {
+            sample_rate_ = 0;
+        } else if (sample_rate_ > 1) {
+            sample_rate_ = 1;
+        } else if (sample_rate_ < min_rate) {
+            sample_rate_ = min_rate;
+        }
+        // NOLINTEND(cppcoreguidelines-avoid-magic-numbers,readability-magic-numbers)
+    }
+    class scope {
+    public:
+        explicit scope(std::atomic<bool> &concurrent) : concurrent_(&concurrent)
+        {
+            *concurrent_ = true;
+        }
+
+        scope(const scope &) = delete;
+        scope &operator=(const scope &) = delete;
+        scope(scope &&oth) noexcept
+        {
+            concurrent_ = oth.concurrent_;
+            oth.concurrent_ = nullptr;
+        }
+        scope &operator=(scope &&oth)
+        {
+            concurrent_ = oth.concurrent_;
+            oth.concurrent_ = nullptr;
+
+            return *this;
+        }
+
+        ~scope()
+        {
+            if (concurrent_ != nullptr) {
+                *concurrent_ = false;
+            }
+        }
+
+    protected:
+        std::atomic<bool> *concurrent_;
+    };
+
+    std::optional<scope> get()
+    {
+        const std::lock_guard<std::mutex> lock_guard(mtx_);
+
+        std::optional<scope> result = std::nullopt;
+
+        if (!concurrent_ && floor(request_ * sample_rate_) !=
+                                floor((request_ + 1) * sample_rate_)) {
+            result = {scope{concurrent_}};
+        }
+
+        if (request_ < UINT_MAX) {
+            request_++;
+        } else {
+            request_ = 1;
+        }
+
+        return result;
+    }
+
+protected:
+    unsigned request_{1};
+    double sample_rate_;
+    std::atomic<bool> concurrent_{false};
+    std::mutex mtx_;
+};
+} // namespace dds

appsec/src/helper/service.cpp

@@ -10,7 +10,8 @@ namespace dds {
 
 service::service(std::shared_ptr<engine> engine,
     std::shared_ptr<service_config> service_config,
-    dds::remote_config::client_handler::ptr &&client_handler)
+    dds::remote_config::client_handler::ptr &&client_handler,
+    const schema_extraction_settings &schema_extraction_settings)
     : engine_(std::move(engine)), service_config_(std::move(service_config)),
       client_handler_(std::move(client_handler))
 {
@@ -22,6 +23,14 @@ service::service(std::shared_ptr<engine> engine,
     if (client_handler_) {
         client_handler_->start();
     }
+
+    double sample_rate = schema_extraction_settings.sample_rate;
+
+    if (!schema_extraction_settings.enabled) {
+        sample_rate = 0;
+    }
+
+    schema_sampler_ = std::make_shared<sampler>(sample_rate);
 }
 
 service::ptr service::from_settings(service_identifier &&id,
@@ -38,7 +47,7 @@ service::ptr service::from_settings(service_identifier &&id,
         std::move(id), eng_settings, service_config, rc_settings, engine_ptr,
         dynamic_enablement);
 
-    return std::make_shared<service>(
-        engine_ptr, std::move(service_config), std::move(client_handler));
+    return std::make_shared<service>(engine_ptr, std::move(service_config),
+        std::move(client_handler), eng_settings.schema_extraction);
 }
 } // namespace dds

appsec/src/helper/service.hpp

@@ -8,6 +8,7 @@
 #include "engine.hpp"
 #include "exception.hpp"
 #include "remote_config/client_handler.hpp"
+#include "sampler.hpp"
 #include "service_config.hpp"
 #include "service_identifier.hpp"
 #include "std_logging.hpp"
@@ -26,7 +27,8 @@ class service {
 
     service(std::shared_ptr<engine> engine,
         std::shared_ptr<service_config> service_config,
-        dds::remote_config::client_handler::ptr &&client_handler);
+        dds::remote_config::client_handler::ptr &&client_handler,
+        const schema_extraction_settings &schema_extraction_settings = {});
 
     service(const service &) = delete;
     service &operator=(const service &) = delete;
@@ -68,10 +70,13 @@ class service {
         return service_config_;
     }
 
+    std::shared_ptr<sampler> get_schema_sampler() { return schema_sampler_; }
+
 protected:
     std::shared_ptr<engine> engine_{};
     std::shared_ptr<service_config> service_config_{};
     dds::remote_config::client_handler::ptr client_handler_{};
+    std::shared_ptr<sampler> schema_sampler_;
 };
 
 } // namespace dds