Imported Upstream version 4.4.0upstream/4.4.0

1 files changed, 129 insertions, 63 deletions

diff --git a/tcpsrv.c b/tcpsrv.c
index b943439..1161949 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -1,8 +1,11 @@
 /* tcpsrv.c
  *
- * Common code for plain TCP based servers. This is currently being
+ * Common code for plain TCP syslog based servers. This is currently being
  * utilized by imtcp and imgssapi.
  *
+ * NOTE: this is *not* a generic TCP server, but one for syslog servers. For
+ *       generic stream servers, please use ./runtime/strmsrv.c!
+ *
  * There are actually two classes within the tcpserver code: one is
  * the tcpsrv itself, the other one is its sessions. This is a helper
  * class to tcpsrv.
@@ -17,7 +20,7 @@
  *
  * File begun on 2007-12-21 by RGerhards (extracted from syslogd.c)
  *
- * Copyright 2007, 2008 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007, 2008, 2009 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of rsyslog.
  *
@@ -66,6 +69,7 @@
 #include "netstrm.h"
 #include "nssel.h"
 #include "errmsg.h"
+#include "unicode-helper.h"
 
 MODULE_TYPE_LIB
 
@@ -85,43 +89,61 @@ DEFobjCurrIf(netstrm)
 DEFobjCurrIf(nssel)
 
 
-/* configure TCP listener settings. This is called during command
- * line parsing. The argument following -t is supplied as an argument.
- * The format of this argument is
- * "<port-to-use>, <nbr-of-sessions>"
- * Typically, there is no whitespace between port and session number.
- * (but it may be...).
- * NOTE: you can not use dbgprintf() in here - the dbgprintf() system is
- * not yet initilized when this function is called.
- * rgerhards, 2007-06-21
- * The port in cOptarg is handed over to us - the caller MUST NOT free it!
+/* add new listener port to listener port list
+ * rgerhards, 2009-05-21
+ */
+static inline rsRetVal
+addNewLstnPort(tcpsrv_t *pThis, uchar *pszPort)
+{
+	tcpLstnPortList_t *pEntry;
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, tcpsrv);
+
+	/* create entry */
+	CHKmalloc(pEntry = malloc(sizeof(tcpLstnPortList_t)));
+	pEntry->pszPort = pszPort;
+	pEntry->pSrv = pThis;
+	CHKmalloc(pEntry->pszInputName = ustrdup(pThis->pszInputName));
+	pEntry->lenInputName = ustrlen(pEntry->pszInputName);
+
+	/* and add to list */
+	pEntry->pNext = pThis->pLstnPorts;
+	pThis->pLstnPorts = pEntry;
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* configure TCP listener settings.
+ * Note: pszPort is handed over to us - the caller MUST NOT free it!
  * rgerhards, 2008-03-20
  */
