diff --git a/internal/rpc/phone_calls.go b/internal/rpc/phone_calls.go index 4f303776..a59c8e0a 100644 --- a/internal/rpc/phone_calls.go +++ b/internal/rpc/phone_calls.go @@ -135,7 +135,10 @@ func (r *Router) onPhoneReceivedCall(ctx context.Context, peer tg.InputPhoneCall if transitioned { // ⚠ P1-2：receiveDate 推送必须在 P1 就位。主叫只有收到带 receive_date 的 // phoneCallWaiting 才会把 20s receive 定时器换成 90s ring 定时器。 - r.pushPhoneCall(ctx, call.AdminID, call, "phone call ringing") + // 这条主叫视角更新只属于 requestCall 的来源设备；账号级广播会在 DrKLO + // 多账号同机时按相同 call_id 覆盖被叫的 pending phoneCallRequested，丢失 + // g_a_hash 并在接听时触发 Ga hash mismatch。 + r.pushPhoneCallToDevice(ctx, call.AdminID, call.CallerDevice, call, "phone call ringing") } return true, nil } diff --git a/internal/rpc/phone_push.go b/internal/rpc/phone_push.go index f6fe9b88..e829f2cb 100644 --- a/internal/rpc/phone_push.go +++ b/internal/rpc/phone_push.go @@ -5,6 +5,7 @@ import ( "github.com/iamxvbaba/td/proto" "github.com/iamxvbaba/td/tg" + "go.uber.org/zap" "telesrv/internal/domain" ) @@ -34,6 +35,35 @@ func (r *Router) pushPhoneCall(ctx context.Context, targetUserID int64, call dom return r.pushUserMessage(ctx, targetUserID, logMessage, r.phoneCallUpdates(ctx, call, targetUserID)) } +// pushPhoneCallToDevice 只把 phoneCall 状态推给一台精确的物理 session。 +// +// 这是 fail-closed 路径：目标锚点缺失、session 已断开或编码/发送失败时都不得 +// 回退为 user 级广播。呼出通话的 ringing 更新只属于 requestCall 来源设备； +// 扩大投递范围会在 DrKLO 多账号同机时污染另一账号的全局 pending 来电对象。 +func (r *Router) pushPhoneCallToDevice(ctx context.Context, targetUserID int64, device domain.SessionRef, call domain.PhoneCall, logMessage string) { + if targetUserID == 0 || device.RawAuthKeyID == ([8]byte{}) || device.SessionID == 0 || r.deps.Sessions == nil { + if r.log != nil { + r.log.Debug(logMessage, + zap.Int64("target_user_id", targetUserID), + zap.Int64("call_id", call.ID), + zap.Int64("target_session_id", device.SessionID), + zap.String("delivery", "skipped_invalid_device_anchor"), + ) + } + return + } + + updates := r.phoneCallUpdates(ctx, call, targetUserID) + if err := r.deps.Sessions.PushToSessionForAuthKey(ctx, device.RawAuthKeyID, device.SessionID, proto.MessageFromServer, updates); err != nil && r.log != nil { + r.log.Debug(logMessage, + zap.Int64("target_user_id", targetUserID), + zap.Int64("call_id", call.ID), + zap.Int64("target_session_id", device.SessionID), + zap.Error(err), + ) + } +} + // pushPhoneCallStopRinging 向被叫其它设备推合成 phoneCallDiscarded 停振铃（P0-1 修正）。 // ctx 必须是接听设备的请求上下文：except 语义恰好把赢家排除在外。 func (r *Router) pushPhoneCallStopRinging(ctx context.Context, call domain.PhoneCall) int { diff --git a/internal/rpc/phone_rpc_test.go b/internal/rpc/phone_rpc_test.go index 97f53331..f96e579f 100644 --- a/internal/rpc/phone_rpc_test.go +++ b/internal/rpc/phone_rpc_test.go @@ -25,6 +25,7 @@ import ( // phonePushRecord 记录一次定向推送（目标用户、被排除的 session、载荷）。 type phonePushRecord struct { userID int64 + rawAuthKeyID [8]byte targetSession int64 excludeSession int64 msg bin.Encoder @@ -32,8 +33,9 @@ type phonePushRecord struct { // phoneCaptureSessions 是带完整推送日志的 SessionBinder fake（captureSessions 只留最后一条）。 type phoneCaptureSessions struct { - mu sync.Mutex - log []phonePushRecord + mu sync.Mutex + log []phonePushRecord + pushErr error } func (s *phoneCaptureSessions) BindAuthKeyForSession([8]byte, int64, [8]byte) {} @@ -47,11 +49,11 @@ func (s *phoneCaptureSessions) UserIDResolvedForAuthKey([8]byte, int64) (int64, func (s *phoneCaptureSessions) UnbindAuthKey([8]byte) int { return 0 } func (s *phoneCaptureSessions) SetReceivesUpdatesForAuthKey([8]byte, int64, bool) {} -func (s *phoneCaptureSessions) PushToSessionForAuthKey(_ context.Context, _ [8]byte, sessionID int64, _ proto.MessageType, msg tg.UpdatesClass) error { +func (s *phoneCaptureSessions) PushToSessionForAuthKey(_ context.Context, rawAuthKeyID [8]byte, sessionID int64, _ proto.MessageType, msg tg.UpdatesClass) error { s.mu.Lock() defer s.mu.Unlock() - s.log = append(s.log, phonePushRecord{targetSession: sessionID, msg: msg}) - return nil + s.log = append(s.log, phonePushRecord{rawAuthKeyID: rawAuthKeyID, targetSession: sessionID, msg: msg}) + return s.pushErr } func (s *phoneCaptureSessions) PushToUserExceptAuthKeySession(_ context.Context, userID int64, _ [8]byte, excludeSessionID int64, _ proto.MessageType, msg tg.UpdatesClass) (int, error) { @@ -73,6 +75,12 @@ func (s *phoneCaptureSessions) reset() { s.log = nil } +func (s *phoneCaptureSessions) setPushError(err error) { + s.mu.Lock() + defer s.mu.Unlock() + s.pushErr = err +} + // stubPrivacy 只为 CanSee 服务；其余接口方法不在通话链路使用。 type stubPrivacy struct { deny map[domain.PrivacyKey]bool @@ -104,8 +112,15 @@ type phoneFixture struct { } const ( - phoneCallerSession = int64(101) - phoneCalleeSession = int64(202) + phoneCallerSession = int64(101) + phoneCalleeSession = int64(202) + phoneOtherCalleeSession = int64(303) +) + +var ( + phoneCallerRawAuthKey = [8]byte{0x11, 0x01} + phoneCalleeRawAuthKey = [8]byte{0x22, 0x02} + phoneOtherCalleeRawAuthKey = [8]byte{0x33, 0x03} ) func newPhoneFixture(t *testing.T, privacy PrivacyService) *phoneFixture { @@ -133,11 +148,16 @@ func newPhoneFixture(t *testing.T, privacy PrivacyService) *phoneFixture { } func (f *phoneFixture) callerCtx() context.Context { - return WithSessionID(WithUserID(f.ctx, f.caller.ID), phoneCallerSession) + return WithSessionID(WithRawAuthKeyID(WithUserID(f.ctx, f.caller.ID), phoneCallerRawAuthKey), phoneCallerSession) } func (f *phoneFixture) calleeCtx() context.Context { - return WithSessionID(WithUserID(f.ctx, f.callee.ID), phoneCalleeSession) + return WithSessionID(WithRawAuthKeyID(WithUserID(f.ctx, f.callee.ID), phoneCalleeRawAuthKey), phoneCalleeSession) + +} + +func (f *phoneFixture) otherCalleeCtx() context.Context { + return WithSessionID(WithRawAuthKeyID(WithUserID(f.ctx, f.callee.ID), phoneOtherCalleeRawAuthKey), phoneOtherCalleeSession) } func phoneTestProtocol() tg.PhoneCallProtocol { @@ -226,8 +246,8 @@ func TestPhoneCallRPCHappyPath(t *testing.T) { t.Fatalf("receivedCall = %v err=%v", ok, err) } pushes = f.sessions.records() - if len(pushes) != 1 || pushes[0].userID != f.caller.ID { - t.Fatalf("receivedCall pushes = %+v, want one to caller", pushes) + if len(pushes) != 1 || pushes[0].rawAuthKeyID != phoneCallerRawAuthKey || pushes[0].targetSession != phoneCallerSession { + t.Fatalf("receivedCall pushes = %+v, want caller device %x/%d", pushes, phoneCallerRawAuthKey, phoneCallerSession) } ringing, ok := phoneCallPayload(t, pushes[0]).(*tg.PhoneCallWaiting) if !ok || ringing.ReceiveDate == 0 { @@ -236,6 +256,15 @@ func TestPhoneCallRPCHappyPath(t *testing.T) { } f.sessions.reset() + // 其它被叫设备晚到的 receivedCall 幂等成功，但不得再次推 ringing。 + if ok, err := f.router.onPhoneReceivedCall(f.otherCalleeCtx(), tg.InputPhoneCall{ID: callID, AccessHash: accessHash}); err != nil || !ok { + t.Fatalf("duplicate receivedCall = %v err=%v", ok, err) + } + if pushes := f.sessions.records(); len(pushes) != 0 { + t.Fatalf("duplicate receivedCall pushes = %+v, want none", pushes) + } + f.sessions.reset() + // --- acceptCall（被叫赢家设备） --- acceptRes, err := f.router.onPhoneAcceptCall(f.calleeCtx(), &tg.PhoneAcceptCallRequest{ Peer: tg.InputPhoneCall{ID: callID, AccessHash: accessHash}, @@ -388,6 +417,35 @@ func TestPhoneCallRPCHappyPath(t *testing.T) { } } +func TestPhoneReceivedCallDevicePushFailureDoesNotBroadcast(t *testing.T) { + f := newPhoneFixture(t, stubPrivacy{}) + _, gaHash, _ := phoneTestKeys() + + res, err := f.router.onPhoneRequestCall(f.callerCtx(), &tg.PhoneRequestCallRequest{ + UserID: &tg.InputUser{UserID: f.callee.ID, AccessHash: f.callee.AccessHash}, + RandomID: 43, + GAHash: gaHash, + Protocol: phoneTestProtocol(), + }) + if err != nil { + t.Fatalf("requestCall: %v", err) + } + waiting := res.PhoneCall.(*tg.PhoneCallWaiting) + f.sessions.reset() + f.sessions.setPushError(errors.New("caller session gone")) + + if ok, err := f.router.onPhoneReceivedCall(f.calleeCtx(), tg.InputPhoneCall{ID: waiting.ID, AccessHash: waiting.AccessHash}); err != nil || !ok { + t.Fatalf("receivedCall = %v err=%v", ok, err) + } + pushes := f.sessions.records() + if len(pushes) != 1 || pushes[0].rawAuthKeyID != phoneCallerRawAuthKey || pushes[0].targetSession != phoneCallerSession { + t.Fatalf("receivedCall pushes = %+v, want one failed attempt to caller device", pushes) + } + if pushes[0].userID != 0 { + t.Fatalf("receivedCall failure fell back to user broadcast: %+v", pushes) + } +} + func TestPhoneCallRPCValidation(t *testing.T) { f := newPhoneFixture(t, stubPrivacy{}) _, gaHash, gb := phoneTestKeys()