Merge branch 'main' into write_msg_pubsub

Author: seetadev

examples/doc-examples/example_encryption_insecure.py

@@ -24,6 +24,12 @@ async def main():
     insecure_transport = InsecureTransport(
         # local_key_pair: The key pair used for libp2p identity
         local_key_pair=key_pair,
+        # secure_bytes_provider: Optional function to generate secure random bytes
+        # (defaults to secrets.token_bytes)
+        secure_bytes_provider=None,  # Use default implementation
+        # peerstore: Optional peerstore to store peer IDs and public keys
+        # (defaults to None)
+        peerstore=None,
     )
 
     # Create a security options dictionary mapping protocol ID to transport

libp2p/__init__.py

@@ -200,7 +200,9 @@ def new_swarm(
             key_pair, noise_privkey=noise_key_pair.private_key
         ),
         TProtocol(secio.ID): secio.Transport(key_pair),
-        TProtocol(PLAINTEXT_PROTOCOL_ID): InsecureTransport(key_pair),
+        TProtocol(PLAINTEXT_PROTOCOL_ID): InsecureTransport(
+            key_pair, peerstore=peerstore_opt
+        ),
     }
 
     # Use given muxer preference if provided, otherwise use global default

libp2p/kad_dht/common.py

@@ -0,0 +1,14 @@
+"""
+Shared constants and protocol parameters for the Kademlia DHT.
+"""
+
+from libp2p.custom_types import (
+    TProtocol,
+)
+
+# Constants for the Kademlia algorithm
+ALPHA = 3  # Concurrency parameter
+PROTOCOL_ID = TProtocol("/ipfs/kad/1.0.0")
+QUERY_TIMEOUT = 10
+
+TTL = DEFAULT_TTL = 24 * 60 * 60  # 24 hours in seconds

libp2p/kad_dht/kad_dht.py

@@ -5,7 +5,9 @@
 implementation based on the Kademlia algorithm and protocol.
 """
 
-from enum import Enum
+from enum import (
+    Enum,
+)
 import logging
 import time
 
@@ -18,9 +20,6 @@
 from libp2p.abc import (
     IHost,
 )
-from libp2p.custom_types import (
-    TProtocol,
-)
 from libp2p.network.stream.net_stream import (
     INetStream,
 )
@@ -34,6 +33,11 @@
     Service,
 )
 
+from .common import (
+    ALPHA,
+    PROTOCOL_ID,
+    QUERY_TIMEOUT,
+)
 from .pb.kademlia_pb2 import (
     Message,
 )
@@ -53,11 +57,7 @@
 logger = logging.getLogger("kademlia-example.kad_dht")
 # logger = logging.getLogger("libp2p.kademlia")
 # Default parameters
-PROTOCOL_ID = TProtocol("/ipfs/kad/1.0.0")
-ROUTING_TABLE_REFRESH_INTERVAL = 1 * 60  # 1 min in seconds for testing
-TTL = 24 * 60 * 60  # 24 hours in seconds
-ALPHA = 3
-QUERY_TIMEOUT = 10  # seconds
+ROUTING_TABLE_REFRESH_INTERVAL = 60  # 1 min in seconds for testing
 
 
 class DHTMode(Enum):

libp2p/kad_dht/peer_routing.py

@@ -15,16 +15,17 @@
     INetStream,
     IPeerRouting,
 )
-from libp2p.custom_types import (
-    TProtocol,
-)
 from libp2p.peer.id import (
     ID,
 )
 from libp2p.peer.peerinfo import (
     PeerInfo,
 )
 
+from .common import (
+    ALPHA,
+    PROTOCOL_ID,
+)
 from .pb.kademlia_pb2 import (
     Message,
 )
@@ -38,10 +39,7 @@
 # logger = logging.getLogger("libp2p.kademlia.peer_routing")
 logger = logging.getLogger("kademlia-example.peer_routing")
 
-# Constants for the Kademlia algorithm
-ALPHA = 3  # Concurrency parameter
 MAX_PEER_LOOKUP_ROUNDS = 20  # Maximum number of rounds in peer lookup
-PROTOCOL_ID = TProtocol("/ipfs/kad/1.0.0")
 
 
 class PeerRouting(IPeerRouting):
@@ -62,7 +60,6 @@ def __init__(self, host: IHost, routing_table: RoutingTable):
         """
         self.host = host
         self.routing_table = routing_table
