Compare commits

...

5 Commits

8 changed files with 265 additions and 64 deletions

View File

@ -159,15 +159,17 @@ func (s *authServer) parseToken(ctx context.Context, tokensString string) (claim
if err != nil {
return nil, err
}
isAdmin := authverify.IsManagerUserID(claims.UserID, s.config.Share.IMAdminUserID)
if isAdmin {
return claims, nil
}
m, err := s.authDatabase.GetTokensWithoutError(ctx, claims.UserID, claims.PlatformID)
if err != nil {
return nil, err
}
if len(m) == 0 {
isAdmin := authverify.IsManagerUserID(claims.UserID, s.config.Share.IMAdminUserID)
if isAdmin {
if err = s.authDatabase.GetTemporaryTokensWithoutError(ctx, claims.UserID, claims.PlatformID, tokensString); err == nil {
return claims, nil
}
}
return nil, servererrs.ErrTokenNotExist.Wrap()
}
if v, ok := m[tokensString]; ok {
@ -179,6 +181,13 @@ func (s *authServer) parseToken(ctx context.Context, tokensString string) (claim
default:
return nil, errs.Wrap(errs.ErrTokenUnknown)
}
} else {
isAdmin := authverify.IsManagerUserID(claims.UserID, s.config.Share.IMAdminUserID)
if isAdmin {
if err = s.authDatabase.GetTemporaryTokensWithoutError(ctx, claims.UserID, claims.PlatformID, tokensString); err == nil {
return claims, nil
}
}
}
return nil, servererrs.ErrTokenNotExist.Wrap()
}

View File

