diff options
Diffstat (limited to 'examples/c-captured/c-captured.c')
-rw-r--r-- | examples/c-captured/c-captured.c | 1391 |
1 files changed, 1391 insertions, 0 deletions
diff --git a/examples/c-captured/c-captured.c b/examples/c-captured/c-captured.c new file mode 100644 index 000000000..98df7c4b8 --- /dev/null +++ b/examples/c-captured/c-captured.c @@ -0,0 +1,1391 @@ +#include <arpa/inet.h> +#include <errno.h> +#include <linux/limits.h> +#include <netinet/ip_icmp.h> +#include <netinet/tcp.h> +#include <netinet/udp.h> +#include <pcap/pcap.h> +#include <signal.h> +#include <stdbool.h> // ndpi_typedefs.h +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/socket.h> +#include <sys/stat.h> +#include <sys/types.h> +#include <time.h> +#include <unistd.h> + +#include <ndpi_typedefs.h> +#include <ndpi_api.h> + +#include "nDPIsrvd.h" +#include "utarray.h" +#include "utils.h" + +// #define VERBOSE +#define DEFAULT_DATADIR "/tmp/nDPId-captured" + +struct packet_data +{ + nDPIsrvd_ull error_event_id; + nDPIsrvd_ull packet_id; + nDPIsrvd_ull packet_datalink; + nDPIsrvd_ull packet_ts_sec; + nDPIsrvd_ull packet_ts_usec; + nDPIsrvd_ull packet_len; + int base64_packet_size; + union + { + char * base64_packet; + char const * base64_packet_const; + }; +}; + +struct flow_packet_data +{ + nDPIsrvd_ull packet_ts_sec; + nDPIsrvd_ull packet_ts_usec; + nDPIsrvd_ull packet_len; + int base64_packet_size; + union + { + char * base64_packet; + char const * base64_packet_const; + }; +}; + +struct global_user_data +{ + UT_array * packets; // packet_data +}; + +struct flow_user_data +{ + uint8_t new_seen : 1; + uint8_t detection_finished : 1; + uint8_t guessed : 1; + uint8_t detected : 1; + uint8_t risky : 1; + uint8_t midstream : 1; + nDPIsrvd_ull flow_datalink; + nDPIsrvd_ull flow_max_packets; + nDPIsrvd_ull flow_tot_l4_payload_len; + UT_array * packets; // flow_packet_data +}; + +static struct nDPIsrvd_socket * ndpisrvd_socket = NULL; +static int main_thread_shutdown = 0; + +static char * pidfile = NULL; +static char * serv_optarg = NULL; +static nDPIsrvd_ull pcap_filename_rotation = 0; +static time_t pcap_filename_last_rotation = 0; +static struct tm pcap_filename_last_rotation_tm = {}; +static char * user = NULL; +static char * group = NULL; +static uint8_t logging_mode = 0; +static uint8_t capture_mode = 0; +static char * datadir = NULL; +static uint8_t process_guessed = 0; +static uint8_t process_undetected = 0; +static ndpi_risk process_risky = NDPI_NO_RISK; +static uint8_t process_midstream = 0; +static uint8_t ignore_empty_flows = 0; + +size_t const max_packet_len = 65535; + +#ifdef ENABLE_MEMORY_PROFILING +void nDPIsrvd_memprof_log_alloc(size_t alloc_size) +{ + (void)alloc_size; +} + +void nDPIsrvd_memprof_log_free(size_t free_size) +{ + (void)free_size; +} + +void nDPIsrvd_memprof_log(char const * const format, ...) +{ + va_list ap; + + va_start(ap, format); + fprintf(stderr, "%s", "nDPIsrvd MemoryProfiler: "); + vfprintf(stderr, format, ap); + fprintf(stderr, "%s\n", ""); + va_end(ap); +} +#endif + +static int pcap_open_or_append(int packet_datalink, + const char * const filename, + pcap_t ** const p, + pcap_dumper_t ** const pd) +{ + *p = pcap_open_dead(packet_datalink, max_packet_len); + if (*p == NULL) + { + return 1; + } + + if (access(filename, F_OK) == 0) + { + *pd = pcap_dump_open_append(*p, filename); + } + else + { + *pd = pcap_dump_open(*p, filename); + } + + if (*pd == NULL) + { + logger(1, "pcap error %s", pcap_geterr(*p)); + pcap_close(*p); + return 1; + } + + return 0; +} + +static void decode_base64(pcap_dumper_t * const pd, + struct packet_data * const pd_elt, + struct flow_packet_data * const fd_elt) +{ + char const * base64_packet; + size_t base64_packet_size; + nDPIsrvd_ull packet_ts_sec, packet_ts_usec; + unsigned char pkt_buf[max_packet_len]; + size_t pkt_buf_len = sizeof(pkt_buf); + + if (pd_elt != NULL) + { + base64_packet = pd_elt->base64_packet; + base64_packet_size = pd_elt->base64_packet_size; + packet_ts_sec = pd_elt->packet_ts_sec; + packet_ts_usec = pd_elt->packet_ts_usec; + } + else if (fd_elt != NULL) + { + base64_packet = fd_elt->base64_packet; + base64_packet_size = fd_elt->base64_packet_size; + packet_ts_sec = fd_elt->packet_ts_sec; + packet_ts_usec = fd_elt->packet_ts_usec; + } + else + { + logger(1, "%s", "BUG: Can not decode base64 packet."); + return; + } + + if (nDPIsrvd_base64decode(base64_packet, base64_packet_size, pkt_buf, &pkt_buf_len) != 0 || pkt_buf_len == 0) + { + logger(1, "packet base64 decode failed (%zu bytes): %s", base64_packet_size, base64_packet); + } + else + { + struct pcap_pkthdr phdr; + phdr.ts.tv_sec = packet_ts_sec; + phdr.ts.tv_usec = packet_ts_usec; + phdr.caplen = pkt_buf_len; + phdr.len = pkt_buf_len; + pcap_dump((unsigned char *)pd, &phdr, pkt_buf); + } +} + +static void packet_data_copy(void * dst, const void * src) +{ + struct packet_data * const pd_dst = (struct packet_data *)dst; + struct packet_data const * const pd_src = (struct packet_data *)src; + *pd_dst = *pd_src; + if (pd_src->base64_packet != NULL && pd_src->base64_packet_size > 0) + { + pd_dst->base64_packet = strndup(pd_src->base64_packet, pd_src->base64_packet_size); + } + else + { + pd_dst->base64_packet = NULL; + pd_dst->base64_packet_size = 0; + } +} + +static void packet_data_dtor(void * elt) +{ + struct packet_data * const pd_elt = (struct packet_data *)elt; + if (pd_elt->base64_packet != NULL) + { + free(pd_elt->base64_packet); + pd_elt->base64_packet = NULL; + pd_elt->base64_packet_size = 0; + } +} + +static void flow_packet_data_copy(void * dst, const void * src) +{ + struct flow_packet_data * const pd_dst = (struct flow_packet_data *)dst; + struct flow_packet_data const * const pd_src = (struct flow_packet_data *)src; + *pd_dst = *pd_src; + if (pd_src->base64_packet != NULL && pd_src->base64_packet_size > 0) + { + pd_dst->base64_packet = strndup(pd_src->base64_packet, pd_src->base64_packet_size); + } + else + { + pd_dst->base64_packet = NULL; + pd_dst->base64_packet_size = 0; + } +} + +static void flow_packet_data_dtor(void * elt) +{ + struct flow_packet_data * const pd_elt = (struct flow_packet_data *)elt; + if (pd_elt->base64_packet != NULL) + { + free(pd_elt->base64_packet); + pd_elt->base64_packet = NULL; + pd_elt->base64_packet_size = 0; + } +} + +static const UT_icd packet_data_icd = {sizeof(struct packet_data), NULL, packet_data_copy, packet_data_dtor}; +static const UT_icd flow_packet_data_icd = {sizeof(struct flow_packet_data), + NULL, + flow_packet_data_copy, + flow_packet_data_dtor}; + +static int utarray_packets_init(struct global_user_data * const global_user) +{ + if (capture_mode == 0) + { + return -1; + } + + if (global_user->packets == NULL) + { + utarray_new(global_user->packets, &packet_data_icd); + } + + return global_user->packets != NULL; +} + +static void utarray_packets_free(struct global_user_data * const global_user) +{ + if (capture_mode == 0) + { + return; + } + + if (global_user->packets != NULL) + { + utarray_free(global_user->packets); + global_user->packets = NULL; + } +} + +static int utarray_packets_add(struct global_user_data * const global_user, struct packet_data const * const pd) +{ + if (capture_mode == 0) + { + return -1; + } + + unsigned int array_len = utarray_len(global_user->packets); + + utarray_push_back(global_user->packets, pd); + + return utarray_len(global_user->packets) == array_len + 1u; +} + +static struct packet_data * utarray_packets_get(struct global_user_data * const global_user, nDPIsrvd_ull packet_id) +{ + if (capture_mode == 0) + { + return NULL; + } + + for (size_t i = 0; i < utarray_len(global_user->packets); ++i) + { + struct packet_data * const pd_elt = (struct packet_data *)utarray_eltptr(global_user->packets, i); + + if (pd_elt->packet_id == packet_id) + { + return pd_elt; + } + } + + return NULL; +} + +static int utarray_flow_packets_init(struct flow_user_data * const flow_user) +{ + if (capture_mode == 0) + { + return -1; + } + + if (flow_user->packets == NULL) + { + utarray_new(flow_user->packets, &flow_packet_data_icd); + } + + return flow_user->packets != NULL; +} + +static void utarray_flow_packets_free(struct flow_user_data * const flow_user) +{ + if (capture_mode == 0) + { + return; + } + + if (flow_user->packets != NULL) + { + utarray_free(flow_user->packets); + flow_user->packets = NULL; + } +} + +static int utarray_flow_packets_add(struct flow_user_data * const flow_user, struct flow_packet_data const * const pd) +{ + if (capture_mode == 0) + { + return -1; + } + + unsigned int array_len = utarray_len(flow_user->packets); + + utarray_push_back(flow_user->packets, pd); + + return utarray_len(flow_user->packets) == array_len + 1u; +} + +static void set_ndpi_risk(ndpi_risk * const risk, nDPIsrvd_ull risk_to_add) +{ + if (risk_to_add == 0) + { + *risk = (ndpi_risk)-1; + } + else + { + *risk |= 1ull << --risk_to_add; + } +} + +static void unset_ndpi_risk(ndpi_risk * const risk, nDPIsrvd_ull risk_to_del) +{ + if (risk_to_del == 0) + { + *risk = 0; + } + else + { + *risk &= ~(1ull << --risk_to_del); + } +} + +static int has_ndpi_risk(ndpi_risk * const risk, nDPIsrvd_ull risk_to_check) +{ + return (*risk & (1ull << --risk_to_check)) != 0; +} + +static int generate_pcap_appendix(char * const appendix, size_t appendix_size) +{ + if (pcap_filename_rotation > 0) + { + time_t current_time = time(NULL); + + if (current_time >= pcap_filename_last_rotation + (time_t)pcap_filename_rotation) + { + pcap_filename_last_rotation = current_time; + if (localtime_r(&pcap_filename_last_rotation, &pcap_filename_last_rotation_tm) == NULL) + { + return 1; + } + } + + if (strftime(appendix, appendix_size, "-%d_%m_%y-%H_%M_%S", &pcap_filename_last_rotation_tm) == 0) + { + return 1; + } + } + + return 0; +} + +static char * packet_generate_pcap_filename(char * const dest, size_t size, nDPIsrvd_ull packet_datalink) +{ + char appendix[32] = {}; + + if (generate_pcap_appendix(appendix, sizeof(appendix)) != 0) + { + return NULL; + } + + int ret = snprintf(dest, size, "%s/packet%s-dlt%llu.pcap", datadir, appendix, packet_datalink); + if (ret <= 0 || (size_t)ret > size) + { + return NULL; + } + + return dest; +} + +static char * flow_generate_pcap_filename(struct flow_user_data const * const flow_user, char * const dest, size_t size) +{ + char appendix[32] = {}; + + if (generate_pcap_appendix(appendix, sizeof(appendix)) != 0) + { + return NULL; + } + + if (flow_user->guessed != 0 || flow_user->detected == 0 || flow_user->risky != 0 || flow_user->midstream != 0) + { + char const * flow_type = NULL; + + if (flow_user->midstream != 0) + { + flow_type = "midstream"; + } + else if (flow_user->guessed != 0) + { + flow_type = "guessed"; + } + else if (flow_user->detected == 0) + { + flow_type = "undetected"; + } + else if (flow_user->risky != 0) + { + flow_type = "risky"; + } + else + { + flow_type = "unknown-type"; + } + + int ret = snprintf(dest, size, "%s/flow-%s%s.pcap", datadir, flow_type, appendix); + if (ret <= 0 || (size_t)ret > size) + { + return NULL; + } + } + else + { + return NULL; + } + + return dest; +} + +static int packet_write_pcap_file(struct global_user_data const * const global_user) +{ + UT_array * const pd_array = global_user->packets; + + if (utarray_len(pd_array) == 0) + { + logger(0, "Can not dump packets to pcap; packet array empty"); + return 1; + } + + while (utarray_len(pd_array) > 0) + { + struct packet_data * const pd_elt = (struct packet_data *)utarray_front(pd_array); + if (pd_elt == NULL) + { + return 1; + } + nDPIsrvd_ull packet_datalink = pd_elt->packet_datalink; + + char filename[PATH_MAX]; + if (packet_generate_pcap_filename(filename, sizeof(filename), packet_datalink) == NULL) + { + logger(1, "%s", "Internal error. Could not generate PCAP filename, exit .."); + return 1; + } + + pcap_t * p = NULL; + pcap_dumper_t * pd = NULL; + if (pcap_open_or_append(packet_datalink, filename, &p, &pd) != 0) + { + logger(1, "Can not dump packets to pcap; file open/append failed"); + return 1; + } + + for (size_t i = 0; i < utarray_len(pd_array); ++i) + { + struct packet_data * const pd_elt_dmp = (struct packet_data *)utarray_eltptr(pd_array, i); + if (pd_elt_dmp == NULL) + { + return 1; + } + + decode_base64(pd, pd_elt_dmp, NULL); + } +#ifdef VERBOSE + printf("packets dumped to %s\n", filename); +#endif + pcap_dump_close(pd); + pcap_close(p); + + size_t i = 0; + while (utarray_len(pd_array) > 0) + { + struct packet_data * const pd_elt_rm = (struct packet_data *)utarray_eltptr(pd_array, i); + if (pd_elt_rm == NULL) + { + return 1; + } + + if (pd_elt_rm->packet_datalink == packet_datalink) + { +#ifdef VERBOSE + printf("Remove packet index %zu with datalink %llu\n", i, packet_datalink); +#endif + utarray_erase(pd_array, i, 1); + } + else + { + i++; + } + } + } + + return 0; +} + +static int flow_write_pcap_file(struct flow_user_data const * const flow_user, char const * const filename) +{ + UT_array const * const pd_array = flow_user->packets; + int packet_datalink = flow_user->flow_datalink; + pcap_t * p = NULL; + pcap_dumper_t * pd = NULL; + + if (utarray_len(pd_array) == 0) + { + logger(1, "Can not dump flow packets to pcap; flow packet array empty"); + return 0; + } + + if (pcap_open_or_append(packet_datalink, filename, &p, &pd) != 0) + { + logger(1, "Can not dump flow packets to pcap; file open/append failed"); + return 0; + } + + struct flow_packet_data * pd_elt = (struct flow_packet_data *)utarray_front(pd_array); + do + { + if (pd_elt == NULL) + { + break; + } + + decode_base64(pd, NULL, pd_elt); + } while ((pd_elt = (struct flow_packet_data *)utarray_next(pd_array, pd_elt)) != NULL); + + pcap_dump_close(pd); + pcap_close(p); + + return 0; +} + +#ifdef VERBOSE +static void flow_packet_data_print(struct flow_user_data const * const flow_user) +{ + UT_array const * const pd_array = flow_user->packets; + + printf("packet-data array size(): %u\n", pd_array->n); + struct flow_packet_data * pd_elt = (struct flow_packet_data *)utarray_front(pd_array); + do + { + if (pd_elt == NULL) + { + break; + } + printf("\tpacket-data base64 length: %d\n", pd_elt->base64_packet_size); + } while ((pd_elt = (struct flow_packet_data *)utarray_next(pd_array, pd_elt)) != NULL); +} +#else +#define flow_packet_data_print(pd_array) +#endif + +static enum nDPIsrvd_conversion_return perror_ull(enum nDPIsrvd_conversion_return retval, char const * const prefix) +{ + switch (retval) + { + case CONVERSION_OK: + break; + + case CONVERISON_KEY_NOT_FOUND: + logger(1, "%s: Key not found.", prefix); + break; + case CONVERSION_NOT_A_NUMBER: + logger(1, "%s: Not a valid number.", prefix); + break; + case CONVERSION_RANGE_EXCEEDED: + logger(1, "%s: Number too large.", prefix); + break; + + default: + logger(1, "Internal error, invalid conversion return value."); + break; + } + + return retval; +} + +static void log_event(struct nDPIsrvd_socket const * const sock, + struct nDPIsrvd_flow * const flow, + char const * const event_name) +{ + struct nDPIsrvd_json_token const * const src_ip = TOKEN_GET_SZ(sock, "src_ip"); + struct nDPIsrvd_json_token const * const dst_ip = TOKEN_GET_SZ(sock, "dst_ip"); + struct nDPIsrvd_json_token const * const l4_proto = TOKEN_GET_SZ(sock, "l4_proto"); + struct nDPIsrvd_json_token const * const src_port = TOKEN_GET_SZ(sock, "src_port"); + struct nDPIsrvd_json_token const * const dst_port = TOKEN_GET_SZ(sock, "dst_port"); + char const *src_ip_str = NULL, *dst_ip_str = NULL, *l4_proto_str = NULL; + size_t src_ip_len, dst_ip_len, l4_proto_len; + char src_port_str[8] = {}, dst_port_str[8] = {}; + char const * const na = "n/a"; + const int na_len = (int)strlen(na); + + if (src_ip == NULL || dst_ip == NULL) + { + logger(1, "Flow %llu: Missing essential source/destination IP address.", flow->id_as_ull); + } + else + { + src_ip_str = TOKEN_GET_VALUE(sock, src_ip, &src_ip_len); + dst_ip_str = TOKEN_GET_VALUE(sock, dst_ip, &dst_ip_len); + } + if (l4_proto != NULL) + { + l4_proto_str = TOKEN_GET_VALUE(sock, l4_proto, &l4_proto_len); + } + if (src_port != NULL && dst_port != NULL) + { + size_t src_port_len = 0; + size_t dst_port_len = 0; + char const * const tmp_src_port_str = TOKEN_GET_VALUE(sock, src_port, &src_port_len); + char const * const tmp_dst_port_str = TOKEN_GET_VALUE(sock, dst_port, &dst_port_len); + if (tmp_src_port_str != NULL && tmp_dst_port_str != NULL) + { + snprintf(src_port_str, nDPIsrvd_ARRAY_LENGTH(src_port_str), ":%.*s", (int)src_port_len, tmp_src_port_str); + snprintf(dst_port_str, nDPIsrvd_ARRAY_LENGTH(dst_port_str), ":%.*s", (int)dst_port_len, tmp_dst_port_str); + } + } + + if (is_console_logger_enabled() != 0) + { + printf("Flow %llu %s: %.*s %.*s%s -> %.*s%s\n", + flow->id_as_ull, + event_name, + (l4_proto_str != NULL ? (int)l4_proto_len : na_len), + (l4_proto_str != NULL ? l4_proto_str : na), + (src_ip_str != NULL ? (int)src_ip_len : na_len), + (src_ip_str != NULL ? src_ip_str : na), + src_port_str, + (dst_ip_str != NULL ? (int)dst_ip_len : na_len), + (dst_ip_str != NULL ? dst_ip_str : na), + dst_port_str); + } + else + { + logger(0, + "Flow %llu %s: %.*s %.*s%s -> %.*s%s", + flow->id_as_ull, + event_name, + (l4_proto_str != NULL ? (int)l4_proto_len : na_len), + (l4_proto_str != NULL ? l4_proto_str : na), + (src_ip_str != NULL ? (int)src_ip_len : na_len), + (src_ip_str != NULL ? src_ip_str : na), + src_port_str, + (dst_ip_str != NULL ? (int)dst_ip_len : na_len), + (dst_ip_str != NULL ? dst_ip_str : na), + dst_port_str); + } +} + +static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_socket * const sock, + struct nDPIsrvd_instance * const instance, + struct nDPIsrvd_thread_data * const thread_data, + struct nDPIsrvd_flow * const flow) +{ + (void)instance; + (void)thread_data; + + struct global_user_data * const global_user = (struct global_user_data *)&sock->global_user_data[0]; + + if (flow == NULL) + { + nDPIsrvd_ull packet_id = 0ull; + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "packet_id"), &packet_id), "packet_id"); + + if (utarray_packets_init(global_user) == 0) + { + logger(1, "Memory allocation for packet data failed."); + return CALLBACK_ERROR; + } + + if (TOKEN_GET_SZ(sock, "error_event_name") != NULL) + { + logger(1, "Received an error event for packet id %llu.", packet_id); + + if (TOKEN_GET_SZ(sock, "error_event_id") == NULL) + { + logger(1, "Missing error event id."); + return CALLBACK_ERROR; + } + + nDPIsrvd_ull error_event_id = 0; + TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "error_event_id"), &error_event_id); + struct packet_data pd = {.error_event_id = error_event_id, .packet_id = packet_id}; + if (utarray_packets_add(global_user, &pd) == 0) + { + logger(1, "Could not add packet to array with id %llu.", packet_id); + return CALLBACK_ERROR; + } + } + else if (TOKEN_VALUE_EQUALS_SZ(sock, TOKEN_GET_SZ(sock, "packet_event_name"), "packet") != 0) + { + logger(1, "Received an packet event for packet id %llu.", packet_id); + + if (capture_mode != 0) + { + struct packet_data * const pd = utarray_packets_get(global_user, packet_id); + if (pd == NULL) + { + logger(1, "Received an packet event w/o a previous error event for packet id %llu.", packet_id); + return CALLBACK_OK; + } + + if (pd->packet_id != packet_id) + { + logger(1, + "Received a packet event with a different packet id then the one seen in the error event: " + "%llu != %llu.", + packet_id, + pd->packet_id); + return CALLBACK_OK; + } + + struct nDPIsrvd_json_token const * const pkt = TOKEN_GET_SZ(sock, "pkt"); + if (pkt == NULL) + { + logger(1, "%s", "No packet data available."); + logger(1, + "JSON message: '%.*s'", + nDPIsrvd_json_buffer_length(sock), + nDPIsrvd_json_buffer_string(sock)); + return CALLBACK_OK; + } + + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "pkt_datalink"), &pd->packet_datalink), + "pkt_datalink"); + + nDPIsrvd_ull thread_ts_usec = 0ull; + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "thread_ts_usec"), &thread_ts_usec), + "thread_ts_usec"); + + nDPIsrvd_ull pkt_len = 0ull; + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "pkt_caplen"), &pkt_len), "pkt_caplen"); + + pd->packet_ts_sec = thread_ts_usec / (1000 * 1000); + pd->packet_ts_usec = (thread_ts_usec % (1000 * 1000)); + pd->packet_len = pkt_len; + pd->base64_packet_size = nDPIsrvd_get_token_size(sock, pkt); + pd->base64_packet_const = strndup(nDPIsrvd_get_token_value(sock, pkt), pd->base64_packet_size); + + if (packet_write_pcap_file(global_user) != 0) + { + logger(1, "%s", "Could not dump non-flow packet data"); + return CALLBACK_OK; + } + } + + utarray_packets_free(global_user); + } + + return CALLBACK_OK; + } + + struct flow_user_data * const flow_user = (struct flow_user_data *)flow->flow_user_data; + + if (flow_user == NULL || flow_user->detection_finished != 0) + { + return CALLBACK_OK; + } + + if (TOKEN_VALUE_EQUALS_SZ(sock, TOKEN_GET_SZ(sock, "flow_state"), "finished") != 0) + { + flow_user->detection_finished = 1; + } + + if (TOKEN_VALUE_EQUALS_SZ(sock, TOKEN_GET_SZ(sock, "packet_event_name"), "packet-flow") != 0) + { + struct nDPIsrvd_json_token const * const pkt = TOKEN_GET_SZ(sock, "pkt"); + if (pkt == NULL) + { + logger(1, "%s", "No packet data available."); + logger(1, "JSON message: '%.*s'", nDPIsrvd_json_buffer_length(sock), nDPIsrvd_json_buffer_string(sock)); + return CALLBACK_OK; + } + + if (utarray_flow_packets_init(flow_user) == 0) + { + logger(1, "%s", "Memory allocation for captured packets failed."); + return CALLBACK_ERROR; + } + + nDPIsrvd_ull thread_ts_usec = 0ull; + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "thread_ts_usec"), &thread_ts_usec), "thread_ts_usec"); + + nDPIsrvd_ull pkt_len = 0ull; + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "pkt_caplen"), &pkt_len), "pkt_caplen"); + + struct flow_packet_data pd = {.packet_ts_sec = thread_ts_usec / (1000 * 1000), + .packet_ts_usec = (thread_ts_usec % (1000 * 1000)), + .packet_len = pkt_len, + .base64_packet_size = nDPIsrvd_get_token_size(sock, pkt), + .base64_packet_const = nDPIsrvd_get_token_value(sock, pkt)}; + if (utarray_flow_packets_add(flow_user, &pd) == 0) + { + logger(1, "%s", "Memory allocation to add a captured packet failed."); + return CALLBACK_ERROR; + } + } + + { + struct nDPIsrvd_json_token const * const flow_event_name = TOKEN_GET_SZ(sock, "flow_event_name"); + + if (flow_event_name != NULL) + { + nDPIsrvd_ull nmb = 0; + + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "flow_src_tot_l4_payload_len"), &nmb), + "flow_src_tot_l4_payload_len"); + flow_user->flow_tot_l4_payload_len += nmb; + + nmb = 0; + + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "flow_dst_tot_l4_payload_len"), &nmb), + "flow_dst_tot_l4_payload_len"); + flow_user->flow_tot_l4_payload_len += nmb; + } + + if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "new") != 0) + { + flow_user->new_seen = 1; + + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "flow_datalink"), &flow_user->flow_datalink), + "flow_datalink"); + perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "flow_max_packets"), &flow_user->flow_max_packets), + "flow_max_packets"); + if (TOKEN_VALUE_EQUALS_SZ(sock, TOKEN_GET_SZ(sock, "midstream"), "1") != 0) + { + flow_user->midstream = 1; + } + + return CALLBACK_OK; + } + else if (flow_user->new_seen == 0) + { + return CALLBACK_OK; + } + else if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "guessed") != 0) + { + flow_user->guessed = 1; + } + else if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "not-detected") != 0) + { + flow_user->detected = 0; + flow_user->detection_finished = 1; + } + else if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detected") != 0 || + TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detection-update")) + { + 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; + + flow_user->detected = 1; + + if (flow_risk != NULL) + { + 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 && + numeric_risk_value < NDPI_MAX_RISK && has_ndpi_risk(&process_risky, numeric_risk_value) != 0) + { + flow_user->risky = 1; + } + } + } + } + + if (flow_user->detection_finished != 0 && + ((flow_user->guessed != 0 && process_guessed != 0) || + (flow_user->detected == 0 && process_undetected != 0) || (flow_user->risky != 0 && process_risky != 0) || + (flow_user->midstream != 0 && process_midstream != 0))) + { + if (flow_user->guessed != 0 && flow_user->detected != 0) + { + log_event(sock, flow, "BUG: guessed and detected at the same time"); + } + + if (logging_mode != 0) + { + if (flow_user->guessed != 0) + log_event(sock, flow, "guessed"); + if (flow_user->detected == 0) + log_event(sock, flow, "not-detected"); + if (flow_user->risky != 0) + log_event(sock, flow, "risky"); + if (flow_user->midstream != 0) + log_event(sock, flow, "midstream"); + } + + if (flow_user->packets == NULL || flow_user->flow_max_packets == 0 || utarray_len(flow_user->packets) == 0) + { + if (capture_mode != 0) + { + log_event(sock, flow, "No packets captured"); + } + } + else if (capture_mode != 0) + { + flow_packet_data_print(flow_user); + if (ignore_empty_flows == 0 || flow_user->flow_tot_l4_payload_len > 0) + { + char pcap_filename[PATH_MAX]; + if (flow_generate_pcap_filename(flow_user, pcap_filename, sizeof(pcap_filename)) == NULL) + { + log_event(sock, flow, "Internal error. Could not generate PCAP filename, exit .."); + return CALLBACK_ERROR; + } +#ifdef VERBOSE + printf("Flow %llu saved to %s\n", flow->id_as_ull, pcap_filename); +#endif + errno = 0; + if (flow_write_pcap_file(flow_user, pcap_filename) != 0) + { + logger(1, "Could not dump packet data to pcap file %s: %s", pcap_filename, strerror(errno)); + return CALLBACK_OK; + } + } + } + + utarray_flow_packets_free(flow_user); + } + } + + return CALLBACK_OK; +} + +static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_socket const * sock, + struct nDPIsrvd_instance const * instance, + struct nDPIsrvd_thread_data const * thread_data, + struct nDPIsrvd_flow const * flow, + void * user_data) +{ + (void)sock; + (void)instance; + (void)thread_data; + (void)user_data; + + struct flow_user_data const * const flow_user = (struct flow_user_data const *)flow->flow_user_data; + + fprintf(stderr, + "[Flow %4llu][ptr: " +#ifdef __LP64__ + "0x%016llx" +#else + "0x%08lx" +#endif + "][last-seen: %13llu][finished: %u][detected: %u][risky: " + "%u][total-L4-payload-length: " + "%4llu][packets-captured: %u]\n", + flow->id_as_ull, +#ifdef __LP64__ + (unsigned long long int)flow, +#else + (unsigned long int)flow, +#endif + flow->last_seen, + flow_user->detection_finished, + flow_user->detected, + flow_user->risky, + flow_user->flow_tot_l4_payload_len, + flow_user->packets != NULL ? utarray_len(flow_user->packets) : 0); + + logger(0, + "[Flow %4llu][ptr: " +#ifdef __LP64__ + "0x%016llx" +#else + "0x%08lx" +#endif + "][last-seen: %13llu][finished: %u][detected: %u][risky: " + "%u][total-L4-payload-length: " + "%4llu][packets-captured: %u]", + flow->id_as_ull, +#ifdef __LP64__ + (unsigned long long int)flow, +#else + (unsigned long int)flow, +#endif + flow->last_seen, + flow_user->detection_finished, + flow_user->detected, + flow_user->risky, + flow_user->flow_tot_l4_payload_len, + flow_user->packets != NULL ? utarray_len(flow_user->packets) : 0); +} + +static void sighandler(int signum) +{ + if (signum == SIGUSR1) + { + nDPIsrvd_flow_info(ndpisrvd_socket, nDPIsrvd_write_flow_info_cb, NULL); + } + else if (main_thread_shutdown == 0) + { + main_thread_shutdown = 1; + } +} + +static void captured_flow_cleanup_callback(struct nDPIsrvd_socket * const sock, + struct nDPIsrvd_instance * const instance, + struct nDPIsrvd_thread_data * const thread_data, + struct nDPIsrvd_flow * const flow, + enum nDPIsrvd_cleanup_reason reason) +{ + (void)sock; + (void)instance; + (void)thread_data; + (void)reason; + + struct flow_user_data * const ud = (struct flow_user_data *)flow->flow_user_data; + if (ud != NULL && ud->packets != NULL) + { + utarray_free(ud->packets); + ud->packets = NULL; + } +} + +static void print_usage(char const * const arg0) +{ + static char const usage[] = + "Usage: %s " + "[-c] [-d] [-p pidfile] [-s host] [-r rotate-every-n-seconds]\n" + "\t \t[-u user] [-g group] [-D dir] [-G] [-U] [-R risk] [-M]\n\n" + "\t-c\tLog all messages to stdout/stderr instead of syslog.\n" + "\t-d\tForking into background after initialization.\n" + "\t-p\tWrite the daemon PID to the given file path.\n" + "\t-s\tDestination where nDPIsrvd is listening on.\n" + "\t \tCan be either a path to UNIX socket or an IPv4/TCP-Port IPv6/TCP-Port tuple.\n" + "\t-r\tRotate PCAP files every n seconds\n" + "\t-u\tChange user.\n" + "\t-g\tChange group.\n" + "\t-l\tLogging mode: Log events and some flow metadata to syslog.\n" + "\t-L\tCapture mode: Capture and dump packets.\n" + "\t-D\tDatadir - Where to store PCAP files.\n" + "\t-G\tGuessed - Dump guessed flows to a PCAP file.\n" + "\t-U\tUndetected - Dump undetected flows to a PCAP file.\n" + "\t-R\tRisky - Dump risky flows to a PCAP file. See additional help below.\n" + "\t-M\tMidstream - Dump midstream flows to a PCAP file.\n" + "\t-E\tEmpty - Ignore flows w/o any layer 4 payload\n\n" + "\tPossible options for `-R' (can be specified multiple times, processed from left to right, ~ disables a " + "risk):\n" + "\t \tExample: -R0 -R~15 would enable all risks except risk with id 15\n"; + + fprintf(stderr, usage, arg0); +#ifndef LIBNDPI_STATIC + fprintf(stderr, "\t\t%d - %s\n", 0, "Capture all risks"); +#else + fprintf(stderr, "\t\t%d - %s\n\t\t", 0, "Capture all risks"); +#endif + for (int risk = NDPI_NO_RISK + 1; risk < NDPI_MAX_RISK; ++risk) + { +#ifndef LIBNDPI_STATIC + fprintf(stderr, "\t\t%d - %s%s", risk, ndpi_risk2str(risk), (risk == NDPI_MAX_RISK - 1 ? "\n\n" : "\n")); +#else + fprintf(stderr, "%d%s", risk, (risk == NDPI_MAX_RISK - 1 ? "\n" : ",")); +#endif + } +} + +static int parse_options(int argc, char ** argv) +{ + int opt; + + while ((opt = getopt(argc, argv, "hcdp:s:r:u:g:lLD:GUR:ME")) != -1) + { + switch (opt) + { + case 'c': + enable_console_logger(); + break; + case 'd': + daemonize_enable(); + break; + case 'p': + free(pidfile); + pidfile = strdup(optarg); + break; + case 's': + free(serv_optarg); + serv_optarg = strdup(optarg); + break; + case 'r': + if (perror_ull(str_value_to_ull(optarg, &pcap_filename_rotation), "pcap_filename_rotation") != + CONVERSION_OK) + { + fprintf(stderr, "%s: Argument for `-r' is not a number: %s\n", argv[0], optarg); + return 1; + } + break; + case 'u': + free(user); + user = strdup(optarg); + break; + case 'g': + free(group); + group = strdup(optarg); + break; + case 'l': + logging_mode = 1; + break; + case 'L': + capture_mode = 1; + break; + case 'D': + free(datadir); + datadir = strdup(optarg); + break; + case 'G': + process_guessed = 1; + break; + case 'U': + process_undetected = 1; + break; + case 'R': + { + char * value = (optarg[0] == '~' ? optarg + 1 : optarg); + nDPIsrvd_ull risk; + if (perror_ull(str_value_to_ull(value, &risk), "process_risky") != CONVERSION_OK) + { + fprintf(stderr, "%s: Argument for `-R' is not a number: %s\n", argv[0], optarg); + return 1; + } + if (risk >= NDPI_MAX_RISK) + { + fprintf(stderr, "%s: Invalid risk set: %s\n", argv[0], optarg); + return 1; + } + if (optarg[0] == '~') + { + unset_ndpi_risk(&process_risky, risk); + } + else + { + set_ndpi_risk(&process_risky, risk); + } + break; + } + case 'M': + process_midstream = 1; + break; + case 'E': + ignore_empty_flows = 1; + break; + default: + print_usage(argv[0]); + return 1; + } + } + + if (logging_mode == 0 && capture_mode == 0) + { + fprintf(stderr, "%s: Neither `-l' nor `-L' given. See application usage for more information.\n", argv[0]); + return 1; + } + + if (serv_optarg == NULL) + { + serv_optarg = strdup(DISTRIBUTOR_UNIX_SOCKET); + } + + if (nDPIsrvd_setup_address(&ndpisrvd_socket->address, serv_optarg) != 0) + { + fprintf(stderr, "%s: Could not parse address `%s'\n", argv[0], serv_optarg); + return 1; + } + + if (datadir == NULL) + { + datadir = strdup(DEFAULT_DATADIR); + } + + if (process_guessed == 0 && process_undetected == 0 && process_risky == 0 && process_midstream == 0) + { + fprintf(stderr, "%s: Nothing to capture. Use at least one of -G / -U / -R / -M flags.\n", argv[0]); + return 1; + } + + if (optind < argc) + { + fprintf(stderr, "Unexpected argument after options\n\n"); + print_usage(argv[0]); + return 1; + } + + if (capture_mode != 0) + { + errno = 0; + if (datadir[0] != '/') + { + fprintf(stderr, + "%s: PCAP capture directory must be absolut i.e. starting with `/', path given: `%s'\n", + argv[0], + datadir); + return 1; + } + if (mkdir(datadir, S_IRWXU) != 0 && errno != EEXIST) + { + fprintf(stderr, "%s: Could not create directory %s: %s\n", argv[0], datadir, strerror(errno)); + return 1; + } + { + struct stat datadir_stat; + if (stat(datadir, &datadir_stat) != 0 || S_ISDIR(datadir_stat.st_mode) == 0) + { + fprintf(stderr, "%s: Data directory %s is not a directory\n", argv[0], datadir); + return 1; + } + } + } + + return 0; +} + +static int mainloop(void) +{ + enum nDPIsrvd_read_return read_ret = READ_OK; + + while (main_thread_shutdown == 0) + { + read_ret = nDPIsrvd_read(ndpisrvd_socket); + if (errno == EINTR) + { + continue; + } + if (read_ret == READ_TIMEOUT) + { + logger(0, + "No data received during the last %llu second(s).\n", + (long long unsigned int)ndpisrvd_socket->read_timeout.tv_sec); + continue; + } + if (read_ret != READ_OK) + { + logger(1, "Could not read from socket: %s", nDPIsrvd_enum_to_string(read_ret)); + break; + } + + enum nDPIsrvd_parse_return parse_ret = nDPIsrvd_parse_all(ndpisrvd_socket); + if (parse_ret != PARSE_NEED_MORE_DATA) + { + logger(1, "Could not parse json message: %s", nDPIsrvd_enum_to_string(parse_ret)); + break; + } + } + + if (main_thread_shutdown == 0 && read_ret != READ_OK) + { + return 1; + } + + return 0; +} + +int main(int argc, char ** argv) +{ + init_logging("nDPIsrvd-captured"); + + ndpisrvd_socket = nDPIsrvd_socket_init(sizeof(struct global_user_data), + 0, + 0, + sizeof(struct flow_user_data), + captured_json_callback, + NULL, + captured_flow_cleanup_callback); + if (ndpisrvd_socket == NULL) + { + fprintf(stderr, "%s: nDPIsrvd socket memory allocation failed!\n", argv[0]); + return 1; + } + + if (parse_options(argc, argv) != 0) + { + return 1; + } + + logger(0, "Recv buffer size: %u", NETWORK_BUFFER_MAX_SIZE); + logger(0, "Connecting to `%s'..", serv_optarg); + + if (nDPIsrvd_connect(ndpisrvd_socket) != CONNECT_OK) + { + fprintf(stderr, "%s: nDPIsrvd socket connect to %s failed!\n", argv[0], serv_optarg); + nDPIsrvd_socket_free(&ndpisrvd_socket); + return 1; + } + + signal(SIGUSR1, sighandler); + signal(SIGINT, sighandler); + signal(SIGTERM, sighandler); + signal(SIGPIPE, sighandler); + + if (daemonize_with_pidfile(pidfile) != 0) + { + return 1; + } + + if (capture_mode != 0 && chmod_chown(datadir, S_IRWXU | S_IRGRP | S_IXGRP, user, group) != 0) + { + logger(1, "Could not chmod/chown `%s': %s", datadir, strerror(errno)); + return 1; + } + + errno = 0; + if (user != NULL && change_user_group(user, group, pidfile) != 0) + { + if (errno != 0) + { + logger(1, "Change user/group failed: %s", strerror(errno)); + } + else + { + logger(1, "Change user/group failed."); + } + return 1; + } + + if (nDPIsrvd_set_read_timeout(ndpisrvd_socket, 180, 0) != 0) + { + return 1; + } + + int retval = mainloop(); + + utarray_packets_free((struct global_user_data *)ndpisrvd_socket->global_user_data); + nDPIsrvd_socket_free(&ndpisrvd_socket); + daemonize_shutdown(pidfile); + shutdown_logging(); + + return retval; +} |