github_mirrors/attachment
3
0
Fork 1
mirror of https://github.com/openappsec/attachment.git synced 2026-01-17 16:00:26 +03:00
Code Issues Packages Projects Releases Wiki Activity

sync code

Browse Source
This commit is contained in:
Daniel Eisenberg
2026-01-06 16:31:33 +02:00
parent 7ce1fba437
commit 599aa34732
82 changed files with 7557 additions and 807 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
core/attachments/http_configuration/http_configuration.cc
View File

@@ -106,7 +106,7 @@ HttpAttachmentConfiguration::save(cereal::JSONOutputArchive &archive) const
"waiting_for_verdict_thread_timeout_msec",
getNumericalValue("waiting_for_verdict_thread_timeout_msec")
),
cereal::make_nvp("nginx_inspection_mode", getNumericalValue("inspection_mode")),
cereal::make_nvp("nginx_inspection_mode", getNumericalValue("nginx_inspection_mode")),
cereal::make_nvp("num_of_nginx_ipc_elements", getNumericalValue("num_of_nginx_ipc_elements")),
cereal::make_nvp("keep_alive_interval_msec", getNumericalValue("keep_alive_interval_msec")),
cereal::make_nvp("min_retries_for_verdict", getNumericalValue("min_retries_for_verdict")),
@@ -114,7 +114,12 @@ HttpAttachmentConfiguration::save(cereal::JSONOutputArchive &archive) const
cereal::make_nvp("hold_verdict_retries", getNumericalValue("hold_verdict_retries")),
cereal::make_nvp("hold_verdict_polling_time", getNumericalValue("hold_verdict_polling_time")),
cereal::make_nvp("body_size_trigger", getNumericalValue("body_size_trigger")),
cereal::make_nvp("remove_server_header", getNumericalValue("remove_server_header"))
cereal::make_nvp("remove_server_header", getNumericalValue("remove_server_header")),
cereal::make_nvp("decompression_pool_size", getNumericalValue("decompression_pool_size")),
cereal::make_nvp("recompression_pool_size", getNumericalValue("recompression_pool_size")),
cereal::make_nvp("is_paired_affinity_enabled", getNumericalValue("is_paired_affinity_enabled")),
cereal::make_nvp("is_async_mode_enabled", getNumericalValue("is_async_mode_enabled")),
cereal::make_nvp("is_brotli_inspection_enabled", getNumericalValue("is_brotli_inspection_enabled"))
);
}
@@ -173,6 +178,21 @@ HttpAttachmentConfiguration::load(cereal::JSONInputArchive &archive)
loadNumericalValue(archive, "hold_verdict_polling_time", 1);
loadNumericalValue(archive, "body_size_trigger", 200000);
loadNumericalValue(archive, "remove_server_header", 0);
loadNumericalValue(archive, "decompression_pool_size", 262144);
loadNumericalValue(archive, "recompression_pool_size", 16384);
loadNumericalValue(archive, "is_paired_affinity_enabled", 0);
loadNumericalValue(archive, "is_brotli_inspection_enabled", 0);
int g_env_async_mode = 1;
char *env_async_mode = getenv("CP_ASYNC_MODE");
if (env_async_mode != NULL) {
if (strcmp(env_async_mode, "true") == 0 || strcmp(env_async_mode, "1") == 0) {
g_env_async_mode = 1;
} else {
g_env_async_mode = 0;
}
}
loadNumericalValue(archive, "is_async_mode_enabled", g_env_async_mode);
}
bool

241
core/compression/compression_utils.cc
View File

