From 30380b374416d8ec8ac7d7037026a62e1ff5aeb0 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Mon, 4 Jul 2022 14:03:09 +0800 Subject: [PATCH 1/3] fix invalid read/write --- source/dnode/mgmt/node_mgmt/src/dmProc.c | 20 +++++++++---------- source/dnode/mgmt/node_mgmt/src/dmTransport.c | 2 +- source/libs/transport/src/transSvr.c | 5 +++-- 3 files changed, 14 insertions(+), 13 deletions(-) diff --git a/source/dnode/mgmt/node_mgmt/src/dmProc.c b/source/dnode/mgmt/node_mgmt/src/dmProc.c index 72878d0d85..cbf13924d7 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmProc.c +++ b/source/dnode/mgmt/node_mgmt/src/dmProc.c @@ -87,8 +87,8 @@ static SProcQueue *dmInitProcQueue(SProc *proc, char *ptr, int32_t size) { static void dmCleanupProcQueue(SProcQueue *queue) {} static inline int32_t dmPushToProcQueue(SProc *proc, SProcQueue *queue, SRpcMsg *pMsg, EProcFuncType ftype) { - const void *pHead = pMsg; - const void *pBody = pMsg->pCont; + const void * pHead = pMsg; + const void * pBody = pMsg->pCont; const int16_t rawHeadLen = sizeof(SRpcMsg); const int32_t rawBodyLen = pMsg->contLen; const int16_t headLen = CEIL8(rawHeadLen); @@ -257,7 +257,7 @@ int32_t dmInitProc(struct SMgmtWrapper *pWrapper) { proc->wrapper = pWrapper; proc->name = pWrapper->name; - SShm *shm = &proc->shm; + SShm * shm = &proc->shm; int32_t cstart = 0; int32_t csize = CEIL8(shm->size / 2); int32_t pstart = csize; @@ -281,13 +281,13 @@ int32_t dmInitProc(struct SMgmtWrapper *pWrapper) { } static void *dmConsumChildQueue(void *param) { - SProc *proc = param; + SProc * proc = param; SMgmtWrapper *pWrapper = proc->wrapper; - SProcQueue *queue = proc->cqueue; + SProcQueue * queue = proc->cqueue; int32_t numOfMsgs = 0; int32_t code = 0; EProcFuncType ftype = DND_FUNC_REQ; - SRpcMsg *pMsg = NULL; + SRpcMsg * pMsg = NULL; dDebug("node:%s, start to consume from cqueue", proc->name); do { @@ -324,13 +324,13 @@ static void *dmConsumChildQueue(void *param) { } static void *dmConsumParentQueue(void *param) { - SProc *proc = param; + SProc * proc = param; SMgmtWrapper *pWrapper = proc->wrapper; - SProcQueue *queue = proc->pqueue; + SProcQueue * queue = proc->pqueue; int32_t numOfMsgs = 0; int32_t code = 0; EProcFuncType ftype = DND_FUNC_REQ; - SRpcMsg *pMsg = NULL; + SRpcMsg * pMsg = NULL; dDebug("node:%s, start to consume from pqueue", proc->name); do { @@ -353,7 +353,7 @@ static void *dmConsumParentQueue(void *param) { rpcRegisterBrokenLinkArg(pMsg); } else if (ftype == DND_FUNC_RELEASE) { dmRemoveProcRpcHandle(proc, pMsg->info.handle); - rpcReleaseHandle(pMsg->info.handle, (int8_t)pMsg->code); + rpcReleaseHandle(&pMsg->info, TAOS_CONN_SERVER); } else { dError("node:%s, invalid ftype:%d from pqueue", proc->name, ftype); rpcFreeCont(pMsg->pCont); diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index 35d478177a..df3c9c4e88 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -245,7 +245,7 @@ static inline void dmReleaseHandle(SRpcHandleInfo *pHandle, int8_t type) { SRpcMsg msg = {.code = type, .info = *pHandle}; dmPutToProcPQueue(&pWrapper->proc, &msg, DND_FUNC_RELEASE); } else { - rpcReleaseHandle(pHandle->handle, type); + rpcReleaseHandle(pHandle, type); } } diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index d32156dd0d..4f33c8cdc1 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -1029,8 +1029,9 @@ void transUnrefSrvHandle(void* handle) { } void transReleaseSrvHandle(void* handle) { - SExHandle* exh = handle; - int64_t refId = exh->refId; + SRpcHandleInfo* info = handle; + SExHandle* exh = info->handle; + int64_t refId = info->refId; ASYNC_CHECK_HANDLE(exh, refId); From d4f1614242dd8a5f3dc2a93b3082cc65b74fb8e4 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Mon, 4 Jul 2022 14:28:50 +0800 Subject: [PATCH 2/3] fix invalid read/write --- source/libs/transport/src/transCli.c | 23 ++++++++++++++++------- 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index c140de24d8..a21e97d049 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -262,13 +262,17 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) { #define REQUEST_PERSIS_HANDLE(msg) ((msg)->info.persistHandle == 1) #define REQUEST_RELEASE_HANDLE(cmsg) ((cmsg)->type == Release) +#define EPSET_IS_VALID(epSet) ((epSet) != NULL && (epSet)->numOfEps != 0) #define EPSET_GET_SIZE(epSet) (epSet)->numOfEps #define EPSET_GET_INUSE_IP(epSet) ((epSet)->eps[(epSet)->inUse].fqdn) #define EPSET_GET_INUSE_PORT(epSet) ((epSet)->eps[(epSet)->inUse].port) -#define EPSET_FORWARD_INUSE(epSet) \ - do { \ - (epSet)->inUse = (++((epSet)->inUse)) % ((epSet)->numOfEps); \ +#define EPSET_FORWARD_INUSE(epSet) \ + do { \ + if ((epSet)->numOfEps != 0) { \ + (epSet)->inUse = (++((epSet)->inUse)) % ((epSet)->numOfEps); \ + } \ } while (0) + #define EPSET_DEBUG_STR(epSet, tbuf) \ do { \ int len = snprintf(tbuf, sizeof(tbuf), "epset:{"); \ @@ -512,7 +516,6 @@ static void allocConnRef(SCliConn* conn, bool update) { } static void addConnToPool(void* pool, SCliConn* conn) { if (conn->status == ConnInPool) { - // assert(0); return; } SCliThrd* thrd = conn->hostThrd; @@ -668,7 +671,6 @@ static void cliSendCb(uv_write_t* req, int status) { void cliSend(SCliConn* pConn) { CONN_HANDLE_BROKEN(pConn); - // assert(taosArrayGetSize(pConn->cliMsgs) > 0); assert(!transQueueEmpty(&pConn->cliMsgs)); SCliMsg* pCliMsg = NULL; @@ -810,6 +812,11 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) { STrans* pTransInst = pThrd->pTransInst; cliMayCvtFqdnToIp(&pCtx->epSet, &pThrd->cvtAddr); + if (!EPSET_IS_VALID(&pCtx->epSet)) { + destroyCmsg(pMsg); + tError("invalid epset"); + return; + } bool ignore = false; SCliConn* conn = cliGetConn(pMsg, pThrd, &ignore); @@ -1077,12 +1084,14 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { } else { cliCompareAndSwap(&pCtx->retryLimit, TRANS_RETRY_COUNT_LIMIT, TRANS_RETRY_COUNT_LIMIT); if (pCtx->retryCnt < pCtx->retryLimit) { - addConnToPool(pThrd->pool, pConn); if (pResp->contLen == 0) { EPSET_FORWARD_INUSE(&pCtx->epSet); } else { - tDeserializeSEpSet(pResp->pCont, pResp->contLen, &pCtx->epSet); + if (tDeserializeSEpSet(pResp->pCont, pResp->contLen, &pCtx->epSet) < 0) { + tError("%s conn %p failed to deserialize epset", CONN_GET_INST_LABEL(pConn)); + } } + addConnToPool(pThrd->pool, pConn); transFreeMsg(pResp->pCont); cliSchedMsgToNextNode(pMsg, pThrd); return -1; From df36cd165b92ca7ec2355d497be91df003338561 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Mon, 4 Jul 2022 15:01:51 +0800 Subject: [PATCH 3/3] fix invalid read/write --- source/libs/transport/test/transUT.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/libs/transport/test/transUT.cpp b/source/libs/transport/test/transUT.cpp index 86c4830284..b55f771ebd 100644 --- a/source/libs/transport/test/transUT.cpp +++ b/source/libs/transport/test/transUT.cpp @@ -175,7 +175,7 @@ static void processReleaseHandleCb(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) rpcMsg.code = 0; rpcSendResponse(&rpcMsg); - rpcReleaseHandle(pMsg->info.handle, TAOS_CONN_SERVER); + rpcReleaseHandle(&pMsg->info, TAOS_CONN_SERVER); } static void processRegisterFailure(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) { {