-        self.protocol_id = PROTOCOL_ID
 
     async def find_peer(self, peer_id: ID) -> PeerInfo | None:
         """
@@ -247,7 +244,7 @@ async def _query_peer_for_closest(self, peer: ID, target_key: bytes) -> list[ID]
             # Open a stream to the peer using the Kademlia protocol
             logger.debug(f"Opening stream to {peer} for closest peers query")
             try:
-                stream = await self.host.new_stream(peer, [self.protocol_id])
+                stream = await self.host.new_stream(peer, [PROTOCOL_ID])
                 logger.debug(f"Stream opened to {peer}")
             except Exception as e:
                 logger.warning(f"Failed to open stream to {peer}: {e}")

libp2p/kad_dht/provider_store.py

@@ -29,6 +29,11 @@
     PeerInfo,
 )
 
+from .common import (
+    ALPHA,
+    PROTOCOL_ID,
+    QUERY_TIMEOUT,
+)
 from .pb.kademlia_pb2 import (
     Message,
 )
@@ -40,9 +45,6 @@
 PROVIDER_RECORD_REPUBLISH_INTERVAL = 22 * 60 * 60  # 22 hours in seconds
 PROVIDER_RECORD_EXPIRATION_INTERVAL = 48 * 60 * 60  # 48 hours in seconds
 PROVIDER_ADDRESS_TTL = 30 * 60  # 30 minutes in seconds
-PROTOCOL_ID = TProtocol("/ipfs/kad/1.0.0")
-ALPHA = 3  # Number of parallel queries/advertisements
-QUERY_TIMEOUT = 10  # Timeout for each query in seconds
 
 
 class ProviderRecord:

libp2p/kad_dht/routing_table.py

@@ -13,17 +13,19 @@
 from libp2p.abc import (
     IHost,
 )
-from libp2p.custom_types import (
-    TProtocol,
+from libp2p.kad_dht.utils import (
+    xor_distance,
 )
-from libp2p.kad_dht.utils import xor_distance
 from libp2p.peer.id import (
     ID,
 )
 from libp2p.peer.peerinfo import (
     PeerInfo,
 )
 
+from .common import (
+    PROTOCOL_ID,
+)
 from .pb.kademlia_pb2 import (
     Message,
 )
@@ -242,12 +244,9 @@ async def _ping_peer(self, peer_id: ID) -> bool:
         if not peer_info:
             raise ValueError(f"Peer {peer_id} not in bucket")
 
-        # Default protocol ID for Kademlia DHT
-        protocol_id = TProtocol("/ipfs/kad/1.0.0")
-
         try:
             # Open a stream to the peer with the DHT protocol
-            stream = await self.host.new_stream(peer_id, [protocol_id])
+            stream = await self.host.new_stream(peer_id, [PROTOCOL_ID])
 
             try:
                 # Create ping protobuf message

libp2p/kad_dht/value_store.py

@@ -19,17 +19,17 @@
     ID,
 )
 
+from .common import (
+    DEFAULT_TTL,
+    PROTOCOL_ID,
+)
 from .pb.kademlia_pb2 import (
     Message,
 )
 
 # logger = logging.getLogger("libp2p.kademlia.value_store")
 logger = logging.getLogger("kademlia-example.value_store")
 
-# Default time to live for values in seconds (24 hours)
-DEFAULT_TTL = 24 * 60 * 60
-PROTOCOL_ID = TProtocol("/ipfs/kad/1.0.0")
-
 
 class ValueStore:
     """