-static void
-configureTCPListen(tcpsrv_t *pThis, char *cOptarg)
+static rsRetVal
+configureTCPListen(tcpsrv_t *pThis, uchar *pszPort)
 {
-	register int i;
-	register char *pArg = cOptarg;
+	int i;
+	uchar *pPort = pszPort;
+	DEFiRet;
 
-	assert(cOptarg != NULL);
+	assert(pszPort != NULL);
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
 
 	/* extract port */
 	i = 0;
-	while(isdigit((int) *pArg)) {
-		i = i * 10 + *pArg++ - '0';
+	while(isdigit((int) *pPort)) {
+		i = i * 10 + *pPort++ - '0';
 	}
 
-	if(pThis->TCPLstnPort != NULL) {
-		free(pThis->TCPLstnPort);
-		pThis->TCPLstnPort = NULL;
-	}
-		
-	if( i >= 0 && i <= 65535) {
-		pThis->TCPLstnPort = cOptarg;
+	if(i >= 0 && i <= 65535) {
+		CHKiRet(addNewLstnPort(pThis, pszPort));
 	} else {
-		errmsg.LogError(0, NO_ERRCODE, "Invalid TCP listen port %s - changed to 514.\n", cOptarg);
+		errmsg.LogError(0, NO_ERRCODE, "Invalid TCP listen port %s - ignored.\n", pszPort);
 	}
+
+finalize_it:
+	RETiRet;
 }
 
 
@@ -202,6 +224,8 @@ TCPSessGetNxtSess(tcpsrv_t *pThis, int iCurr)
 static void deinit_tcp_listener(tcpsrv_t *pThis)
 {
 	int i;
+	tcpLstnPortList_t *pEntry;
+	tcpLstnPortList_t *pDel;
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
 
@@ -219,8 +243,15 @@ static void deinit_tcp_listener(tcpsrv_t *pThis)
 		pThis->pSessions = NULL; /* just to make sure... */
 	}
 
-	if(pThis->TCPLstnPort != NULL)
-		free(pThis->TCPLstnPort);
+	/* free list of tcp listen ports */
+	pEntry = pThis->pLstnPorts;
+	while(pEntry != NULL) {
+		free(pEntry->pszPort);
+		free(pEntry->pszInputName);
+		pDel = pEntry;
+		pEntry = pEntry->pNext;
+		free(pDel);
+	}
 
 	/* finally close our listen streams */
 	for(i = 0 ; i < pThis->iLstnMax ; ++i) {
@@ -235,7 +266,8 @@ static void deinit_tcp_listener(tcpsrv_t *pThis)
 static rsRetVal
 addTcpLstn(void *pUsr, netstrm_t *pLstn)
 {
-	tcpsrv_t *pThis = (tcpsrv_t*) pUsr;
+	tcpLstnPortList_t *pPortList = (tcpLstnPortList_t *) pUsr;
+	tcpsrv_t *pThis = pPortList->pSrv;
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
@@ -245,6 +277,7 @@ addTcpLstn(void *pUsr, netstrm_t *pLstn)
 		ABORT_FINALIZE(RS_RET_MAX_LSTN_REACHED);
 
 	pThis->ppLstn[pThis->iLstnMax] = pLstn;
+	pThis->ppLstnPort[pThis->iLstnMax] = pPortList;
 	++pThis->iLstnMax;
 
 finalize_it:
@@ -252,19 +285,20 @@ finalize_it:
 }
 
 
-/* Initialize TCP sockets (for listener) and listens on them */
-static rsRetVal
-create_tcp_socket(tcpsrv_t *pThis)
+/* Initialize TCP listener socket for a single port
+ * rgerhards, 2009-05-21
+ */
+static inline rsRetVal
+initTCPListener(tcpsrv_t *pThis, tcpLstnPortList_t *pPortEntry)
 {
 	DEFiRet;
 	uchar *TCPLstnPort;
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
+	assert(pPortEntry != NULL);
 
-	if(!strcmp((char*)pThis->TCPLstnPort, "0"))
-		TCPLstnPort = (uchar*)"514";
-		// TODO: we need to enable the caller to set a port (based on who is
-		// using this, 514 may be totally unsuitable... --- rgerhards, 2008-04-22
+	if(!ustrcmp(pPortEntry->pszPort, UCHAR_CONSTANT("0")))
+		TCPLstnPort = UCHAR_CONSTANT("514");
 		/* use default - we can not do service db update, because there is
 		 * no IANA-assignment for syslog/tcp. In the long term, we might
 		 * re-use RFC 3195 port of 601, but that would probably break to
@@ -272,10 +306,31 @@ create_tcp_socket(tcpsrv_t *pThis)
 		 * rgerhards, 2007-06-28
 		 */
 	else
-		TCPLstnPort = (uchar*)pThis->TCPLstnPort;
+		TCPLstnPort = pPortEntry->pszPort;
 
 	/* TODO: add capability to specify local listen address! */
-	CHKiRet(netstrm.LstnInit(pThis->pNS, (void*)pThis, addTcpLstn, TCPLstnPort, NULL, pThis->iSessMax));
+	CHKiRet(netstrm.LstnInit(pThis->pNS, (void*)pPortEntry, addTcpLstn, TCPLstnPort, NULL, pThis->iSessMax));
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* Initialize TCP sockets (for listener) and listens on them */
+static rsRetVal
+create_tcp_socket(tcpsrv_t *pThis)
+{
+	tcpLstnPortList_t *pEntry;
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, tcpsrv);
+
+	/* init all configured ports */
+	pEntry = pThis->pLstnPorts;
+	while(pEntry != NULL) {
+		CHKiRet(initTCPListener(pThis, pEntry));
+		pEntry = pEntry->pNext;
+	}
 
 	/* OK, we had success. Now it is also time to
 	 * initialize our connections
@@ -297,7 +352,7 @@ finalize_it:
 /* Accept new TCP connection; make entry in session table. If there
  * is no more space left in the connection table, the new TCP
  * connection is immediately dropped.
- * ppSess has a pointer to the newly created session, if it succeds.
+ * ppSess has a pointer to the newly created session, if it succeeds.
  * If it does not succeed, no session is created and ppSess is
  * undefined. If the user has provided an OnSessAccept Callback,
  * this one is executed immediately after creation of the 
@@ -305,7 +360,7 @@ finalize_it:
  * rgerhards, 2008-03-02
  */
 static rsRetVal
-SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, netstrm_t *pStrm)
+SessAccept(tcpsrv_t *pThis, tcpLstnPortList_t *pLstnInfo, tcps_sess_t **ppSess, netstrm_t *pStrm)
 {
 	DEFiRet;
 	tcps_sess_t *pSess = NULL;
@@ -316,6 +371,7 @@ SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, netstrm_t *pStrm)
 	uchar *fromHostIP = NULL;
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
+	assert(pLstnInfo != NULL);
 
 	CHKiRet(netstrm.AcceptConnReq(pStrm, &pNewStrm));
 
@@ -325,13 +381,15 @@ SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, netstrm_t *pStrm)
 		errno = 0;
 		errmsg.LogError(0, RS_RET_MAX_SESS_REACHED, "too many tcp sessions - dropping incoming request");
 		ABORT_FINALIZE(RS_RET_MAX_SESS_REACHED);
-	} else {
-		/* we found a free spot and can construct our session object */
-		CHKiRet(tcps_sess.Construct(&pSess));
-		CHKiRet(tcps_sess.SetTcpsrv(pSess, pThis));
 	}
 
-	/* OK, we have a "good" index... */
+	/* we found a free spot and can construct our session object */
+	CHKiRet(tcps_sess.Construct(&pSess));
+	CHKiRet(tcps_sess.SetTcpsrv(pSess, pThis));
+	CHKiRet(tcps_sess.SetLstnInfo(pSess, pLstnInfo));
+	if(pThis->OnMsgReceive != NULL)
+		CHKiRet(tcps_sess.SetOnMsgReceive(pSess, pThis->OnMsgReceive));
+
 	/* get the host name */
 	CHKiRet(netstrm.GetRemoteHName(pNewStrm, &fromHostFQDN));
 	CHKiRet(netstrm.GetRemoteIP(pNewStrm, &fromHostIP));
@@ -376,12 +434,10 @@ finalize_it:
 	if(iRet != RS_RET_OK) {
 		if(pSess != NULL)
 			tcps_sess.Destruct(&pSess);
-		if(fromHostFQDN != NULL)
-			free(fromHostFQDN);
-		if(fromHostIP != NULL)
-			free(fromHostIP);
 		if(pNewStrm != NULL)
 			netstrm.Destruct(&pNewStrm);
+		free(fromHostFQDN);
+		free(fromHostIP);
 	}
 
 	RETiRet;
@@ -445,7 +501,7 @@ Run(tcpsrv_t *pThis)
 			CHKiRet(nssel.IsReady(pSel, pThis->ppLstn[i], NSDSEL_RD, &bIsReady, &nfds));
 			if(bIsReady) {
 				dbgprintf("New connect on NSD %p.\n", pThis->ppLstn[i]);
-				SessAccept(pThis, &pNewSess, pThis->ppLstn[i]);
+				SessAccept(pThis, pThis->ppLstnPort[i], &pNewSess, pThis->ppLstn[i]);
 				--nfds; /* indicate we have processed one */
 			}
 		}
