objects for receive-side socket abstraction specified

The objects for receiver-side socket abstraction have now be
specified. The project as whole does not yet compile and
definitely not run, but I'd like to commit some milestones along
this way.

1 files changed, 37 insertions, 198 deletions

diff --git a/tcpsrv.c b/tcpsrv.c
index 96048e31..a276bf19 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -63,6 +63,7 @@
 #include "tcpsrv.h"
 #include "obj.h"
 #include "glbl.h"
+#include "netstrm.h"
 #include "errmsg.h"
 
 MODULE_TYPE_LIB
@@ -77,30 +78,7 @@ DEFobjCurrIf(glbl)
 DEFobjCurrIf(tcps_sess)
 DEFobjCurrIf(errmsg)
 DEFobjCurrIf(net)
-
-
-
-/* code to free all sockets within a socket table.
- * A socket table is a descriptor table where the zero
- * element has the count of elements. This is used for
- * listening sockets. The socket table itself is also
- * freed.
- * A POINTER to this structure must be provided, thus
- * double indirection!
- * rgerhards, 2007-06-28
- */
-static void freeAllSockets(int **socks)
-{
-	assert(socks != NULL);
-	assert(*socks != NULL);
-	while(**socks) {
-		dbgprintf("Closing socket %d.\n", (*socks)[**socks]);
-		close((*socks)[**socks]);
-		(**socks)--;
-	}
-	free(*socks);
-	*socks = NULL;
-}
+DEFobjCurrIf(netstrm)
 
 
 /* configure TCP listener settings. This is called during command
@@ -236,33 +214,24 @@ static void deinit_tcp_listener(tcpsrv_t *pThis)
 	if(pThis->TCPLstnPort != NULL)
 		free(pThis->TCPLstnPort);
 
-	/* finally close the listen sockets themselfs */
-	freeAllSockets(&pThis->pSocksLstn);
+	/* finally close our listen stream */
+	netstrm.Destruct(&pThis->pLstn);
 }
 
 
