<commit_before>/*
 * The contents of this file are subject to the MonetDB Public License
 * Version 1.1 (the "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 * http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html
 *
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
 * License for the specific language governing rights and limitations
 * under the License.
 *
 * The Original Code is the MonetDB Database System.
 *
 * The Initial Developer of the Original Code is CWI.
 * Portions created by CWI are Copyright (C) 1997-July 2008 CWI.
 * Copyright August 2008-2010 MonetDB B.V.
 * All Rights Reserved.
 */

#include "sql_config.h"

#include <stdio.h>
#include <string.h>  /* strerror, strchr, strcmp */
#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <netdb.h>
#include <netinet/in.h>
#ifdef HAVE_ALLOCA_H
#include <alloca.h>
#endif

#include <stream.h>
#include <stream_socket.h>
#include <mal_sabaoth.h>
#include <utils/utils.h> /* freeConfFile */
#include <utils/properties.h> /* readProps */

#include "merovingian.h"
#include "forkmserver.h"
#include "proxy.h"
#include "multiplex-funnel.h"

static multiplex *mero_multiplex_funnel = NULL;

static err
handleClient(int sock, char isusock)
{
	stream *fdin, *fout;
	str buf = alloca(sizeof(char) * 8096);
	char *user = NULL, *algo = NULL, *passwd = NULL, *lang = NULL;
	char *database = NULL, *s;
	char *host = NULL;
	sabdb *top = NULL;
	sabdb *stat = NULL;
	struct sockaddr_in saddr;
	socklen_t saddrlen = sizeof(struct sockaddr_in);
	err e;
	confkeyval *ckv, *kv;
	char mydoproxy;
	sabdb redirs[24];  /* do we need more? */
	int r = 0;

	fdin = socket_rastream(sock, "merovingian<-client (read)");
	if (fdin == 0)
		return(newErr("merovingian-client inputstream problems"));
	fdin = block_stream(fdin);

	fout = socket_wastream(sock, "merovingian->client (write)");
	if (fout == 0) {
		close_stream(fdin);
		return(newErr("merovingian-client outputstream problems"));
	}
	fout = block_stream(fout);

	if (isusock) {
		host = "(local)";
	} else if (getpeername(sock, (struct sockaddr *)&saddr, &saddrlen) == -1) {
		Mfprintf(stderr, "couldn't get peername of client: %s\n",
				strerror(errno));
		host = "(unknown)";
	} else {
		size_t len;
		struct hostent *hoste = 
			gethostbyaddr(&saddr.sin_addr.s_addr, 4, saddr.sin_family);
		if (hoste == NULL) {
			len = (3 + 1 + 3 + 1 + 3 + 1 + 3 + 1 + 5) + 1;
			host = alloca(sizeof(char) * len);
			snprintf(host, len, "%u.%u.%u.%u:%u",
					(unsigned) ((ntohl(saddr.sin_addr.s_addr) >> 24) & 0xff),
					(unsigned) ((ntohl(saddr.sin_addr.s_addr) >> 16) & 0xff),
					(unsigned) ((ntohl(saddr.sin_addr.s_addr) >> 8) & 0xff),
					(unsigned) (ntohl(saddr.sin_addr.s_addr) & 0xff),
					(unsigned) (ntohs(saddr.sin_port)));
		} else {
			len = strlen(hoste->h_name) + 1 + 5 + 1;
			host = alloca(sizeof(char) * len);
			snprintf(host, len, "%s:%u",
					hoste->h_name, (unsigned) (ntohs(saddr.sin_port)));
		}
	}

	/* note that we claim to speak proto 8 here */
	mnstr_printf(fout, "%s:merovingian:8:%s:%s:",
			"void",  /* some bs */
			"md5,plain", /* we actually don't look at the password */
#ifdef WORDS_BIGENDIAN
			"BIG"
#else
			"LIT"
#endif
			);
	mnstr_flush(fout);

	/* get response */
	buf[0] = '\0';
	if (mnstr_read_block(fdin, buf, 8095, 1) < 0) {
		/* we didn't get a terminated block :/ */
		e = newErr("client %s sent challenge in incomplete block: %s",
				host, buf);
		mnstr_printf(fout, "!merovingian: client sent something this "
				"server could not understand, sorry\n", user);
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(e);
	}
	buf[8095] = '\0';

	/* decode BIG/LIT:user:{cypher}passwordchal:lang:database: line */

	user = buf;
	/* byte order */
	s = strchr(user, ':');
	if (s) {
		*s = 0;
		/* we don't use this in merovingian */
		/* mnstr_set_byteorder(fin->s, strcmp(user, "BIG") == 0); */
		user = s + 1;
	} else {
		e = newErr("client %s challenge error: %s", host, buf);
		mnstr_printf(fout, "!merovingian: incomplete challenge '%s'\n", user);
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(e);
	}

	/* passwd */
	s = strchr(user, ':');
	if (s) {
		*s = 0;
		passwd = s + 1;
		/* decode algorithm, i.e. {plain}mypasswordchallenge */
		if (*passwd != '{') {
			e = newErr("client %s challenge error: %s", host, buf);
			mnstr_printf(fout, "!merovingian: invalid password entry\n");
			mnstr_flush(fout);
			close_stream(fout);
			close_stream(fdin);
			return(e);
		}
		algo = passwd + 1;
		s = strchr(algo, '}');
		if (!s) {
			e = newErr("client %s challenge error: %s", host, buf);
			mnstr_printf(fout, "!merovingian: invalid password entry\n");
			mnstr_flush(fout);
			close_stream(fout);
			close_stream(fdin);
			return(e);
		}
		*s = 0;
		passwd = s + 1;
	} else {
		e = newErr("client %s challenge error: %s", host, buf);
		mnstr_printf(fout, "!merovingian: incomplete challenge '%s'\n", user);
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(e);
	}

	/* lang */
	s = strchr(passwd, ':');
	if (s) {
		*s = 0;
		lang = s + 1;
	} else {
		e = newErr("client %s challenge error: %s", host, buf);
		mnstr_printf(fout, "!merovingian: incomplete challenge, missing language\n");
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(e);
	}

	/* database */
	s = strchr(lang, ':');
	if (s) {
		*s = 0;
		database = s + 1;
		/* since we don't know where the string ends, we need to look
		 * for another : */
		s = strchr(database, ':');
		if (s == NULL) {
			e = newErr("client %s challenge error: %s", host, buf);
			mnstr_printf(fout, "!merovingian: incomplete challenge, missing trailing colon\n");
			mnstr_flush(fout);
			close_stream(fout);
			close_stream(fdin);
			return(e);
		} else {
			*s = '\0';
		}
	}

	if (*database == '\0') {
		/* we need to have a database, if we haven't gotten one,
		 * complain */
		mnstr_printf(fout, "!merovingian: please specify a database\n");
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(newErr("client %s specified no database", host));
	}

	if (strcmp(lang, "multiplex-funnel") == 0) {
		/* SQL multiplexer with funnelling capabilities */
		/* find/start/attach funnel */
		/* FIXME: maintain a list, search, match name, etc.
		if (strcmp(_multiplex_funnel->name, database) == 0)
		*/
		if (mero_multiplex_funnel == NULL) {
			mero_multiplex_funnel = multiplexInit(database);
			if (pthread_create(&mero_multiplex_funnel->tid,
					NULL, (void *(*)(void *))multiplexThread,
					(void *)mero_multiplex_funnel) < 0)
			{
				Mfprintf(stderr, "starting multiplex-funnel %s failed\n",
						database);
			}
		}
		multiplexAddClient(mero_multiplex_funnel, sock, fout, fdin, host);

		return(NO_ERR);
	}

	if (strcmp(lang, "resolve") == 0) {
		/* ensure the pattern ends with '/\*' such that we force a
		 * remote entry, including those for local databases, this
		 * way we will get a redirect back to merovingian for such
		 * database if it is proxied and hence not remotely
		 * available */
		size_t len = strlen(database);
		if (len > 2 &&
				database[len - 2] != '/' &&
				database[len - 1] != '*')
		{
			char *n = alloca(sizeof(char) * len + 2 + 1);
			snprintf(n, len + 2 + 1, "%s/*", database);
			database = n;
		}
	}

	if ((e = forkMserver(database, &top, 0)) != NO_ERR) {
		if (top == NULL) {
			mnstr_printf(fout, "!merovingian: no such database '%s', please create it first\n", database);
		} else {
			mnstr_printf(fout, "!merovingian: internal error while starting mserver, please refer to the logs\n");
		}
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);<commit_msg>allow multiple multiplex-funnels to exist<commit_after>/*
 * The contents of this file are subject to the MonetDB Public License
 * Version 1.1 (the "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 * http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html
 *
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
 * License for the specific language governing rights and limitations
 * under the License.
 *
 * The Original Code is the MonetDB Database System.
 *
 * The Initial Developer of the Original Code is CWI.
 * Portions created by CWI are Copyright (C) 1997-July 2008 CWI.
 * Copyright August 2008-2010 MonetDB B.V.
 * All Rights Reserved.
 */

#include "sql_config.h"

#include <stdio.h>
#include <string.h>  /* strerror, strchr, strcmp */
#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <netdb.h>
#include <netinet/in.h>
#ifdef HAVE_ALLOCA_H
#include <alloca.h>
#endif

#include <stream.h>
#include <stream_socket.h>
#include <mal_sabaoth.h>
#include <utils/utils.h> /* freeConfFile */
#include <utils/properties.h> /* readProps */

#include "merovingian.h"
#include "forkmserver.h"
#include "proxy.h"
#include "multiplex-funnel.h"

typedef struct _mplist {
	multiplex *mpf;
	struct _mplist *next;
} mplist;

static mplist *mero_multiplex_funnel = NULL;

static err
handleClient(int sock, char isusock)
{
	stream *fdin, *fout;
	str buf = alloca(sizeof(char) * 8096);
	char *user = NULL, *algo = NULL, *passwd = NULL, *lang = NULL;
	char *database = NULL, *s;
	char *host = NULL;
	sabdb *top = NULL;
	sabdb *stat = NULL;
	struct sockaddr_in saddr;
	socklen_t saddrlen = sizeof(struct sockaddr_in);
	err e;
	confkeyval *ckv, *kv;
	char mydoproxy;
	sabdb redirs[24];  /* do we need more? */
	int r = 0;

	fdin = socket_rastream(sock, "merovingian<-client (read)");
	if (fdin == 0)
		return(newErr("merovingian-client inputstream problems"));
	fdin = block_stream(fdin);

	fout = socket_wastream(sock, "merovingian->client (write)");
	if (fout == 0) {
		close_stream(fdin);
		return(newErr("merovingian-client outputstream problems"));
	}
	fout = block_stream(fout);

	if (isusock) {
		host = "(local)";
	} else if (getpeername(sock, (struct sockaddr *)&saddr, &saddrlen) == -1) {
		Mfprintf(stderr, "couldn't get peername of client: %s\n",
				strerror(errno));
		host = "(unknown)";
	} else {
		size_t len;
		struct hostent *hoste = 
			gethostbyaddr(&saddr.sin_addr.s_addr, 4, saddr.sin_family);
		if (hoste == NULL) {
			len = (3 + 1 + 3 + 1 + 3 + 1 + 3 + 1 + 5) + 1;
			host = alloca(sizeof(char) * len);
			snprintf(host, len, "%u.%u.%u.%u:%u",
					(unsigned) ((ntohl(saddr.sin_addr.s_addr) >> 24) & 0xff),
					(unsigned) ((ntohl(saddr.sin_addr.s_addr) >> 16) & 0xff),
					(unsigned) ((ntohl(saddr.sin_addr.s_addr) >> 8) & 0xff),
					(unsigned) (ntohl(saddr.sin_addr.s_addr) & 0xff),
					(unsigned) (ntohs(saddr.sin_port)));
		} else {
			len = strlen(hoste->h_name) + 1 + 5 + 1;
			host = alloca(sizeof(char) * len);
			snprintf(host, len, "%s:%u",
					hoste->h_name, (unsigned) (ntohs(saddr.sin_port)));
		}
	}

	/* note that we claim to speak proto 8 here */
	mnstr_printf(fout, "%s:merovingian:8:%s:%s:",
			"void",  /* some bs */
			"md5,plain", /* we actually don't look at the password */
#ifdef WORDS_BIGENDIAN
			"BIG"
#else
			"LIT"
#endif
			);
	mnstr_flush(fout);

	/* get response */
	buf[0] = '\0';
	if (mnstr_read_block(fdin, buf, 8095, 1) < 0) {
		/* we didn't get a terminated block :/ */
		e = newErr("client %s sent challenge in incomplete block: %s",
				host, buf);
		mnstr_printf(fout, "!merovingian: client sent something this "
				"server could not understand, sorry\n", user);
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(e);
	}
	buf[8095] = '\0';

	/* decode BIG/LIT:user:{cypher}passwordchal:lang:database: line */

	user = buf;
	/* byte order */
	s = strchr(user, ':');
	if (s) {
		*s = 0;
		/* we don't use this in merovingian */
		/* mnstr_set_byteorder(fin->s, strcmp(user, "BIG") == 0); */
		user = s + 1;
	} else {
		e = newErr("client %s challenge error: %s", host, buf);
		mnstr_printf(fout, "!merovingian: incomplete challenge '%s'\n", user);
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(e);
	}

	/* passwd */
	s = strchr(user, ':');
	if (s) {
		*s = 0;
		passwd = s + 1;
		/* decode algorithm, i.e. {plain}mypasswordchallenge */
		if (*passwd != '{') {
			e = newErr("client %s challenge error: %s", host, buf);
			mnstr_printf(fout, "!merovingian: invalid password entry\n");
			mnstr_flush(fout);
			close_stream(fout);
			close_stream(fdin);
			return(e);
		}
		algo = passwd + 1;
		s = strchr(algo, '}');
		if (!s) {
			e = newErr("client %s challenge error: %s", host, buf);
			mnstr_printf(fout, "!merovingian: invalid password entry\n");
			mnstr_flush(fout);
			close_stream(fout);
			close_stream(fdin);
			return(e);
		}
		*s = 0;
		passwd = s + 1;
	} else {
		e = newErr("client %s challenge error: %s", host, buf);
		mnstr_printf(fout, "!merovingian: incomplete challenge '%s'\n", user);
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(e);
	}

	/* lang */
	s = strchr(passwd, ':');
	if (s) {
		*s = 0;
		lang = s + 1;
	} else {
		e = newErr("client %s challenge error: %s", host, buf);
		mnstr_printf(fout, "!merovingian: incomplete challenge, missing language\n");
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(e);
	}

	/* database */
	s = strchr(lang, ':');
	if (s) {
		*s = 0;
		database = s + 1;
		/* since we don't know where the string ends, we need to look
		 * for another : */
		s = strchr(database, ':');
		if (s == NULL) {
			e = newErr("client %s challenge error: %s", host, buf);
			mnstr_printf(fout, "!merovingian: incomplete challenge, missing trailing colon\n");
			mnstr_flush(fout);
			close_stream(fout);
			close_stream(fdin);
			return(e);
		} else {
			*s = '\0';
		}
	}

	if (*database == '\0') {
		/* we need to have a database, if we haven't gotten one,
		 * complain */
		mnstr_printf(fout, "!merovingian: please specify a database\n");
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
		return(newErr("client %s specified no database", host));
	}

	if (strcmp(lang, "multiplex-funnel") == 0) {
		/* SQL multiplexer with funnelling capabilities */
		/* find/start/attach funnel */
		mplist *w;
		for (w = mero_multiplex_funnel; w != NULL; w = w->next) {
			if (strcmp(w->mpf->pool, database) == 0)
				break;
		}
		if (w == NULL) {
			w = malloc(sizeof(mplist));
			w->next = mero_multiplex_funnel;
			w->mpf = multiplexInit(database);
			mero_multiplex_funnel = w;
			if (pthread_create(&w->mpf->tid,
					NULL, (void *(*)(void *))multiplexThread,
					(void *)w->mpf) < 0)
			{
				Mfprintf(stderr, "starting multiplex-funnel %s failed\n",
						database);
			}
		}
		multiplexAddClient(w->mpf, sock, fout, fdin, host);

		return(NO_ERR);
	}

	if (strcmp(lang, "resolve") == 0) {
		/* ensure the pattern ends with '/\*' such that we force a
		 * remote entry, including those for local databases, this
		 * way we will get a redirect back to merovingian for such
		 * database if it is proxied and hence not remotely
		 * available */
		size_t len = strlen(database);
		if (len > 2 &&
				database[len - 2] != '/' &&
				database[len - 1] != '*')
		{
			char *n = alloca(sizeof(char) * len + 2 + 1);
			snprintf(n, len + 2 + 1, "%s/*", database);
			database = n;
		}
	}

	if ((e = forkMserver(database, &top, 0)) != NO_ERR) {
		if (top == NULL) {
			mnstr_printf(fout, "!merovingian: no such database '%s', please create it first\n", database);
		} else {
			mnstr_printf(fout, "!merovingian: internal error while starting mserver, please refer to the logs\n");
		}
		mnstr_flush(fout);
		close_stream(fout);
		close_stream(fdin);
{'fraction_non_alphanumeric': 0.16336351548154246, 'fraction_numerical': 0.0120306464889508, 'mean_word_length': 3.6535061873895107, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 64, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7315288', 'n_tokens_mistral': 6693, 'n_tokens_neox': 5960, 'n_words': 1594}