relpsess: skip graceful close for half-open clients

src/copen.c

@@ -158,6 +158,9 @@ BEGINcommand(S, Init)
 	CHKRet(relpSessSendResponse(pSess, pFrame->txnr, pszSrvOffers, lenSrvOffers));
 
 	pSess->bServerConnOpen = 1;
+	if(pSess->pEngine->onSessOpen != NULL) {
+		pSess->pEngine->onSessOpen(pSess->pUsr, pSess);
+	}
 
 finalize_it:
 	if(pszSrvOffers != NULL)
@@ -168,6 +171,9 @@ BEGINcommand(S, Init)
 		relpOffersDestruct(&pSrvOffers);
 
 	if(iRet != RELP_RET_OK) {
+		if(pSess->pEngine->onSessOpenFail != NULL) {
+			pSess->pEngine->onSessOpenFail(pSess->pUsr, pSess, iRet);
+		}
 		if(iRet == RELP_RET_RQD_FEAT_MISSING) {
 			strncpy(szErrMsg, "500 required command not supported by client", sizeof(szErrMsg));
 			lenErrMsg = 44;

src/librelp.h

@@ -220,6 +220,62 @@ relpRetVal relpEngineSetOnErr(relpEngine_t *pThis,
 				void (*pCB)(void*pUsr, char *objinfo, char*errmsg, relpRetVal errcode) );
 relpRetVal relpEngineSetOnGenericErr(relpEngine_t *pThis,
 				void (*pCB)(char *objinfo, char*errmsg, relpRetVal errcode) );
+/**
+ * Set a session-open callback.
+ *
+ * Callback signature:
+ *   void cb(void *pUsr, const relpSess_t *pSess)
+ *
+ * The callback is invoked after a server-side RELP session successfully
+ * completes the open handshake and is ready for traffic. The callback runs
+ * on the relpEngineRun() thread. The relpSess_t pointer is only valid for the
+ * duration of the callback and must not be retained.
+ *
+ * Callback parameters:
+ *
+ * pUsr  - user pointer from relpSrvSetUsrPtr()
+ * pSess - session being opened (read-only, ephemeral)
+ */
+relpRetVal relpEngineSetOnSessOpen(relpEngine_t *pThis,
+				void (*pCB)(void*pUsr, const relpSess_t *pSess) );
+/**
+ * Set a session-close callback.
+ *
+ * Callback signature:
+ *   void cb(void *pUsr, const relpSess_t *pSess, relpRetVal reason)
+ *
+ * The callback fires when a session is removed from the engine, regardless
+ * of the close cause (protocol close, I/O error, or engine shutdown). The
+ * reason parameter contains the relpRetVal that triggered teardown; engine
+ * shutdown uses RELP_RET_SESSION_CLOSED. The callback runs on the
+ * relpEngineRun() thread and pSess is valid only during the callback.
+ *
+ * Callback parameters:
+ *
+ * pUsr   - user pointer from relpSrvSetUsrPtr()
+ * pSess  - session being closed (read-only, ephemeral)
+ * reason - relpRetVal that caused teardown
+ */
+relpRetVal relpEngineSetOnSessClose(relpEngine_t *pThis,
+				void (*pCB)(void*pUsr, const relpSess_t *pSess, relpRetVal reason) );
+/**
+ * Set a session-open-failed callback.
+ *
+ * Callback signature:
+ *   void cb(void *pUsr, const relpSess_t *pSess, relpRetVal reason)
+ *
+ * The callback fires when the server-side open handshake fails. It is
+ * invoked before the error response is sent. The callback runs on the
+ * relpEngineRun() thread and pSess is valid only during the callback.
+ *
+ * Callback parameters:
+ *
+ * pUsr   - user pointer from relpSrvSetUsrPtr()
+ * pSess  - session that failed to open (read-only, ephemeral)
+ * reason - relpRetVal describing the failure
+ */
+relpRetVal relpEngineSetOnSessOpenFail(relpEngine_t *pThis,
+				void (*pCB)(void*pUsr, const relpSess_t *pSess, relpRetVal reason) );
 
 /* exposed server property set functions */
 relpRetVal relpSrvSetLstnPort(relpSrv_t *pThis, unsigned char *pLstnPort);

src/relp.c

@@ -220,11 +220,12 @@ relpEngineAddToSess(relpEngine_t *const pThis, relpSess_t *pSess)
  * rgerhards, 2008-03-17
  */
 static relpRetVal
-relpEngineDelSess(relpEngine_t *const pThis, relpEngSessLst_t *pSessLstEntry)
+relpEngineDelSess(relpEngine_t *const pThis, relpEngSessLst_t *pSessLstEntry, relpRetVal reason)
 {
 	ENTER_RELPFUNC;
 	RELPOBJ_assert(pThis, Engine);
 	assert(pSessLstEntry != NULL);
+	relpSess_t *const pSess = pSessLstEntry->pSess;
 
 #	if defined(HAVE_EPOLL_CREATE1) || defined(HAVE_EPOLL_CREATE)
 	delSessFromEpoll(pThis, pSessLstEntry);
@@ -234,6 +235,10 @@ relpEngineDelSess(relpEngine_t *const pThis, relpEngSessLst_t *pSessLstEntry)
 	--pThis->lenSessLst;
 	pthread_mutex_unlock(&pThis->mutSessLst);
 
+	if(pThis->onSessClose != NULL) {
+		pThis->onSessClose(pSess->pUsr, pSess, reason);
+	}
+
 	relpSessDestruct(&pSessLstEntry->pSess);
 	free(pSessLstEntry);
 
@@ -296,8 +301,7 @@ relpEngineDestruct(relpEngine_t **ppThis)
 	/* now destruct all currently existing sessions */
 	for(pSessL = pThis->pSessLstRoot ; pSessL != NULL ; pSessL = pSessLNxt) {
 		pSessLNxt = pSessL->pNext;
-		relpSessDestruct(&pSessL->pSess);
-		free(pSessL);
+		relpEngineDelSess(pThis, pSessL, RELP_RET_SESSION_CLOSED);
 	}
 
 	/* and now all existing servers... */
@@ -559,6 +563,35 @@ relpEngineSetOnGenericErr(relpEngine_t *const pThis, void (*pCB)(char *objinfo,
 	LEAVE_RELPFUNC;
 }
 
+relpRetVal PART_OF_API
+relpEngineSetOnSessOpen(relpEngine_t *const pThis, void (*pCB)(void*pUsr, const relpSess_t *pSess) )
+{
+	ENTER_RELPFUNC;
+	RELPOBJ_assert(pThis, Engine);
+	pThis->onSessOpen = pCB;
+	LEAVE_RELPFUNC;
+}
+
+relpRetVal PART_OF_API
+relpEngineSetOnSessClose(relpEngine_t *const pThis,
+		void (*pCB)(void*pUsr, const relpSess_t *pSess, relpRetVal reason) )
+{
+	ENTER_RELPFUNC;
+	RELPOBJ_assert(pThis, Engine);
+	pThis->onSessClose = pCB;
+	LEAVE_RELPFUNC;
+}
+
+relpRetVal PART_OF_API
+relpEngineSetOnSessOpenFail(relpEngine_t *const pThis,
+		void (*pCB)(void*pUsr, const relpSess_t *pSess, relpRetVal reason) )
+{
+	ENTER_RELPFUNC;
+	RELPOBJ_assert(pThis, Engine);
+	pThis->onSessOpenFail = pCB;
+	LEAVE_RELPFUNC;
+}
+
 /* Deprecated, use relpEngineListnerConstruct() family of functions.
  * See there for further information.
  */
@@ -665,7 +698,7 @@ doRecv(relpEngine_t *const pThis, relpEngSessLst_t *pSessEtry, int sock)
 	if(localRet != RELP_RET_OK) {
 		pThis->dbgprint((char*)"relp session %d iRet %d, tearing it down\n",
 				sock, localRet);
-		relpEngineDelSess(pThis, pSessEtry);
+		relpEngineDelSess(pThis, pSessEtry, localRet);
 	}
 	return localRet;
 }
@@ -684,7 +717,7 @@ doSend(relpEngine_t *const pThis, relpEngSessLst_t *pSessEtry, int sock)
 	if(localRet != RELP_RET_OK) {
 		pThis->dbgprint((char*)"relp session %d iRet %d during send, tearing it down\n",
 				sock, localRet);
-		relpEngineDelSess(pThis, pSessEtry);
+		relpEngineDelSess(pThis, pSessEtry, localRet);
 	}
 }
 
@@ -801,7 +834,7 @@ handleSessIO(relpEngine_t *const pThis, epolld_t *epd)
 				if(localRet != RELP_RET_OK) {
 					pThis->dbgprint((char*)"relp session %d handshake iRet %d, tearing it down\n",
 							epd->sock, localRet);
-					relpEngineDelSess(pThis, pSessEtry);
+				    relpEngineDelSess(pThis, pSessEtry, localRet);
 				}
 #			else
 					pThis->dbgprint((char*)"librelp error: handshake retry requested in "
@@ -1001,7 +1034,7 @@ engineEventLoopRun(relpEngine_t *const pThis)
 								pThis->dbgprint((char*)"relp session %d handshake "
 										"iRet %d, tearing it down\n",
 										sock, localRet);
-								relpEngineDelSess(pThis, pSessEtry);
+							    relpEngineDelSess(pThis, pSessEtry, localRet);
 							}
 #						else
 							pThis->dbgprint((char*)"librelp error: handshake retry "

src/relp.h

@@ -130,11 +130,14 @@ struct relpEngine_s {
                                   unsigned char *pMsg, size_t lenMsg); /**< callback for "syslog" cmd */
         relpRetVal (*onSyslogRcv2)(void*, unsigned char*pHostname, unsigned char *pIP,
                                   unsigned char *pMsg, size_t lenMsg); /**< callback for "syslog" cmd */
-       relpRetVal (*onSyslogRcv3)(void*, unsigned char*pHostname, unsigned char *pIP,
+        relpRetVal (*onSyslogRcv3)(void*, unsigned char*pHostname, unsigned char *pIP,
                                   unsigned char *pPort, unsigned char *pMsg, size_t lenMsg); /**< callback for "syslog" cmd */
         void (*onAuthErr)(void*pUsr, char *authinfo, char*errmsg, relpRetVal errcode);
         void (*onErr)(void*pUsr, char *objinfo, char*errmsg, relpRetVal errcode);
         void (*onGenericErr)(char *objinfo, char*errmsg, relpRetVal errcode);
+        void (*onSessOpen)(void*pUsr, const relpSess_t *pSess);
+        void (*onSessClose)(void*pUsr, const relpSess_t *pSess, relpRetVal reason);
+        void (*onSessOpenFail)(void*pUsr, const relpSess_t *pSess, relpRetVal reason);
 	int protocolVersion; /**< version of the relp protocol supported by this engine */
 
 	/* Flags */

src/relpsess.c

@@ -89,6 +89,13 @@ callOnErr(const relpSess_t *__restrict__ const pThis,
 	}
 }
 
+static int
+relpSessCltCanGracefullyDisconnect(const relpSess_t *const pThis)
+{
+	return pThis->sessState == eRelpSessState_READY_TO_SEND
+	    || pThis->sessState == eRelpSessState_WINDOW_FULL;
+}
+
 
 /* helper to free permittedPeer structure */
 static inline void
@@ -180,8 +187,7 @@ relpSessDestruct(relpSess_t **ppThis)
 			}
 		} else {
 			/* we are at the client side of the connection */
-			if(   pThis->sessState != eRelpSessState_DISCONNECTED
-			   && pThis->sessState != eRelpSessState_BROKEN) {
+			if(relpSessCltCanGracefullyDisconnect(pThis)) {
 				relpSessCltDoDisconnect(pThis);
 			}
 		}
@@ -613,15 +619,15 @@ relpSessWaitState(relpSess_t *const pThis, const relpSessState_t stateExpected,
 		clock_gettime(CLOCK_REALTIME, &tCurr);
 	}
 
-finalize_it:
+	finalize_it:
 	pThis->pEngine->dbgprint((char*)"relpSessWaitState returns %d\n", iRet);
-	if(	iRet == RELP_RET_TIMED_OUT ||
-		iRet == RELP_RET_SESSION_BROKEN ||
-		relpEngineShouldStop(pThis->pEngine)) {
+	if(iRet == RELP_RET_TIMED_OUT || relpEngineShouldStop(pThis->pEngine)) {
 		/* the session is broken! */
 		callOnErr(pThis, (char*) "error waiting on required session state, session broken",
 			RELP_RET_SESSION_BROKEN);
 		pThis->sessState = eRelpSessState_BROKEN;
+	} else if(iRet == RELP_RET_SESSION_BROKEN) {
+		pThis->sessState = eRelpSessState_BROKEN;
 	}
 
 	LEAVE_RELPFUNC;

tests/Makefile.am

@@ -49,12 +49,14 @@ TESTS=  selftest_receive_watchdog.sh \
 	selftest_receive_usage.sh \
 	basic.sh \
 	basic-realistic.sh \
+	session-callbacks.sh \
 	receiver-abort.sh \
 	long-msg.sh \
 	oversize-msg-abort-errmsg.sh \
 	oversize-msg-accept-errmsg.sh \
 	truncate-oversize-msg.sh \
 	send-noconnect.sh \
+	send-dummyserver-no-waitstate-errmsg.sh \
 	receive-emptyconnect.sh
 if ENABLE_TLS_GENERIC
 TESTS+=$(TLS_TESTS)
@@ -78,6 +80,8 @@ EXTRA_DIST=$(TESTS) \
 	set-envvars.in \
 	dummyclient.py \
 	dummyserver.py \
+	invalid-open.py \
+	session-callbacks.sh \
 	test-framework.sh \
 	receive.c \
 	send.c \
@@ -95,4 +99,3 @@ EXTRA_DIST=$(TESTS) \
 	tls-certs/ossl-server-cert.pem \
 	tls-certs/ossl-server-key.pem \
 	tls-certs/ossl-server-certchain.pem
-

tests/invalid-open.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+
+import os
+import socket
+import time
+
+
+def main() -> None:
+    port = int(os.environ["TESTPORT"])
+    frame = b"1 open 12 relp_version\n"
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.settimeout(2)
+    sock.connect(("127.0.0.1", port))
+    sock.sendall(frame)
+    try:
+        sock.recv(1024)
+    except socket.timeout:
+        pass
+    time.sleep(0.1)
+    sock.close()
+
+
+if __name__ == "__main__":
+    main()