netsukuku/src/daemon.c

/* This file is part of Netsukuku
 * (c) Copyright 2005 Andrea Lo Pumo aka AlpT <alpt@freaknet.org>
 *
 * This source code is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License as published 
 * by the Free Software Foundation; either version 2 of the License,
 * or (at your option) any later version.
 *
 * This source code is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * Please refer to the GNU Public License for more details.
 *
 * You should have received a copy of the GNU Public License along with
 * this source code; if not, write to:
 * Free Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */

#include "includes.h"

#include "common.h"
#include "inet.h"
#include "request.h"
#include "if.h"
#include "pkts.h"
#include "bmap.h"
#include "daemon.h"
#include "netsukuku.h"
#include "accept.h"

extern int errno;

/* 
 * prepare_listen_socket: 
 * It creates a new socket of the desired `family' and binds it to the 
 * specified `port'. It sets also the reuseaddr and NONBLOCK
 * socket options, because this new socket shall be used to listen() and
 * accept().
 * If `dev' is not null, the socket will be binded to the device named 
 * `dev'->dev_name with the SO_BINDTODEVICE socket option.
 * The created socket is returned.
 */
int prepare_listen_socket(int family, int socktype, u_short port, 
		interface *dev)
{
	struct addrinfo hints, *ai, *aitop;
	char strport[NI_MAXSERV];
	int err, s;

	setzero(&hints, sizeof(struct addrinfo));
	hints.ai_family=family;
	hints.ai_socktype=socktype;
	hints.ai_flags=AI_PASSIVE;
	snprintf(strport, NI_MAXSERV, "%u", port);
	
	err=getaddrinfo(NULL, strport, &hints, &aitop);
	if(err) {
		error("Getaddrinfo error: %s", gai_strerror(err));
		return -1;
	}

	for (ai = aitop; ai; ai = ai->ai_next) {
		if (ai->ai_family != AF_INET && ai->ai_family != AF_INET6)
			continue;
		
		s = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
		if (s == -1)
			/* Maybe we can use another socket...*/
			continue;

		/* Bind the created socket to the device named dev->dev_name */
		if(dev && (set_bindtodevice_sk(s, dev->dev_name) < 0)) {
			inet_close(&s);
			continue;
		}

		if(set_reuseaddr_sk(s) < 0) {
			inet_close(&s);
			continue;
		}

		/* Let's bind it! */
		if(bind(s, ai->ai_addr, ai->ai_addrlen) < 0) {
			error("Cannot bind the port %d: %s. Trying another "
					"socket...", port, strerror(errno));
			inet_close(&s);
			continue;
		}
		freeaddrinfo(aitop);
		return s;
	}
	
	error("Cannot open inbound socket on port %d: %s", port, strerror(errno));
	freeaddrinfo(aitop);
	return -1;
}


/*
 * sockets_all_ifs
 *
 * creates a socket for each interface which is in the `ifs' array. 
 * The array has `ifs_n' members.
 * Each created socket is stored in the `dev_sk' array, which has `ifs_n'#
 * members.
 * The created socket will be bound to the relative interface.
 * In `max_sk_idx' is stored the index of the `dev_sk' array, which has the
 * biggest dev_sk.
 *
 * On error 0 is returned, otherwise the number of utilised interfaces is
 * returned.
 * If the error is fatal, a negative value is returned.
 */
int sockets_all_ifs(int family, int socktype, u_short port, 
			interface *ifs, int ifs_n, 
			int *dev_sk, int *max_sk_idx)
{
	int i, n, e=0;

	*max_sk_idx=0;

	for(i=0, n=0; i<ifs_n; i++) {
		dev_sk[i] = prepare_listen_socket(family, socktype, port,
				&ifs[i]);
		
		if(dev_sk[i] < 0) {
			error("Cannot create a socket on the %s interface! "
					"Ignoring it", ifs[i].dev_name);
			dev_sk[i]=0;
			e++;
			continue;
		}

		if(dev_sk[i] >= dev_sk[*max_sk_idx])
			*max_sk_idx=i;
		
		n++;
	}

	if(e == ifs_n)
		return -1;

	return n;
}

/*
 *  udp_exec_pkt: passes the received udp packet to pkt_exec().
 * `passed_argv' is a pointer to a udp_exec_pkt_argv struct 
 */
void *udp_exec_pkt(void *passed_argv)
{
	struct udp_exec_pkt_argv argv;
	
	PACKET rpkt;
	const char *ntop;

	memcpy(&argv, passed_argv, sizeof(struct udp_exec_pkt_argv));
	memcpy(&rpkt, argv.recv_pkt, sizeof(PACKET));

	if(argv.flags & UDP_THREAD_FOR_EACH_PKT)
		pthread_mutex_unlock(&udp_exec_lock);
	
	/* Drop any packet we sent in broadcast */
	if(!memcmp(rpkt.from.data, me.cur_ip.data, MAX_IP_SZ)) {
		pkt_free(&rpkt, 0);
		return 0;
	}

	if(add_accept(rpkt.from, 1)) {
		ntop=inet_to_str(rpkt.from);
		debug(DBG_NORMAL, "ACPT: dropped UDP pkt from %s: "
				"Accept table full.", ntop);
		return 0;
	} 

	pkt_exec(rpkt, argv.acpt_idx);
	pkt_free(&rpkt, 0);
	
	return 0;
}

