@@ -1,3 +1,10 @@

+Tue Feb 17 17:08:18 EET 2009 (edwin)

+------------------------------------

+ * clamd/others.c, clamd/others.h, clamd/scanner.c,

+ clamd/server-th.c, shared/optparser.c, unit_tests/check_clamd.c,

+ unit_tests/test-clamd.conf: ReadTimeout handling, and introduce

+ CommandReadTimeout.

+

 Tue Feb 17 12:07:37 CET 2009 (tk)

 ---------------------------------

  * sigtool/sigtool.c: better error messages (bb#1396)

@@ -216,19 +216,22 @@ int poll_fd(int fd, int timeout_sec, int check_signals)

     int ret;

     struct fd_data fds = FDS_INIT;

-    if (fds_add(&fds, fd, 1) == -1)

+    if (fds_add(&fds, fd, 1, timeout_sec) == -1)

 	return -1;

     pthread_mutex_lock(&fds.buf_mutex);

-    ret = fds_poll_recv(&fds, timeout_sec, check_signals);

+    do {

+	ret = fds_poll_recv(&fds, timeout_sec, check_signals);

+    } while (ret == -1 && errno == EINTR);

     pthread_mutex_unlock(&fds.buf_mutex);

     fds_free(&fds);

     return ret;

 }

-static void cleanup_fds(struct fd_data *data)

+void fds_cleanup(struct fd_data *data)

 {

     struct fd_buf *newbuf;

     unsigned i,j;

+

     for (i=0,j=0;i < data->nfds; i++) {

 	if (data->buf[i].fd < 0) {

 	    if (data->buf[i].buffer)

@@ -239,9 +242,12 @@ static void cleanup_fds(struct fd_data *data)

 	    data->buf[j] = data->buf[i];

 	j++;

     }

+    if (j == data->nfds)

+	return;

     for (i = j ; i < data->nfds; i++)

 	data->buf[i].fd = -1;

     data->nfds = j;

+    logg("*CLEANUP_FDS: %u fds\n", data->nfds);

     /* Shrink buffer */

     newbuf = realloc(data->buf, j*sizeof(*newbuf));

     if (newbuf)

@@ -314,7 +320,7 @@ static int read_fd_data(struct fd_buf *buf)

    return n;

 }

-static int buf_init(struct fd_buf *buf, int listen_only)

+static int buf_init(struct fd_buf *buf, int listen_only, int timeout)

 {

     buf->off = 0;

     buf->got_newdata = 0;

@@ -343,10 +349,16 @@ static int buf_init(struct fd_buf *buf, int listen_only)

 	buf->bufsize = 0;

 	buf->buffer = NULL;

     }

+    if (timeout) {

+	time(&buf->timeout_at);

+	buf->timeout_at += timeout;

+    } else {

+	buf->timeout_at = 0;

+    }

     return 0;

 }

-int fds_add(struct fd_data *data, int fd, int listen_only)

