From fa99b5b88dd07122e87f25f086746710ce926bd1 Mon Sep 17 00:00:00 2001 From: DL6ER Date: Wed, 8 Mar 2017 11:32:25 +0100 Subject: [PATCH] [MAJOR change] FTL is now multi-threaded. There will be one core process (which catches signals handles them), one log processing thread (waiting for log changes), one listening thread (waiting for incoming connections) and one thread for each client that is connected (waiting for requests to be answered by FTL) --- FTL.h | 4 -- log.c | 13 +++++ main.c | 39 +++++-------- request.c | 133 ++++++++++++++++++++++++------------------- routines.h | 13 +++-- socket.c | 161 +++++++++++++++++++++++++++++------------------------ 6 files changed, 199 insertions(+), 164 deletions(-) diff --git a/FTL.h b/FTL.h index 615818c6..21ec14a1 100644 --- a/FTL.h +++ b/FTL.h @@ -139,10 +139,6 @@ FILE *logfile; FILE *dnsmasqlog; int dnsmasqlogpos; volatile sig_atomic_t killed; -int clientsocket; - -char socketrecvbuffer[SOCKETBUFFERLEN]; -char socketsendbuffer[SOCKETBUFFERLEN]; char ** setupVarsArray; int setupVarsElements; diff --git a/log.c b/log.c index 1156a0c1..4ce00b6c 100644 --- a/log.c +++ b/log.c @@ -73,6 +73,19 @@ void logg_str_str(const char* str, char* str2, char* str3) printf("[%d-%02d-%02d %02d:%02d:%02d.%03i] %s%s (%s)\n", tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec, millisec, str, str2, str3); } +void logg_str_str_int(const char* str, char* str2, char* str3, int i) +{ + time_t t = time(NULL); + struct tm tm = *localtime(&t); + struct timeval tv; + gettimeofday(&tv, NULL); + int millisec = tv.tv_usec/1000; + fprintf(logfile, "[%d-%02d-%02d %02d:%02d:%02d.%03i] %s%s%s%i\n", tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec, millisec, str, str2, str3, i); + fflush(logfile); + if(debug) + printf("[%d-%02d-%02d %02d:%02d:%02d.%03i] %s%s%s%i\n", tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec, millisec, str, str2, str3, i); +} + void logg_struct_resize(const char* str, int from, int to) { time_t t = time(NULL); diff --git a/main.c b/main.c index 96c41223..67081cc0 100644 --- a/main.c +++ b/main.c @@ -42,41 +42,30 @@ int main (int argc, char* argv[]) { log_counter_info(); check_setupVarsconf(); - bool clientconnected = false; - pthread_t piholelogthread; - - if(pthread_create( &piholelogthread, NULL, &pihole_log_thread, NULL ) != 0) + if(pthread_create( &piholelogthread, NULL, pihole_log_thread, NULL ) != 0) { logg("Unable to open Pi-hole log processing thread. Exiting..."); killed = 1; } - while(!killed) + pthread_t listenthread; + if(pthread_create( &listenthread, NULL, listenting_thread, NULL ) != 0) { - // Daemon loop - check_socket(); - - if (clientsocket > 0) - { - clientconnected = true; - read_socket(); - sleepms(5); - } - else if(clientconnected) - { - clientconnected = false; - if(debug) - logg("Client disconnected"); - } - else - { - listen_socket(); - } + logg("Unable to open Socket listening thread. Exiting..."); + killed = 1; } + while(!killed) + { + sleepms(100); + } + + logg("Shutting down..."); - close_sockets(); + pthread_cancel(piholelogthread); + pthread_cancel(listenthread); +// close_sockets(); logg("########## FTL terminated! ##########"); fclose(logfile); return 0; diff --git a/request.c b/request.c index 8536a3f0..b2ce994c 100644 --- a/request.c +++ b/request.c @@ -16,26 +16,30 @@ int cmpdomains(int *elem1, int *elem2); #define min(a,b) ({ __typeof__ (a) _a = (a); __typeof__ (b) _b = (b); _a < _b ? _a : _b; }) #define max(a,b) ({ __typeof__ (a) _a = (a); __typeof__ (b) _b = (b); _a > _b ? _a : _b; }) -void process_request(void) +void process_request(char *client_message, int *sock) { char EOT[2]; EOT[0] = 0x04; EOT[1] = 0x00; - if(command(">stats")) + char server_message[SOCKETBUFFERLEN]; + bool processed = false; + if(command(client_message, ">stats")) { + processed = true; float percentage = 0.0; // Avoid 1/0 condition if(counters.queries > 0) { percentage = 1e2*counters.blocked/counters.queries; } - sprintf(socketsendbuffer,"domains_being_blocked %i\ndns_queries_today %i\nads_blocked_today %i\nads_percentage_today %f\n",counters.gravity,counters.queries,counters.blocked,percentage); - swrite(); + sprintf(server_message,"domains_being_blocked %i\ndns_queries_today %i\nads_blocked_today %i\nads_percentage_today %f\n",counters.gravity,counters.queries,counters.blocked,percentage); + swrite(server_message, *sock); if(debug) - logg("Sent stats data to client"); + logg_int("Sent stats data to client, ID: ", *sock); } - else if(command(">overTime")) + else if(command(client_message, ">overTime")) { + processed = true; int i; bool sendit = false; for(i=0; i < counters.overTime; i++) @@ -46,20 +50,21 @@ void process_request(void) } if(sendit) { - sprintf(socketsendbuffer,"%i %i %i\n",overTime[i].timestamp,overTime[i].total,overTime[i].blocked); - swrite(); + sprintf(server_message,"%i %i %i\n",overTime[i].timestamp,overTime[i].total,overTime[i].blocked); + swrite(server_message, *sock); } } if(debug) - logg("Sent overTime data to client"); + logg_int("Sent overTime data to client, ID: ", *sock); } - else if(command(">top-domains") || command(">top-ads")) + else if(command(client_message, ">top-domains") || command(client_message, ">top-ads")) { + processed = true; int i, temparray[counters.domains][2], count=10, num; - bool blocked = command(">top-ads"); + bool blocked = command(client_message, ">top-ads"); // Match both top-domains and top-ads - if(sscanf(socketrecvbuffer, ">%*[^(](%i)", &num) > 0) + if(sscanf(client_message, ">%*[^(](%i)", &num) > 0) { // User wants a different number of requests count = num; @@ -123,25 +128,26 @@ void process_request(void) if(blocked && showblocked && domains[j].blockedcount > 0) { - sprintf(socketsendbuffer,"%i %i %s\n",i,domains[j].blockedcount,domains[j].domain); - swrite(); + sprintf(server_message,"%i %i %s\n",i,domains[j].blockedcount,domains[j].domain); + swrite(server_message, *sock); } else if(!blocked && showpermitted && (domains[j].count - domains[j].blockedcount) > 0) { - sprintf(socketsendbuffer,"%i %i %s\n",i,(domains[j].count - domains[j].blockedcount),domains[j].domain); - swrite(); + sprintf(server_message,"%i %i %s\n",i,(domains[j].count - domains[j].blockedcount),domains[j].domain); + swrite(server_message, *sock); } } if(excludedomains != NULL) clearSetupVarsArray(); if(debug) - logg("Sent top lists data to client"); + logg_int("Sent top lists data to client, ID: ", *sock); } - else if(command(">top-clients")) + else if(command(client_message, ">top-clients")) { + processed = true; int i, temparray[counters.clients][2], count=10, num; - if(sscanf(socketrecvbuffer, ">%*[^(](%i)", &num) > 0) + if(sscanf(client_message, ">%*[^(](%i)", &num) > 0) { // User wants a different number of requests count = num; @@ -178,16 +184,17 @@ void process_request(void) } } - sprintf(socketsendbuffer,"%i %i %s %s\n",i,clients[j].count,clients[j].ip,clients[j].name); - swrite(); + sprintf(server_message,"%i %i %s %s\n",i,clients[j].count,clients[j].ip,clients[j].name); + swrite(server_message, *sock); } if(excludeclients != NULL) clearSetupVarsArray(); if(debug) - logg("Sent top clients data to client"); + logg_int("Sent top clients data to client, ID: ", *sock); } - else if(command(">forward-dest")) + else if(command(client_message, ">forward-dest")) { + processed = true; int i, temparray[counters.forwarded][2]; for(i=0; i < counters.forwarded; i++) { @@ -202,28 +209,30 @@ void process_request(void) { // Get sorted indices int j = temparray[counters.forwarded-i-1][0]; - sprintf(socketsendbuffer,"%i %i %s %s\n",i,forwarded[j].count,forwarded[j].ip,forwarded[j].name); - swrite(); + sprintf(server_message,"%i %i %s %s\n",i,forwarded[j].count,forwarded[j].ip,forwarded[j].name); + swrite(server_message, *sock); } if(debug) - logg("Sent forwarded destinations data to client"); + logg_int("Sent forwarded destinations data to client, ID: ", *sock); } - else if(command(">querytypes")) + else if(command(client_message, ">querytypes")) { - sprintf(socketsendbuffer,"A (IPv4): %i\nAAAA (IPv6): %i\nPTR: %i\nSRV: %i\n",counters.IPv4,counters.IPv6,counters.PTR,counters.SRV); - swrite(); + processed = true; + sprintf(server_message,"A (IPv4): %i\nAAAA (IPv6): %i\nPTR: %i\nSRV: %i\n",counters.IPv4,counters.IPv6,counters.PTR,counters.SRV); + swrite(server_message, *sock); if(debug) - logg("Sent query type data to client"); + logg_int("Sent query type data to client, ID: ", *sock); } - else if(command(">getallqueries")) + else if(command(client_message, ">getallqueries")) { + processed = true; // Do we want a more specific version of this command (domain/client/time interval filtered)? int from = 0, until = 0; bool filtertime = false; - if(command(">getallqueries-time")) + if(command(client_message, ">getallqueries-time")) { // Get from to until boundaries - sscanf(socketrecvbuffer, ">getallqueries-time %i %i",&from, &until); + sscanf(client_message, ">getallqueries-time %i %i",&from, &until); if(debug) { logg_int("Showing only limited time interval starting at ",from); @@ -234,11 +243,11 @@ void process_request(void) char *domainname; bool filterdomainname = false; - if(command(">getallqueries-domain")) + if(command(client_message, ">getallqueries-domain")) { domainname = calloc(128, sizeof(char)); // Get domain name we want to see only (limit length to 127 chars) - sscanf(socketrecvbuffer, ">getallqueries-domain %127s", domainname); + sscanf(client_message, ">getallqueries-domain %127s", domainname); if(debug) logg_str("Showing only queries with domain ", domainname); filterdomainname = true; @@ -246,11 +255,11 @@ void process_request(void) char *clientname; bool filterclientname = false; - if(command(">getallqueries-client")) + if(command(client_message, ">getallqueries-client")) { clientname = calloc(128, sizeof(char)); // Get client name we want to see only (limit length to 127 chars) - sscanf(socketrecvbuffer, ">getallqueries-client %127s", clientname); + sscanf(client_message, ">getallqueries-client %127s", clientname); if(debug) logg_str("Showing only queries with client ", clientname); filterclientname = true; @@ -258,7 +267,7 @@ void process_request(void) int ibeg = 0, num; // Test for integer that specifies number of entries to be shown - if(sscanf(socketrecvbuffer, ">%*[^(](%i)", &num) > 0) + if(sscanf(client_message, ">%*[^(](%i)", &num) > 0) { // User wants a different number of requests // Don't allow a start index that is smaller than zero @@ -357,15 +366,15 @@ void process_request(void) if(!privacymode) { if(strlen(clients[queries[i].clientID].name) > 0) - sprintf(socketsendbuffer,"%i %s %s %s %i\n",queries[i].timestamp,type,domains[queries[i].domainID].domain,clients[queries[i].clientID].name,queries[i].status); + sprintf(server_message,"%i %s %s %s %i\n",queries[i].timestamp,type,domains[queries[i].domainID].domain,clients[queries[i].clientID].name,queries[i].status); else - sprintf(socketsendbuffer,"%i %s %s %s %i\n",queries[i].timestamp,type,domains[queries[i].domainID].domain,clients[queries[i].clientID].ip,queries[i].status); + sprintf(server_message,"%i %s %s %s %i\n",queries[i].timestamp,type,domains[queries[i].domainID].domain,clients[queries[i].clientID].ip,queries[i].status); } else { - sprintf(socketsendbuffer,"%i %s %s hidden %i\n",queries[i].timestamp,type,domains[queries[i].domainID].domain,queries[i].status); + sprintf(server_message,"%i %s %s hidden %i\n",queries[i].timestamp,type,domains[queries[i].domainID].domain,queries[i].status); } - swrite(); + swrite(server_message, *sock); } // Free allocated memory @@ -375,13 +384,14 @@ void process_request(void) free(domainname); if(debug) - logg("Sent all queries data to client"); + logg_int("Sent all queries data to client, ID: ", *sock); } - else if(command(">recentBlocked")) + else if(command(client_message, ">recentBlocked")) { + processed = true; int i, num=1; // Test for integer that specifies number of entries to be shown - if(sscanf(socketrecvbuffer, ">%*[^(](%i)", &num) > 0) + if(sscanf(client_message, ">%*[^(](%i)", &num) > 0) { // User wants a different number of requests if(num >= counters.queries) @@ -397,8 +407,8 @@ void process_request(void) if(queries[i].status == 1 || queries[i].status == 4) { found++; - sprintf(socketsendbuffer,"%s\n",domains[queries[i].domainID].domain); - swrite(); + sprintf(server_message,"%s\n",domains[queries[i].domainID].domain); + swrite(server_message, *sock); } if(found >= num) { @@ -406,27 +416,34 @@ void process_request(void) } } } + // End of queryable commands + if(processed) + { + // Send EOM + seom(server_message, *sock); + } + else + { + sprintf(server_message,"unknown command: %s\n",client_message); + swrite(server_message, *sock); + } + // Test only at the end if we want to quit or kill // so things can be processed before - else if(command(">quit") || command(EOT)) + if(command(client_message, ">quit") || command(client_message, EOT)) { - close(clientsocket); - clientsocket = 0; - if(debug) - logg("Clients wants to quit"); + close(*sock); + *sock = 0; } - else if(command(">kill")) + else if(command(client_message, ">kill")) { killed = 1; } - // Send EOM - if(clientsocket) - seom(); } -bool command(const char* cmd) +bool command(char *client_message, const char* cmd) { - if(strstr(socketrecvbuffer,cmd) != NULL) + if(strstr(client_message,cmd) != NULL) return true; else return false; diff --git a/routines.h b/routines.h index f831e0f8..aadc3a58 100644 --- a/routines.h +++ b/routines.h @@ -18,6 +18,7 @@ void open_FTL_log(void); void logg(const char* str); void logg_int(const char* star, int i); void logg_str(const char* str, char* str2); +void logg_str_str_int(const char* str, char* str2, char* str3, int i); void logg_struct_resize(const char* str, int from, int to); void logg_str_str(const char* str, char* str2, char* str3); void log_counter_info(void); @@ -32,15 +33,17 @@ void *pihole_log_thread(void *val); void memory_check(int which); void init_socket(void); -bool listen_socket(void); +int listen_socket(void); bool check_socket(void); void read_socket(void); void close_sockets(void); -void seom(void); -void swrite(void); +void seom(char server_message[], int sock); +void swrite(char server_message[], int sock); +void *connection_handler_thread(void *socket_desc); +void *listenting_thread(void *args); -void process_request(void); -bool command(const char* cmd); +void process_request(char *client_message, int *sock); +bool command(char *client_message, const char* cmd); void formatNumber(bool raw, int n, char* buffer); void read_gravity_files(void); diff --git a/socket.c b/socket.c index cccad110..758429e5 100644 --- a/socket.c +++ b/socket.c @@ -34,11 +34,6 @@ void init_socket(void) exit(1); } - struct timeval timeout; - timeout.tv_sec = 0; - timeout.tv_usec = 1000; - setsockopt (sockfd, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(timeout)); - struct sockaddr_in serv_addr; // The function bzero() sets all values in a buffer to zero. bzero((char *) &serv_addr, sizeof(serv_addr)); @@ -82,73 +77,6 @@ void init_socket(void) } } -bool check_socket(void) -{ - char buffer[32]; - if (recv(clientsocket, buffer, sizeof(buffer), MSG_PEEK | MSG_DONTWAIT) == 0) - { - close(clientsocket); - clientsocket = 0; - return false; - } - else - { - return true; - } -} - -bool listen_socket(void) -{ - struct sockaddr_in cli_addr; - // The function bzero() sets all values in a buffer to zero. - bzero((char *) &cli_addr, sizeof(cli_addr)); - socklen_t clilen = sizeof(cli_addr); - clientsocket = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen); - // if (newsockfd < 0) - // printf("ERROR on accept"); - if (clientsocket > 0) - { - if(debug) - logg_str("Client connected: ", inet_ntoa (cli_addr.sin_addr)); - - // const char * msg = "This is the Pi-hole FTL daemon, enter \"quit\" to quit\n\n"; - // write(clientsocket, msg, strlen(msg)); - return true; - } - else - { - return false; - } -} - -void read_socket(void) -{ - bzero(socketrecvbuffer,SOCKETBUFFERLEN); - ssize_t n = recv(clientsocket,socketrecvbuffer,SOCKETBUFFERLEN-1, MSG_DONTWAIT); - if (n > 0) - { - process_request(); - } -} - -void close_sockets(void) -{ - close(clientsocket); - close(sockfd); -} - -void seom(void) -{ - sprintf(socketsendbuffer,"---EOM---\n\n"); - swrite(); -} - -void swrite(void) -{ - if(!write(clientsocket, socketsendbuffer, strlen(socketsendbuffer))) - logg_int("WARNING: Socket write returned error code ", errno); -} - void saveport(int port) { FILE *f; @@ -164,3 +92,92 @@ void saveport(int port) } logg_int("Listening on port ", port); } + +void seom(char server_message[SOCKETBUFFERLEN], int sock) +{ + sprintf(server_message,"---EOM---\n\n"); + swrite(server_message, sock); +} + +void swrite(char server_message[SOCKETBUFFERLEN], int sock) +{ + if(!write(sock, server_message, strlen(server_message))) + logg_int("WARNING: Socket write returned error code ", errno); +} + +int listen_socket(void) +{ + struct sockaddr_in cli_addr; + // The function bzero() sets all values in a buffer to zero. + bzero((char *) &cli_addr, sizeof(cli_addr)); + socklen_t clilen = sizeof(cli_addr); + int clientsocket = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen); + // if (newsockfd < 0) + // printf("ERROR on accept"); + if(debug) + logg_str_str_int("Client connected: ", inet_ntoa (cli_addr.sin_addr), ", ID: ", clientsocket); + + return clientsocket; +} + +void *listenting_thread(void *args) +{ + int *newsock; + while(!killed) + { + int csck = listen_socket(); + + // + newsock = calloc(1,sizeof(int)); + *newsock = csck; + + pthread_t connection_thread; + if(pthread_create( &connection_thread, NULL, connection_handler_thread, (void*) newsock ) != 0) + { + logg("Unable to open Pi-hole log processing thread. Exiting..."); + killed = 1; + } + } + return 0; +} + +void *connection_handler_thread(void *socket_desc) +{ + //Get the socket descriptor + int sock = *(int*)socket_desc; + // Store copy only for displaying the debug messages + int sockID = sock; + char client_message[SOCKETBUFFERLEN] = ""; + + //Receive from client + ssize_t n; + while((n = recv(sock,client_message,SOCKETBUFFERLEN-1, 0))) + { + if (n > 0) + { + char *message = calloc(strlen(client_message)+1,sizeof(char)); + strcpy(message, client_message); + process_request(message, &sock); + free(message); + if(sock == 0) + { + // Client disconnected by seding EOT or ">quit" + break; + } + } + else if(n == -1) + { + if(debug) + logg_int("Client connection interrupted, ID: ", sockID); + } + } + if(debug) + logg_int("Client disconnected, ID: ", sockID); + + //Free the socket pointer + if(sock != 0) + close(sock); + free(socket_desc); + + return 0; +}