@ -17,6 +17,8 @@ package msg
import (
"context"
"encoding/json"
"strconv"
"strings"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
@ -49,38 +51,94 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
if err != nil {
return nil, err
}
// Get message and add fault tolerance handling
_, _, msgs, err := m.MsgDatabase.GetMsgBySeqs(ctx, req.UserID, req.ConversationID, []int64{req.Seq})
if err != nil {
return nil, err
}
if len(msgs) == 0 || msgs[0] == nil {
return nil, errs.ErrRecordNotFound.WrapMsg("msg not found")
}
if msgs[0].ContentType == constant.MsgRevokeNotification {
return nil, servererrs.ErrMsgAlreadyRevoke.WrapMsg("msg already revoke")
// Log the error but continue execution
log.ZWarn(ctx, "GetMsgBySeqs error when revoking message", err,
"userID", req.UserID, "conversationID", req.ConversationID, "seq", req.Seq)
} else if len(msgs) == 0 || msgs[0] == nil {
// Check if seq is within valid range for the current conversation
maxSeq, err := m.MsgDatabase.GetMaxSeq(ctx, req.ConversationID)
if err != nil {
log.ZWarn(ctx, "GetMaxSeq error when revoking message", err,
"conversationID", req.ConversationID)
} else if req.Seq > maxSeq {
return nil, errs.ErrArgs.WrapMsg("seq exceeds maxSeq")
}
// Log warning but continue execution
log.ZWarn(ctx, "Message not found when revoking, but will proceed", nil,
"userID", req.UserID, "conversationID", req.ConversationID, "seq", req.Seq)
}
data, _ := json.Marshal(msgs[0])
// If message doesn't exist, create a minimal substitute for revocation notification
var msgToRevoke *sdkws.MsgData
if len(msgs) == 0 || msgs[0] == nil {
// Create a minimal message object with necessary fields
msgToRevoke = &sdkws.MsgData{
SendID: req.UserID, // Use revoker as sender
Seq: req.Seq,
SessionType: getSessionTypeFromConversationID(req.ConversationID), // Helper function to get session type
ClientMsgID: "missing_" + strconv.FormatInt(req.Seq, 10), // Generate a temporary ID
SendTime: time.Now().UnixMilli() - 1000, // Set to a slightly earlier time
}
// Set GroupID or RecvID based on session type
if msgToRevoke.SessionType == constant.ReadGroupChatType {
// Extract group ID from conversation ID
if strings.HasPrefix(req.ConversationID, "sg_") {
msgToRevoke.GroupID = req.ConversationID[3:] // Remove "sg_" prefix
}
} else {
// For single chat, parse receiver ID from conversation ID
if strings.HasPrefix(req.ConversationID, "si_") || strings.HasPrefix(req.ConversationID, "sp_") {
parts := strings.Split(req.ConversationID[3:], "_")
if len(parts) == 2 {
// Conversation ID format is typically: si_senderID_receiverID
// If current user is the sender, then receiver is the other party
if parts[0] == req.UserID {
msgToRevoke.RecvID = parts[1]
} else {
msgToRevoke.RecvID = parts[0]
}
} else {
// Set empty if parsing fails
msgToRevoke.RecvID = ""
}
} else {
msgToRevoke.RecvID = ""
}
}
} else {
msgToRevoke = msgs[0]
if msgToRevoke.ContentType == constant.MsgRevokeNotification {
return nil, servererrs.ErrMsgAlreadyRevoke.WrapMsg("msg already revoke")
}
}
data, _ := json.Marshal(msgToRevoke)
log.ZDebug(ctx, "GetMsgBySeqs", "conversationID", req.ConversationID, "seq", req.Seq, "msg", string(data))
var role int32
if !authverify.IsAppManagerUid(ctx, m.config.Share.IMAdminUserID) {
sessionType := msgs[0].SessionType
sessionType := msgToRevoke.SessionType
switch sessionType {
case constant.SingleChatType:
if err := authverify.CheckAccessV3(ctx, msgs[0].SendID, m.config.Share.IMAdminUserID); err != nil {
if err := authverify.CheckAccessV3(ctx, msgToRevoke.SendID, m.config.Share.IMAdminUserID); err != nil {
return nil, err
}
role = user.AppMangerLevel
case constant.ReadGroupChatType:
members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, msgs[0].GroupID, datautil.Distinct([]string{req.UserID, msgs[0].SendID}))
members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, msgToRevoke.GroupID, datautil.Distinct([]string{req.UserID, msgToRevoke.SendID}))
if err != nil {
return nil, err
}
if req.UserID != msgs[0].SendID {
if req.UserID != msgToRevoke.SendID {
switch members[req.UserID].RoleLevel {
case constant.GroupOwner:
case constant.GroupAdmin:
if sendMember, ok := members[msgs[0].SendID]; ok {
if sendMember, ok := members[msgToRevoke.SendID]; ok {
if sendMember.RoleLevel != constant.GroupOrdinaryUsers {
return nil, errs.ErrNoPermission.WrapMsg("no permission")
}
@ -114,20 +172,31 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
}
tips := sdkws.RevokeMsgTips{
RevokerUserID: revokerUserID,
ClientMsgID: msgs[0].ClientMsgID,
ClientMsgID: msgToRevoke.ClientMsgID,
RevokeTime: now,
Seq: req.Seq,
SesstionType: msgs[0].SessionType,
SesstionType: msgToRevoke.SessionType,
ConversationID: req.ConversationID,
IsAdminRevoke: flag,
}
var recvID string
if msgs[0].SessionType == constant.ReadGroupChatType {
recvID = msgs[0].GroupID
if msgToRevoke.SessionType == constant.ReadGroupChatType {
recvID = msgToRevoke.GroupID
} else {
recvID = msgs[0].RecvID
recvID = msgToRevoke.RecvID
}
m.notificationSender.NotificationWithSessionType(ctx, req.UserID, recvID, constant.MsgRevokeNotification, msgs[0].SessionType, &tips)
m.notificationSender.NotificationWithSessionType(ctx, req.UserID, recvID, constant.MsgRevokeNotification, msgToRevoke.SessionType, &tips)
m.webhookAfterRevokeMsg(ctx, &m.config.WebhooksConfig.AfterRevokeMsg, req)
return &msg.RevokeMsgResp{}, nil
}
func getSessionTypeFromConversationID(conversationID string) int32 {
// Conversation ID format is typically: "single chat prefix{userID}" or "group chat prefix{groupID}"
if strings.HasPrefix(conversationID, "sp_") || strings.HasPrefix(conversationID, "si_") {
return constant.SingleChatType
} else if strings.HasPrefix(conversationID, "sg_") {
return constant.ReadGroupChatType
}
// Default to single chat type
return constant.SingleChatType
}

View File

@ -1,8 +1,9 @@
package cachekey
import (
"github.com/openimsdk/protocol/constant"
"strings"
"github.com/openimsdk/protocol/constant"
)
const (
@ -13,6 +14,10 @@ func GetTokenKey(userID string, platformID int) string {
return UidPidToken + userID + ":" + constant.PlatformIDToName(platformID)
}
func GetTemporaryTokenKey(userID string, platformID int, token string) string {
return UidPidToken + ":TEMPORARY:" + userID + ":" + constant.PlatformIDToName(platformID) + ":" + token
}
func GetAllPlatformTokenKey(userID string) []string {
res := make([]string, len(constant.PlatformID2Name))
for k := range constant.PlatformID2Name {

View File

@ -27,7 +27,6 @@ type tokenCache struct {
func (x *tokenCache) getTokenKey(userID string, platformID int, token string) string {
return cachekey.GetTokenKey(userID, platformID) + ":" + token
}
func (x *tokenCache) SetTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error {
@ -57,6 +56,14 @@ func (x *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, p
return mm, nil
}
func (x *tokenCache) HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) error {
key := cachekey.GetTemporaryTokenKey(userID, platformID, token)
if _, err := x.cache.Get(ctx, []string{key}); err != nil {
return err
}
return nil
}
func (x *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) {
prefix := cachekey.UidPidToken + userID + ":"
tokens, err := x.cache.Prefix(ctx, prefix)
@ -128,3 +135,32 @@ func (x *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, pla
func (x *tokenCache) getExpireTime(t int64) time.Duration {
return time.Hour * 24 * time.Duration(t)
}
func (x *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error {
keys := make([]string, 0, len(tokens))
for platformID, ts := range tokens {
for _, t := range ts {
keys = append(keys, x.getTokenKey(userID, platformID, t))
}
}
return x.cache.Del(ctx, keys)
}
func (x *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error {
keys := make([]string, 0, len(fields))
for _, f := range fields {
keys = append(keys, x.getTokenKey(userID, platformID, f))
}
if err := x.cache.Del(ctx, keys); err != nil {
return err
}
for _, f := range fields {
k := cachekey.GetTemporaryTokenKey(userID, platformID, f)
if err := x.cache.Set(ctx, k, "", time.Minute*5); err != nil {
return errs.Wrap(err)
}
}
return nil
}

View File

@ -9,6 +9,7 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
"github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/utils/datautil"
"github.com/redis/go-redis/v9"
)
@ -55,6 +56,14 @@ func (c *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, p
return mm, nil
}
func (c *tokenCache) HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) error {
err := c.rdb.Get(ctx, cachekey.GetTemporaryTokenKey(userID, platformID, token)).Err()
if err != nil {
return errs.Wrap(err)
}
return nil
}
func (c *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) {
var (
res = make(map[int]map[string]int)
@ -101,13 +110,19 @@ func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, pla
}
func (c *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error {
pipe := c.rdb.Pipeline()
for k, v := range tokens {
pipe.HSet(ctx, k, v)
}
_, err := pipe.Exec(ctx)
if err != nil {
return errs.Wrap(err)
keys := datautil.Keys(tokens)
if err := ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
pipe := c.rdb.Pipeline()
for k, v := range tokens {
pipe.HSet(ctx, k, v)
}
_, err := pipe.Exec(ctx)
if err != nil {
return errs.Wrap(err)
}
return nil
}); err != nil {
return err
}
return nil
}
@ -119,3 +134,47 @@ func (c *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, pla
func (c *tokenCache) getExpireTime(t int64) time.Duration {
return time.Hour * 24 * time.Duration(t)
}
// DeleteTokenByTokenMap tokens key is platformID, value is token slice
func (c *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error {
var (
keys = make([]string, 0, len(tokens))
keyMap = make(map[string][]string)
)
for k, v := range tokens {
k1 := cachekey.GetTokenKey(userID, k)
keys = append(keys, k1)
keyMap[k1] = v
}
if err := ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
pipe := c.rdb.Pipeline()
for k, v := range tokens {
pipe.HDel(ctx, cachekey.GetTokenKey(userID, k), v...)
}
_, err := pipe.Exec(ctx)
if err != nil {
return errs.Wrap(err)
}
return nil
}); err != nil {
return err
}
return nil
}
func (c *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error {
key := cachekey.GetTokenKey(userID, platformID)
if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil {
return errs.Wrap(err)
}
for _, f := range fields {
k := cachekey.GetTemporaryTokenKey(userID, platformID, f)
if err := c.rdb.Set(ctx, k, "", time.Minute*5).Err(); err != nil {
return errs.Wrap(err)
}
}
return nil
}

View File

@ -9,8 +9,11 @@ type TokenModel interface {
// SetTokenFlagEx set token and flag with expire time
SetTokenFlagEx(ctx context.Context, userID string, platformID int, token string, flag int) error
GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) error
GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error)
SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error
DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error
DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error
DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error
}

View File

@ -17,6 +17,8 @@ import (
type AuthDatabase interface {
// If the result is empty, no error is returned.
GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
GetTemporaryTokensWithoutError(ctx context.Context, userID string, platformID int, token string) error
// Create token
CreateToken(ctx context.Context, userID string, platformID int) (string, error)
@ -52,6 +54,10 @@ func (a *authDatabase) GetTokensWithoutError(ctx context.Context, userID string,
return a.cache.GetTokensWithoutError(ctx, userID, platformID)
}
func (a *authDatabase) GetTemporaryTokensWithoutError(ctx context.Context, userID string, platformID int, token string) error {
return a.cache.HasTemporaryToken(ctx, userID, platformID, token)
}
func (a *authDatabase) SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error {
return a.cache.SetTokenMapByUidPid(ctx, userID, platformID, m)
}
@ -85,23 +91,30 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
return "", err
}
deleteTokenKey, kickedTokenKey, err := a.checkToken(ctx, tokens, platformID)
deleteTokenKey, kickedTokenKey, adminTokens, err := a.checkToken(ctx, tokens, platformID)
if err != nil {
return "", err
}
if len(deleteTokenKey) != 0 {
err = a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey)
err = a.cache.DeleteTokenByTokenMap(ctx, userID, deleteTokenKey)
if err != nil {
return "", err
}
}
if len(kickedTokenKey) != 0 {
for _, k := range kickedTokenKey {
err := a.cache.SetTokenFlagEx(ctx, userID, platformID, k, constant.KickedToken)
if err != nil {
return "", err
for plt, ks := range kickedTokenKey {
for _, k := range ks {
err := a.cache.SetTokenFlagEx(ctx, userID, plt, k, constant.KickedToken)
if err != nil {
return "", err
}
log.ZDebug(ctx, "kicked token in create token", "token", k)
}
log.ZDebug(ctx, "kicked token in create token", "token", k)
}
}
if len(adminTokens) != 0 {
if err = a.cache.DeleteAndSetTemporary(ctx, userID, constant.AdminPlatformID, adminTokens); err != nil {
return "", err
}
}
@ -119,12 +132,13 @@ func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformI
return tokenString, nil
}
func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string]int, platformID int) ([]string, []string, error) {
// todo: Move the logic for handling old data to another location.
// checkToken will check token by tokenPolicy and return deleteToken,kickToken,deleteAdminToken
func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string]int, platformID int) (map[int][]string, map[int][]string, []string, error) {
// todo: Asynchronous deletion of old data.
var (
loginTokenMap = make(map[int][]string) // The length of the value of the map must be greater than 0
deleteToken = make([]string, 0)
kickToken = make([]string, 0)
deleteToken = make(map[int][]string)
kickToken = make(map[int][]string)
adminToken = make([]string, 0)
unkickTerminal = ""
)
@ -133,7 +147,7 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
for k, v := range tks {
_, err := tokenverify.GetClaimFromToken(k, authverify.Secret(a.accessSecret))
if err != nil || v != constant.NormalToken {
deleteToken = append(deleteToken, k)
deleteToken[plfID] = append(deleteToken[plfID], k)
} else {
if plfID != constant.AdminPlatformID {
loginTokenMap[plfID] = append(loginTokenMap[plfID], k)
@ -153,14 +167,15 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
}
limit := a.multiLogin.MaxNumOneEnd
if l > limit {
kickToken = append(kickToken, ts[:l-limit]...)
kickToken[plt] = ts[:l-limit]
}
}
case constant.AllLoginButSameTermKick:
for plt, ts := range loginTokenMap {
kickToken = append(kickToken, ts[:len(ts)-1]...)
kickToken[plt] = ts[:len(ts)-1]
if plt == platformID {
kickToken = append(kickToken, ts[len(ts)-1])
kickToken[plt] = append(kickToken[plt], ts[len(ts)-1])
}
}
case constant.PCAndOther:
@ -168,29 +183,33 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
if constant.PlatformIDToClass(platformID) != unkickTerminal {
for plt, ts := range loginTokenMap {
if constant.PlatformIDToClass(plt) != unkickTerminal {
kickToken = append(kickToken, ts...)
kickToken[plt] = ts
}
}
} else {
var (
preKick []string
isReserve = true
preKickToken string
preKickPlt int
reserveToken = false
)
for plt, ts := range loginTokenMap {
if constant.PlatformIDToClass(plt) != unkickTerminal {
// Keep a token from another end
if isReserve {
isReserve = false
kickToken = append(kickToken, ts[:len(ts)-1]...)
preKick = append(preKick, ts[len(ts)-1])
if !reserveToken {
reserveToken = true
kickToken[plt] = ts[:len(ts)-1]
preKickToken = ts[len(ts)-1]
preKickPlt = plt
continue
} else {
// Prioritize keeping Android
if plt == constant.AndroidPlatformID {
kickToken = append(kickToken, preKick...)
kickToken = append(kickToken, ts[:len(ts)-1]...)
if preKickToken != "" {
kickToken[preKickPlt] = append(kickToken[preKickPlt], preKickToken)
}
kickToken[plt] = ts[:len(ts)-1]
} else {
kickToken = append(kickToken, ts...)
kickToken[plt] = ts
}
}
}
@ -203,19 +222,19 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
for plt, ts := range loginTokenMap {
if constant.PlatformIDToClass(plt) == constant.PlatformIDToClass(platformID) {
kickToken = append(kickToken, ts...)
kickToken[plt] = ts
} else {
if _, ok := reserved[constant.PlatformIDToClass(plt)]; !ok {
reserved[constant.PlatformIDToClass(plt)] = struct{}{}
kickToken = append(kickToken, ts[:len(ts)-1]...)
kickToken[plt] = ts[:len(ts)-1]
continue
} else {
kickToken = append(kickToken, ts...)
kickToken[plt] = ts
}
}
}
default:
return nil, nil, errs.New("unknown multiLogin policy").Wrap()
return nil, nil, nil, errs.New("unknown multiLogin policy").Wrap()
}
//var adminTokenMaxNum = a.multiLogin.MaxNumOneEnd
@ -226,8 +245,9 @@ func (a *authDatabase) checkToken(ctx context.Context, tokens map[int]map[string
//if l > adminTokenMaxNum {
// kickToken = append(kickToken, adminToken[:l-adminTokenMaxNum]...)
//}
var deleteAdminToken []string
if platformID == constant.AdminPlatformID {
kickToken = append(kickToken, adminToken...)
deleteAdminToken = adminToken
}
return deleteToken, kickToken, nil
return deleteToken, kickToken, deleteAdminToken, nil
}

View File

@ -127,7 +127,7 @@ func (x *CacheMgo) Del(ctx context.Context, key []string) error {
return nil
}
_, err := x.coll.DeleteMany(ctx, bson.M{"key": bson.M{"$in": key}})
return err
return errs.Wrap(err)
}
func (x *CacheMgo) lockKey(key string) string {