Merge pull request #3578 from ktsaou/master

netdata web server protection against slowloris

9 files changed, 119 insertions, 28 deletions

diff --git a/src/common.c b/src/common.c
index fc4acc4777..94fd5e429d 100644
--- a/src/common.c
+++ b/src/common.c
@@ -19,6 +19,7 @@ char *netdata_configured_home_dir    = NULL;
 char *netdata_configured_host_prefix = NULL;
 char *netdata_configured_timezone    = NULL;
 
+struct rlimit rlimit_nofile = { .rlim_cur = 1024, .rlim_max = 1024 };
 int enable_ksm = 1;
 
 volatile sig_atomic_t netdata_exit = 0;
diff --git a/src/common.h b/src/common.h
index dba96f8858..15fc50a6a2 100644
--- a/src/common.h
+++ b/src/common.h
@@ -323,6 +323,8 @@ extern int memory_file_save(const char *filename, void *mem, size_t size);
 
 extern int fd_is_valid(int fd);
 
+extern struct rlimit rlimit_nofile;
+
 extern int enable_ksm;
 
 extern int sleep_usec(usec_t usec);
diff --git a/src/main.c b/src/main.c
index 08f07ebb9f..798c7f0fcf 100644
--- a/src/main.c
+++ b/src/main.c
@@ -1005,6 +1005,11 @@ int main(int argc, char **argv) {
     }
 #endif /* NETDATA_INTERNAL_CHECKS */
 
+    // get the max file limit
+    if(getrlimit(RLIMIT_NOFILE, &rlimit_nofile) != 0)
+        error("getrlimit(RLIMIT_NOFILE) failed");
+    else
+        info("resources control: allowed file descriptors: soft = %zu, max = %zu", rlimit_nofile.rlim_cur, rlimit_nofile.rlim_max);
 
     // fork, switch user, create pid file, set process priority
     if(become_daemon(dont_fork, user) == -1)
diff --git a/src/socket.c b/src/socket.c
index 5980bb611d..8bede73fd9 100644
--- a/src/socket.c
+++ b/src/socket.c
@@ -978,6 +978,12 @@ inline POLLINFO *poll_add_fd(POLLJOB *p
 
     if(unlikely(fd < 0)) return NULL;
 
+    //if(p->limit && p->used >= p->limit) {
+    //    info("Max sockets limit reached (%zu sockets), dropping connection", p->used);
+    //    close(fd);
+    //    return NULL;
+    //}
+
     if(unlikely(!p->first_free)) {
         size_t new_slots = p->slots + POLL_FDS_INCREASE_STEP;
         debug(D_POLLFD, "POLLFD: ADD: increasing size (current = %zu, new = %zu, used = %zu, min = %zu, max = %zu)", p->slots, new_slots, p->used, p->min, p->max);
@@ -1199,7 +1205,28 @@ static void poll_events_process(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, sho
         pi->last_received_t = now;
         pi->recv_count++;
 
-        if(likely(pi->flags & POLLINFO_FLAG_SERVER_SOCKET)) {
+        if(likely(pi->flags & POLLINFO_FLAG_CLIENT_SOCKET)) {
+            // read data from client TCP socket
+            debug(D_POLLFD, "POLLFD: LISTENER: reading data from TCP client slot %zu (fd %d)", i, fd);
+
+            pf->events = 0;
+            if (pi->rcv_callback(pi, &pf->events) == -1) {
+                poll_close_fd(&p->inf[i]);
+                return;
+            }
+            pf = &p->fds[i];
+            pi = &p->inf[i];
+
+#ifdef NETDATA_INTERNAL_CHECKS
+            // this is common - it is used for web server file copies
+            if(unlikely(!(pf->events & (POLLIN|POLLOUT)))) {
+                error("POLLFD: LISTENER: after reading, client slot %zu (fd %d) from '%s:%s' was left without expecting input or output. ", i, fd, pi->client_ip?pi->client_ip:"<undefined-ip>", pi->client_port?pi->client_port:"<undefined-port>");
+                //poll_close_fd(pi);
+                //return;
+            }
+#endif
+        }
+        else if(likely(pi->flags & POLLINFO_FLAG_SERVER_SOCKET)) {
             // new connection
             // debug(D_POLLFD, "POLLFD: LISTENER: accepting connections from slot %zu (fd %d)", i, fd);
 
@@ -1220,7 +1247,11 @@ static void poll_events_process(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, sho
 
                             debug(D_POLLFD, "POLLFD: LISTENER: accept4() slot %zu (fd %d) failed.", i, fd);
 
-                            if(errno != EWOULDBLOCK && errno != EAGAIN)
+                            if(unlikely(errno == EMFILE)) {
+                                error("POLLFD: LISTENER: too many open files - sleeping for 1ms - used by this thread %zu, max for this thread %zu", p->used, p->limit);
+                                usleep(1000); // 10ms
+                            }
+                            else if(unlikely(errno != EWOULDBLOCK && errno != EAGAIN))
                                 error("POLLFD: LISTENER: accept() failed.");
 
                             break;
@@ -1245,7 +1276,7 @@ static void poll_events_process(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, sho
                             pf = &p->fds[i];
                             pi = &p->inf[i];
                         }
-                    } while (nfd >= 0);
+                    } while (nfd >= 0 && (!p->limit || p->used < p->limit));
                     break;
                 }
 
@@ -1268,26 +1299,6 @@ static void poll_events_process(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, sho
                 }
             }
         }
-
-        if(likely(pi->flags & POLLINFO_FLAG_CLIENT_SOCKET)) {
-            // read data from client TCP socket
-            debug(D_POLLFD, "POLLFD: LISTENER: reading data from TCP client slot %zu (fd %d)", i, fd);
-
-            pf->events = 0;
-            if (pi->rcv_callback(pi, &pf->events) == -1) {
-                poll_close_fd(pi);
-                return;
-            }
-
-#ifdef NETDATA_INTERNAL_CHECKS
-            // this is common - it is used for web server file copies
-            if(unlikely(!(pf->events & (POLLIN|POLLOUT)))) {
-                error("POLLFD: LISTENER: after reading, client slot %zu (fd %d) from '%s:%s' was left without expecting input or output. ", i, fd, pi->client_ip?pi->client_ip:"<undefined-ip>", pi->client_port?pi->client_port:"<undefined-port>");
-                //poll_close_fd(pi);
-                //return;
-            }
-#endif
-        }
     }
 
     if(unlikely(revents & POLLOUT)) {
@@ -1299,9 +1310,11 @@ static void poll_events_process(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, sho
 
         pf->events = 0;
         if (pi->snd_callback(pi, &pf->events) == -1) {
-            poll_close_fd(pi);
+            poll_close_fd(&p->inf[i]);
             return;
         }
+        pf = &p->fds[i];
+        pi = &p->inf[i];
 
 #ifdef NETDATA_INTERNAL_CHECKS
         // this is common - it is used for streaming
@@ -1347,6 +1360,7 @@ void poll_events(LISTEN_SOCKETS *sockets
         , time_t tcp_idle_timeout_seconds
         , time_t timer_milliseconds
         , void *timer_data
+        , size_t max_tcp_sockets
 ) {
     if(!sockets || !sockets->opened) {
         error("POLLFD: internal error: no listening sockets are opened");
@@ -1361,6 +1375,7 @@ void poll_events(LISTEN_SOCKETS *sockets
             .slots = 0,
             .used = 0,
             .max = 0,
+            .limit = max_tcp_sockets,
             .fds = NULL,
             .inf = NULL,
             .first_free = NULL,
@@ -1401,6 +1416,8 @@ void poll_events(LISTEN_SOCKETS *sockets
         info("POLLFD: LISTENER: listening on '%s'", (sockets->fds_names[i])?sockets->fds_names[i]:"UNKNOWN");
     }
 
+    int listen_sockets_active = 1;
+
     int timeout_ms = 1000; // in milliseconds
     time_t last_check = now_boottime_sec();
 
@@ -1432,6 +1449,17 @@ void poll_events(LISTEN_SOCKETS *sockets
                 timeout_ms = (int)(dt_usec / USEC_PER_MS);
         }
 
+        // enable or disable the TCP listening sockets, based on the current number of sockets used and the limit set
+        if((listen_sockets_active && (p.limit && p.used >= p.limit)) || (!listen_sockets_active && (!p.limit || p.used < p.limit))) {
+            listen_sockets_active = !listen_sockets_active;
+            info("%s listening sockets (used TCP sockets %zu, max allowed for this worker %zu)", (listen_sockets_active)?"ENABLING":"DISABLING", p.used, p.limit);
+            for (i = 0; i <= p.max; i++) {
+                if(p.inf[i].flags & POLLINFO_FLAG_SERVER_SOCKET && p.inf[i].socktype == SOCK_STREAM) {
+                    p.fds[i].events = (short int) ((listen_sockets_active) ? POLLIN : 0);
+                }
+            }
+        }
+
         debug(D_POLLFD, "POLLFD: LISTENER: Waiting on %zu sockets for %zu ms...", p.max + 1, (size_t)timeout_ms);
         retval = poll(p.fds, p.max + 1, timeout_ms);
         time_t now = now_boottime_sec();
diff --git a/src/socket.h b/src/socket.h
index 8dd043709c..7b3e726ece 100644
--- a/src/socket.h
+++ b/src/socket.h
@@ -101,6 +101,8 @@ struct poll {
     size_t min;
     size_t max;
 
+    size_t limit;
+
     time_t complete_request_timeout;
     time_t idle_timeout;
     time_t checks_every;
@@ -154,6 +156,7 @@ extern void poll_events(LISTEN_SOCKETS *sockets
         , time_t tcp_idle_timeout_seconds
         , time_t timer_milliseconds
         , void *timer_data
+        , size_t max_tcp_sockets
 );
 
 #endif //NETDATA_SOCKET_H
diff --git a/src/statsd.c b/src/statsd.c
index b0d0547d55..44ebd88947 100644
--- a/src/statsd.c
+++ b/src/statsd.c
@@ -222,6 +222,8 @@ typedef struct statsd_app {
 
 struct collection_thread_status {
     int status;
+    size_t max_sockets;
+
     netdata_thread_t thread;
     struct rusage rusage;
     RRDSET *st_cpu;
@@ -359,7 +361,7 @@ static int statsd_metric_compare(void* a, void* b) {
     else return strcmp(((STATSD_METRIC *)a)->name, ((STATSD_METRIC *)b)->name);
 }
 
-static inline STATSD_METRIC *stasd_metric_index_find(STATSD_INDEX *index, const char *name, uint32_t hash) {
+static inline STATSD_METRIC *statsd_metric_index_find(STATSD_INDEX *index, const char *name, uint32_t hash) {
     STATSD_METRIC tmp;
     tmp.name = name;
     tmp.hash = (hash)?hash:simple_hash(tmp.name);
@@ -372,7 +374,7 @@ static inline STATSD_METRIC *statsd_find_or_add_metric(STATSD_INDEX *index, cons
 
     uint32_t hash = simple_hash(name);
 
-    STATSD_METRIC *m = stasd_metric_index_find(index, name, hash);
+    STATSD_METRIC *m = statsd_metric_index_find(index, name, hash);
     if(unlikely(!m)) {
         debug(D_STATSD, "Creating new %s metric '%s'", index->name, name);
 
@@ -982,6 +984,7 @@ void *statsd_collector_thread(void *ptr) {
             , statsd.tcp_idle_timeout  // tcp idle timeout, 0 = disabled
             , statsd.update_every * 1000
             , ptr // timer_data
+            , status->max_sockets
     );
 
     netdata_thread_cleanup_pop(1);
@@ -2169,6 +2172,8 @@ void *statsd_main(void *ptr) {
     if(config_get_boolean(CONFIG_SECTION_STATSD, "gaps on timers (deleteTimers)", 0))
         statsd.timers.default_options |= STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED;
 
+    size_t max_sockets = (size_t)config_get_number(CONFIG_SECTION_STATSD, "statsd server max TCP sockets", (long long int)(rlimit_nofile.rlim_cur / 4));
+
 #ifdef STATSD_MULTITHREADED
     statsd.threads = (int)config_get_number(CONFIG_SECTION_STATSD, "threads", processors);
     if(statsd.threads < 1) {
@@ -2202,6 +2207,7 @@ void *statsd_main(void *ptr) {
 
     int i;
     for(i = 0; i < statsd.threads ;i++) {
+        statsd.collection_threads_status[i].max_sockets = max_sockets / statsd.threads;
         char tag[NETDATA_THREAD_TAG_MAX + 1];
         snprintfz(tag, NETDATA_THREAD_TAG_MAX, "STATSD_COLLECTOR[%d]", i + 1);
         netdata_thread_create(&statsd.collection_threads_status[i].thread, tag, NETDATA_THREAD_OPTION_DEFAULT, statsd_collector_thread, &statsd.collection_threads_status[i]);
diff --git a/src/web_client.c b/src/web_client.c
index 3a6b13e5f4..477fb3d576 100644
--- a/src/web_client.c
+++ b/src/web_client.c
@@ -159,6 +159,9 @@ void web_client_request_done(struct web_client *w) {
     w->response.sent = 0;
     w->response.code = 0;
 
+    w->header_parse_tries = 0;
+    w->header_parse_last_size = 0;
+
     web_client_enable_wait_receive(w);
     web_client_disable_wait_send(w);
 
@@ -809,7 +812,31 @@ typedef enum {
 } HTTP_VALIDATION;
 
 static inline HTTP_VALIDATION http_request_validate(struct web_client *w) {
-    char *s = w->response.data->buffer, *encoded_url = NULL;
+    char *s = (char *)buffer_tostring(w->response.data), *encoded_url = NULL;
+
+    size_t last_pos = w->header_parse_last_size;
+    if(last_pos > 4) last_pos -= 4; // allow searching for \r\n\r\n
+    else last_pos = 0;
+
+    w->header_parse_tries++;
+    w->header_parse_last_size = buffer_strlen(w->response.data);
+
+    if(w->header_parse_tries > 1) {
+        if(w->header_parse_last_size < last_pos)
+            last_pos = 0;
+
+        if(strstr(&s[last_pos], "\r\n\r\n") == NULL) {
+            if(w->header_parse_tries > 10) {
+                info("Disabling slow client after %zu attempts to read the request (%zu bytes received)", w->header_parse_tries, buffer_strlen(w->response.data));
+                w->header_parse_tries = 0;
+                w->header_parse_last_size = 0;
+                web_client_disable_wait_receive(w);
+                return HTTP_VALIDATION_NOT_SUPPORTED;
+            }
+
+            return HTTP_VALIDATION_INCOMPLETE;
+        }
+    }
 
     // is is a valid request?
     if(!strncmp(s, "GET ", 4)) {
@@ -825,6 +852,8 @@ static inline HTTP_VALIDATION http_request_validate(struct web_client *w) {
         w->mode = WEB_CLIENT_MODE_STREAM;
     }
     else {
+        w->header_parse_tries = 0;
+        w->header_parse_last_size = 0;
         web_client_disable_wait_receive(w);
         return HTTP_VALIDATION_NOT_SUPPORTED;
     }
@@ -872,6 +901,8 @@ static inline HTTP_VALIDATION http_request_validate(struct web_client *w) {
                 // FIXME -- we should avoid it
                 strncpyz(w->last_url, w->decoded_url, NETDATA_WEB_REQUEST_URL_SIZE);
 
+                w->header_parse_tries = 0;
+                w->header_parse_last_size = 0;
                 web_client_disable_wait_receive(w);
                 return HTTP_VALIDATION_OK;
             }
diff --git a/src/web_client.h b/src/web_client.h
index d7c48d5094..b495c37e17 100644
--- a/src/web_client.h
+++ b/src/web_client.h
@@ -128,6 +128,9 @@ struct web_client {
     WEB_CLIENT_MODE mode;           // the operational mode of the client
     WEB_CLIENT_ACL acl;             // the access list of the client
 
+    size_t header_parse_tries;
+    size_t header_parse_last_size;
+
     int tcp_cork;                   // 1 = we have a cork on the socket
 
     int ifd;
diff --git a/src/web_server.c b/src/web_server.c
index 1d194565fc..31c5464113 100644
--- a/src/web_server.c
+++ b/src/web_server.c
@@ -902,6 +902,8 @@ struct web_server_static_threaded_worker {
     int id;
     int running;
 
+    size_t max_sockets;
+
     volatile size_t connected;
     volatile size_t disconnected;
     volatile size_t receptions;
@@ -1078,7 +1080,12 @@ static int web_server_rcv_callback(POLLINFO *pi, short int *events) {
                         , (void *) w
                 );
 
-                w->pollinfo_filecopy_slot = fpi->slot;
+                if(fpi)
+                    w->pollinfo_filecopy_slot = fpi->slot;
+                else {
+                    error("Failed to add filecopy fd. Closing client.");
+                    return -1;
+                }
             }
         }
     }
@@ -1192,6 +1199,7 @@ void *socket_listen_main_static_threaded_worker(void *ptr) {
                         , web_client_timeout
                         , default_rrd_update_every * 1000 // timer_milliseconds
                         , ptr // timer_data
+                        , worker_private->max_sockets
             );
 
     netdata_thread_cleanup_pop(1);
@@ -1257,6 +1265,8 @@ void *socket_listen_main_static_threaded(void *ptr) {
             static_threaded_workers_count = config_get_number(CONFIG_SECTION_WEB, "web server threads", def_thread_count);
             if(static_threaded_workers_count < 1) static_threaded_workers_count = 1;
 
+            size_t max_sockets = (size_t)config_get_number(CONFIG_SECTION_WEB, "web server max sockets", (long long int)(rlimit_nofile.rlim_cur / 2));
+
             static_workers_private_data = callocz((size_t)static_threaded_workers_count, sizeof(struct web_server_static_threaded_worker));
 
             web_server_is_multithreaded = (static_threaded_workers_count > 1);
@@ -1264,6 +1274,7 @@ void *socket_listen_main_static_threaded(void *ptr) {
             int i;
             for(i = 1; i < static_threaded_workers_count; i++) {
                 static_workers_private_data[i].id = i;
+                static_workers_private_data[i].max_sockets = max_sockets / static_threaded_workers_count;
 
                 char tag[50 + 1];
                 snprintfz(tag, 50, "WEB_SERVER[static%d]", i+1);
@@ -1273,6 +1284,7 @@ void *socket_listen_main_static_threaded(void *ptr) {
             }
 
             // and the main one
+            static_workers_private_data[0].max_sockets = max_sockets / static_threaded_workers_count;
             socket_listen_main_static_threaded_worker((void *)&static_workers_private_data[0]);
 
     netdata_thread_cleanup_pop(1);