+int fds_add(struct fd_data *data, int fd, int listen_only, int timeout)

 {

     struct fd_buf *buf;

     unsigned n;

@@ -359,7 +371,7 @@ int fds_add(struct fd_data *data, int fd, int listen_only)

     for (n = 0; n < data->nfds; n++)

 	if (data->buf[n].fd == fd) {

 	    /* clear stale data in buffer */

-	    if (buf_init(&data->buf[n], listen_only) < 0)

+	    if (buf_init(&data->buf[n], listen_only, timeout) < 0)

 		return -1;

 	    return 0;

 	}

@@ -373,7 +385,7 @@ int fds_add(struct fd_data *data, int fd, int listen_only)

     data->buf = buf;

     data->nfds = n;

     data->buf[n-1].buffer = NULL;

-    if (buf_init(&data->buf[n-1], listen_only) < 0)

+    if (buf_init(&data->buf[n-1], listen_only, timeout) < 0)

 	return -1;

     data->buf[n-1].fd = fd;

     return 0;

@@ -412,12 +424,40 @@ int fds_poll_recv(struct fd_data *data, int timeout, int check_signals)

     unsigned fdsok = data->nfds;

     size_t i;

     int retval;

+    time_t now, closest_timeout;

     /* we must have at least one fd, the control fd! */

+    fds_cleanup(data);

     if (!data->nfds)

 	return 0;

-    for (i=0;i < data->nfds;i++)

+

+    for (i=0;i < data->nfds;i++) {

 	data->buf[i].got_newdata = 0;

+    }

+

+    if (timeout > 0)

+	closest_timeout = now + timeout;

+    else

+	closest_timeout = 0;

+    time(&now);

+    for (i=0;i < data->nfds; i++) {

+	time_t timeout_at = data->buf[i].timeout_at;

+	if (timeout_at && timeout_at < now) {

+	    /* timed out */

+	    data->buf[i].got_newdata = -2;

+	    /* we must return immediately from poll/select, we have a timeout! */

+	    closest_timeout = now;

+	} else {

+	    if (!closest_timeout)

+		closest_timeout = timeout_at;

+	    else if (timeout_at < closest_timeout)

+		closest_timeout = timeout_at;

+	}

+    }

+    if (closest_timeout)

+	timeout = closest_timeout - now;

+    else

+	timeout = -1;

 #ifdef HAVE_POLL

     /* Use poll() if available, preferred because:

      *  - can poll any number of FDs

@@ -577,9 +617,6 @@ int fds_poll_recv(struct fd_data *data, int timeout, int check_signals)

 #endif

     }

-    /* Remove closed / error fds */

-    if (fdsok != data->poll_data_nfds)

-	cleanup_fds(data);

     return retval;

 }

@@ -53,6 +53,7 @@ struct fd_buf {

     uint32_t chunksize;

     long quota;

     char *dumpname;

+    time_t timeout_at; /* 0 - no timeout */

     jobgroup_t *group;

 };

