user auth

include/common/tmsg.h

@@ -73,7 +73,8 @@ typedef uint16_t tmsg_t;
 enum { CONN_TYPE__QUERY = 1, CONN_TYPE__TMQ, CONN_TYPE__MAX };
 
 enum {
-  HEARTBEAT_KEY_DBINFO = 1,
+  HEARTBEAT_KEY_USER_AUTHINFO = 1,
+  HEARTBEAT_KEY_DBINFO,
   HEARTBEAT_KEY_STBINFO,
   HEARTBEAT_KEY_MQ_TMP,
 };
@@ -669,10 +670,20 @@ typedef struct {
   SArray* pArray;  // Array of SUseDbRsp
 } SUseDbBatchRsp;
 
+
 int32_t tSerializeSUseDbBatchRsp(void* buf, int32_t bufLen, SUseDbBatchRsp* pRsp);
 int32_t tDeserializeSUseDbBatchRsp(void* buf, int32_t bufLen, SUseDbBatchRsp* pRsp);
 void    tFreeSUseDbBatchRsp(SUseDbBatchRsp* pRsp);
 
+typedef struct {
+  SArray* pArray;  // Array of SGetUserAuthRsp
+} SUserAuthBatchRsp;
+
+int32_t tSerializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchRsp* pRsp);
+int32_t tDeserializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchRsp* pRsp);
+void    tFreeSUserAuthBatchRsp(SUserAuthBatchRsp* pRsp);
+
+
 typedef struct {
   char db[TSDB_DB_FNAME_LEN];
 } SCompactDbReq;

include/libs/catalog/catalog.h

@@ -40,9 +40,11 @@ enum {
   CTG_DBG_STB_RENT_NUM,
 };
 
-#define USER_AUTH_READ  1
-#define USER_AUTH_WRITE 2
-#define USER_AUTH_ALL   4
+typedef enum {
+  AUTH_TYPE_READ = 1,
+  AUTH_TYPE_WRITE,
+  AUTH_TYPE_OTHER,
+} AUTH_TYPE;
 
 typedef struct SCatalogReq {
   SArray *pTableName;     // element is SNAME
@@ -81,6 +83,11 @@ typedef struct SDbVgVersion {
   int32_t numOfTable; // unit is TSDB_TABLE_NUM_UNIT
 } SDbVgVersion;
 
+typedef struct SUserAuthVersion {
+  char    user[TSDB_USER_LEN];
+  int32_t version;
+} SUserAuthVersion;
+
 typedef SDbCfgRsp SDbCfgInfo;
 typedef SUserIndexRsp SIndexInfo;
 
@@ -223,13 +230,17 @@ int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableMetaVersion **stable
 
 int32_t catalogGetExpiredDBs(SCatalog* pCatalog, SDbVgVersion **dbs, uint32_t *num);
 
+int32_t catalogGetExpiredUsers(SCatalog* pCtg, SUserAuthVersion **users, uint32_t *num);
+
 int32_t catalogGetDBCfg(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, SDbCfgInfo* pDbCfg);
 
 int32_t catalogGetIndexInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* indexName, SIndexInfo* pInfo);
 
 int32_t catalogGetUdfInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* funcName, SFuncInfo** pInfo);
 
