Merge pull request #890 from lla-dane/feat/lip2p-record

libp2p-record module in reference with go-libp2p-record

Author: seetadev

README.md

@@ -33,8 +33,8 @@ ______________________________________________________________________
 | **Transport**                          | **Status** |                                     **Source**                                      |
 | -------------------------------------- | :--------: | :---------------------------------------------------------------------------------: |
 | **`libp2p-tcp`**                       |     ✅     | [source](https://github.com/libp2p/py-libp2p/blob/main/libp2p/transport/tcp/tcp.py) |
-| **`libp2p-quic`**                      |     ✅     | [source](https://github.com/libp2p/py-libp2p/tree/main/libp2p/transport/quic) |
-| **`libp2p-websocket`**                 |     ✅     | [source](https://github.com/libp2p/py-libp2p/tree/main/libp2p/transport/websocket) |
+| **`libp2p-quic`**                      |     ✅     |    [source](https://github.com/libp2p/py-libp2p/tree/main/libp2p/transport/quic)    |
+| **`libp2p-websocket`**                 |     ✅     | [source](https://github.com/libp2p/py-libp2p/tree/main/libp2p/transport/websocket)  |
 | **`libp2p-webrtc-browser-to-server`**  |     🌱     |                                                                                     |
 | **`libp2p-webrtc-private-to-private`** |     🌱     |                                                                                     |
 

docs/libp2p.peer.rst

@@ -4,6 +4,14 @@ libp2p.peer package
 Submodules
 ----------
 
+libp2p.peer.envelope module
+---------------------------
+
+.. automodule:: libp2p.peer.envelope
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
 libp2p.peer.id module
 ---------------------
 
@@ -12,6 +20,14 @@ libp2p.peer.id module
    :undoc-members:
    :show-inheritance:
 
+libp2p.peer.peer_record module
+------------------------------
+
+.. automodule:: libp2p.peer.peer_record
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
 libp2p.peer.peerdata module
 ---------------------------
 

docs/libp2p.records.rst

@@ -0,0 +1,37 @@
+libp2p.records package
+======================
+
+Submodules
+----------
+
+libp2p.records.pubkey module
+----------------------------
+
+.. automodule:: libp2p.records.pubkey
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+libp2p.records.record module
+----------------------------
+
+.. automodule:: libp2p.records.record
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+libp2p.records.utils module
+---------------------------
+
+.. automodule:: libp2p.records.utils
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+libp2p.records.validator module
+-------------------------------
+
+.. automodule:: libp2p.records.validator
+   :members:
+   :undoc-members:
+   :show-inheritance:

docs/libp2p.rst

@@ -17,6 +17,7 @@ Subpackages
    libp2p.peer
    libp2p.protocol_muxer
    libp2p.pubsub
+   libp2p.records
    libp2p.relay
    libp2p.security
    libp2p.stream_muxer

libp2p/kad_dht/common.py

@@ -9,6 +9,13 @@
 # Constants for the Kademlia algorithm
 ALPHA = 3  # Concurrency parameter
 PROTOCOL_ID = TProtocol("/ipfs/kad/1.0.0")
+PROTOCOL_PREFIX = TProtocol("/ipfs")
 QUERY_TIMEOUT = 10
 
 TTL = DEFAULT_TTL = 24 * 60 * 60  # 24 hours in seconds
+
+# Default parameters
+BUCKET_SIZE = 20  # k in the Kademlia paper
+MAXIMUM_BUCKETS = 256  # Maximum number of buckets (for 256-bit keys)
+PEER_REFRESH_INTERVAL = 60  # Interval to refresh peers in seconds
+STALE_PEER_THRESHOLD = 3600  # Time in seconds after which a peer is considered stale

libp2p/kad_dht/kad_dht.py

@@ -21,6 +21,7 @@
 from libp2p.abc import (
     IHost,
 )
+from libp2p.custom_types import TProtocol
 from libp2p.discovery.random_walk.rt_refresh_manager import RTRefreshManager
 from libp2p.kad_dht.utils import maybe_consume_signed_record
 from libp2p.network.stream.net_stream import (
@@ -34,13 +35,17 @@
     PeerInfo,
 )
 from libp2p.peer.peerstore import env_to_send_in_RPC
+from libp2p.records.pubkey import PublicKeyValidator
+from libp2p.records.validator import NamespacedValidator, Validator
 from libp2p.tools.async_service import (
     Service,
 )
 
 from .common import (
     ALPHA,
+    BUCKET_SIZE,
     PROTOCOL_ID,
+    PROTOCOL_PREFIX,
     QUERY_TIMEOUT,
 )
 from .pb.kademlia_pb2 import (
@@ -92,7 +97,17 @@ class KadDHT(Service):
 
     """
 
-    def __init__(self, host: IHost, mode: DHTMode, enable_random_walk: bool = False):
+    def __init__(
+        self,
+        host: IHost,
+        mode: DHTMode,
+        enable_random_walk: bool = False,
+        validator: NamespacedValidator | None = None,
+        validator_changed: bool = False,
+        protocol_prefix: TProtocol = PROTOCOL_PREFIX,
+        enable_providers: bool = True,
+        enable_values: bool = True,
+    ):
         """
         Initialize a new Kademlia DHT node.
 
@@ -115,6 +130,18 @@ def __init__(self, host: IHost, mode: DHTMode, enable_random_walk: bool = False)
         # Initialize the routing table
         self.routing_table = RoutingTable(self.local_peer_id, self.host)
 
+        self.protocol_prefix = protocol_prefix
+        self.enable_providers = enable_providers
+        self.enable_values = enable_values
+        self.validator = validator
+
+        if validator is None:
+            self.validator = NamespacedValidator({"pk": PublicKeyValidator()})
+
+        # If true implies that the validator has been changed and that
+        # Defaults should not be used
+        self.validator_changed = validator_changed
+
         # Initialize peer routing
         self.peer_routing = PeerRouting(host, self.routing_table)
 
@@ -208,6 +235,84 @@ async def stop(self) -> None:
         else:
             logger.info("RT Refresh Manager was not running (Random Walk disabled)")
 
+    def apply_fallbacks(self) -> None:
+        """
+        Apply fallback validators if not explicitely changed by the user
+
+        This sets default validators like 'pk' and 'ipns' if they are missing and
+        the default validator set hasn't been overridden.
+        """
+        if not self.validator_changed:
+            if not isinstance(self.validator, NamespacedValidator):
+                raise ValueError(
+                    "Default validator was changed without marking it True"
+                )
+
+            if "pk" not in self.validator._validators:
+                self.validator._validators["pk"] = PublicKeyValidator()
+
+            # TODO: Do the same thing for ipns, but need to implement first.
+
+    def validate_config(self) -> None:
+        """
+        Validate the DHT config.
+        """
+        if self.protocol_prefix != PROTOCOL_PREFIX:
+            return  # Skip validation for non-standart prefixes
+
+        for bucket in self.routing_table.buckets:
+            if bucket.bucket_size != BUCKET_SIZE:
+                raise ValueError(
+                    f"{PROTOCOL_PREFIX} prefix must use bucket size {BUCKET_SIZE}"
+                )
+
+        if not self.enable_providers:
+            raise ValueError(f"{PROTOCOL_PREFIX} prefix must have providers enabled")
+
+        if not self.enable_values:
+            raise ValueError(f"{PROTOCOL_PREFIX} prefix must have values enabled")
+
+        if not isinstance(self.validator, NamespacedValidator):
+            raise ValueError(
+                f"{PROTOCOL_PREFIX} prefix must use a namespace type validator"
+            )
+
+        vmap = self.validator._validators
+
+        # TODO: Need to add ipns also in the check
+        if set(vmap.keys()) != {"pk"}:
+            raise ValueError(f"{PROTOCOL_PREFIX} must have 'pk' and 'ipns' validators")
+
+        pk_validator = vmap.get("pk")
+        if not isinstance(pk_validator, PublicKeyValidator):
+            raise TypeError("'pk' namesapce must use PublicKeyValidator")
+
+        # TODO: ipns checks
+
+    def set_validator(self, val: NamespacedValidator) -> None:
+        """
+        Set a custom validator for the DHT config.
+
+        This marks the validator as explicitly changed, so the default
+        validators (pk and ipns) will not be automatically applied later.
+        """
+        self.validator = val
+        self.validator_changed = True
+        return
+
+    def set_namespace_validator(self, ns: str, val: Validator) -> None:
+        """
+        Adds a validator under a specofic namespace to the current DHT config.
+
+        Raises an error if the current validator is not a NamespacedValidator
+        """
+        if not isinstance(self.validator, NamespacedValidator):
+            raise TypeError(
+                "Can only add namespaced validators to a NamespacedValidator"
+            )
+
+        self.validator._validators["ns"] = val
+
     async def switch_mode(self, new_mode: DHTMode) -> DHTMode:
         """
         Switch the DHT mode.
@@ -511,8 +616,8 @@ async def handle_stream(self, stream: INetStream) -> None:
                         await stream.close()
                         return
 
-                    value = self.value_store.get(key)
-                    if value:
+                    value_record = self.value_store.get(key)
+                    if value_record:
                         logger.debug(f"Found value for key {key.hex()}")
 
                         # Create response using protobuf
@@ -521,9 +626,7 @@ async def handle_stream(self, stream: INetStream) -> None:
 
                         # Create record
                         response.key = key
-                        response.record.key = key
-                        response.record.value = value
-                        response.record.timeReceived = str(time.time())
+                        response.record.CopyFrom(value_record)
 
                         # Create sender_signed_peer_record
                         envelope_bytes, _ = env_to_send_in_RPC(self.host)
@@ -666,6 +769,26 @@ async def put_value(self, key: bytes, value: bytes) -> None:
         """
         logger.debug(f"Storing value for key {key.hex()}")
 
+        if key.decode("utf-8").startswith("/"):
+            if self.validator is not None:
+                # Dont allow local users to put bad values
+                self.validator.validate(key.decode("utf-8"), value)
+
+                old_value_record = self.value_store.get(key)
+                if old_value_record is not None and old_value_record.value != value:
+                    # Select which value is better
+                    try:
+                        index = self.validator.select(
+                            key.decode("utf-8"), [value, old_value_record.value]
+                        )
+                        if index != 0:
+                            raise ValueError(
+                                "Refusing to replace newer value with the older one"
+                            )
+                    except Exception as e:
+                        logger.debug(f"Validation error for key {key.hex()}: {e}")
+                        raise
+
         # 1. Store locally first
         self.value_store.put(key, value)
         try:
@@ -716,10 +839,10 @@ async def get_value(self, key: bytes) -> bytes | None:
         logger.debug(f"Getting value for key: {key.hex()}")
 
         # 1. Check local store first
-        value = self.value_store.get(key)
-        if value:
+        value_record = self.value_store.get(key)
+        if value_record:
             logger.debug("Found value locally")
-            return value
+            return value_record.value
 
         # 2. Get closest peers, excluding self
         closest_peers = [

libp2p/kad_dht/pb/kademlia.proto

@@ -39,3 +39,4 @@ message Message {
 
     optional bytes senderRecord = 11;   // Envelope(PeerRecord) encoded
 }
+`

libp2p/kad_dht/routing_table.py

@@ -25,7 +25,11 @@
 )
 
 from .common import (
+    BUCKET_SIZE,
+    MAXIMUM_BUCKETS,
+    PEER_REFRESH_INTERVAL,
     PROTOCOL_ID,
+    STALE_PEER_THRESHOLD,
 )
 from .pb.kademlia_pb2 import (
     Message,
@@ -34,12 +38,6 @@
 # logger = logging.getLogger("libp2p.kademlia.routing_table")
 logger = logging.getLogger("kademlia-example.routing_table")
 
-# Default parameters
-BUCKET_SIZE = 20  # k in the Kademlia paper
-MAXIMUM_BUCKETS = 256  # Maximum number of buckets (for 256-bit keys)
-PEER_REFRESH_INTERVAL = 60  # Interval to refresh peers in seconds
-STALE_PEER_THRESHOLD = 3600  # Time in seconds after which a peer is considered stale
-
 
 def peer_id_to_key(peer_id: ID) -> bytes:
     """