Finished (mostly) C guessed/undetected flow dumper.

Signed-off-by: Toni Uhlig <matzeton@googlemail.com>

3 files changed, 469 insertions, 69 deletions

diff --git a/dependencies/nDPIsrvd.h b/dependencies/nDPIsrvd.h
index 02d359522..21067a676 100644
--- a/dependencies/nDPIsrvd.h
+++ b/dependencies/nDPIsrvd.h
@@ -2,6 +2,7 @@
 #define NDPISRVD_H 1
 
 #include <arpa/inet.h>
+#include <ctype.h>
 #include <errno.h>
 #include <stdint.h>
 #include <stdio.h>
@@ -15,20 +16,11 @@
 #include "jsmn/jsmn.h"
 #include "uthash.h"
 
-// FIXME: Unify with event enums in nDPId.c
-enum nDPIsrvd_event
-{
-    EVENT_INVALID = 0,
-    EVENT_FLOW,
-    EVENT_PACKET,
-    EVENT_COUNT
-};
-
 struct nDPIsrvd_flow
 {
-    char id[20];
-    void * user_data;
+    char id[24];
     UT_hash_handle hh;
+    uint8_t user_data[0];
 };
 
 struct nDPIsrvd_socket
@@ -74,16 +66,11 @@ struct nDPIsrvd_socket
 
     struct
     {
-        enum nDPIsrvd_event event;
         char const * event_name;
         int event_name_len;
         char const * flow_id;
         int flow_id_len;
     } current;
-
-    struct nDPIsrvd_flow * flows;
-
-    void * user_data;
 };
 
 #define FIRST_ENUM_VALUE 1
@@ -126,7 +113,81 @@ enum nDPIsrvd_callback_return
     CALLBACK_LAST_ENUM_VALUE
 };
 
-typedef enum nDPIsrvd_callback_return (*json_callback)(struct nDPIsrvd_socket * const sock, void * user_data);
+typedef enum nDPIsrvd_callback_return (*json_callback)(struct nDPIsrvd_socket * const sock, void * const user_data);
+
+/* Slightly modified code: https://en.wikibooks.org/wiki/Algorithm_Implementation/Miscellaneous/Base64 */
+#define WHITESPACE 64
+#define EQUALS 65
+#define INVALID 66
+int nDPIsrvd_base64decode(char * in, size_t inLen, unsigned char * out, size_t * outLen)
+{
+    char * end = in + inLen;
+    char iter = 0;
+    uint32_t buf = 0;
+    size_t len = 0;
+
+    /* treat ASCII char 92 '\\' as whitespace because libnDPI escapes all strings by prepending '/' with a '\\' */
+    static const unsigned char d[] = {66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 64, 66, 66, 66, 66, 66, 66, 66, 66, 66,
+                                      66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66,
+                                      66, 66, 66, 62, 66, 66, 66, 63, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 66, 66,
+                                      66, 65, 66, 66, 66, 0,  1,  2,  3,  4,  5,  6,  7,  8,  9,  10, 11, 12, 13, 14,
+                                      15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 66, 64, 66, 66, 66, 66, 26, 27, 28,
+                                      29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48,
+                                      49, 50, 51, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66,
+                                      66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66,
+                                      66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66,
+                                      66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66,
+                                      66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66,
+                                      66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66,
+                                      66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66, 66};
+
+    while (in < end)
+    {
+        unsigned char c = d[*(unsigned char *)in++];
+
+        switch (c)
+        {
+            case WHITESPACE:
+                continue; /* skip whitespace */
+            case INVALID:
+                return 1; /* invalid input, return error */
+            case EQUALS:  /* pad character, end of data */
+                in = end;
+                continue;
+            default:
+                buf = buf << 6 | c;
+                iter++; // increment the number of iteration
+                /* If the buffer is full, split it into bytes */
+                if (iter == 4)
+                {
+                    if ((len += 3) > *outLen)
+                        return 1; /* buffer overflow */
+                    *(out++) = (buf >> 16) & 255;
+                    *(out++) = (buf >> 8) & 255;
+                    *(out++) = buf & 255;
+                    buf = 0;
+                    iter = 0;
+                }
+        }
+    }
+
+    if (iter == 3)
+    {
+        if ((len += 2) > *outLen)
+            return 1; /* buffer overflow */
+        *(out++) = (buf >> 10) & 255;
+        *(out++) = (buf >> 2) & 255;
+    }
+    else if (iter == 2)
+    {
+        if (++len > *outLen)
+            return 1; /* buffer overflow */
+        *(out++) = (buf >> 4) & 255;
+    }
+
+    *outLen = len; /* modify to reflect the actual output size */
+    return 0;
+}
 
 static inline char const * nDPIsrvd_enum_to_string(int enum_value)
 {
@@ -167,15 +228,19 @@ static inline struct nDPIsrvd_socket * nDPIsrvd_init(void)
     return sock;
 }
 
-static inline void nDPIsrvd_free(struct nDPIsrvd_socket ** const sock)
+static inline void nDPIsrvd_free(struct nDPIsrvd_socket ** const sock, struct nDPIsrvd_flow ** const flow_table)
 {
     struct nDPIsrvd_flow * current_flow;
     struct nDPIsrvd_flow * tmp;
 
-    HASH_ITER(hh, (*sock)->flows, current_flow, tmp)
+    if (flow_table != NULL)
     {
-        HASH_DEL((*sock)->flows, current_flow);
-        free(current_flow);
+        HASH_ITER(hh, *flow_table, current_flow, tmp)
+        {
+            HASH_DEL(*flow_table, current_flow);
+            free(current_flow);
+        }
+        *flow_table = NULL;
     }
 
     free(*sock);
@@ -256,6 +321,13 @@ static inline enum nDPIsrvd_read_return nDPIsrvd_read(struct nDPIsrvd_socket * c
     return READ_OK;
 }
 
+static inline int token_event_equals(struct nDPIsrvd_socket const * const sock, char const * const event_value)
+{
+    return sock->current.event_name != NULL && sock->current.event_name_len > 0 &&
+           (int)strlen(event_value) == sock->current.event_name_len &&
+           strncmp(sock->current.event_name, event_value, sock->current.event_name_len) == 0;
+}
+
 static inline int token_is_key(struct nDPIsrvd_socket const * const sock)
 {
     return sock->jsmn.current_token % 2;
@@ -318,7 +390,9 @@ static inline int value_equals(struct nDPIsrvd_socket const * const sock, char c
            strncmp(name, sock->jsmn.key_value.value, sock->jsmn.key_value.value_length) == 0;
 }
 
-static inline void nDPIsrvd_handle_flow(struct nDPIsrvd_socket * const sock)
+static inline struct nDPIsrvd_flow * nDPIsrvd_get_flow(struct nDPIsrvd_socket * const sock,
+                                                       struct nDPIsrvd_flow ** const flow_table,
+                                                       size_t user_data_size)
 {
     if (token_is_start(sock) == 1)
     {
@@ -330,16 +404,20 @@ static inline void nDPIsrvd_handle_flow(struct nDPIsrvd_socket * const sock)
         {
             if (strncmp(sock->current.event_name, "new", sock->current.event_name_len) == 0)
             {
-                struct nDPIsrvd_flow * f = (struct nDPIsrvd_flow *)malloc(sizeof(*f));
+                struct nDPIsrvd_flow * f = (struct nDPIsrvd_flow *)calloc(1, sizeof(*f) + user_data_size);
+                if (f == NULL)
+                {
+                    return NULL;
+                }
                 snprintf(f->id, sizeof(f->id), "%.*s", sock->current.flow_id_len, sock->current.flow_id);
-                f->user_data = NULL;
-                HASH_ADD(hh, sock->flows, id, sizeof(f->id), f);
+                HASH_ADD(hh, *flow_table, id, (size_t)sock->current.flow_id_len, f);
+                return f;
             }
-            else if (strncmp(sock->current.event_name, "end", sock->current.event_name_len) == 0 ||
-                     strncmp(sock->current.event_name, "idle", sock->current.event_name_len) == 0)
+            else
             {
                 struct nDPIsrvd_flow * f = NULL;
-                HASH_FIND(hh, sock->flows, sock->current.flow_id, (size_t)sock->current.flow_id_len, f);
+                HASH_FIND(hh, *flow_table, sock->current.flow_id, (size_t)sock->current.flow_id_len, f);
+                return f;
             }
         }
     }
@@ -347,13 +425,11 @@ static inline void nDPIsrvd_handle_flow(struct nDPIsrvd_socket * const sock)
     {
         if (key_equals(sock, "packet_event_name") == 1)
         {
-            sock->current.event = EVENT_PACKET;
             sock->current.event_name = sock->jsmn.key_value.value;
             sock->current.event_name_len = sock->jsmn.key_value.value_length;
         }
         else if (key_equals(sock, "flow_event_name") == 1)
         {
-            sock->current.event = EVENT_FLOW;
             sock->current.event_name = sock->jsmn.key_value.value;
             sock->current.event_name_len = sock->jsmn.key_value.value_length;
         }
@@ -363,6 +439,8 @@ static inline void nDPIsrvd_handle_flow(struct nDPIsrvd_socket * const sock)
             sock->current.flow_id_len = sock->jsmn.key_value.value_length;
         }
     }
+
+    return NULL;
 }
 
 static inline enum nDPIsrvd_parse_return nDPIsrvd_parse(struct nDPIsrvd_socket * const sock,
@@ -420,7 +498,7 @@ static inline enum nDPIsrvd_parse_return nDPIsrvd_parse(struct nDPIsrvd_socket *
         sock->jsmn.key_value.value = NULL;
         sock->jsmn.key_value.value_length = 0;
         sock->jsmn.current_token = 0;
-        nDPIsrvd_handle_flow(sock);
+
         if (cb(sock, user_data) != CALLBACK_OK)
         {
             return PARSE_CALLBACK_ERROR;
@@ -448,7 +526,6 @@ static inline enum nDPIsrvd_parse_return nDPIsrvd_parse(struct nDPIsrvd_socket *
                 {
                     return PARSE_JSMN_ERROR;
                 }
-                nDPIsrvd_handle_flow(sock);
                 if (cb(sock, user_data) != CALLBACK_OK)
                 {
                     return PARSE_CALLBACK_ERROR;
@@ -465,7 +542,6 @@ static inline enum nDPIsrvd_parse_return nDPIsrvd_parse(struct nDPIsrvd_socket *
         {
             return PARSE_CALLBACK_ERROR;
         }
-        nDPIsrvd_handle_flow(sock);
 
         sock->jsmn.current_token = -1;
         sock->jsmn.tokens_found = 0;
diff --git a/examples/c-captured/c-captured.c b/examples/c-captured/c-captured.c
index 5c7715eff..a7370ccbd 100644
--- a/examples/c-captured/c-captured.c
+++ b/examples/c-captured/c-captured.c
@@ -1,5 +1,6 @@
 #include <arpa/inet.h>
 #include <errno.h>
+#include <pcap/pcap.h>
 #include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -9,9 +10,46 @@
 #include <unistd.h>
 
 #include "nDPIsrvd.h"
-#include "jsmn/jsmn.h"
+#include "utarray.h"
 
-//#define VERBOSE 1
+//#define VERBOSE
+
+struct packet_data
+{
+    uint64_t packet_ts;
+    size_t packet_len;
+    size_t base64_packet_size;
+    union {
+        char * base64_packet;
+        char const * base64_packet_const;
+    };
+};
+
+struct flow_user_data
+{
+    uint8_t guessed;
+    uint8_t detected;
+    int pkt_datalink;
+    UT_array * packets;
+};
+
+struct callback_tmp_data
+{
+    uint8_t guessed;
+    uint8_t detected;
+    int pkt_datalink;
+
+    uint8_t flow_end_or_idle;
+    uint8_t is_packet_flow;
+
+    struct packet_data pkt;
+};
+
+struct callback_user_data
+{
+    struct nDPIsrvd_flow * flow_table;
+    struct callback_tmp_data tmp;
+};
 
 struct nDPIsrvd_socket * sock = NULL;
 static int main_thread_shutdown = 0;
@@ -19,55 +57,322 @@ static char const serv_listen_path[] = DISTRIBUTOR_UNIX_SOCKET;
 static char const serv_listen_addr[INET_ADDRSTRLEN] = DISTRIBUTOR_HOST;
 static uint16_t const serv_listen_port = DISTRIBUTOR_PORT;
 
-enum nDPIsrvd_callback_return nDPIsrvd_json_callback(struct nDPIsrvd_socket * const sock, void * user_data)
+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 const UT_icd packet_data_icd = {sizeof(struct packet_data), NULL, packet_data_copy, packet_data_dtor};
+
+static char * generate_pcap_filename(struct nDPIsrvd_flow const * const flow,
+                                     struct flow_user_data const * const flow_user,
+                                     char * const dest,
+                                     size_t size)
+{
+    if (flow_user->guessed != 0 || flow_user->detected == 0)
+    {
+        int ret = snprintf(dest, size, "flow-%s-%s.pcap", (flow_user->guessed != 0 ? "guessed" : "undetected"), flow->id);
+        if (ret <= 0 || (size_t)ret > size)
+        {
+            return NULL;
+        }
+    }
+    else
+    {
+        return NULL;
+    }
+
+    return dest;
+}
+
+static int packet_write_pcap_file(UT_array const * const pd_array, int pkt_datalink, char const * const filename)
 {
-    (void)user_data;
+    size_t const max_packet_len = 65535;
+
+    if (pd_array->icd.copy != packet_data_copy || pd_array->icd.dtor != packet_data_dtor)
+    {
+        return 1;
+    }
+
+    if (utarray_len(pd_array) == 0)
+    {
+        printf("no packets received via json, can not dump anything to pcap\n");
+        return 0;
+    }
 
-    if (token_is_start(sock) == 1)
+    pcap_t * p = pcap_open_dead(pkt_datalink, max_packet_len);
+    if (p == NULL)
     {
+        return 1;
+    }
+    pcap_dumper_t * pd = pcap_dump_open(p, filename);
+    if (pd == NULL)
+    {
+        fprintf(stderr, "pcap error %s\n", pcap_geterr(p));
+        pcap_close(p);
+        return 1;
+    }
+
+    struct packet_data * pd_elt = (struct packet_data *)utarray_front(pd_array);
+    do
+    {
+        if (pd_elt == NULL)
+        {
+            break;
+        }
+
+        unsigned char pkt_buf[max_packet_len];
+        size_t pkt_buf_len = sizeof(pkt_buf);
+        if (nDPIsrvd_base64decode(pd_elt->base64_packet, pd_elt->base64_packet_size, pkt_buf, &pkt_buf_len) != 0 ||
+            pkt_buf_len == 0)
+        {
+            printf("packet base64 decode failed (%zu bytes): %s\n", pd_elt->base64_packet_size, pd_elt->base64_packet);
+        }
+        else
+        {
+            struct pcap_pkthdr phdr;
+            phdr.ts.tv_sec = 0;
+            phdr.ts.tv_usec = 0;
+            phdr.caplen = pkt_buf_len;
+            phdr.len = pkt_buf_len;
+            pcap_dump((unsigned char *)pd, &phdr, pkt_buf);
+        }
+    } while ((pd_elt = (struct packet_data *)utarray_next(pd_array, pd_elt)) != NULL);
+
+    pcap_dump_close(pd);
+    pcap_close(p);
+
+    return 0;
+}
+
 #ifdef VERBOSE
-        /* Start of a JSON string. */
-        printf("JSON ");
-#endif
+static void packet_data_print(UT_array const * const pd_array)
+{
+    if (pd_array->icd.copy != packet_data_copy || pd_array->icd.dtor != packet_data_dtor)
+    {
+        return;
     }
-    else if (token_is_end(sock) == 1)
+
+    printf("packet-data array size(): %u\n", pd_array->n);
+    struct packet_data * pd_elt = (struct packet_data *)utarray_front(pd_array);
+    do
     {
+        if (pd_elt == NULL)
+        {
+            break;
+        }
+        printf("\tpacket-data base64 length: %zu\n", pd_elt->base64_packet_size);
+    } while ((pd_elt = (struct packet_data *)utarray_next(pd_array, pd_elt)) != NULL);
+}
+#else
+#define packet_data_print(pd_array)
+#endif
+
+enum nDPIsrvd_callback_return nDPIsrvd_json_callback(struct nDPIsrvd_socket * const sock, void * const user_data)
+{
+    struct callback_user_data * const cb_user_data = (struct callback_user_data *)user_data;
+    struct nDPIsrvd_flow * flow = nDPIsrvd_get_flow(sock, &cb_user_data->flow_table, sizeof(struct flow_user_data));
+    struct flow_user_data * flow_user = (struct flow_user_data *)(flow != NULL ? flow->user_data : NULL);
+
+    if (token_is_start(sock) == 1) /* Start of a JSON string. */
+    {
+        memset(&cb_user_data->tmp, 0, sizeof(cb_user_data->tmp));
+        cb_user_data->tmp.pkt_datalink = -1;
 #ifdef VERBOSE
-        /* End of a JSON string. */
-        printf("EoF\n");
+        printf("JSON ");
 #endif
+        return CALLBACK_OK;
     }
-    else if (token_is_key_value_pair(sock) == 1)
+    else if (token_is_end(sock) == 1) /* End of a JSON string. */
     {
-        if (key_equals(sock, "flow_event_name") == 1)
+        if (flow != NULL)
         {
-            if (value_equals(sock, "guessed") == 1)
-            {
-                printf("Guessed flow.\n");
-            }
-            else if (value_equals(sock, "not-detected") == 1)
+            if (cb_user_data->tmp.is_packet_flow == 1)
             {
-                printf("Not detected flow.\n");
+                if (flow_user->packets == NULL)
+                {
+                    utarray_new(flow_user->packets, &packet_data_icd);
+                }
+                if (flow_user->packets != NULL)
+                {
+                    utarray_push_back(flow_user->packets, &cb_user_data->tmp.pkt);
+                }
+                flow_user->pkt_datalink = cb_user_data->tmp.pkt_datalink;
+            } else {
+                if (cb_user_data->tmp.guessed != 0) {
+                    flow_user->guessed = cb_user_data->tmp.guessed;
+                }
+                if (cb_user_data->tmp.detected != 0) {
+                    flow_user->detected = cb_user_data->tmp.detected;
+                }
             }
-            else if (value_equals(sock, "detected") == 1)
+            if (cb_user_data->tmp.flow_end_or_idle == 1 &&
+                (flow_user->guessed != 0 || flow_user->detected == 0))
             {
-                printf("Detected flow.\n");
+                if (flow_user->packets != NULL)
+                {
+                    packet_data_print(flow_user->packets);
+                    char pcap_filename[64];
+                    if (generate_pcap_filename(flow, flow_user, pcap_filename, sizeof(pcap_filename)) == NULL)
+                    {
+                        fprintf(stderr, "%s\n", "Internal error, exit ..");
+                        return CALLBACK_ERROR;
+                    }
+                    printf("dump flow with id %s to %s\n", flow->id, pcap_filename);
+                    if (packet_write_pcap_file(flow_user->packets, flow_user->pkt_datalink, pcap_filename) != 0)
+                    {
+                        return CALLBACK_ERROR;
+                    }
+                    utarray_free(flow_user->packets);
+                    flow_user->packets = NULL;
+                }
             }
+#ifdef VERBOSE
+            printf("GUESSED: %u, DETECTED: %u ", flow_user->guessed, flow_user->detected);
+#endif
         }
 #ifdef VERBOSE
-        printf("[%.*s : %.*s] ",
-               sock->jsmn.key_value.key_length,
-               sock->jsmn.key_value.key,
-               sock->jsmn.key_value.value_length,
-               sock->jsmn.key_value.value);
+        printf("EoF\n");
 #endif
+        return CALLBACK_OK;
     }
-    else
+
+    if (token_is_key_value_pair(sock) != 1)
     {
         fprintf(stderr, "%s\n", "Internal error, exit ..");
         return CALLBACK_ERROR;
     }
 
+    if (key_equals(sock, "packet_event_name") == 1)
+    {
+        if (value_equals(sock, "packet-flow") == 1)
+        {
+            cb_user_data->tmp.is_packet_flow = 1;
+        }
+    }
+    else if (key_equals(sock, "pkt") == 1)
+    {
+        cb_user_data->tmp.pkt.base64_packet_const = sock->jsmn.key_value.value;
+        cb_user_data->tmp.pkt.base64_packet_size = sock->jsmn.key_value.value_length;
+    }
+    else if (key_equals(sock, "pkt_ts") == 1)
+    {
+        char * endptr = NULL;
+        unsigned long long int value = strtoull(sock->jsmn.key_value.value, &endptr, 10);
+        if (sock->jsmn.key_value.value == endptr)
+        {
+            fprintf(stderr,
+                    "pkt_ts `%.*s': Value `%.*s' is not a valid number.\n",
+                    sock->jsmn.key_value.key_length,
+                    sock->jsmn.key_value.key,
+                    sock->jsmn.key_value.value_length,
+                    sock->jsmn.key_value.value);
+            return CALLBACK_ERROR;
+        }
+        if (errno == ERANGE)
+        {
+            fprintf(stderr,
+                    "pkt_ts `%.*s': Number too large.\n",
+                    sock->jsmn.key_value.key_length,
+                    sock->jsmn.key_value.key);
+            return CALLBACK_ERROR;
+        }
+        cb_user_data->tmp.pkt.packet_ts = value;
+    }
+    else if (key_equals(sock, "pkt_len") == 1)
+    {
+        char * endptr = NULL;
+        unsigned long long int value = strtoull(sock->jsmn.key_value.value, &endptr, 10);
+        if (sock->jsmn.key_value.value == endptr)
+        {
+            fprintf(stderr,
+                    "pkt_len `%.*s': Value `%.*s' is not a valid number.\n",
+                    sock->jsmn.key_value.key_length,
+                    sock->jsmn.key_value.key,
+                    sock->jsmn.key_value.value_length,
+                    sock->jsmn.key_value.value);
+            return CALLBACK_ERROR;
+        }
+        if (errno == ERANGE)
+        {
+            fprintf(stderr,
+                    "pkt_len `%.*s': Number too large.\n",
+                    sock->jsmn.key_value.key_length,
+                    sock->jsmn.key_value.key);
+            return CALLBACK_ERROR;
+        }
+        cb_user_data->tmp.pkt.packet_len = value;
+    }
+    else if (key_equals(sock, "pkt_datalink") == 1)
+    {
+        char * endptr = NULL;
+        unsigned long long int value = strtoull(sock->jsmn.key_value.value, &endptr, 10);
+        if (sock->jsmn.key_value.value == endptr)
+        {
+            fprintf(stderr,
+                    "pkt_datalink `%.*s': Value `%.*s' is not a valid number.\n",
+                    sock->jsmn.key_value.key_length,
+                    sock->jsmn.key_value.key,
+                    sock->jsmn.key_value.value_length,
+                    sock->jsmn.key_value.value);
+            return CALLBACK_ERROR;
+        }
+        if (errno == ERANGE || value > (unsigned long long int)((uint32_t)-1))
+        {
+            fprintf(stderr,
+                    "pkt_datalink `%.*s': Number too large.\n",
+                    sock->jsmn.key_value.key_length,
+                    sock->jsmn.key_value.key);
+            return CALLBACK_ERROR;
+        }
+        cb_user_data->tmp.pkt_datalink = value;
+    }
+    else if (key_equals(sock, "flow_event_name") == 1)
+    {
+        if (value_equals(sock, "end") == 1 || value_equals(sock, "idle") == 1)
+        {
+            cb_user_data->tmp.flow_end_or_idle = 1;
+        }
+        else if (value_equals(sock, "guessed") == 1)
+        {
+            cb_user_data->tmp.guessed = 1;
+        }
+        else if (value_equals(sock, "detected") == 1)
+        {
+            cb_user_data->tmp.detected = 1;
+        }
+    }
+
+#ifdef VERBOSE
+    printf("[%.*s : %.*s] ",
+           sock->jsmn.key_value.key_length,
+           sock->jsmn.key_value.key,
+           sock->jsmn.key_value.value_length,
+           sock->jsmn.key_value.value);
+#endif
+
     return CALLBACK_OK;
 }
 
@@ -83,6 +388,9 @@ static void sighandler(int signum)
 
 int main(int argc, char ** argv)
 {
+    struct callback_user_data cb_user_data;
+
+    memset(&cb_user_data, 0, sizeof(cb_user_data));
     sock = nDPIsrvd_init();
     if (sock == NULL)
     {
@@ -100,12 +408,16 @@ int main(int argc, char ** argv)
     {
         printf("Connecting to UNIX socket: %s\n", argv[1]);
         connect_ret = nDPIsrvd_connect_unix(sock, argv[1]);
-    } else if (argc == 1) {
+    }
+    else if (argc == 1)
+    {
         if (access(serv_listen_path, R_OK) == 0)
         {
             printf("Connecting to %s\n", serv_listen_path);
             connect_ret = nDPIsrvd_connect_unix(sock, serv_listen_path);
-        } else {
+        }
+        else
+        {
             printf("Connecting to %s:%u\n", serv_listen_addr, serv_listen_port);
             connect_ret = nDPIsrvd_connect_ip(sock, serv_listen_addr, serv_listen_port);
         }
@@ -114,7 +426,7 @@ int main(int argc, char ** argv)
     if (connect_ret != CONNECT_OK)
     {
         fprintf(stderr, "%s: nDPIsrvd socket connect failed!\n", argv[0]);
-        nDPIsrvd_free(&sock);
+        nDPIsrvd_free(&sock, &cb_user_data.flow_table);
         return 1;
     }
 
@@ -124,18 +436,30 @@ int main(int argc, char ** argv)
         enum nDPIsrvd_read_return read_ret = nDPIsrvd_read(sock);
         if (read_ret != READ_OK)
         {
+            fprintf(stderr, "%s: nDPIsrvd read failed with: %s\n", argv[0], nDPIsrvd_enum_to_string(read_ret));
             break;
         }
 
-        enum nDPIsrvd_parse_return parse_ret = nDPIsrvd_parse(sock, nDPIsrvd_json_callback, NULL);
-        switch (parse_ret)
+        enum nDPIsrvd_parse_return parse_ret = nDPIsrvd_parse(sock, nDPIsrvd_json_callback, &cb_user_data);
+        if (parse_ret != PARSE_OK)
         {
-            default:
-                break;
+            fprintf(stderr, "%s: nDPIsrvd parse failed with: %s\n", argv[0], nDPIsrvd_enum_to_string(parse_ret));
+            break;
         }
     }
 
-    nDPIsrvd_free(&sock);
+    struct nDPIsrvd_flow * current_flow;
+    struct nDPIsrvd_flow * tmp;
+    HASH_ITER(hh, cb_user_data.flow_table, current_flow, tmp)
+    {
+        struct flow_user_data * const ud = (struct flow_user_data *)current_flow->user_data;
+        if (ud != NULL && ud->packets != NULL)
+        {
+            utarray_free(ud->packets);
+            ud->packets = NULL;
+        }
+    }
+    nDPIsrvd_free(&sock, &cb_user_data.flow_table);
 
     return 0;
 }
diff --git a/examples/go-dashboard/main.go b/examples/go-dashboard/main.go
index 8201ca9df..efd01974b 100644
--- a/examples/go-dashboard/main.go
+++ b/examples/go-dashboard/main.go
@@ -183,7 +183,7 @@ func main() {
 					break
 				}
 
-				if jsonStr[jsonLen+nDPIsrvd_JSON_BYTES-1] != '}' {
+				if jsonStr[jsonLen+nDPIsrvd_JSON_BYTES-2] != '}' || jsonStr[jsonLen+nDPIsrvd_JSON_BYTES-1] != '\n' {
 					ErrorLogger.Printf("BUG: JSON invalid closing character at position %d: '%s'\n",
 						jsonLen+nDPIsrvd_JSON_BYTES,
 						string(jsonStr[jsonLen+nDPIsrvd_JSON_BYTES-1]))