/*
 * udp_daemon: Takes care of receiving udp packets.
 * `passed_argv' is a pointer to a udp_daemon_argv struct
 */
void *udp_daemon(void *passed_argv)
{
	struct udp_daemon_argv argv;
	struct udp_exec_pkt_argv exec_pkt_argv;
	
	interface *ifs;
	int max_sk_idx, dev_sk[me.cur_ifs_n];

	PACKET rpkt;
	fd_set fdset;
	int ret, i, err;
	u_short udp_port;
	
	pthread_t thread;
	pthread_attr_t t_attr;
	
#ifdef DEBUG
	int select_errors=0;
#endif
	
	memcpy(&argv, passed_argv, sizeof(struct udp_daemon_argv));
	udp_port=argv.port;
	setzero(&exec_pkt_argv, sizeof(struct udp_exec_pkt_argv));

	if(argv.flags & UDP_THREAD_FOR_EACH_PKT) {
		pthread_attr_init(&t_attr);
		pthread_attr_setdetachstate(&t_attr, PTHREAD_CREATE_DETACHED);
		exec_pkt_argv.flags|=UDP_THREAD_FOR_EACH_PKT;
	}

	debug(DBG_SOFT, "Preparing the udp listening socket on port %d", udp_port);
	
	err=sockets_all_ifs(my_family, SOCK_DGRAM, udp_port, me.cur_ifs,
				me.cur_ifs_n, dev_sk, &max_sk_idx);
	if(!err)
		return NULL;
	else if(err < 0)
		fatal("Creation of the %s daemon aborted. "
			"Is there another ntkd running?", "udp");
	
	debug(DBG_NORMAL, "Udp daemon on port %d up & running", udp_port);
	pthread_mutex_unlock(&udp_daemon_lock);

	pthread_mutex_init(&udp_exec_lock, 0);
	
	for(;;) {
		FD_ZERO(&fdset);

		if(!me.cur_ifs_n) {
			/* All the devices have been removed while ntkd was
			 * running, sleep well */
			sleep(1);
			continue;
		}

		for(i=0; i < me.cur_ifs_n; i++)
			if(dev_sk[i])
				FD_SET(dev_sk[i], &fdset);

		ret=select(dev_sk[max_sk_idx]+1, &fdset, NULL, NULL, NULL);
		if(sigterm_timestamp)
			/* NetsukukuD has been closed */
			break;
		if (ret < 0) {
#ifdef DEBUG
			if(select_errors > 20)
				break;
			select_errors++;
#endif
			error("daemon_udp: select error: %s", strerror(errno));
			continue;
		}

		for(i=0; i < me.cur_ifs_n; i++) {
			ifs=&me.cur_ifs[i];
			if(!dev_sk[i])
				continue;
			
			if(!FD_ISSET(dev_sk[i], &fdset))
				continue;

			setzero(&rpkt, sizeof(PACKET));
			pkt_addsk(&rpkt, my_family, dev_sk[i], SKT_UDP);
			pkt_add_dev(&rpkt, ifs, 0);
			rpkt.flags=MSG_WAITALL;
			pkt_addport(&rpkt, udp_port);

			if(pkt_recv(&rpkt) < 0) {
				pkt_free(&rpkt, 0);
				continue;
			}

			exec_pkt_argv.acpt_idx=accept_idx;
			exec_pkt_argv.acpt_sidx=accept_sidx;

			if(argv.flags & UDP_THREAD_FOR_EACH_PKT) {
				exec_pkt_argv.recv_pkt=&rpkt;
				pthread_mutex_lock(&udp_exec_lock);
				pthread_create(&thread, &t_attr, udp_exec_pkt,
						&exec_pkt_argv);
				pthread_mutex_lock(&udp_exec_lock);
				pthread_mutex_unlock(&udp_exec_lock);
			} else {
				exec_pkt_argv.recv_pkt=&rpkt;
				udp_exec_pkt(&exec_pkt_argv);
			}
		}
	}

	destroy_accept_tbl();
	return NULL;
}

void *tcp_recv_loop(void *recv_pkt)
{
	PACKET rpkt;
	int acpt_idx, acpt_sidx;

	acpt_idx=accept_idx;
	acpt_sidx=accept_sidx;
	memcpy(&rpkt, recv_pkt, sizeof(PACKET));
	pthread_mutex_unlock(&tcp_exec_lock);

#if 0
	add_accept_pid(getpid(), acpt_idx, acpt_sidx);
#endif

	while( pkt_recv(&rpkt) != -1 ) {
		if(pkt_exec(rpkt, acpt_idx) < 0) {
			goto close;
			break;
		} else
			pkt_free(&rpkt, 0);
	}

close:
	pkt_free(&rpkt, 1);
	close_accept(acpt_idx, acpt_sidx);

	return NULL;
}

