mserver.c

一个内存数据库的源代码这是服务器端还有客户端

#line 375 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
#include "mal_config.h"
#include "mserver.h"

#include  <sys/types.h>
#ifdef HAVE_SYS_SOCKET_H
# include <sys/socket.h>
#endif
#ifdef HAVE_WIN32
# include <winsock.h>
#endif
#ifdef HAVE_SYS_UN_H
#include <sys/un.h>
#endif

#define SOCKPTR struct sockaddr *
#if HAVE_SOCKLEN_T
#define SOCKLEN socklen_t
#else
#define SOCKLEN int
#endif

#ifdef NATIVE_WIN32
#define s_close(s)	closesocket(s)
#else
#define s_close(s)	close(s)
#endif

#ifdef HAVE_OPENSSL
#include <openssl/ssl.h>
#include <openssl/err.h>
#endif

static char seedChars[] = {'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j',
	'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x',
	'y', 'z', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L',
	'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z',
	'1', '2', '3', '4', '5', '6', '7', '8', '9', '0'};


static void generateChallenge(str buf, int min, int max) {
	size_t size;
	size_t chr;
	size_t i;

	/* don't seed the randomiser, or you get the same challenge during
	 * the same second */
	/* srand(time(NULL)); */
	size = rand();
	size = (size % (max - min)) + min;
	for (i = 0; i < size; i++) {
		chr = rand();
		chr %= 62;
		buf[i] = seedChars[chr];
	}
	buf[i] = '\0';
}

static void 
doChallenge(str cmd, int msgsock) {
	if (cmd == NULL) {
		char *buf = (char *) GDKmalloc(BLOCK);
		char *challenge;
		char *algos;
		stream *fdin = block_stream( socket_rastream(msgsock, "Server read"));
		stream *fdout = block_stream( socket_wastream(msgsock, "Server write"));
		bstream *bs;

		if (!fdin || !fdout) {
			GDKsyserror("SERVERlisten:fdin or fdout problems");
			return;
		}

		assert(buf);

		/* generate the challenge string */
		challenge = alloca(sizeof(char) * (12 + 1));
		generateChallenge(challenge, 8, 12);
		if (AUTHgetHashAlgorithms(&algos) != MAL_SUCCEED) assert (0);
		/* note that we claim to speak proto 8 here */
		stream_printf(fdout, "%s:mserver:8:%s:%s",
				challenge,
				algos,
#ifdef WORDS_BIGENDIAN
				"BIG"
#else
				"LIT"
#endif
				);
		stream_flush(fdout);
		/* get response */
		stream_read_block(fdin, buf, BLOCK, 1);

		/* in embedded mode we allow just one client */
		if (GDKembedded && MCcountClients() > 1) {
			stream_close(fdin);
			stream_destroy(fdin);
			stream_close(fdout);
			stream_destroy(fdout);
			return;
		}
#ifdef DEBUG_SERVER
		printf("mserver:Client accepted %s\n", name);
		fflush(stdout);

		THRprintf(GDKout, "#SERVERlisten:client accepted %d\n", msgsock);
		THRprintf(GDKout, "#SERVERlisten:client string %s\n", name);
#endif
		bs = bstream_create(fdin, 128 * BLOCK);

		bs->eof = 1;
		MSscheduleClient(buf, challenge, bs, fdout);
	} else {
		/* in M4 it is possible to execute a command directly */
	}
}

static char *threadcommand;
static int usock = -1;
static MT_Id listener[8]; 
static int lastlistener=0;
static int serveractive=TRUE;

