user auth

include/common/tmsg.h

@@ -424,7 +424,9 @@ int32_t tDeserializeSGetUserAuthReq(void* buf, int32_t bufLen, SGetUserAuthReq*
 
 typedef struct {
   char      user[TSDB_USER_LEN];
+  int32_t   version;
   int8_t    superAuth;
+  SHashObj* createdDbs;
   SHashObj* readDbs;
   SHashObj* writeDbs;
 } SGetUserAuthRsp;

include/libs/catalog/catalog.h

@@ -40,6 +40,9 @@ enum {
   CTG_DBG_STB_RENT_NUM,
 };
 
+#define USER_AUTH_READ  1
+#define USER_AUTH_WRITE 2
+#define USER_AUTH_ALL   4
 
 typedef struct SCatalogReq {
   SArray *pTableName;     // element is SNAME
@@ -57,6 +60,7 @@ typedef struct SMetaData {
 typedef struct SCatalogCfg {
   uint32_t maxTblCacheNum;
   uint32_t maxDBCacheNum;
+  uint32_t maxUserCacheNum;
   uint32_t dbRentSec;
   uint32_t stbRentSec;
 } SCatalogCfg;
@@ -225,6 +229,8 @@ int32_t catalogGetIndexInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps,
 
 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);
+
 
 /**
  * Destroy catalog and relase all resources

source/libs/catalog/inc/catalogInt.h

@@ -54,6 +54,7 @@ enum {
   CTG_ACT_REMOVE_DB,
   CTG_ACT_REMOVE_STB,
   CTG_ACT_REMOVE_TBL,
+  CTG_ACT_UPDATE_USER,
   CTG_ACT_MAX
 };
 
@@ -95,8 +96,18 @@ typedef struct SCtgRentMgmt {
   SCtgRentSlot  *slots;
 } SCtgRentMgmt;
 
+typedef struct SCtgUserAuth {
+  int32_t   version;
+  SRWLatch  lock;
+  bool      superUser;
+  SHashObj *createdDbs;
+  SHashObj *readDbs;
+  SHashObj *writeDbs;
+} SCtgUserAuth;
+
 typedef struct SCatalog {
   uint64_t         clusterId;  
+  SHashObj        *userCache;    //key:user, value:SCtgUserAuth
   SHashObj        *dbCache;      //key:dbname, value:SCtgDBCache
   SCtgRentMgmt     dbRent;
   SCtgRentMgmt     stbRent;
@@ -124,6 +135,8 @@ typedef struct SCtgCacheStat {
   uint64_t vgMissNum;
   uint64_t tblHitNum;
   uint64_t tblMissNum;
+  uint64_t userHitNum;
+  uint64_t userMissNum;
 } SCtgCacheStat;
 
 typedef struct SCatalogStat {
@@ -169,6 +182,11 @@ typedef struct SCtgRemoveTblMsg {
   uint64_t dbId;
 } SCtgRemoveTblMsg;
 
+typedef struct SCtgUpdateUserMsg {
+  SCatalog* pCtg;
+  SGetUserAuthRsp userAuth;
+} SCtgUpdateTblMsg;
+
 
 typedef struct SCtgMetaAction {
   int32_t  act;
@@ -234,6 +252,8 @@ typedef struct SCtgAction {
 #define CTG_FLAG_SYS_DB       0x8
 #define CTG_FLAG_FORCE_UPDATE 0x10
 
+#define CTG_FLAG_SET(_flag, _v) ((_flag) |= (_v))
+
 #define CTG_FLAG_IS_STB(_flag) ((_flag) & CTG_FLAG_STB)
 #define CTG_FLAG_IS_NOT_STB(_flag) ((_flag) & CTG_FLAG_NOT_STB)
 #define CTG_FLAG_IS_UNKNOWN_STB(_flag) ((_flag) & CTG_FLAG_UNKNOWN_STB)

source/libs/catalog/src/catalog.c

@@ -51,6 +51,11 @@ SCtgAction gCtgAction[CTG_ACT_MAX] = {{
                             CTG_ACT_REMOVE_TBL,
                             "remove tbMeta",
                             ctgActRemoveTbl
+                          },
+                          {
+                            CTG_ACT_UPDATE_USER,
+                            "update user",
+                            ctgActUpdateUser
                           }
 };
 
@@ -357,6 +362,30 @@ _return:
   CTG_RET(code);
 }
 
+int32_t ctgPushUpdateUserMsgInQueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq) {
+  int32_t code = 0;
+  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_USER, .syncReq = syncReq};
+  SCtgUpdateUserMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateUserMsg));
+  if (NULL == msg) {
+    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateUserMsg));
+    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
+  }
+
+  msg->pCtg = pCtg;
+  msg->userAuth = *pAuth;
+
+  action.data = msg;
+
+  CTG_ERR_JRET(ctgPushAction(pCtg, &action));
+
+  return TSDB_CODE_SUCCESS;
+  
+_return:
+
+  taosMemoryFreeClear(msg);
+  
+  CTG_RET(code);
+}
 
 int32_t ctgAcquireVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache, bool *inCache) {
   CTG_LOCK(CTG_READ, &dbCache->vgLock);
@@ -687,6 +716,43 @@ int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEp
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char *user, SGetUserAuthRsp *authRsp) {
+  char *msg = NULL;
+  int32_t msgLen = 0;
+
+  ctgDebug("try to get user auth from mnode, user:%s", user);
+
+  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)]((void *)user, &msg, 0, &msgLen);
+  if (code) {
+    ctgError("Build get user auth msg failed, code:%x, db:%s", code, user);
+    CTG_ERR_RET(code);
+  }
+  
+  SRpcMsg rpcMsg = {
+      .msgType = TDMT_MND_GET_USER_AUTH,
+      .pCont   = msg,
+      .contLen = msgLen,
+  };
+
+  SRpcMsg rpcRsp = {0};
+
+  rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
+  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
+    ctgError("error rsp for get user auth, error:%s, user:%s", tstrerror(rpcRsp.code), user);
+    CTG_ERR_RET(rpcRsp.code);
+  }
+
+  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)](authRsp, rpcRsp.pCont, rpcRsp.contLen);
+  if (code) {
+    ctgError("Process get user auth rsp failed, code:%x, user:%s", code, user);
+    CTG_ERR_RET(code);
+  }
+
+  ctgDebug("Got user auth from mnode, user:%s", user);
+
+  return TSDB_CODE_SUCCESS;
+}
+
 
 int32_t ctgIsTableMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist) {
   if (NULL == pCtg->dbCache) {
@@ -859,6 +925,55 @@ 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) {
+  if (NULL == pCtg->userCache) {
+    ctgDebug("empty user auth cache, user:%s", user);
+    goto _return;
+  }
+  
+  SCtgUserAuth *pUser = (SCtgUserAuth *)taosHashGet(pCtg->userCache, user, strlen(user));
+  if (NULL == pUser) {
+    ctgDebug("user not in cache, user:%s", user);
+    goto _return;
+  }
+
+  *inCache = true;
+
+  ctgDebug("Got user from cache, user:%s", user);
+  CTG_CACHE_STAT_ADD(userHitNum, 1);
+  
+  if (pUser->superUser) {
+    CTG_FLAG_SET(auth, USER_AUTH_ALL);
+    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);
+    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->writeDbs && taosHashGet(pUser->writeDbs, dbFName, strlen(dbFName))) {
+    CTG_FLAG_SET(auth, USER_AUTH_WRITE);
+  }
+
+  CTG_UNLOCK(CTG_READ, &pUser->lock);
+  
+  return TSDB_CODE_SUCCESS;
+
+_return:
+
+  *inCache = false;
+  CTG_CACHE_STAT_ADD(userMissNum, 1);
+  
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t ctgGetTableMetaFromMnodeImpl(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, char *dbFName, char* tbName, STableMetaOutput* output) {
   SBuildTableMetaInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = tbName};
   char *msg = NULL;
@@ -1952,6 +2067,44 @@ _return:
   CTG_RET(code);
 }
 
+int32_t ctgGetUserDbAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, int32_t* auth) {
+  bool inCache = false;
+  int32_t code = 0;
+  *auth = 0;
+  
+  CTG_ERR_RET(ctgGetUserDbAuthFromCache(pCtg, user, dbFName, &inCache, auth));
+
+  if (inCache) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SGetUserAuthRsp authRsp = {0};
+  CTG_ERR_RET(ctgGetUserDbAuthFromMnode(pCtg, pRpc, pMgmtEps, user, &authRsp));
+  
+  if (authRsp.superAuth) {
+    CTG_FLAG_SET(auth, USER_AUTH_ALL);
+    goto _return;
+  }
+
+  if (authRsp.createdDbs && taosHashGet(authRsp.createdDbs, dbFName, strlen(dbFName))) {
+    CTG_FLAG_SET(auth, USER_AUTH_ALL);
+    goto _return;
+  }
+
+  if (authRsp.readDbs && taosHashGet(authRsp.readDbs, dbFName, strlen(dbFName))) {
+    CTG_FLAG_SET(auth, USER_AUTH_READ);
+  }
+
+  if (authRsp.writeDbs && taosHashGet(authRsp.writeDbs, dbFName, strlen(dbFName))) {
+    CTG_FLAG_SET(auth, USER_AUTH_WRITE);
+  }
+
+_return:
+
+  ctgPushUpdateUserMsgInQueue(pCtg, &authRsp, false);
+
+  return TSDB_CODE_SUCCESS;
+}
 
 
 int32_t ctgActUpdateVg(SCtgMetaAction *action) {
@@ -2121,6 +2274,67 @@ _return:
   CTG_RET(code);
 }
 
+int32_t ctgActUpdateUser(SCtgMetaAction *action) {
+  int32_t code = 0;
+  SCtgUpdateUserMsg *msg = action->data;
+  SCatalog* pCtg = msg->pCtg;
+
+  if (NULL == pCtg->userCache) {
+    pCtg->userCache = taosHashInit(gCtgMgmt.cfg.maxUserCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
+    if (NULL == pCtg->userCache) {
+      ctgError("taosHashInit %d user cache failed", gCtgMgmt.cfg.maxUserCacheNum);
+      CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+  }
+  
+  SCtgUserAuth *pUser = (SCtgUserAuth *)taosHashGet(pCtg->userCache, msg->userAuth.user, strlen(msg->userAuth.user));
+  if (NULL == pUser) {
+    SCtgUserAuth userAuth = {0};
+
+    userAuth.version = msg->userAuth.version;
+    userAuth.superUser = msg->userAuth.superAuth;
+    userAuth.createdDbs = msg->userAuth.createdDbs;
+    userAuth.readDbs = msg->userAuth.readDbs;
+    userAuth.writeDbs = msg->userAuth.writeDbs;
+
+    if (taosHashPut(pCtg->userCache, msg->userAuth.user, strlen(msg->userAuth.user), &userAuth, sizeof(userAuth))) {
+      ctgError("taosHashPut user %s to cache failed", msg->userAuth.user);
+      CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
+    }
+
+    return TSDB_CODE_SUCCESS;
+  }
+
+  pUser->version = msg->userAuth.version;
+
+  CTG_LOCK(CTG_WRITE, &pUser->lock);
+
+  taosHashCleanup(pUser->createdDbs);
+  pUser->createdDbs = msg->userAuth.createdDbs;
+  msg->userAuth.createdDbs = NULL;
+
+  taosHashCleanup(pUser->readDbs);
+  pUser->readDbs = msg->userAuth.readDbs;
+  msg->userAuth.readDbs = NULL;
+
+  taosHashCleanup(pUser->writeDbs);
+  pUser->writeDbs = msg->userAuth.writeDbs;
+  msg->userAuth.writeDbs = NULL;
+
+  CTG_UNLOCK(CTG_WRITE, &pUser->lock);
+
+_return:
+
+
+  taosHashCleanup(msg->userAuth.createdDbs);
+  taosHashCleanup(msg->userAuth.readDbs);
+  taosHashCleanup(msg->userAuth.writeDbs);
+ 
+  taosMemoryFreeClear(msg);
+  
+  CTG_RET(code);
+}
+
 
 void* ctgUpdateThreadFunc(void* param) {
   setThreadName("catalog");
@@ -2880,6 +3094,21 @@ _return:
   CTG_API_LEAVE(code);
 }
 
+int32_t catalogGetUserDbAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, int32_t* auth) {
+  CTG_API_ENTER();
+  
+  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == user || NULL == dbFName || NULL == auth) {
+    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
+  }
+
+  int32_t code = 0;
+  CTG_ERR_JRET(ctgGetUserDbAuth(pCtg, pRpc, pMgmtEps, user, dbFName, auth));
+  
+_return:
+
+  CTG_API_LEAVE(code);
+}
+
 
 void catalogDestroy(void) {
   qInfo("start to destroy catalog");

source/libs/qcom/src/querymsg.c

@@ -181,6 +181,25 @@ int32_t queryBuildRetrieveFuncMsg(void *input, char **msg, int32_t msgSize, int3
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t queryBuildGetUserAuthMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) {
+  if (NULL == msg || NULL == msgLen) {
+    return TSDB_CODE_TSC_INVALID_INPUT;
+  }
+
+  SGetUserAuthReq req = {0};
+  strncpy(req.user, input, sizeof(req.user));
+
+  int32_t bufLen = tSerializeSGetUserAuthReq(NULL, 0, &req);
+  void   *pBuf = rpcMallocCont(bufLen);
+  tSerializeSGetUserAuthReq(pBuf, bufLen, &req);
+
+  *msg = pBuf;
+  *msgLen = bufLen;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 int32_t queryProcessUseDBRsp(void *output, char *msg, int32_t msgSize) {
   SUseDbOutput *pOut = output;
   SUseDbRsp     usedbRsp = {0};
@@ -419,6 +438,20 @@ int32_t queryProcessRetrieveFuncRsp(void *output, char *msg, int32_t msgSize) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t queryProcessGetUserAuthRsp(void *output, char *msg, int32_t msgSize) {
+  if (NULL == output || NULL == msg || msgSize <= 0) {
+    return TSDB_CODE_TSC_INVALID_INPUT;
+  }
+
+  if (tDeserializeSGetUserAuthRsp(msg, msgSize, (SGetUserAuthRsp *)output) != 0) {
+    qError("tDeserializeSGetUserAuthRsp failed, msgSize:%d", msgSize);
+    return TSDB_CODE_INVALID_MSG;
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 void initQueryModuleMsgHandle() {
   queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_META)] = queryBuildTableMetaReqMsg;
   queryBuildMsg[TMSG_INDEX(TDMT_MND_TABLE_META)] = queryBuildTableMetaReqMsg;
@@ -427,6 +460,8 @@ void initQueryModuleMsgHandle() {
   queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_DB_CFG)] = queryBuildGetDBCfgMsg;
   queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_INDEX)]  = queryBuildGetIndexMsg;
   queryBuildMsg[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)]  = queryBuildRetrieveFuncMsg;
+  queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)]  = queryBuildGetUserAuthMsg;
+  
 
   queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_META)] = queryProcessTableMetaRsp;
   queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_META)] = queryProcessTableMetaRsp;
@@ -435,6 +470,7 @@ void initQueryModuleMsgHandle() {
   queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_DB_CFG)] = queryProcessGetDbCfgRsp;
   queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_INDEX)]  = queryProcessGetIndexRsp;
   queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)]  = queryProcessRetrieveFuncRsp;
+  queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)]  = queryProcessGetUserAuthRsp;
 }
 
 #pragma GCC diagnostic pop