c-captured extension

 * capture packets after error event occurred
 * add "logging" and "capture" mode

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

1 files changed, 629 insertions, 137 deletions

diff --git a/examples/c-captured/c-captured.c b/examples/c-captured/c-captured.c
index 2bd291232..c914c6991 100644
--- a/examples/c-captured/c-captured.c
+++ b/examples/c-captured/c-captured.c
@@ -29,6 +29,9 @@
 
 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;
@@ -40,9 +43,26 @@ struct packet_data
     };
 };
 
+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 flow_new_seen;
     uint8_t detection_finished;
     uint8_t guessed;
     uint8_t detected;
@@ -51,7 +71,7 @@ struct flow_user_data
     nDPIsrvd_ull flow_datalink;
     nDPIsrvd_ull flow_max_packets;
     nDPIsrvd_ull flow_tot_l4_payload_len;
-    UT_array * packets;
+    UT_array * packets; // flow_packet_data
 };
 
 static struct nDPIsrvd_socket * sock = NULL;
@@ -64,6 +84,8 @@ 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;
@@ -71,6 +93,8 @@ 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)
 {
@@ -94,6 +118,81 @@ void nDPIsrvd_memprof_log(char const * const format, ...)
 }
 #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)
+    {
+        syslog(LOG_DAEMON | LOG_ERR, "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
+    {
+        syslog(LOG_DAEMON | LOG_ERR, "%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)
+    {
+        syslog(LOG_DAEMON | LOG_ERR, "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;
@@ -121,7 +220,144 @@ static void packet_data_dtor(void * elt)
     }
 }
 
+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)
 {
@@ -152,13 +388,8 @@ static int has_ndpi_risk(ndpi_risk * const risk, nDPIsrvd_ull risk_to_check)
     return (*risk & (1ull << --risk_to_check)) != 0;
 }
 
-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)
+static int generate_pcap_appendix(char * const appendix, size_t appendix_size)
 {
-    char appendix[32] = {};
-
     if (pcap_filename_rotation > 0)
     {
         time_t current_time = time(NULL);
@@ -168,22 +399,44 @@ static char * generate_pcap_filename(struct nDPIsrvd_flow const * const flow,
             pcap_filename_last_rotation = current_time;
             if (localtime_r(&pcap_filename_last_rotation, &pcap_filename_last_rotation_tm) == NULL)
             {
-                return NULL;
+                return 1;
             }
         }
 
-        if (strftime(appendix, sizeof(appendix), "%d_%m_%y-%H_%M_%S", &pcap_filename_last_rotation_tm) == 0)
+        if (strftime(appendix, appendix_size, "-%d_%m_%y-%H_%M_%S", &pcap_filename_last_rotation_tm) == 0)
         {
-            return NULL;
+            return 1;
         }
     }
-    else
 
+    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)
     {
-        if (snprintf(appendix, sizeof(appendix), "%llu", flow->id_as_ull) <= 0)
-        {
-            return NULL;
-        }
+        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)
@@ -211,7 +464,7 @@ static char * generate_pcap_filename(struct nDPIsrvd_flow const * const flow,
             flow_type = "unknown-type";
         }
 
-        int ret = snprintf(dest, size, "%s/flow-%s-%s.pcap", datadir, flow_type, appendix);
+        int ret = snprintf(dest, size, "%s/flow-%s%s.pcap", datadir, flow_type, appendix);
         if (ret <= 0 || (size_t)ret > size)
         {
             return NULL;
@@ -225,45 +478,102 @@ static char * generate_pcap_filename(struct nDPIsrvd_flow const * const flow,
     return dest;
 }
 
-static int packet_write_pcap_file(UT_array const * const pd_array, int pkt_datalink, char const * const filename)
+static int packet_write_pcap_file(struct global_user_data const * const global_user)
 {
-    size_t const max_packet_len = 65535;
+    UT_array * const pd_array = global_user->packets;
 
-    if (pd_array->icd.copy != packet_data_copy || pd_array->icd.dtor != packet_data_dtor)
+    if (utarray_len(pd_array) == 0)
     {
+        syslog(LOG_DAEMON, "Can not dump packets to pcap; packet array empty");
         return 1;
     }
 
-    if (utarray_len(pd_array) == 0)
+    while (utarray_len(pd_array) > 0)
     {
-        syslog(LOG_DAEMON, "no packets received via json, can not dump anything to pcap");
-        return 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;
 
-    pcap_t * p = pcap_open_dead(pkt_datalink, max_packet_len);
-    if (p == NULL)
-    {
-        return 1;
-    }
+        char filename[PATH_MAX];
+        if (packet_generate_pcap_filename(filename, sizeof(filename), packet_datalink) == NULL)
+        {
+            syslog(LOG_DAEMON | LOG_ERR, "%s", "Internal error. Could not generate PCAP filename, exit ..");
+            return 1;
+        }
 
-    pcap_dumper_t * pd;
-    if (access(filename, F_OK) == 0)
-    {
-        pd = pcap_dump_open_append(p, filename);
+        pcap_t * p = NULL;
+        pcap_dumper_t * pd = NULL;
+        if (pcap_open_or_append(packet_datalink, filename, &p, &pd) != 0)
+        {
+            syslog(LOG_DAEMON | LOG_ERR, "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", pcap_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++;
+            }
+        }
     }
-    else
+
+    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)
     {
-        pd = pcap_dump_open(p, filename);
+        syslog(LOG_DAEMON | LOG_ERR, "Can not dump flow packets to pcap; flow packet array empty");
+        return 0;
     }
 
-    if (pd == NULL)
+    if (pcap_open_or_append(packet_datalink, filename, &p, &pd) != 0)
     {
-        syslog(LOG_DAEMON | LOG_ERR, "pcap error %s", pcap_geterr(p));
-        pcap_close(p);
-        return 1;
+        syslog(LOG_DAEMON | LOG_ERR, "Can not dump flow packets to pcap; file open/append failed");
+        return 0;
     }
 
-    struct packet_data * pd_elt = (struct packet_data *)utarray_front(pd_array);
+    struct flow_packet_data * pd_elt = (struct flow_packet_data *)utarray_front(pd_array);
     do
     {
         if (pd_elt == NULL)
@@ -271,26 +581,8 @@ static int packet_write_pcap_file(UT_array const * const pd_array, int pkt_datal
             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)
-        {
-            syslog(LOG_DAEMON | LOG_ERR,
-                   "packet base64 decode failed (%d bytes): %s",
-                   pd_elt->base64_packet_size,
-                   pd_elt->base64_packet);
-        }
-        else
-        {
-            struct pcap_pkthdr phdr;
-            phdr.ts.tv_sec = pd_elt->packet_ts_sec;
-            phdr.ts.tv_usec = pd_elt->packet_ts_usec;
-            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);
+        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);
@@ -299,15 +591,12 @@ static int packet_write_pcap_file(UT_array const * const pd_array, int pkt_datal
 }
 
 #ifdef VERBOSE
-static void packet_data_print(UT_array const * const pd_array)
+static void flow_packet_data_print(struct flow_user_data const * const flow_user)
 {
-    if (pd_array->icd.copy != packet_data_copy || pd_array->icd.dtor != packet_data_dtor)
-    {
-        return;
-    }
+    UT_array const * const pd_array = flow_user->packets;
 
     printf("packet-data array size(): %u\n", pd_array->n);
-    struct packet_data * pd_elt = (struct packet_data *)utarray_front(pd_array);
+    struct flow_packet_data * pd_elt = (struct flow_packet_data *)utarray_front(pd_array);
     do
     {
         if (pd_elt == NULL)
@@ -315,10 +604,10 @@ static void packet_data_print(UT_array const * const pd_array)
             break;
         }
         printf("\tpacket-data base64 length: %d\n", pd_elt->base64_packet_size);
-    } while ((pd_elt = (struct packet_data *)utarray_next(pd_array, pd_elt)) != NULL);
+    } while ((pd_elt = (struct flow_packet_data *)utarray_next(pd_array, pd_elt)) != NULL);
 }
 #else
-#define packet_data_print(pd_array)
+#define flow_packet_data_print(pd_array)
 #endif
 
 static enum nDPIsrvd_conversion_return perror_ull(enum nDPIsrvd_conversion_return retval, char const * const prefix)
@@ -346,6 +635,60 @@ static enum nDPIsrvd_conversion_return perror_ull(enum nDPIsrvd_conversion_retur
     return retval;
 }
 
+static void syslog_event(struct nDPIsrvd_socket * 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)
+    {
+        syslog(LOG_DAEMON | LOG_ERR, "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, dst_port_len;
+        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);
+        }
+    }
+
+    syslog(LOG_DAEMON | LOG_ERR,
+           "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,
@@ -354,9 +697,101 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
     (void)instance;
     (void)thread_data;
 
+    struct global_user_data * const global_user = (struct global_user_data *)&sock->global_user_data[0];
+
     if (flow == NULL)
     {
-        return CALLBACK_OK; // We do not care for non-flow events for NOW except for packet-flow events.
+        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)
+        {
+            syslog(LOG_DAEMON | LOG_ERR, "Memory allocation for packet data failed.");
+            return CALLBACK_ERROR;
+        }
+
+        if (TOKEN_GET_SZ(sock, "error_event_name") != NULL)
+        {
+            syslog(LOG_DAEMON | LOG_ERR, "Received an error event for packet id %llu.", packet_id);
+
+            if (TOKEN_GET_SZ(sock, "error_event_id") == NULL)
+            {
+                syslog(LOG_DAEMON | LOG_ERR, "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)
+            {
+                syslog(LOG_DAEMON | LOG_ERR, "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)
+        {
+            syslog(LOG_DAEMON | LOG_ERR, "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)
+                {
+                    syslog(LOG_DAEMON | LOG_ERR,
+                           "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)
+                {
+                    syslog(LOG_DAEMON | LOG_ERR,
+                           "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)
+                {
+                    syslog(LOG_DAEMON | LOG_ERR, "%s", "No packet data available.");
+                    syslog(LOG_DAEMON | LOG_ERR,
+                           "JSON String: '%.*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)
+                {
+                    syslog(LOG_DAEMON | LOG_ERR, "%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;
@@ -366,20 +801,25 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
         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)
         {
             syslog(LOG_DAEMON | LOG_ERR, "%s", "No packet data available.");
-            syslog(LOG_DAEMON | LOG_ERR, "JSON String: '%.*s'", nDPIsrvd_json_buffer_length(sock), nDPIsrvd_json_buffer_string(sock));
+            syslog(LOG_DAEMON | LOG_ERR,
+                   "JSON String: '%.*s'",
+                   nDPIsrvd_json_buffer_length(sock),
+                   nDPIsrvd_json_buffer_string(sock));
             return CALLBACK_OK;
         }
-        if (flow_user->packets == NULL)
-        {
-            utarray_new(flow_user->packets, &packet_data_icd);
-        }
-        if (flow_user->packets == NULL)
+
+        if (utarray_flow_packets_init(flow_user) == 0)
         {
             syslog(LOG_DAEMON | LOG_ERR, "%s", "Memory allocation for captured packets failed.");
             return CALLBACK_ERROR;
@@ -391,18 +831,16 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
         nDPIsrvd_ull pkt_len = 0ull;
         perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "pkt_caplen"), &pkt_len), "pkt_caplen");
 
-        nDPIsrvd_ull pkt_l4_len = 0ull;
-        perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "pkt_l4_len"), &pkt_l4_len), "pkt_l4_len");
-
-        nDPIsrvd_ull pkt_l4_offset = 0ull;
-        perror_ull(TOKEN_VALUE_TO_ULL(sock, TOKEN_GET_SZ(sock, "pkt_l4_offset"), &pkt_l4_offset), "pkt_l4_offset");
-
-        struct 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)};
-        utarray_push_back(flow_user->packets, &pd);
+        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)
+        {
+            syslog(LOG_DAEMON | LOG_ERR, "%s", "Memory allocation to add a captured packet failed.");
+            return CALLBACK_ERROR;
+        }
     }
 
     {
@@ -425,7 +863,6 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
 
         if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "new") != 0)
         {
-            flow_user->flow_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),
@@ -437,25 +874,35 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
 
             return CALLBACK_OK;
         }
+        else if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "end") != 0)
+        {
+            struct nDPIsrvd_json_token const * const ndpi_proto = TOKEN_GET_SZ(sock, "ndpi", "proto");
+
+            if (ndpi_proto != NULL)
+            {
+                flow_user->detected = 1;
+            }
+        }
         else if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "guessed") != 0)
         {
             flow_user->guessed = 1;
-            flow_user->detection_finished = 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") != 0)
+                 TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detection-update") != 0 ||
+                 TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "update") != 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;
 
-            flow_user->detected = 1;
-            flow_user->detection_finished = 1;
+            if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "update") == 0)
+            {
+                flow_user->detected = 1;
+            }
 
             if (flow_risk != NULL)
             {
@@ -466,49 +913,58 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
                     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->detected = 1;
                         flow_user->risky = 1;
                     }
                 }
             }
         }
 