str
SERVERlistenThread(int *Sock)
{
	char *msg = 0;
	char *cmd = threadcommand;

	if( lastlistener < 8 )
		listener[lastlistener++] = MT_getpid();
	(void)usock;
	do {
		int retval;
		struct timeval tv;
		fd_set fds;
		int msgsock;
		int sock = *Sock;

		FD_ZERO(&fds);
		if (sock >= 0)
			FD_SET(sock, &fds);
#ifdef HAVE_SYS_UN_H
		if (usock >= 0)
			FD_SET(usock, &fds);
#endif
		/* Wait up to 0.5 seconds. */
		tv.tv_sec = 0;
		tv.tv_usec = 500;

		/* temporarily use msgsock to record the larger of sock and usock */
		msgsock = sock;
#ifdef HAVE_SYS_UN_H
		if (usock > sock)
			msgsock = usock;
#endif
		retval = select(msgsock + 1, &fds, NULL, NULL, &tv);
		if (retval == 0) {
			/* nothing interesting has happened */
			continue;
		}
		if (retval < 0) {
			if (MT_geterrno() != EINTR) {
				msg = "select failed";
				goto error;
			}
			continue;
		}
		if (sock >= 0 && FD_ISSET(sock, &fds)) {
			if ((msgsock = accept(sock, (SOCKPTR) 0, (SOCKLEN *) 0)) < 0) {
				if (MT_geterrno() != EINTR || serveractive==FALSE) {
					msg = "accept failed\n";
					goto error;
				}
				continue;
			}
#ifdef HAVE_SYS_UN_H
		} else if (usock >= 0 && FD_ISSET(usock, &fds)) {
			if ((msgsock = accept(usock, (SOCKPTR) 0, (SOCKLEN *) 0)) < 0) {
				if (MT_geterrno() != EINTR) {
					msg = "accept failed\n";
					goto error;
				}
				continue;
			}
#endif
		} else
			continue;
#ifdef DEBUG_SERVER
		printf("server:accepted\n");
		fflush(stdout);
#endif
		doChallenge(cmd, msgsock);
	} while (1);
error:
	throw(IO, "mserver.listen", msg);
}

/**
 * Small utility function to call the sabaoth marchConnection function
 * with the right arguments.  If the socket is bound to 0.0.0.0 the
 * hostname address is used, to make the info usable for servers outside
 * localhost.
 */
void SERVERannounce(struct in_addr addr, int port, int ssl) {
	str buf = alloca(sizeof(char) * 1024);
	str host = NULL;
	bit bssl = ssl != 0 ? 1 : 0;

	if (addr.s_addr == INADDR_ANY) {
		host = alloca(sizeof(char) * (90 + 1));
		gethostname(host, 90);
		host[90] = '\0';
	} else {
		/* avoid doing this, it requires some includes that probably
		 * give trouble on windowz
		host = inet_ntoa(addr);
		 */
		host = alloca(sizeof(char) * ((3 + 1 + 3 + 1 + 3 + 1 + 3) + 1));
		sprintf(host, "%u.%u.%u.%u",
				(unsigned) ((ntohl(addr.s_addr) >> 24) & 0xff),
				(unsigned) ((ntohl(addr.s_addr) >> 16) & 0xff),
				(unsigned) ((ntohl(addr.s_addr) >> 8) & 0xff),
				(unsigned) (ntohl(addr.s_addr) & 0xff));
	}
	if ((buf = SABAOTHmarchConnection(&ssl, &host, &port, &bssl)) != MAL_SUCCEED)
		GDKfree(buf);
}

str
SERVERlisten(int *Port, str *Usockfile, int *Maxusers, str *Cmd)
{
	struct sockaddr_in server;
	int sock = -1;
	int *Sock = GDKmalloc(sizeof(int));

#ifdef HAVE_SYS_UN_H
	struct sockaddr_un userver;
#endif
	SOCKLEN length = 0;
	int on = 1;
	int i = 0;
	MT_Id pid, *pidp = &pid;
	int port;
	int maxusers;
	char msg[512], *cmd, *usockfile, host[512];

	port = *Port;
	if (Usockfile == NULL || *Usockfile == 0 || strcmp(*Usockfile, str_nil) == 0)
		usockfile = NULL;
	else {
#ifdef HAVE_SYS_UN_H
		usockfile = GDKstrdup(*Usockfile);
#else
		usockfile = NULL;
		throw(IO, "mserver.listen", "Unix domain sockets are not supported");
#endif
	}
	maxusers = *Maxusers;
	cmd = Cmd ? GDKstrdup(*Cmd) : NULL;

	maxusers = (maxusers ? maxusers : SERVERMAXUSERS);

	if (port < 0 && usockfile == NULL)
		throw(IO, "mserver.listen", "No port or socket file specified");

	if (port >= 0) {
		sock = socket(AF_INET, SOCK_STREAM, 0);
		if (sock < 0)
			throw(IO, "mserver.listen",
					"Creation of stream socket failedi: %s",
					strerror(errno));

		/* In auto-sensing mode, start at the default */
		/* FIXME: should be configurable via the config file */
		if (*Port == 0)
			port = SERVERPORT;

		setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof on);

		server.sin_family = AF_INET;
		/* FIXME: this should be configurable via the config file */
		server.sin_addr.s_addr = INADDR_ANY;
		for (i = 0; i < 8; i++)
			server.sin_zero[i] = 0;
		length = (SOCKLEN) sizeof(server);

		do {
			server.sin_port = htons((unsigned short) ((port) & 0xFFFF));
			if (bind(sock, (SOCKPTR) &server, length) < 0) {
				/* FIXME: this should come from the config file */
				if (
#ifdef EADDRINUSE
errno == EADDRINUSE &&
#else 
#ifdef WSAEADDRINUSE 
errno == WSAEADDRINUSE &&
#endif
#endif
 					*Port == 0 && port < 60000) {
					port++;
					continue;
				}
				throw(IO, "mserver.listen",
						"Binding to stream socket port %d failed: %s",
						port, strerror(errno));
			} else {
				break;
			}
		} while (1);

		if (getsockname(sock, (SOCKPTR) &server, &length) < 0)
			throw(IO, "mserver.listen",
					"Failed getting socket name: %s",
					strerror(errno));
		listen(sock, maxusers);
	}
