nDPId: improved error messages if UNIX/UDP endpoint refuses connections/datagrams

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

CMakeLists.txt

@@ -160,6 +160,10 @@ endif()
 
 set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wall -Wextra")
 set(NDPID_DEFS -DJSMN_STATIC=1 -DJSMN_STRICT=1 -DJSMN_PARENT_LINKS=1)
+set(NDPID_DEPS_INC "${CMAKE_SOURCE_DIR}"
+                   "${CMAKE_SOURCE_DIR}/dependencies"
+                   "${CMAKE_SOURCE_DIR}/dependencies/jsmn"
+                   "${CMAKE_SOURCE_DIR}/dependencies/uthash/src")
 if(ENABLE_MEMORY_PROFILING)
     message(WARNING "ENABLE_MEMORY_PROFILING should not be used in production environments.")
     add_definitions("-DENABLE_MEMORY_PROFILING=1"
@@ -250,29 +254,21 @@ find_package(PCAP "1.8.1" REQUIRED)
 
 target_compile_options(nDPId PRIVATE "-pthread")
 target_compile_definitions(nDPId PRIVATE -D_GNU_SOURCE=1 -DGIT_VERSION=\"${GIT_VERSION}\" ${NDPID_DEFS} ${ZLIB_DEFS})
-target_include_directories(nDPId PRIVATE "${STATIC_LIBNDPI_INC}" "${DEFAULT_NDPI_INCLUDE}")
+target_include_directories(nDPId PRIVATE "${STATIC_LIBNDPI_INC}" "${DEFAULT_NDPI_INCLUDE}" ${NDPID_DEPS_INC})
 target_link_libraries(nDPId "${STATIC_LIBNDPI_LIB}" "${pkgcfg_lib_NDPI_ndpi}"
                             "${pkgcfg_lib_PCRE_pcre}" "${pkgcfg_lib_MAXMINDDB_maxminddb}" "${pkgcfg_lib_ZLIB_z}"
                             "${GCRYPT_LIBRARY}" "${GCRYPT_ERROR_LIBRARY}" "${PCAP_LIBRARY}" "${LIBM_LIB}"
                             "-pthread")
 
 target_compile_definitions(nDPIsrvd PRIVATE -D_GNU_SOURCE=1 -DGIT_VERSION=\"${GIT_VERSION}\" ${NDPID_DEFS})
-target_include_directories(nDPIsrvd PRIVATE
-                           "${CMAKE_SOURCE_DIR}"
-                           "${CMAKE_SOURCE_DIR}/dependencies"
-                           "${CMAKE_SOURCE_DIR}/dependencies/jsmn"
-                           "${CMAKE_SOURCE_DIR}/dependencies/uthash/src")
+target_include_directories(nDPIsrvd PRIVATE ${NDPID_DEPS_INC})
 
-target_include_directories(nDPId-test PRIVATE
-                           "${CMAKE_SOURCE_DIR}"
-                           "${CMAKE_SOURCE_DIR}/dependencies"
-                           "${CMAKE_SOURCE_DIR}/dependencies/jsmn"
-                           "${CMAKE_SOURCE_DIR}/dependencies/uthash/src")
+target_include_directories(nDPId-test PRIVATE ${NDPID_DEPS_INC})
 target_compile_options(nDPId-test PRIVATE "-Wno-unused-function" "-pthread")
 target_compile_definitions(nDPId-test PRIVATE -D_GNU_SOURCE=1 -DNO_MAIN=1 -DGIT_VERSION=\"${GIT_VERSION}\"
                                               ${NDPID_DEFS} ${ZLIB_DEFS} ${NDPID_TEST_MPROF_DEFS})
 target_include_directories(nDPId-test PRIVATE
-                           "${STATIC_LIBNDPI_INC}" "${DEFAULT_NDPI_INCLUDE}" "${CMAKE_SOURCE_DIR}/dependencies/uthash/src")
+                           "${STATIC_LIBNDPI_INC}" "${DEFAULT_NDPI_INCLUDE}" ${NDPID_DEPS_INC})
 target_link_libraries(nDPId-test "${STATIC_LIBNDPI_LIB}" "${pkgcfg_lib_NDPI_ndpi}"
                                  "${pkgcfg_lib_PCRE_pcre}" "${pkgcfg_lib_MAXMINDDB_maxminddb}" "${pkgcfg_lib_ZLIB_z}"
                                  "${GCRYPT_LIBRARY}" "${GCRYPT_ERROR_LIBRARY}" "${PCAP_LIBRARY}" "${LIBM_LIB}"
@@ -281,11 +277,7 @@ target_link_libraries(nDPId-test "${STATIC_LIBNDPI_LIB}" "${pkgcfg_lib_NDPI_ndpi
 if(BUILD_EXAMPLES)
     add_executable(nDPIsrvd-collectd examples/c-collectd/c-collectd.c)
     target_compile_definitions(nDPIsrvd-collectd PRIVATE ${NDPID_DEFS})
-    target_include_directories(nDPIsrvd-collectd PRIVATE
-                               "${CMAKE_SOURCE_DIR}"
-                               "${CMAKE_SOURCE_DIR}/dependencies"
-                               "${CMAKE_SOURCE_DIR}/dependencies/jsmn"
-                               "${CMAKE_SOURCE_DIR}/dependencies/uthash/src")
+    target_include_directories(nDPIsrvd-collectd PRIVATE ${NDPID_DEPS_INC})
 
     add_executable(nDPIsrvd-captured examples/c-captured/c-captured.c utils.c)
     if(BUILD_NDPI)
@@ -293,27 +285,18 @@ if(BUILD_EXAMPLES)
     endif()
     target_compile_definitions(nDPIsrvd-captured PRIVATE ${NDPID_DEFS})
     target_include_directories(nDPIsrvd-captured PRIVATE
-                               "${STATIC_LIBNDPI_INC}" "${DEFAULT_NDPI_INCLUDE}" "${CMAKE_SOURCE_DIR}"
-                               "${CMAKE_SOURCE_DIR}/dependencies"
-                               "${CMAKE_SOURCE_DIR}/dependencies/jsmn"
-                               "${CMAKE_SOURCE_DIR}/dependencies/uthash/src")
+                               "${STATIC_LIBNDPI_INC}" "${DEFAULT_NDPI_INCLUDE}" "${CMAKE_SOURCE_DIR}" ${NDPID_DEPS_INC})
     target_link_libraries(nDPIsrvd-captured "${pkgcfg_lib_NDPI_ndpi}"
                                             "${pkgcfg_lib_PCRE_pcre}" "${pkgcfg_lib_MAXMINDDB_maxminddb}"
                                             "${GCRYPT_LIBRARY}" "${GCRYPT_ERROR_LIBRARY}" "${PCAP_LIBRARY}")
 
     add_executable(nDPIsrvd-json-dump examples/c-json-stdout/c-json-stdout.c)
     target_compile_definitions(nDPIsrvd-json-dump PRIVATE ${NDPID_DEFS})
-    target_include_directories(nDPIsrvd-json-dump PRIVATE
-                               "${CMAKE_SOURCE_DIR}"
-                               "${CMAKE_SOURCE_DIR}/dependencies/jsmn")
+    target_include_directories(nDPIsrvd-json-dump PRIVATE ${NDPID_DEPS_INC})
 
     add_executable(nDPIsrvd-simple examples/c-simple/c-simple.c)
     target_compile_definitions(nDPIsrvd-simple PRIVATE ${NDPID_DEFS})
-    target_include_directories(nDPIsrvd-simple PRIVATE
-                               "${CMAKE_SOURCE_DIR}"
-                               "${CMAKE_SOURCE_DIR}/dependencies"
-                               "${CMAKE_SOURCE_DIR}/dependencies/jsmn"
-                               "${CMAKE_SOURCE_DIR}/dependencies/uthash/src")
+    target_include_directories(nDPIsrvd-simple PRIVATE ${NDPID_DEPS_INC})
     target_link_libraries(nDPIsrvd-simple "${pkgcfg_lib_NDPI_ndpi}"
                                           "${pkgcfg_lib_PCRE_pcre}" "${pkgcfg_lib_MAXMINDDB_maxminddb}"
                                           "${GCRYPT_LIBRARY}" "${GCRYPT_ERROR_LIBRARY}" "${PCAP_LIBRARY}")

nDPId.c

@@ -25,6 +25,7 @@
 #endif
 
 #include "config.h"
+#include "nDPIsrvd.h"
 #include "utils.h"
 
 #ifndef UNIX_PATH_MAX
@@ -354,6 +355,7 @@ static char const * const daemon_event_name_table[DAEMON_EVENT_COUNT] = {
 };
 
 static struct nDPId_reader_thread reader_threads[nDPId_MAX_READER_THREADS] = {};
+static struct nDPIsrvd_address collector_address;
 static volatile int nDPId_main_thread_shutdown = 0;
 static volatile uint64_t global_flow_id = 1;
 static int ip4_interface_avail = 0, ip6_interface_avail = 0;
@@ -388,7 +390,7 @@ static struct
     char * custom_categories_file;
     char * custom_ja3_file;
     char * custom_sha1_file;
-    char collector_sockpath[UNIX_PATH_MAX];
+    char collector_address[UNIX_PATH_MAX];
 #ifdef ENABLE_ZLIB
     uint8_t enable_zlib_compression;
 #endif
@@ -416,7 +418,7 @@ static struct
     unsigned long long int max_packets_per_flow_to_process;
 } nDPId_options = {.pidfile = nDPId_PIDFILE,
                    .user = "nobody",
-                   .collector_sockpath = COLLECTOR_UNIX_SOCKET,
+                   .collector_address = COLLECTOR_UNIX_SOCKET,
                    .max_flows_per_thread = nDPId_MAX_FLOWS_PER_THREAD / 2,
                    .max_idle_flows_per_thread = nDPId_MAX_IDLE_FLOWS_PER_THREAD / 2,
                    .tick_resolution = nDPId_TICK_RESOLUTION,
@@ -1833,7 +1835,9 @@ static void jsonize_daemon(struct nDPId_reader_thread * const reader_thread, enu
             ndpi_serialize_string_uint64(&workflow->ndpi_serializer,
                                          "total-skipped-flows",
                                          workflow->total_skipped_flows);
-            ndpi_serialize_string_uint64(&workflow->ndpi_serializer, "total-l4-payload-len", workflow->total_l4_payload_len);
+            ndpi_serialize_string_uint64(&workflow->ndpi_serializer,
+                                         "total-l4-payload-len",
+                                         workflow->total_l4_payload_len);
             ndpi_serialize_string_uint64(&workflow->ndpi_serializer,
                                          "total-not-detected-flows",
                                          workflow->total_not_detected_flows);
@@ -1906,14 +1910,13 @@ static void jsonize_flow(struct nDPId_workflow * const workflow, struct nDPId_fl
 
 static int connect_to_collector(struct nDPId_reader_thread * const reader_thread)
 {
-    struct sockaddr_un saddr;
-
     if (reader_thread->collector_sockfd >= 0)
     {
         close(reader_thread->collector_sockfd);
     }
 
-    reader_thread->collector_sockfd = socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0);
+    int sock_type = (collector_address.raw.sa_family == AF_UNIX ? SOCK_STREAM : SOCK_DGRAM);
+    reader_thread->collector_sockfd = socket(collector_address.raw.sa_family, sock_type | SOCK_CLOEXEC, 0);
     if (reader_thread->collector_sockfd < 0)
     {
         reader_thread->collector_sock_reconnect = 1;
@@ -1926,14 +1929,7 @@ static int connect_to_collector(struct nDPId_reader_thread * const reader_thread
         return 1;
     }
 
-    saddr.sun_family = AF_UNIX;
-    int written = snprintf(saddr.sun_path, sizeof(saddr.sun_path), "%s", nDPId_options.collector_sockpath);
-    if (written < 0)
-    {
-        return 1;
-    }
-
-    if (connect(reader_thread->collector_sockfd, (struct sockaddr *)&saddr, sizeof(saddr)) < 0)
+    if (connect(reader_thread->collector_sockfd, &collector_address.raw, collector_address.size) < 0)
     {
         reader_thread->collector_sock_reconnect = 1;
         return 1;
@@ -1989,9 +1985,10 @@ static void send_to_collector(struct nDPId_reader_thread * const reader_thread,
         if (connect_to_collector(reader_thread) == 0)
         {
             logger(1,
-                   "[%8llu, %zu] Reconnected to nDPIsrvd Collector",
+                   "[%8llu, %zu] Reconnected to nDPIsrvd Collector at %s",
                    workflow->packets_captured,
-                   reader_thread->array_index);
+                   reader_thread->array_index,
+                   nDPId_options.collector_address);
             jsonize_daemon(reader_thread, DAEMON_EVENT_RECONNECT);
         }
     }
@@ -2011,7 +2008,19 @@ static void send_to_collector(struct nDPId_reader_thread * const reader_thread,
         }
         if (saved_errno != EAGAIN)
         {
-            reader_thread->collector_sock_reconnect = 1;
+            if (saved_errno == ECONNREFUSED)
+            {
+                logger(1,
+                       "[%8llu, %zu] %s to %s refused by endpoint",
+                       workflow->packets_captured,
+                       reader_thread->array_index,
+                       (collector_address.raw.sa_family == AF_UNIX ? "Connection" : "Datagram"),
+                       nDPId_options.collector_address);
+            }
+            if (collector_address.raw.sa_family == AF_UNIX)
+            {
+                reader_thread->collector_sock_reconnect = 1;
+            }
         }
         else
         {
@@ -2025,11 +2034,15 @@ static void send_to_collector(struct nDPId_reader_thread * const reader_thread,
                 if (written < 0)
                 {
                     logger(1,
-                           "[%8llu, %zu] Send data (blocking I/O) to nDPIsrvd Collector failed: %s",
+                           "[%8llu, %zu] Send data (blocking I/O) to nDPIsrvd Collector at %s failed: %s",
                            workflow->packets_captured,
                            reader_thread->array_index,
+                           nDPId_options.collector_address,
                            strerror(saved_errno));
-                    reader_thread->collector_sock_reconnect = 1;
+                    if (collector_address.raw.sa_family == AF_UNIX)
+                    {
+                        reader_thread->collector_sock_reconnect = 1;
+                    }
                     break;
                 }
                 else
@@ -3995,7 +4008,7 @@ static void * processing_thread(void * const ndpi_thread_arg)
         logger(1,
                "Thread %zu: Could not connect to nDPIsrvd Collector at %s, will try again later. Error: %s",
                reader_thread->array_index,
-               nDPId_options.collector_sockpath,
+               nDPId_options.collector_address,
                (errno != 0 ? strerror(errno) : "Internal Error."));
     }
     else
@@ -4344,7 +4357,7 @@ static int nDPId_parse_options(int argc, char ** argv)
         "Usage: %s "
         "[-i pcap-file/interface] [-I] [-E] [-B bpf-filter]\n"
         "\t  \t"
-        "[-l] [-L logfile] [-c path-to-unix-sock] "
+        "[-l] [-L logfile] [-c address] "
         "[-d] [-p pidfile]\n"
         "\t  \t"
         "[-u user] [-g group] "
@@ -4361,7 +4374,7 @@ static int nDPId_parse_options(int argc, char ** argv)
         "\t-B\tSet an optional PCAP filter string. (BPF format)\n"
         "\t-l\tLog all messages to stderr.\n"
         "\t-L\tLog all messages to a log file.\n"
-        "\t-c\tPath to the UNIX socket (nDPIsrvd Collector).\n"
+        "\t-c\tPath to a UNIX socket (nDPIsrvd Collector) or a custom UDP endpoint.\n"
         "\t-d\tForking into background after initialization.\n"
         "\t-p\tWrite the daemon PID to the given file path.\n"
         "\t-u\tChange UID to the numeric value of user.\n"
@@ -4384,7 +4397,7 @@ static int nDPId_parse_options(int argc, char ** argv)
         "\t-v\tversion\n"
         "\t-h\tthis\n\n";
 
-    while ((opt = getopt(argc, argv, "hi:IEB:lL:c:dp:u:g:P:C:J:S:a:zo:vh")) != -1)
+    while ((opt = getopt(argc, argv, "i:IEB:lL:c:dp:u:g:P:C:J:S:a:zo:vh")) != -1)
     {
         switch (opt)
         {
@@ -4410,8 +4423,8 @@ static int nDPId_parse_options(int argc, char ** argv)
                 }
                 break;
             case 'c':
-                strncpy(nDPId_options.collector_sockpath, optarg, sizeof(nDPId_options.collector_sockpath) - 1);
-                nDPId_options.collector_sockpath[sizeof(nDPId_options.collector_sockpath) - 1] = '\0';
+                strncpy(nDPId_options.collector_address, optarg, sizeof(nDPId_options.collector_address) - 1);
+                nDPId_options.collector_address[sizeof(nDPId_options.collector_address) - 1] = '\0';
                 break;
             case 'd':
                 daemonize_enable();
@@ -4594,20 +4607,10 @@ static int validate_options(void)
         }
     }
 #endif
-    if (is_path_absolute("Collector socket", nDPId_options.collector_sockpath) != 0)
+    if (nDPIsrvd_setup_address(&collector_address, nDPId_options.collector_address) != 0)
     {
         retval = 1;
-    }
-    {
-        struct sockaddr_un saddr;
-        if (strlen(nDPId_options.collector_sockpath) >= sizeof(saddr.sun_path))
-        {
-            logger_early(1,
-                         "Collector socket path too long, current/max: %zu/%zu",
-                         strlen(nDPId_options.collector_sockpath),
-                         sizeof(saddr.sun_path) - 1);
-            retval = 1;
-        }
+        logger_early(1, "Collector socket invalid address: %s.", nDPId_options.collector_address);
     }
     if (nDPId_options.instance_alias == NULL)
     {