diff --git a/services/gateway-service/internal/transport/http/callbackapi/tencent_im_callback.go b/services/gateway-service/internal/transport/http/callbackapi/tencent_im_callback.go index 0f30fd9f..53bbbc10 100644 --- a/services/gateway-service/internal/transport/http/callbackapi/tencent_im_callback.go +++ b/services/gateway-service/internal/transport/http/callbackapi/tencent_im_callback.go @@ -2,6 +2,7 @@ package callbackapi import ( "context" + "crypto/sha256" "encoding/json" "fmt" "hyapp/services/gateway-service/internal/transport/http/httpkit" @@ -302,8 +303,25 @@ func (h *Handler) tencentIMCallbackAuthenticated(request *http.Request) bool { return true } - return subtleStringEqual(request.URL.Query().Get("CallbackAuthToken"), token) || - subtleStringEqual(request.Header.Get("X-Tencent-IM-Callback-Token"), token) + if subtleStringEqual(request.URL.Query().Get("CallbackAuthToken"), token) || + subtleStringEqual(request.Header.Get("X-Tencent-IM-Callback-Token"), token) { + // 保留既有直传 token 协议，避免已接入的内部探针在控制台切换鉴权方式时失效。 + return true + } + + requestTimeRaw := strings.TrimSpace(request.URL.Query().Get("RequestTime")) + requestTime, err := strconv.ParseInt(requestTimeRaw, 10, 64) + if err != nil { + return false + } + if deltaSeconds := time.Now().Unix() - requestTime; deltaSeconds < -60 || deltaSeconds > 60 { + // 腾讯云签名只绑定秒级时间戳；限制一分钟窗口可阻止已截获回调被长期重放。 + return false + } + + digest := sha256.Sum256([]byte(token + requestTimeRaw)) + expectedSign := fmt.Sprintf("%x", digest) + return subtleStringEqual(request.URL.Query().Get("Sign"), expectedSign) } func (h *Handler) tencentIMSDKAppIDMatched(request *http.Request) bool { diff --git a/services/gateway-service/internal/transport/http/response_test.go b/services/gateway-service/internal/transport/http/response_test.go index 1f967c6c..a2f940f2 100644 --- a/services/gateway-service/internal/transport/http/response_test.go +++ b/services/gateway-service/internal/transport/http/response_test.go @@ -8,6 +8,7 @@ import ( "encoding/base64" "encoding/json" "errors" + "fmt" "hyapp/services/gateway-service/internal/transport/http/httpkit" "hyapp/services/gateway-service/internal/transport/http/roomapi" "io" @@ -12192,6 +12193,32 @@ func TestTencentIMCallbackAuthAndUnknownCommand(t *testing.T) { assertTencentCallback(t, unknownRecorder, http.StatusOK, 0) } +func TestTencentIMCallbackAcceptsSignedAuthAndRejectsExpiredRequest(t *testing.T) { + const callbackToken = "callback-token" + router := NewHandlerWithConfig(&fakeRoomClient{}, &fakeRoomGuardClient{}, nil, nil, TencentIMConfig{ + SDKAppID: 1400000000, + CallbackAuthToken: callbackToken, + }).Routes(auth.NewVerifier("secret")) + body := []byte(`{"CallbackCommand":"Group.CallbackAfterSendMsg","GroupId":"room-1","From_Account":"42"}`) + + requestTime := strconv.FormatInt(time.Now().Unix(), 10) + request := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackAfterSendMsg&RequestTime="+requestTime+"&Sign="+tencentIMCallbackSignature(callbackToken, requestTime), bytes.NewReader(body)) + recorder := httptest.NewRecorder() + router.ServeHTTP(recorder, request) + assertTencentCallback(t, recorder, http.StatusOK, 0) + + expiredTime := strconv.FormatInt(time.Now().Add(-2*time.Minute).Unix(), 10) + expiredRequest := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackAfterSendMsg&RequestTime="+expiredTime+"&Sign="+tencentIMCallbackSignature(callbackToken, expiredTime), bytes.NewReader(body)) + expiredRecorder := httptest.NewRecorder() + router.ServeHTTP(expiredRecorder, expiredRequest) + assertTencentCallback(t, expiredRecorder, http.StatusOK, 1) +} + +func tencentIMCallbackSignature(token string, requestTime string) string { + digest := sha256.Sum256([]byte(token + requestTime)) + return fmt.Sprintf("%x", digest) +} + func assertEnvelope(t *testing.T, recorder *httptest.ResponseRecorder, statusCode int, code string, requestID string) { t.Helper() diff --git a/services/room-service/internal/room/service/create_room.go b/services/room-service/internal/room/service/create_room.go index 9d6439bf..750667e5 100644 --- a/services/room-service/internal/room/service/create_room.go +++ b/services/room-service/internal/room/service/create_room.go @@ -2,6 +2,7 @@ package service import ( "context" + "errors" "fmt" "log/slog" "strconv" @@ -249,6 +250,9 @@ func (s *Service) CreateRoom(ctx context.Context, req *roomv1.CreateRoomRequest) OutboxRecords: []outbox.Record{createdEvent}, RoomCoverUploadConsumption: coverConsumption, }); err != nil { + if errors.Is(err, ErrRoomCoverUploadNotConsumable) { + return nil, xerr.New(xerr.Conflict, "room cover upload is already consumed") + } return nil, err } saveMutationMS = elapsedMS(saveStartedAt) diff --git a/services/room-service/internal/room/service/mic.go b/services/room-service/internal/room/service/mic.go index 7f87c40d..36423e33 100644 --- a/services/room-service/internal/room/service/mic.go +++ b/services/room-service/internal/room/service/mic.go @@ -377,6 +377,10 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM return mutationResult{}, nil, err } } + if current.MuteUsers[cmd.TargetUserID] { + // MicUp 后到 RTC 发流确认之间仍可能发生房间禁言；确认入口必须再次校验，不能复活已撤销的发言权限。 + return mutationResult{}, nil, xerr.New(xerr.PermissionDenied, "user is muted") + } seat, exists := current.SeatByUser(cmd.TargetUserID) if !exists { @@ -608,6 +612,10 @@ func (s *Service) SetMicMute(ctx context.Context, req *roomv1.SetMicMuteRequest) if !exists { return mutationResult{}, nil, xerr.New(xerr.NotFound, "target not on seat") } + if !cmd.Muted && current.MuteUsers[cmd.TargetUserID] { + // 房间禁言优先级高于用户自己的麦克风开关；解除房间禁言前禁止目标或管理员把 RTC 麦位重新打开。 + return mutationResult{}, nil, xerr.New(xerr.PermissionDenied, "user is muted") + } if seat.MicMuted == cmd.Muted { return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil } diff --git a/services/room-service/internal/room/service/mic_test.go b/services/room-service/internal/room/service/mic_test.go index 1e8030f3..f578ed84 100644 --- a/services/room-service/internal/room/service/mic_test.go +++ b/services/room-service/internal/room/service/mic_test.go @@ -70,12 +70,64 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { if !unmutedSpeak.GetAllowed() || unmutedSpeak.GetReason() != "" { t.Fatalf("unmuted user must recover public chat permission: %+v", unmutedSpeak) } - if _, err := svc.MicUp(ctx, &roomv1.MicUpRequest{ + upResp, err := svc.MicUp(ctx, &roomv1.MicUpRequest{ Meta: rocketMeta(roomID, mutedUserID, "unmuted-mic-up"), SeatNo: 2, - }); err != nil { + }) + if err != nil { t.Fatalf("unmuted user must recover mic up permission: %v", err) } + + mutedOnSeat, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ + Meta: rocketMeta(roomID, ownerID, "mute-current-rtc-seat"), + TargetUserId: mutedUserID, + Muted: true, + }) + if err != nil { + t.Fatalf("mute current RTC seat failed: %v", err) + } + var currentSeat *roomv1.SeatState + for _, seat := range mutedOnSeat.GetRoom().GetMicSeats() { + if seat.GetUserId() == mutedUserID { + currentSeat = seat + break + } + } + if currentSeat == nil || !currentSeat.GetMicMuted() || currentSeat.GetSeatNo() != 2 { + t.Fatalf("room mute must project to the current RTC seat without removing it: %+v", currentSeat) + } + if _, err := svc.SetMicMute(ctx, &roomv1.SetMicMuteRequest{ + Meta: rocketMeta(roomID, mutedUserID, "muted-user-open-mic"), + TargetUserId: mutedUserID, + Muted: false, + }); !xerr.IsCode(err, xerr.PermissionDenied) { + t.Fatalf("room-muted user must not reopen RTC mic: %v", err) + } + if _, err := svc.ConfirmMicPublishing(ctx, &roomv1.ConfirmMicPublishingRequest{ + Meta: rocketMeta(roomID, mutedUserID, "muted-publish-confirm"), + TargetUserId: mutedUserID, + MicSessionId: upResp.GetMicSessionId(), + RoomVersion: upResp.GetResult().GetRoomVersion(), + EventTimeMs: now.Now().UnixMilli(), + Source: "client", + }); !xerr.IsCode(err, xerr.PermissionDenied) { + t.Fatalf("room mute between MicUp and RTC confirmation must reject publishing: %v", err) + } + + if _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ + Meta: rocketMeta(roomID, ownerID, "unmute-current-rtc-seat"), + TargetUserId: mutedUserID, + Muted: false, + }); err != nil { + t.Fatalf("unmute current RTC seat failed: %v", err) + } + if _, err := svc.SetMicMute(ctx, &roomv1.SetMicMuteRequest{ + Meta: rocketMeta(roomID, mutedUserID, "unmuted-user-open-mic"), + TargetUserId: mutedUserID, + Muted: false, + }); err != nil { + t.Fatalf("room unmute must restore explicit RTC mic control: %v", err) + } } func TestMicUpReturnsExistingSeatForDuplicateUserCommand(t *testing.T) { diff --git a/services/room-service/internal/room/service/moderation.go b/services/room-service/internal/room/service/moderation.go index d222eaa0..e35bbdca 100644 --- a/services/room-service/internal/room/service/moderation.go +++ b/services/room-service/internal/room/service/moderation.go @@ -17,7 +17,7 @@ import ( // MuteUser 修改用户禁言状态。 func (s *Service) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*roomv1.MuteUserResponse, error) { ctx = contextFromMeta(ctx, req.GetMeta()) - // 禁言状态由 room-service 持有，腾讯云 IM 发言前回调必须同步查询 CheckSpeakPermission。 + // 禁言状态由 room-service 持有：腾讯云 IM 发言前回调查 CheckSpeakPermission，RTC 麦位则同步为服务端静音态。 cmd := command.MuteUser{ Base: baseFromMeta(req.GetMeta()), TargetUserID: req.GetTargetUserId(), @@ -51,11 +51,20 @@ func (s *Service) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*r return mutationResult{snapshot: current.ToProto()}, nil, nil } + var mutedSeat *state.MicSeat if cmd.Muted { - // map set 表达禁言集合，true 才会被快照导出。 + // map set 是公屏、上麦和开麦权限的统一事实；麦位静音只是这个事实的 RTC 投影。 current.MuteUsers[cmd.TargetUserID] = true + if seat, exists := current.SeatByUser(cmd.TargetUserID); exists && !seat.MicMuted { + // 房间禁言不能只依赖目标客户端执行 muteLocalAudio；先把当前麦位标成 muted， + // 让快照和 room_mic_mute_changed 都保持静音，目标用户解除房间禁言前也不能自行开麦。 + index := current.SeatIndex(seat.SeatNo) + current.MicSeats[index].MicMuted = true + updated := current.MicSeats[index] + mutedSeat = &updated + } } else { - // 解除禁言直接删除集合成员。 + // 解除房间禁言只恢复“允许开麦”；不替用户自动打开本地麦克风，当前麦位仍需显式 SetMicMute(false)。 delete(current.MuteUsers, cmd.TargetUserID) } @@ -70,6 +79,28 @@ func (s *Service) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*r if err != nil { return mutationResult{}, nil, err } + records := make([]outbox.Record, 0, 2) + if mutedSeat != nil { + micChanged := roomMicChangedEvent(&roomeventsv1.RoomMicChanged{ + ActorUserId: cmd.ActorUserID(), + TargetUserId: cmd.TargetUserID, + FromSeat: mutedSeat.SeatNo, + ToSeat: mutedSeat.SeatNo, + Action: "mic_mute", + MicSessionId: mutedSeat.MicSessionID, + PublishState: mutedSeat.PublishState, + PublishDeadlineMs: mutedSeat.PublishDeadlineMS, + MicMuted: true, + SeatStatus: mutedSeat.SeatStatus(), + TargetGiftValue: roomUserGiftValue(current, cmd.TargetUserID), + }, command.UserDisplayProfile{}) + micEvent, err := outbox.Build(current.RoomID, "RoomMicChanged", current.Version, now, micChanged) + if err != nil { + return mutationResult{}, nil, err + } + records = append(records, micEvent) + } + records = append(records, muteEvent) return mutationResult{ snapshot: current.ToProto(), @@ -81,7 +112,7 @@ func (s *Service) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*r TargetUserID: cmd.TargetUserID, RoomVersion: current.Version, }, - }, []outbox.Record{muteEvent}, nil + }, records, nil }) if err != nil { return nil, err diff --git a/services/room-service/internal/room/service/pipeline.go b/services/room-service/internal/room/service/pipeline.go index c4a119b2..4c86f64e 100644 --- a/services/room-service/internal/room/service/pipeline.go +++ b/services/room-service/internal/room/service/pipeline.go @@ -161,6 +161,9 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl } return replayMutationResult(committed, snapshotWithApp(ctx, current.ToProto())) } + if errors.Is(err, ErrRoomCoverUploadNotConsumable) { + return nil, xerr.New(xerr.Conflict, "room cover upload is already consumed") + } if err != nil { // COMMIT can succeed in MySQL while its acknowledgement is lost to a connection reset. The gift // operation is then already committed and recovery workers correctly stop claiming it, so returning @@ -213,6 +216,9 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl RoomCoverUploadConsumption: result.roomCoverUploadConsumption, }) commitCancel() + if errors.Is(err, ErrRoomCoverUploadNotConsumable) { + return nil, xerr.New(xerr.Conflict, "room cover upload is already consumed") + } if err != nil && !s.mutationCommitConfirmed(ctx, commandRecord) { return nil, err } diff --git a/services/room-service/internal/room/service/recovery.go b/services/room-service/internal/room/service/recovery.go index c4e9255c..31529c3d 100644 --- a/services/room-service/internal/room/service/recovery.go +++ b/services/room-service/internal/room/service/recovery.go @@ -457,9 +457,14 @@ func replay(current *state.RoomState, cmd command.Command, committedAtMS int64) current.Version++ case *command.MuteUser: if typed.Muted { - // 回放禁言集合，供 CheckSpeakPermission 使用。 + // 回放禁言集合及其 RTC 麦位投影；否则重启后快照会把仍被房间禁言的用户显示成可开麦。 current.MuteUsers[typed.TargetUserID] = true + if seat, exists := current.SeatByUser(typed.TargetUserID); exists { + index := current.SeatIndex(seat.SeatNo) + current.MicSeats[index].MicMuted = true + } } else { + // 与在线命令一致，解除房间禁言不隐式打开用户麦克风。 delete(current.MuteUsers, typed.TargetUserID) } current.Version++ diff --git a/services/room-service/internal/room/service/repository.go b/services/room-service/internal/room/service/repository.go index 248094cd..169661ff 100644 --- a/services/room-service/internal/room/service/repository.go +++ b/services/room-service/internal/room/service/repository.go @@ -225,6 +225,10 @@ type RoomMediaUploadRegistration struct { // ErrRoomMediaUploadCommandConflict 表示上传 command_id 已绑定另一份文件或用途。 var ErrRoomMediaUploadCommandConflict = errors.New("room media upload command payload conflict") +// ErrRoomCoverUploadNotConsumable 表示凭证不存在、已被消费或与当前 actor 不匹配。 +// service 必须映射成稳定业务冲突，不能把事务条件未命中作为数据库内部错误暴露。 +var ErrRoomCoverUploadNotConsumable = errors.New("room cover upload is not consumable") + // RoomCoverUploadConsumption 是创建/更新命令对预上传封面的单次消费条件。 // SQL 还会用 app_code、空 room_id、purpose 和 active 状态做条件更新，防止跨租户、跨用户和重复使用。 type RoomCoverUploadConsumption struct { diff --git a/services/room-service/internal/storage/mysql/command_log.go b/services/room-service/internal/storage/mysql/command_log.go index 34803009..b890275d 100644 --- a/services/room-service/internal/storage/mysql/command_log.go +++ b/services/room-service/internal/storage/mysql/command_log.go @@ -180,7 +180,7 @@ func (r *Repository) SaveMutation(ctx context.Context, commit roomservice.Mutati if err != nil { return err } - return fmt.Errorf("room cover upload cannot be consumed: room_id=%s cover_upload_id=%s", commit.Command.RoomID, consumption.CoverUploadID) + return fmt.Errorf("%w: room_id=%s cover_upload_id=%s", roomservice.ErrRoomCoverUploadNotConsumable, commit.Command.RoomID, consumption.CoverUploadID) } }