#ifdef HAVE_SYS_UN_H
	if (usockfile) {
		usock = socket(AF_UNIX, SOCK_STREAM, 0);
		if (usock < 0) {
			unlink(usockfile);
			throw(IO, "mserver.listen",
					"Creation of Unix socket failed: %s",
					strerror(errno));
		}
		userver.sun_family = AF_UNIX;
		strncpy(userver.sun_path, usockfile, sizeof(userver.sun_path));

		length = (SOCKLEN) sizeof(userver);
		if (bind(usock, (SOCKPTR) & userver, length) < 0) {
			unlink(usockfile);
			throw(IO, "mserver.listen",
					"Binding to Unix socket file %s failed: %s",
					usockfile, strerror(errno));
		}
		listen(usock, maxusers);
	}
#endif

#ifdef DEBUG_SERVER
	THRprintf(GDKout, "#SERVERlisten:Network started at %d\n", port);
#endif
	threadcommand = cmd;
	*Sock = sock;

	if (MT_create_thread(pidp, (void (*)(void *)) SERVERlistenThread, Sock) < 0) {
		throw(MAL, "mserver.listen", "Starting thread failed");
	}
#ifdef HAVE_SYS_UN_H
	if (usockfile)
		unlink(usockfile);
#endif
	gethostname(host, (int) 512);
	(void) msg;
#ifdef DEBUG_SERVER
	snprintf(msg, (int) 512, "#Ready to accept connections on %s:%d\n", host, port);
	stream_printf(GDKout, "%s", msg);
#endif
	SERVERannounce(server.sin_addr, port, 0);
	return MAL_SUCCEED;
}

#ifdef HAVE_OPENSSL
static char *
ssl_error(const char *name, int err, int ret)
{
	char *errstr, *s;
	char buf[120];
	unsigned long e;

	switch (err) {
	case SSL_ERROR_ZERO_RETURN:
		errstr = "TLS/SSL connection has been closed";
		break;
	case SSL_ERROR_WANT_READ:
		errstr = "The operation did not complete (read)";
		break;
	case SSL_ERROR_WANT_WRITE:
		errstr = "The operation did not complete (write)";
		break;
	case SSL_ERROR_WANT_X509_LOOKUP:
		errstr = "The operation did not complete (X509 lookup)";
		break;
	case SSL_ERROR_WANT_CONNECT:
		errstr = "The operation did not complete (connect)";
		break;
	case SSL_ERROR_SYSCALL:
		e = ERR_get_error();
		if (e == 0) {
			if (ret == 0) {
				errstr = "EOF occurred in violation of protocol";
			} else if (ret == -1) {
				/* the underlying BIO reported an I/O error */
				errstr = "I/O error";
			} else {	/* possible? */
				errstr = "Some I/O error occurred";
			}
		} else {
			errstr = ERR_error_string(e, buf);
		}
		break;
	case SSL_ERROR_SSL:
		e = ERR_get_error();
		if (e != 0)
			errstr = ERR_error_string(e, buf);
		else {		/* possible? */
			errstr = "A failure in the SSL library occurred";
		}
		break;
	default:
		errstr = "Invalid error code";
	}
	s = (char*) GDKmalloc(strlen(errstr) + strlen(name) + 4);
	sprintf(s, "%s: %s\n", name, errstr);	/* we allocated enough, so it fits */
	return s;
}

static MT_Lock *mutex_buf;

