From 2353cafe4406a3d0b1c206d54cf0987596efda38 Mon Sep 17 00:00:00 2001 From: highesttt Date: Fri, 28 Aug 2026 20:43:38 -0400 Subject: [PATCH] chatsync: classify requests using trust state --- pkg/connector/chatsync.go | 119 ++++++++++++++++++++++-- pkg/connector/chatsync_test.go | 42 +++++++++ pkg/connector/dbmeta.go | 1 + pkg/connector/handlematrix.go | 8 ++ pkg/connector/handletwit.go | 16 ++++ pkg/twittermeow/data/types/messaging.go | 2 + pkg/twittermeow/xchat_convert.go | 2 + pkg/twittermeow/xchat_processor.go | 1 + pkg/twittermeow/xchat_send_test.go | 8 ++ 9 files changed, 189 insertions(+), 10 deletions(-) diff --git a/pkg/connector/chatsync.go b/pkg/connector/chatsync.go index b0f134f..21dd90f 100644 --- a/pkg/connector/chatsync.go +++ b/pkg/connector/chatsync.go @@ -51,6 +51,96 @@ func isXChatPortalForLogin(portal *bridgev2.Portal, loginID networkid.UserLoginI return ok && meta.IsXChatConversation() } +func xchatInboxItemTrust(item *response.XChatInboxItem) *bool { + defer func() { _ = recover() }() + if item == nil { + return nil + } + encodedEvents := make([]string, 0, len(item.LatestMessageEvents)+len(item.EncodedMessageEvents)+len(item.LatestConversationKeyChangeEvents)+1) + encodedEvents = append(encodedEvents, item.LatestMessageEvents...) + encodedEvents = append(encodedEvents, item.EncodedMessageEvents...) + encodedEvents = append(encodedEvents, item.LatestConversationKeyChangeEvents...) + if item.LatestNotifiableMessageCreateEvent != "" { + encodedEvents = append(encodedEvents, item.LatestNotifiableMessageCreateEvent) + } + + var trusted *bool + latestSequenceID := "" + for _, encoded := range encodedEvents { + decoded, err := base64.StdEncoding.DecodeString(encoded) + if err != nil { + continue + } + var evt payload.MessageEvent + if err = payload.Decode(decoded, &evt); err != nil || evt.IsTrusted == nil { + continue + } + sequenceID := ptr.Val(evt.SequenceId) + if trusted != nil && (sequenceID == "" && latestSequenceID != "" || + sequenceID != "" && latestSequenceID != "" && compareIntStrings(sequenceID, latestSequenceID) < 0) { + continue + } + value := *evt.IsTrusted + trusted = &value + latestSequenceID = sequenceID + } + return trusted +} + +func applyXChatTrustToChatInfo(info *bridgev2.ChatInfo, trusted bool) { + info.MessageRequest = ptr.Ptr(!trusted) + info.ExtraUpdates = bridgev2.MergeExtraUpdaters(info.ExtraUpdates, func(_ context.Context, portal *bridgev2.Portal) bool { + meta, ok := portal.Metadata.(*PortalMetadata) + if !ok || meta == nil || (meta.XChatTrusted != nil && *meta.XChatTrusted == trusted) { + return false + } + meta.XChatTrusted = ptr.Ptr(trusted) + return true + }) +} + +func (tc *TwitterClient) syncXChatTrust( + ctx context.Context, + conversationID string, + trusted *bool, + timestamp time.Time, + streamOrder int64, +) bool { + if trusted == nil { + return true + } + if ctx == nil { + ctx = context.TODO() + } + portalKey := tc.MakePortalKeyFromID(conversationID) + portal, err := tc.connector.br.GetPortalByKey(ctx, portalKey) + if err != nil { + zerolog.Ctx(ctx).Warn().Err(err).Msg("Failed to get XChat portal for trust update") + return false + } + meta, _ := portal.Metadata.(*PortalMetadata) + messageRequest := !*trusted + if meta != nil && meta.XChatTrusted != nil && *meta.XChatTrusted == *trusted && portal.MessageRequest == messageRequest { + return true + } + + chatInfo := &bridgev2.ChatInfo{} + applyXChatTrustToChatInfo(chatInfo, *trusted) + if current := bridgev2.GetRemoteEventFromContext(ctx); current != nil && current.GetPortalKey().ID == portalKey.ID { + portal.UpdateInfo(ctx, chatInfo, tc.userLogin, nil, timestamp) + return true + } + return tc.queueXChatRemoteEventWithPortalRepair(ctx, conversationID, &simplevent.ChatInfoChange{ + EventMeta: simplevent.EventMeta{ + Type: bridgev2.RemoteEventChatInfoChange, + PortalKey: portalKey, + Timestamp: timestamp, + StreamOrder: streamOrder, + }, + ChatInfoChange: &bridgev2.ChatInfoChange{ChatInfo: chatInfo}, + }) +} + // TODO: Remove this repair after affected bridges have been reconnected func (tc *TwitterClient) repairExistingXChatMessageRequests(ctx context.Context) error { if tc.xchatRequestsRepaired { @@ -68,8 +158,13 @@ func (tc *TwitterClient) repairExistingXChatMessageRequests(ctx context.Context) } xchatRooms++ + meta := portal.Metadata.(*PortalMetadata) + messageRequest := false + if meta.XChatTrusted != nil { + messageRequest = !*meta.XChatTrusted + } wasMessageRequest := portal.MessageRequest - portal.MessageRequest = false + portal.MessageRequest = messageRequest internals := (*bridgev2.PortalInternals)(portal) stateKey, bridgeInfo := internals.GetBridgeInfo() for _, eventType := range []event.Type{event.StateBridge, event.StateHalfShotBridge} { @@ -81,7 +176,7 @@ func (tc *TwitterClient) repairExistingXChatMessageRequests(ctx context.Context) return fmt.Errorf("repair XChat message-request room state") } } - if !wasMessageRequest { + if wasMessageRequest == messageRequest { continue } if err = portal.Save(ctx); err != nil { @@ -102,6 +197,9 @@ func shouldEmitChatInfoUpdate(chatInfo *bridgev2.ChatInfo, portalRoomType databa if chatInfo == nil { return false } + if chatInfo.MessageRequest != nil || chatInfo.ExtraUpdates != nil { + return true + } // DM room title/avatar are member-derived, so always emit ChatInfoChange // for existing DM rooms to refresh stale member profile info. @@ -242,9 +340,12 @@ func (tc *TwitterClient) xchatItemToConversation(ctx context.Context, item *resp conv := &types.Conversation{ ConversationID: detail.ConversationID, - Trusted: true, // XChat conversations are always trusted + Trusted: true, Muted: detail.IsMuted, } + if trusted := xchatInboxItemTrust(item); trusted != nil { + conv.Trusted = *trusted + } // Determine conversation type based on conversation ID and metadata. if strings.HasPrefix(detail.ConversationID, "g") || detail.GroupMetadata != nil || @@ -384,11 +485,7 @@ func (tc *TwitterClient) xchatItemToChatInfo(ctx context.Context, item *response } } - // MessageRequest is true for untrusted conversations (message requests) - var messageRequest *bool - if conv != nil { - messageRequest = ptr.Ptr(!conv.Trusted) - } + trusted := xchatInboxItemTrust(item) membersAreFull := len(detail.GroupMembersResults) > 0 && len(memberMap) > 0 for _, memberResult := range detail.GroupMembersResults { if memberID, _ := xchatUserFromResult(memberResult); memberID == "" { @@ -418,8 +515,10 @@ func (tc *TwitterClient) xchatItemToChatInfo(ctx context.Context, item *response TotalMemberCount: len(memberMap), MemberMap: memberMap, }, - CanBackfill: true, - MessageRequest: messageRequest, + CanBackfill: true, + } + if trusted != nil { + applyXChatTrustToChatInfo(info, *trusted) } if isGroup { diff --git a/pkg/connector/chatsync_test.go b/pkg/connector/chatsync_test.go index 15701a8..078d81e 100644 --- a/pkg/connector/chatsync_test.go +++ b/pkg/connector/chatsync_test.go @@ -2,11 +2,53 @@ package connector import ( "context" + "encoding/base64" "testing" + "go.mau.fi/util/ptr" + "maunium.net/go/mautrix/bridgev2" + "maunium.net/go/mautrix/bridgev2/database" + + "go.mau.fi/mautrix-twitter/pkg/twittermeow/data/payload" "go.mau.fi/mautrix-twitter/pkg/twittermeow/data/response" ) +func TestXChatItemTrustControlsMessageRequest(t *testing.T) { + for _, trusted := range []*bool{ptr.Ptr(false), ptr.Ptr(true), nil} { + client := &TwitterClient{} + item := &response.XChatInboxItem{ + ConversationDetail: response.XChatConversationDetail{ConversationID: "g123"}, + LatestMessageEvents: []string{"AgABMA=="}, + } + if trusted != nil { + encoded, err := payload.Encode(&payload.MessageEvent{IsTrusted: trusted}) + if err != nil { + t.Fatal(err) + } + item.LatestMessageEvents = []string{base64.StdEncoding.EncodeToString(encoded)} + } + conv := client.xchatItemToConversation(t.Context(), item, nil) + if conv.Trusted != (trusted == nil || *trusted) { + t.Fatalf("Trusted = %t", conv.Trusted) + } + info := client.xchatItemToChatInfo(t.Context(), item, nil, conv) + if trusted == nil { + if info.MessageRequest != nil || info.ExtraUpdates != nil { + t.Fatal("missing trust changed message-request state") + } + return + } + if info.MessageRequest == nil || *info.MessageRequest == *trusted { + t.Fatalf("MessageRequest = %v", info.MessageRequest) + } + meta := &PortalMetadata{} + portal := &bridgev2.Portal{Portal: &database.Portal{Metadata: meta}} + if !info.ExtraUpdates(t.Context(), portal) || meta.XChatTrusted == nil || *meta.XChatTrusted != *trusted { + t.Fatalf("XChatTrusted = %v", meta.XChatTrusted) + } + } +} + func TestXChatItemToConversationPreservesPlaintextGroupName(t *testing.T) { tc := &TwitterClient{} item := &response.XChatInboxItem{ diff --git a/pkg/connector/dbmeta.go b/pkg/connector/dbmeta.go index a9c8ce4..1aed3e1 100644 --- a/pkg/connector/dbmeta.go +++ b/pkg/connector/dbmeta.go @@ -51,6 +51,7 @@ type PortalMetadata struct { // Server token for XChat API ConversationToken string `json:"conversation_token,omitempty"` + XChatTrusted *bool `json:"xchat_trusted,omitempty"` } // CanUseXChat returns true if this conversation has encryption keys diff --git a/pkg/connector/handlematrix.go b/pkg/connector/handlematrix.go index a2cd7cc..e23371c 100644 --- a/pkg/connector/handlematrix.go +++ b/pkg/connector/handlematrix.go @@ -1022,5 +1022,13 @@ func (tc *TwitterClient) HandleMatrixAcceptMessageRequest(ctx context.Context, m if errors.Is(err, twittermeow.ErrConversationDoesntExist) { err = nil } + if err == nil && msg != nil && msg.Portal != nil { + meta, ok := msg.Portal.Metadata.(*PortalMetadata) + if ok && meta.IsXChatConversation() { + chatInfo := &bridgev2.ChatInfo{} + applyXChatTrustToChatInfo(chatInfo, true) + msg.Portal.UpdateInfo(ctx, chatInfo, tc.userLogin, nil, time.Time{}) + } + } return err } diff --git a/pkg/connector/handletwit.go b/pkg/connector/handletwit.go index 82934ea..cfe2402 100644 --- a/pkg/connector/handletwit.go +++ b/pkg/connector/handletwit.go @@ -229,6 +229,9 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit return false } } + if !tc.syncXChatTrust(ctx, evt.ConversationID, evt.Trusted, methods.ParseMsecTimestamp(evt.Time), streamOrder) { + return false + } txnID := evt.RequestID if txnID == "" { @@ -283,6 +286,9 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit return false } } + if !tc.syncXChatTrust(ctx, evt.ConversationID, evt.Trusted, methods.ParseMsecTimestamp(evt.Time), streamOrder) { + return false + } txnID := evt.RequestID if txnID == "" { @@ -536,6 +542,15 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit Msg("Failed to ensure portal for ConversationCreate event") return false } + if !tc.syncXChatTrust( + ctx, + evt.ConversationID, + evt.Trusted, + methods.ParseMsecTimestamp(evt.Time), + methods.ParseSnowflakeInt(evt.ID), + ) { + return false + } // If the portal was just created or doesn't have a room yet, trigger a resync // to ensure it gets created and backfilled @@ -579,6 +594,7 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit Msg("Failed to get chat info for trusted conversation") return false } + applyXChatTrustToChatInfo(chatInfo, true) return xchatRemoteEventHandled(tc.userLogin.QueueRemoteEvent(&simplevent.ChatResync{ EventMeta: simplevent.EventMeta{ diff --git a/pkg/twittermeow/data/types/messaging.go b/pkg/twittermeow/data/types/messaging.go index bf521ca..dcdb086 100644 --- a/pkg/twittermeow/data/types/messaging.go +++ b/pkg/twittermeow/data/types/messaging.go @@ -30,6 +30,7 @@ type Message struct { SequenceID string `json:"sequence_id,omitempty"` RequestID string `json:"request_id,omitempty"` ConversationID string `json:"conversation_id,omitempty"` + Trusted *bool `json:"trusted,omitempty"` ConversationKeyVersion string `json:"conversation_key_version,omitempty"` MessageData MessageData `json:"message_data,omitempty"` MessageReactions []MessageReaction `json:"message_reactions,omitempty"` @@ -92,6 +93,7 @@ type ConversationCreate struct { AffectsSort bool `json:"affects_sort,omitempty"` ConversationID string `json:"conversation_id,omitempty"` RequestID string `json:"request_id,omitempty"` + Trusted *bool `json:"trusted,omitempty"` } type ConversationDelete struct { diff --git a/pkg/twittermeow/xchat_convert.go b/pkg/twittermeow/xchat_convert.go index 5716119..a4ca64c 100644 --- a/pkg/twittermeow/xchat_convert.go +++ b/pkg/twittermeow/xchat_convert.go @@ -58,6 +58,7 @@ func convertXChatMessageToTwitterMessage(evt *payload.MessageEvent, contents *pa SequenceID: seqID, RequestID: ptr.Val(evt.MessageId), ConversationID: ptr.Val(evt.ConversationId), + Trusted: evt.IsTrusted, ConversationKeyVersion: keyVersion, MessageData: msgData, } @@ -96,6 +97,7 @@ func convertXChatMessageEdit(evt *payload.MessageEvent, edit *payload.MessageEdi SequenceID: ptr.Val(evt.SequenceId), RequestID: ptr.Val(evt.MessageId), ConversationID: ptr.Val(evt.ConversationId), + Trusted: evt.IsTrusted, ConversationKeyVersion: keyVersion, MessageData: msgData, }) diff --git a/pkg/twittermeow/xchat_processor.go b/pkg/twittermeow/xchat_processor.go index 64f7ac5..9af9f47 100644 --- a/pkg/twittermeow/xchat_processor.go +++ b/pkg/twittermeow/xchat_processor.go @@ -546,6 +546,7 @@ func (p *XChatEventProcessor) emitConversationCreate(ctx context.Context, evt *p Time: ptr.Val(evt.CreatedAtMsec), ConversationID: ptr.Val(evt.ConversationId), RequestID: ptr.Val(evt.MessageId), + Trusted: evt.IsTrusted, }) } diff --git a/pkg/twittermeow/xchat_send_test.go b/pkg/twittermeow/xchat_send_test.go index 79370b3..76df2ff 100644 --- a/pkg/twittermeow/xchat_send_test.go +++ b/pkg/twittermeow/xchat_send_test.go @@ -11,9 +11,17 @@ import ( "github.com/rs/zerolog" "go.mau.fi/mautrix-twitter/pkg/twittermeow/cookies" + "go.mau.fi/mautrix-twitter/pkg/twittermeow/data/payload" "go.mau.fi/mautrix-twitter/pkg/twittermeow/data/response" ) +func TestConvertXChatMessagePreservesTrust(t *testing.T) { + trusted := false + if got := convertXChatMessageToTwitterMessage(&payload.MessageEvent{IsTrusted: &trusted}, &payload.MessageContents{}, ""); got.Trusted == nil || *got.Trusted { + t.Fatalf("Trusted = %v", got.Trusted) + } +} + func TestRefreshConversationKeysDoesNotBlockOnConversationDataCallback(t *testing.T) { client := NewClient(cookies.NewCookies(nil), nil, zerolog.Nop()) client.HTTP = &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {