scottk/nDPId
1
0
Fork 0
mirror of https://github.com/optim-enterprises-bv/nDPId.git synced 2025-11-02 03:07:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

c-analysed: improved feature extraction from "analyse" events

Browse Source 
* c-captured: update detected risks on "detection-update" events
 * c-collectd: added missing flow breed
 * c-collectd: PUTVAL macros are more flexible now

Signed-off-by: Toni Uhlig <matzeton@googlemail.com>
This commit is contained in:
Toni Uhlig
2022-10-01 17:26:02 +02:00
parent 14f6b87551
commit b6060b897e
8 changed files with 214 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
CMakeLists.txt
View File

@@ -275,7 +275,8 @@ target_link_libraries(nDPId-test "${STATIC_LIBNDPI_LIB}" "${pkgcfg_lib_NDPI_ndpi
if(BUILD_EXAMPLES)
add_executable(nDPIsrvd-collectd examples/c-collectd/c-collectd.c)
target_compile_definitions(nDPIsrvd-collectd PRIVATE ${NDPID_DEFS})
target_include_directories(nDPIsrvd-collectd PRIVATE ${NDPID_DEPS_INC})
target_include_directories(nDPIsrvd-collectd PRIVATE
"${STATIC_LIBNDPI_INC}" "${DEFAULT_NDPI_INCLUDE}" "${CMAKE_SOURCE_DIR}" ${NDPID_DEPS_INC})
add_executable(nDPIsrvd-captured examples/c-captured/c-captured.c utils.c)
if(BUILD_NDPI)
@@ -301,10 +302,10 @@ if(BUILD_EXAMPLES)
target_include_directories(nDPIsrvd-simple PRIVATE ${NDPID_DEPS_INC})
if(ENABLE_COVERAGE)
add_dependencies(coverage nDPIsrvd-collectd nDPIsrvd-captured nDPIsrvd-json-dump nDPIsrvd-simple)
add_dependencies(coverage nDPIsrvd-analysed nDPIsrvd-collectd nDPIsrvd-captured nDPIsrvd-json-dump nDPIsrvd-simple)
endif()
install(TARGETS nDPIsrvd-collectd nDPIsrvd-captured nDPIsrvd-json-dump nDPIsrvd-simple DESTINATION bin)
install(TARGETS nDPIsrvd-analysed nDPIsrvd-collectd nDPIsrvd-captured nDPIsrvd-json-dump nDPIsrvd-simple DESTINATION bin)
install(FILES examples/c-collectd/plugin_nDPIsrvd.conf examples/c-collectd/rrdgraph.sh DESTINATION share/nDPId/nDPIsrvd-collectd)
install(DIRECTORY examples/c-collectd/www DESTINATION share/nDPId/nDPIsrvd-collectd)
endif()

11
dependencies/nDPIsrvd.h vendored
View File

@@ -917,15 +917,20 @@ static inline struct nDPIsrvd_json_token const * nDPIsrvd_get_next_token(struct
*next_index = start->token_index;
}
for (int i = *next_index + 2; i < sock->jsmn.tokens_found; i += 2)
for (int i = *next_index + 1; i < sock->jsmn.tokens_found; ++i)
{
if (sock->jsmn.tokens[i].parent != start->token_index)
{
continue;
}
if (sock->jsmn.tokens[i].type != JSMN_STRING && sock->jsmn.tokens[i].type != JSMN_PRIMITIVE)
{
continue;
}
size_t key_len;
char const * const key = nDPIsrvd_jsmn_token_to_string(sock, &sock->jsmn.tokens[i - 1], &key_len);
char const * const key = nDPIsrvd_jsmn_token_to_string(sock, &sock->jsmn.tokens[i], &key_len);
if (key == NULL)
{
break;
@@ -1087,7 +1092,7 @@ static inline int nDPIsrvd_walk_tokens(
jsmntok_t const * const t = &sock->jsmn.tokens[b];
char const * const js = sock->buffer.json_string;
if (depth >= 12)
if (depth >= 16)
{
return 0;
}

5
examples/README.md
View File

@@ -3,6 +3,11 @@
Some ready-2-use/ready-2-extend examples/utils.
All examples are prefixed with their used LANG.
## c-analysed
A feature extractor useful for ML/DL use cases.
It generates CSV files from flow "analyse" events.
## c-captured
A capture daemon suitable for low-resource devices.

183
examples/c-analysed/c-analysed.c
View File

@@ -8,7 +8,8 @@
#include "utils.h"
#define MIN(a, b) (a > b ? b : a)
#define BUFFER_REMAINING(siz) (NETWORK_BUFFER_MAX_SIZE - siz)
#define BUFFER_REMAINING(siz) (NETWORK_BUFFER_MAX_SIZE / 3 - siz)
typedef char csv_buf_t[(NETWORK_BUFFER_MAX_SIZE / 3) + 1];
static int main_thread_shutdown = 0;
static struct nDPIsrvd_socket * sock = NULL;
@@ -113,6 +114,8 @@ static void sighandler(int signum)
struct nDPIsrvd_instance * itmp;
int verification_failed = 0;
fflush(csv_fp);
if (signum == SIGUSR1)
{
nDPIsrvd_flow_info(sock, nDPIsrvd_write_flow_info_cb, NULL);
@@ -141,10 +144,7 @@ static void sighandler(int signum)
}
}
static void csv_buf_add(char csv_buf[NETWORK_BUFFER_MAX_SIZE + 1],
size_t * const csv_buf_used,
char const * const str,
size_t siz_len)
static void csv_buf_add(csv_buf_t buf, size_t * const csv_buf_used, char const * const str, size_t siz_len)
{
size_t len;
@@ -155,7 +155,7 @@ static void csv_buf_add(char csv_buf[NETWORK_BUFFER_MAX_SIZE + 1],
{
return;
}
strncat(csv_buf, str, len);
strncat(buf, str, len);
}
else
{
@@ -165,17 +165,14 @@ static void csv_buf_add(char csv_buf[NETWORK_BUFFER_MAX_SIZE + 1],
*csv_buf_used += len;
if (BUFFER_REMAINING(*csv_buf_used) > 0)
{
csv_buf[*csv_buf_used] = ',';
buf[*csv_buf_used] = ',';
(*csv_buf_used)++;
}
csv_buf[*csv_buf_used] = '\0';
buf[*csv_buf_used] = '\0';
}
static int json_value_to_csv(struct nDPIsrvd_socket * const sock,
char csv_buf[NETWORK_BUFFER_MAX_SIZE + 1],
size_t * const csv_buf_used,
char const * const json_key,
...)
static int json_value_to_csv(
struct nDPIsrvd_socket * const sock, csv_buf_t buf, size_t * const csv_buf_used, char const * const json_key, ...)
{
va_list ap;
nDPIsrvd_hashkey key;
@@ -200,16 +197,13 @@ static int json_value_to_csv(struct nDPIsrvd_socket * const sock,
ret++;
}
csv_buf_add(csv_buf, csv_buf_used, val, val_length);
csv_buf_add(buf, csv_buf_used, val, val_length);
return ret;
}
static int json_array_to_csv(struct nDPIsrvd_socket * const sock,
char csv_buf[NETWORK_BUFFER_MAX_SIZE + 1],
size_t * const csv_buf_used,
char const * const json_key,
...)
static int json_array_to_csv(
struct nDPIsrvd_socket * const sock, csv_buf_t buf, size_t * const csv_buf_used, char const * const json_key, ...)
{
va_list ap;
nDPIsrvd_hashkey key;
@@ -224,23 +218,23 @@ static int json_array_to_csv(struct nDPIsrvd_socket * const sock,
if (token == NULL)
{
ret++;
csv_buf_add(csv_buf, csv_buf_used, NULL, 0);
csv_buf_add(buf, csv_buf_used, NULL, 0);
}
{
struct nDPIsrvd_json_token next = {};
csv_buf_add(csv_buf, csv_buf_used, "\"", 1);
csv_buf[--(*csv_buf_used)] = '\0';
csv_buf_add(buf, csv_buf_used, "\"", 1);
buf[--(*csv_buf_used)] = '\0';
while (nDPIsrvd_token_iterate(sock, token, &next) == 0)
{
size_t val_length = 0;
char const * const val = TOKEN_GET_VALUE(sock, &next, &val_length);
csv_buf_add(csv_buf, csv_buf_used, val, val_length);
csv_buf_add(buf, csv_buf_used, val, val_length);
}
csv_buf[--(*csv_buf_used)] = '\0';
csv_buf_add(csv_buf, csv_buf_used, "\"", 1);
buf[--(*csv_buf_used)] = '\0';
csv_buf_add(buf, csv_buf_used, "\"", 1);
}
return ret;
@@ -251,7 +245,7 @@ static enum nDPIsrvd_callback_return simple_json_callback(struct nDPIsrvd_socket
struct nDPIsrvd_thread_data * const thread_data,
struct nDPIsrvd_flow * const flow)
{
char csv_buf[NETWORK_BUFFER_MAX_SIZE + 1];
csv_buf_t buf;
size_t csv_buf_used = 0;
(void)instance;
@@ -273,88 +267,88 @@ static enum nDPIsrvd_callback_return simple_json_callback(struct nDPIsrvd_socket
return CALLBACK_ERROR;
}
csv_buf[0] = '\0';
buf[0] = '\0';
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_datalink", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "l3_proto", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "src_ip", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "dst_ip", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "l4_proto", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "src_port", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "dst_port", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "flow_datalink", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "l3_proto", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "src_ip", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "dst_ip", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "l4_proto", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "src_port", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "dst_port", NULL);
if (json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_state", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_src_packets_processed", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_dst_packets_processed", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_first_seen", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_src_last_pkt_time", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_dst_last_pkt_time", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_src_min_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_dst_min_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_src_max_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_dst_max_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_src_tot_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "flow_dst_tot_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "midstream", NULL) != 0)
if (json_value_to_csv(sock, buf, &csv_buf_used, "flow_state", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_src_packets_processed", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_dst_packets_processed", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_first_seen", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_src_last_pkt_time", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_dst_last_pkt_time", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_src_min_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_dst_min_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_src_max_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_dst_max_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_src_tot_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "flow_dst_tot_l4_payload_len", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "midstream", NULL) != 0)
{
return CALLBACK_ERROR;
}
if (json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "iat", "min", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "iat", "avg", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "iat", "max", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "iat", "stddev", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "iat", "var", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "iat", "ent", NULL) != 0)
if (json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "iat", "min", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "iat", "avg", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "iat", "max", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "iat", "stddev", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "iat", "var", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "iat", "ent", NULL) != 0)
{
return CALLBACK_ERROR;
}
if (json_array_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "iat", "data", NULL) != 0)
if (json_array_to_csv(sock, buf, &csv_buf_used, "data_analysis", "iat", "data", NULL) != 0)
{
return CALLBACK_ERROR;
}
if (json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "pktlen", "min", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "pktlen", "avg", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "pktlen", "max", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "pktlen", "stddev", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "pktlen", "var", NULL) != 0 ||
json_value_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "pktlen", "ent", NULL) != 0)
if (json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "pktlen", "min", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "pktlen", "avg", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "pktlen", "max", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "pktlen", "stddev", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "pktlen", "var", NULL) != 0 ||
json_value_to_csv(sock, buf, &csv_buf_used, "data_analysis", "pktlen", "ent", NULL) != 0)
{
return CALLBACK_ERROR;
}
if (json_array_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "pktlen", "data", NULL) != 0)
if (json_array_to_csv(sock, buf, &csv_buf_used, "data_analysis", "pktlen", "data", NULL) != 0)
{
return CALLBACK_ERROR;
}
if (json_array_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "bins", "c_to_s", NULL) != 0)
if (json_array_to_csv(sock, buf, &csv_buf_used, "data_analysis", "bins", "c_to_s", NULL) != 0)
{
return CALLBACK_ERROR;
}
if (json_array_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "bins", "s_to_c", NULL) != 0)
if (json_array_to_csv(sock, buf, &csv_buf_used, "data_analysis", "bins", "s_to_c", NULL) != 0)
{
return CALLBACK_ERROR;
}
if (json_array_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "directions", NULL) != 0)
if (json_array_to_csv(sock, buf, &csv_buf_used, "data_analysis", "directions", NULL) != 0)
{
return CALLBACK_ERROR;
}
if (json_array_to_csv(sock, csv_buf, &csv_buf_used, "data_analysis", "entropies", NULL) != 0)
if (json_array_to_csv(sock, buf, &csv_buf_used, "data_analysis", "entropies", NULL) != 0)
{
return CALLBACK_ERROR;
}
json_value_to_csv(sock, csv_buf, &csv_buf_used, "ndpi", "proto", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "ndpi", "proto_id", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "ndpi", "encrypted", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "ndpi", "breed", NULL);
json_value_to_csv(sock, csv_buf, &csv_buf_used, "ndpi", "category", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "ndpi", "proto", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "ndpi", "proto_id", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "ndpi", "encrypted", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "ndpi", "breed", NULL);
json_value_to_csv(sock, buf, &csv_buf_used, "ndpi", "category", NULL);
{
struct nDPIsrvd_json_token const * const token = TOKEN_GET_SZ(sock, "ndpi", "confidence");
struct nDPIsrvd_json_token const * current = NULL;
@@ -362,8 +356,8 @@ static enum nDPIsrvd_callback_return simple_json_callback(struct nDPIsrvd_socket
if (token == NULL)
{
csv_buf_add(csv_buf, &csv_buf_used, NULL, 0);
csv_buf_add(csv_buf, &csv_buf_used, NULL, 0);
csv_buf_add(buf, &csv_buf_used, NULL, 0);
csv_buf_add(buf, &csv_buf_used, NULL, 0);
}
else
{
@@ -373,18 +367,47 @@ static enum nDPIsrvd_callback_return simple_json_callback(struct nDPIsrvd_socket
char const * const key = TOKEN_GET_KEY(sock, current, &key_length);
char const * const value = TOKEN_GET_VALUE(sock, current, &value_length);
csv_buf_add(csv_buf, &csv_buf_used, key, key_length);
csv_buf_add(csv_buf, &csv_buf_used, value, value_length);
csv_buf_add(buf, &csv_buf_used, key, key_length);
csv_buf_add(buf, &csv_buf_used, value, value_length);
}
}
}
if (csv_buf_used > 0 && csv_buf[csv_buf_used - 1] == ',')
{
csv_buf[--csv_buf_used] = '\0';
csv_buf_t risks;
size_t csv_risks_used = 0;
struct nDPIsrvd_json_token const * const flow_risk = TOKEN_GET_SZ(sock, "ndpi", "flow_risk");
struct nDPIsrvd_json_token const * current = NULL;
int next_child_index = -1;
risks[csv_risks_used++] = '"';
risks[csv_risks_used] = '\0';
if (flow_risk != NULL)
{
while ((current = nDPIsrvd_get_next_token(sock, flow_risk, &next_child_index)) != NULL)
{
size_t key_length = 0;
char const * const key = TOKEN_GET_KEY(sock, current, &key_length);
csv_buf_add(risks, &csv_risks_used, key, key_length);
}
}
if (csv_risks_used > 1)
{
risks[csv_risks_used - 1] = '"';
}
else if (BUFFER_REMAINING(csv_risks_used) > 0)
{
risks[csv_risks_used++] = '"';
}
csv_buf_add(buf, &csv_buf_used, risks, csv_risks_used);
}
fprintf(csv_fp, "%.*s\n", (int)csv_buf_used, csv_buf);
if (csv_buf_used > 0 && buf[csv_buf_used - 1] == ',')
{
buf[--csv_buf_used] = '\0';
}
fprintf(csv_fp, "%.*s\n", (int)csv_buf_used, buf);
return CALLBACK_OK;
}
@@ -458,7 +481,7 @@ static int parse_options(int argc, char ** argv)
csv_fp = fopen(csv_outfile, "a+");
if (csv_fp == NULL)
{
fprintf(stderr, "%s: Could not open file `%s' for appending\n", argv[0], csv_outfile);
fprintf(stderr, "%s: Could not open file `%s' for appending: %s\n", argv[0], csv_outfile, strerror(errno));
return 1;
}
@@ -471,7 +494,7 @@ static int parse_options(int argc, char ** argv)
"flow_src_tot_l4_payload_len,flow_dst_tot_l4_payload_len,midstream,iat_min,iat_avg,iat_max,iat_stddev,"
"iat_var,iat_ent,iat_data,pktlen_min,pktlen_avg,pktlen_max,pktlen_stddev,pktlen_var,pktlen_ent,pktlen_"
"data,bins_c_to_s,bins_s_to_c,directions,entropies,proto,proto_id,encrypted,breed,category,"
"confidence_id,confidence\n");
"confidence_id,confidence,risks\n");
}
if (serv_optarg == NULL)

