fix(pubsub): verify message from matches signing key and prevent seen-cache poisoning

Closes a strict-sign origin spoofing vulnerability in the pubsub stack. A peer
with any valid signing key could publish a signed message whose `from` named a
different peer; receivers attributed the message to the impersonated peer and
the forged `from || seqno` poisoned the seen-cache, suppressing the legitimate
follow-up message from the real author.

Three changes:

* PubsubCrypto.pubsubValidate now rejects messages missing/empty `from`, `key`
  or `signature`, returns false (never throws) on un-parseable key/from, and
  requires `PeerId.fromPubKey(msg.key) == PeerId(msg.from)` in addition to the
  existing signature check.

* PubsubApiImpl.PublisherImpl.publishExt rejects any caller-supplied `from`
  that does not match the publisher's signing key, throwing
  IllegalArgumentException. Null `from` continues to default to the signing
  key's PeerId as before; unsigned publishers (no privKey) are unchanged.

* AbstractRouter no longer leaves a seenMessages entry behind for a message
  that fails validation. On both sync and async validation failure the entry
  inserted speculatively at the start of `onInbound` is evicted, so a later
  legitimate message with the same id is still processed and delivered.

Adds PubsubStrictSignSpoofTest covering forged-from rejection, missing-field
rejection, publishExt mismatch rejection, the matching-from happy path, and an
end-to-end regression test that the seen-cache is no longer poisoned by a
rejected forgery.

Author: lucassaldanha

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

@@ -226,7 +226,11 @@ abstract class AbstractRouter(
                 true
             } catch (e: Exception) {
                 logger.debug("Invalid pubsub message from peer {}: {}", peer, it, e)
-                seenMessages[it] = Optional.of(ValidationResult.Invalid)
+                // Evict the speculatively-inserted seen-cache entry so that a later
+                // legitimate message with the same id (e.g. same from||seqno) is not
+                // suppressed by a forged predecessor that failed validation.
+                // See security issue 01 (pubsub strict-sign spoof / seen-cache poisoning).
+                seenMessages -= it.messageId
                 notifyUnseenInvalidMessage(peer, it)
                 false
             }
@@ -240,8 +244,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,16 @@ 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. Otherwise we'd be signing a
+            // message that claims a different origin — see security issue 01
+            // (pubsub strict-sign spoof).
+            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()
     )