fix: bound retained chat-reaction message IDs and validate relay identity

Explorer/Assets/DCL/Chat/MessageBus/LiveKitChatMessagesBus.cs

@@ -26,6 +26,11 @@ namespace DCL.Chat.MessageBus
 {
     public class LiveKitChatMessagesBus : IChatMessagesBus
     {
+        // Hard ceiling on the dedup stamps retained per period. A flood of distinct timestamps
+        // restarts the window instead of growing it, bounding the memory a sender can make this
+        // cache hold.
+        private const int MAX_DEDUP_ENTRIES = 2048;
+
         private readonly IMessagePipesHub messagePipesHub;
         private readonly IMessageDeduplication<double> messageDeduplication;
         private readonly CancellationTokenSource cancellationTokenSource = new ();
@@ -52,7 +57,7 @@ public LiveKitChatMessagesBus(IMessagePipesHub messagePipesHub,
             IRoomHub roomHub)
         {
             this.messagePipesHub = messagePipesHub;
-            messageDeduplication = new MessageDeduplication<double>();
+            messageDeduplication = new MessageDeduplication<double>(MAX_DEDUP_ENTRIES);
             this.userBlockingCache = userBlockingCache;
             this.identityCache = identityCache;
             this.messageFactory = messageFactory;

Explorer/Assets/DCL/Chat/MessageBus/Tests/LiveKitChatMessagesBusShould.cs

@@ -19,6 +19,7 @@
 using NUnit.Framework;
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using ChatMessage = DCL.Chat.History.ChatMessage;
 using ChatPacket = Decentraland.Kernel.Comms.Rfc4.Chat;
 
@@ -60,7 +61,7 @@ public void SetUp()
             FeatureFlagsConfiguration.Initialize(new FeatureFlagsConfiguration(FeatureFlagsResultDto.Empty));
             OfficialWalletsHelper.Initialize(new OfficialWalletsHelper());
             FeaturesRegistry.Initialize(new FeaturesRegistry(appArgs, false));
-            CommunitiesFeatureAccess.Initialize(new CommunitiesFeatureAccess(identityCache, appArgs));
+            CommunitiesFeatureAccess.Initialize(new CommunitiesFeatureAccess(identityCache, appArgs, CancellationToken.None));
             RoomMetadataCurrentScene.InitializeTest();
 
             pipesHub = new FakeMessagePipesHub();

Explorer/Assets/DCL/Chat/_Refactor/ChatReactions/Core/ChatReactionsFactory.cs

@@ -6,6 +6,7 @@
 using DCL.Chat.ChatReactions.Simulation.World;
 using DCL.Chat.History;
 using DCL.Diagnostics;
+using DCL.FeatureFlags;
 using DCL.Friends.UserBlocking;
 using DCL.Multiplayer.Connections.DecentralandUrls;
 using DCL.Multiplayer.Connections.Messaging.Hubs;
@@ -155,6 +156,15 @@ private static IReactionMessageBus CreateReactionBus(
             ChatReactionsConfig reactionsConfig,
             int maxValidEmojiIndex)
         {
+            // Gated here rather than only in the UI: subscribing the pipes would let any
+            // co-located peer drive reaction decoding and retention on a client that cannot
+            // show reactions at all.
+            if (!FeatureFlagsConfiguration.Instance.IsEnabled(FeatureFlagsStrings.CHAT_REACTIONS_ENABLED))
+            {
+                ReportHub.Log(ReportCategory.CHAT_MESSAGES, "[ChatPlugin] Chat reactions disabled — using NullReactionMessageBus (no pipes subscribed)");
+                return new NullReactionMessageBus();
+            }
+
             string serverEnv = environment switch
             {
                 DecentralandEnvironment.Org => "prd",

Explorer/Assets/DCL/Chat/_Refactor/ChatReactions/Networking/MultiplayerReactionMessageBus.cs

@@ -12,6 +12,7 @@
 using DCL.Multiplayer.Connections.Messaging.Pipe;
 using DCL.Multiplayer.Deduplication;
 using DCL.Utilities;
+using DCL.Web3;
 using DCL.Web3.Identities;
 using Decentraland.Kernel.Comms.Rfc4;
 using LiveKit.Proto;
@@ -27,15 +28,25 @@ public sealed class MultiplayerReactionMessageBus : IReactionMessageBus
         // The effective bound is tightened to the actual atlas tile count via the constructor.
         private const int MAX_VALID_EMOJI_INDEX = 4096;
 
+        // Message IDs are produced locally in one of two shapes: ChatUtils.GetId
+        // ("{42-char address}:{invariant double}", ~67 characters) or a 36-character GUID for
+        // system messages. 96 clears both with headroom while keeping an ID that arrives from
+        // the network bounded before it is used to build anything.
+        private const int MAX_MESSAGE_ID_LENGTH = 96;
+
+        // Hard ceiling on the dedup keys either cache retains. A flood of distinct keys restarts
+        // the window instead of growing it, bounding the memory a peer can make this bus hold.
+        private const int MAX_DEDUP_ENTRIES = 2048;
+
         private readonly IMessagePipesHub messagePipesHub;
         private readonly IUserBlockingCache userBlockingCache;
         private readonly IWeb3IdentityCache identityCache;
         private readonly string routingUser;
         private readonly int maxValidEmojiIndex;
         private readonly int situationalReceiveCountCap;
         private readonly CancellationTokenSource cts = new ();
-        private readonly IMessageDeduplication<float> situationalDedup = new MessageDeduplication<float>();
-        private readonly IMessageDeduplication<string> chatReactionDedup = new MessageDeduplication<string>();
+        private readonly IMessageDeduplication<float> situationalDedup = new MessageDeduplication<float>(MAX_DEDUP_ENTRIES);
+        private readonly IMessageDeduplication<string> chatReactionDedup = new MessageDeduplication<string>(MAX_DEDUP_ENTRIES);
         private readonly PerSenderRateLimiter situationalRateLimiter;
         private readonly PerSenderRateLimiter chatReactionRateLimiter;
 
@@ -187,25 +198,25 @@ private void OnChatReactionReceived(ReceivedMessage<ChatReaction> receivedMessag
         {
             using (receivedMessage)
             {
-                // For community messages relayed through the message-router, FromWalletId
-                // is the relay's identity, not the original sender. Fall back to Payload.Address
-                // only in that case. For direct messages (nearby/DM), always use the trusted
-                // transport-level FromWalletId to prevent identity spoofing.
-                // NOTE: Add server-stamped ForwardedFrom to the ChatReaction protocol (like Chat
-                // has) so the relay writes the verified sender identity, not the client.
-                string walletId = receivedMessage.FromWalletId == routingUser
-                    && !string.IsNullOrEmpty(receivedMessage.Payload.Address)
-                    ? receivedMessage.Payload.Address
-                    : receivedMessage.FromWalletId;
-
-                ReportHub.Log(ReportCategory.CHAT_MESSAGES, $"[MultiplayerReactionBus] OnChatReactionReceived raw: emoji={receivedMessage.Payload.EmojiIndex} messageId={receivedMessage.Payload.MessageId} address={receivedMessage.Payload.Address} fromWallet={receivedMessage.FromWalletId}");
-
                 if (cts.IsCancellationRequested)
                 {
                     ReportHub.LogWarning(ReportCategory.CHAT_MESSAGES, "[MultiplayerReactionBus] OnChatReactionReceived skipped — CTS cancelled");
                     return;
                 }
 
+                string messageId = receivedMessage.Payload.MessageId;
+
+                // Checked before the dedup keys exist, so an ID no local message could carry
+                // never reaches the dedup cache. Dropped silently: ReportHub.LogWarning is not
+                // compiled out, so naming the ID here would allocate it once per packet in
+                // retail builds.
+                if (string.IsNullOrEmpty(messageId) || messageId.Length > MAX_MESSAGE_ID_LENGTH)
+                    return;
+
+                string walletId = ResolveSenderWalletId(receivedMessage);
+
+                ReportHub.Log(ReportCategory.CHAT_MESSAGES, $"[MultiplayerReactionBus] OnChatReactionReceived raw: emoji={receivedMessage.Payload.EmojiIndex} messageId={messageId} fromWallet={receivedMessage.FromWalletId} resolved={walletId}");
+
                 if (IsUserBlocked(walletId))
                 {
                     ReportHub.Log(ReportCategory.CHAT_MESSAGES, $"[MultiplayerReactionBus] OnChatReactionReceived skipped — user blocked: {walletId}");
@@ -223,8 +234,8 @@ private void OnChatReactionReceived(ReceivedMessage<ChatReaction> receivedMessag
 
                 // Use raw value in dedup key so add/remove have distinct keys.
                 // Evict the opposite key so toggling (add→remove→add) isn't blocked.
-                string dedupKey = $"{receivedMessage.Payload.MessageId}:{rawEmojiIndex}";
-                string oppositeKey = $"{receivedMessage.Payload.MessageId}:{ReactionWireEncoding.Encode(emojiIndex, !isRemoval)}";
+                string dedupKey = $"{messageId}:{rawEmojiIndex}";
+                string oppositeKey = $"{messageId}:{ReactionWireEncoding.Encode(emojiIndex, !isRemoval)}";
 
                 if (!chatReactionDedup.TryPass(walletId, dedupKey))
                 {
@@ -239,18 +250,37 @@ private void OnChatReactionReceived(ReceivedMessage<ChatReaction> receivedMessag
                 if (!chatReactionRateLimiter.TryPass(walletId, UnityEngine.Time.unscaledTime))
                     return;
 
-                ReportHub.Log(ReportCategory.CHAT_MESSAGES, $"[MultiplayerReactionBus] Received chat reaction: emoji={emojiIndex} isRemoval={isRemoval} messageId={receivedMessage.Payload.MessageId} from={walletId}");
+                ReportHub.Log(ReportCategory.CHAT_MESSAGES, $"[MultiplayerReactionBus] Received chat reaction: emoji={emojiIndex} isRemoval={isRemoval} messageId={messageId} from={walletId}");
 
                 ReactionReceived?.Invoke(new ReactionReceivedArgs(
                     walletId,
                     emojiIndex,
                     1,
                     ReactionType.Message,
-                    receivedMessage.Payload.MessageId,
+                    messageId,
                     isRemoval));
             }
         }
 
+        private string ResolveSenderWalletId(ReceivedMessage<ChatReaction> receivedMessage)
+        {
+            // Community reactions relayed through the message-router arrive carrying the relay's
+            // identity in FromWalletId, so the original sender can only come from the payload.
+            // Reading it solely from the router, and solely when it is a canonical wallet
+            // address, keeps a direct peer from asserting an arbitrary identity and keeps the
+            // block, dedup and rate-limit keys bounded.
+            // NOTE: Add a server-stamped ForwardedFrom to the ChatReaction protocol (like Chat
+            // has) so the relay writes the verified sender identity, not the client.
+            if (receivedMessage.FromWalletId != routingUser)
+                return receivedMessage.FromWalletId;
+
+            string relayedAddress = receivedMessage.Payload.Address;
+
+            return Web3Address.IsValidWalletAddress(relayedAddress)
+                ? relayedAddress
+                : receivedMessage.FromWalletId;
+        }
+
         private bool IsUserBlocked(string userAddress) =>
             userBlockingCache.UserIsBlocked(userAddress);
     }

Explorer/Assets/DCL/Chat/_Refactor/ChatReactions/Networking/NullReactionMessageBus.cs

@@ -0,0 +1,27 @@
+using System;
+
+namespace DCL.Chat.ChatReactions.Networking
+{
+    /// <summary>
+    ///     Inert bus used while the chat-reactions feature flag is off. It subscribes to no
+    ///     message pipe and drops every send, so no reaction traffic is decoded, retained or
+    ///     emitted while the feature is unavailable.
+    /// </summary>
+    public sealed class NullReactionMessageBus : IReactionMessageBus
+    {
+        /// <summary>
+        ///     Never raised — nothing is subscribed, so no reaction can arrive.
+        /// </summary>
+        public event Action<ReactionReceivedArgs> ReactionReceived
+        {
+            add { }
+            remove { }
+        }
+
+        public void Dispose() { }
+
+        public void SendSituationalReaction(int emojiIndex, int count = 1, float overrideTimestamp = 0f) { }
+
+        public void SendMessageReaction(int emojiIndex, string messageId, ReactionChannelRouting routing) { }
+    }
+}

Explorer/Assets/DCL/Chat/_Refactor/ChatReactions/Tests/MultiplayerReactionMessageBusShould.cs

@@ -21,6 +21,17 @@ namespace DCL.Chat.ChatReactions.Tests
     [TestFixture]
     public class MultiplayerReactionMessageBusShould
     {
+        private const string ROUTING_USER = "message-router-test-0";
+
+        // A canonical wallet address, so Web3Address.IsValidWalletAddress accepts it.
+        private const string RELAYED_WALLET = "0xabcdefabcdefabcdefabcdefabcdefabcdefabcd";
+
+        // One more than MultiplayerReactionMessageBus.MAX_DEDUP_ENTRIES, so a flood of this many
+        // distinct keys is guaranteed to drive the dedup cache to its ceiling.
+        private const int FLOOD_BEYOND_DEDUP_CAPACITY = 2049;
+
+        private const int OVERSIZED_MESSAGE_ID_LENGTH = 5000;
+
         private FakeMessagePipesHub pipesHub = null!;
         private IMultiPool multiPool = null!;
         private List<ReactionReceivedArgs> received = null!;
@@ -162,6 +173,105 @@ public void DeduplicateSameChatReactionAcrossPipes()
             Assert.That(received.Count, Is.EqualTo(1));
         }
 
+        [Test]
+        public void RejectChatReactionWithOversizedMessageId()
+        {
+            // Arrange
+            CreateBus();
+
+            // Act — an ID far longer than any the client itself can produce
+            DeliverChatReaction("0xattacker", wireEmojiIndex: 1, messageId: new string('x', OVERSIZED_MESSAGE_ID_LENGTH));
+
+            // Assert
+            Assert.That(received, Is.Empty);
+        }
+
+        [Test]
+        public void RejectChatReactionWithEmptyMessageId()
+        {
+            // Arrange
+            CreateBus();
+
+            // Act
+            DeliverChatReaction("0xattacker", wireEmojiIndex: 1, messageId: string.Empty);
+
+            // Assert
+            Assert.That(received, Is.Empty);
+        }
+
+        [Test]
+        public void NotRetainOversizedMessageIdsInTheDedupCache()
+        {
+            // Arrange — a budget wide enough that the rate limiter cannot mask dedup behaviour
+            WidenReceiveBudget();
+            CreateBus();
+
+            var oversizedId = new string('x', OVERSIZED_MESSAGE_ID_LENGTH);
+            DeliverChatReaction("0xattacker", wireEmojiIndex: 1, messageId: "msg1");
+
+            // Act — were these retained, they would fill the dedup cache and drop the window
+            for (int i = 0; i < FLOOD_BEYOND_DEDUP_CAPACITY; i++)
+                DeliverChatReaction("0xattacker", wireEmojiIndex: 1, messageId: oversizedId + i);
+
+            DeliverChatReaction("0xattacker", wireEmojiIndex: 1, messageId: "msg1");
+
+            // Assert — the first legitimate reaction is still the only one that got through,
+            // so its dedup key survived the whole flood
+            Assert.That(received.Count, Is.EqualTo(1));
+        }
+
+        [Test]
+        public void BoundTheDedupCacheUnderAFloodOfDistinctMessageIds()
+        {
+            // Arrange — a budget wide enough that the rate limiter cannot mask dedup behaviour
+            WidenReceiveBudget();
+            CreateBus();
+
+            DeliverChatReaction("0xattacker", wireEmojiIndex: 1, messageId: "msg-keep");
+
+            // Act — well-formed IDs are retained, so this drives the cache to its ceiling
+            for (int i = 0; i < FLOOD_BEYOND_DEDUP_CAPACITY; i++)
+                DeliverChatReaction("0xattacker", wireEmojiIndex: 1, messageId: $"msg-flood-{i}");
+
+            DeliverChatReaction("0xattacker", wireEmojiIndex: 1, messageId: "msg-keep");
+
+            // Assert — hitting the ceiling dropped the window instead of growing it, so the
+            // original key is no longer held and its reaction passes dedup a second time
+            Assert.That(received.Count, Is.EqualTo(FLOOD_BEYOND_DEDUP_CAPACITY + 2));
+            Assert.That(received[received.Count - 1].MessageId, Is.EqualTo("msg-keep"));
+        }
+
+        [Test]
+        public void AttributeRelayedReactionToTheRouterWhenPayloadAddressIsMalformed()
+        {
+            // Arrange
+            CreateBus();
+
+            // Act — the relay forwards a payload whose address is not a wallet address at all
+            DeliverChatReaction(ROUTING_USER, wireEmojiIndex: 1, messageId: "msg1",
+                address: new string('x', OVERSIZED_MESSAGE_ID_LENGTH));
+
+            // Assert — attribution falls back to the transport-level identity
+            Assert.That(received.Count, Is.EqualTo(1));
+            Assert.That(received[0].WalletId, Is.EqualTo(ROUTING_USER));
+        }
+
+        [Test]
+        public void HonourRelayedAddressOnlyWhenItComesFromTheRouter()
+        {
+            // Arrange
+            CreateBus();
+
+            // Act
+            DeliverChatReaction(ROUTING_USER, wireEmojiIndex: 1, messageId: "msg1", address: RELAYED_WALLET);
+            DeliverChatReaction("0xattacker", wireEmojiIndex: 1, messageId: "msg2", address: RELAYED_WALLET);
+
+            // Assert
+            Assert.That(received.Count, Is.EqualTo(2));
+            Assert.That(received[0].WalletId, Is.EqualTo(RELAYED_WALLET), "The router's forwarded sender is trusted");
+            Assert.That(received[1].WalletId, Is.EqualTo("0xattacker"), "A direct peer cannot claim another identity");
+        }
+
         // ── Test helpers ─────────────────────────────────────────
 
         private void CreateBus(int maxValidEmojiIndex = 4096)
@@ -170,24 +280,35 @@ private void CreateBus(int maxValidEmojiIndex = 4096)
                 pipesHub,
                 Substitute.For<IUserBlockingCache>(),
                 Substitute.For<IWeb3IdentityCache>(),
-                routingUser: "message-router-test-0",
+                routingUser: ROUTING_USER,
                 config: config,
                 maxValidEmojiIndex: maxValidEmojiIndex);
 
             newBus.ReactionReceived += args => received.Add(args);
             bus = newBus;
         }
 
+        /// <summary>
+        ///     Raises the per-sender receive budget past any flood these tests emit, so an
+        ///     assertion about retention can only be explained by the dedup cache.
+        /// </summary>
+        private void WidenReceiveBudget()
+        {
+            config.ChatReactionReceiveRatePerSecond = 60f;
+            config.ChatReactionReceiveBurst = FLOOD_BEYOND_DEDUP_CAPACITY * 2;
+        }
+
         private void DeliverSituational(string fromWallet, int emojiIndex, int count, float timestamp)
         {
             var payload = new Reaction { EmojiIndex = emojiIndex, Count = count, Timestamp = timestamp };
             pipesHub.Island.Deliver(Packet.MessageOneofCase.Reaction,
                 new ReceivedMessage<Reaction>(payload, new Packet(), fromWallet, multiPool, RoomSource.Island, string.Empty));
         }
 
-        private void DeliverChatReaction(string fromWallet, int wireEmojiIndex, string messageId, FakeMessagePipe? pipe = null)
+        private void DeliverChatReaction(string fromWallet, int wireEmojiIndex, string messageId,
+            FakeMessagePipe? pipe = null, string address = "")
         {
-            var payload = new ChatReaction { EmojiIndex = wireEmojiIndex, MessageId = messageId, Address = string.Empty };
+            var payload = new ChatReaction { EmojiIndex = wireEmojiIndex, MessageId = messageId, Address = address };
             (pipe ?? pipesHub.Island).Deliver(Packet.MessageOneofCase.ChatReaction,
                 new ReceivedMessage<ChatReaction>(payload, new Packet(), fromWallet, multiPool, RoomSource.Island, string.Empty));
         }