Merge pull request libp2p#762 from acul71/identify-fix-varint-go

feat: add length-prefixed support to identify protocol

Author: seetadev

examples/identify/identify.py

@@ -8,9 +8,10 @@
 from libp2p import (
     new_host,
 )
-from libp2p.identity.identify.identify import ID as IDENTIFY_PROTOCOL_ID
-from libp2p.identity.identify.pb.identify_pb2 import (
-    Identify,
+from libp2p.identity.identify.identify import (
+    ID as IDENTIFY_PROTOCOL_ID,
+    identify_handler_for,
+    parse_identify_response,
 )
 from libp2p.peer.peerinfo import (
     info_from_p2p_addr,
@@ -50,19 +51,32 @@ def print_identify_response(identify_response):
     )
 
 
-async def run(port: int, destination: str) -> None:
+async def run(port: int, destination: str, use_varint_format: bool = True) -> None:
     localhost_ip = "0.0.0.0"
 
     if not destination:
         # Create first host (listener)
         listen_addr = multiaddr.Multiaddr(f"/ip4/{localhost_ip}/tcp/{port}")
         host_a = new_host()
 
+        # Set up identify handler with specified format
+        identify_handler = identify_handler_for(
+            host_a, use_varint_format=use_varint_format
+        )
+        host_a.set_stream_handler(IDENTIFY_PROTOCOL_ID, identify_handler)
+
         async with host_a.run(listen_addrs=[listen_addr]):
+            # Get the actual address and replace 0.0.0.0 with 127.0.0.1 for client
+            # connections
+            server_addr = str(host_a.get_addrs()[0])
+            client_addr = server_addr.replace("/ip4/0.0.0.0/", "/ip4/127.0.0.1/")
+
+            format_name = "length-prefixed" if use_varint_format else "raw protobuf"
             print(
-                "First host listening. Run this from another console:\n\n"
+                f"First host listening (using {format_name} format). "
+                f"Run this from another console:\n\n"
                 f"identify-demo "
-                f"-d {host_a.get_addrs()[0]}\n"
+                f"-d {client_addr}\n"
             )
             print("Waiting for incoming identify request...")
             await trio.sleep_forever()
@@ -84,11 +98,18 @@ async def run(port: int, destination: str) -> None:
 
             try:
                 print("Starting identify protocol...")
-                response = await stream.read()
+
+                # Read the complete response (could be either format)
+                # Read a larger chunk to get all the data before stream closes
+                response = await stream.read(8192)  # Read enough data in one go
+
                 await stream.close()
-                identify_msg = Identify()
-                identify_msg.ParseFromString(response)
+
+                # Parse the response using the robust protocol-level function
+                # This handles both old and new formats automatically
+                identify_msg = parse_identify_response(response)
                 print_identify_response(identify_msg)
+
             except Exception as e:
                 print(f"Identify protocol error: {e}")
 
@@ -98,9 +119,12 @@ async def run(port: int, destination: str) -> None:
 def main() -> None:
     description = """
     This program demonstrates the libp2p identify protocol.
-    First run identify-demo -p <PORT>' to start a listener.
+    First run 'identify-demo -p <PORT> [--raw-format]' to start a listener.
     Then run 'identify-demo <ANOTHER_PORT> -d <DESTINATION>'
     where <DESTINATION> is the multiaddress shown by the listener.
+
+    Use --raw-format to send raw protobuf messages (old format) instead of
+    length-prefixed protobuf messages (new format, default).
     """
 
     example_maddr = (
@@ -115,10 +139,22 @@ def main() -> None:
         type=str,
         help=f"destination multiaddr string, e.g. {example_maddr}",
     )
+    parser.add_argument(
+        "--raw-format",
+        action="store_true",
+        help=(
+            "use raw protobuf format (old format) instead of "
+            "length-prefixed (new format)"
+        ),
+    )
     args = parser.parse_args()
 
+    # Determine format: raw format if --raw-format is specified, otherwise
+    # length-prefixed
+    use_varint_format = not args.raw_format
+
     try:
-        trio.run(run, *(args.port, args.destination))
+        trio.run(run, *(args.port, args.destination, use_varint_format))
     except KeyboardInterrupt:
         pass
 

examples/identify_push/identify_push_listener_dialer.py

@@ -57,18 +57,56 @@
 logger = logging.getLogger("libp2p.identity.identify-push-example")
 
 
-def custom_identify_push_handler_for(host):
+def custom_identify_push_handler_for(host, use_varint_format: bool = True):
     """
     Create a custom handler for the identify/push protocol that logs and prints
     the identity information received from the dialer.
+
+    Args:
+        host: The libp2p host
+        use_varint_format: If True, expect length-prefixed format; if False, expect
+            raw protobuf
+
     """
 
     async def handle_identify_push(stream: INetStream) -> None:
         peer_id = stream.muxed_conn.peer_id
 
         try:
-            # Read the identify message from the stream
-            data = await stream.read()
+            if use_varint_format:
+                # Read length-prefixed identify message from the stream
+                from libp2p.utils.varint import decode_varint_from_bytes
+
+                # First read the varint length prefix
+                length_bytes = b""
+                while True:
+                    b = await stream.read(1)
+                    if not b:
+                        break
+                    length_bytes += b
+                    if b[0] & 0x80 == 0:
+                        break
+
+                if not length_bytes:
+                    logger.warning("No length prefix received from peer %s", peer_id)
+                    return
+
+                msg_length = decode_varint_from_bytes(length_bytes)
+
+                # Read the protobuf message
+                data = await stream.read(msg_length)
+                if len(data) != msg_length:
+                    logger.warning("Incomplete message received from peer %s", peer_id)
+                    return
+            else:
+                # Read raw protobuf message from the stream
+                data = b""
+                while True:
+                    chunk = await stream.read(4096)
+                    if not chunk:
+                        break
+                    data += chunk
+
             identify_msg = Identify()
             identify_msg.ParseFromString(data)
 
@@ -129,19 +167,28 @@ async def handle_identify_push(stream: INetStream) -> None:
     return handle_identify_push
 
 
-async def run_listener(port: int) -> None:
+async def run_listener(port: int, use_varint_format: bool = True) -> None:
     """Run a host in listener mode."""
-    print(f"\n==== Starting Identify-Push Listener on port {port} ====\n")
+    format_name = "length-prefixed" if use_varint_format else "raw protobuf"
+    print(
+        f"\n==== Starting Identify-Push Listener on port {port} "
+        f"(using {format_name} format) ====\n"
+    )
 
     # Create key pair for the listener
     key_pair = create_new_key_pair()
 
     # Create the listener host
     host = new_host(key_pair=key_pair)
 
-    # Set up the identify and identify/push handlers
-    host.set_stream_handler(ID_IDENTIFY, identify_handler_for(host))
-    host.set_stream_handler(ID_IDENTIFY_PUSH, custom_identify_push_handler_for(host))
+    # Set up the identify and identify/push handlers with specified format
+    host.set_stream_handler(
+        ID_IDENTIFY, identify_handler_for(host, use_varint_format=use_varint_format)
+    )
+    host.set_stream_handler(
+        ID_IDENTIFY_PUSH,
+        identify_push_handler_for(host, use_varint_format=use_varint_format),
+    )
 
     # Start listening
     listen_addr = multiaddr.Multiaddr(f"/ip4/0.0.0.0/tcp/{port}")
@@ -165,19 +212,30 @@ async def run_listener(port: int) -> None:
         await trio.sleep_forever()
 
 
-async def run_dialer(port: int, destination: str) -> None:
+async def run_dialer(
+    port: int, destination: str, use_varint_format: bool = True
+) -> None:
     """Run a host in dialer mode that connects to a listener."""
-    print(f"\n==== Starting Identify-Push Dialer on port {port} ====\n")
+    format_name = "length-prefixed" if use_varint_format else "raw protobuf"
+    print(
+        f"\n==== Starting Identify-Push Dialer on port {port} "
+        f"(using {format_name} format) ====\n"
+    )
 
     # Create key pair for the dialer
     key_pair = create_new_key_pair()
 
     # Create the dialer host
     host = new_host(key_pair=key_pair)
 
-    # Set up the identify and identify/push handlers
-    host.set_stream_handler(ID_IDENTIFY, identify_handler_for(host))
-    host.set_stream_handler(ID_IDENTIFY_PUSH, identify_push_handler_for(host))
+    # Set up the identify and identify/push handlers with specified format
+    host.set_stream_handler(
+        ID_IDENTIFY, identify_handler_for(host, use_varint_format=use_varint_format)
+    )
+    host.set_stream_handler(
+        ID_IDENTIFY_PUSH,
+        identify_push_handler_for(host, use_varint_format=use_varint_format),
+    )
 
     # Start listening on a different port
     listen_addr = multiaddr.Multiaddr(f"/ip4/0.0.0.0/tcp/{port}")
@@ -206,7 +264,9 @@ async def run_dialer(port: int, destination: str) -> None:
 
             try:
                 # Call push_identify_to_peer which returns a boolean
-                success = await push_identify_to_peer(host, peer_info.peer_id)
+                success = await push_identify_to_peer(
+                    host, peer_info.peer_id, use_varint_format=use_varint_format
+                )
 
                 if success:
                     logger.info("Identify push completed successfully!")
@@ -240,29 +300,40 @@ def main() -> None:
     This program demonstrates the libp2p identify/push protocol.
     Without arguments, it runs as a listener on random port.
     With -d parameter, it runs as a dialer on random port.
-    """
 
-    example = (
-        "/ip4/127.0.0.1/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
-    )
+    Use --raw-format to send raw protobuf messages (old format) instead of
+    length-prefixed protobuf messages (new format, default).
+    """
 
     parser = argparse.ArgumentParser(description=description)
     parser.add_argument("-p", "--port", default=0, type=int, help="source port number")
     parser.add_argument(
         "-d",
         "--destination",
         type=str,
-        help=f"destination multiaddr string, e.g. {example}",
+        help="destination multiaddr string",
+    )
+    parser.add_argument(
+        "--raw-format",
+        action="store_true",
+        help=(
+            "use raw protobuf format (old format) instead of "
+            "length-prefixed (new format)"
+        ),
     )
     args = parser.parse_args()
 
+    # Determine format: raw format if --raw-format is specified, otherwise
+    # length-prefixed
+    use_varint_format = not args.raw_format
+
     try:
         if args.destination:
             # Run in dialer mode with random available port if not specified
-            trio.run(run_dialer, args.port, args.destination)
+            trio.run(run_dialer, args.port, args.destination, use_varint_format)
         else:
             # Run in listener mode with random available port if not specified
-            trio.run(run_listener, args.port)
+            trio.run(run_listener, args.port, use_varint_format)
     except KeyboardInterrupt:
         print("\nInterrupted by user")
         logger.info("Interrupted by user")

libp2p/host/defaults.py

@@ -26,5 +26,8 @@
 
 def get_default_protocols(host: IHost) -> "OrderedDict[TProtocol, StreamHandlerFn]":
     return OrderedDict(
-        ((IdentifyID, identify_handler_for(host)), (PingID, handle_ping))
+        (
+            (IdentifyID, identify_handler_for(host, use_varint_format=True)),
+            (PingID, handle_ping),
+        )
     )

libp2p/identity/identify/identify.py

@@ -16,7 +16,9 @@
     StreamClosed,
 )
 from libp2p.utils import (
+    decode_varint_with_size,
     get_agent_version,
+    varint,
 )
 
 from .pb.identify_pb2 import (
@@ -72,7 +74,47 @@ def _mk_identify_protobuf(
     )
 
 
-def identify_handler_for(host: IHost) -> StreamHandlerFn:
+def parse_identify_response(response: bytes) -> Identify:
+    """
+    Parse identify response that could be either:
+    - Old format: raw protobuf
+    - New format: length-prefixed protobuf
+
+    This function provides backward and forward compatibility.
+    """
+    # Try new format first: length-prefixed protobuf
+    if len(response) >= 1:
+        length, varint_size = decode_varint_with_size(response)
+        if varint_size > 0 and length > 0 and varint_size + length <= len(response):
+            protobuf_data = response[varint_size : varint_size + length]
+            try:
+                identify_response = Identify()
+                identify_response.ParseFromString(protobuf_data)
+                # Sanity check: must have agent_version (protocol_version is optional)
+                if identify_response.agent_version:
+                    logger.debug(
+                        "Parsed length-prefixed identify response (new format)"
+                    )
+                    return identify_response
+            except Exception:
+                pass  # Fall through to old format
+
+    # Fall back to old format: raw protobuf
+    try:
+        identify_response = Identify()
+        identify_response.ParseFromString(response)
+        logger.debug("Parsed raw protobuf identify response (old format)")
+        return identify_response
+    except Exception as e:
+        logger.error(f"Failed to parse identify response: {e}")
+        logger.error(f"Response length: {len(response)}")
+        logger.error(f"Response hex: {response.hex()}")
+        raise
+
+
+def identify_handler_for(
+    host: IHost, use_varint_format: bool = False
+) -> StreamHandlerFn:
     async def handle_identify(stream: INetStream) -> None:
         # get observed address from ``stream``
         peer_id = (
@@ -100,7 +142,21 @@ async def handle_identify(stream: INetStream) -> None:
         response = protobuf.SerializeToString()
 
         try:
-            await stream.write(response)
+            if use_varint_format:
+                # Send length-prefixed protobuf message (new format)
+                await stream.write(varint.encode_uvarint(len(response)))
+                await stream.write(response)
+                logger.debug(
+                    "Sent new format (length-prefixed) identify response to %s",
+                    peer_id,
+                )
+            else:
+                # Send raw protobuf message (old format for backward compatibility)
+                await stream.write(response)
+                logger.debug(
+                    "Sent old format (raw protobuf) identify response to %s",
+                    peer_id,
+                )
         except StreamClosed:
             logger.debug("Fail to respond to %s request: stream closed", ID)
         else: