mtproto,rpc: support Android legacy startup

(cherry picked from commit 06b6ae2bb2f90bd7cc1d6a8404a82aff507e6821)

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,

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)

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 {

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})

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)),