mbus.c

jpeg and mpeg 编解码技术源代码

/*
 * 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	0x12345678

struct 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 16

static 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;