Add DBUS suspicious flow event notification daemon.

* nDPIsrvd.h: support for closing/resetting a nDPIsrvd_socket (required for a reconnect)

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

CMakeLists.txt

@@ -69,8 +69,12 @@ option(ENABLE_MEMORY_PROFILING "Enable dynamic memory tracking." OFF)
 option(ENABLE_ZLIB "Enable zlib support for nDPId (experimental)." OFF)
 option(ENABLE_SYSTEMD "Install systemd components." OFF)
 option(BUILD_EXAMPLES "Build C examples." ON)
+option(ENABLE_DBUS "Build DBus notification example." OFF)
 option(BUILD_NDPI "Clone and build nDPI from github." OFF)
 if(BUILD_NDPI)
+    if(APPLE)
+        message(WARNING "Building libnDPI from CMake is not supported on Apple and may fail.")
+    endif()
     option(BUILD_NDPI_FORCE_GIT_UPDATE "Forcefully instruments nDPI build script to update the git submodule." OFF)
     unset(NDPI_NO_PKGCONFIG CACHE)
     unset(STATIC_LIBNDPI_INSTALLDIR CACHE)
@@ -160,6 +164,9 @@ if(ENABLE_ZLIB)
     set(ZLIB_DEFS "-DENABLE_ZLIB=1")
     pkg_check_modules(ZLIB REQUIRED zlib)
 endif()
+if(ENABLE_DBUS)
+    pkg_check_modules(DBUS REQUIRED dbus-1)
+endif()
 if(NDPI_WITH_GCRYPT)
     message(STATUS "nDPI: Enable GCRYPT")
     set(NDPI_ADDITIONAL_ARGS "${NDPI_ADDITIONAL_ARGS} --with-local-libgcrypt")
@@ -366,6 +373,18 @@ if(BUILD_EXAMPLES)
         add_dependencies(coverage nDPIsrvd-analysed nDPIsrvd-collectd nDPIsrvd-captured nDPIsrvd-json-dump nDPIsrvd-simple)
     endif()
 
+    if(ENABLE_DBUS)
+        add_executable(nDPIsrvd-notifyd examples/c-notifyd/c-notifyd.c utils.c)
+        if(BUILD_NDPI)
+            add_dependencies(nDPIsrvd-notifyd libnDPI)
+        endif()
+        target_include_directories(nDPIsrvd-notifyd PRIVATE
+                                   "${STATIC_LIBNDPI_INC}" "${DEFAULT_NDPI_INCLUDE}" "${CMAKE_SOURCE_DIR}" "${NDPID_DEPS_INC}"
+                                   "${DBUS_INCLUDE_DIRS}")
+        target_link_libraries(nDPIsrvd-notifyd "${DBUS_LIBRARIES}")
+        install(TARGETS nDPIsrvd-notifyd DESTINATION bin)
+    endif()
+
     install(TARGETS nDPIsrvd-analysed nDPIsrvd-collectd nDPIsrvd-captured nDPIsrvd-json-dump nDPIsrvd-simple DESTINATION bin)
     install(FILES examples/c-collectd/plugin_nDPIsrvd.conf examples/c-collectd/rrdgraph.sh DESTINATION share/nDPId/nDPIsrvd-collectd)
     install(DIRECTORY examples/c-collectd/www DESTINATION share/nDPId/nDPIsrvd-collectd)
@@ -429,6 +448,7 @@ if(STATIC_LIBNDPI_INSTALLDIR)
 message(STATUS "STATIC_LIBNDPI_INSTALLDIR: ${STATIC_LIBNDPI_INSTALLDIR}")
 endif()
 message(STATUS "BUILD_NDPI...............: ${BUILD_NDPI}")
+message(STATUS "ENABLE_DBUS..............: ${ENABLE_DBUS}")
 if(BUILD_NDPI)
 message(STATUS "NDPI_ADDITIONAL_ARGS.....: ${NDPI_ADDITIONAL_ARGS}")
 endif()

dependencies/nDPIsrvd.h

@@ -399,14 +399,19 @@ static inline void nDPIsrvd_buffer_free(struct nDPIsrvd_buffer * const buffer)
     buffer->max = 0;
 }
 