-        if (flow_user->flow_new_seen == 0)
-        {
-            return CALLBACK_OK;
-        }
-
-        if (flow_user->packets == NULL || flow_user->flow_max_packets == 0 || utarray_len(flow_user->packets) == 0)
-        {
-            syslog(LOG_DAEMON | LOG_ERR, "flow %llu: No packets captured.", flow->id_as_ull);
-            return CALLBACK_OK;
-        }
-
         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)))
         {
-            packet_data_print(flow_user->packets);
-            if (ignore_empty_flows == 0 || flow_user->flow_tot_l4_payload_len > 0)
+            if (logging_mode != 0)
+            {
+                if (flow_user->guessed != 0)
+                    syslog_event(sock, flow, "guessed");
+                if (flow_user->detected == 0)
+                    syslog_event(sock, flow, "not-detected");
+                if (flow_user->risky != 0)
+                    syslog_event(sock, flow, "risky");
+                if (flow_user->midstream != 0)
+                    syslog_event(sock, flow, "midstream");
+            }
+
+            if (capture_mode != 0 && (flow_user->packets == NULL || flow_user->flow_max_packets == 0 ||
+                                      utarray_len(flow_user->packets) == 0))
             {
-                char pcap_filename[PATH_MAX];
-                if (generate_pcap_filename(flow, flow_user, pcap_filename, sizeof(pcap_filename)) == NULL)
+                syslog(LOG_DAEMON | LOG_ERR, "Flow %llu: No packets captured.", flow->id_as_ull);
+            }
+            else if (capture_mode != 0)
+            {
+                flow_packet_data_print(flow_user);
+                if (ignore_empty_flows == 0 || flow_user->flow_tot_l4_payload_len > 0)
                 {
-                    syslog(LOG_DAEMON | LOG_ERR, "%s", "Internal error. Could not generate PCAP filename, exit ..");
-                    return CALLBACK_ERROR;
-                }
+                    char pcap_filename[PATH_MAX];
+                    if (flow_generate_pcap_filename(flow_user, pcap_filename, sizeof(pcap_filename)) == NULL)
+                    {
+                        syslog(LOG_DAEMON | LOG_ERR, "%s", "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);
+                    printf("Flow %llu saved to %s\n", flow->id_as_ull, pcap_filename);
 #endif
-                if (packet_write_pcap_file(flow_user->packets, flow_user->flow_datalink, pcap_filename) != 0)
-                {
-                    syslog(LOG_DAEMON | LOG_ERR, "Could not packet data to pcap file %s", pcap_filename);
-                    return CALLBACK_ERROR;
+                    if (flow_write_pcap_file(flow_user, pcap_filename) != 0)
+                    {
+                        syslog(LOG_DAEMON | LOG_ERR, "Could not dump packet data to pcap file %s", pcap_filename);
+                        return CALLBACK_OK;
+                    }
                 }
             }
 
-            utarray_free(flow_user->packets);
-            flow_user->packets = NULL;
+            utarray_flow_packets_free(flow_user);
         }
     }
 
@@ -535,7 +991,7 @@ static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_socket const * sock,
 #else
             "0x%08lx"
 #endif
-            "][last-seen: %13llu][new-seen: %u][finished: %u][detected: %u][risky: "
+            "][last-seen: %13llu][finished: %u][detected: %u][risky: "
             "%u][total-L4-payload-length: "
             "%4llu][packets-captured: %u]\n",
             flow->id_as_ull,
@@ -545,7 +1001,6 @@ static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_socket const * sock,
             (unsigned long int)flow,
 #endif
             flow->last_seen,
-            flow_user->flow_new_seen,
             flow_user->detection_finished,
             flow_user->detected,
             flow_user->risky,
@@ -559,7 +1014,7 @@ static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_socket const * sock,
 #else
            "0x%08lx"
 #endif
-           "][last-seen: %13llu][new-seen: %u][finished: %u][detected: %u][risky: "
+           "][last-seen: %13llu][finished: %u][detected: %u][risky: "
            "%u][total-L4-payload-length: "
            "%4llu][packets-captured: %u]",
            flow->id_as_ull,
@@ -569,7 +1024,6 @@ static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_socket const * sock,
            (unsigned long int)flow,
 #endif
            flow->last_seen,
-           flow_user->flow_new_seen,
            flow_user->detection_finished,
            flow_user->detected,
            flow_user->risky,
@@ -621,6 +1075,8 @@ static void print_usage(char const * const arg0)
         "\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"
@@ -651,7 +1107,7 @@ static int parse_options(int argc, char ** argv)
 {
     int opt;
 
-    while ((opt = getopt(argc, argv, "hdp:s:r:u:g:D:GUR:ME")) != -1)
+    while ((opt = getopt(argc, argv, "hdp:s:r:u:g:lLD:GUR:ME")) != -1)
     {
         switch (opt)
         {
@@ -682,6 +1138,12 @@ static int parse_options(int argc, char ** argv)
                 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);
@@ -728,6 +1190,12 @@ static int parse_options(int argc, char ** argv)
         }
     }
 
+    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);
@@ -757,19 +1225,30 @@ static int parse_options(int argc, char ** argv)
         return 1;
     }
 