@@ -22,6 +22,8 @@
#include <strings.h>
#include <string.h>
#include <zlib.h>
#include <brotli/encode.h>
#include <brotli/decode.h>
using namespace std;
@@ -29,6 +31,10 @@ using DebugFunction = void(*)(const char *);
static const int max_debug_level = static_cast<int>(CompressionUtilsDebugLevel::COMPRESSION_DBG_LEVEL_ASSERTION);
static const int max_retries = 3;
static const size_t default_brotli_buffer_size = 16384;
static const size_t brotli_decompression_probe_size = 64;
static void
defaultPrint(const char *debug_message)
{
@@ -104,12 +110,23 @@ static const int zlib_no_flush = Z_NO_FLUSH;
struct CompressionStream
{
CompressionStream() { bzero(&stream, sizeof(z_stream)); }
CompressionStream()
:
br_encoder_state(nullptr),
br_decoder_state(nullptr)
{
bzero(&stream, sizeof(z_stream));
}
~CompressionStream() { fini(); }
tuple<basic_string<unsigned char>, bool>
decompress(const unsigned char *data, uint32_t size)
{
if (state == TYPE::UNINITIALIZED && size > 0 && isBrotli(data, size)) return decompressBrotli(data, size);
if (state == TYPE::DECOMPRESS_BROTLI) return decompressBrotli(data, size);
initInflate();
if (state != TYPE::DECOMPRESS) throw runtime_error("Could not start decompression");
@@ -138,7 +155,7 @@ struct CompressionStream
res.append(work_space.data(), stream.total_out - old_total_out);
} else {
++retries;
if (retries > 3) {
if (retries > max_retries) {
fini();
throw runtime_error("No results from inflate more than three times");
}
@@ -156,6 +173,7 @@ struct CompressionStream
basic_string<unsigned char>
compress(CompressionType type, const unsigned char *data, uint32_t size, int is_last_chunk)
{
if (type == CompressionType::BROTLI) return compressBrotli(data, size, is_last_chunk);
initDeflate(type);
if (state != TYPE::COMPRESS) throw runtime_error("Could not start compression");
@@ -183,7 +201,7 @@ struct CompressionStream
res.append(work_space.data(), stream.total_out - old_total_out);
} else {
++retries;
if (retries > 3) {
if (retries > max_retries) {
fini();
throw runtime_error("No results from deflate more than three times");
}
@@ -201,7 +219,7 @@ private:
void
initInflate()
{
if (state != TYPE::UNINITIALIZAED) return;
if (state != TYPE::UNINITIALIZED) return;
auto init_status = inflateInit2(&stream, default_num_window_bits + 32);
if (init_status != zlib_ok_return_value) {
@@ -216,7 +234,7 @@ private:
void
initDeflate(CompressionType type)
{
if (state != TYPE::UNINITIALIZAED) return;
if (state != TYPE::UNINITIALIZED) return;
int num_history_window_bits;
switch (type) {
@@ -228,6 +246,10 @@ private:
num_history_window_bits = default_num_window_bits;
break;
}
case CompressionType::BROTLI: {
zlibDbgAssertion << "Brotli compression should use compressBrotli()";
return;
}
default: {
zlibDbgAssertion
<< "Invalid compression type value: "
@@ -253,6 +275,190 @@ private:
state = TYPE::COMPRESS;
}
basic_string<unsigned char>
compressBrotli(const unsigned char *data, uint32_t size, int is_last_chunk)
{
if (state == TYPE::UNINITIALIZED) {
br_encoder_state = BrotliEncoderCreateInstance(nullptr, nullptr, nullptr);
if (!br_encoder_state) throw runtime_error("Failed to create Brotli encoder state");
BrotliEncoderSetParameter(br_encoder_state, BROTLI_PARAM_QUALITY, BROTLI_DEFAULT_QUALITY);
BrotliEncoderSetParameter(br_encoder_state, BROTLI_PARAM_LGWIN, BROTLI_DEFAULT_WINDOW);
state = TYPE::COMPRESS_BROTLI;
} else if (state != TYPE::COMPRESS_BROTLI) {
throw runtime_error("Compression stream in inconsistent state for Brotli compression");
}
basic_string<unsigned char> output;
vector<uint8_t> buffer(16384);
int retries = 0;
const uint8_t* next_in = data;
size_t available_in = size;
while (available_in > 0 || is_last_chunk) {
size_t available_out = buffer.size();
uint8_t* next_out = buffer.data();
BrotliEncoderOperation op = is_last_chunk ? BROTLI_OPERATION_FINISH : BROTLI_OPERATION_PROCESS;
auto brotli_success = BrotliEncoderCompressStream(
br_encoder_state,
op,
&available_in,
&next_in,
&available_out,
&next_out,
nullptr
);
if (brotli_success == BROTLI_FALSE) {
fini();
throw runtime_error("Brotli compression error");
}
size_t bytes_written = buffer.size() - available_out;
if (bytes_written > 0) {
output.append(buffer.data(), bytes_written);
retries = 0;
} else {
retries++;
if (retries > max_retries) {
fini();
throw runtime_error("Brotli compression error: Exceeded retry limit.");
}
}
if (BrotliEncoderIsFinished(br_encoder_state)) break;
if (available_in == 0 && !is_last_chunk) break;
}
if (is_last_chunk) fini();
return output;
}
tuple<basic_string<unsigned char>, bool>
decompressBrotli(const unsigned char *data, uint32_t size)
{
if (state != TYPE::DECOMPRESS_BROTLI) {
br_decoder_state = BrotliDecoderCreateInstance(nullptr, nullptr, nullptr);
if (!br_decoder_state) throw runtime_error("Failed to create Brotli decoder state");
BrotliDecoderSetParameter(br_decoder_state, BROTLI_DECODER_PARAM_LARGE_WINDOW, 1u);
state = TYPE::DECOMPRESS_BROTLI;
}
basic_string<unsigned char> output;
const uint8_t* next_in = data;
size_t available_in = size;
size_t buffer_size = max<size_t>(size * 4, default_brotli_buffer_size);
vector<uint8_t> buffer(buffer_size);
// Use a constant ratio for max buffer size relative to input size
const size_t max_buffer_size = 256 * 1024 * 1024; // 256 MB max buffer size
while (true) {
size_t available_out = buffer.size();
uint8_t* next_out = buffer.data();
BrotliDecoderResult result = BrotliDecoderDecompressStream(
br_decoder_state,
&available_in,
&next_in,
&available_out,
&next_out,
nullptr
);
if (result == BROTLI_DECODER_RESULT_ERROR) {
fini();
auto error_msg = string(BrotliDecoderErrorString(BrotliDecoderGetErrorCode(br_decoder_state)));
throw runtime_error("Brotli decompression error: " + error_msg);
}
// Handle any produced output
size_t bytes_produced = buffer.size() - available_out;
if (bytes_produced > 0) {
output.append(buffer.data(), bytes_produced);
}
if (result == BROTLI_DECODER_RESULT_SUCCESS) {
bool is_finished = BrotliDecoderIsFinished(br_decoder_state);
if (is_finished) fini();
return make_tuple(output, is_finished);
}
if (result == BROTLI_DECODER_RESULT_NEEDS_MORE_OUTPUT) {
// Check if we've exceeded the maximum buffer size limit
if (buffer.size() >= max_buffer_size) {
fini();
throw runtime_error("Brotli decompression buffer size limit exceeded - possibly corrupted data");
}
// Resize buffer to accommodate more output
size_t new_size = min(buffer.size() * 2, max_buffer_size);
buffer.resize(new_size);
continue; // Continue with the same input, new buffer
}
// If we reach here, we need more input but have no more to provide
if (available_in == 0) {
// No more input data available, return what we have so far
return make_tuple(output, false);
}
}
return make_tuple(output, false);
}
bool
isBrotli(const unsigned char *data, uint32_t size)
{
if (size < 4) return false;
BrotliDecoderState* test_decoder = BrotliDecoderCreateInstance(nullptr, nullptr, nullptr);
if (!test_decoder) return false;
const uint8_t* next_in = data;
size_t available_in = min<size_t>(size, brotli_decompression_probe_size);
uint8_t output[brotli_decompression_probe_size];
size_t available_out = sizeof(output);
uint8_t* next_out = output;
BrotliDecoderResult result = BrotliDecoderDecompressStream(
test_decoder,
&available_in,
&next_in,
&available_out,
&next_out,
nullptr
);
bool is_brotli = false;
if (
result != BROTLI_DECODER_RESULT_ERROR &&
(
available_out < sizeof(output) ||
available_in < min<size_t>(size, brotli_decompression_probe_size)
)
) {
is_brotli = true;
}
BrotliDecoderDestroyInstance(test_decoder);
if (is_brotli) {
br_decoder_state = BrotliDecoderCreateInstance(nullptr, nullptr, nullptr);
BrotliDecoderSetParameter(br_decoder_state, BROTLI_DECODER_PARAM_LARGE_WINDOW, 1u);
state = TYPE::DECOMPRESS_BROTLI;
return true;
}
return false;
}
void
fini()
{
@@ -261,11 +467,21 @@ private:
if (state == TYPE::DECOMPRESS) end_stream_res = inflateEnd(&stream);
if (state == TYPE::COMPRESS) end_stream_res = deflateEnd(&stream);
if (end_stream_res != zlib_ok_return_value) {
if (br_encoder_state) {
BrotliEncoderDestroyInstance(br_encoder_state);
br_encoder_state = nullptr;
}
if (br_decoder_state) {
BrotliDecoderDestroyInstance(br_decoder_state);
br_decoder_state = nullptr;
}
if (end_stream_res != zlib_ok_return_value && end_stream_res != Z_DATA_ERROR) {
zlibDbgError << "Failed to clean state: " << getZlibError(end_stream_res);
}
state = TYPE::UNINITIALIZAED;
state = TYPE::UNINITIALIZED;
}
string
@@ -288,7 +504,16 @@ private:
}
z_stream stream;
enum class TYPE { UNINITIALIZAED, COMPRESS, DECOMPRESS } state = TYPE::UNINITIALIZAED;
enum class TYPE {
UNINITIALIZED,
COMPRESS,
DECOMPRESS,
COMPRESS_BROTLI,
DECOMPRESS_BROTLI
} state = TYPE::UNINITIALIZED;
BrotliEncoderState* br_encoder_state = nullptr;
BrotliDecoderState* br_decoder_state = nullptr;
};
void

40
core/include/attachments/nano_attachment.h
View File

@@ -225,35 +225,35 @@ void FreeAttachmentResponseContent(
);
///
/// @brief Compresses HttpBody and return allocated compressed body.
/// @brief Compresses NanoHttpBody and return allocated compressed body.
///
/// @param attachment The NanoAttachment object associated with the session.
/// @param session_data The HttpSessionData object representing the session.
/// @param bodies The bodies pointer to be compressed.
///
HttpBody * compressBody(
NanoHttpBody * compressBody(
NanoAttachment *attachment,
HttpSessionData *session_data,
HttpBody *bodies
NanoHttpBody *bodies
);
///
/// @brief Compresses HttpBody and return allocated compressed body.
/// @brief Compresses NanoHttpBody and return allocated compressed body.
///
/// @param attachment The NanoAttachment object associated with the session.
/// @param session_data The HttpSessionData object representing the session.
/// @param bodies The bodies pointer to be decompressed.
///
HttpBody * decompressBody(
NanoHttpBody * decompressBody(
NanoAttachment *attachment,
HttpSessionData *session_data,
HttpBody *bodies
NanoHttpBody *bodies
);
///
/// @brief Free allocated compressed body.
///
/// This function frees the allocated resources of HttpBody object.
/// This function frees the allocated resources of NanoHttpBody object.
///
/// @param attachment The NanoAttachment object associated with the session.
/// @param session_data The HttpSessionData object representing the session.
@@ -263,31 +263,7 @@ void
freeCompressedBody(
NanoAttachment *attachment,
HttpSessionData *session_data,
HttpBody *bodies
NanoHttpBody *bodies
);
///
/// @brief Gets the request processing timeout in milliseconds.
///
/// This function retrieves the configured timeout value for request processing
/// from the NanoAttachment configuration.
///
/// @param attachment A pointer to the NanoAttachment structure.
///
/// @return The request processing timeout in milliseconds.
///
uint32_t GetRequestProcessingTimeout(NanoAttachment *attachment);
///
/// @brief Gets the response processing timeout in milliseconds.
///
/// This function retrieves the configured timeout value for response processing
/// from the NanoAttachment configuration.
///
/// @param attachment A pointer to the NanoAttachment structure.
///
/// @return The response processing timeout in milliseconds.
///
uint32_t GetResponseProcessingTimeout(NanoAttachment *attachment);
#endif // __NANO_ATTACHMENT_H__

191
core/include/attachments/nano_attachment_common.h
View File

@@ -7,10 +7,16 @@
#include <sys/types.h>
#include <assert.h>
#include <sched.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include "compression_utils.h"
typedef uint32_t SessionID;
typedef void* DataBuffer;
typedef int64_t NanoHttpCpInjectPos;
#define MAX_NGINX_UID_LEN 32
#define MAX_SHARED_MEM_PATH_LEN 128
@@ -175,7 +181,9 @@ typedef enum AttachmentDataType
RESPONSE_END,
CONTENT_LENGTH,
METRIC_DATA_FROM_PLUGIN,
REQUEST_DELAYED_VERDICT
REQUEST_DELAYED_VERDICT,
COUNT
} AttachmentDataType;
#ifdef __cplusplus
@@ -207,9 +215,23 @@ typedef enum ServiceVerdict
TRAFFIC_VERDICT_INJECT,
TRAFFIC_VERDICT_IRRELEVANT,
TRAFFIC_VERDICT_RECONF,
TRAFFIC_VERDICT_DELAYED
TRAFFIC_VERDICT_DELAYED,
LIMIT_RESPONSE_HEADERS,
TRAFFIC_VERDICT_CUSTOM_RESPONSE
} ServiceVerdict;
#ifdef __cplusplus
typedef enum class AttachmentContentType
#else
typedef enum AttachmentContentType
#endif
{
CONTENT_TYPE_APPLICATION_JSON,
CONTENT_TYPE_TEXT_HTML,
CONTENT_TYPE_TEXT_PLAIN,
CONTENT_TYPE_OTHER
} AttachmentContentType;
#ifdef __cplusplus
typedef enum class AttachmentVerdict
#else
@@ -234,7 +256,7 @@ typedef enum HttpModificationType
} HttpModificationType;
typedef struct __attribute__((__packed__)) HttpInjectData {
int64_t injection_pos;
NanoHttpCpInjectPos injection_pos;
HttpModificationType mod_type;
uint16_t injection_size;
uint8_t is_header;
@@ -263,6 +285,13 @@ typedef struct __attribute__((__packed__)) HttpWebResponseData {
} response_data;
} HttpWebResponseData;
typedef struct __attribute__((__packed__)) HttpJsonResponseData {
uint16_t response_code;
uint16_t body_size;
AttachmentContentType content_type;
char body[0];
} HttpJsonResponseData;
typedef struct {
size_t len;
unsigned char *data;
@@ -308,6 +337,8 @@ typedef enum HttpMetaDataType
PARSED_HOST_DATA,
PARSED_URI_SIZE,
PARSED_URI_DATA,
WAF_TAG_SIZE,
WAF_TAG_DATA,
META_DATA_COUNT
} HttpMetaDataType;
@@ -402,10 +433,10 @@ typedef struct ResHttpHeaders {
uint64_t content_length;
} ResHttpHeaders;
typedef struct HttpBody {
typedef struct NanoHttpBody {
nano_str_t *data;
size_t bodies_count;
} HttpBody;
} NanoHttpBody;
typedef struct AttachmentData {
SessionID session_id;
@@ -417,6 +448,7 @@ typedef struct AttachmentData {
typedef union __attribute__((__packed__)) HttpModifyData {
HttpInjectData inject_data[0];
HttpWebResponseData web_response_data[0];
HttpJsonResponseData json_response_data[0];
} HttpModifyData;
typedef struct __attribute__((__packed__)) HttpReplyFromService {
@@ -479,6 +511,12 @@ typedef struct NanoResponseModifications {
NanoHttpModificationList *modifications;
} NanoResponseModifications;
typedef struct __attribute__((__packed__)) NanoHttpRequestData {
uint16_t data_type;
uint32_t session_id;
unsigned char data[0];
} NanoHttpRequestData;
typedef struct __attribute__((__packed__)) NanoHttpMetricData {
uint16_t data_type;
#ifdef __cplusplus
@@ -488,4 +526,147 @@ typedef struct __attribute__((__packed__)) NanoHttpMetricData {
#endif
} NanoHttpMetricData;
// Simple but reliable hash function for generating consistent, well-distributed offsets
// Uses a basic polynomial hash that avoids large intermediate values
static inline uint32_t hash_string(const char *str) {
uint32_t hash = 0;
while (*str) {
hash = (hash * 31 + (unsigned char)*str++) % 10000; // Keep values under 10000
}
return hash; // Return bounded hash - modulo will be applied by caller
}
static inline int set_affinity_by_uid(uint32_t uid) {
int num_cores = sysconf(_SC_NPROCESSORS_CONF);
// Debug print for troubleshooting
fprintf(stderr, "[DEBUG] set_affinity_by_uid: num_cores=%d, uid=%u\n", num_cores, uid);
uint32_t core_num = (uid - 1) % num_cores; // Ensure core_num is within bounds
cpu_set_t mask, mask_check;
CPU_ZERO(&mask);
CPU_ZERO(&mask_check);
CPU_SET(core_num, &mask);
pid_t pid = getpid(); // Use process PID, not thread ID
if (sched_setaffinity(pid, sizeof(mask), &mask) != 0) {
return -1; // Error setting affinity
}
if (sched_getaffinity(pid, sizeof(mask_check), &mask_check) != 0) {
return -2; // Error getting affinity
}
// Compare mask and mask_check
int i;
for (i = 0; i < num_cores; ++i) {
if (CPU_ISSET(i, &mask) != CPU_ISSET(i, &mask_check)) {
return -3; // Affinity not set as expected
}
}
return 0; // Success
}
static inline int set_affinity_by_uid_with_offset(uint32_t uid, uint32_t offset) {
int num_cores = sysconf(_SC_NPROCESSORS_CONF);
// Debug print for troubleshooting
fprintf(
stderr, "[DEBUG] set_affinity_by_uid_with_offset: num_cores=%d, uid=%u, offset=%u\n", num_cores, uid, offset);
// Prevent integer overflow by applying modulo to offset first
uint32_t safe_offset = offset % num_cores;
uint32_t core_num = ((uid - 1) + safe_offset) % num_cores;
cpu_set_t mask, mask_check;
CPU_ZERO(&mask);
CPU_ZERO(&mask_check);
CPU_SET(core_num, &mask);
pid_t pid = getpid(); // Use process PID, not thread ID
if (sched_setaffinity(pid, sizeof(mask), &mask) != 0) {
return -1; // Error setting affinity
}
if (sched_getaffinity(pid, sizeof(mask_check), &mask_check) != 0) {
return -2; // Error getting affinity
}
// Compare mask and mask_check
int i;
for (i = 0; i < num_cores; ++i) {
if (CPU_ISSET(i, &mask) != CPU_ISSET(i, &mask_check)) {
return -3; // Affinity not set as expected
}
}
return 0; // Success
}
static inline int set_affinity_by_uid_with_offset_fixed_cores(uint32_t uid, uint32_t offset, int num_cores) {
// Debug print for troubleshooting
fprintf(
stderr,
"[DEBUG] set_affinity_by_uid_with_offset_fixed_cores: num_cores=%d, uid=%u, offset=%u\n",
num_cores,
uid,
offset
);
// Prevent integer overflow by applying modulo to offset first
uint32_t safe_offset = offset % num_cores;
uint32_t core_num = ((uid - 1) + safe_offset) % num_cores;
cpu_set_t mask, mask_check;
CPU_ZERO(&mask);
CPU_ZERO(&mask_check);
CPU_SET(core_num, &mask);
pid_t pid = getpid(); // Use process PID, not thread ID
if (sched_setaffinity(pid, sizeof(mask), &mask) != 0) {
return -1; // Error setting affinity
}
if (sched_getaffinity(pid, sizeof(mask_check), &mask_check) != 0) {
return -2; // Error getting affinity
}
// Compare mask and mask_check
int i;
for (i = 0; i < num_cores; ++i) {
if (CPU_ISSET(i, &mask) != CPU_ISSET(i, &mask_check)) {
return -3; // Affinity not set as expected
}
}
return 0; // Success
}
static inline int set_affinity_to_core(int target_core) {
// Debug print for troubleshooting
fprintf(stderr, "[DEBUG] set_affinity_to_core: target_core=%d\n", target_core);
cpu_set_t mask, mask_check;
CPU_ZERO(&mask);
CPU_ZERO(&mask_check);
CPU_SET(target_core, &mask);
pid_t pid = getpid(); // Use process PID, not thread ID
if (sched_setaffinity(pid, sizeof(mask), &mask) != 0) {
return -1; // Error setting affinity
}
if (sched_getaffinity(pid, sizeof(mask_check), &mask_check) != 0) {
return -2; // Error getting affinity
}
// Compare mask and mask_check
int num_cores = sysconf(_SC_NPROCESSORS_CONF);
int i;
for (i = 0; i < num_cores; ++i) {
if (CPU_ISSET(i, &mask) != CPU_ISSET(i, &mask_check)) {
return -3; // Affinity not set as expected
}
}
return 0; // Success
}
static inline int reset_affinity() {
int num_cores = sysconf(_SC_NPROCESSORS_CONF);
// Debug print for troubleshooting
fprintf(stderr, "[DEBUG] reset_affinity: num_cores=%d\n", num_cores);
cpu_set_t mask;
CPU_ZERO(&mask);
int i;
for (i = 0; i < num_cores; ++i) CPU_SET(i, &mask);
pid_t pid = getpid(); // Use process PID, not thread ID
if (sched_setaffinity(pid, sizeof(mask), &mask) != 0) {
return -1; // Error setting affinity
}
return 0; // Success
}
#endif // __NANO_ATTACHMENT_COMMON_H__

282
core/include/attachments/nginx_attachment_common.h
View File

@@ -15,286 +15,10 @@
#ifndef __NGINX_ATTACHMENT_COMMON_H__
#define __NGINX_ATTACHMENT_COMMON_H__
#include <stddef.h>
#include <stdint.h>
#include <sys/types.h>
#include <assert.h>
// This file has been deprecated. Do not add anything here.
// Any future additions should be added to nano_attachment_common.h
// For any inquiries please contact Daniel Yashin.
#define MAX_NGINX_UID_LEN 32
#define NUM_OF_NGINX_IPC_ELEMENTS 200
#define DEFAULT_KEEP_ALIVE_INTERVAL_MSEC 300000
#define SHARED_MEM_PATH "/dev/shm/"
#define SHARED_REGISTRATION_SIGNAL_PATH SHARED_MEM_PATH "check-point/cp-nano-attachment-registration"
#define SHARED_KEEP_ALIVE_PATH SHARED_MEM_PATH "check-point/cp-nano-attachment-registration-expiration-socket"
#define SHARED_VERDICT_SIGNAL_PATH SHARED_MEM_PATH "check-point/cp-nano-http-transaction-handler"
#define SHARED_ATTACHMENT_CONF_PATH SHARED_MEM_PATH "cp_nano_http_attachment_conf"
#define DEFAULT_STATIC_RESOURCES_PATH SHARED_MEM_PATH "static_resources"
#define INJECT_POS_IRRELEVANT -1
#define CORRUPTED_SESSION_ID 0
#define METRIC_PERIODIC_TIMEOUT 600
extern char shared_verdict_signal_path[];
extern int workers_amount_to_send;
typedef int64_t ngx_http_cp_inject_pos_t;
#ifdef __cplusplus
typedef enum class ngx_http_modification_type
#else
typedef enum ngx_http_modification_type
#endif
{
APPEND,
INJECT,
REPLACE
} ngx_http_modification_type_e;
#ifdef __cplusplus
typedef enum class ngx_http_chunk_type
#else
typedef enum ngx_http_chunk_type
#endif
{
REQUEST_START,
REQUEST_HEADER,
REQUEST_BODY,
REQUEST_END,
RESPONSE_CODE,
RESPONSE_HEADER,
RESPONSE_BODY,
RESPONSE_END,
CONTENT_LENGTH,
METRIC_DATA_FROM_PLUGIN,
HOLD_DATA,
COUNT
} ngx_http_chunk_type_e;
#ifdef __cplusplus
typedef enum class ngx_http_plugin_metric_type
#else
typedef enum ngx_http_plugin_metric_type
#endif
{
TRANSPARENTS_COUNT,
TOTAL_TRANSPARENTS_TIME,
INSPECTION_OPEN_FAILURES_COUNT,
INSPECTION_CLOSE_FAILURES_COUNT,
INSPECTION_SUCCESSES_COUNT,
INJECT_VERDICTS_COUNT,
DROP_VERDICTS_COUNT,
ACCEPT_VERDICTS_COUNT,
IRRELEVANT_VERDICTS_COUNT,
RECONF_VERDICTS_COUNT,
INSPECT_VERDICTS_COUNT,
HOLD_VERDICTS_COUNT,
AVERAGE_OVERALL_PPROCESSING_TIME_UNTIL_VERDICT,
MAX_OVERALL_PPROCESSING_TIME_UNTIL_VERDICT,
MIN_OVERALL_PPROCESSING_TIME_UNTIL_VERDICT,
AVERAGE_REQ_PPROCESSING_TIME_UNTIL_VERDICT,
MAX_REQ_PPROCESSING_TIME_UNTIL_VERDICT,
MIN_REQ_PPROCESSING_TIME_UNTIL_VERDICT,
AVERAGE_RES_PPROCESSING_TIME_UNTIL_VERDICT,
MAX_RES_PPROCESSING_TIME_UNTIL_VERDICT,
MIN_RES_PPROCESSING_TIME_UNTIL_VERDICT,
THREAD_TIMEOUT,
REG_THREAD_TIMEOUT,
REQ_HEADER_THREAD_TIMEOUT,
REQ_BODY_THREAD_TIMEOUT,
AVERAGE_REQ_BODY_SIZE_UPON_TIMEOUT,
MAX_REQ_BODY_SIZE_UPON_TIMEOUT,
MIN_REQ_BODY_SIZE_UPON_TIMEOUT,
RES_HEADER_THREAD_TIMEOUT,
RES_BODY_THREAD_TIMEOUT,
HOLD_THREAD_TIMEOUT,
AVERAGE_RES_BODY_SIZE_UPON_TIMEOUT,
MAX_RES_BODY_SIZE_UPON_TIMEOUT,
MIN_RES_BODY_SIZE_UPON_TIMEOUT,
THREAD_FAILURE,
REQ_PROCCESSING_TIMEOUT,
RES_PROCCESSING_TIMEOUT,
REQ_FAILED_TO_REACH_UPSTREAM,
REQ_FAILED_COMPRESSION_COUNT,
RES_FAILED_COMPRESSION_COUNT,
REQ_FAILED_DECOMPRESSION_COUNT,
RES_FAILED_DECOMPRESSION_COUNT,
REQ_SUCCESSFUL_COMPRESSION_COUNT,
RES_SUCCESSFUL_COMPRESSION_COUNT,
REQ_SUCCESSFUL_DECOMPRESSION_COUNT,
RES_SUCCESSFUL_DECOMPRESSION_COUNT,
CORRUPTED_ZIP_SKIPPED_SESSION_COUNT,
CPU_USAGE,
AVERAGE_VM_MEMORY_USAGE,
AVERAGE_RSS_MEMORY_USAGE,
MAX_VM_MEMORY_USAGE,
MAX_RSS_MEMORY_USAGE,
REQUEST_OVERALL_SIZE_COUNT,
RESPONSE_OVERALL_SIZE_COUNT,
METRIC_TYPES_COUNT
} ngx_http_plugin_metric_type_e;
#ifdef __cplusplus
typedef enum class ngx_http_cp_verdict
#else
typedef enum ngx_http_cp_verdict
#endif
{
TRAFFIC_VERDICT_INSPECT,
TRAFFIC_VERDICT_ACCEPT,
TRAFFIC_VERDICT_DROP,
TRAFFIC_VERDICT_INJECT,
TRAFFIC_VERDICT_IRRELEVANT,
TRAFFIC_VERDICT_RECONF,
TRAFFIC_VERDICT_WAIT,
LIMIT_RESPONSE_HEADERS
} ngx_http_cp_verdict_e;
#ifdef __cplusplus
typedef enum class ngx_http_cp_debug_level
#else
typedef enum ngx_http_cp_debug_level
#endif
{
DBG_LEVEL_TRACE,
DBG_LEVEL_DEBUG,
DBG_LEVEL_INFO,
DBG_LEVEL_WARNING,
DBG_LEVEL_ERROR,
#ifndef __cplusplus
DBG_LEVEL_ASSERT,
#endif
DBG_LEVEL_COUNT
} ngx_http_cp_debug_level_e;
#ifdef __cplusplus
typedef enum class ngx_http_meta_data
#else
typedef enum ngx_http_meta_data
#endif
{
HTTP_PROTOCOL_SIZE,
HTTP_PROTOCOL_DATA,
HTTP_METHOD_SIZE,
HTTP_METHOD_DATA,
HOST_NAME_SIZE,
HOST_NAME_DATA,
LISTENING_ADDR_SIZE,
LISTENING_ADDR_DATA,
LISTENING_PORT,
URI_SIZE,
URI_DATA,
CLIENT_ADDR_SIZE,
CLIENT_ADDR_DATA,
CLIENT_PORT,
PARSED_HOST_SIZE,
PARSED_HOST_DATA,
PARSED_URI_SIZE,
PARSED_URI_DATA,
WAF_TAG_SIZE,
WAF_TAG_DATA,
META_DATA_COUNT
} ngx_http_meta_data_e;
#ifdef __cplusplus
typedef enum class ngx_http_header_data
#else
typedef enum ngx_http_header_data
#endif
{
HEADER_KEY_SIZE,
HEADER_KEY_DATA,
HEADER_VAL_SIZE,
HEADER_VAL_DATA,
HEADER_DATA_COUNT
} ngx_http_header_data_e;
typedef enum ngx_http_inspection_mode
{
NON_BLOCKING_THREAD,
BLOCKING_THREAD,
NO_THREAD,
INSPECTION_MODE_COUNT
} ngx_http_inspection_mode_e;
#ifdef __cplusplus
typedef enum class ngx_web_response_type
#else
typedef enum ngx_web_response_type
#endif
{
CUSTOM_WEB_RESPONSE,
CUSTOM_WEB_BLOCK_PAGE_RESPONSE,
RESPONSE_CODE_ONLY,
REDIRECT_WEB_RESPONSE,
NO_WEB_RESPONSE
} ngx_web_response_type_e;
typedef struct __attribute__((__packed__)) ngx_http_cp_inject_data {
ngx_http_cp_inject_pos_t injection_pos;
ngx_http_modification_type_e mod_type;
uint16_t injection_size;
uint8_t is_header;
uint8_t orig_buff_index;
char data[0];
} ngx_http_cp_inject_data_t;
typedef struct __attribute__((__packed__)) ngx_http_cp_web_response_data {
uint8_t web_repsonse_type;
uint8_t uuid_size;
union {
struct __attribute__((__packed__)) ngx_http_cp_custom_web_response_data {
uint16_t response_code;
uint8_t title_size;
uint8_t body_size;
char data[0];
} custom_response_data;
struct __attribute__((__packed__)) ngx_http_cp_redirect_data {
uint8_t unused_dummy;
uint8_t add_event_id;
uint16_t redirect_location_size;
char redirect_location[0];
} redirect_data;
} response_data;
} ngx_http_cp_web_response_data_t;
static_assert(
sizeof(((ngx_http_cp_web_response_data_t*)0)->response_data.custom_response_data) ==
sizeof(((ngx_http_cp_web_response_data_t*)0)->response_data.redirect_data),
"custom_response_data must be equal to redirect_data in size"
);
typedef union __attribute__((__packed__)) ngx_http_cp_modify_data {
ngx_http_cp_inject_data_t inject_data[0];
ngx_http_cp_web_response_data_t web_response_data[0];
} ngx_http_cp_modify_data_t;
typedef struct __attribute__((__packed__)) ngx_http_cp_reply_from_service {
uint16_t verdict;
uint32_t session_id;
uint8_t modification_count;
ngx_http_cp_modify_data_t modify_data[0];
} ngx_http_cp_reply_from_service_t;
typedef struct __attribute__((__packed__)) ngx_http_cp_request_data {
uint16_t data_type;
uint32_t session_id;
unsigned char data[0];
} ngx_http_cp_request_data_t;
typedef struct __attribute__((__packed__)) ngx_http_cp_metric_data {
uint16_t data_type;
#ifdef __cplusplus
uint64_t data[static_cast<int>(ngx_http_plugin_metric_type::METRIC_TYPES_COUNT)];
#else
uint64_t data[METRIC_TYPES_COUNT];
#endif
} ngx_http_cp_metric_data_t;
#endif // __NGINX_ATTACHMENT_COMMON_H__

9
core/include/attachments/nginx_attachment_util.h
View File

@@ -17,7 +17,7 @@
#include <stdio.h>
#include "nginx_attachment_common.h"
#include "nano_attachment_common.h"
#ifdef __cplusplus
extern "C" {
@@ -29,7 +29,7 @@ typedef const char * c_str;
int initAttachmentConfig(c_str conf_file);
ngx_http_inspection_mode_e getInspectionMode();
NanoHttpInspectionMode getInspectionMode();
unsigned int getNumOfNginxIpcElements();
unsigned int getKeepAliveIntervalMsec();
unsigned int getDbgLevel();
@@ -61,11 +61,16 @@ unsigned int getMinRetriesForVerdict();
unsigned int getMaxRetriesForVerdict();
unsigned int getReqBodySizeTrigger();
unsigned int getRemoveResServerHeader();
unsigned int getDecompressionPoolSize();
unsigned int getRecompressionPoolSize();
unsigned int getIsBrotliInspectionEnabled();
unsigned int getWaitingForVerdictThreadTimeout();
int isIPAddress(c_str ip_str);
int isSkipSource(c_str ip_str);
unsigned int isPairedAffinityEnabled();
unsigned int isAsyncModeEnabled();
#ifdef __cplusplus
}

41
core/shmem_ipc/shmem_ipc.c
View File

@@ -199,6 +199,11 @@ void
resetIpc(SharedMemoryIPC *ipc, uint16_t num_of_data_segments)
{
writeDebug(TraceLevel, "Reseting IPC queues\n");
if (!ipc || !ipc->rx_queue || !ipc->tx_queue) {
writeDebug(WarningLevel, "resetIpc called with NULL ipc pointer\n");
return;
}
resetRingQueue(ipc->rx_queue, num_of_data_segments);
resetRingQueue(ipc->tx_queue, num_of_data_segments);
}
@@ -208,6 +213,11 @@ destroyIpc(SharedMemoryIPC *shmem, int is_owner)
{
writeDebug(TraceLevel, "Destroying IPC queues\n");
if (!shmem) {
writeDebug(WarningLevel, "Destroying IPC queues called with NULL shmem pointer\n");
return;
}
if (shmem->rx_queue != NULL) {
destroySharedRingQueue(shmem->rx_queue, is_owner, isTowardsOwner(is_owner, 0));
shmem->rx_queue = NULL;
@@ -225,6 +235,10 @@ dumpIpcMemory(SharedMemoryIPC *ipc)
{
writeDebug(WarningLevel, "Ipc memory dump:\n");
writeDebug(WarningLevel, "RX queue:\n");
if (!ipc || !ipc->rx_queue) {
writeDebug(WarningLevel, "RX queue is NULL\n");
return;
}
dumpRingQueueShmem(ipc->rx_queue);
writeDebug(WarningLevel, "TX queue:\n");
dumpRingQueueShmem(ipc->tx_queue);
@@ -234,6 +248,10 @@ int
sendData(SharedMemoryIPC *ipc, const uint16_t data_to_send_size, const char *data_to_send)
{
writeDebug(TraceLevel, "Sending data of size %u\n", data_to_send_size);
if (!ipc || !ipc->tx_queue) {
writeDebug(WarningLevel, "sendData called with NULL ipc pointer\n");
return -1;
}
return pushToQueue(ipc->tx_queue, data_to_send, data_to_send_size);
}
@@ -247,12 +265,22 @@ sendChunkedData(
{
writeDebug(TraceLevel, "Sending %u chunks of data\n", num_of_data_elem);
if (!ipc) {
writeDebug(WarningLevel, "sendChunkedData called with NULL ipc pointer\n");
return -1;
}
return pushBuffersToQueue(ipc->tx_queue, data_elem_to_send, data_to_send_sizes, num_of_data_elem);
}
int
receiveData(SharedMemoryIPC *ipc, uint16_t *received_data_size, const char **received_data)
{
if (!ipc) {
writeDebug(WarningLevel, "receiveData called with NULL ipc pointer\n");
return -1;
}
int res = peekToQueue(ipc->rx_queue, received_data, received_data_size);
writeDebug(TraceLevel, "Received data from queue. Res: %d, data size: %u\n", res, *received_data_size);
return res;
@@ -261,6 +289,10 @@ receiveData(SharedMemoryIPC *ipc, uint16_t *received_data_size, const char **rec
int
popData(SharedMemoryIPC *ipc)
{
if (!ipc) {
writeDebug(WarningLevel, "popData called with NULL ipc pointer\n");
return -1;
}
int res = popFromQueue(ipc->rx_queue);
writeDebug(TraceLevel, "Popped data from queue. Res: %d\n", res);
return res;
@@ -269,6 +301,10 @@ popData(SharedMemoryIPC *ipc)
int
isDataAvailable(SharedMemoryIPC *ipc)
{
if (!ipc) {
writeDebug(WarningLevel, "isDataAvailable called with NULL ipc pointer\n");
return 0;
}
int res = !isQueueEmpty(ipc->rx_queue);
writeDebug(TraceLevel, "Checking if there is data pending to be read. Res: %d\n", res);
return res;
@@ -277,6 +313,11 @@ isDataAvailable(SharedMemoryIPC *ipc)
int
isCorruptedShmem(SharedMemoryIPC *ipc, int is_owner)
{
if (!ipc) {
writeDebug(WarningLevel, "isCorruptedShmem called with NULL ipc pointer\n");
return 1;
}
if (isCorruptedQueue(ipc->rx_queue, isTowardsOwner(is_owner, 0)) ||
isCorruptedQueue(ipc->tx_queue, isTowardsOwner(is_owner, 1))
) {

29
core/shmem_ipc_2/shmem_ipc.c
View File

@@ -297,6 +297,9 @@ initIpc(
void
resetIpc(SharedMemoryIPC *ipc, uint16_t num_of_data_segments)
{
if (!ipc || !ipc->rx_queue || !ipc->tx_queue) {
return;
}
writeDebug(&(ipc->logging_data), TraceLevel, "Reseting IPC queues\n");
resetRingQueue(&(ipc->logging_data), ipc->rx_queue, num_of_data_segments);
resetRingQueue(&(ipc->logging_data), ipc->tx_queue, num_of_data_segments);
@@ -335,6 +338,9 @@ destroyIpc(SharedMemoryIPC *shmem, int is_owner)
void
dumpIpcMemory(SharedMemoryIPC *ipc)
{
if (!ipc) {
return;
}
writeDebug(&(ipc->logging_data), WarningLevel, "Ipc memory dump:\n");
writeDebug(&(ipc->logging_data), WarningLevel, "RX queue:\n");
dumpRingQueueShmem(&(ipc->logging_data), ipc->rx_queue);
@@ -345,6 +351,9 @@ dumpIpcMemory(SharedMemoryIPC *ipc)
int
sendData(SharedMemoryIPC *ipc, const uint16_t data_to_send_size, const char *data_to_send)
{
if (!ipc || !ipc->tx_queue) {
return -1;
}
writeDebug(&(ipc->logging_data), TraceLevel, "Sending data of size %u\n", data_to_send_size);
return pushToQueue(&(ipc->logging_data), ipc->tx_queue, ipc->global_data, data_to_send, data_to_send_size);
}
@@ -357,6 +366,10 @@ sendChunkedData(
const uint8_t num_of_data_elem
)
{
if (!ipc || !ipc->tx_queue) {
return -1;
}
writeDebug(&(ipc->logging_data), TraceLevel, "Sending %u chunks of data\n", num_of_data_elem);
return pushBuffersToQueue(
@@ -372,6 +385,10 @@ sendChunkedData(
int
receiveData(SharedMemoryIPC *ipc, uint16_t *received_data_size, const char **received_data)
{
if (!ipc) {
return -1;
}
int res = peekToQueue(&(ipc->logging_data), ipc->rx_queue, ipc->global_data, received_data, received_data_size);
writeDebug(
&(ipc->logging_data),
@@ -386,6 +403,10 @@ receiveData(SharedMemoryIPC *ipc, uint16_t *received_data_size, const char **rec
int
popData(SharedMemoryIPC *ipc)
{
if (!ipc || !ipc->rx_queue) {
return -1;
}
int res = popFromQueue(&(ipc->logging_data), ipc->rx_queue, ipc->global_data);
writeDebug(&(ipc->logging_data), TraceLevel, "Popped data from queue. Res: %d\n", res);
return res;
@@ -394,6 +415,10 @@ popData(SharedMemoryIPC *ipc)
int
isDataAvailable(SharedMemoryIPC *ipc)
{
if (!ipc || !ipc->rx_queue) {
return 0;
}
int res = !isQueueEmpty(ipc->rx_queue);
writeDebug(&(ipc->logging_data), TraceLevel, "Checking if there is data pending to be read. Res: %d\n", res);
return res;
@@ -402,6 +427,10 @@ isDataAvailable(SharedMemoryIPC *ipc)
int
isCorruptedShmem(SharedMemoryIPC *ipc, int is_owner)
{
if (!ipc) {
return 1;
}
if (isCorruptedQueue(&(ipc->logging_data), ipc->rx_queue, ipc->global_data, isTowardsOwner(is_owner, 0)) ||
isCorruptedQueue(&(ipc->logging_data), ipc->tx_queue, ipc->global_data, isTowardsOwner(is_owner, 1))
) {