bugfix: TCP (and GSSAPI) octet-counted frame did not work correctly in all

    situations. If the header was split across two packet reads, it was
    invalidly processed, causing loss or modification of messages.

7 files changed, 101 insertions, 119 deletions

diff --git a/ChangeLog b/ChangeLog
index 53a2f716..648c6cca 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -5,6 +5,12 @@ Version 3.12.3 (rgerhards), 2008-03-??
 - bugfix: $ModDir should not be reset on $ResetConfig - this can cause a lot
   of confusion and there is no real good reason to do so. Also conflicts with
   the new -M option and environment setting.
+- bugfix: TCP and GSSAPI framing mode variable was uninitialized, leading to
+  wrong framing (caused, among others, interop problems)
+- bugfix: TCP (and GSSAPI) octet-counted frame did not work correctly in all
+  situations. If the header was split across two packet reads, it was invalidly
+  processed, causing loss or modification of messages.
+- bugfix: memory leak in imfile
 - some more internal cleanup ;)
 ---------------------------------------------------------------------------
 Version 3.12.2 (rgerhards), 2008-03-13
diff --git a/plugins/imgssapi/imgssapi.c b/plugins/imgssapi/imgssapi.c
index 8d406a7e..74d5d5c5 100644
--- a/plugins/imgssapi/imgssapi.c
+++ b/plugins/imgssapi/imgssapi.c
@@ -480,7 +480,7 @@ OnSessAcceptGSS(tcpsrv_t *pThis, tcps_sess_t *pSess)
 				    (GSS_ROUTINE_ERROR(maj_stat) == GSS_S_DEFECTIVE_TOKEN)) {
 					dbgprintf("GSS-API Reverting to plain TCP\n");
 					dbgprintf("tcp session socket with new data: #%d\n", fdSess);
-					if(tcps_sess.DataRcvd(pSess, buf, ret) == 0) {
+					if(tcps_sess.DataRcvd(pSess, buf, ret) != RS_RET_OK) {
 						errmsg.LogError(NO_ERRCODE, "Tearing down TCP Session %p - see "
 							    "previous messages for reason(s)\n", pSess);
 						ABORT_FINALIZE(RS_RET_ERR); // TODO: define good error codes
diff --git a/plugins/omrelp/omrelp.c b/plugins/omrelp/omrelp.c
index 22482c23..9dd75a84 100644
--- a/plugins/omrelp/omrelp.c
+++ b/plugins/omrelp/omrelp.c
@@ -152,6 +152,8 @@ static rsRetVal TCPSendFrame(void *pvData, char *msg, size_t len)
 
 	lenSend = send(pData->sock, msg, len, 0);
 	dbgprintf("TCP sent %ld bytes, requested %ld\n", (long) lenSend, (long) len);
+if(lenSend > 2000)
+	dbgprintf("TCP (msgoverflow) sent %ld bytes, requested %ld\n", (long) lenSend, (long) len);
 
 	if(lenSend == -1) {
 		/* we have an error case - check what we can live with */
@@ -500,10 +502,10 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
 	CHKiRet(tcpclt.SetSendInit(pData->pTCPClt, TCPSendInit));
 	CHKiRet(tcpclt.SetSendFrame(pData->pTCPClt, TCPSendFrame));
 	CHKiRet(tcpclt.SetSendPrepRetry(pData->pTCPClt, TCPSendPrepRetry));
-	CHKiRet(tcpclt.SetFraming(pData->pTCPClt, tcp_framing));
+	CHKiRet(tcpclt.SetFraming(pData->pTCPClt, TCP_FRAMING_OCTET_COUNTING));
 
 	/* TODO: do we need to call freeInstance if we failed - this is a general question for
-	 * all output modules. I'll address it lates as the interface evolves. rgerhards, 2007-07-25
+	 * all output modules. I'll address it later as the interface evolves. rgerhards, 2007-07-25
 	 */
 CODE_STD_FINALIZERparseSelectorAct
 ENDparseSelectorAct
diff --git a/queue.c b/queue.c
index 1eded530..acf21e32 100644
--- a/queue.c
+++ b/queue.c
@@ -934,7 +934,7 @@ queueUngetObj(queue_t *pThis, obj_t *pUsr, int bLockMutex)
 	DEFVARS_mutexProtection;
 
 	ISOBJ_TYPE_assert(pThis, queue);
-	ISOBJ_assert(pUsr); /* TODO: we aborted right at this place at least twice -- race? 2008-02-28, -03-10
+	ISOBJ_assert(pUsr); /* TODO: we aborted right at this place at least 3 times -- race? 2008-02-28, -03-10, -03-15
 			       The second time I noticed it the queue was in destruction with NO worker threads
 			       running. The pUsr ptr was totally off and provided no clue what it may be pointing
 			       at (except that it looked like the static data pool). Both times, the abort happend
@@ -2031,7 +2031,6 @@ queueEnqObj(queue_t *pThis, flowControl_t flowCtlType, void *pUsr)
 	 * result, that non-blockable sources like UDP syslog receive priority in the system.
 	 * It's a side effect, but a good one ;) -- rgerhards, 2008-03-14
 	 */
-dbgprintf("enqueueMsg: flowctl mode: %d, queue size %d, FullDlyMrk %d\n", flowCtlType, pThis->iQueueSize, pThis->iFullDlyMrk);
 	if(flowCtlType == eFLOWCTL_FULL_DELAY) {
 		while(pThis->iQueueSize >= pThis->iFullDlyMrk) {
 			dbgoprint((obj_t*) pThis, "enqueueMsg: FullDelay mark reached for full delayble message - blocking.\n");
diff --git a/tcps_sess.c b/tcps_sess.c
index c8ac3a37..2a6b300a 100644
--- a/tcps_sess.c
+++ b/tcps_sess.c
@@ -237,6 +237,86 @@ Close(tcps_sess_t *pThis)
 }
 
 
+/* process the data received. As TCP is stream based, we need to process the
+ * data inside a state machine. The actual data received is passed in byte-by-byte
+ * from DataRcvd, and this function here compiles messages from them and submits
+ * the end result to the queue. Introducing this function fixes a long-term bug ;)
+ * rgerhards, 2008-03-14
+ */
+static rsRetVal
+processDataRcvd(tcps_sess_t *pThis, char c)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, tcps_sess);
+
+dbgprintf("processDataRcvd char %c, state %d\n", c, pThis->inputState);
+	if(pThis->inputState == eAtStrtFram) {
+		if(isdigit((int) c)) {
+			pThis->inputState = eInOctetCnt;
+			pThis->iOctetsRemain = 0;
+			pThis->eFraming = TCP_FRAMING_OCTET_COUNTING;
+		} else {
+			pThis->inputState = eInMsg;
+			pThis->eFraming = TCP_FRAMING_OCTET_STUFFING;
+		}
+	}
+
+	if(pThis->inputState == eInOctetCnt) {
+		if(isdigit(c)) {
+			pThis->iOctetsRemain = pThis->iOctetsRemain * 10 + c - '0';
+		} else { /* done with the octet count, so this must be the SP terminator */
+			dbgprintf("TCP Message with octet-counter, size %d.\n", pThis->iOctetsRemain);
+			if(c != ' ') {
+				errmsg.LogError(NO_ERRCODE, "Framing Error in received TCP message: "
+					    "delimiter is not SP but has ASCII value %d.\n", c);
+			}
+			if(pThis->iOctetsRemain < 1) {
+				/* TODO: handle the case where the octet count is 0! */
+				dbgprintf("Framing Error: invalid octet count\n");
+				errmsg.LogError(NO_ERRCODE, "Framing Error in received TCP message: "
+					    "invalid octet count %d.\n", pThis->iOctetsRemain);
+			}
+			pThis->inputState = eInMsg;
+		}
+	} else {
+		assert(pThis->inputState == eInMsg);
+		if(pThis->iMsg >= MAXLINE) {
+			/* emergency, we now need to flush, no matter if we are at end of message or not... */
+			dbgprintf("error: message received is larger than MAXLINE, we split it\n");
+			parseAndSubmitMessage(pThis->fromHost, pThis->msg, pThis->iMsg, MSG_PARSE_HOSTNAME, NOFLAG);
+			pThis->iMsg = 0;
+			/* we might think if it is better to ignore the rest of the
+			 * message than to treat it as a new one. Maybe this is a good
+			 * candidate for a configuration parameter...
+			 * rgerhards, 2006-12-04
+			 */
+		}
+
+		if(c == '\n' && pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delemiter? */
+			parseAndSubmitMessage(pThis->fromHost, pThis->msg, pThis->iMsg, MSG_PARSE_HOSTNAME, NOFLAG);
+			pThis->iMsg = 0;
+			pThis->inputState = eAtStrtFram;
+		} else {
+			/* IMPORTANT: here we copy the actual frame content to the message! */
+			*(pThis->msg + pThis->iMsg++) = c;
+		}
+
+		if(pThis->eFraming == TCP_FRAMING_OCTET_COUNTING) {
+			/* do we need to find end-of-frame via octet counting? */
+			pThis->iOctetsRemain--;
+			if(pThis->iOctetsRemain < 1) {
+				/* we have end of frame! */
+				parseAndSubmitMessage(pThis->fromHost, pThis->msg, pThis->iMsg, MSG_PARSE_HOSTNAME, NOFLAG);
+				pThis->iMsg = 0;
+				pThis->inputState = eAtStrtFram;
+			}
+		}
+	}
+
+	RETiRet;
+}
+
+
 /* Processes the data received via a TCP session. If there
  * is no other way to handle it, data is discarded.
  * Input parameter data is the data received, iLen is its
@@ -244,16 +324,6 @@ Close(tcps_sess_t *pThis)
  * is errors must be handled by caller!). iTCPSess must be
  * the index of the TCP session that received the data.
  * rgerhards 2005-07-04
- * Changed this functions interface. We now return a status of
- * what shall happen with the session. This is information for
- * the caller. If 1 is returned, the session should remain open
- * and additional data be accepted. If we return 0, the TCP
- * session is to be closed by the caller. This functionality is
- * needed in order to support framing errors, from which there
- * is no recovery possible other than session termination and
- * re-establishment. The need for this functionality thus is
- * primarily rooted in support for -transport-tls I-D framing.
- * rgerhards, 2006-12-07
  * And another change while generalizing. We now return either
  * RS_RET_OK, which means the session should be kept open
  * or anything else, which means it must be closed.
@@ -263,7 +333,6 @@ static rsRetVal
 DataRcvd(tcps_sess_t *pThis, char *pData, size_t iLen)
 {
 	DEFiRet;
-	register int iMsg;
 	char *pMsg;
 	char *pEnd;
 
@@ -284,112 +353,14 @@ DataRcvd(tcps_sess_t *pThis, char *pData, size_t iLen)
 	  * - printline() the buffer
 	  * - continue with copying
 	  */
-	iMsg = pThis->iMsg; /* copy for speed */
 	pMsg = pThis->msg; /* just a shortcut */
 	pEnd = pData + iLen; /* this is one off, which is intensional */
 
 	while(pData < pEnd) {
-		/* Check if we are at a new frame */
-		if(pThis->bAtStrtOfFram) {
-			/* we need to look at the message and detect
-			 * the framing mode used
-			 *//*
-			 * Contrary to -transport-tls, we accept leading zeros in the message
-			 * length. We do this in the spirit of "Be liberal in what you accept,
-			 * and conservative in what you send". We expect that including leading
-			 * zeros could be a common coding error.
-			 * rgerhards, 2006-12-07
-			 * The chairs of the IETF syslog-sec WG have announced that it is
-			 * consensus to do the octet count on the SYSLOG-MSG part only. I am
-			 * now changing the code to reflect this. Hopefully, it will not change
-			 * once again (there can no compatibility layer programmed for this).
-			 * To be on the save side, I just comment the code out. I mark these
-			 * comments with "IETF20061218".
-			 * rgerhards, 2006-12-19
-			 */
-			if(isdigit((int) *pData)) {
-				int iCnt;	/* the frame count specified */
-				pThis->eFraming = TCP_FRAMING_OCTET_COUNTING;
-				/* in this mode, we have OCTET-COUNT SP MSG - so we now need
-				 * to extract the OCTET-COUNT and the SP and then extract
-				 * the msg.
-				 */
-				iCnt = 0;
-				/* IETF20061218 int iNbrOctets = 0; / * number of octets already consumed */
-				while(isdigit((int) *pData)) {
-					iCnt = iCnt * 10 + *pData - '0';
-					/* IETF20061218 ++iNbrOctets; */
-					++pData;
-				}
-				dbgprintf("TCP Message with octet-counter, size %d.\n", iCnt);
-				if(*pData == ' ') {
-					++pData;	/* skip over SP */
-					/* IETF20061218 ++iNbrOctets; */
-				} else {
-					/* TODO: handle "invalid frame" case */
-					errmsg.LogError(NO_ERRCODE, "Framing Error in received TCP message: "
-					            "delimiter is not SP but has ASCII value %d.\n",
-						    *pData);
-					return(0); /* unconditional error exit */
-				}
-				/* IETF20061218 pThis->iOctetsRemain = iCnt - iNbrOctets; */
-				pThis->iOctetsRemain = iCnt;
-				if(pThis->iOctetsRemain < 1) {
-					/* TODO: handle the case where the octet count is 0 or negative! */
-					dbgprintf("Framing Error: invalid octet count\n");
-					errmsg.LogError(NO_ERRCODE, "Framing Error in received TCP message: "
-					            "invalid octet count %d.\n",
-				 		    pThis->iOctetsRemain);
-					return(0); /* unconditional error exit */
-				}
-			} else {
-				pThis->eFraming = TCP_FRAMING_OCTET_STUFFING;
-				/* No need to do anything else here in this case */
-			}
-			pThis->bAtStrtOfFram = 0; /* done frame header */
-		}
-	
-		/* now copy message until end of record */
-
-		if(iMsg >= MAXLINE) {
-			/* emergency, we now need to flush, no matter if
-			 * we are at end of message or not...
-			 */
-			parseAndSubmitMessage(pThis->fromHost, pMsg, iMsg, MSG_PARSE_HOSTNAME, NOFLAG);
-			iMsg = 0;
-			/* we might think if it is better to ignore the rest of the
-		 	 * message than to treat it as a new one. Maybe this is a good
-			 * candidate for a configuration parameter...
-			 * rgerhards, 2006-12-04
-			 */
-		}
-
-		if(*pData == '\n' &&
-		   pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delemiter? */
-			parseAndSubmitMessage(pThis->fromHost, pMsg, iMsg, MSG_PARSE_HOSTNAME, NOFLAG);
-			iMsg = 0;
-			pThis->bAtStrtOfFram = 1;
-			++pData;
-		} else {
-			/* IMPORTANT: here we copy the actual frame content to the message! */
-			*(pMsg + iMsg++) = *pData++;
-		}
-
-		if(pThis->eFraming == TCP_FRAMING_OCTET_COUNTING) {
-			/* do we need to find end-of-frame via octet counting? */
-			pThis->iOctetsRemain--;
-			if(pThis->iOctetsRemain < 1) {
-				/* we have end of frame! */
-				parseAndSubmitMessage(pThis->fromHost, pMsg, iMsg, MSG_PARSE_HOSTNAME, NOFLAG);
-				iMsg = 0;
-				pThis->bAtStrtOfFram = 1;
-			}
-		}
+		CHKiRet(processDataRcvd(pThis, *pData++));
 	}
 
-	pThis->iMsg = iMsg; /* persist value */
-
-	return(1);	/* successful return */
+finalize_it:
 	RETiRet;
 }
 
diff --git a/tcps_sess.h b/tcps_sess.h
index 2bdee80b..0433fdfb 100644
--- a/tcps_sess.h
+++ b/tcps_sess.h
@@ -41,6 +41,11 @@ typedef struct tcps_sess_s {
 	int sock;
 	int iMsg; /* index of next char to store in msg */
 	int bAtStrtOfFram;	/* are we at the very beginning of a new frame? */
+	enum {
+		eAtStrtFram,
+		eInOctetCnt,
+		eInMsg
+	} inputState;		/* our current state */
 	int iOctetsRemain;	/* Number of Octets remaining in message */
 	TCPFRAMINGMODE eFraming;
 	char msg[MAXLINE+1];
diff --git a/tcpsrv.c b/tcpsrv.c
index f3fad684..4553656e 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -612,13 +612,12 @@ Run(tcpsrv_t *pThis)
 					tcps_sess.Destruct(&pThis->pSessions[iTCPSess]);
 				} else {
 					/* valid data received, process it! */
-					if(tcps_sess.DataRcvd(pThis->pSessions[iTCPSess], buf, state) == 0) {
+					if(tcps_sess.DataRcvd(pThis->pSessions[iTCPSess], buf, state) != RS_RET_OK) {
 						/* in this case, something went awfully wrong.
 						 * We are instructed to terminate the session.
 						 */
 						errmsg.LogError(NO_ERRCODE, "Tearing down TCP Session %d - see "
-							    "previous messages for reason(s)\n",
-							    iTCPSess);
+							    "previous messages for reason(s)\n", iTCPSess);
 						pThis->pOnErrClose(pThis->pSessions[iTCPSess]);
 						tcps_sess.Destruct(&pThis->pSessions[iTCPSess]);
 					}