-/* Initialize TCP sockets (for listener)
- * This function returns either NULL (which means it failed) or 
- * a pointer to an array of file descriptiors. If the pointer is
- * returned, the zeroest element [0] contains the count of valid
- * descriptors. The descriptors themself follow in range
- * [1] ... [num-descriptors]. It is guaranteed that each of these
- * descriptors is valid, at least when this function returns.
- * Please note that technically the array may be larger than the number
- * of valid pointers stored in it. The memory overhead is minimal, so
- * we do not bother to re-allocate an array of the exact size. Logically,
- * the array still contains the exactly correct number of descriptors.
- */
-static int *create_tcp_socket(tcpsrv_t *pThis)
+/* Initialize TCP sockets (for listener) and listens on them */
+static rsRetVal
+create_tcp_socket(tcpsrv_t *pThis)
 {
-        struct addrinfo hints, *res, *r;
-        int error, maxs, *s, *socks, on = 1;
-	char *TCPLstnPort;
+	DEFiRet;
+	uchar *TCPLstnPort;
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
 
-	if(!strcmp(pThis->TCPLstnPort, "0"))
-		TCPLstnPort = "514";
+	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
 		/* 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
@@ -270,136 +239,13 @@ static int *create_tcp_socket(tcpsrv_t *pThis)
 		 * rgerhards, 2007-06-28
 		 */
 	else
-		TCPLstnPort = pThis->TCPLstnPort;
-	dbgprintf("creating tcp socket on port %s\n", TCPLstnPort);
-        memset(&hints, 0, sizeof(hints));
-        hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV;
-        hints.ai_family = glbl.GetDefPFFamily();
-        hints.ai_socktype = SOCK_STREAM;
-
-        error = getaddrinfo(NULL, TCPLstnPort, &hints, &res);
-        if(error) {
-               errmsg.LogError(NO_ERRCODE, "%s", gai_strerror(error));
-	       return NULL;
-	}
-
-        /* Count max number of sockets we may open */
-        for (maxs = 0, r = res; r != NULL ; r = r->ai_next, maxs++)
-		/* EMPTY */;
-        socks = malloc((maxs+1) * sizeof(int));
-        if (socks == NULL) {
-               errmsg.LogError(NO_ERRCODE, "couldn't allocate memory for TCP listen sockets, suspending TCP message reception.");
-               freeaddrinfo(res);
-               return NULL;
-        }
-
-        *socks = 0;   /* num of sockets counter at start of array */
-        s = socks + 1;
-	for (r = res; r != NULL ; r = r->ai_next) {
-               *s = socket(r->ai_family, r->ai_socktype, r->ai_protocol);
-        	if (*s < 0) {
-			if(!(r->ai_family == PF_INET6 && errno == EAFNOSUPPORT))
-				errmsg.LogError(NO_ERRCODE, "create_tcp_socket(), socket");
-				/* it is debatable if PF_INET with EAFNOSUPPORT should
-				 * also be ignored...
-				 */
-                        continue;
-                }
-
-#ifdef IPV6_V6ONLY
-                if (r->ai_family == AF_INET6) {
-                	int iOn = 1;
-			if (setsockopt(*s, IPPROTO_IPV6, IPV6_V6ONLY,
-			      (char *)&iOn, sizeof (iOn)) < 0) {
-			errmsg.LogError(NO_ERRCODE, "TCP setsockopt");
-			close(*s);
-			*s = -1;
-			continue;
-                	}
-                }
-#endif
-       		if (setsockopt(*s, SOL_SOCKET, SO_REUSEADDR,
-			       (char *) &on, sizeof(on)) < 0 ) {
-			errmsg.LogError(NO_ERRCODE, "TCP setsockopt(REUSEADDR)");
-                        close(*s);
-			*s = -1;
-			continue;
-		}
-
-		/* We need to enable BSD compatibility. Otherwise an attacker
-		 * could flood our log files by sending us tons of ICMP errors.
-		 */
-#ifndef OS_BSD	
-		if(net.should_use_so_bsdcompat()) {
-			if (setsockopt(*s, SOL_SOCKET, SO_BSDCOMPAT,
-					(char *) &on, sizeof(on)) < 0) {
-				errmsg.LogError(NO_ERRCODE, "TCP setsockopt(BSDCOMPAT)");
-                                close(*s);
-				*s = -1;
-				continue;
-			}
-		}
-#endif
-
-	        if( (bind(*s, r->ai_addr, r->ai_addrlen) < 0)
-#ifndef IPV6_V6ONLY
-		     && (errno != EADDRINUSE)
-#endif
-	           ) {
-                        errmsg.LogError(NO_ERRCODE, "TCP bind");
-                	close(*s);
-			*s = -1;
-                        continue;
-                }
-
-		if( listen(*s,pThis->iSessMax / 10 + 5) < 0) {
-			/* If the listen fails, it most probably fails because we ask
-			 * for a too-large backlog. So in this case we first set back
-			 * to a fixed, reasonable, limit that should work. Only if
-			 * that fails, too, we give up.
-			 */
-			errmsg.LogError(NO_ERRCODE, "listen with a backlog of %d failed - retrying with default of 32.",
-				    pThis->iSessMax / 10 + 5);
-			if(listen(*s, 32) < 0) {
-				errmsg.LogError(NO_ERRCODE, "TCP listen, suspending tcp inet");
-	                	close(*s);
-				*s = -1;
-               		        continue;
-			}
-		}
+		TCPLstnPort = (uchar*)pThis->TCPLstnPort;
 
-		(*socks)++;
-		s++;
-	}
-
-        if(res != NULL)
-               freeaddrinfo(res);
-
-	if(Debug && *socks != maxs)
-		dbgprintf("We could initialize %d TCP listen sockets out of %d we received "
-		 	"- this may or may not be an error indication.\n", *socks, maxs);
-
-        if(*socks == 0) {
-		errmsg.LogError(NO_ERRCODE, "No TCP listen socket could successfully be initialized, "
-			 "message reception via TCP disabled.\n");
-        	free(socks);
-		return(NULL);
-	}
+	/* TODO: add capability to specify local listen address! */
+	CHKiRet(netstrm.LstnInit(pThis->pLstn, TCPLstnPort, NULL, pThis->iSessMax));
 
-	/* OK, we had success. Now it is also time to
-	 * initialize our connections
-	 */
-	if(TCPSessTblInit(pThis) != 0) {
-		/* OK, we are in some trouble - we could not initialize the
-		 * session table, so we can not continue. We need to free all
-		 * we have assigned so far, because we can not really use it...
-		 */
-		errmsg.LogError(NO_ERRCODE, "Could not initialize TCP session table, suspending TCP message reception.");
-		freeAllSockets(&socks); /* prevent a socket leak */
-		return(NULL);
-	}
-
-	return(socks);
+finalize_it:
+	RETiRet;
 }
 
 
