Network viewer: filter by username (#16911)

* add users to network-connections function

* cleaner and simpler filters; removed IPs and ports, added serverPort

3 files changed, 1369 insertions, 0 deletions

diff --git a/src/libnetdata/maps/local-sockets.h b/src/libnetdata/maps/local-sockets.h
new file mode 100644
index 0000000000..70282259f2
--- /dev/null
+++ b/src/libnetdata/maps/local-sockets.h
@@ -0,0 +1,1235 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#ifndef NETDATA_LOCAL_SOCKETS_H
+#define NETDATA_LOCAL_SOCKETS_H
+
+#include "libnetdata/libnetdata.h"
+
+// disable libmnl for the moment
+#undef HAVE_LIBMNL
+
+#ifdef HAVE_LIBMNL
+#include <linux/inet_diag.h>
+#include <linux/sock_diag.h>
+#include <linux/unix_diag.h>
+#include <linux/netlink.h>
+#include <libmnl/libmnl.h>
+#endif
+
+#define UID_UNSET (uid_t)(UINT32_MAX)
+
+// --------------------------------------------------------------------------------------------------------------------
+// hashtable for keeping the namespaces
+// key and value is the namespace inode
+
+#define SIMPLE_HASHTABLE_VALUE_TYPE uint64_t
+#define SIMPLE_HASHTABLE_NAME _NET_NS
+#include "libnetdata/simple_hashtable.h"
+
+// --------------------------------------------------------------------------------------------------------------------
+// hashtable for keeping the sockets of PIDs
+// key is the inode
+
+struct pid_socket;
+#define SIMPLE_HASHTABLE_VALUE_TYPE struct pid_socket
+#define SIMPLE_HASHTABLE_NAME _PID_SOCKET
+#include "libnetdata/simple_hashtable.h"
+
+// --------------------------------------------------------------------------------------------------------------------
+// hashtable for keeping all the sockets
+// key is the inode
+
+struct local_socket;
+#define SIMPLE_HASHTABLE_VALUE_TYPE struct local_socket
+#define SIMPLE_HASHTABLE_NAME _LOCAL_SOCKET
+#include "libnetdata/simple_hashtable.h"
+
+// --------------------------------------------------------------------------------------------------------------------
+// hashtable for keeping all local IPs
+// key is XXH3_64bits hash of the IP
+
+union ipv46;
+#define SIMPLE_HASHTABLE_VALUE_TYPE union ipv46
+#define SIMPLE_HASHTABLE_NAME _LOCAL_IP
+#include "libnetdata/simple_hashtable.h"
+
+// --------------------------------------------------------------------------------------------------------------------
+// hashtable for keeping all listening ports
+// key is XXH3_64bits hash of the family, protocol, port number, namespace
+
+struct local_port;
+#define SIMPLE_HASHTABLE_VALUE_TYPE struct local_port
+#define SIMPLE_HASHTABLE_NAME _LISTENING_PORT
+#include "libnetdata/simple_hashtable.h"
+
+// --------------------------------------------------------------------------------------------------------------------
+
+struct local_socket_state;
+typedef void (*local_sockets_cb_t)(struct local_socket_state *state, struct local_socket *n, void *data);
+
+typedef struct local_socket_state {
+    struct {
+        bool listening;
+        bool inbound;
+        bool outbound;
+        bool local;
+        bool tcp4;
+        bool tcp6;
+        bool udp4;
+        bool udp6;
+        bool pid;
+        bool cmdline;
+        bool comm;
+        bool uid;
+        bool namespaces;
+        size_t max_errors;
+
+        local_sockets_cb_t cb;
+        void *data;
+
+        const char *host_prefix;
+    } config;
+
+    struct {
+        size_t pid_fds_processed;
+        size_t pid_fds_opendir_failed;
+        size_t pid_fds_readlink_failed;
+        size_t pid_fds_parse_failed;
+        size_t errors_encountered;
+    } stats;
+
+#ifdef HAVE_LIBMNL
+    bool use_nl;
+    struct mnl_socket *nl;
+    uint16_t tmp_protocol;
+#endif
+
+    uint64_t proc_self_net_ns_inode;
+
+    SIMPLE_HASHTABLE_NET_NS ns_hashtable;
+    SIMPLE_HASHTABLE_PID_SOCKET pid_sockets_hashtable;
+    SIMPLE_HASHTABLE_LOCAL_SOCKET sockets_hashtable;
+    SIMPLE_HASHTABLE_LOCAL_IP local_ips_hashtable;
+    SIMPLE_HASHTABLE_LISTENING_PORT listening_ports_hashtable;
+} LS_STATE;
+
+// --------------------------------------------------------------------------------------------------------------------
+
+typedef enum __attribute__((packed)) {
+    SOCKET_DIRECTION_NONE = 0,
+    SOCKET_DIRECTION_LISTEN = (1 << 0),         // a listening socket
+    SOCKET_DIRECTION_INBOUND = (1 << 1),        // an inbound socket connecting a remote system to a local listening socket
+    SOCKET_DIRECTION_OUTBOUND = (1 << 2),       // a socket initiated by this system, connecting to another system
+    SOCKET_DIRECTION_LOCAL_INBOUND = (1 << 3),  // the socket connecting 2 localhost applications
+    SOCKET_DIRECTION_LOCAL_OUTBOUND = (1 << 4), // the socket connecting 2 localhost applications
+} SOCKET_DIRECTION;
+
+#ifndef TASK_COMM_LEN
+#define TASK_COMM_LEN 16
+#endif
+
+struct pid_socket {
+    uint64_t inode;
+    pid_t pid;
+    uid_t uid;
+    uint64_t net_ns_inode;
+    char *cmdline;
+    char comm[TASK_COMM_LEN];
+};
+
+struct local_port {
+    uint16_t protocol;
+    uint16_t family;
+    uint16_t port;
+    uint64_t net_ns_inode;
+};
+
+union ipv46 {
+    uint32_t ipv4;
+    struct in6_addr ipv6;
+};
+
+struct socket_endpoint {
+    uint16_t protocol;
+    uint16_t family;
+    uint16_t port;
+    union ipv46 ip;
+};
+
+static inline void ipv6_to_in6_addr(const char *ipv6_str, struct in6_addr *d) {
+    char buf[9];
+
+    for (size_t k = 0; k < 4; ++k) {
+        memcpy(buf, ipv6_str + (k * 8), 8);
+        buf[sizeof(buf) - 1] = '\0';
+        d->s6_addr32[k] = strtoul(buf, NULL, 16);
+    }
+}
+
+typedef struct local_socket {
+    uint64_t inode;
+    uint64_t net_ns_inode;
+
+    int state;
+    struct socket_endpoint local;
+    struct socket_endpoint remote;
+    pid_t pid;
+
+    SOCKET_DIRECTION direction;
+
+    uint8_t timer;
+    uint8_t retransmits;
+    uint32_t expires;
+    uint32_t rqueue;
+    uint32_t wqueue;
+    uid_t uid;
+
+    char comm[TASK_COMM_LEN];
+    char *cmdline;
+
+    struct local_port local_port_key;
+
+    XXH64_hash_t local_ip_hash;
+    XXH64_hash_t remote_ip_hash;
+    XXH64_hash_t local_port_hash;
+} LOCAL_SOCKET;
+
+// --------------------------------------------------------------------------------------------------------------------
+
+static inline void local_sockets_log(LS_STATE *ls, const char *format, ...) __attribute__ ((format(__printf__, 2, 3)));
+static inline void local_sockets_log(LS_STATE *ls, const char *format, ...) {
+    if(++ls->stats.errors_encountered == ls->config.max_errors) {
+        nd_log(NDLS_COLLECTORS, NDLP_ERR, "LOCAL-SOCKETS: max number of logs reached. Not logging anymore");
+        return;
+    }
+
+    if(ls->stats.errors_encountered > ls->config.max_errors)
+        return;
+
+    char buf[16384];
+    va_list args;
+    va_start(args, format);
+    vsnprintf(buf, sizeof(buf), format, args);
+    va_end(args);
+
+    nd_log(NDLS_COLLECTORS, NDLP_ERR, "LOCAL-SOCKETS: %s", buf);
+}
+
+// --------------------------------------------------------------------------------------------------------------------
+
+static void local_sockets_foreach_local_socket_call_cb(LS_STATE *ls) {
+    for(SIMPLE_HASHTABLE_SLOT_LOCAL_SOCKET *sl = simple_hashtable_first_read_only_LOCAL_SOCKET(&ls->sockets_hashtable);
+         sl;
+         sl = simple_hashtable_next_read_only_LOCAL_SOCKET(&ls->sockets_hashtable, sl)) {
+        LOCAL_SOCKET *n = SIMPLE_HASHTABLE_SLOT_DATA(sl);
+        if(!n) continue;
+
+        if((ls->config.listening && n->direction & SOCKET_DIRECTION_LISTEN) ||
+            (ls->config.local && n->direction & (SOCKET_DIRECTION_LOCAL_INBOUND|SOCKET_DIRECTION_LOCAL_OUTBOUND)) ||
+            (ls->config.inbound && n->direction & SOCKET_DIRECTION_INBOUND) ||
+            (ls->config.outbound && n->direction & SOCKET_DIRECTION_OUTBOUND)
+        ) {
+            // we have to call the callback for this socket
+            if (ls->config.cb)
+                ls->config.cb(ls, n, ls->config.data);
+        }
+    }
+}
+
+// --------------------------------------------------------------------------------------------------------------------
+
+static inline void local_sockets_fix_cmdline(char* str) {
+    char *s = str;
+
+    // map invalid characters to underscores
+    while(*s) {
+        if(*s == '|' || iscntrl(*s)) *s = '_';
+        s++;
+    }
+}
+
+// --------------------------------------------------------------------------------------------------------------------
+
+static inline bool
+local_sockets_read_proc_inode_link(LS_STATE *ls, const char *filename, uint64_t *inode, const char *type) {
+    char link_target[FILENAME_MAX + 1];
+
+    *inode = 0;
+
+    ssize_t len = readlink(filename, link_target, sizeof(link_target) - 1);
+    if (len == -1) {
+        local_sockets_log(ls, "cannot read '%s' link '%s'", type, filename);
+
+        ls->stats.pid_fds_readlink_failed++;
+        return false;
+    }
+    link_target[len] = '\0';
+
+    len = strlen(type);
+    if(strncmp(link_target, type, len) == 0 && link_target[len] == ':' && link_target[len + 1] == '[' && isdigit(link_target[len + 2])) {
+        *inode = strtoull(&link_target[len + 2], NULL, 10);
+        // ll_log(ls, "read link of type '%s' '%s' from '%s', inode = %"PRIu64, type, link_target, filename, *inode);
+        return true;
+    }
+    else {
+        // ll_log(ls, "cannot read '%s' link '%s' from '%s'", type, link_target, filename);
+        ls->stats.pid_fds_processed++;
+        return false;
+    }
+}
+
+static inline bool local_sockets_is_path_a_pid(const char *s) {
+    if(!s || !*s) return false;
+
+    while(*s) {
+        if(!isdigit(*s++))
+            return false;
+    }
+
+    return true;
+}
+
+static inline bool local_sockets_find_all_sockets_in_proc(LS_STATE *ls, const char *proc_filename) {
+    DIR *proc_dir;
+    struct dirent *proc_entry;
+    char filename[FILENAME_MAX + 1];
+    char comm[TASK_COMM_LEN];
+    char cmdline[8192];
+    const char *cmdline_trimmed;
+    uint64_t net_ns_inode;
+
+    proc_dir = opendir(proc_filename);
+    if (proc_dir == NULL) {
+        local_sockets_log(ls, "cannot opendir() '%s'", proc_filename);
+        ls->stats.pid_fds_readlink_failed++;
+        return false;
+    }
+
+    while ((proc_entry = readdir(proc_dir)) != NULL) {
+        if(proc_entry->d_type != DT_DIR)
+            continue;
+
+        if(!strcmp(proc_entry->d_name, ".") || !strcmp(proc_entry->d_name, ".."))
+            continue;
+
+        if(!local_sockets_is_path_a_pid(proc_entry->d_name))
+            continue;
+
+        // Build the path to the fd directory of the process
+        snprintfz(filename, FILENAME_MAX, "%s/%s/fd/", proc_filename, proc_entry->d_name);
+        DIR *fd_dir = opendir(filename);
+        if (fd_dir == NULL) {
+            local_sockets_log(ls, "cannot opendir() '%s'", filename);
+            ls->stats.pid_fds_opendir_failed++;
+            continue;
+        }
+
+        comm[0] = '\0';
+        cmdline[0] = '\0';
+        cmdline_trimmed = NULL;
+        pid_t pid = (pid_t)strtoul(proc_entry->d_name, NULL, 10);
+        if(!pid) {
+            local_sockets_log(ls, "cannot parse pid of '%s'", proc_entry->d_name);
+            closedir(fd_dir);
+            continue;
+        }
+        net_ns_inode = 0;
+        uid_t uid = UID_UNSET;
+
+        struct dirent *fd_entry;
+        while ((fd_entry = readdir(fd_dir)) != NULL) {
+            if(fd_entry->d_type != DT_LNK)
+                continue;
+
+            snprintfz(filename, sizeof(filename), "%s/%s/fd/%s", proc_filename, proc_entry->d_name, fd_entry->d_name);
+            uint64_t inode = 0;
+            if(!local_sockets_read_proc_inode_link(ls, filename, &inode, "socket"))
+                continue;
+
+            SIMPLE_HASHTABLE_SLOT_PID_SOCKET *sl = simple_hashtable_get_slot_PID_SOCKET(&ls->pid_sockets_hashtable, inode, &inode, true);
+            struct pid_socket *ps = SIMPLE_HASHTABLE_SLOT_DATA(sl);
+            if(!ps || (ps->pid == 1 && pid != 1)) {
+                if(uid == UID_UNSET && ls->config.uid) {
+                    char status_buf[512];
+                    snprintfz(filename, sizeof(filename), "%s/%s/status", proc_filename, proc_entry->d_name);
+                    if (read_txt_file(filename, status_buf, sizeof(status_buf)))
+                        local_sockets_log(ls, "cannot open file: %s\n", filename);
+                    else {
+                        char *u = strstr(status_buf, "Uid:");
+                        if(u) {
+                            u += 4;
+                            while(isspace(*u)) u++;                     // skip spaces
+                            while(*u >= '0' && *u <= '9') u++;          // skip the first number (real uid)
+                            while(isspace(*u)) u++;                     // skip spaces again
+                            uid = strtol(u, NULL, 10);   // parse the 2nd number (effective uid)
+                        }
+                    }
+                }
+                if(!comm[0] && ls->config.comm) {
+                    snprintfz(filename, sizeof(filename), "%s/%s/comm", proc_filename, proc_entry->d_name);
+                    if (read_txt_file(filename, comm, sizeof(comm)))
+                        local_sockets_log(ls, "cannot open file: %s\n", filename);
+                    else {
+                        size_t clen = strlen(comm);
+                        if(comm[clen - 1] == '\n')
+                            comm[clen - 1] = '\0';
+                    }
+                }
+                if(!cmdline[0] && ls->config.cmdline) {
+                    snprintfz(filename, sizeof(filename), "%s/%s/cmdline", proc_filename, proc_entry->d_name);
+                    if (read_proc_cmdline(filename, cmdline, sizeof(cmdline)))
+                        local_sockets_log(ls, "cannot open file: %s\n", filename);
+                    else {
+                        local_sockets_fix_cmdline(cmdline);
+                        cmdline_trimmed = trim(cmdline);
+                    }
+                }
+                if(!net_ns_inode && ls->config.namespaces) {
+                    snprintfz(filename, sizeof(filename), "%s/%s/ns/net", proc_filename, proc_entry->d_name);
+                    if(local_sockets_read_proc_inode_link(ls, filename, &net_ns_inode, "net")) {
+                        SIMPLE_HASHTABLE_SLOT_NET_NS *sl_ns = simple_hashtable_get_slot_NET_NS(&ls->ns_hashtable, net_ns_inode, (uint64_t *)net_ns_inode, true);
+                        simple_hashtable_set_slot_NET_NS(&ls->ns_hashtable, sl_ns, net_ns_inode, (uint64_t *)net_ns_inode);
+                    }
+                }
+
+                if(!ps)
+                    ps = callocz(1, sizeof(*ps));
+
+                ps->inode = inode;
+                ps->pid = pid;
+                ps->uid = uid;
+                ps->net_ns_inode = net_ns_inode;
+                strncpyz(ps->comm, comm, sizeof(ps->comm) - 1);
+
+                if(ps->cmdline)
+                    freez(ps->cmdline);
+
+                ps->cmdline = cmdline_trimmed ? strdupz(cmdline_trimmed) : NULL;
+                simple_hashtable_set_slot_PID_SOCKET(&ls->pid_sockets_hashtable, sl, inode, ps);
+            }
+        }
+
+        closedir(fd_dir);
+    }
+
+    closedir(proc_dir);
+    return true;
+}
+
+// --------------------------------------------------------------------------------------------------------------------
+
+static bool local_sockets_is_ipv4_mapped_ipv6_address(const struct in6_addr *addr) {
+    // An IPv4-mapped IPv6 address starts with 80 bits of zeros followed by 16 bits of ones
+    static const unsigned char ipv4_mapped_prefix[12] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xFF };
+    return memcmp(addr->s6_addr, ipv4_mapped_prefix, 12) == 0;
+}
+
+static bool local_sockets_is_loopback_address(struct socket_endpoint *se) {
+    if (se->family == AF_INET) {
+        // For IPv4, loopback addresses are in the 127.0.0.0/8 range
+        return (ntohl(se->ip.ipv4) >> 24) == 127; // Check if the first byte is 127
+    } else if (se->family == AF_INET6) {
+        // Check if the address is an IPv4-mapped IPv6 address
+        if (local_sockets_is_ipv4_mapped_ipv6_address(&se->ip.ipv6)) {
+            // Extract the last 32 bits (IPv4 address) and check if it's in the 127.0.0.0/8 range
+            uint8_t *ip6 = (uint8_t *)&se->ip.ipv6;
+            const uint32_t ipv4_addr = *((const uint32_t *)(ip6 + 12));
+            return (ntohl(ipv4_addr) >> 24) == 127;
+        }
+
+        // For IPv6, loopback address is ::1
+        return memcmp(&se->ip.ipv6, &in6addr_loopback, sizeof(se->ip.ipv6)) == 0;
+    }
+
+    return false;
+}
+
+static inline bool local_sockets_is_ipv4_reserved_address(uint32_t ip) {
+    // Check for the reserved address ranges
+    ip = ntohl(ip);
+    return (
+            (ip >> 24 == 10) ||                         // Private IP range (A class)
+            (ip >> 20 == (172 << 4) + 1) ||             // Private IP range (B class)
+            (ip >> 16 == (192 << 8) + 168) ||           // Private IP range (C class)
+            (ip >> 24 == 127) ||                        // Loopback address (127.0.0.0)
+            (ip >> 24 == 0) ||                          // Reserved (0.0.0.0)
+            (ip >> 24 == 169 && (ip >> 16) == 254) ||   // Link-local address (169.254.0.0)
+            (ip >> 16 == (192 << 8) + 0)                // Test-Net (192.0.0.0)
+            );
+}
+
+static inline bool local_sockets_is_private_address(struct socket_endpoint *se) {
+    if (se->family == AF_INET) {
+        return local_sockets_is_ipv4_reserved_address(se->ip.ipv4);
+    }
+    else if (se->family == AF_INET6) {
+        uint8_t *ip6 = (uint8_t *)&se->ip.ipv6;
+
+        // Check if the address is an IPv4-mapped IPv6 address
+        if (local_sockets_is_ipv4_mapped_ipv6_address(&se->ip.ipv6)) {
+            // Extract the last 32 bits (IPv4 address) and check if it's in the 127.0.0.0/8 range
+            const uint32_t ipv4_addr = *((const uint32_t *)(ip6 + 12));
+            return local_sockets_is_ipv4_reserved_address(ipv4_addr);
+        }
+
+        // Check for link-local addresses (fe80::/10)
+        if ((ip6[0] == 0xFE) && ((ip6[1] & 0xC0) == 0x80))
+            return true;
+
+        // Check for Unique Local Addresses (ULA) (fc00::/7)
+        if ((ip6[0] & 0xFE) == 0xFC)
+            return true;
+
+        // Check for multicast addresses (ff00::/8)
+        if (ip6[0] == 0xFF)
+            return true;
+
+        // For IPv6, loopback address is :: or ::1
+        return memcmp(&se->ip.ipv6, &in6addr_any, sizeof(se->ip.ipv6)) == 0 ||
+               memcmp(&se->ip.ipv6, &in6addr_loopback, sizeof(se->ip.ipv6)) == 0;
+    }
+
+    return false;
+}
+
+static bool local_sockets_is_multicast_address(struct socket_endpoint *se) {
+    if (se->family == AF_INET) {
+        // For IPv4, check if the address is 0.0.0.0
+        uint32_t ip = htonl(se->ip.ipv4);
+        return (ip >= 0xE0000000 && ip <= 0xEFFFFFFF);   // Multicast address range (224.0.0.0/4)
+    }
+    else if (se->family == AF_INET6) {
+        // For IPv6, check if the address is ff00::/8
+        uint8_t *ip6 = (uint8_t *)&se->ip.ipv6;
+        return ip6[0] == 0xff;
+    }
+
+    return false;
+}
+
+static bool local_sockets_is_zero_address(struct socket_endpoint *se) {
+    if (se->family == AF_INET) {
+        // For IPv4, check if the address is 0.0.0.0
+        return se->ip.ipv4 == 0;
+    }
+    else if (se->family == AF_INET6) {
+        // For IPv6, check if the address is ::
+        return memcmp(&se->ip.ipv6, &in6addr_any, sizeof(se->ip.ipv6)) == 0;
+    }
+
+    return false;
+}
+
+static inline const char *local_sockets_address_space(struct socket_endpoint *se) {
+    if(local_sockets_is_zero_address(se))
+        return "zero";
+    else if(local_sockets_is_loopback_address(se))
+        return "loopback";
+    else if(local_sockets_is_multicast_address(se))
+        return "multicast";
+    else if(local_sockets_is_private_address(se))
+        return "private";
+    else
+        return "public";
+}
+
+// --------------------------------------------------------------------------------------------------------------------
+
+static inline void local_sockets_index_listening_port(LS_STATE *ls, LOCAL_SOCKET *n) {
+    if(n->direction & SOCKET_DIRECTION_LISTEN) {
+        // for the listening sockets, keep a hashtable with all the local ports
+        // so that we will be able to detect INBOUND sockets
+
+        SIMPLE_HASHTABLE_SLOT_LISTENING_PORT *sl_port =
+            simple_hashtable_get_slot_LISTENING_PORT(&ls->listening_ports_hashtable, n->local_port_hash, &n->local_port_key, true);
+
+        struct local_port *port = SIMPLE_HASHTABLE_SLOT_DATA(sl_port);
+        if(!port)
+            simple_hashtable_set_slot_LISTENING_PORT(&ls->listening_ports_hashtable, sl_port, n->local_port_hash, &n->local_port_key);
+    }
+}
+
+static inline bool local_sockets_add_socket(LS_STATE *ls, LOCAL_SOCKET *tmp) {
+    if(!tmp->inode) return false;
+
+    SIMPLE_HASHTABLE_SLOT_LOCAL_SOCKET *sl = simple_hashtable_get_slot_LOCAL_SOCKET(&ls->sockets_hashtable, tmp->inode, &tmp->inode, true);
+    LOCAL_SOCKET *n = SIMPLE_HASHTABLE_SLOT_DATA(sl);
+    if(n) {
+        local_sockets_log(ls, "inode %" PRIu64" already exists in hashtable - ignoring duplicate", tmp->inode);
+        return false;
+    }
+
+    n = (LOCAL_SOCKET *)callocz(1, sizeof(LOCAL_SOCKET));
+    *n = *tmp; // copy all contents
+
+    // fix the key
+    n->local_port_key.port = n->local.port;
+    n->local_port_key.family = n->local.family;
+    n->local_port_key.protocol = n->local.protocol;
+    n->local_port_key.net_ns_inode = ls->proc_self_net_ns_inode;
+
+    n->local_ip_hash = XXH3_64bits(&n->local.ip, sizeof(n->local.ip));
+    n->remote_ip_hash = XXH3_64bits(&n->remote.ip, sizeof(n->remote.ip));
+    n->local_port_hash = XXH3_64bits(&n->local_port_key, sizeof(n->local_port_key));
+
+    // --- look up a pid for it -----------------------------------------------------------------------------------
+
+    SIMPLE_HASHTABLE_SLOT_PID_SOCKET *sl_pid = simple_hashtable_get_slot_PID_SOCKET(&ls->pid_sockets_hashtable, n->inode, &n->inode, false);
+    struct pid_socket *ps = SIMPLE_HASHTABLE_SLOT_DATA(sl_pid);
+    if(ps) {
+        n->net_ns_inode = ps->net_ns_inode;
+        n->pid = ps->pid;
+
+        if(ps->uid != UID_UNSET && n->uid == UID_UNSET)
+            n->uid = ps->uid;
+
+        if(ps->cmdline)
+            n->cmdline = strdupz(ps->cmdline);
+        strncpyz(n->comm, ps->comm, sizeof(n->comm) - 1);
+    }
+
+    // --- index it -----------------------------------------------------------------------------------------------
+
+    simple_hashtable_set_slot_LOCAL_SOCKET(&ls->sockets_hashtable, sl, n->inode, n);
+
+    if(!local_sockets_is_zero_address(&n->local)) {
+        // put all the local IPs into the local_ips hashtable
+        // so, we learn all local IPs the system has
+
+        SIMPLE_HASHTABLE_SLOT_LOCAL_IP *sl_ip =
+            simple_hashtable_get_slot_LOCAL_IP(&ls->local_ips_hashtable, n->local_ip_hash, &n->local.ip, true);
+
+        union ipv46 *ip = SIMPLE_HASHTABLE_SLOT_DATA(sl_ip);
+        if(!ip)
+            simple_hashtable_set_slot_LOCAL_IP(&ls->local_ips_hashtable, sl_ip, n->local_ip_hash, &n->local.ip);
+    }
+
+    // --- 1st phase for direction detection ----------------------------------------------------------------------
+
+    if((n->local.protocol == IPPROTO_TCP && n->state == TCP_LISTEN) ||
+        local_sockets_is_zero_address(&n->local) ||
+        local_sockets_is_zero_address(&n->remote)) {
+        // the socket is either in a TCP LISTEN, or
+        // the remote address is zero
+        n->direction |= SOCKET_DIRECTION_LISTEN;
+    }
+    else {
+        // we can't say yet if it is inbound or outboud
+        // so, mark it as both inbound and outbound
+        n->direction |= SOCKET_DIRECTION_INBOUND | SOCKET_DIRECTION_OUTBOUND;
+    }
+
+    // --- index it in LISTENING_PORT -----------------------------------------------------------------------------
+
+    local_sockets_index_listening_port(ls, n);
+
+    return true;
+}
+
+#ifdef HAVE_LIBMNL
+
+static inline void local_sockets_netlink_init(LS_STATE *ls) {
+    ls->use_nl = true;
+    ls->nl = mnl_socket_open(NETLINK_INET_DIAG);
+    if (!ls->nl) {
+        local_sockets_log(ls, "cannot open netlink socket");
+        ls->use_nl = false;
+    }
+
+    if (mnl_socket_bind(ls->nl, 0, MNL_SOCKET_AUTOPID) < 0) {
+        local_sockets_log(ls, "cannot bind netlink socket");
+        ls->use_nl = false;
+    }
+}
+
+static inline void local_sockets_netlink_cleanup(LS_STATE *ls) {
+    if(ls->nl) {
+        mnl_socket_close(ls->nl);
+        ls->nl = NULL;
+    }
+}
+
+static inline int local_sockets_netlink_cb_data(const struct nlmsghdr *nlh, void *data) {
+    LS_STATE *ls = data;
+
+    struct inet_diag_msg *diag_msg = mnl_nlmsg_get_payload(nlh);
+
+    LOCAL_SOCKET n = {
+        .inode = diag_msg->idiag_inode,
+        .direction = SOCKET_DIRECTION_NONE,
+        .state = diag_msg->idiag_state,
+        .local = {
+            .protocol = ls->tmp_protocol,
+            .family = diag_msg->idiag_family,
+            .port = diag_msg->id.idiag_sport,
+        },
+        .remote = {
+            .protocol = ls->tmp_protocol,
+            .family = diag_msg->idiag_family,
+            .port = diag_msg->id.idiag_dport,
+        },
+        .timer = diag_msg->idiag_timer,
+        .retransmits = diag_msg->idiag_retrans,
+        .expires = diag_msg->idiag_expires,
+        .rqueue = diag_msg->idiag_rqueue,
+        .wqueue = diag_msg->idiag_wqueue,
+        .uid = diag_msg->idiag_uid,
+    };
+
+    if (diag_msg->idiag_family == AF_INET) {
+        memcpy(&n.local.ip.ipv4, diag_msg->id.idiag_src, sizeof(n.local.ip.ipv4));
+        memcpy(&n.remote.ip.ipv4, diag_msg->id.idiag_dst, sizeof(n.remote.ip.ipv4));
+    }
+    else if (diag_msg->idiag_family == AF_INET6) {
+        memcpy(&n.local.ip.ipv6, diag_msg->id.idiag_src, sizeof(n.local.ip.ipv6));
+        memcpy(&n.remote.ip.ipv6, diag_msg->id.idiag_dst, sizeof(n.remote.ip.ipv6));
+    }
+
+    local_sockets_add_socket(ls, &n);
+
+    return MNL_CB_OK;
+}
+
+static inline bool local_sockets_netlink_get_sockets(LS_STATE *ls, uint16_t family, uint16_t protocol) {
+    ls->tmp_protocol = protocol;
+
+    char buf[MNL_SOCKET_BUFFER_SIZE];
+    struct nlmsghdr *nlh;
+    struct inet_diag_req_v2 req;
+    unsigned int seq, portid = mnl_socket_get_portid(ls->nl);
+
+    memset(&req, 0, sizeof(req));
+    req.sdiag_family = family;
+    req.sdiag_protocol = protocol;
+    req.idiag_states = -1;
+
+    nlh = mnl_nlmsg_put_header(buf);
+    nlh->nlmsg_type = SOCK_DIAG_BY_FAMILY;
+    nlh->nlmsg_flags = NLM_F_DUMP | NLM_F_REQUEST;
+    nlh->nlmsg_seq = seq = time(NULL);
+    mnl_nlmsg_put_extra_header(nlh, sizeof(req));
+    memcpy(mnl_nlmsg_get_payload(nlh), &req, sizeof(req));
+
+    if (mnl_socket_sendto(ls->nl, nlh, nlh->nlmsg_len) < 0) {
+        local_sockets_log(ls, "mnl_socket_send failed");
+        return false;
+    }
+
+    ssize_t ret;
+    while ((ret = mnl_socket_recvfrom(ls->nl, buf, sizeof(buf))) > 0) {
+        ret = mnl_cb_run(buf, ret, seq, portid, local_sockets_netlink_cb_data, ls);
+        if (ret <= MNL_CB_STOP)
+            break;
+    }
+    if (ret == -1) {
+        local_sockets_log(ls, "mnl_socket_recvfrom");
+        return false;
+    }
+
+    return true;
+}
+#endif // HAVE_LIBMNL
+
+static inline bool local_sockets_read_proc_net_x(LS_STATE *ls, const char *filename, uint16_t family, uint16_t protocol) {
+    if(family != AF_INET && family != AF_INET6)
+        return false;
+
+    FILE *fp;
+    char *line = NULL;
+    size_t len = 0;
+    ssize_t read;
+
+    fp = fopen(filename, "r");
+    if (fp == NULL)
+        return false;
+
+    ssize_t min_line_length = (family == AF_INET) ? 105 : 155;
+    size_t counter = 0;
+
+    // Read line by line
+    while ((read = getline(&line, &len, fp)) != -1) {
+        if(counter++ == 0) continue; // skip the first line
+
+        if(read < min_line_length) {
+            local_sockets_log(ls, "too small line No %zu of filename '%s': %s", counter, filename, line);
+            continue;
+        }
+
+        unsigned int local_address, local_port, state, remote_address, remote_port;
+        uint64_t  inode = 0;
+        char local_address6[33], remote_address6[33];
+
+        if(family == AF_INET) {
+            if (sscanf(line, "%*d: %X:%X %X:%X %X %*X:%*X %*X:%*X %*X %*d %*d %"PRIu64,
+                       &local_address, &local_port, &remote_address, &remote_port, &state, &inode) != 6) {
+                local_sockets_log(ls, "cannot parse ipv4 line No %zu of filename '%s': %s", counter, filename, line);
+                continue;
+            }
+        }
+        else if(family == AF_INET6) {
+            if(sscanf(line, "%*d: %32[0-9A-Fa-f]:%X %32[0-9A-Fa-f]:%X %X %*X:%*X %*X:%*X %*X %*d %*d %"PRIu64,
+                       local_address6, &local_port, remote_address6, &remote_port, &state, &inode) != 6) {
+                local_sockets_log(ls, "cannot parse ipv6 line No %zu of filename '%s': %s", counter, filename, line);
+                continue;
+            }
+        }
+
+        LOCAL_SOCKET n = {
+            .inode = inode,
+            .direction = SOCKET_DIRECTION_NONE,
+            .state = (int)state,
+            .local = {
+                .family = family,
+                .protocol = protocol,
+                .port = local_port,
+            },
+            .remote = {
+                .family = family,
+                .protocol = protocol,
+                .port = remote_port,
+            },
+            .uid = UID_UNSET,
+        };
+
+        if(family == AF_INET) {
+            n.local.ip.ipv4 = local_address;
+            n.remote.ip.ipv4 = remote_address;
+        }
+        else if(family == AF_INET6) {
+            ipv6_to_in6_addr(local_address6, &n.local.ip.ipv6);
+            ipv6_to_in6_addr(remote_address6, &n.remote.ip.ipv6);
+        }
+
+        local_sockets_add_socket(ls, &n);
+    }
+
+    fclose(fp);
+
+    if (line)
+        freez(line);
+
+    return true;
+}
+
+// --------------------------------------------------------------------------------------------------------------------
+
+static inline void local_sockets_detect_directions(LS_STATE *ls) {
+    for(SIMPLE_HASHTABLE_SLOT_LOCAL_SOCKET *sl = simple_hashtable_first_read_only_LOCAL_SOCKET(&ls->sockets_hashtable);
+         sl ;
+         sl = simple_hashtable_next_read_only_LOCAL_SOCKET(&ls->sockets_hashtable, sl)) {
+        LOCAL_SOCKET *n = SIMPLE_HASHTABLE_SLOT_DATA(sl);
+        if (!n) continue;
+
+        if ((n->direction & (SOCKET_DIRECTION_INBOUND|SOCKET_DIRECTION_OUTBOUND)) !=
+                            (SOCKET_DIRECTION_INBOUND|SOCKET_DIRECTION_OUTBOUND))
+            continue;
+
+        // check if the local port is one of our listening ports
+        {
+            SIMPLE_HASHTABLE_SLOT_LISTENING_PORT *sl_port =
+                simple_hashtable_get_slot_LISTENING_PORT(&ls->listening_ports_hashtable, n->local_port_hash, &n->local_port_key, false);
+
+            struct local_port *port = SIMPLE_HASHTABLE_SLOT_DATA(sl_port); // do not reference this pointer - is invalid
+            if(port) {
+                // the local port of this socket is a port we listen to
+                n->direction &= ~SOCKET_DIRECTION_OUTBOUND;
+            }
+            else
+                n->direction &= ~SOCKET_DIRECTION_INBOUND;
+        }
+
+        // check if the remote IP is one of our local IPs
+        {
+            SIMPLE_HASHTABLE_SLOT_LOCAL_IP *sl_ip =
+                simple_hashtable_get_slot_LOCAL_IP(&ls->local_ips_hashtable, n->remote_ip_hash, &n->remote.ip, false);
+
+            union ipv46 *d = SIMPLE_HASHTABLE_SLOT_DATA(sl_ip);
+            if (d) {
+                // the remote IP of this socket is one of our local IPs
+                if(n->direction & SOCKET_DIRECTION_INBOUND) {
+                    n->direction &= ~SOCKET_DIRECTION_INBOUND;
+                    n->direction |= SOCKET_DIRECTION_LOCAL_INBOUND;
+                }
+                else if(n->direction & SOCKET_DIRECTION_OUTBOUND) {
+                    n->direction &= ~SOCKET_DIRECTION_OUTBOUND;
+                    n->direction |= SOCKET_DIRECTION_LOCAL_OUTBOUND;
+                }
+                continue;
+            }
+        }
+
+        if (local_sockets_is_loopback_address(&n->local) ||
+            local_sockets_is_loopback_address(&n->remote)) {
+            // both IP addresses are loopback
+            if(n->direction & SOCKET_DIRECTION_INBOUND) {
+                n->direction &= ~SOCKET_DIRECTION_INBOUND;
+                n->direction |= SOCKET_DIRECTION_LOCAL_INBOUND;
+            }
+            else if(n->direction & SOCKET_DIRECTION_OUTBOUND) {
+                n->direction &= ~SOCKET_DIRECTION_OUTBOUND;
+                n->direction |= SOCKET_DIRECTION_LOCAL_OUTBOUND;
+            }
+        }
+    }
+}
+
+// --------------------------------------------------------------------------------------------------------------------
+
+static inline void local_sockets_init(LS_STATE *ls) {
+    simple_hashtable_init_NET_NS(&ls->ns_hashtable, 1024);
+    simple_hashtable_init_PID_SOCKET(&ls->pid_sockets_hashtable, 65535);
+    simple_hashtable_init_LOCAL_SOCKET(&ls->sockets_hashtable, 65535);
+    simple_hashtable_init_LOCAL_IP(&ls->local_ips_hashtable, 4096);
+    simple_hashtable_init_LISTENING_PORT(&ls->listening_ports_hashtable, 4096);
+}
+
+static inline void local_sockets_cleanup(LS_STATE *ls) {
+    // free the sockets hashtable data
+    for(SIMPLE_HASHTABLE_SLOT_LOCAL_SOCKET *sl = simple_hashtable_first_read_only_LOCAL_SOCKET(&ls->sockets_hashtable);
+         sl;
+         sl = simple_hashtable_next_read_only_LOCAL_SOCKET(&ls->sockets_hashtable, sl)) {
+        LOCAL_SOCKET *n = SIMPLE_HASHTABLE_SLOT_DATA(sl);
+        if(!n) continue;
+
+        freez(n->cmdline);
+        freez(n);
+    }
+
+    // free the pid_socket hashtable data
+    for(SIMPLE_HASHTABLE_SLOT_PID_SOCKET *sl = simple_hashtable_first_read_only_PID_SOCKET(&a