From b4357848093c82435e505282c1de9ca5d35a7b7d Mon Sep 17 00:00:00 2001 From: A Date: Mon, 8 Jun 2026 21:42:46 +0800 Subject: [PATCH] mtproto,rpc: support Android legacy startup (cherry picked from commit 06b6ae2bb2f90bd7cc1d6a8404a82aff507e6821) --- internal/mtprotoedge/encrypted.go | 9 ++ internal/mtprotoedge/encrypted_test.go | 30 ++++++- internal/mtprotoedge/exchange.go | 55 +++++++++--- internal/mtprotoedge/exchange_test.go | 118 +++++++++++++++++++++++++ internal/mtprotoedge/helpers_test.go | 7 +- internal/rpc/compat.go | 44 +++++++++ internal/rpc/langpack.go | 58 ++++++++++++ internal/rpc/langpack_compat_test.go | 54 +++++++++++ internal/rpc/router.go | 3 + 9 files changed, 361 insertions(+), 17 deletions(-) create mode 100644 internal/rpc/compat.go create mode 100644 internal/rpc/langpack_compat_test.go diff --git a/internal/mtprotoedge/encrypted.go b/internal/mtprotoedge/encrypted.go index 399eaf7c..9192b4db 100644 --- a/internal/mtprotoedge/encrypted.go +++ b/internal/mtprotoedge/encrypted.go @@ -42,6 +42,11 @@ func newConnState() *connState { } } +func (cs *connState) reset() { + next := newConnState() + *cs = *next +} + const ( maxTrackedClientMsgIDs = 400 @@ -86,6 +91,9 @@ func (s *Server) handleEncrypted(ctx context.Context, tc transport.Conn, cs *con // 首个加密消息或 session 变化时（重新）注册连接到 SessionManager。 if current == nil || current.sessionID != data.SessionID { + if current != nil { + cs.reset() + } if current != nil { s.conns.Unregister(current) current.Close() @@ -599,6 +607,7 @@ func clientMessageNeedsAck(typeID uint32) bool { switch typeID { case proto.MessageContainerTypeID, mt.MsgsAckTypeID, + mt.PingDelayDisconnectRequestTypeID, mt.HTTPWaitRequestTypeID, mt.BadMsgNotificationTypeID, mt.BadServerSaltTypeID, diff --git a/internal/mtprotoedge/encrypted_test.go b/internal/mtprotoedge/encrypted_test.go index 248c6c75..b07d8b10 100644 --- a/internal/mtprotoedge/encrypted_test.go +++ b/internal/mtprotoedge/encrypted_test.go @@ -278,14 +278,14 @@ func TestOldMessageInFreshContainerAccepted(t *testing.T) { } } -func TestPingDelayDisconnectOddSeqAccepted(t *testing.T) { +func TestPingDelayDisconnectEvenSeqAccepted(t *testing.T) { const dc = 2 addr, pub, _ := startTestServer(t, Options{DC: dc}) conn, auth, cipher := dialHandshake(t, addr, dc, pub) clientMsgID := proto.NewMessageIDGen(time.Now) reqMsgID := clientMsgID.New(proto.MessageFromClient) - sendEncryptedWithSeq(t, conn, cipher, auth, reqMsgID, 1, &mt.PingDelayDisconnectRequest{ + sendEncryptedWithSeq(t, conn, cipher, auth, reqMsgID, 0, &mt.PingDelayDisconnectRequest{ PingID: 9, DisconnectDelay: 60, }) @@ -414,6 +414,32 @@ func TestBadMsgSeqTooHigh(t *testing.T) { } } +func TestSessionChangeResetsClientSeqState(t *testing.T) { + const dc = 2 + addr, pub, _ := startTestServer(t, Options{DC: dc}) + conn, auth, cipher := dialHandshake(t, addr, dc, pub) + + clientMsgID := proto.NewMessageIDGen(time.Now) + firstMsgID := clientMsgID.New(proto.MessageFromClient) + sendEncryptedWithSeq(t, conn, cipher, auth, firstMsgID, 1, &tg.HelpGetConfigRequest{}) + collectReplies(t, conn, cipher, auth.AuthKey, mt.MsgsAckTypeID) + + nextSessionID := auth.SessionID + 1 + if nextSessionID == 0 { + nextSessionID++ + } + secondMsgID := clientMsgID.New(proto.MessageFromClient) + body := encodeClientMessageBodyForTest(t, &tg.HelpGetConfigRequest{}) + sendEncryptedWithSessionSaltAndSeq(t, conn, cipher, auth, nextSessionID, auth.ServerSalt, secondMsgID, 1, body) + + replies := collectReplies(t, conn, cipher, auth.AuthKey, mt.MsgsAckTypeID) + if _, ok := replies[mt.BadMsgNotificationTypeID]; ok { + t.Fatalf("session change with fresh seq_no produced bad_msg_notification") + } + mustHave(t, replies, mt.NewSessionCreatedTypeID, "new_session_created after session change") + mustHave(t, replies, mt.MsgsAckTypeID, "msgs_ack after session change") +} + func readBadMsgNotification(t *testing.T, conn transport.Conn, cipher crypto.Cipher, key crypto.AuthKey) mt.BadMsgNotification { t.Helper() replies := collectReplies(t, conn, cipher, key, mt.BadMsgNotificationTypeID) diff --git a/internal/mtprotoedge/exchange.go b/internal/mtprotoedge/exchange.go index d9703c50..ac99fe20 100644 --- a/internal/mtprotoedge/exchange.go +++ b/internal/mtprotoedge/exchange.go @@ -12,6 +12,8 @@ import ( "github.com/gotd/td/bin" "github.com/gotd/td/crypto" "github.com/gotd/td/exchange" + "github.com/gotd/td/mt" + "github.com/gotd/td/proto" "github.com/gotd/td/proto/codec" "github.com/gotd/td/transport" @@ -123,23 +125,48 @@ func (c *bufferedConn) push(b *bin.Buffer) { // Recv 优先返回已 push 的帧（FIFO），耗尽后读取底层连接。 func (c *bufferedConn) Recv(ctx context.Context, b *bin.Buffer) error { - c.mu.Lock() - if len(c.pending) > 0 { - e := c.pending[0] - c.pending = c.pending[1:] - c.last.ResetTo(e.Copy()) - c.mu.Unlock() - b.ResetTo(e.Buf) + for { + c.mu.Lock() + if len(c.pending) > 0 { + e := c.pending[0] + c.pending = c.pending[1:] + c.last.ResetTo(e.Copy()) + c.mu.Unlock() + b.ResetTo(e.Buf) + } else { + c.mu.Unlock() + if err := c.Conn.Recv(ctx, b); err != nil { + return err + } + c.mu.Lock() + c.last.ResetTo(b.Copy()) + c.mu.Unlock() + } + + if isUnencryptedMsgsAckFrame(b) { + continue + } return nil } - c.mu.Unlock() - if err := c.Conn.Recv(ctx, b); err != nil { - return err +} + +func isUnencryptedMsgsAckFrame(frame *bin.Buffer) bool { + authKeyID, err := peekAuthKeyID(frame) + if err != nil || authKeyID != emptyAuthKeyID { + return false } - c.mu.Lock() - c.last.ResetTo(b.Copy()) - c.mu.Unlock() - return nil + + var msg proto.UnencryptedMessage + copy := &bin.Buffer{Buf: frame.Copy()} + if err := msg.Decode(copy); err != nil { + return false + } + payload := &bin.Buffer{Buf: msg.MessageData} + id, err := payload.PeekID() + if err != nil { + return false + } + return id == mt.MsgsAckTypeID } func (c *bufferedConn) lastFrame() *bin.Buffer { diff --git a/internal/mtprotoedge/exchange_test.go b/internal/mtprotoedge/exchange_test.go index d03c885e..3cfb410f 100644 --- a/internal/mtprotoedge/exchange_test.go +++ b/internal/mtprotoedge/exchange_test.go @@ -103,6 +103,124 @@ func TestKeyExchange(t *testing.T) { } } +func TestKeyExchangeIgnoresUnencryptedMsgsAck(t *testing.T) { + const dc = 2 + + rsaKey, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatalf("gen rsa: %v", err) + } + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + keys := memory.NewAuthKeyStore() + srv := New(Options{ + Logger: zaptest.NewLogger(t), + DC: dc, + RSAKey: rsaKey, + AuthKeys: keys, + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + serveErr := make(chan error, 1) + go func() { serveErr <- srv.Serve(ctx, ln) }() + + raw, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + conn, err := transport.Intermediate.Handshake(raw) + if err != nil { + t.Fatalf("transport handshake: %v", err) + } + + pub := exchange.PublicKey{RSA: &rsaKey.PublicKey} + exchCtx, ec := context.WithTimeout(context.Background(), 10*time.Second) + defer ec() + res, err := exchange.NewExchanger(&ackingExchangeConn{Conn: conn, t: t}, dc). + WithRand(rand.Reader). + WithLogger(zaptest.NewLogger(t).Named("client")). + Client([]exchange.PublicKey{pub}). + Run(exchCtx) + if err != nil { + t.Fatalf("client exchange: %v", err) + } + + deadline := time.Now().Add(5 * time.Second) + for { + if _, found, _ := keys.Get(context.Background(), res.AuthKey.ID); found { + break + } + if time.Now().After(deadline) { + t.Fatalf("server did not store auth key %x", res.AuthKey.ID) + } + time.Sleep(20 * time.Millisecond) + } + + cancel() + select { + case err := <-serveErr: + if err != nil { + t.Fatalf("serve: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("server did not stop after ctx cancel") + } +} + +type ackingExchangeConn struct { + transport.Conn + t *testing.T +} + +func (c *ackingExchangeConn) Recv(ctx context.Context, b *bin.Buffer) error { + if err := c.Conn.Recv(ctx, b); err != nil { + return err + } + c.ackHandshakeMessage(b) + return nil +} + +func (c *ackingExchangeConn) ackHandshakeMessage(frame *bin.Buffer) { + var msg tgproto.UnencryptedMessage + copy := &bin.Buffer{Buf: frame.Copy()} + if err := msg.Decode(copy); err != nil { + return + } + payload := &bin.Buffer{Buf: msg.MessageData} + id, err := payload.PeekID() + if err != nil { + return + } + switch id { + case mt.ResPQTypeID, mt.ServerDHParamsOkTypeID: + default: + return + } + + var ackPayload bin.Buffer + if err := (&mt.MsgsAck{MsgIDs: []int64{msg.MessageID}}).Encode(&ackPayload); err != nil { + c.t.Fatalf("encode msgs_ack: %v", err) + } + var ackFrame bin.Buffer + if err := (tgproto.UnencryptedMessage{ + MessageID: int64(tgproto.NewMessageID(time.Now(), tgproto.MessageFromClient)), + MessageData: ackPayload.Raw(), + }).Encode(&ackFrame); err != nil { + c.t.Fatalf("encode msgs_ack frame: %v", err) + } + sendCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := c.Conn.Send(sendCtx, &ackFrame); err != nil { + c.t.Fatalf("send msgs_ack: %v", err) + } +} + func TestReconnectFakeReqPQThenEncryptedFrame(t *testing.T) { const dc = 2 addr, pub, _ := startTestServer(t, Options{DC: dc}) diff --git a/internal/mtprotoedge/helpers_test.go b/internal/mtprotoedge/helpers_test.go index 88a384be..f75521af 100644 --- a/internal/mtprotoedge/helpers_test.go +++ b/internal/mtprotoedge/helpers_test.go @@ -104,11 +104,16 @@ func sendEncryptedWithSeq(t *testing.T, conn transport.Conn, cipher crypto.Ciphe } func sendEncryptedWithSaltAndSeq(t *testing.T, conn transport.Conn, cipher crypto.Cipher, auth exchange.ClientExchangeResult, salt, msgID int64, seqNo int32, body []byte) { + t.Helper() + sendEncryptedWithSessionSaltAndSeq(t, conn, cipher, auth, auth.SessionID, salt, msgID, seqNo, body) +} + +func sendEncryptedWithSessionSaltAndSeq(t *testing.T, conn transport.Conn, cipher crypto.Cipher, auth exchange.ClientExchangeResult, sessionID, salt, msgID int64, seqNo int32, body []byte) { t.Helper() var buf bin.Buffer if err := cipher.Encrypt(auth.AuthKey, crypto.EncryptedMessageData{ Salt: salt, - SessionID: auth.SessionID, + SessionID: sessionID, MessageID: msgID, SeqNo: seqNo, MessageDataLen: int32(len(body)), diff --git a/internal/rpc/compat.go b/internal/rpc/compat.go new file mode 100644 index 00000000..7ab07302 --- /dev/null +++ b/internal/rpc/compat.go @@ -0,0 +1,44 @@ +package rpc + +import ( + "context" + "fmt" + "time" + + "go.uber.org/zap" + + "github.com/gotd/td/bin" +) + +// dispatchCompat handles explicitly allowlisted legacy TL constructors that are +// still emitted by supported clients but are absent from gotd's pinned schema. +func (r *Router) dispatchCompat(ctx context.Context, b *bin.Buffer, id uint32) (bin.Encoder, bool, error) { + start := time.Now() + var ( + enc bin.Encoder + name string + err error + ) + + switch id { + case legacyLangpackGetLanguagesTypeID: + name = "langpack.getLanguages#800fd57d" + enc, err = r.handleLegacyLangpackGetLanguages(ctx, b) + default: + return nil, false, nil + } + + fields := append([]zap.Field{ + zap.String("method", name), + zap.String("type_id", fmt.Sprintf("%#x", id)), + zap.Bool("compat", true), + zap.Duration("dur", time.Since(start)), + }, r.contextLogFields(ctx)...) + if err != nil { + fields = append(fields, zap.Error(err)) + r.log.Info("RPC compat handled", fields...) + } else { + r.log.Debug("RPC compat handled", fields...) + } + return enc, true, err +} diff --git a/internal/rpc/langpack.go b/internal/rpc/langpack.go index 5914ef5a..38f5ba1f 100644 --- a/internal/rpc/langpack.go +++ b/internal/rpc/langpack.go @@ -2,12 +2,19 @@ package rpc import ( "context" + "strings" + "github.com/gotd/td/bin" "github.com/gotd/td/tg" ) +const legacyLangpackGetLanguagesTypeID uint32 = 0x800fd57d + // registerLangpack 注册 langpack.* RPC handler。 func (r *Router) registerLangpack(d *tg.ServerDispatcher) { + d.OnLangpackGetLanguages(func(ctx context.Context, langPack string) ([]tg.LangPackLanguage, error) { + return r.langpackLanguages(ctx, langPack), nil + }) d.OnLangpackGetLangPack(func(ctx context.Context, req *tg.LangpackGetLangPackRequest) (*tg.LangPackDifference, error) { if r.deps.LangPack == nil { return &tg.LangPackDifference{LangCode: req.LangCode}, nil @@ -39,3 +46,54 @@ func (r *Router) registerLangpack(d *tg.ServerDispatcher) { return tgLangPackStrings(pack.Strings), nil }) } + +func (r *Router) handleLegacyLangpackGetLanguages(ctx context.Context, b *bin.Buffer) (bin.Encoder, error) { + if err := b.ConsumeID(legacyLangpackGetLanguagesTypeID); err != nil { + return nil, err + } + return &tg.LangPackLanguageVector{Elems: r.langpackLanguages(ctx, "")}, nil +} + +func (r *Router) langpackLanguages(ctx context.Context, langPack string) []tg.LangPackLanguage { + if langPack == "" { + langPack = langPackFromClient(ctx) + } + _ = langPack + return []tg.LangPackLanguage{ + { + Official: true, + Name: "English", + NativeName: "English", + LangCode: "en", + PluralCode: "en", + StringsCount: 0, + TranslatedCount: 0, + TranslationsURL: "", + }, + { + Official: true, + Name: "Chinese (Simplified)", + NativeName: "Chinese (Simplified)", + LangCode: "zh-hans", + PluralCode: "zh", + StringsCount: 0, + TranslatedCount: 0, + TranslationsURL: "", + }, + } +} + +func langPackFromClient(ctx context.Context) string { + info, ok := ClientInfoFrom(ctx) + if !ok { + return "tdesktop" + } + if info.LangPack != "" { + return info.LangPack + } + client := strings.ToLower(info.DeviceModel + " " + info.SystemVersion + " " + info.AppVersion) + if strings.Contains(client, "android") { + return "android" + } + return "tdesktop" +} diff --git a/internal/rpc/langpack_compat_test.go b/internal/rpc/langpack_compat_test.go new file mode 100644 index 00000000..3b87b1f2 --- /dev/null +++ b/internal/rpc/langpack_compat_test.go @@ -0,0 +1,54 @@ +package rpc + +import ( + "context" + "testing" + + "go.uber.org/zap/zaptest" + + "github.com/gotd/td/bin" + "github.com/gotd/td/clock" + "github.com/gotd/td/tg" +) + +func TestLangpackGetLanguagesCurrentAndLegacy(t *testing.T) { + r := New(Config{DC: 2, IP: "127.0.0.1", Port: 2398}, Deps{}, zaptest.NewLogger(t), clock.System) + + t.Run("current layer", func(t *testing.T) { + var in bin.Buffer + if err := (&tg.LangpackGetLanguagesRequest{LangPack: "tdesktop"}).Encode(&in); err != nil { + t.Fatalf("encode request: %v", err) + } + assertLangpackLanguages(t, r, context.Background(), &in) + }) + + t.Run("legacy android no args", func(t *testing.T) { + var in bin.Buffer + in.PutID(legacyLangpackGetLanguagesTypeID) + ctx := WithClientInfo(context.Background(), ClientInfo{ + DeviceModel: "Android", + AppVersion: "12.7.3", + LangCode: "en", + }) + assertLangpackLanguages(t, r, ctx, &in) + }) +} + +func assertLangpackLanguages(t *testing.T, r *Router, ctx context.Context, in *bin.Buffer) { + t.Helper() + enc, err := r.Dispatch(ctx, [8]byte{}, 0, in) + if err != nil { + t.Fatalf("dispatch langpack.getLanguages: %v", err) + } + var out bin.Buffer + if err := enc.Encode(&out); err != nil { + t.Fatalf("encode response: %v", err) + } + var langs tg.LangPackLanguageVector + if err := langs.Decode(&out); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(langs.Elems) == 0 || langs.Elems[0].LangCode != "en" { + t.Fatalf("languages = %+v, want English entry", langs.Elems) + } +} diff --git a/internal/rpc/router.go b/internal/rpc/router.go index d8a4977d..391de403 100644 --- a/internal/rpc/router.go +++ b/internal/rpc/router.go @@ -306,6 +306,9 @@ func (r *Router) dispatch(ctx context.Context, b *bin.Buffer, depth int) (bin.En return r.dispatch(ctx, &bin.Buffer{Buf: inner.data}, depth+1) default: + if enc, ok, err := r.dispatchCompat(ctx, b, id); ok { + return enc, err + } start := time.Now() enc, err := r.dispatcher.Handle(ctx, b) dur := time.Since(start)