libp2p/security/insecure/transport.py

@@ -1,11 +1,15 @@
+from collections.abc import Callable
+
 from libp2p.abc import (
+    IPeerStore,
     IRawConnection,
     ISecureConn,
 )
 from libp2p.crypto.exceptions import (
     MissingDeserializerError,
 )
 from libp2p.crypto.keys import (
+    KeyPair,
     PrivateKey,
     PublicKey,
 )
@@ -30,11 +34,15 @@
 from libp2p.peer.id import (
     ID,
 )
+from libp2p.peer.peerstore import (
+    PeerStoreError,
+)
 from libp2p.security.base_session import (
     BaseSession,
 )
 from libp2p.security.base_transport import (
     BaseSecureTransport,
+    default_secure_bytes_provider,
 )
 from libp2p.security.exceptions import (
     HandshakeFailure,
@@ -102,6 +110,7 @@ async def run_handshake(
     conn: IRawConnection,
     is_initiator: bool,
     remote_peer_id: ID | None,
+    peerstore: IPeerStore | None = None,
 ) -> ISecureConn:
     """Raise `HandshakeFailure` when handshake failed."""
     msg = make_exchange_message(local_private_key.get_public_key())
@@ -164,7 +173,14 @@ async def run_handshake(
         conn=conn,
     )
 
-    # TODO: Store `pubkey` and `peer_id` to `PeerStore`
+    # Store `pubkey` and `peer_id` to `PeerStore`
+    if peerstore is not None:
+        try:
+            peerstore.add_pubkey(received_peer_id, received_pubkey)
+        except PeerStoreError:
+            # If peer ID and pubkey don't match, it would have already been caught above
+            # This might happen if the peer is already in the store
+            pass
 
     return secure_conn
 
@@ -175,6 +191,18 @@ class InsecureTransport(BaseSecureTransport):
     transport does not add any additional security.
     """
 
+    def __init__(
+        self,
+        local_key_pair: KeyPair,
+        secure_bytes_provider: Callable[[int], bytes] | None = None,
+        peerstore: IPeerStore | None = None,
+    ) -> None:
+        # If secure_bytes_provider is None, use the default one
+        if secure_bytes_provider is None:
+            secure_bytes_provider = default_secure_bytes_provider
+        super().__init__(local_key_pair, secure_bytes_provider)
+        self.peerstore = peerstore
+
     async def secure_inbound(self, conn: IRawConnection) -> ISecureConn:
         """
         Secure the connection, either locally or by communicating with opposing
@@ -183,8 +211,9 @@ async def secure_inbound(self, conn: IRawConnection) -> ISecureConn:
 
         :return: secure connection object (that implements secure_conn_interface)
         """
+        # For inbound connections, we don't know the remote peer ID yet
         return await run_handshake(
-            self.local_peer, self.local_private_key, conn, False, None
+            self.local_peer, self.local_private_key, conn, False, None, self.peerstore
         )
 
     async def secure_outbound(self, conn: IRawConnection, peer_id: ID) -> ISecureConn:
@@ -195,7 +224,7 @@ async def secure_outbound(self, conn: IRawConnection, peer_id: ID) -> ISecureCon
         :return: secure connection object (that implements secure_conn_interface)
         """
         return await run_handshake(
-            self.local_peer, self.local_private_key, conn, True, peer_id
+            self.local_peer, self.local_private_key, conn, True, peer_id, self.peerstore
         )
 
 

newsfragments/631.feature.rst

@@ -0,0 +1,7 @@
+Store public key and peer ID in peerstore during handshake
+
+Modified the InsecureTransport class to accept an optional peerstore parameter and updated the handshake process to store the received public key and peer ID in the peerstore when available.
+
+Added test cases to verify:
+1. The peerstore remains unchanged when handshake fails due to peer ID mismatch
+2. The handshake correctly adds a public key to a peer ID that already exists in the peerstore but doesn't have a public key yet