-int32_t catalogGetUserDbAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, int32_t* auth);
+int32_t catalogChkAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, AUTH_TYPE type, bool *pass);
+
+int32_t catalogUpdateUserAuthInfo(SCatalog* pCtg, SGetUserAuthRsp* pAuth);
 
 
 /**

source/client/src/clientHb.c

@@ -28,6 +28,27 @@ static int32_t hbMqHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq
 
 static int32_t hbMqHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { return 0; }
 
+static int32_t hbProcessUserAuthInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
+  int32_t code = 0;
+
+  SUserAuthBatchRsp batchRsp = {0};
+  if (tDeserializeSUserAuthBatchRsp(value, valueLen, &batchRsp) != 0) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    return -1;
+  }
+
+  int32_t numOfBatchs = taosArrayGetSize(batchRsp.pArray);
+  for (int32_t i = 0; i < numOfBatchs; ++i) {
+    SGetUserAuthRsp *rsp = taosArrayGet(batchRsp.pArray, i);
+    tscDebug("hb user auth rsp, user:%s, version:%d", rsp->user, rsp->version);
+
+    catalogUpdateUserAuthInfo(pCatalog, rsp);
+  }
+
+  tFreeSUserAuthBatchRsp(&batchRsp);
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
   int32_t code = 0;
 
@@ -148,6 +169,24 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
   for (int32_t i = 0; i < kvNum; ++i) {
     SKv *kv = taosArrayGet(pRsp->info, i);
     switch (kv->key) {
+      case HEARTBEAT_KEY_USER_AUTHINFO: {
+        if (kv->valueLen <= 0 || NULL == kv->value) {
+          tscError("invalid hb user auth info, len:%d, value:%p", kv->valueLen, kv->value);
+          break;
+        }
+
+        int64_t         *clusterId = (int64_t *)info->param;
+        struct SCatalog *pCatalog = NULL;
+
+        int32_t code = catalogGetHandle(*clusterId, &pCatalog);
+        if (code != TSDB_CODE_SUCCESS) {
+          tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", *clusterId, tstrerror(code));
+          break;
+        }
+
+        hbProcessUserAuthInfoRsp(kv->value, kv->valueLen, pCatalog);
+        break;
+      }
       case HEARTBEAT_KEY_DBINFO: {
         if (kv->valueLen <= 0 || NULL == kv->value) {
           tscError("invalid hb db info, len:%d, value:%p", kv->valueLen, kv->value);
@@ -327,6 +366,39 @@ int32_t hbGetQueryBasicInfo(SClientHbKey *connKey, SClientHbReq *req) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t hbGetExpiredUserInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
+  SUserAuthVersion *users = NULL;
+  uint32_t          userNum = 0;
+  int32_t           code = 0;
+
+  code = catalogGetExpiredUsers(pCatalog, &users, &userNum);
+  if (TSDB_CODE_SUCCESS != code) {
+    return code;
+  }
+
+  if (userNum <= 0) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  for (int32_t i = 0; i < userNum; ++i) {
+    SUserAuthVersion *user = &users[i];
+    user->version = htonl(user->version);
+  }
+
+  SKv kv = {
+      .key = HEARTBEAT_KEY_USER_AUTHINFO,
+      .valueLen = sizeof(SUserAuthVersion) * userNum,
+      .value = users,
+  };
+
+  tscDebug("hb got %d expired users, valueLen:%d", userNum, kv.valueLen);
+
+  taosHashPut(req->info, &kv.key, sizeof(kv.key), &kv, sizeof(kv));
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 int32_t hbGetExpiredDBInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
   SDbVgVersion *dbs = NULL;
   uint32_t      dbNum = 0;
@@ -407,6 +479,11 @@ int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req
 
   hbGetQueryBasicInfo(connKey, req);
 
+  code = hbGetExpiredUserInfo(connKey, pCatalog, req);
+  if (TSDB_CODE_SUCCESS != code) {
+    return code;
+  }
+
   code = hbGetExpiredDBInfo(connKey, pCatalog, req);
   if (TSDB_CODE_SUCCESS != code) {
     return code;

source/common/src/tmsg.c

@@ -1145,31 +1145,47 @@ int32_t tDeserializeSGetUserAuthReq(void *buf, int32_t bufLen, SGetUserAuthReq *
   return 0;
 }
 
-int32_t tSerializeSGetUserAuthRsp(void *buf, int32_t bufLen, SGetUserAuthRsp *pRsp) {
-  SCoder encoder = {0};
-  tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
-
-  if (tStartEncode(&encoder) < 0) return -1;
-  if (tEncodeCStr(&encoder, pRsp->user) < 0) return -1;
-  if (tEncodeI8(&encoder, pRsp->superAuth) < 0) return -1;
+int32_t tSerializeSGetUserAuthRspImpl(SCoder *pEncoder, SGetUserAuthRsp *pRsp) {
+  if (tEncodeCStr(pEncoder, pRsp->user) < 0) return -1;
+  if (tEncodeI8(pEncoder, pRsp->superAuth) < 0) return -1;
+  if (tEncodeI32(pEncoder, pRsp->version) < 0) return -1;
 
+  int32_t numOfCreatedDbs = taosHashGetSize(pRsp->createdDbs);
   int32_t numOfReadDbs = taosHashGetSize(pRsp->readDbs);
   int32_t numOfWriteDbs = taosHashGetSize(pRsp->writeDbs);
-  if (tEncodeI32(&encoder, numOfReadDbs) < 0) return -1;
-  if (tEncodeI32(&encoder, numOfWriteDbs) < 0) return -1;
+  if (tEncodeI32(pEncoder, numOfCreatedDbs) < 0) return -1;
+  if (tEncodeI32(pEncoder, numOfReadDbs) < 0) return -1;
+  if (tEncodeI32(pEncoder, numOfWriteDbs) < 0) return -1;
 
-  char *db = taosHashIterate(pRsp->readDbs, NULL);
+  char *db = taosHashIterate(pRsp->createdDbs, NULL);
   while (db != NULL) {
-    if (tEncodeCStr(&encoder, db) < 0) return -1;
+    if (tEncodeCStr(pEncoder, db) < 0) return -1;
+    db = taosHashIterate(pRsp->createdDbs, db);
+  }
+
+  db = taosHashIterate(pRsp->readDbs, NULL);
+  while (db != NULL) {
+    if (tEncodeCStr(pEncoder, db) < 0) return -1;
     db = taosHashIterate(pRsp->readDbs, db);
   }
 
   db = taosHashIterate(pRsp->writeDbs, NULL);
   while (db != NULL) {
-    if (tEncodeCStr(&encoder, db) < 0) return -1;
+    if (tEncodeCStr(pEncoder, db) < 0) return -1;
     db = taosHashIterate(pRsp->writeDbs, db);
   }
 
+  return 0;
+}
+
+int32_t tSerializeSGetUserAuthRsp(void *buf, int32_t bufLen, SGetUserAuthRsp *pRsp) {
+  SCoder encoder = {0};
+  tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+
+  if (tSerializeSGetUserAuthRspImpl(&encoder, pRsp) < 0) return -1;
+
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -1177,39 +1193,58 @@ int32_t tSerializeSGetUserAuthRsp(void *buf, int32_t bufLen, SGetUserAuthRsp *pR
   return tlen;
 }
 
-int32_t tDeserializeSGetUserAuthRsp(void *buf, int32_t bufLen, SGetUserAuthRsp *pRsp) {
-  pRsp->readDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, false);
-  pRsp->writeDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, false);
+int32_t tDeserializeSGetUserAuthRspImpl(SCoder *pDecoder, SGetUserAuthRsp *pRsp) {
+  pRsp->createdDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
+  pRsp->readDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
+  pRsp->writeDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
   if (pRsp->readDbs == NULL || pRsp->writeDbs == NULL) {
     return -1;
   }
 
-  SCoder decoder = {0};
-  tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
-
-  if (tStartDecode(&decoder) < 0) return -1;
-  if (tDecodeCStrTo(&decoder, pRsp->user) < 0) return -1;
-  if (tDecodeI8(&decoder, &pRsp->superAuth) < 0) return -1;
+  if (tDecodeCStrTo(pDecoder, pRsp->user) < 0) return -1;
+  if (tDecodeI8(pDecoder, &pRsp->superAuth) < 0) return -1;
+  if (tDecodeI32(pDecoder, &pRsp->version) < 0) return -1;
 
+  int32_t numOfCreatedDbs = 0;
   int32_t numOfReadDbs = 0;
   int32_t numOfWriteDbs = 0;
-  if (tDecodeI32(&decoder, &numOfReadDbs) < 0) return -1;
-  if (tDecodeI32(&decoder, &numOfWriteDbs) < 0) return -1;
+  if (tDecodeI32(pDecoder, &numOfCreatedDbs) < 0) return -1;
+  if (tDecodeI32(pDecoder, &numOfReadDbs) < 0) return -1;
+  if (tDecodeI32(pDecoder, &numOfWriteDbs) < 0) return -1;
+
+  for (int32_t i = 0; i < numOfCreatedDbs; ++i) {
+    char db[TSDB_DB_FNAME_LEN] = {0};
+    if (tDecodeCStrTo(pDecoder, db) < 0) return -1;
+    int32_t len = strlen(db) + 1;
+    taosHashPut(pRsp->createdDbs, db, len, db, len);
+  }
 
   for (int32_t i = 0; i < numOfReadDbs; ++i) {
     char db[TSDB_DB_FNAME_LEN] = {0};
-    if (tDecodeCStrTo(&decoder, db) < 0) return -1;
+    if (tDecodeCStrTo(pDecoder, db) < 0) return -1;
     int32_t len = strlen(db) + 1;
     taosHashPut(pRsp->readDbs, db, len, db, len);
   }
 
   for (int32_t i = 0; i < numOfWriteDbs; ++i) {
     char db[TSDB_DB_FNAME_LEN] = {0};
-    if (tDecodeCStrTo(&decoder, db) < 0) return -1;
+    if (tDecodeCStrTo(pDecoder, db) < 0) return -1;
     int32_t len = strlen(db) + 1;
     taosHashPut(pRsp->writeDbs, db, len, db, len);
   }
 
+  return 0;
+}
+
+
+int32_t tDeserializeSGetUserAuthRsp(void *buf, int32_t bufLen, SGetUserAuthRsp *pRsp) {
+  SCoder decoder = {0};
+  tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+
+  if (tDeserializeSGetUserAuthRspImpl(&decoder, pRsp) < 0) return -1;
+
   tEndDecode(&decoder);
 
   tCoderClear(&decoder);
@@ -1217,6 +1252,7 @@ int32_t tDeserializeSGetUserAuthRsp(void *buf, int32_t bufLen, SGetUserAuthRsp *
 }
 
 void tFreeSGetUserAuthRsp(SGetUserAuthRsp *pRsp) {
+  taosHashCleanup(pRsp->createdDbs);
   taosHashCleanup(pRsp->readDbs);
   taosHashCleanup(pRsp->writeDbs);
 }
@@ -2036,6 +2072,62 @@ void tFreeSUseDbBatchRsp(SUseDbBatchRsp *pRsp) {
   taosArrayDestroy(pRsp->pArray);
 }
 
+int32_t tSerializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchRsp* pRsp){
+  SCoder encoder = {0};
+  tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+
+  int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
+  if (tEncodeI32(&encoder, numOfBatch) < 0) return -1;
+  for (int32_t i = 0; i < numOfBatch; ++i) {
+    SGetUserAuthRsp *pUserAuthRsp = taosArrayGet(pRsp->pArray, i);
+    if (tSerializeSGetUserAuthRspImpl(&encoder, pUserAuthRsp) < 0) return -1;
+  }
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tCoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSUserAuthBatchRsp(void* buf, int32_t bufLen, SUserAuthBatchRsp* pRsp){
+  SCoder decoder = {0};
+  tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+
+  int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
+  if (tDecodeI32(&decoder, &numOfBatch) < 0) return -1;
+
+  pRsp->pArray = taosArrayInit(numOfBatch, sizeof(SGetUserAuthRsp));
+  if (pRsp->pArray == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
+  for (int32_t i = 0; i < numOfBatch; ++i) {
+    SGetUserAuthRsp rsp = {0};
+    if (tDeserializeSGetUserAuthRspImpl(&decoder, &rsp) < 0) return -1;
+    taosArrayPush(pRsp->pArray, &rsp);
+  }
+  tEndDecode(&decoder);
+
+  tCoderClear(&decoder);
+  return 0;
+}
+
+void    tFreeSUserAuthBatchRsp(SUserAuthBatchRsp* pRsp){
+  int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
+  for (int32_t i = 0; i < numOfBatch; ++i) {
+    SGetUserAuthRsp *pUserAuthRsp = taosArrayGet(pRsp->pArray, i);
+    tFreeSGetUserAuthRsp(pUserAuthRsp);
+  }
+
+  taosArrayDestroy(pRsp->pArray);
+}
+
+
 int32_t tSerializeSDbCfgReq(void *buf, int32_t bufLen, SDbCfgReq *pReq) {
   SCoder encoder = {0};
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);

source/dnode/mnode/impl/inc/mndDef.h

@@ -255,6 +255,7 @@ typedef struct {
   int64_t   updateTime;
   int8_t    superUser;
   int32_t   acctId;
+  int32_t   authVersion;
   SHashObj* readDbs;
   SHashObj* writeDbs;
 } SUserObj;

source/dnode/mnode/impl/inc/mndUser.h

@@ -29,6 +29,7 @@ void      mndReleaseUser(SMnode *pMnode, SUserObj *pUser);
 
 // for trans test
 SSdbRaw *mndUserActionEncode(SUserObj *pUser);
+int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_t numOfUses, void **ppRsp, int32_t *pRspLen);
 
 #ifdef __cplusplus
 }

source/dnode/mnode/impl/src/mndProfile.c

@@ -403,6 +403,16 @@ static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHb
     SKv *kv = pIter;
 
     switch (kv->key) {
+      case HEARTBEAT_KEY_USER_AUTHINFO: {
+        void *  rspMsg = NULL;
+        int32_t rspLen = 0;
+        mndValidateUserAuthInfo(pMnode, kv->value, kv->valueLen / sizeof(SUserAuthVersion), &rspMsg, &rspLen);
+        if (rspMsg && rspLen > 0) {
+          SKv kv1 = {.key = HEARTBEAT_KEY_USER_AUTHINFO, .valueLen = rspLen, .value = rspMsg};
+          taosArrayPush(hbRsp.info, &kv1);
+        }
+        break;
+      }
       case HEARTBEAT_KEY_DBINFO: {
         void *  rspMsg = NULL;
         int32_t rspLen = 0;

source/dnode/mnode/impl/src/mndUser.c

@@ -451,13 +451,16 @@ static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       goto _OVER;
     }
+    newUser.authVersion++;
   } else if (alterReq.alterType == TSDB_ALTER_USER_REMOVE_READ_DB) {
     if (taosHashRemove(newUser.readDbs, alterReq.dbname, len) != 0) {
       terrno = TSDB_CODE_MND_DB_NOT_EXIST;
       goto _OVER;
     }
+    newUser.authVersion++;
   } else if (alterReq.alterType == TSDB_ALTER_USER_CLEAR_READ_DB) {
     taosHashClear(newUser.readDbs);
+    newUser.authVersion++;
   } else if (alterReq.alterType == TSDB_ALTER_USER_ADD_WRITE_DB) {
     if (pDb == NULL) {
       terrno = TSDB_CODE_MND_DB_NOT_EXIST;
@@ -467,13 +470,16 @@ static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       goto _OVER;
     }
+    newUser.authVersion++;
   } else if (alterReq.alterType == TSDB_ALTER_USER_REMOVE_WRITE_DB) {
     if (taosHashRemove(newUser.writeDbs, alterReq.dbname, len) != 0) {
       terrno = TSDB_CODE_MND_DB_NOT_EXIST;
       goto _OVER;
     }
+    newUser.authVersion++;
   } else if (alterReq.alterType == TSDB_ALTER_USER_CLEAR_WRITE_DB) {
     taosHashClear(newUser.writeDbs);
+    newUser.authVersion++;
   } else {
     terrno = TSDB_CODE_MND_INVALID_ALTER_OPER;
     goto _OVER;
@@ -576,6 +582,36 @@ _OVER:
   return code;
 }
 
+static int32_t mndSetUserAuthRsp(SMnode       *pMnode, SUserObj *pUser, SGetUserAuthRsp *pRsp) {
+  memcpy(pRsp->user, pUser->user, TSDB_USER_LEN);
+  pRsp->superAuth = pUser->superUser;
+  pRsp->version = pUser->authVersion;
+  pRsp->readDbs = mndDupDbHash(pUser->readDbs);
+  pRsp->writeDbs = mndDupDbHash(pUser->writeDbs);
+  pRsp->createdDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
+  if (NULL == pRsp->createdDbs) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+  
+  SSdb *pSdb = pMnode->pSdb;
+  void *pIter = NULL;
+  while (1) {
+    SDbObj *pDb = NULL;
+    pIter = sdbFetch(pSdb, SDB_DB, pIter, (void **)&pDb);
+    if (pIter == NULL) break;
+
+    if (strcmp(pDb->createUser, pUser->user) == 0) {
+      int32_t len = strlen(pDb->name) + 1;
+      taosHashPut(pRsp->createdDbs, pDb->name, len, pDb->name, len);
+    }
+
+    sdbRelease(pSdb, pDb);
+  }
+
+  return 0;
+}
+
 static int32_t mndProcessGetUserAuthReq(SNodeMsg *pReq) {
   SMnode         *pMnode = pReq->pNode;
   int32_t         code = -1;
@@ -596,25 +632,9 @@ static int32_t mndProcessGetUserAuthReq(SNodeMsg *pReq) {
     goto _OVER;
   }
 
-  memcpy(authRsp.user, pUser->user, TSDB_USER_LEN);
-  authRsp.superAuth = pUser->superUser;
-  authRsp.readDbs = mndDupDbHash(pUser->readDbs);
-  authRsp.writeDbs = mndDupDbHash(pUser->writeDbs);
-
-  SSdb *pSdb = pMnode->pSdb;
-  void *pIter = NULL;
-  while (1) {
-    SDbObj *pDb = NULL;
-    pIter = sdbFetch(pSdb, SDB_DB, pIter, (void **)&pDb);
-    if (pIter == NULL) break;
-
-    if (strcmp(pDb->createUser, pUser->user) == 0) {
-      int32_t len = strlen(pDb->name) + 1;
-      taosHashPut(authRsp.readDbs, pDb->name, len, pDb->name, len);
-      taosHashPut(authRsp.writeDbs, pDb->name, len, pDb->name, len);
-    }
-
-    sdbRelease(pSdb, pDb);
+  code = mndSetUserAuthRsp(pMnode, pUser, &authRsp);
+  if (code) {
+    goto _OVER;
   }
 
   int32_t contLen = tSerializeSGetUserAuthRsp(NULL, 0, &authRsp);
@@ -631,6 +651,7 @@ static int32_t mndProcessGetUserAuthReq(SNodeMsg *pReq) {
   code = 0;
 
 _OVER:
+  
   mndReleaseUser(pMnode, pUser);
   tFreeSGetUserAuthRsp(&authRsp);
 
@@ -681,3 +702,72 @@ static void mndCancelGetNextUser(SMnode *pMnode, void *pIter) {
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetch(pSdb, pIter);
 }
+
+int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_t numOfUses, void **ppRsp, int32_t *pRspLen) {
+  SUserAuthBatchRsp batchRsp = {0};
+  batchRsp.pArray = taosArrayInit(numOfUses, sizeof(SGetUserAuthRsp));
+  if (batchRsp.pArray == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
+  int32_t code = 0;
+  for (int32_t i = 0; i < numOfUses; ++i) {
+    SUserObj *pUser = mndAcquireUser(pMnode, pUsers[i].user);
+    if (pUser == NULL) {
+      mError("user:%s, failed to auth user since %s", pUsers[i].user, terrstr());
+      continue;
+    }
+
+    if (pUser->authVersion <= pUsers[i].version) {
+      mndReleaseUser(pMnode, pUser);
+      continue;
+    }
+    
+    SGetUserAuthRsp rsp = {0};
+    code = mndSetUserAuthRsp(pMnode, pUser, &rsp);
+    if (code) {
+      mndReleaseUser(pMnode, pUser);
+      tFreeSGetUserAuthRsp(&rsp);
+      goto _OVER;
+    }
+
+
+    taosArrayPush(batchRsp.pArray, &rsp);
+    mndReleaseUser(pMnode, pUser);
+  }
+
+  if (taosArrayGetSize(batchRsp.pArray) <= 0) {
+    *ppRsp = NULL;
+    *pRspLen = 0;
+    
+    tFreeSUserAuthBatchRsp(&batchRsp);
+    return 0;
+  }
+
+  int32_t rspLen = tSerializeSUserAuthBatchRsp(NULL, 0, &batchRsp);
+  void   *pRsp = taosMemoryMalloc(rspLen);
+  if (pRsp == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    tFreeSUserAuthBatchRsp(&batchRsp);
+    return -1;
+  }
+  tSerializeSUserAuthBatchRsp(pRsp, rspLen, &batchRsp);
+
+  *ppRsp = pRsp;
+  *pRspLen = rspLen;
+
+  tFreeSUserAuthBatchRsp(&batchRsp);
+  return 0;
+
+_OVER:
+
+  *ppRsp = NULL;
+  *pRspLen = 0;
+  
+  tFreeSUserAuthBatchRsp(&batchRsp);
+  return code;
+}
+
+
+

source/libs/catalog/inc/catalogInt.h

@@ -185,7 +185,7 @@ typedef struct SCtgRemoveTblMsg {
 typedef struct SCtgUpdateUserMsg {
   SCatalog* pCtg;
   SGetUserAuthRsp userAuth;
-} SCtgUpdateTblMsg;
+} SCtgUpdateUserMsg;
 
 
 typedef struct SCtgMetaAction {

source/libs/catalog/src/catalog.c

@@ -24,6 +24,7 @@ int32_t ctgActUpdateTbl(SCtgMetaAction *action);
 int32_t ctgActRemoveDB(SCtgMetaAction *action);
 int32_t ctgActRemoveStb(SCtgMetaAction *action);
 int32_t ctgActRemoveTbl(SCtgMetaAction *action);
+int32_t ctgActUpdateUser(SCtgMetaAction *action);
 
 extern SCtgDebug gCTGDebug;
 SCatalogMgmt gCtgMgmt = {0};
@@ -382,6 +383,7 @@ int32_t ctgPushUpdateUserMsgInQueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool
   
 _return:
 
+  tFreeSGetUserAuthRsp(pAuth);
   taosMemoryFreeClear(msg);
   
   CTG_RET(code);
@@ -925,7 +927,7 @@ int32_t ctgGetTableTypeFromCache(SCatalog* pCtg, const char* dbFName, const char
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t ctgGetUserDbAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, bool *inCache, int32_t *auth) {
+int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass) {
   if (NULL == pCtg->userCache) {
     ctgDebug("empty user auth cache, user:%s", user);
     goto _return;
@@ -943,23 +945,23 @@ int32_t ctgGetUserDbAuthFromCache(SCatalog* pCtg, const char* user, const char*
   CTG_CACHE_STAT_ADD(userHitNum, 1);
   
   if (pUser->superUser) {
-    CTG_FLAG_SET(auth, USER_AUTH_ALL);
+    *pass = true;
     return TSDB_CODE_SUCCESS;
   }
 
   CTG_LOCK(CTG_READ, &pUser->lock);
   if (pUser->createdDbs && taosHashGet(pUser->createdDbs, dbFName, strlen(dbFName))) {
-    CTG_FLAG_SET(auth, USER_AUTH_ALL);
+    *pass = true;
     CTG_UNLOCK(CTG_READ, &pUser->lock);
     return TSDB_CODE_SUCCESS;
   }
   
-  if (pUser->readDbs && taosHashGet(pUser->readDbs, dbFName, strlen(dbFName))) {
-    CTG_FLAG_SET(auth, USER_AUTH_READ);
+  if (pUser->readDbs && taosHashGet(pUser->readDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_READ) {
+    *pass = true;
   }
   
-  if (pUser->writeDbs && taosHashGet(pUser->writeDbs, dbFName, strlen(dbFName))) {
-    CTG_FLAG_SET(auth, USER_AUTH_WRITE);
+  if (pUser->writeDbs && taosHashGet(pUser->writeDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_WRITE) {
+    *pass = true;
   }
 
   CTG_UNLOCK(CTG_READ, &pUser->lock);
@@ -2067,12 +2069,13 @@ _return:
   CTG_RET(code);
 }
 
-int32_t ctgGetUserDbAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, int32_t* auth) {
+int32_t ctgChkAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, AUTH_TYPE type, bool *pass) {
   bool inCache = false;
   int32_t code = 0;
-  *auth = 0;
   
-  CTG_ERR_RET(ctgGetUserDbAuthFromCache(pCtg, user, dbFName, &inCache, auth));
+  *pass = false;
+  
+  CTG_ERR_RET(ctgChkAuthFromCache(pCtg, user, dbFName, type, &inCache, pass));
 
   if (inCache) {
     return TSDB_CODE_SUCCESS;
@@ -2082,21 +2085,21 @@ int32_t ctgGetUserDbAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, con
   CTG_ERR_RET(ctgGetUserDbAuthFromMnode(pCtg, pRpc, pMgmtEps, user, &authRsp));
   
   if (authRsp.superAuth) {
-    CTG_FLAG_SET(auth, USER_AUTH_ALL);
+    *pass = true;
     goto _return;
   }
 
   if (authRsp.createdDbs && taosHashGet(authRsp.createdDbs, dbFName, strlen(dbFName))) {
-    CTG_FLAG_SET(auth, USER_AUTH_ALL);
+    *pass = true;
     goto _return;
   }
 
-  if (authRsp.readDbs && taosHashGet(authRsp.readDbs, dbFName, strlen(dbFName))) {
-    CTG_FLAG_SET(auth, USER_AUTH_READ);
+  if (authRsp.readDbs && taosHashGet(authRsp.readDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_READ) {
+    *pass = true;
   }
 
-  if (authRsp.writeDbs && taosHashGet(authRsp.writeDbs, dbFName, strlen(dbFName))) {
-    CTG_FLAG_SET(auth, USER_AUTH_WRITE);
+  if (authRsp.writeDbs && taosHashGet(authRsp.writeDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_WRITE) {
+    *pass = true;
   }
 
 _return:
@@ -3050,6 +3053,35 @@ int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num)
   CTG_API_LEAVE(ctgMetaRentGet(&pCtg->dbRent, (void **)dbs, num, sizeof(SDbVgVersion)));
 }
 
+int32_t catalogGetExpiredUsers(SCatalog* pCtg, SUserAuthVersion **users, uint32_t *num) {
+  CTG_API_ENTER();
+  
+  if (NULL == pCtg || NULL == users || NULL == num) {
+    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
+  }
+
+  *num = taosHashGetSize(pCtg->userCache);
+  if (*num > 0) {
+    *users = taosMemoryCalloc(*num, sizeof(SUserAuthVersion));
+    if (NULL == *users) {
+      ctgError("calloc %d userAuthVersion failed", *num);
+      CTG_API_LEAVE(TSDB_CODE_OUT_OF_MEMORY);
+    }
+  }
+
+  uint32_t i = 0;
+  SCtgUserAuth *pAuth = taosHashIterate(pCtg->userCache, NULL);
+  while (pAuth != NULL) {
+    void *key = taosHashGetKey(pAuth, NULL);
+    strncpy((*users)[i].user, key, sizeof((*users)[i].user));
+    (*users)[i].version = pAuth->version;
+    pAuth = taosHashIterate(pCtg->userCache, pAuth);
+  }
+
+  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
+}
+
+
 int32_t catalogGetDBCfg(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, SDbCfgInfo* pDbCfg) {
   CTG_API_ENTER();
   
@@ -3094,21 +3126,31 @@ _return:
   CTG_API_LEAVE(code);
 }
 
-int32_t catalogGetUserDbAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, int32_t* auth) {
+int32_t catalogChkAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, AUTH_TYPE type, bool *pass) {
   CTG_API_ENTER();
   
-  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == user || NULL == dbFName || NULL == auth) {
+  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == user || NULL == dbFName || NULL == pass) {
     CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
   }
 
   int32_t code = 0;
-  CTG_ERR_JRET(ctgGetUserDbAuth(pCtg, pRpc, pMgmtEps, user, dbFName, auth));
+  CTG_ERR_JRET(ctgChkAuth(pCtg, pRpc, pMgmtEps, user, dbFName, type, pass));
   
 _return:
 
   CTG_API_LEAVE(code);
 }
 
+int32_t catalogUpdateUserAuthInfo(SCatalog* pCtg, SGetUserAuthRsp* pAuth) {
+  CTG_API_ENTER();
+
+  if (NULL == pCtg || NULL == pAuth) {
+    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
+  }
+
+  CTG_API_LEAVE(ctgPushUpdateUserMsgInQueue(pCtg, pAuth, false));
+}
+
 
 void catalogDestroy(void) {
   qInfo("start to destroy catalog");