@@ -455,7 +511,7 @@ Run(tcpsrv_t *pThis)
 		while(nfds && iTCPSess != -1) {
 			CHKiRet(nssel.IsReady(pSel, pThis->pSessions[iTCPSess]->pStrm, NSDSEL_RD, &bIsReady, &nfds));
 			if(bIsReady) {
-				char buf[8*1024]; /* reception buffer - may hold a partial or multiple messages */
+				char buf[128*1024]; /* reception buffer - may hold a partial or multiple messages */
 				dbgprintf("netstream %p with new data\n", pThis->pSessions[iTCPSess]->pStrm);
 
 				/* Receive message */
@@ -514,6 +570,7 @@ finalize_it: /* this is a very special case - this time only we do not exit the
 BEGINobjConstruct(tcpsrv) /* be sure to specify the object type also in END macro! */
 	pThis->iSessMax = TCPSESS_MAX_DEFAULT; /* TODO: useful default ;) */
 	pThis->addtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER;
+	pThis->OnMsgReceive = NULL;
 ENDobjConstruct(tcpsrv)
 
 
@@ -536,6 +593,7 @@ tcpsrvConstructFinalize(tcpsrv_t *pThis)
 
 	/* set up listeners */
 	CHKmalloc(pThis->ppLstn = calloc(TCPLSTN_MAX_DEFAULT, sizeof(netstrm_t*)));
+	CHKmalloc(pThis->ppLstnPort = calloc(TCPLSTN_MAX_DEFAULT, sizeof(tcpLstnPortList_t*)));
 	iRet = pThis->OpenLstnSocks(pThis);
 
 finalize_it:
@@ -557,12 +615,10 @@ CODESTARTobjDestruct(tcpsrv)
 
 	if(pThis->pNS != NULL)
 		netstrms.Destruct(&pThis->pNS);
-	if(pThis->pszDrvrAuthMode != NULL)
-		free(pThis->pszDrvrAuthMode);
-	if(pThis->ppLstn != NULL)
-		free(pThis->ppLstn);
-	if(pThis->pszInputName != NULL)
-		free(pThis->pszInputName);
+	free(pThis->pszDrvrAuthMode);
+	free(pThis->ppLstn);
+	free(pThis->ppLstnPort);
+	free(pThis->pszInputName);
 ENDobjDestruct(tcpsrv)
 
 
@@ -660,6 +716,16 @@ SetUsrP(tcpsrv_t *pThis, void *pUsr)
 	RETiRet;
 }
 
+static rsRetVal
+SetOnMsgReceive(tcpsrv_t *pThis, rsRetVal (*OnMsgReceive)(tcps_sess_t*, uchar*, int))
+{
+	DEFiRet;
+	assert(OnMsgReceive != NULL);
+	pThis->OnMsgReceive = OnMsgReceive;
+	RETiRet;
+}
+
+
 
 /* Set additional framing to use (if any) -- rgerhards, 2008-12-10 */
 static rsRetVal
@@ -682,9 +748,8 @@ SetInputName(tcpsrv_t *pThis, uchar *name)
 	if(name == NULL)
 		pszName = NULL;
 	else
-		CHKmalloc(pszName = (uchar*)strdup((char*)name));
-	if(pThis->pszInputName != NULL)
-		free(pThis->pszInputName);
+		CHKmalloc(pszName = ustrdup(name));
+	free(pThis->pszInputName);
 	pThis->pszInputName = pszName;
 finalize_it:
 	RETiRet;
@@ -713,7 +778,7 @@ SetDrvrAuthMode(tcpsrv_t *pThis, uchar *mode)
 {
 	DEFiRet;
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
-	CHKmalloc(pThis->pszDrvrAuthMode = (uchar*)strdup((char*)mode));
+	CHKmalloc(pThis->pszDrvrAuthMode = ustrdup(mode));
 finalize_it:
 	RETiRet;
 }
@@ -768,7 +833,7 @@ CODESTARTobjQueryInterface(tcpsrv)
 	pIf->ConstructFinalize = tcpsrvConstructFinalize;
 	pIf->Destruct = tcpsrvDestruct;
 
-	pIf->SessAccept = SessAccept;
+	//pIf->SessAccept = SessAccept;
 	pIf->configureTCPListen = configureTCPListen;
 	pIf->create_tcp_socket = create_tcp_socket;
 	pIf->Run = Run;
@@ -790,6 +855,7 @@ CODESTARTobjQueryInterface(tcpsrv)
 	pIf->SetCBOnDestruct = SetCBOnDestruct;
 	pIf->SetCBOnRegularClose = SetCBOnRegularClose;
 	pIf->SetCBOnErrClose = SetCBOnErrClose;
+	pIf->SetOnMsgReceive = SetOnMsgReceive;
 
 finalize_it:
 ENDobjQueryInterface(tcpsrv)