MM-64531: [Shared Channels] Users on different remote servers should not communicate unless the remotes have established secure connection. (#30985) (#33434)

Automatic Merge

server/platform/services/sharedchannel/service.go

@@ -329,6 +329,29 @@ func (scs *Service) postUnshareNotification(channelID string, creatorID string,
 	}
 }
 
+// IsRemoteClusterDirectlyConnected checks if a remote cluster has a direct connection to the current server
+func (scs *Service) IsRemoteClusterDirectlyConnected(remoteId string) bool {
+	if remoteId == "" {
+		return true // Local server is always "directly connected"
+	}
+
+	// Check if the remote cluster exists and confirmed
+	rc, err := scs.server.GetStore().RemoteCluster().Get(remoteId, false)
+	if err != nil {
+		return false
+	}
+
+	isConfirmed := rc.IsConfirmed()
+	hasCreator := rc.CreatorId != ""
+
+	// For a direct connection, the remote cluster must be confirmed AND have a creator
+	// (someone on this server initiated or accepted the connection)
+	// Remote clusters known only through synthetic users won't have a creator
+	directConnection := isConfirmed && hasCreator
+
+	return directConnection
+}
+
 // OnReceiveSyncMessageForTesting is a wrapper to expose onReceiveSyncMessage for testing purposes
 // isGlobalUserSyncEnabled checks if the global user sync feature is enabled
 func (scs *Service) isGlobalUserSyncEnabled() bool {

server/platform/services/sharedchannel/sync_recv.go

@@ -298,7 +298,15 @@ func (scs *Service) upsertSyncUser(c request.CTX, user *model.User, channel *mod
 	var userSaved *model.User
 	if euser == nil {
 		// new user.  Make sure the remoteID is correct and insert the record
+		// Preserve original remote ID before overwriting RemoteId
+		originalRemoteId := user.GetRemoteID()
 		user.RemoteId = model.NewPointer(rc.RemoteId)
+		if user.Props == nil || user.Props[model.UserPropsKeyOriginalRemoteId] == "" {
+			if originalRemoteId == "" {
+				originalRemoteId = rc.RemoteId // If no original RemoteId, use current sync sender
+			}
+			user.SetProp(model.UserPropsKeyOriginalRemoteId, originalRemoteId)
+		}
 		if userSaved, err = scs.insertSyncUser(c, user, channel, rc); err != nil {
 			return nil, err
 		}