-    errno = 0;
-    if (datadir[0] != '/')
+    if (capture_mode != 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;
+        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;
@@ -817,8 +1296,15 @@ static int mainloop(void)
 
 int main(int argc, char ** argv)
 {
-    sock = nDPIsrvd_socket_init(
-        0, 0, 0, sizeof(struct flow_user_data), captured_json_callback, NULL, captured_flow_cleanup_callback);
+    init_logging("nDPIsrvd-captured");
+
+    sock = nDPIsrvd_socket_init(sizeof(struct global_user_data),
+                                0,
+                                0,
+                                sizeof(struct flow_user_data),
+                                captured_json_callback,
+                                NULL,
+                                captured_flow_cleanup_callback);
     if (sock == NULL)
     {
         fprintf(stderr, "%s: nDPIsrvd socket memory allocation failed!\n", argv[0]);
@@ -866,8 +1352,14 @@ int main(int argc, char ** argv)
     }
     chmod(datadir, S_IRWXU);
 
+    if (nDPIsrvd_set_read_timeout(sock, 180, 0) != 0)
+    {
+        return 1;
+    }
+
     int retval = mainloop();
 
+    utarray_packets_free((struct global_user_data *)sock->global_user_data);
     nDPIsrvd_socket_free(&sock);
     daemonize_shutdown(pidfile);
     closelog();