gossip validate

example/03-gossip/gossip_anonymous_example.cpp

@@ -77,8 +77,7 @@ int main(int argc, char **argv) {
       libp2p::injector::useTransportAdaptors<
           libp2p::transport::QuicTransport>());
 
-  auto io_context =
-      injector.create<std::shared_ptr<boost::asio::io_context>>();
+  auto io_context = injector.create<std::shared_ptr<boost::asio::io_context>>();
   auto host = injector.create<std::shared_ptr<libp2p::host::BasicHost>>();
   auto gossip =
       injector.create<std::shared_ptr<libp2p::protocol::gossip::Gossip>>();
@@ -99,7 +98,7 @@ int main(int argc, char **argv) {
   });
 
   // Subscribe to topic
-  auto topic = gossip->subscribe("example");
+  auto topic = gossip->subscribe("example", false);
 
   // Receiver task
   libp2p::coroSpawn(*io_context, [&]() -> libp2p::Coro<void> {
@@ -133,4 +132,3 @@ int main(int argc, char **argv) {
   io_context->run();
   return EXIT_SUCCESS;
 }
-

example/03-gossip/gossip_chat_example.cpp

@@ -95,7 +95,7 @@ int main(int argc, char **argv) {
   });
 
   // Subscribe to a simple topic named "example".
-  auto topic = gossip->subscribe("example");
+  auto topic = gossip->subscribe("example", false);
 
   // Receiver task: prints all messages arriving on the topic.
   libp2p::coroSpawn(*io_context, [&]() -> libp2p::Coro<void> {

include/libp2p/protocol/gossip/config.hpp

@@ -11,6 +11,7 @@
 #include <libp2p/peer/stream_protocols.hpp>
 #include <qtils/bytes_std_hash.hpp>
 #include <unordered_map>
+#include <unordered_set>
 
 namespace libp2p::protocol::gossip {
   using TopicHash = Bytes;
@@ -51,6 +52,14 @@ namespace libp2p::protocol::gossip {
     std::optional<Bytes> signature;
 
     std::optional<PeerId> received_from;
+    std::optional<MessageId> message_id;
+    std::unordered_set<PeerId> duplicate_peers;
+  };
+
+  enum class ValidationResult {
+    Accept,
+    Reject,
+    Ignore,
   };
 
   using MessageIdFn = std::function<MessageId(const Message &)>;

include/libp2p/protocol/gossip/gossip.hpp

@@ -119,6 +119,7 @@ namespace libp2p::protocol::gossip {
    */
   struct MessageCacheEntry {
     MessagePtr message;
+    bool validated;
     std::unordered_map<PeerId, size_t> iwant;
   };
 
@@ -147,16 +148,17 @@ namespace libp2p::protocol::gossip {
    public:
     /** Receive next message published to this topic (awaitable). */
     CoroOutcome<Bytes> receive();
-    CoroOutcome<Message> receiveMessage();
+    CoroOutcome<MessagePtr> receiveMessage();
     /** Publish a payload to this topic (signed locally). */
     void publish(BytesIn message);
     /** Count outbound peers currently in the mesh. */
     size_t meshOutCount() const;
 
     std::weak_ptr<Gossip> weak_gossip_;
     TopicHash topic_hash_;
+    bool validate_ = false;
     bool publish_only_ = false;
-    CoroOutcomeChannel<Message> receive_channel_;
+    CoroOutcomeChannel<MessagePtr> receive_channel_;
     History history_;
     TopicBackoff backoff_;
     std::unordered_set<PeerPtr> peers_;       // all peers subscribed
@@ -260,9 +262,10 @@ namespace libp2p::protocol::gossip {
     void start();
 
     /** Subscribe locally to a topic by hash (creates Topic if new). */
-    std::shared_ptr<Topic> subscribe(TopicHash topic_hash);
+    std::shared_ptr<Topic> subscribe(TopicHash topic_hash, bool validate);
     /** Subscribe locally to a topic by string name. */
-    std::shared_ptr<Topic> subscribe(std::string_view topic_hash);
+    std::shared_ptr<Topic> subscribe(std::string_view topic_hash,
+                                     bool validate);
 
     /**
      * Publish message to topic.
@@ -273,13 +276,14 @@ namespace libp2p::protocol::gossip {
     /** Publish a payload to a topic (signed, deduped, broadcast). */
     void publish(Topic &topic, BytesIn data);
 
+    void validate(MessagePtr message, ValidationResult result);
+
     /**
      * Broadcast a message to peers following mesh/flood rules and optional
      * IDONTWANT side channel.
      */
     void broadcast(Topic &topic,
                    std::optional<PeerId> from,
-                   const MessageId &message_id,
                    const MessagePtr &message);
 
     /** Decode and process an inbound RPC frame from a peer. */
@@ -328,6 +332,7 @@ namespace libp2p::protocol::gossip {
                       const gossipsub::pb::RPC &pb_message);
 
     std::shared_ptr<Topic> getOrCreateTopic(const TopicHash &topic_hash,
+                                            bool validate,
                                             bool publish_only);
 
     // Dependencies and state

include/libp2p/protocol/gossip/score.hpp

@@ -19,6 +19,22 @@ namespace libp2p::protocol::gossip::score {
 
   constexpr std::chrono::seconds kTimeCacheDuration{120};
 
+  /// The reason a Gossipsub message has been rejected.
+  enum class RejectReason {
+    /// The message failed the configured validation during decoding.
+    ValidationError,
+    /// The message source is us.
+    SelfOrigin,
+    /// The peer that sent the message was blacklisted.
+    BlackListedPeer,
+    /// The source (from field) of the message was blacklisted.
+    BlackListedSource,
+    /// The validation was ignored.
+    ValidationIgnored,
+    /// The validation failed.
+    ValidationFailed,
+  };
+
   struct DeliveryStatusUnknown {};
   struct DeliveryStatusValid {
     Time time;
@@ -226,6 +242,51 @@ namespace libp2p::protocol::gossip::score {
       }
     }
 
+    void rejectInvalidMessage(const PeerId &from, const TopicHash &topic_hash) {
+      markInvalidMessageDelivery(from, topic_hash);
+    }
+
+    // Reject a message.
+    void rejectMessage(const PeerId &from,
+                       const MessageId &message_id,
+                       const TopicHash &topic_hash,
+                       RejectReason reason) {
+      // these messages are not tracked, but the peer is penalized as they are
+      // invalid
+      if (reason == RejectReason::ValidationError
+          or reason == RejectReason::SelfOrigin) {
+        rejectInvalidMessage(from, topic_hash);
+        return;
+      }
+      // we ignore those messages, so do nothing.
+      if (reason == RejectReason::BlackListedPeer
+          or reason == RejectReason::BlackListedSource) {
+        return;
+      }
+      auto &record = deliveries_.getOrDefault(message_id);
+      // Multiple peers can now reject the same message as we track which peers
+      // send us the message. If we have already updated the status, return.
+      if (not std::holds_alternative<DeliveryStatusUnknown>(record.status)) {
+        return;
+      }
+      if (reason == RejectReason::ValidationIgnored) {
+        // we were explicitly instructed by the validator to ignore the message
+        // but not penalize the peer
+        record.status = DeliveryStatusIgnored{};
+        record.peers.clear();
+        return;
+      }
+      // mark the message as invalid and penalize peers that have already
+      // forwarded it.
+      record.status = DeliveryStatusInvalid{};
+      // release the delivery time tracking map to free some memory early
+      auto peers = std::exchange(record.peers, {});
+      markInvalidMessageDelivery(from, topic_hash);
+      for (auto &peer_id : peers) {
+        markInvalidMessageDelivery(peer_id, topic_hash);
+      }
+    }
+
     void connect(const PeerId &peer_id) {
       peer_stats_[peer_id].expires_at.reset();
     }