static void
locking_function(int mode, int n, const char *file, int line)
{
	(void) file;
	(void) line;
	if (mode & CRYPTO_LOCK){
		MT_set_lock(mutex_buf[n], "locking_function");
	} else {
		MT_unset_lock(mutex_buf[n], "locking_function");
	}
}

static unsigned long
id_function(void)
{
	return (unsigned long) MT_getpid();
}
#endif /* HAVE_OPENSSL */

bat *
SERVERprelude(void)
{
#ifdef HAVE_OPENSSL
	int i, nlocks;

	nlocks = CRYPTO_num_locks();
	mutex_buf = GDKmalloc(nlocks * sizeof(*mutex_buf));
	if (mutex_buf == NULL) {
		GDKsyserror("SERVERprelude: failed to allocate %d mutexes\n", nlocks);
		return NULL;
	}
	for (i = 0; i < nlocks; i++)
		MT_lock_init(&mutex_buf[i]);
	CRYPTO_set_locking_callback(locking_function);
	CRYPTO_set_id_callback(id_function);
#endif
	return NULL;
}

void
SERVERepilogue(void)
{
#ifdef HAVE_OPENSSL
	if (mutex_buf) {
		int i, nlocks;

		nlocks = CRYPTO_num_locks();
		CRYPTO_set_id_callback(NULL);
		CRYPTO_set_locking_callback(NULL);
		for (i = 0; i < nlocks; i++)
			MT_destroy_lock(mutex_buf[i]);
		GDKfree(mutex_buf);
		mutex_buf = NULL;
	}
	/* callString(mal_clients, "sabaoth.retreatConnection(...);", 0); */
#endif /* HAVE_OPENSSL */
	/* callString(mal_clients, "sabaoth.retreatConnection(...);", 0); */
}

str
SERVERlistenSSL(int *Port, int *Maxusers, str keyfile, str certfile, str cmd)
{
#ifdef HAVE_OPENSSL
	struct sockaddr_in server;
	int sock = -1;
	SOCKLEN length = 0;
	int on = 1;
	int msgsock;
	int i = 0;

	int port = *Port;
	int maxusers = *Maxusers;
	SSL_CTX *ctx = NULL;
	SSL *ssl = NULL;
	char *msg = NULL;

	if (!port)
		port = SERVERSSLPORT;
	if (!maxusers)
		maxusers = SERVERMAXUSERS;

	ctx = SSL_CTX_new(SSLv23_method());
	if (ctx == NULL) {
		msg = "creation of SSL context failed\n";
		goto fail;
	}
	if (SSL_CTX_set_cipher_list(ctx, "ALL:!LOW") == 0) {
		msg = "SSL_CTX_set_cipher_list failed\n";
		goto fail;
	}
	if (keyfile && *keyfile && certfile && *certfile) {
		if (SSL_CTX_use_PrivateKey_file(ctx, keyfile, SSL_FILETYPE_PEM) < 1) {
			msg = "SSL_CTX_use_PrivateKey_file failed\n";
			goto fail;
		}
		if (SSL_CTX_use_certificate_chain_file(ctx, certfile) < 1) {
			msg = "SSL_CTX_use_certificate_chain_file failed\n";
			goto fail;
		}
	}
	if (keyfile)
		GDKfree(keyfile);
	if (certfile)
		GDKfree(certfile);

	SSL_CTX_set_verify(ctx, SSL_VERIFY_NONE, NULL);

	sock = socket(AF_INET, SOCK_STREAM, 0);
	if (sock < 0) {
		GDKsyserror("SERVERlistenSSL:creation of stream socket failed\n");
		goto fail;
	}


#line 911 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
	server.sin_family = AF_INET;
	server.sin_addr.s_addr = INADDR_ANY;
	server.sin_port = htons((unsigned short) ((port) & 0xFFFF));
	for (i = 0; i < 8; i++)
		server.sin_zero[i] = 0;

	setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof on);

	length = (SOCKLEN) sizeof(server);
	if (bind(sock, (SOCKPTR) & server, length) < 0) {
		GDKsyserror("SERVERlistenSSL:binding to stream socket (%d) failed\n", port);
		goto fail;
	}


#line 927 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
	if (getsockname(sock, (SOCKPTR) & server, &length) < 0) {
		msg = "getting socket name\n";
		goto fail;
	}
	listen(sock, maxusers);

#ifdef DEBUG_SERVER