3
examples/c-captured/c-captured.c
View File

@@ -444,7 +444,8 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
flow_user->detected = 0;
flow_user->detection_finished = 1;
}
else if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detected") != 0)
else if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detected") != 0 ||
TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detection-update") != 0)
{
struct nDPIsrvd_json_token const * const flow_risk = TOKEN_GET_SZ(sock, "ndpi", "flow_risk");
struct nDPIsrvd_json_token const * current = NULL;

128
examples/c-collectd/c-collectd.c
View File

@@ -1,3 +1,4 @@
#include <arpa/inet.h>
#include <errno.h>
#include <signal.h>
#include <stdio.h>
@@ -8,6 +9,8 @@
#include <sys/timerfd.h>
#include <unistd.h>
#include <ndpi_typedefs.h>
#include "nDPIsrvd.h"
#define DEFAULT_COLLECTD_EXEC_INST "nDPIsrvd"
@@ -28,6 +31,7 @@ struct flow_user_data
{
nDPIsrvd_ull last_flow_src_l4_payload_len;
nDPIsrvd_ull last_flow_dst_l4_payload_len;
nDPIsrvd_ull detected_risks;
};
static int main_thread_shutdown = 0;
@@ -62,6 +66,7 @@ static struct
uint64_t flow_breed_fun_count;
uint64_t flow_breed_unsafe_count;
uint64_t flow_breed_potentially_dangerous_count;
uint64_t flow_breed_tracker_ads_count;
uint64_t flow_breed_dangerous_count;
uint64_t flow_breed_unrated_count;
uint64_t flow_breed_unknown_count;
@@ -103,6 +108,9 @@ static struct
uint64_t flow_l4_udp_count;
uint64_t flow_l4_icmp_count;
uint64_t flow_l4_other_count;
nDPIsrvd_ull flow_risk_count[NDPI_MAX_RISK];
nDPIsrvd_ull flow_risk_unknown_count;
} collectd_statistics = {};
struct json_stat_map
@@ -128,6 +136,7 @@ static struct json_stat_map const breeds_map[] = {{"Safe", &collectd_statistics.
{"Unsafe", &collectd_statistics.flow_breed_unsafe_count},
{"Potentially Dangerous",
&collectd_statistics.flow_breed_potentially_dangerous_count},
{"Tracker/Ads", &collectd_statistics.flow_breed_tracker_ads_count},
{"Dangerous", &collectd_statistics.flow_breed_dangerous_count},
{"Unrated", &collectd_statistics.flow_breed_unrated_count},
{NULL, &collectd_statistics.flow_breed_unknown_count}};
@@ -322,26 +331,33 @@ static int parse_options(int argc, char ** argv, struct nDPIsrvd_socket * const
}
#ifdef GENERATE_TIMESTAMP
#define COLLECTD_PUTVAL_N_FORMAT(name) "PUTVAL \"%s/exec-%s/gauge-" #name "\" interval=%llu %llu:%llu\n"
#define COLLECTD_PUTVAL_PREFIX "PUTVAL \"%s/exec-%s/gauge-"
#define COLLECTD_PUTVAL_SUFFIX "\" interval=%llu %llu:%llu\n"
#define COLLECTD_PUTVAL_N(value) \
collectd_hostname, instance_name, collectd_interval_ull, (unsigned long long int)now, \
collectd_hostname, instance_name, #value, collectd_interval_ull, (unsigned long long int)now, \
(unsigned long long int)collectd_statistics.value
#define COLLECTD_PUTVAL_N2(name, value) \
collectd_hostname, instance_name, name, collectd_interval_ull, (unsigned long long int)now, \
(unsigned long long int)collectd_statistics.value
#else
#define COLLECTD_PUTVAL_N_FORMAT(name) "PUTVAL \"%s/exec-%s/gauge-" #name "\" interval=%llu N:%llu\n"
#define COLLECTD_PUTVAL_PREFIX "PUTVAL \"%s/exec-%s/gauge-"
#define COLLECTD_PUTVAL_SUFFIX "\" interval=%llu N:%llu\n"
#define COLLECTD_PUTVAL_N(value) \
collectd_hostname, instance_name, collectd_interval_ull, (unsigned long long int)collectd_statistics.value
collectd_hostname, instance_name, #value, collectd_interval_ull, (unsigned long long int)collectd_statistics.value
#define COLLECTD_PUTVAL_N2(name, value) \
collectd_hostname, instance_name, name, collectd_interval_ull, (unsigned long long int)collectd_statistics.value
#endif
#define COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_PREFIX "%s" COLLECTD_PUTVAL_SUFFIX
static void print_collectd_exec_output(void)
{
size_t i;
#ifdef GENERATE_TIMESTAMP
time_t now = time(NULL);
#endif
printf(COLLECTD_PUTVAL_N_FORMAT(flow_new_count) COLLECTD_PUTVAL_N_FORMAT(flow_end_count)
COLLECTD_PUTVAL_N_FORMAT(flow_idle_count) COLLECTD_PUTVAL_N_FORMAT(flow_guessed_count)
COLLECTD_PUTVAL_N_FORMAT(flow_detected_count) COLLECTD_PUTVAL_N_FORMAT(flow_detection_update_count)
COLLECTD_PUTVAL_N_FORMAT(flow_not_detected_count) COLLECTD_PUTVAL_N_FORMAT(flow_src_total_bytes)
COLLECTD_PUTVAL_N_FORMAT(flow_dst_total_bytes) COLLECTD_PUTVAL_N_FORMAT(flow_risky_count),
printf(COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT(),
COLLECTD_PUTVAL_N(flow_new_count),
COLLECTD_PUTVAL_N(flow_end_count),
@@ -354,12 +370,9 @@ static void print_collectd_exec_output(void)
COLLECTD_PUTVAL_N(flow_dst_total_bytes),
COLLECTD_PUTVAL_N(flow_risky_count));
printf(COLLECTD_PUTVAL_N_FORMAT(flow_breed_safe_count) COLLECTD_PUTVAL_N_FORMAT(flow_breed_acceptable_count)
COLLECTD_PUTVAL_N_FORMAT(flow_breed_fun_count) COLLECTD_PUTVAL_N_FORMAT(flow_breed_unsafe_count)
COLLECTD_PUTVAL_N_FORMAT(flow_breed_potentially_dangerous_count)
COLLECTD_PUTVAL_N_FORMAT(flow_breed_dangerous_count)
COLLECTD_PUTVAL_N_FORMAT(flow_breed_unrated_count)
COLLECTD_PUTVAL_N_FORMAT(flow_breed_unknown_count),
printf(COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT(),
COLLECTD_PUTVAL_N(flow_breed_safe_count),
COLLECTD_PUTVAL_N(flow_breed_acceptable_count),
@@ -370,30 +383,16 @@ static void print_collectd_exec_output(void)
COLLECTD_PUTVAL_N(flow_breed_unrated_count),
COLLECTD_PUTVAL_N(flow_breed_unknown_count));
printf(COLLECTD_PUTVAL_N_FORMAT(flow_category_media_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_vpn_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_email_count) COLLECTD_PUTVAL_N_FORMAT(
flow_category_data_transfer_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_web_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_social_network_count) COLLECTD_PUTVAL_N_FORMAT(
flow_category_download_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_game_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_chat_count) COLLECTD_PUTVAL_N_FORMAT(
flow_category_voip_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_database_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_remote_access_count) COLLECTD_PUTVAL_N_FORMAT(
flow_category_cloud_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_network_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_collaborative_count) COLLECTD_PUTVAL_N_FORMAT(
flow_category_rpc_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_streaming_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_system_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_software_update_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_music_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_video_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_shopping_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_productivity_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_file_sharing_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_mining_count)
COLLECTD_PUTVAL_N_FORMAT(flow_category_malware_count)
COLLECTD_PUTVAL_N_FORMAT(
flow_category_advertisment_count)
COLLECTD_PUTVAL_N_FORMAT(
flow_category_unknown_count),
printf(COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT(),
COLLECTD_PUTVAL_N(flow_category_media_count),
COLLECTD_PUTVAL_N(flow_category_vpn_count),
@@ -424,10 +423,9 @@ static void print_collectd_exec_output(void)
COLLECTD_PUTVAL_N(flow_category_advertisment_count),
COLLECTD_PUTVAL_N(flow_category_unknown_count));
printf(COLLECTD_PUTVAL_N_FORMAT(flow_l3_ip4_count) COLLECTD_PUTVAL_N_FORMAT(flow_l3_ip6_count)
COLLECTD_PUTVAL_N_FORMAT(flow_l3_other_count) COLLECTD_PUTVAL_N_FORMAT(flow_l4_tcp_count)
COLLECTD_PUTVAL_N_FORMAT(flow_l4_udp_count) COLLECTD_PUTVAL_N_FORMAT(flow_l4_icmp_count)
COLLECTD_PUTVAL_N_FORMAT(flow_l4_other_count),
printf(COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT() COLLECTD_PUTVAL_N_FORMAT()
COLLECTD_PUTVAL_N_FORMAT(),
COLLECTD_PUTVAL_N(flow_l3_ip4_count),
COLLECTD_PUTVAL_N(flow_l3_ip6_count),
@@ -435,7 +433,15 @@ static void print_collectd_exec_output(void)
COLLECTD_PUTVAL_N(flow_l4_tcp_count),
COLLECTD_PUTVAL_N(flow_l4_udp_count),
COLLECTD_PUTVAL_N(flow_l4_icmp_count),
COLLECTD_PUTVAL_N(flow_l4_other_count));
COLLECTD_PUTVAL_N(flow_l4_other_count),
COLLECTD_PUTVAL_N(flow_risk_unknown_count));
for (i = 0; i < NDPI_MAX_RISK; ++i)
{
char gauge_name[BUFSIZ];
snprintf(gauge_name, sizeof(gauge_name), "flow_risk_%zu_count", i);
printf(COLLECTD_PUTVAL_N_FORMAT(), COLLECTD_PUTVAL_N2(gauge_name, flow_risk_count[i]));
}
memset(&collectd_statistics, 0, sizeof(collectd_statistics));
}
@@ -611,13 +617,43 @@ static enum nDPIsrvd_callback_return collectd_json_callback(struct nDPIsrvd_sock
collectd_statistics.flow_l4_other_count++;
}
}
else if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detected") != 0)
else if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detected") != 0 ||
TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detection-update") != 0 ||
TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "update") != 0)
{
if (TOKEN_GET_SZ(sock, "flow_risk") != NULL)
struct nDPIsrvd_json_token const * const flow_risk = TOKEN_GET_SZ(sock, "ndpi", "flow_risk");
struct nDPIsrvd_json_token const * current = NULL;
int next_child_index = -1;
if (flow_risk != NULL)
{
if (flow_user_data->detected_risks == 0)
{
collectd_statistics.flow_risky_count++;
}
while ((current = nDPIsrvd_get_next_token(sock, flow_risk, &next_child_index)) != NULL)
{
nDPIsrvd_ull numeric_risk_value = (nDPIsrvd_ull)-1;
if (str_value_to_ull(TOKEN_GET_KEY(sock, current, NULL), &numeric_risk_value) == CONVERSION_OK)
{
if ((flow_user_data->detected_risks & (1 << numeric_risk_value)) == 0)
{
if (numeric_risk_value < NDPI_MAX_RISK)
{
collectd_statistics.flow_risk_count[numeric_risk_value]++;
}
else
{
collectd_statistics.flow_risk_unknown_count++;
}
}
flow_user_data->detected_risks |= (1 << numeric_risk_value);
}
}
}
struct nDPIsrvd_json_token const * const breed = TOKEN_GET_SZ(sock, "ndpi", "breed");
collectd_map_token_to_stat(sock, breed, breeds_map, nDPIsrvd_ARRAY_LENGTH(breeds_map));

13
nDPId-test.c
View File

@@ -186,7 +186,7 @@ static int setup_pipe(int pipefd[PIPE_FDS])
static void * nDPIsrvd_mainloop_thread(void * const arg)
{
(void)arg;
int nDPIsrvd_shutdown = 0;
int epollfd = create_evq();
struct remote_desc * mock_json_desc = NULL;
struct remote_desc * mock_test_desc = NULL;
@@ -242,7 +242,7 @@ static void * nDPIsrvd_mainloop_thread(void * const arg)
pthread_mutex_lock(&nDPIsrvd_start_mutex);
while (1)
while (nDPIsrvd_shutdown == 0)
{
int nready = epoll_wait(epollfd, events, events_size, -1);
@@ -258,10 +258,13 @@ static void * nDPIsrvd_mainloop_thread(void * const arg)
{
if ((events[i].events & EPOLLHUP) != 0 || (events[i].events & EPOLLERR) != 0)
{
goto error;
logger(1, "nDPIsrvd distributor %d connection closed", events[i].data.fd);
handle_data_event(epollfd, &events[i]);
nDPIsrvd_shutdown++;
}
else if (handle_data_event(epollfd, &events[i]) != 0)
{
logger(1, "nDPIsrvd data event handler failed for distributor %d", events[i].data.fd);
THREAD_ERROR_GOTO(arg);
}
}
@@ -303,7 +306,7 @@ static enum nDPIsrvd_callback_return update_flow_packets_processed(struct nDPIsr
TOKEN_GET_SZ(sock, "flow_src_packets_processed"), TOKEN_GET_SZ(sock, "flow_dst_packets_processed")};
flow_stats->total_packets_processed = 0;
for (enum nDPId_flow_direction dir = 0; dir < FD_COUNT; ++dir)
for (int dir = 0; dir < FD_COUNT; ++dir)
{
if (flow_total_packets_processed[dir] != NULL)
{
@@ -330,7 +333,7 @@ static enum nDPIsrvd_callback_return update_flow_l4_payload_len(struct nDPIsrvd_
TOKEN_GET_SZ(sock, "flow_src_tot_l4_payload_len"), TOKEN_GET_SZ(sock, "flow_dst_tot_l4_payload_len")};
flow_stats->flow_total_l4_data_len = 0;
for (enum nDPId_flow_direction dir = 0; dir < FD_COUNT; ++dir)
for (int dir = 0; dir < FD_COUNT; ++dir)
{
if (flow_total_l4_payload_len[dir] != NULL)
{

1
packages/openwrt/net/nDPId-testing/Makefile
View File

@@ -102,6 +102,7 @@ define Package/nDPId-testing/install
$(INSTALL_DIR) $(1)/usr/bin
$(INSTALL_BIN) $(PKG_INSTALL_DIR)/usr/bin/nDPId-test $(1)/usr/bin/nDPId-testing-test
$(INSTALL_BIN) $(PKG_INSTALL_DIR)/usr/bin/nDPIsrvd $(1)/usr/bin/nDPIsrvd-testing
$(INSTALL_BIN) $(PKG_INSTALL_DIR)/usr/bin/nDPIsrvd-analysed $(1)/usr/bin/nDPIsrvd-testing-analysed
$(INSTALL_BIN) $(PKG_INSTALL_DIR)/usr/bin/nDPIsrvd-captured $(1)/usr/bin/nDPIsrvd-testing-captured
$(INSTALL_BIN) $(PKG_INSTALL_DIR)/usr/bin/nDPIsrvd-collectd $(1)/usr/bin/nDPIsrvd-testing-collectd
$(INSTALL_BIN) $(PKG_INSTALL_DIR)/usr/bin/nDPIsrvd-json-dump $(1)/usr/bin/nDPIsrvd-testing-json-dump