Merge pull request #889 from lla-dane/pubsub-record

 Signed-Peer-Record support in Pubsub/Gossipsub message transfer

Author: seetadev

libp2p/pubsub/floodsub.py

@@ -15,6 +15,7 @@
 from libp2p.peer.id import (
     ID,
 )
+from libp2p.peer.peerstore import env_to_send_in_RPC
 
 from .exceptions import (
     PubsubRouterError,
@@ -103,6 +104,11 @@ async def publish(self, msg_forwarder: ID, pubsub_msg: rpc_pb2.Message) -> None:
         )
         rpc_msg = rpc_pb2.RPC(publish=[pubsub_msg])
 
+        # Add the senderRecord of the peer in the RPC msg
+        if isinstance(self.pubsub, Pubsub):
+            envelope_bytes, _ = env_to_send_in_RPC(self.pubsub.host)
+            rpc_msg.senderRecord = envelope_bytes
+
         logger.debug("publishing message %s", pubsub_msg)
 
         if self.pubsub is None:

libp2p/pubsub/gossipsub.py

@@ -34,10 +34,12 @@
 )
 from libp2p.peer.peerstore import (
     PERMANENT_ADDR_TTL,
+    env_to_send_in_RPC,
 )
 from libp2p.pubsub import (
     floodsub,
 )
+from libp2p.pubsub.utils import maybe_consume_signed_record
 from libp2p.tools.async_service import (
     Service,
 )
@@ -226,6 +228,12 @@ async def handle_rpc(self, rpc: rpc_pb2.RPC, sender_peer_id: ID) -> None:
         :param rpc: RPC message
         :param sender_peer_id: id of the peer who sent the message
         """
+        # Process the senderRecord if sent
+        if isinstance(self.pubsub, Pubsub):
+            if not maybe_consume_signed_record(rpc, self.pubsub.host, sender_peer_id):
+                logger.error("Received an invalid-signed-record, ignoring the message")
+                return
+
         control_message = rpc.control
 
         # Relay each rpc control message to the appropriate handler
@@ -253,6 +261,11 @@ async def publish(self, msg_forwarder: ID, pubsub_msg: rpc_pb2.Message) -> None:
         )
         rpc_msg = rpc_pb2.RPC(publish=[pubsub_msg])
 
+        # Add the senderRecord of the peer in the RPC msg
+        if isinstance(self.pubsub, Pubsub):
+            envelope_bytes, _ = env_to_send_in_RPC(self.pubsub.host)
+            rpc_msg.senderRecord = envelope_bytes
+
         logger.debug("publishing message %s", pubsub_msg)
 
         for peer_id in peers_gen:
@@ -818,6 +831,13 @@ async def handle_iwant(
         # 1) Package these messages into a single packet
         packet: rpc_pb2.RPC = rpc_pb2.RPC()
 
+        # Here the an RPC message is being created and published in response
+        # to the iwant control msg, so we will send a freshly created senderRecord
+        # with the RPC msg
+        if isinstance(self.pubsub, Pubsub):
+            envelope_bytes, _ = env_to_send_in_RPC(self.pubsub.host)
+            packet.senderRecord = envelope_bytes
+
         packet.publish.extend(msgs_to_forward)
 
         if self.pubsub is None:
@@ -973,6 +993,12 @@ async def emit_control_message(
             raise NoPubsubAttached
         # Add control message to packet
         packet: rpc_pb2.RPC = rpc_pb2.RPC()
+
+        # Add the sender's peer-record in the RPC msg
+        if isinstance(self.pubsub, Pubsub):
+            envelope_bytes, _ = env_to_send_in_RPC(self.pubsub.host)
+            packet.senderRecord = envelope_bytes
+
         packet.control.CopyFrom(control_msg)
 
         # Get stream for peer from pubsub

libp2p/pubsub/pb/rpc.proto

@@ -14,6 +14,7 @@ message RPC {
 	}
 
 	optional ControlMessage control = 3;
+	optional bytes senderRecord = 4;
 }
 
 message Message {