Merge branch 'main' into fix/885-Update-default-Bind-address

libp2p/custom_types.py

@@ -37,3 +37,4 @@ SyncValidatorFn = Callable[[ID, rpc_pb2.Message], bool]
 AsyncValidatorFn = Callable[[ID, rpc_pb2.Message], Awaitable[bool]]
 ValidatorFn = Union[SyncValidatorFn, AsyncValidatorFn]
 UnsubscribeFn = Callable[[], Awaitable[None]]
+MessageID = NewType("MessageID", str)

libp2p/pubsub/gossipsub.py

@@ -1,6 +1,3 @@
-from ast import (
-    literal_eval,
-)
 from collections import (
     defaultdict,
 )
@@ -22,6 +19,7 @@ from libp2p.abc import (
     IPubsubRouter,
 )
 from libp2p.custom_types import (
+    MessageID,
     TProtocol,
 )
 from libp2p.peer.id import (
@@ -56,6 +54,10 @@ from .pb import (
 from .pubsub import (
     Pubsub,
 )
+from .utils import (
+    parse_message_id_safe,
+    safe_parse_message_id,
+)
 
 PROTOCOL_ID = TProtocol("/meshsub/1.0.0")
 PROTOCOL_ID_V11 = TProtocol("/meshsub/1.1.0")
@@ -794,8 +796,8 @@ class GossipSub(IPubsubRouter, Service):
 
         # Add all unknown message ids (ids that appear in ihave_msg but not in
         # seen_seqnos) to list of messages we want to request
-        msg_ids_wanted: list[str] = [
-            msg_id
+        msg_ids_wanted: list[MessageID] = [
+            parse_message_id_safe(msg_id)
             for msg_id in ihave_msg.messageIDs
             if msg_id not in seen_seqnos_and_peers
         ]
@@ -811,9 +813,9 @@ class GossipSub(IPubsubRouter, Service):
         Forwards all request messages that are present in mcache to the
         requesting peer.
         """
-        # FIXME: Update type of message ID
-        # FIXME: Find a better way to parse the msg ids
-        msg_ids: list[Any] = [literal_eval(msg) for msg in iwant_msg.messageIDs]
+        msg_ids: list[tuple[bytes, bytes]] = [
+            safe_parse_message_id(msg) for msg in iwant_msg.messageIDs
+        ]
         msgs_to_forward: list[rpc_pb2.Message] = []
         for msg_id_iwant in msg_ids:
             # Check if the wanted message ID is present in mcache

libp2p/pubsub/utils.py

@@ -1,6 +1,10 @@
+import ast
 import logging
 
 from libp2p.abc import IHost
+from libp2p.custom_types import (
+    MessageID,
+)
 from libp2p.peer.envelope import consume_envelope
 from libp2p.peer.id import ID
 from libp2p.pubsub.pb.rpc_pb2 import RPC
@@ -48,3 +52,29 @@ def maybe_consume_signed_record(msg: RPC, host: IHost, peer_id: ID) -> bool:
             logger.error("Failed to update the Certified-Addr-Book: %s", e)
             return False
     return True
+
+
+def parse_message_id_safe(msg_id_str: str) -> MessageID:
+    """Safely handle message ID as string."""
+    return MessageID(msg_id_str)
+
+
+def safe_parse_message_id(msg_id_str: str) -> tuple[bytes, bytes]:
+    """
+    Safely parse message ID using ast.literal_eval with validation.
+    :param msg_id_str: String representation of message ID
+    :return: Tuple of (seqno, from_id) as bytes
+    :raises ValueError: If parsing fails
+    """
+    try:
+        parsed = ast.literal_eval(msg_id_str)
+        if not isinstance(parsed, tuple) or len(parsed) != 2:
+            raise ValueError("Invalid message ID format")
+
+        seqno, from_id = parsed
+        if not isinstance(seqno, bytes) or not isinstance(from_id, bytes):
+            raise ValueError("Message ID components must be bytes")
+
+        return (seqno, from_id)
+    except (ValueError, SyntaxError) as e:
+        raise ValueError(f"Invalid message ID format: {e}")