/* * FILE:     mbus.c * AUTHOR:   Colin Perkins * MODIFIED: Orion Hodson *           Markus Germeier *  * Copyright (c) 1997-2000 University College London * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, is permitted provided that the following conditions  * are met: * 1. Redistributions of source code must retain the above copyright *    notice, this list of conditions and the following disclaimer. * 2. Redistributions in binary form must reproduce the above copyright *    notice, this list of conditions and the following disclaimer in the *    documentation and/or other materials provided with the distribution. * 3. All advertising materials mentioning features or use of this software *    must display the following acknowledgement: *      This product includes software developed by the Computer Science *      Department at University College London * 4. Neither the name of the University nor of the Department may be used *    to endorse or promote products derived from this software without *    specific prior written permission. * THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS ``AS IS'' AND * ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF * SUCH DAMAGE. */#include "config_unix.h"#include "config_win32.h"#include "debug.h"#include "memory.h"#include "net_udp.h"#include "hmac.h"#include "qfDES.h"#include "base64.h"#include "gettimeofday.h"#include "vsnprintf.h"#include "mbus.h"#include "mbus_config.h"#include "mbus_parser.h"#include "mbus_addr.h"#define MBUS_BUF_SIZE	  1500#define MBUS_ACK_BUF_SIZE 1500#define MBUS_MAX_ADDR	    10#define MBUS_MAX_QLEN	    50 /* Number of messages we can queue with mbus_qmsg() */#define MBUS_MAGIC	0x87654321#define MBUS_MSG_MAGIC	0x12345678struct mbus_msg {	struct mbus_msg	*next;	struct timeval	 send_time;	/* Time the message was sent, to trigger a retransmit */	struct timeval	 comp_time;	/* Time the message was composed, the timestamp in the packet header */	char		*dest;	int		 reliable;	int		 complete;	/* Indicates that we've finished adding cmds to this message */	int		 seqnum;	int		 retransmit_count;	int		 message_size;	int		 num_cmds;	char		*cmd_list[MBUS_MAX_QLEN];	char		*arg_list[MBUS_MAX_QLEN];	uint32_t	 idx_list[MBUS_MAX_QLEN];	uint32_t	 magic;		/* For debugging... */};struct mbus {	socket_udp	 	 *s;	char		 	 *addr;				/* Addresses we respond to. 					*/	int		 	  max_other_addr;	int		 	  num_other_addr;	char			**other_addr;			/* Addresses of other entities on the mbus. 			*/        struct timeval          **other_hello;                  /* Time of last mbus.hello we received from other entities      */	int		 	  seqnum;	struct mbus_msg	 	 *cmd_queue;			/* Queue of messages waiting to be sent */	struct mbus_msg	 	 *waiting_ack;			/* The last reliable message sent, if we have not yet got the ACK */	char		 	 *hashkey;	int		 	  hashkeylen;	char		 	 *encrkey;	int		 	  encrkeylen;	struct timeval	 	  last_heartbeat;		/* Last time we sent a heartbeat message */	struct mbus_config	 *cfg;	void (*cmd_handler)(char *src, char *cmd, char *arg, void *dat);	void (*err_handler)(int seqnum, int reason);	uint32_t		  magic;			/* For debugging...                                             */	uint32_t		  index;	uint32_t		  index_sent;};static void mbus_validate(struct mbus *m){#ifdef DEBUG	int	i;	ASSERT(m->num_other_addr <= m->max_other_addr);	ASSERT(m->num_other_addr >= 0);	for (i = 0; i < m->num_other_addr; i++) {		ASSERT(m->other_addr[i]  != NULL);		ASSERT(m->other_hello[i] != NULL);	}	for (i = m->num_other_addr + 1; i < m->max_other_addr; i++) {		ASSERT(m->other_addr[i]  == NULL);		ASSERT(m->other_hello[i] == NULL);	}#endif	ASSERT(m->magic == MBUS_MAGIC);	xmemchk();}static void mbus_msg_validate(struct mbus_msg *m){#ifdef DEBUG	int	i;	ASSERT((m->num_cmds < MBUS_MAX_QLEN) && (m->num_cmds >= 0));	for (i = 0; i < m->num_cmds; i++) {		ASSERT(m->cmd_list[i] != NULL);		ASSERT(m->arg_list[i] != NULL);		if (i > 0) {			ASSERT(m->idx_list[i] > m->idx_list[i-1]);		}	}	for (i = m->num_cmds + 1; i < MBUS_MAX_QLEN; i++) {		ASSERT(m->cmd_list[i] == NULL);		ASSERT(m->arg_list[i] == NULL);	}		ASSERT(m->dest != NULL);#endif	ASSERT(m->magic == MBUS_MSG_MAGIC);}static void store_other_addr(struct mbus *m, char *a){	/* This takes the address a and ensures it is stored in the   */	/* m->other_addr field of the mbus structure. The other_addr  */	/* field should probably be a hash table, but for now we hope */	/* that there are not too many entities on the mbus, so the   */	/* list is small.                                             */	int	i;	mbus_validate(m);	for (i = 0; i < m->num_other_addr; i++) {		if (mbus_addr_match(m->other_addr[i], a)) {			/* Already in the list... */			gettimeofday(m->other_hello[i],NULL);			return;		}	}	if (m->num_other_addr == m->max_other_addr) {		/* Expand the list... */		m->max_other_addr *= 2;		m->other_addr = (char **) xrealloc(m->other_addr, m->max_other_addr * sizeof(char *));		m->other_hello = (struct timeval **) xrealloc(m->other_hello, m->max_other_addr * sizeof(struct timeval *));	}	m->other_hello[m->num_other_addr]=(struct timeval *)xmalloc(sizeof(struct timeval));	gettimeofday(m->other_hello[m->num_other_addr],NULL);	m->other_addr[m->num_other_addr++] = xstrdup(a);}static void remove_other_addr(struct mbus *m, char *a){	/* Removes the address a from the m->other_addr field of the */	/* mbus structure.                                           */	int	i, j;	mbus_validate(m);	for (i = 0; i < m->num_other_addr; i++) {		if (mbus_addr_match(m->other_addr[i], a)) {			xfree(m->other_addr[i]);			xfree(m->other_hello[i]);			for (j = i+1; j < m->num_other_addr; j++) {				m->other_addr[j-1] = m->other_addr[j];				m->other_hello[j-1] = m->other_hello[j];			}			m->other_addr[m->num_other_addr  - 1] = NULL;			m->other_hello[m->num_other_addr - 1] = NULL;			m->num_other_addr--;		}	}}static void remove_inactiv_other_addr(struct mbus *m, struct timeval t, int interval){	/* Remove addresses we haven't heard from for about 5 * interval */	/* Count backwards so it is safe to remove entries               */	int i;    	mbus_validate(m);	for (i=m->num_other_addr-1; i>=0; i--){		if ((t.tv_sec-(m->other_hello[i]->tv_sec)) > 5 * interval) {			debug_msg("remove dead entity (%s)\n", m->other_addr[i]);			remove_other_addr(m, m->other_addr[i]);		}	}}int mbus_addr_valid(struct mbus *m, char *addr){	int	i;	mbus_validate(m);	for (i = 0; i < m->num_other_addr; i++) {		if (mbus_addr_match(m->other_addr[i], addr)) {			return TRUE;		}	}	return FALSE;}static int mbus_addr_unique(struct mbus *m, char *addr){	int     i, n = 0;	mbus_validate(m);	for (i = 0; i < m->num_other_addr; i++) {		if (mbus_addr_match(m->other_addr[i], addr)) {			n++;		}	}	return n==1;}/* The mb_* functions are used to build an mbus message up in the *//* mb_buffer, and to add authentication and encryption before the *//* message is sent.                                               */char	 mb_cryptbuf[MBUS_BUF_SIZE];char	*mb_buffer;char	*mb_bufpos;#define MBUS_AUTH_LEN 16static void mb_header(int seqnum, int ts, char reliable, const char *src, const char *dst, int ackseq){	xmemchk();	mb_buffer   = (char *) xmalloc(MBUS_BUF_SIZE + 1);	memset(mb_buffer,   0, MBUS_BUF_SIZE);	memset(mb_buffer, ' ', MBUS_AUTH_LEN);	mb_bufpos = mb_buffer + MBUS_AUTH_LEN;	sprintf(mb_bufpos, "\nmbus/1.0 %6d %9d %c (%s) %s ", seqnum, ts, reliable, src, dst);	mb_bufpos += 33 + strlen(src) + strlen(dst);	if (ackseq == -1) {		sprintf(mb_bufpos, "()\n");		mb_bufpos += 3;	} else {		sprintf(mb_bufpos, "(%6d)\n", ackseq);		mb_bufpos += 9;	}}static void mb_add_command(const char *cmnd, const char *args){	int offset = strlen(cmnd) + strlen(args) + 5;	ASSERT((mb_bufpos + offset - mb_buffer) < MBUS_BUF_SIZE);	sprintf(mb_bufpos, "%s (%s)\n", cmnd, args);	mb_bufpos += offset - 1; /* The -1 in offset means we're not NUL terminated - fix in mb_send */}static void mb_send(struct mbus *m){	char		digest[16];	int		len;	unsigned char	initVec[8] = {0,0,0,0,0,0,0,0}; 	mbus_validate(m);	*(mb_bufpos++) = '\0';	ASSERT((mb_bufpos - mb_buffer) < MBUS_BUF_SIZE);	ASSERT(strlen(mb_buffer) < MBUS_BUF_SIZE);	/* Pad to a multiple of 8 bytes, so the encryption can work... */	while (((mb_bufpos - mb_buffer) % 8) != 0) {		*(mb_bufpos++) = '\0';	}	len = mb_bufpos - mb_buffer;	ASSERT(len < MBUS_BUF_SIZE);	ASSERT(strlen(mb_buffer) < MBUS_BUF_SIZE);	xmemchk();	if (m->hashkey != NULL) {		/* Authenticate... */		hmac_md5(mb_buffer + MBUS_AUTH_LEN+1, strlen(mb_buffer) - (MBUS_AUTH_LEN+1), m->hashkey, m->hashkeylen, digest);		base64encode(digest, 12, mb_buffer, MBUS_AUTH_LEN);	}	xmemchk();	if (m->encrkey != NULL) {		/* Encrypt... */		memset(mb_cryptbuf, 0, MBUS_BUF_SIZE);		memcpy(mb_cryptbuf, mb_buffer, len);		ASSERT((len % 8) == 0);		ASSERT(len < MBUS_BUF_SIZE);		ASSERT(m->encrkeylen == 8);		xmemchk();		qfDES_CBC_e(m->encrkey, mb_cryptbuf, len, initVec);		xmemchk();		memcpy(mb_buffer, mb_cryptbuf, len);	}	xmemchk();	udp_send(m->s, mb_buffer, len);	xfree(mb_buffer);}static void resend(struct mbus *m, struct mbus_msg *curr) {	/* Don't need to check for buffer overflows: this was done in mbus_send() when */	/* this message was first transmitted. If it was okay then, it's okay now.     */	int	 i;	mbus_validate(m);	mb_header(curr->seqnum, curr->comp_time.tv_sec, (char)(curr->reliable?'R':'U'), m->addr, curr->dest, -1);	for (i = 0; i < curr->num_cmds; i++) {		mb_add_command(curr->cmd_list[i], curr->arg_list[i]);	}	mb_send(m);	curr->retransmit_count++;}void mbus_retransmit(struct mbus *m){	struct mbus_msg	*curr = m->waiting_ack;	struct timeval	time;	long		diff;	mbus_validate(m);	if (!mbus_waiting_ack(m)) {		return;	}	mbus_msg_validate(curr);	gettimeofday(&time, NULL);	/* diff is time in milliseconds that the message has been awaiting an ACK */	diff = ((time.tv_sec * 1000) + (time.tv_usec / 1000)) - ((curr->send_time.tv_sec * 1000) + (curr->send_time.tv_usec / 1000));	if (diff > 10000) {		debug_msg("Reliable mbus message failed!\n");		if (m->err_handler == NULL) {			abort();		}		m->err_handler(curr->seqnum, MBUS_MESSAGE_LOST);		/* if we don't delete this failed message, the error handler                   gets triggered every time we call mbus_retransmit */		while (m->waiting_ack->num_cmds > 0) {		    m->waiting_ack->num_cmds--;		    xfree(m->waiting_ack->cmd_list[m->waiting_ack->num_cmds]);		    xfree(m->waiting_ack->arg_list[m->waiting_ack->num_cmds]);		}		xfree(m->waiting_ack->dest);		xfree(m->waiting_ack);		m->waiting_ack = NULL;		return;	} 	/* Note: We only send one retransmission each time, to avoid	 * overflowing the receiver with a burst of requests...	 */	if ((diff > 750) && (curr->retransmit_count == 2)) {		resend(m, curr);		return;	} 	if ((diff > 500) && (curr->retransmit_count == 1)) {		resend(m, curr);		return;	} 	if ((diff > 250) && (curr->retransmit_count == 0)) {		resend(m, curr);		return;	}}void mbus_heartbeat(struct mbus *m, int interval){	struct timeval	curr_time;	char	*a = (char *) xmalloc(3);	sprintf(a, "()");	mbus_validate(m);	gettimeofday(&curr_time, NULL);	if (curr_time.tv_sec - m->last_heartbeat.tv_sec >= interval) {		mb_header(++m->seqnum, (int) curr_time.tv_sec, 'U', m->addr, "()", -1);		mb_add_command("mbus.hello", "");		mb_send(m);		m->last_heartbeat = curr_time;		/* Remove dead sources */		remove_inactiv_other_addr(m, curr_time, interval);	}	xfree(a);}int mbus_waiting_ack(struct mbus *m){	mbus_validate(m);	return m->waiting_ack != NULL;}int mbus_sent_all(struct mbus *m){	mbus_validate(m);	return (m->cmd_queue == NULL) && (m->waiting_ack == NULL);}struct mbus *mbus_init(void  (*cmd_handler)(char *src, char *cmd, char *arg, void *dat), 		       void  (*err_handler)(int seqnum, int reason),		       char  *addr){	struct mbus		*m;	struct mbus_key	 	 k;	struct mbus_parser	*mp;	int		 	 i;	char            	*net_addr, *tmp;	uint16_t         	 net_port;	int              	 net_scope;	m = (struct mbus *) xmalloc(sizeof(struct mbus));	if (m == NULL) {		debug_msg("Unable to allocate memory for mbus\n");		return NULL;	}	m->cfg = mbus_create_config();	mbus_lock_config_file(m->cfg);	net_addr = (char *) xmalloc(20);	mbus_get_net_addr(m->cfg, net_addr, &net_port, &net_scope);	m->s		  = udp_init(net_addr, net_port, net_port, net_scope);        if (m->s == NULL) {                debug_msg("Unable to initialize mbus address\n");                xfree(m);                return NULL;        }	m->seqnum         = 0;	m->cmd_handler    = cmd_handler;	m->err_handler	  = err_handler;	m->num_other_addr = 0;	m->max_other_addr = 10;	m->other_addr     = (char **) xmalloc(sizeof(char *) * 10);	m->other_hello    = (struct timeval **) xmalloc(sizeof(struct timeval *) * 10);	for (i = 0; i < 10; i++) {		m->other_addr[i]  = NULL;		m->other_hello[i] = NULL;	}	m->cmd_queue	  = NULL;	m->waiting_ack	  = NULL;	m->magic          = MBUS_MAGIC;	m->index          = 0;	m->index_sent     = 0;	mp = mbus_parse_init(xstrdup(addr));	if (!mbus_parse_lst(mp, &tmp)) {		debug_msg("Invalid mbus address\n");		abort();	}	m->addr = xstrdup(tmp);	mbus_parse_done(mp);	ASSERT(m->addr != NULL);	gettimeofday(&(m->last_heartbeat), NULL);	mbus_get_encrkey(m->cfg, &k);	m->encrkey    = k.key;	m->encrkeylen = k.key_len;	mbus_get_hashkey(m->cfg, &k);	m->hashkey    = k.key;	m->hashkeylen = k.key_len;	mbus_unlock_config_file(m->cfg);	xfree(net_addr);	return m;}void mbus_cmd_handler(struct mbus *m, void  (*cmd_handler)(char *src, char *cmd, char *arg, void *dat)){	mbus_validate(m);	m->cmd_handler = cmd_handler;}static void mbus_flush_msgs(struct mbus_msg **queue){        struct mbus_msg *curr, *next;        int i;	        curr = *queue;        while(curr) {                next = curr->next;                xfree(curr->dest);                for(i = 0; i < curr->num_cmds; i++) {                        xfree(curr->cmd_list[i]);                        xfree(curr->arg_list[i]);                }		xfree(curr);