Merge branch 'develop' into feature/quic-issues

Author: lucassaldanha

libp2p/src/main/kotlin/io/libp2p/mux/mplex/MplexFrameCodec.kt

@@ -73,7 +73,12 @@ class MplexFrameCodec(
             val streamId = header.shr(3)
             val data = msg.readSlice(lenData.toInt())
             data.retain() // MessageToMessageCodec releases original buffer, but it needs to be relayed
-            val flag = MplexFlag.getByValue(streamTag)
+            val flag = try {
+                MplexFlag.getByValue(streamTag)
+            } catch (e: IllegalArgumentException) {
+                data.release()
+                throw e
+            }
             val mplexFrame = MplexFrame(MplexId(ctx.channel().id(), streamId, !flag.isInitiator), flag, data)
             out.add(mplexFrame)
         }

libp2p/src/main/kotlin/io/libp2p/mux/mplex/MplexHandler.kt

@@ -26,9 +26,18 @@ open class MplexHandler(
     override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
         msg as MplexFrame
         when (msg.flag.type) {
-            MplexFlag.Type.OPEN -> onRemoteOpen(msg.id)
-            MplexFlag.Type.CLOSE -> onRemoteDisconnect(msg.id)
-            MplexFlag.Type.RESET -> onRemoteClose(msg.id)
+            MplexFlag.Type.OPEN -> {
+                onRemoteOpen(msg.id)
+                msg.release()
+            }
+            MplexFlag.Type.CLOSE -> {
+                onRemoteDisconnect(msg.id)
+                msg.release()
+            }
+            MplexFlag.Type.RESET -> {
+                onRemoteClose(msg.id)
+                msg.release()
+            }
             MplexFlag.Type.DATA -> childRead(msg.id, msg.data)
         }
     }

libp2p/src/main/kotlin/io/libp2p/pubsub/AbstractRouter.kt

@@ -83,6 +83,15 @@ abstract class AbstractRouter(
         return true
     }
 
+    /**
+     * Per-router caps on repeated-field counts inside inbound RPCs. Enforced before
+     * protobuf materialisation by an [RpcCountFrameDecoder] inserted into the stream
+     * pipeline. Defaults to [PubsubRpcLimits.NONE] (no pre-decode cap). Subclasses
+     * with configured limits (e.g. [io.libp2p.pubsub.gossip.GossipRouter]) override.
+     */
+    protected open val rpcLimits: PubsubRpcLimits
+        get() = PubsubRpcLimits.NONE
+
     /**
      * Flushes all pending message parts for all peers
      */