+static inline void nDPIsrvd_json_buffer_reset(struct nDPIsrvd_json_buffer * const json_buffer)
+{
+    json_buffer->json_string_start = 0ul;
+    json_buffer->json_string_length = 0ull;
+    json_buffer->json_string = NULL;
+}
+
 static inline int nDPIsrvd_json_buffer_init(struct nDPIsrvd_json_buffer * const json_buffer, size_t json_buffer_size)
 {
     int ret = nDPIsrvd_buffer_init(&json_buffer->buf, json_buffer_size);
     if (ret == 0)
     {
-        json_buffer->json_string_start = 0ul;
-        json_buffer->json_string_length = 0ull;
-        json_buffer->json_string = NULL;
+        nDPIsrvd_json_buffer_reset(json_buffer);
     }
 
     return ret;
@@ -415,9 +420,7 @@ static inline int nDPIsrvd_json_buffer_init(struct nDPIsrvd_json_buffer * const
 static inline void nDPIsrvd_json_buffer_free(struct nDPIsrvd_json_buffer * const json_buffer)
 {
     nDPIsrvd_buffer_free(&json_buffer->buf);
-    json_buffer->json_string_start = 0ul;
-    json_buffer->json_string_length = 0ull;
-    json_buffer->json_string = NULL;
+    nDPIsrvd_json_buffer_reset(json_buffer);
 }
 
 static inline struct nDPIsrvd_socket * nDPIsrvd_socket_init(size_t global_user_data_size,
@@ -468,7 +471,6 @@ static inline struct nDPIsrvd_socket * nDPIsrvd_socket_init(size_t global_user_d
 
     return sock;
 error:
-    nDPIsrvd_json_buffer_free(&sock->buffer);
     nDPIsrvd_socket_free(&sock);
     return NULL;
 }
@@ -585,25 +587,56 @@ static inline void nDPIsrvd_cleanup_instance(struct nDPIsrvd_socket * const sock
     }
 }
 
-static inline void nDPIsrvd_socket_free(struct nDPIsrvd_socket ** const sock)
+static inline void nDPIsrvd_socket_close(struct nDPIsrvd_socket * const sock)
 {
     struct nDPIsrvd_instance * current_instance;
     struct nDPIsrvd_instance * itmp;
     struct nDPIsrvd_json_token * current_json_token;
     struct nDPIsrvd_json_token * jtmp;
 
-    if (sock == NULL || *sock == NULL)
+    if (sock == NULL)
     {
         return;
     }
 
-    if ((*sock)->json.token_table != NULL)
+    if (sock->json.token_table != NULL)
     {
-        HASH_ITER(hh, (*sock)->json.token_table, current_json_token, jtmp)
+        HASH_ITER(hh, sock->json.token_table, current_json_token, jtmp)
         {
-            HASH_DEL((*sock)->json.token_table, current_json_token);
+            HASH_DEL(sock->json.token_table, current_json_token);
         }
-        (*sock)->json.token_table = NULL;
+    }
+
+    if (sock->json.tokens != NULL)
+    {
+        utarray_clear(sock->json.tokens);
+    }
+
+    if (sock->instance_table != NULL)
+    {
+        HASH_ITER(hh, sock->instance_table, current_instance, itmp)
+        {
+            nDPIsrvd_cleanup_instance(sock, current_instance, CLEANUP_REASON_APP_SHUTDOWN);
+        }
+    }
+
+    nDPIsrvd_json_buffer_reset(&sock->buffer);
+    close(sock->fd);
+    sock->fd = -1;
+}
+
+static inline void nDPIsrvd_socket_free(struct nDPIsrvd_socket ** const sock)
+{
+    if (sock == NULL)
+    {
+        return;
+    }
+
+    nDPIsrvd_socket_close(*sock);
+
+    if (*sock == NULL)
+    {
+        return;
     }
 
     if ((*sock)->json.tokens != NULL)
@@ -611,15 +644,11 @@ static inline void nDPIsrvd_socket_free(struct nDPIsrvd_socket ** const sock)
         utarray_free((*sock)->json.tokens);
     }
 
-    HASH_ITER(hh, (*sock)->instance_table, current_instance, itmp)
-    {
-        nDPIsrvd_cleanup_instance(*sock, current_instance, CLEANUP_REASON_APP_SHUTDOWN);
-    }
+    (*sock)->json.tokens = NULL;
+    (*sock)->json.token_table = NULL;
     (*sock)->instance_table = NULL;
-
     nDPIsrvd_json_buffer_free(&(*sock)->buffer);
     nDPIsrvd_free(*sock);
-
     *sock = NULL;
 }
 
@@ -1653,7 +1682,7 @@ static inline int nDPIsrvd_json_buffer_length(struct nDPIsrvd_socket const * con
     return (int)sock->buffer.json_string_length - NETWORK_BUFFER_LENGTH_DIGITS;
 }
 
-static inline char const *nDPIsrvd_json_buffer_string(struct nDPIsrvd_socket const * const sock)
+static inline char const * nDPIsrvd_json_buffer_string(struct nDPIsrvd_socket const * const sock)
 {
     if (sock == NULL)
     {

examples/README.md

@@ -18,6 +18,10 @@ It saves flows that were guessed/undetected/risky/midstream to a PCAP file for m
 
 A collecd-exec compatible middleware that gathers statistic values from nDPId.
 
+## c-notifyd
+
+A notification daemon that sends information about suspicious flow events to DBUS.
+
 ## c-json-stdout
 
 Tiny nDPId json dumper. Does not provide any useful funcationality besides dumping parsed JSON objects.

examples/c-collectd/c-collectd.c

@@ -1,4 +1,4 @@
-#include <arpa/inet.h>
+#include <arpa/inet.h> // ndpi_typedefs.h
 #include <errno.h>
 #include <signal.h>
 #include <stdbool.h> // ndpi_typedefs.h

examples/c-notifyd/c-notifyd.c

@@ -0,0 +1,609 @@
+#include <dbus-1.0/dbus/dbus.h>
+#include <signal.h>
+#include <stdint.h>
+#include <syslog.h>
+
+#include "nDPIsrvd.h"
+#include "utstring.h"
+#include "utils.h"
+
+struct flow_user_data
+{
+    nDPIsrvd_ull detected_risks;
+};
+
+enum dbus_level
+{
+    DBUS_LOW = 0,
+    DBUS_NORMAL,
+    DBUS_CRITICAL
+};
+
+static char const * const flow_severities[] = {"Low", "Medium", "High", "Severe", "Critical", "Emergency"};
+static char const * const flow_breeds[] = {
+    "Safe", "Acceptable", "Fun", "Unsafe", "Potentially Dangerous", "Tracker/Ads", "Dangerous", "Unrated", "???"};
+static char const * const flow_categories[] = {"Unspecified",
+                                               "Media",
+                                               "VPN",
+                                               "Email",
+                                               "DataTransfer",
+                                               "Web",
+                                               "SocialNetwork",
+                                               "Download",
+                                               "Game",
+                                               "Chat",
+                                               "VoIP",
+                                               "Database",
+                                               "RemoteAccess",
+                                               "Cloud",
+                                               "Network",
+                                               "Collaborative",
+                                               "RPC",
+                                               "Streaming",
+                                               "System",
+                                               "SoftwareUpdate",
+                                               "Music",
+                                               "Video",
+                                               "Shopping",
+                                               "Productivity",
+                                               "FileSharing",
+                                               "ConnCheck",
+                                               "IoT-Scada",
+                                               "VirtAssistant",
+                                               "Cybersecurity",
+                                               "AdultContent",
+                                               "Mining",
+                                               "Malware",
+                                               "Advertisement",
+                                               "Banned_Site",
+                                               "Site_Unavailable",
+                                               "Allowed_Site",
+                                               "Antimalware",
+                                               "Crypto_Currency"};
+
+static uint8_t desired_flow_severities[nDPIsrvd_ARRAY_LENGTH(flow_severities)] = {};
+static uint8_t desired_flow_breeds[nDPIsrvd_ARRAY_LENGTH(flow_breeds)] = {};
+static uint8_t desired_flow_categories[nDPIsrvd_ARRAY_LENGTH(flow_categories)] = {};
+
+static unsigned int id = 0;
+static char const * const application = "nDPIsrvd.notifyd";
+
+static int main_thread_shutdown = 0;
+
+static char * pidfile = NULL;
+static char * serv_optarg = NULL;
+
+static void send_to_dbus(char const * const icon,
+                         char const * const urgency,
+                         enum dbus_level level,
+                         char const * const summary,
+                         char const * const body,
+                         int timeout)
+{
+    DBusConnection * connection = dbus_bus_get(DBUS_BUS_SESSION, 0);
+    DBusMessage * message = dbus_message_new_method_call("org.freedesktop.Notifications",
+                                                         "/org/freedesktop/Notifications",
+                                                         "org.freedesktop.Notifications",
+                                                         "Notify");
+    DBusMessageIter iter[4];
+    dbus_message_iter_init_append(message, iter);
+    dbus_message_iter_append_basic(iter, 's', &application);
+    dbus_message_iter_append_basic(iter, 'u', &id);
+    dbus_message_iter_append_basic(iter, 's', &icon);
+    dbus_message_iter_append_basic(iter, 's', &summary);
+    dbus_message_iter_append_basic(iter, 's', &body);
+    dbus_message_iter_open_container(iter, 'a', "s", iter + 1);
+    dbus_message_iter_close_container(iter, iter + 1);
+    dbus_message_iter_open_container(iter, 'a', "{sv}", iter + 1);
+    dbus_message_iter_open_container(iter + 1, 'e', 0, iter + 2);
+    dbus_message_iter_append_basic(iter + 2, 's', &urgency);
+    dbus_message_iter_open_container(iter + 2, 'v', "y", iter + 3);
+    dbus_message_iter_append_basic(iter + 3, 'y', &level);
+    dbus_message_iter_close_container(iter + 2, iter + 3);
+    dbus_message_iter_close_container(iter + 1, iter + 2);
+    dbus_message_iter_close_container(iter, iter + 1);
+    dbus_message_iter_append_basic(iter, 'i', &timeout);
+    dbus_connection_send(connection, message, 0);
+    dbus_connection_flush(connection);
+    dbus_message_unref(message);
+    dbus_connection_unref(connection);
+
+    id++;
+}
+
+static void notify(enum dbus_level level, char const * const summary, int timeout, char const * const body)
+{
+    send_to_dbus("dialog-information", "urgency", level, summary, body, timeout);
+}
+
+__attribute__((format(printf, 4, 5))) static void notifyf(
+    enum dbus_level level, char const * const summary, int timeout, char const * const body_fmt, ...)
+{
+    va_list ap;
+    char buf[BUFSIZ];
+
+    va_start(ap, body_fmt);
+    if (vsnprintf(buf, sizeof(buf), body_fmt, ap) > 0)
+    {
+        notify(level, summary, timeout, buf);
+    }
+    va_end(ap);
+}
+
+static ssize_t get_value_index(char const * const possible_values[],
+                               size_t possible_values_size,
+                               char const * const needle,
+                               size_t needle_len)
+{
+    size_t i;
+
+    for (i = 0; i < possible_values_size; ++i)
+    {
+        if (strncmp(needle, possible_values[i], needle_len) == 0)
+        {
+            break;
+        }
+    }
+
+    if (i == possible_values_size)
+    {
+        return -1;
+    }
+
+    return i;
+}
+
+static void check_value(char const * const possible_values[],
+                        size_t possible_values_size,
+                        char const * const needle,
+                        size_t needle_len)
+{
+    if (get_value_index(possible_values, possible_values_size, needle, needle_len) == -1)
+    {
+        syslog(LOG_DAEMON | LOG_ERR, "BUG: Unknown value: %.*s", (int)needle_len, needle);
+        notifyf(DBUS_CRITICAL, "BUG", 5000, "Unknown value: %.*s", (int)needle_len, needle);
+    }
+}
+
+static enum nDPIsrvd_callback_return notifyd_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 nDPIsrvd_json_token const * const flow_event_name = TOKEN_GET_SZ(sock, "flow_event_name");
+    struct flow_user_data * flow_user_data = NULL;
+
+    if (flow != NULL)
+    {
+        flow_user_data = (struct flow_user_data *)flow->flow_user_data;
+    }
+
+    if (TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detected") != 0 ||
+        TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "detection-update") != 0 ||
+        TOKEN_VALUE_EQUALS_SZ(sock, flow_event_name, "update") != 0)
+    {
+        struct nDPIsrvd_json_token const * const flow_risks = TOKEN_GET_SZ(sock, "ndpi", "flow_risk");
+        struct nDPIsrvd_json_token const * current = NULL;
+        int next_child_index = -1, desired_severity_found = 0;
+        UT_string risks;
+
+        utstring_init(&risks);
+
+        if (flow_risks != NULL)
+        {
+            while ((current = nDPIsrvd_get_next_token(sock, flow_risks, &next_child_index)) != NULL)
+            {
+                nDPIsrvd_ull numeric_risk_value = (nDPIsrvd_ull)-1;
+                size_t flow_risk_key_len = 0;
+                char const * const flow_risk_key = TOKEN_GET_KEY(sock, current, &flow_risk_key_len);
+
+                if (flow_risk_key == NULL || flow_risk_key_len == 0)
+                {
+                    continue;
+                }
+
+                if (str_value_to_ull(flow_risk_key, &numeric_risk_value) == CONVERSION_OK && flow_user_data != NULL &&
+                    (flow_user_data->detected_risks & (1ull << numeric_risk_value)) == 0)
+                {
+                    flow_user_data->detected_risks |= (1ull << (numeric_risk_value - 1));
+
+                    char flow_risk_sz[flow_risk_key_len + 1];
+                    snprintf(flow_risk_sz, sizeof(flow_risk_sz), "%llu", numeric_risk_value);
+                    size_t flow_risk_len = 0;
+                    size_t flow_severity_len = 0;
+                    char const * const flow_risk_str =
+                        TOKEN_GET_VALUE(sock,
+                                        TOKEN_GET_SZ(sock, "ndpi", "flow_risk", flow_risk_sz, "risk"),
+                                        &flow_risk_len);
+                    char const * const flow_severity_str =
+                        TOKEN_GET_VALUE(sock,
+                                        TOKEN_GET_SZ(sock, "ndpi", "flow_risk", flow_risk_sz, "severity"),
+                                        &flow_severity_len);
+
+                    if (flow_risk_str == NULL || flow_risk_len == 0 || flow_severity_str == NULL ||
+                        flow_severity_len == 0)
+                    {
+                        continue;
+                    }
+
+                    ssize_t severity_index = get_value_index(flow_severities,
+                                                             nDPIsrvd_ARRAY_LENGTH(flow_severities),
+                                                             flow_severity_str,
+                                                             flow_severity_len);
+                    if (severity_index != -1 && desired_flow_severities[severity_index] != 0)
+                    {
+                        desired_severity_found = 1;
+                    }
+                    utstring_printf(&risks,
+                                    "Risk: '%.*s'\n"
+                                    "Severity: '%.*s'\n",
+                                    (int)flow_risk_len,
+                                    flow_risk_str,
+                                    (int)flow_severity_len,
+                                    flow_severity_str);
+                    check_value(flow_severities,
+                                nDPIsrvd_ARRAY_LENGTH(flow_severities),
+                                flow_severity_str,
+                                flow_severity_len);
+                }
+            }
+        }
+
+        {
+            size_t flow_breed_len = 0;
+            size_t flow_category_len = 0;
+            char const * const flow_breed_str =
+                TOKEN_GET_VALUE(sock, TOKEN_GET_SZ(sock, "ndpi", "breed"), &flow_breed_len);
+            char const * const flow_category_str =
+                TOKEN_GET_VALUE(sock, TOKEN_GET_SZ(sock, "ndpi", "category"), &flow_category_len);
+
+            if (flow_breed_str != NULL && flow_breed_len != 0 && flow_category_str != NULL && flow_category_len != 0)
+            {
+                ssize_t breed_index =
+                    get_value_index(flow_breeds, nDPIsrvd_ARRAY_LENGTH(flow_breeds), flow_breed_str, flow_breed_len);
+                ssize_t category_index = get_value_index(flow_categories,
+                                                         nDPIsrvd_ARRAY_LENGTH(flow_categories),
+                                                         flow_category_str,
+                                                         flow_category_len);
+
+                if ((breed_index != -1 && desired_flow_breeds[breed_index] != 0) ||
+                    (category_index != -1 && desired_flow_categories[category_index] != 0) ||
+                    desired_severity_found != 0)
+                {
+                    notifyf(DBUS_CRITICAL,
+                            "Flow Notification",
+                            5000,
+                            "Breed: '%.*s', Category: '%.*s'\n%s",
+                            (int)flow_breed_len,
+                            flow_breed_str,
+                            (int)flow_category_len,
+                            flow_category_str,
+                            (utstring_len(&risks) > 0 ? utstring_body(&risks) : "No flow risks detected\n"));
+                }
+
+                check_value(flow_breeds, nDPIsrvd_ARRAY_LENGTH(flow_breeds), flow_breed_str, flow_breed_len);
+                check_value(flow_categories,
+                            nDPIsrvd_ARRAY_LENGTH(flow_categories),
+                            flow_category_str,
+                            flow_category_len);
+            }
+            else if (desired_severity_found != 0)
+            {
+                notifyf(DBUS_CRITICAL, "Risky Flow", 5000, "%s", utstring_body(&risks));
+            }
+        }
+
+        utstring_done(&risks);
+    }
+
+    return CALLBACK_OK;
+}
+
+static void print_usage(char const * const arg0)
+{
+    static char const usage[] =
+        "Usage: %s "
+        "[-s host] [-C category...] [-B breed...] [-S severity...]\n\n"
+        "\t-s\tDestination where nDPIsrvd is listening on.\n"
+        "\t-C\tDesired nDPI category which fires a notificiation.\n"
+        "\t  \tCan be specified multiple times.\n"
+        "\t-B\tDesired nDPI breed which fires a notification.\n"
+        "\t  \tCan be specified multiple times.\n"
+        "\t-S\tDesired nDPI risk severity which fires a notification.\n"
+        "\t  \tCan be specified multiple times.\n"
+        "\n"
+        "Possible values for `-C': %s\n"
+        "Possible values for `-B': %s\n"
+        "Possible values for `-S': %s\n"
+        "\n";
+
+    UT_string flow_categories_str, flow_breeds_str, flow_severities_str;
+    utstring_init(&flow_categories_str);
+    utstring_init(&flow_breeds_str);
+    utstring_init(&flow_severities_str);
+    for (size_t i = 0; i < nDPIsrvd_ARRAY_LENGTH(flow_categories); ++i)
+    {
+        utstring_printf(&flow_categories_str, "%s, ", flow_categories[i]);
+    }
+    flow_categories_str.d[flow_categories_str.i - 2] = '\0';
+    for (size_t i = 0; i < nDPIsrvd_ARRAY_LENGTH(flow_breeds); ++i)
+    {
+        utstring_printf(&flow_breeds_str, "%s, ", flow_breeds[i]);
+    }
+    flow_breeds_str.d[flow_breeds_str.i - 2] = '\0';
+    for (size_t i = 0; i < nDPIsrvd_ARRAY_LENGTH(flow_severities); ++i)
+    {
+        utstring_printf(&flow_severities_str, "%s, ", flow_severities[i]);
+    }
+    flow_severities_str.d[flow_severities_str.i - 2] = '\0';
+    fprintf(stderr,
+            usage,
+            arg0,
+            utstring_body(&flow_categories_str),
+            utstring_body(&flow_breeds_str),
+            utstring_body(&flow_severities_str));
+    utstring_done(&flow_severities_str);
+    utstring_done(&flow_breeds_str);
+    utstring_done(&flow_categories_str);
+}
+
+static int set_defaults(void)
+{
+    char const * const default_severities[] = {"High", "Severe", "Critical", "Emergency"};
+    char const * const default_breeds[] = {"Unsafe", "Potentially Dangerous", "Dangerous", "Unrated"};
+    char const * const default_categories[] = {"Mining", "Malware", "Banned_Site", "Crypto_Currency"};
+
+    for (size_t i = 0; i < nDPIsrvd_ARRAY_LENGTH(default_severities); ++i)
+    {
+        ssize_t index = get_value_index(flow_severities,
+                                        nDPIsrvd_ARRAY_LENGTH(flow_severities),
+                                        default_severities[i],
+                                        strlen(default_severities[i]));
+        if (index == -1)
+        {
+            return 1;
+        }
+        desired_flow_severities[index] = 1;
+    }
+
+    for (size_t i = 0; i < nDPIsrvd_ARRAY_LENGTH(default_breeds); ++i)
+    {
+        ssize_t index = get_value_index(flow_breeds,
+                                        nDPIsrvd_ARRAY_LENGTH(flow_breeds),
+                                        default_breeds[i],
+                                        strlen(default_breeds[i]));
+        if (index == -1)
+        {
+            return 1;
+        }
+        desired_flow_breeds[index] = 1;
+    }
+
+    for (size_t i = 0; i < nDPIsrvd_ARRAY_LENGTH(default_categories); ++i)
+    {
+        ssize_t index = get_value_index(flow_categories,
+                                        nDPIsrvd_ARRAY_LENGTH(flow_categories),
+                                        default_categories[i],
+                                        strlen(default_categories[i]));
+        if (index == -1)
+        {
+            return 1;
+        }
+        desired_flow_categories[index] = 1;
+    }
+
+    return 0;
+}
+
+static int parse_options(int argc, char ** argv, struct nDPIsrvd_socket * const sock)
+{
+    int opt, force_defaults = 1;
+
+    while ((opt = getopt(argc, argv, "hdp:s:C:B:S:")) != -1)
+    {
+        switch (opt)
+        {
+            case 'd':
+                daemonize_enable();
+                break;
+            case 'p':
+                free(pidfile);
+                pidfile = strdup(optarg);
+                break;
+            case 's':
+                free(serv_optarg);
+                serv_optarg = strdup(optarg);
+                break;
+            case 'C':
+            {
+                ssize_t index =
+                    get_value_index(flow_categories, nDPIsrvd_ARRAY_LENGTH(flow_categories), optarg, strlen(optarg));
+                if (index == -1)
+                {
+                    fprintf(stderr, "Invalid argument for `-C': %s\n", optarg);
+                    return 1;
+                }
+                else
+                {
+                    desired_flow_categories[index] = 1;
+                }
+                force_defaults = 0;
+                break;
+            }
+            case 'B':
+            {
+                ssize_t index =
+                    get_value_index(flow_breeds, nDPIsrvd_ARRAY_LENGTH(flow_breeds), optarg, strlen(optarg));
+                if (index == -1)
+                {
+                    fprintf(stderr, "Invalid argument for `-B': %s\n", optarg);
+                    return 1;
+                }
+                else
+                {
+                    desired_flow_breeds[index] = 1;
+                }
+                force_defaults = 0;
+                break;
+            }
+            case 'S':
+            {
+                ssize_t index =
+                    get_value_index(flow_severities, nDPIsrvd_ARRAY_LENGTH(flow_severities), optarg, strlen(optarg));
+                if (index == -1)
+                {
+                    fprintf(stderr, "Invalid argument for `-S': %s\n", optarg);
+                    return 1;
+                }
+                else
+                {
+                    desired_flow_severities[index] = 1;
+                }
+                force_defaults = 0;
+                break;
+            }
+            default:
+                print_usage(argv[0]);
+                return 1;
+        }
+    }
+
+    if (force_defaults != 0 && set_defaults() != 0)
+    {
+        fprintf(stderr, "%s\n", "BUG: Could not set default values.");
+        syslog(LOG_DAEMON | LOG_ERR, "%s\n", "BUG: Could not set default values.");
+        return 1;
+    }
+
+    if (serv_optarg == NULL)
+    {
+        serv_optarg = strdup(DISTRIBUTOR_UNIX_SOCKET);
+    }
+
+    if (nDPIsrvd_setup_address(&sock->address, serv_optarg) != 0)
+    {
+        syslog(LOG_DAEMON | LOG_ERR, "Could not parse address `%s'", serv_optarg);
+        return 1;
+    }
+
+    if (optind < argc)
+    {
+        syslog(LOG_DAEMON | LOG_ERR, "%s", "Unexpected argument after options");
+        return 1;
+    }
+
+    return 0;
+}
+
+static void sighandler(int signum)
+{
+    switch (signum)
+    {
+        case SIGINT:
+            notify(DBUS_LOW, "nDPIsrvd-notifyd", 3000, "Received SIGINT, shutdown.");
+            break;
+        case SIGTERM:
+            notify(DBUS_LOW, "nDPIsrvd-notifyd", 3000, "Received SIGTERM, shutdown.");
+            break;
+        default:
+            notify(DBUS_LOW, "nDPIsrvd-notifyd", 3000, "Received unknown signal, shutdown.");
+            break;
+    }
+
+    main_thread_shutdown++;
+}
+
+int main(int argc, char ** argv)
+{
+    signal(SIGINT, sighandler);
+    signal(SIGTERM, sighandler);
+    signal(SIGPIPE, SIG_IGN);
+
+    openlog("nDPIsrvd-notifyd", LOG_CONS, LOG_DAEMON);
+
+    struct nDPIsrvd_socket * sock =
+        nDPIsrvd_socket_init(0, 0, 0, sizeof(struct flow_user_data), notifyd_json_callback, NULL, NULL);
+    if (sock == NULL)
+    {
+        syslog(LOG_DAEMON | LOG_ERR, "%s", "nDPIsrvd socket memory allocation failed!");
+        return 1;
+    }
+
+    if (parse_options(argc, argv, sock) != 0)
+    {
+        goto failure;
+    }
+
+    if (daemonize_with_pidfile(pidfile) != 0)
+    {
+        return 1;
+    }
+
+    int previous_connect_succeeded = 1;
+    do
+    {
+        if (nDPIsrvd_connect(sock) != CONNECT_OK)
+        {
+            if (previous_connect_succeeded != 0)
+            {
+                notifyf(DBUS_CRITICAL, "nDPIsrvd-notifyd", 3000, "nDPIsrvd socket connect to %s failed!", serv_optarg);
+                syslog(LOG_DAEMON | LOG_ERR, "nDPIsrvd socket connect to %s failed!", serv_optarg);
+                previous_connect_succeeded = 0;
+            }
+            nDPIsrvd_socket_close(sock);
+            sleep(1);
+            continue;
+        }
+        previous_connect_succeeded = 1;
+
+        if (nDPIsrvd_set_read_timeout(sock, 3, 0) != 0)
+        {
+            syslog(LOG_DAEMON | LOG_ERR, "nDPIsrvd set read timeout failed: %s", strerror(errno));
+            goto failure;
+        }
+
+        notifyf(DBUS_NORMAL, "nDPIsrvd-notifyd", 3000, "Connected to '%s'", serv_optarg);
+        syslog(LOG_DAEMON | LOG_NOTICE, "%s", "Initialization succeeded.");
+
+        while (main_thread_shutdown == 0)
+        {
+            enum nDPIsrvd_read_return read_ret = nDPIsrvd_read(sock);
+            if (errno == EINTR)
+            {
+                continue;
+            }
+            if (read_ret == READ_TIMEOUT)
+            {
+                continue;
+            }
+            if (read_ret != READ_OK)
+            {
+                break;
+            }
+
+            enum nDPIsrvd_parse_return parse_ret = nDPIsrvd_parse_all(sock);
+            if (parse_ret != PARSE_NEED_MORE_DATA)
+            {
+                syslog(LOG_DAEMON | LOG_ERR,
+                       "Could not parse json string %s: %.*s\n",
+                       nDPIsrvd_enum_to_string(parse_ret),
+                       nDPIsrvd_json_buffer_length(sock),
+                       nDPIsrvd_json_buffer_string(sock));
+                break;
+            }
+        }
+
+        nDPIsrvd_socket_close(sock);
+        notifyf(DBUS_NORMAL, "nDPIsrvd-notifyd", 3000, "Disconnected from '%s'.", serv_optarg);
+    } while (main_thread_shutdown == 0);
+
+failure:
+    nDPIsrvd_socket_free(&sock);
+    daemonize_shutdown(pidfile);
+    closelog();
+
+    return 0;
+}