@@ -414,34 +260,26 @@ static int *create_tcp_socket(tcpsrv_t *pThis)
  * rgerhards, 2008-03-02
  */
 static rsRetVal
-SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, int fd)
+SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, nsd_t *pNsd)
 {
 	DEFiRet;
 	tcps_sess_t *pSess;
-	int newConn;
+	netstrm_t *pNewStrm = NULL;
 	int iSess = -1;
 	struct sockaddr_storage addr;
-	socklen_t addrlen = sizeof(struct sockaddr_storage);
 	uchar fromHost[NI_MAXHOST];
 	uchar fromHostFQDN[NI_MAXHOST];
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
 
-	newConn = accept(fd, (struct sockaddr*) &addr, &addrlen);
-	if (newConn < 0) {
-		errmsg.LogError(NO_ERRCODE, "tcp accept, ignoring error and connection request");
-		ABORT_FINALIZE(RS_RET_ERR); // TODO: better error code
-		//was: return -1;
-	}
+	CHKiRet(netstrm.AcceptConnReq(pThis->pLstn, pNsd, &pNewStrm));
 
 	/* Add to session list */
 	iSess = TCPSessTblFindFreeSpot(pThis);
 	if(iSess == -1) {
 		errno = 0;
 		errmsg.LogError(NO_ERRCODE, "too many tcp sessions - dropping incoming request");
-		close(newConn);
-		ABORT_FINALIZE(RS_RET_ERR); // TODO: better error code
-		//was: return -1;
+		ABORT_FINALIZE(RS_RET_MAX_SESS_REACHED);
 	} else {
 		/* we found a free spot and can construct our session object */
 		CHKiRet(tcps_sess.Construct(&pSess));
@@ -455,7 +293,6 @@ SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, int fd)
 		 * are now told to discard the connection request.
 		 * Error message has been generated by cvthname.
 		 */
-		close (newConn);
 		ABORT_FINALIZE(RS_RET_ERR); // TODO: better error code
 	}
 
@@ -469,10 +306,8 @@ SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, int fd)
 		dbgprintf("%s is not an allowed sender\n", (char *) fromHostFQDN);
 		if(glbl.GetOption_DisallowWarning()) {
 			errno = 0;
-			errmsg.LogError(NO_ERRCODE, "TCP message from disallowed sender %s discarded",
-				   (char*)fromHost);
+			errmsg.LogError(NO_ERRCODE, "TCP message from disallowed sender %s discarded", fromHost);
 		}
-		close(newConn);
 		ABORT_FINALIZE(RS_RET_HOST_NOT_PERMITTED);
 	}
 
@@ -480,7 +315,8 @@ SessAccept(tcpsrv_t *pThis, tcps_sess_t **ppSess, int fd)
 	 * means we can finally fill in the session object.
 	 */
 	CHKiRet(tcps_sess.SetHost(pSess, fromHost));
-	CHKiRet(tcps_sess.SetSock(pSess, newConn));
+	CHKiRet(tcps_sess.SetStrm(pSess, pNewStrm));
+	pNewStrm = NULL; /* prevent it from being freed in error handler, now done in tcps_sess! */
 	CHKiRet(tcps_sess.SetMsgIdx(pSess, 0));
 	CHKiRet(tcps_sess.ConstructFinalize(pSess));
 
@@ -499,6 +335,8 @@ finalize_it:
 				tcps_sess.Destruct(&pThis->pSessions[iSess]);
 		}
 		iSess = -1; // TODO: change this to be fully iRet compliant ;)
+		if(pNewStrm != NULL)
+			netstrm.Destruct(&pNewStrm);
 	}
 
 	RETiRet;
@@ -517,6 +355,7 @@ Run(tcpsrv_t *pThis)
 	int iTCPSess;
 	fd_set readfds;
 	tcps_sess_t *pNewSess;