@@ -113,6 +122,7 @@ abstract class AbstractRouter(
         with(streamHandler.stream) {
             pushHandler(LimitedProtobufVarint32FrameDecoder(maxMsgSize))
             pushHandler(ProtobufVarint32LengthFieldPrepender())
+            pushHandler(RpcCountFrameDecoder(rpcLimits))
             pushHandler(ProtobufDecoder(Rpc.RPC.getDefaultInstance()))
             pushHandler(ProtobufEncoder())
             handler?.also { pushHandler(it) }
@@ -216,7 +226,9 @@ abstract class AbstractRouter(
                 true
             } catch (e: Exception) {
                 logger.debug("Invalid pubsub message from peer {}: {}", peer, it, e)
-                seenMessages[it] = Optional.of(ValidationResult.Invalid)
+                // Avoid rejecting a future legitimate message with the same id
+                // (e.g. same from||seqno)
+                seenMessages -= it.messageId
                 notifyUnseenInvalidMessage(peer, it)
                 false
             }
@@ -230,8 +242,14 @@ abstract class AbstractRouter(
         validFuts.forEach { (msg, validationFut) ->
             validationFut.thenAcceptAsync(
                 { res ->
-                    seenMessages[msg] = Optional.of(res)
-                    if (res == ValidationResult.Invalid) notifyUnseenInvalidMessage(peer, msg)
+                    if (res == ValidationResult.Invalid) {
+                        // Evict so a later legitimate message with the same id is not
+                        // suppressed by this rejected one.
+                        seenMessages -= msg.messageId
+                        notifyUnseenInvalidMessage(peer, msg)
+                    } else {
+                        seenMessages[msg] = Optional.of(res)
+                    }
                 },
                 executor
             )

libp2p/src/main/kotlin/io/libp2p/pubsub/PubsubApiImpl.kt

@@ -42,6 +42,14 @@ open class PubsubApiImpl(val router: PubsubRouter) : PubsubApi {
             seqId: Long?,
             vararg topics: Topic
         ): CompletableFuture<Unit> {
+            // If the caller supplies a `from` AND we have a signing key, the `from` MUST
+            // resolve to the same PeerId as the signing key
+            if (from != null && privKey != null) {
+                val derived = PeerId.fromPubKey(privKey.publicKey()).bytes
+                require(from.contentEquals(derived)) {
+                    "publishExt 'from' (${PeerId(from)}) does not match the signing key's PeerId (${PeerId(derived)})"
+                }
+            }
             val mFrom = from?.toProtobuf() ?: this.from
             val mSeqId = seqId ?: seqIdGenerator()
 

libp2p/src/main/kotlin/io/libp2p/pubsub/PubsubCrypto.kt

@@ -1,13 +1,17 @@
 package io.libp2p.pubsub
 
+import io.libp2p.core.PeerId
 import io.libp2p.core.crypto.PrivKey
 import io.libp2p.core.crypto.marshalPublicKey
 import io.libp2p.core.crypto.unmarshalPublicKey
 import io.libp2p.etc.types.toProtobuf
+import org.slf4j.LoggerFactory
 import pubsub.pb.Rpc
 
 val SignPrefix = "libp2p-pubsub:".toByteArray()
 
+private val logger = LoggerFactory.getLogger("io.libp2p.pubsub.PubsubCrypto")
+
 fun pubsubSign(msg: Rpc.Message, key: PrivKey): Rpc.Message {
     if (msg.hasKey() || msg.hasSignature()) throw IllegalArgumentException("Message to sign should not contain 'key' or 'signature' fields")
     val signature = key.sign(SignPrefix + msg.toByteArray())
@@ -17,12 +21,59 @@ fun pubsubSign(msg: Rpc.Message, key: PrivKey): Rpc.Message {
         .build()
 }
 
+/**
+ * Strict-sign validation for a pubsub [Rpc.Message]:
+ *
+ *  - `from`, `key` and `signature` MUST all be present and non-empty.
+ *  - `signature` MUST verify under `key` over the SSZ-style serialised message with
+ *    `signature` and `key` cleared.
+ *  - `PeerId.fromPubKey(key)` MUST equal `PeerId(from)`, otherwise an attacker holding
+ *    any valid private key could publish messages whose `from` claims a different peer.
+ *
+ * Returns `false` on any validation failure. Never throws.
+ */
 fun pubsubValidate(msg: Rpc.Message): Boolean {
+    if (!msg.hasFrom() || msg.from.isEmpty) {
+        logger.debug("Rejecting pubsub message with missing/empty 'from'")
+        return false
+    }
+    if (!msg.hasKey() || msg.key.isEmpty) {
+        logger.debug("Rejecting pubsub message with missing/empty 'key'")
+        return false
+    }
+    if (!msg.hasSignature() || msg.signature.isEmpty) {
+        logger.debug("Rejecting pubsub message with missing/empty 'signature'")
+        return false
+    }
+
+    val publicKey = try {
+        unmarshalPublicKey(msg.key.toByteArray())
+    } catch (e: Exception) {
+        logger.debug("Rejecting pubsub message with un-parseable 'key': {}", e.toString())
+        return false
+    }
+
+    val keyDerivedPeerId = PeerId.fromPubKey(publicKey)
+    val claimedFrom = try {
+        PeerId(msg.from.toByteArray())
+    } catch (e: Exception) {
+        logger.debug("Rejecting pubsub message with un-parseable 'from': {}", e.toString())
+        return false
+    }
+    if (keyDerivedPeerId != claimedFrom) {
+        logger.debug(
+            "Rejecting pubsub message: from={} does not match PeerId(key)={}",
+            claimedFrom,
+            keyDerivedPeerId
+        )
+        return false
+    }
+
     val msgToSign = Rpc.Message.newBuilder(msg)
         .clearSignature()
         .clearKey()
         .build()
-    return unmarshalPublicKey(msg.key.toByteArray()).verify(
+    return publicKey.verify(
         SignPrefix + msgToSign.toByteArray(),
         msg.signature.toByteArray()
     )

libp2p/src/main/kotlin/io/libp2p/pubsub/PubsubRpcLimits.kt

@@ -0,0 +1,60 @@
+package io.libp2p.pubsub
+
+/**
+ * Per-router limits on repeated-field counts inside an inbound pubsub RPC. Enforced
+ * at decode time by [RpcMessageCountValidator] to prevent allocation amplification
+ * before [pubsub.pb.Rpc.RPC] is materialised.
+ *
+ * A null field means "no limit" — same semantics as the corresponding nullable
+ * fields on `GossipParams`.
+ */
+data class PubsubRpcLimits(
+    val maxPublishedMessages: Int?,
+    val maxTopicsPerPublishedMessage: Int?,
+    val maxSubscriptions: Int?,
+    val maxIHaveMessageIds: Int?,
+    val maxIWantMessageIds: Int?,
+    val maxGraftMessages: Int?,
+    val maxPruneMessages: Int?,
+    val maxPeersPerPruneMessage: Int?,
+    val maxIDontWantMessages: Int? = null,
+    val maxIDontWantMessageIds: Int? = null,
+    val rejectEmptyPublishEntries: Boolean = true,
+    val rejectEmptyIDontWantEntries: Boolean = true,
+) {
+    /**
+     * True when no configured limit or reject-flag can fire. Lets
+     * [RpcCountFrameDecoder] skip the validator walk entirely on the toggle-off
+     * path. Any new field added to this data class must be considered here.
+     */
+    val isNoop: Boolean =
+        maxPublishedMessages == null &&
+            maxTopicsPerPublishedMessage == null &&
+            maxSubscriptions == null &&
+            maxIHaveMessageIds == null &&
+            maxIWantMessageIds == null &&
+            maxGraftMessages == null &&
+            maxPruneMessages == null &&
+            maxPeersPerPruneMessage == null &&
+            maxIDontWantMessages == null &&
+            maxIDontWantMessageIds == null &&
+            !rejectEmptyPublishEntries &&
+            !rejectEmptyIDontWantEntries
+
+    companion object {
+        val NONE = PubsubRpcLimits(
+            maxPublishedMessages = null,
+            maxTopicsPerPublishedMessage = null,
+            maxSubscriptions = null,
+            maxIHaveMessageIds = null,
+            maxIWantMessageIds = null,
+            maxGraftMessages = null,
+            maxPruneMessages = null,
+            maxPeersPerPruneMessage = null,
+            maxIDontWantMessages = null,
+            maxIDontWantMessageIds = null,
+            rejectEmptyPublishEntries = false,
+            rejectEmptyIDontWantEntries = false,
+        )
+    }
+}

libp2p/src/main/kotlin/io/libp2p/pubsub/RpcCountFrameDecoder.kt

@@ -0,0 +1,59 @@
+package io.libp2p.pubsub
+
+import io.netty.buffer.ByteBuf
+import io.netty.channel.ChannelHandlerContext
+import io.netty.handler.codec.CorruptedFrameException
+import io.netty.handler.codec.MessageToMessageDecoder
+import org.slf4j.LoggerFactory
+
+/**
+ * Pre-decode count cap for inbound pubsub RPC frames. Sits between
+ * [io.libp2p.etc.util.netty.protobuf.LimitedProtobufVarint32FrameDecoder] (byte-size
+ * cap) and [io.netty.handler.codec.protobuf.ProtobufDecoder] (materialisation).
+ *
+ * For each frame, delegates to [RpcMessageCountValidator]. Accepted frames are
+ * forwarded unchanged as a `ByteBuf` to the next handler. Frames rejected because
+ * a configured count limit was exceeded are dropped with a debug log; no
+ * `Rpc$Message` is allocated for them. Frames rejected because the protobuf bytes
+ * themselves are malformed propagate a [CorruptedFrameException] so that
+ * downstream handlers (e.g. [io.libp2p.pubsub.AbstractRouter.onPeerWireException])
+ * can apply the same behaviour penalty they would have on a [ProtobufDecoder]
+ * failure.
+ *
+ * When [limits] is a no-op (see [PubsubRpcLimits.isNoop], e.g. [PubsubRpcLimits.NONE])
+ * the validator is skipped entirely and the buffer is forwarded as-is. Malformed
+ * bytes still surface downstream from [ProtobufDecoder], which already triggers
+ * the same wire-exception path the validator would have used.
+ */
+class RpcCountFrameDecoder(private val limits: PubsubRpcLimits) : MessageToMessageDecoder<ByteBuf>() {
+
+    override fun decode(ctx: ChannelHandlerContext, msg: ByteBuf, out: MutableList<Any>) {
+        if (limits.isNoop) {
+            out.add(msg.retain())
+            return
+        }
+
+        val result = try {
+            RpcMessageCountValidator.validate(msg, limits)
+        } catch (e: Exception) {
+            logger.debug("Dropping pubsub RPC frame due to unexpected validator error", e)
+            return
+        }
+
+        when (result) {
+            RpcMessageCountValidator.Result.Accepted -> {
+                out.add(msg.retain())
+            }
+            is RpcMessageCountValidator.Result.Malformed -> {
+                throw CorruptedFrameException(result.reason)
+            }
+            is RpcMessageCountValidator.Result.Rejected -> {
+                logger.debug("Dropping pubsub RPC frame: {}", result.reason)
+            }
+        }
+    }
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(RpcCountFrameDecoder::class.java)
+    }
+}