diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 8741af54ec..ed744fa315 100644 --- a/include/common/tmsg.h +++ b/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; diff --git a/include/libs/catalog/catalog.h b/include/libs/catalog/catalog.h index 5fc9e08de4..04a24c4f32 100644 --- a/include/libs/catalog/catalog.h +++ b/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); /** diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index b11a49fa1a..fc39e80c1e 100644 --- a/source/client/src/clientHb.c +++ b/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; diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 6235dcf895..f49d0b02f6 100644 --- a/source/common/src/tmsg.c +++ b/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); diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index d516b0bf26..0ec8e5bd29 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/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; diff --git a/source/dnode/mnode/impl/inc/mndUser.h b/source/dnode/mnode/impl/inc/mndUser.h index b3eb7f2f95..2140d0fa67 100644 --- a/source/dnode/mnode/impl/inc/mndUser.h +++ b/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 } diff --git a/source/dnode/mnode/impl/src/mndProfile.c b/source/dnode/mnode/impl/src/mndProfile.c index 59d07fd4aa..2de337537f 100644 --- a/source/dnode/mnode/impl/src/mndProfile.c +++ b/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; diff --git a/source/dnode/mnode/impl/src/mndUser.c b/source/dnode/mnode/impl/src/mndUser.c index 5e15bdeb43..d2a9151167 100644 --- a/source/dnode/mnode/impl/src/mndUser.c +++ b/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; +} + + + diff --git a/source/libs/catalog/inc/catalogInt.h b/source/libs/catalog/inc/catalogInt.h index d6aa03aecb..3e8528e3d9 100644 --- a/source/libs/catalog/inc/catalogInt.h +++ b/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 { diff --git a/source/libs/catalog/src/catalog.c b/source/libs/catalog/src/catalog.c index 39a784d00c..f485f85809 100644 --- a/source/libs/catalog/src/catalog.c +++ b/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");