+	nsdsel_t *pSel;
 
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
 
@@ -525,11 +364,9 @@ Run(tcpsrv_t *pThis)
 	 * right into the sleep below.
 	 */
 	while(1) {
-	        maxfds = 0;
-	        FD_ZERO (&readfds);
+		CHKiRet(nsdsel_ptcp.Construct(&pSel));
 
-		/* Add the TCP listen sockets to the list of read descriptors.
-	    	 */
+		/* Add the TCP listen sockets to the list of read descriptors. */
 		if(pThis->pSocksLstn != NULL && *pThis->pSocksLstn) {
 			for (i = 0; i < *pThis->pSocksLstn; i++) {
 				/* The if() below is theoretically not needed, but I leave it in
@@ -537,8 +374,8 @@ Run(tcpsrv_t *pThis)
 				 * feature is not yet supported by the current code base.
 				 */
 				if (pThis->pSocksLstn[i+1] != -1) {
-					if(Debug)
-						net.debugListenInfo(pThis->pSocksLstn[i+1], "TCP");
+					if(Debug) net.debugListenInfo(pThis->pSocksLstn[i+1], "TCP");
+					CHKiRet(nsdsel_ptcp.Add(pSel, ));
 					FD_SET(pThis->pSocksLstn[i+1], &readfds);
 					if(pThis->pSocksLstn[i+1]>maxfds) maxfds=pThis->pSocksLstn[i+1];
 				}
@@ -566,6 +403,7 @@ Run(tcpsrv_t *pThis)
 		}
 
 		/* wait for io to become ready */
+		/* this is the somewhat weak spot in our socket layer abstraction... */
 		nfds = select(maxfds+1, (fd_set *) &readfds, NULL, NULL, NULL);
 
 		for (i = 0; i < *pThis->pSocksLstn; i++) {
@@ -620,7 +458,6 @@ Run(tcpsrv_t *pThis)
 /* Standard-Constructor
  */
 BEGINobjConstruct(tcpsrv) /* be sure to specify the object type also in END macro! */
-	pThis->pSocksLstn = NULL;
 	pThis->iSessMax = 200; /* TODO: useful default ;) */
 ENDobjConstruct(tcpsrv)
 
@@ -632,7 +469,7 @@ tcpsrvConstructFinalize(tcpsrv_t __attribute__((unused)) *pThis)
 {
 	DEFiRet;
 	ISOBJ_TYPE_assert(pThis, tcpsrv);
-	pThis->pSocksLstn = pThis->OpenLstnSocks(pThis);
+	iRet = pThis->OpenLstnSocks(pThis);
 
 	RETiRet;
 }
@@ -727,7 +564,7 @@ SetCBOnErrClose(tcpsrv_t *pThis, rsRetVal (*pCB)(tcps_sess_t*))
 }
 
 static rsRetVal
-SetCBOpenLstnSocks(tcpsrv_t *pThis, int* (*pCB)(tcpsrv_t*))
+SetCBOpenLstnSocks(tcpsrv_t *pThis, rsRetVal (*pCB)(tcpsrv_t*))
 {
 	DEFiRet;
 	pThis->OpenLstnSocks = pCB;
@@ -793,6 +630,7 @@ CODESTARTObjClassExit(tcpsrv)
 	objRelease(conf, CORE_COMPONENT);
 	objRelease(glbl, CORE_COMPONENT);
 	objRelease(errmsg, CORE_COMPONENT);
+	objRelease(netstrm, LM_NET_FILENAME);
 	objRelease(net, LM_NET_FILENAME);
 ENDObjClassExit(tcpsrv)
 
@@ -805,6 +643,7 @@ BEGINObjClassInit(tcpsrv, 1, OBJ_IS_LOADABLE_MODULE) /* class, version - CHANGE
 	/* request objects we use */
 	CHKiRet(objUse(errmsg, CORE_COMPONENT));
 	CHKiRet(objUse(net, LM_NET_FILENAME));
+	CHKiRet(objUse(netstrm, LM_NET_FILENAME));
 	CHKiRet(objUse(tcps_sess, DONT_LOAD_LIB));
 	CHKiRet(objUse(conf, CORE_COMPONENT));
 	CHKiRet(objUse(glbl, CORE_COMPONENT));