void *tcp_daemon(void *door)
{
	pthread_t thread;
	pthread_attr_t t_attr;
	
	PACKET rpkt;
	struct sockaddr_storage addr;
	socklen_t addrlen = sizeof addr;
	inet_prefix ip;
	
	fd_set fdset;
	int fd, ret, err, i;

	interface *ifs;
	int max_sk_idx, dev_sk[me.cur_ifs_n];
	
	u_short tcp_port=*(u_short *)door;
	const char *ntop;

	pthread_attr_init(&t_attr);
	pthread_attr_setdetachstate(&t_attr, PTHREAD_CREATE_DETACHED);
	
	debug(DBG_SOFT, "Preparing the tcp listening socket on port %d", tcp_port);

	err=sockets_all_ifs(my_family, SOCK_STREAM, tcp_port, me.cur_ifs, 
				me.cur_ifs_n, dev_sk, &max_sk_idx);
	if(!err)
		return NULL;
	else if(err < 0)
		fatal("Creation of the %s daemon aborted. "
			"Is there another ntkd running?", "tcp");

	pthread_mutex_init(&tcp_exec_lock, 0);

	for(i=0; i<me.cur_ifs_n; i++) {
		if(!dev_sk[i])
			continue;
		/* 
		 * While we are accepting the connections we keep the socket non
		 * blocking.
		 */
		if(set_nonblock_sk(dev_sk[i])) {
		  pthread_mutex_unlock(&tcp_daemon_lock);
		  return NULL;
		}
		
 		/* Shhh, it's listening... */
  		if(listen(dev_sk[i], 5) == -1) {
  			inet_close(&dev_sk[i]);
			pthread_mutex_unlock(&tcp_daemon_lock);
  			return NULL;
  		}
  	}
	
	debug(DBG_NORMAL, "Tcp daemon on port %d up & running", tcp_port);
	pthread_mutex_unlock(&tcp_daemon_lock);
	for(;;) {
		FD_ZERO(&fdset);

		if(!me.cur_ifs_n) {
			/* All the devices have been removed while ntkd was
			 * running, sleep well */
			sleep(1);
			continue;
		}

		for(i=0; i < me.cur_ifs_n; i++)
			if(dev_sk[i])
				FD_SET(dev_sk[i], &fdset);

		ret=select(dev_sk[max_sk_idx]+1, &fdset, NULL, NULL, NULL);
		if(sigterm_timestamp)
			/* NetsukukuD has been closed */
			break;
		if(ret < 0 && errno != EINTR)
			error("daemon_tcp: select error: %s", strerror(errno));
		if(ret < 0)
			continue;

		for(i=0; i < me.cur_ifs_n; i++) {
			ifs=&me.cur_ifs[i];
			if(!dev_sk[i])
				continue;

			if(!FD_ISSET(dev_sk[i], &fdset))
				continue;

			fd=accept(dev_sk[i], (struct sockaddr *)&addr, &addrlen);
			if(fd == -1) {
				if (errno != EINTR && errno != EWOULDBLOCK)
					error("daemon_tcp: accept(): %s", strerror(errno));
				continue;
			}

			setzero(&rpkt, sizeof(PACKET));
			pkt_addsk(&rpkt, my_family, fd, SKT_TCP);
			pkt_add_dev(&rpkt, ifs, 0);
			rpkt.flags=MSG_WAITALL;
			pkt_addport(&rpkt, tcp_port);

			ntop=0;
			sockaddr_to_inet((struct sockaddr *)&addr, &ip, 0);
			pkt_addfrom(&rpkt, &ip);
			if(server_opt.dbg_lvl)
				ntop=inet_to_str(ip);

			if((ret=add_accept(ip, 0))) {
				debug(DBG_NORMAL, "ACPT: drop connection with %s: "
						"Accept table full.", ntop);

				/* Omg, we cannot take it anymore, go away: ACK_NEGATIVE */
				pkt_err(rpkt, ret, 1);
				inet_close(&fd);
				continue;
			} else {
				/* 
				 * Ok, the connection is good, send back the
				 * ACK_AFFERMATIVE.
				 */
				pkt_addto(&rpkt, &rpkt.from);
				send_rq(&rpkt, 0, ACK_AFFERMATIVE, 0, 0, 0, 0);
			}

			if(unset_nonblock_sk(fd))
				continue;

			pthread_mutex_lock(&tcp_exec_lock);
			err=pthread_create(&thread, &t_attr, tcp_recv_loop, (void *)&rpkt);
			pthread_detach(thread);
			pthread_mutex_lock(&tcp_exec_lock);
			pthread_mutex_unlock(&tcp_exec_lock);
		}
	}
	return NULL;
}