@@ -75,8 +76,9 @@ struct fd_data {

 int poll_fd(int fd, int timeout_sec, int check_signals);

 void virusaction(const char *filename, const char *virname, const struct optstruct *opts);

 int writen(int fd, void *buff, unsigned int count);

-int fds_add(struct fd_data *data, int fd, int listen_only);

+int fds_add(struct fd_data *data, int fd, int listen_only, int timeout);

 void fds_remove(struct fd_data *data, int fd);

+void fds_cleanup(struct fd_data *data);

 int fds_poll_recv(struct fd_data *data, int timeout, int check_signals);

 void fds_free(struct fd_data *data);

@@ -286,7 +286,7 @@ int scanfd(const int fd, const client_conn_t *conn, unsigned long int *scanned,

 int scanstream(int odesc, unsigned long int *scanned, const struct cl_engine *engine, unsigned int options, const struct optstruct *opts, char term)

 {

 	int ret, sockfd, acceptd;

-	int tmpd, bread, retval, timeout, btread;

+	int tmpd, bread, retval, firsttimeout, timeout, btread;

 	unsigned int port = 0, portscan, min_port, max_port;

 	unsigned long int quota = 0, maxsize = 0;

 	short bound = 0;

@@ -327,7 +327,11 @@ int scanstream(int odesc, unsigned long int *scanned, const struct cl_engine *en

     timeout = optget(opts, "ReadTimeout")->numarg;

     if(timeout == 0)

-    	timeout = -1;

+	timeout = -1;

+

+    firsttimeout = optget(opts, "CommandReadTimeout")->numarg;

+    if (firsttimeout == 0)

+	firsttimeout = -1;

     if(!bound && !portscan) {

 	logg("!ScanStream: Can't find any free port.\n");

@@ -343,7 +347,7 @@ int scanstream(int odesc, unsigned long int *scanned, const struct cl_engine *en

 	}

     }

-    retval = poll_fd(sockfd, timeout, 0);

+    retval = poll_fd(sockfd, firsttimeout, 0);

     if (!retval || retval == -1) {

 	const char *reason = !retval ? "timeout" : "poll";

 	mdprintf(odesc, "Accept %s. ERROR%c", reason, term);

@@ -86,7 +86,7 @@ static void scanner_thread(void *arg)

 #ifndef	C_WINDOWS

 	sigset_t sigset;

 #endif

-	int ret, timeout;

+	int ret;

 	unsigned virus=0, errors = 0;

 #ifndef	C_WINDOWS

@@ -105,10 +105,6 @@ static void scanner_thread(void *arg)

     pthread_sigmask(SIG_SETMASK, &sigset, NULL);

 #endif

-    timeout = optget(conn->opts, "ReadTimeout")->numarg;

-    if(!timeout)

-	timeout = -1;

-

     ret = command(conn, &virus);

     if (ret == -1) {

 	pthread_mutex_lock(&exit_mutex);

@@ -294,11 +290,14 @@ static const char *get_cmd(struct fd_buf *buf, size_t off, size_t *len, char *te

 struct acceptdata {

     struct fd_data fds;

     struct fd_data recv_fds;

+    pthread_cond_t cond_nfds;

+    int max_queue;

+    int commandtimeout;

     int syncpipe_wake_recv[2];

     int syncpipe_wake_accept[2];

 };

-#define ACCEPTDATA_INIT { FDS_INIT, FDS_INIT, {-1, -1}, {-1, -1} }

+#define ACCEPTDATA_INIT { FDS_INIT, FDS_INIT, PTHREAD_COND_INITIALIZER, 0, 0, {-1, -1}, {-1, -1}}

 static void *acceptloop_th(void *arg)

 {

@@ -309,6 +308,8 @@ static void *acceptloop_th(void *arg)

     struct acceptdata *data = (struct acceptdata*)arg;

     struct fd_data *fds = &data->fds;

     struct fd_data *recv_fds = &data->recv_fds;

+    int max_queue = data->max_queue;

+    int commandtimeout = data->commandtimeout;

     pthread_mutex_lock(&fds->buf_mutex);

     for (;;) {

@@ -348,8 +349,20 @@ static void *acceptloop_th(void *arg)

 		buf->fd = -1;

 		continue;

 	    }

-	    /* listen only socket */

-	    new_sd = accept(fds->buf[i].fd, NULL, NULL);

+

+	    /* don't accept unlimited number of connections, or

+	     * we'll run out of file descriptors */

+	    pthread_mutex_lock(&recv_fds->buf_mutex);

+	    while (recv_fds->nfds > max_queue) {

+		pthread_mutex_lock(&exit_mutex);

+		if(progexit) {

+		    pthread_mutex_unlock(&exit_mutex);

+		    break;

+		}

+		pthread_mutex_unlock(&exit_mutex);

+		pthread_cond_wait(&data->cond_nfds, &recv_fds->buf_mutex);

+	    }

+	    pthread_mutex_unlock(&recv_fds->buf_mutex);

 	    pthread_mutex_lock(&exit_mutex);

 	    if(progexit) {

@@ -358,6 +371,10 @@ static void *acceptloop_th(void *arg)

 	    }

 	    pthread_mutex_unlock(&exit_mutex);

+	    /* listen only socket */

+	    new_sd = accept(fds->buf[i].fd, NULL, NULL);

+

+

 	    if (new_sd >= 0) {

 		int ret, flags;

@@ -374,7 +391,7 @@ static void *acceptloop_th(void *arg)

 #endif

 		pthread_mutex_lock(&recv_fds->buf_mutex);

-		ret = fds_add(recv_fds, new_sd, 0);

+		ret = fds_add(recv_fds, new_sd, 0, commandtimeout);

 		pthread_mutex_unlock(&recv_fds->buf_mutex);

 		if (ret == -1) {

@@ -432,7 +449,7 @@ static void *acceptloop_th(void *arg)

 int recvloop_th(int *socketds, unsigned nsockets, struct cl_engine *engine, unsigned int dboptions, const struct optstruct *opts)

 {

-	int max_threads, max_queue, ret = 0;

+	int max_threads, max_queue, readtimeout, ret = 0;

 	unsigned int options = 0;

 	char timestr[32];

 #ifndef	C_WINDOWS

@@ -692,7 +709,9 @@ int recvloop_th(int *socketds, unsigned nsockets, struct cl_engine *engine, unsi

     logg("*Listening daemon: PID: %u\n", (unsigned int) mainpid);

     max_threads = optget(opts, "MaxThreads")->numarg;

-    max_queue = optget(opts, "MaxQueue")->numarg;

+    acceptdata.max_queue = max_queue = optget(opts, "MaxQueue")->numarg;

+    acceptdata.commandtimeout = optget(opts, "CommandReadTimeout")->numarg;

+    readtimeout = optget(opts, "ReadTimeout")->numarg;

     if(optget(opts, "ClamukoScanOnAccess")->enabled)

 #ifdef CLAMUKO

@@ -753,7 +772,7 @@ int recvloop_th(int *socketds, unsigned nsockets, struct cl_engine *engine, unsi

     idletimeout = optget(opts, "IdleTimeout")->numarg;

     for (i=0;i < nsockets;i++)

-	if (fds_add(&acceptdata.fds, socketds[i], 1) == -1) {

+	if (fds_add(&acceptdata.fds, socketds[i], 1, 0) == -1) {

 	    logg("!fds_add failed\n");

 	    cl_engine_free(engine);

 	    return 1;

@@ -766,8 +785,8 @@ int recvloop_th(int *socketds, unsigned nsockets, struct cl_engine *engine, unsi

 	exit(-1);

     }

-    if (fds_add(fds, acceptdata.syncpipe_wake_recv[0], 1) == -1 ||

-	fds_add(&acceptdata.fds, acceptdata.syncpipe_wake_accept[0], 1)) {

+    if (fds_add(fds, acceptdata.syncpipe_wake_recv[0], 1, 0) == -1 ||

+	fds_add(&acceptdata.fds, acceptdata.syncpipe_wake_accept[0], 1, 0)) {

 	logg("!failed to add pipe fd\n");

 	exit(-1);

     }

@@ -787,6 +806,10 @@ int recvloop_th(int *socketds, unsigned nsockets, struct cl_engine *engine, unsi

 	int new_sd;

 	/* Block waiting for connection on any of the sockets */

 	pthread_mutex_lock(&fds->buf_mutex);

+	fds_cleanup(fds);

+	/* signal that we can accept more connections */

+	if (fds->nfds <= max_queue)

+	    pthread_cond_signal(&acceptdata.cond_nfds);

 	new_sd = fds_poll_recv(fds, -1, 1);

 	if (!fds->nfds) {

@@ -806,6 +829,7 @@ int recvloop_th(int *socketds, unsigned nsockets, struct cl_engine *engine, unsi

 	    pthread_mutex_unlock(&exit_mutex);

 	}

+

 	i = (rr_last + 1) % fds->nfds;

 	for (j = 0;  j < fds->nfds && new_sd >= 0; j++, i = (i+1) % fds->nfds) {

 	    size_t pos = 0;

@@ -833,6 +857,12 @@ int recvloop_th(int *socketds, unsigned nsockets, struct cl_engine *engine, unsi

 		}

 	    }

+	    if (buf->fd != -1 && buf->got_newdata == -2) {

+		logg("*RECVTH: client read timed out\n");

+		mdprintf(buf->fd, "COMMAND READ TIMED OUT\n");

+		error = 1;

+	    }

+

 	    rr_last = i;

 	    if (buf->mode == MODE_WAITANCILL) {

 		buf->mode = MODE_COMMAND;

@@ -940,6 +970,9 @@ int recvloop_th(int *socketds, unsigned nsockets, struct cl_engine *engine, unsi

 			    buf->fd = -1;

 			}

 		    }

+		    /* we received a command, set readtimeout */

+		    time(&buf->timeout_at);

+		    buf->timeout_at += readtimeout;

 		    pos += cmdlen+1;

 		    if (conn.mode == MODE_STREAM) {

 			/* TODO: this doesn't belong here */

@@ -187,6 +187,8 @@ const struct clam_option clam_options[] = {

     { "ReadTimeout", NULL, 0, TYPE_NUMBER, MATCH_NUMBER, 120, NULL, 0, OPT_CLAMD, "This option specifies the time (in seconds) after which clamd should\ntimeout if a client doesn't provide any data.", "120" },

+    { "CommandReadTimeout", NULL, 0, TYPE_NUMBER, MATCH_NUMBER, 5, NULL, 0, OPT_CLAMD, "This option specifies the time (in seconds) after which clamd should\ntimeout if a client doesn't provide any initial command after connecting.", "5" },

+

     { "ReadTimeout", NULL, 0, TYPE_NUMBER, MATCH_NUMBER, 120, NULL, 0, OPT_MILTER, "Waiting for data from clamd will timeout after this time (seconds).\nValue of 0 disables the timeout.", "300" },

     { "MaxQueue", NULL, 0, TYPE_NUMBER, MATCH_NUMBER, 100, NULL, 0, OPT_CLAMD, "Maximum number of queued items (including those being processed)\nWARNING: you shouldn't increase this beyond 512, since you may run out of File Descriptors (usual max is 1024)\n", "200" },

@@ -17,6 +17,12 @@

 #include <sys/un.h>

 #include <unistd.h>

+#include <sys/time.h>

+#include <sys/resource.h>

+#include <sys/select.h>

+#include <sys/time.h>

+#include <sys/types.h>

+#include <unistd.h>

 #include <check.h>

 #include "checks_common.h"

 #include "libclamav/version.h"

@@ -510,6 +516,70 @@ START_TEST (test_fildes_unwanted)

 }

 END_TEST

+#define TIMEOUT_REPLY "TIMED OUT WAITING FOR COMMAND\n"

+

+START_TEST (test_connections)

+{

+    int rc;

+    size_t i;

+    struct rlimit rlim;

+    int *sock;

+    int nf, maxfd=0;

+    fail_unless_fmt(getrlimit(RLIMIT_NOFILE, &rlim) != -1,

+		    "Failed to get RLIMIT_NOFILE: %s\n", strerror(errno));

+    nf = rlim.rlim_cur - 5;

+    sock = malloc(sizeof(int)*nf);

+

+    fail_unless(!!sock, "malloc failed\n");

+

+    for (i=0;i<nf;i++) {

+	/* just open connections, and let them time out */

+	conn_setup();

+	sock[i] = sockd;

+	if (sockd > maxfd)

+	    maxfd = sockd;

+    }

+    rc = fork();

+    fail_unless(rc != -1, "fork() failed: %s\n", strerror(errno));

+    if (rc == 0) {

+	char dummy;

+	int ret;

+	fd_set rfds;

+	FD_ZERO(&rfds);

+	for (i=0;i<nf;i++) {

+	    FD_SET(sock[i], &rfds);

+	}

+	while (1) {

+	    ret = select(maxfd+1, &rfds, NULL, NULL, NULL);

+	    if (ret < 0)

+		break;

+	    for (i=0;i<nf;i++) {

+		if (FD_ISSET(sock[i], &rfds)) {

+		    if (recv(sock[i], &dummy, 1, 0) == 0) {

+			close(sock[i]);

+			FD_CLR(sock[i], &rfds);

+		    }

+		}

+	    }

+	}

+	printf("exited\n");

+	free(sock);

+	exit(0);

+    } else {

+	for (i=0;i<nf;i++) {

+	    close(sock[i]);

+	}

+	free(sock);

+	/* now see if clamd is able to do anything else */

+	for (i=0;i<10;i++) {

+	    conn_setup();

+	    test_command("RELOAD", sizeof("RELOAD")-1, NULL, "RELOADING\n", sizeof("RELOADING\n")-1);

+	    conn_teardown();

+	}

+    }

+}

+END_TEST

+

 static Suite *test_clamd_suite(void)

 {

     Suite *s = suite_create("clamd");

@@ -528,6 +598,7 @@ static Suite *test_clamd_suite(void)

     suite_add_tcase(s, tc_stress);

     tcase_add_test(tc_stress, test_fildes_many);

     tcase_add_test(tc_stress, test_fildes_unwanted);

+    tcase_add_test(tc_stress, test_connections);

     return s;

 }

@@ -12,3 +12,5 @@ TCPSocket 331X

 ExitOnOOM yes

 DetectPUA yes

 ScanPDF yes

+CommandReadTimeout 1

+MaxQueue 800