Merge branch 'main' into fix/interop-issues

Author: seetadev

libp2p/__init__.py

@@ -33,8 +33,11 @@
 from libp2p.crypto.keys import (
     KeyPair,
 )
+from libp2p.crypto.ed25519 import (
+    create_new_key_pair as create_new_ed25519_key_pair,
+)
 from libp2p.crypto.rsa import (
-    create_new_key_pair,
+    create_new_key_pair as create_new_rsa_key_pair,
 )
 from libp2p.crypto.x25519 import create_new_key_pair as create_new_x25519_key_pair
 from libp2p.custom_types import (
@@ -154,7 +157,17 @@ def create_mplex_muxer_option() -> TMuxerOptions:
 
 
 def generate_new_rsa_identity() -> KeyPair:
-    return create_new_key_pair()
+    return create_new_rsa_key_pair()
+
+
+def generate_new_ed25519_identity() -> KeyPair:
+    """
+    Generate a new Ed25519 identity key pair.
+
+    Ed25519 is preferred for better interoperability with other libp2p implementations
+    (e.g., Rust, Go) which often disable RSA support.
+    """
+    return create_new_ed25519_key_pair()
 
 
 def generate_peer_id_from(key_pair: KeyPair) -> ID:
@@ -207,9 +220,14 @@ def new_swarm(
           due to its improved performance and features.
           Mplex (/mplex/6.7.0) is retained for backward compatibility
           but may be deprecated in the future.
+
+    Note: Ed25519 keys are used by default for better interoperability with
+          other libp2p implementations (Rust, Go) which often disable RSA support.
     """
     if key_pair is None:
-        key_pair = generate_new_rsa_identity()
+        # Use Ed25519 by default for better interoperability with Rust/Go libp2p
+        # which often compile without RSA support
+        key_pair = generate_new_ed25519_identity()
 
     id_opt = generate_peer_id_from(key_pair)
 

libp2p/stream_muxer/yamux/yamux.py

@@ -653,6 +653,17 @@ async def handle_incoming(self) -> None:
                             self.streams[stream_id] = stream
                             self.stream_buffers[stream_id] = bytearray()
                             self.stream_events[stream_id] = trio.Event()
+
+                            # Read any data that came with the SYN frame
+                            if length > 0:
+                                data = await self.secured_conn.read(length)
+                                self.stream_buffers[stream_id].extend(data)
+                                self.stream_events[stream_id].set()
+                                logger.debug(
+                                    f"Read {length} bytes with SYN "
+                                    f"for stream {stream_id}"
+                                )
+
                             ack_header = struct.pack(
                                 YAMUX_HEADER_FORMAT,
                                 0,
@@ -680,10 +691,20 @@ async def handle_incoming(self) -> None:
                 elif typ == TYPE_DATA and flags & FLAG_ACK:
                     async with self.streams_lock:
                         if stream_id in self.streams:
-                            logger.debug(
-                                f"Received ACK for stream"
-                                f"{stream_id} for peer {self.peer_id}"
-                            )
+                            # Read any data that came with the ACK
+                            if length > 0:
+                                data = await self.secured_conn.read(length)
+                                self.stream_buffers[stream_id].extend(data)
+                                self.stream_events[stream_id].set()
+                                logger.debug(
+                                    f"Received ACK with {length} bytes for stream "
+                                    f"{stream_id} for peer {self.peer_id}"
+                                )
+                            else:
+                                logger.debug(
+                                    f"Received ACK (no data) for stream {stream_id} "
+                                    f"for peer {self.peer_id}"
+                                )
                 elif typ == TYPE_GO_AWAY:
                     error_code = length
                     if error_code == GO_AWAY_NORMAL:

newsfragments/1034.bugfix.rst

@@ -0,0 +1 @@
+Fixed interoperability with rust-libp2p by switching default key generation to Ed25519 and enhancing Yamux to handle data with SYN/ACK frames.

pyproject.toml

@@ -25,12 +25,12 @@ dependencies = [
     "grpcio>=1.41.0",
     "lru-dict>=1.1.6",
     "miniupnpc>=2.3",
-    "multiaddr>=0.0.11",
+    "multiaddr==0.0.11",
     "mypy-protobuf>=3.0.0",
     "noiseprotocol>=0.3.0",
     "protobuf>=4.25.0,<7.0.0",
     "pycryptodome>=3.9.2",
-    "pymultihash>=0.8.2",
+    "pymultihash==0.8.2",
     "pynacl>=1.3.0",
     "rpcudp>=3.0.0",
     "trio-typing>=0.0.4",

tests/core/network/test_swarm.py

@@ -8,8 +8,18 @@
 )
 
 from libp2p import (
+    generate_new_ed25519_identity,
+    generate_new_rsa_identity,
     new_swarm,
 )
+from libp2p.crypto.ed25519 import (
+    Ed25519PrivateKey,
+    Ed25519PublicKey,
+)
+from libp2p.crypto.rsa import (
+    RSAPrivateKey,
+    RSAPublicKey,
+)
 from libp2p.network.exceptions import (
     SwarmException,
 )
@@ -259,6 +269,41 @@ def test_new_swarm_quic_multiaddr_supported():
     assert isinstance(swarm.transport, QUICTransport)
 
 
+def test_new_swarm_defaults_to_ed25519():
+    """Test that new_swarm() generates Ed25519 keys by default (not RSA)."""
+    # Test that new_swarm() without key_pair parameter generates a valid swarm
+    swarm = new_swarm()
+    assert isinstance(swarm, Swarm)
+
+    # The swarm's peer ID should be valid (indicating successful key generation)
+    peer_id = swarm.get_peer_id()
+    assert peer_id is not None
+
+    # Test that explicitly providing Ed25519 keys works
+    ed25519_key_pair = generate_new_ed25519_identity()
+    swarm_ed25519 = new_swarm(key_pair=ed25519_key_pair)
+    assert isinstance(swarm_ed25519, Swarm)
+    assert swarm_ed25519.get_peer_id() is not None
+
+    # Verify that Ed25519 keys are indeed being used by checking key type
+    assert isinstance(ed25519_key_pair.private_key, Ed25519PrivateKey)
+    assert isinstance(ed25519_key_pair.public_key, Ed25519PublicKey)
+
+    # Test that RSA keys can still be explicitly provided
+    rsa_key_pair = generate_new_rsa_identity()
+    swarm_rsa = new_swarm(key_pair=rsa_key_pair)
+    assert isinstance(swarm_rsa, Swarm)
+    assert swarm_rsa.get_peer_id() is not None
+
+    # Verify RSA keys are being used when explicitly provided
+    assert isinstance(rsa_key_pair.private_key, RSAPrivateKey)
+    assert isinstance(rsa_key_pair.public_key, RSAPublicKey)
+
+    # Ensure different key types produce different peer IDs
+    # (This is expected since RSA and Ed25519 generate different keys)
+    assert swarm_ed25519.get_peer_id() != swarm_rsa.get_peer_id()
+
+
 @pytest.mark.trio
 async def test_swarm_listen_multiple_addresses(security_protocol):
     """Test that swarm can listen on multiple addresses simultaneously."""

tests/core/stream_muxer/test_yamux.py

@@ -24,10 +24,12 @@
     MuxedConnUnavailable,
 )
 from libp2p.stream_muxer.yamux.yamux import (
+    FLAG_ACK,
     FLAG_FIN,
     FLAG_RST,
     FLAG_SYN,
     GO_AWAY_PROTOCOL_ERROR,
+    TYPE_DATA,
     TYPE_PING,
     TYPE_WINDOW_UPDATE,
     YAMUX_HEADER_FORMAT,
@@ -624,3 +626,150 @@ async def accept_should_unblock():
         "accept_stream() should have raised MuxedConnUnavailable"
     )
     logging.debug("test_yamux_accept_stream_unblocks_on_error complete")
+
+
+@pytest.mark.trio
+async def test_yamux_syn_with_data(yamux_pair):
+    """Test that data sent with SYN frame is properly received and buffered."""
+    logging.debug("Starting test_yamux_syn_with_data")
+    client_yamux, server_yamux = yamux_pair
+
+    # Manually construct a SYN frame with accompanying data
+    test_data = b"data with SYN frame"
+    stream_id = 1  # Client stream ID (odd number)
+
+    # Create SYN header with data length
+    syn_header = struct.pack(
+        YAMUX_HEADER_FORMAT,
+        0,  # version
+        TYPE_DATA,  # type
+        FLAG_SYN,  # flags
+        stream_id,
+        len(test_data),  # length of accompanying data
+    )
+
+    # Send SYN with data directly
+    await client_yamux.secured_conn.write(syn_header + test_data)
+    logging.debug(f"Sent SYN with {len(test_data)} bytes of data")
+
+    # Server should accept the stream and have data already buffered
+    server_stream = await server_yamux.accept_stream()
+    assert server_stream.stream_id == stream_id
+
+    # Verify the data was buffered and is immediately available
+    received = await server_stream.read(len(test_data))
+    assert received == test_data, "Data sent with SYN should be immediately available"
+    logging.debug("test_yamux_syn_with_data complete")
+
+
+@pytest.mark.trio
+async def test_yamux_ack_with_data(yamux_pair):
+    """Test that data sent with ACK frame is properly received and buffered."""
+    logging.debug("Starting test_yamux_ack_with_data")
+    client_yamux, server_yamux = yamux_pair
+
+    # Client opens a stream (sends SYN)
+    client_stream = await client_yamux.open_stream()
+    stream_id = client_stream.stream_id
+
+    # Wait for server to receive SYN and respond with ACK
+    await trio.sleep(0.1)
+
+    # Now manually send data with an ACK flag from server to client
+    test_data = b"data with ACK frame"
+    ack_header = struct.pack(
+        YAMUX_HEADER_FORMAT,
+        0,  # version
+        TYPE_DATA,  # type
+        FLAG_ACK,  # flags (ACK flag set)
+        stream_id,
+        len(test_data),  # length of accompanying data
+    )
+
+    # Send ACK with data from server to client
+    await server_yamux.secured_conn.write(ack_header + test_data)
+    logging.debug(f"Sent ACK with {len(test_data)} bytes of data")
+
+    # Wait for the data to be processed
+    await trio.sleep(0.1)
+
+    # Verify the data was buffered on the client side
+    # Since the stream is already open, the data should be in the buffer
+    async with client_yamux.streams_lock:
+        assert stream_id in client_yamux.stream_buffers
+        assert len(client_yamux.stream_buffers[stream_id]) >= len(test_data)
+        buffered_data = bytes(client_yamux.stream_buffers[stream_id][: len(test_data)])
+        # Remove the data we just checked
+        client_yamux.stream_buffers[stream_id] = client_yamux.stream_buffers[stream_id][
+            len(test_data) :
+        ]
+
+    assert buffered_data == test_data, "Data sent with ACK should be buffered"
+    logging.debug("test_yamux_ack_with_data complete")
+
+
+@pytest.mark.trio
+async def test_yamux_syn_with_empty_data(yamux_pair):
+    """Test that SYN frame with zero-length data is handled correctly."""
+    logging.debug("Starting test_yamux_syn_with_empty_data")
+    client_yamux, server_yamux = yamux_pair
+
+    # Manually construct a SYN frame with no data (length = 0)
+    stream_id = 3  # Client stream ID (odd number)
+
+    syn_header = struct.pack(
+        YAMUX_HEADER_FORMAT,
+        0,  # version
+        TYPE_DATA,  # type
+        FLAG_SYN,  # flags
+        stream_id,
+        0,  # length = 0, no accompanying data
+    )
+
+    # Send SYN with no data
+    await client_yamux.secured_conn.write(syn_header)
+    logging.debug("Sent SYN with no data")
+
+    # Server should accept the stream
+    server_stream = await server_yamux.accept_stream()
+    assert server_stream.stream_id == stream_id
+
+    # Verify no data is in the buffer
+    async with server_yamux.streams_lock:
+        assert len(server_yamux.stream_buffers[stream_id]) == 0
+
+    logging.debug("test_yamux_syn_with_empty_data complete")
+
+
+@pytest.mark.trio
+async def test_yamux_syn_with_large_data(yamux_pair):
+    """Test that large data sent with SYN frame is properly handled."""
+    logging.debug("Starting test_yamux_syn_with_large_data")
+    client_yamux, server_yamux = yamux_pair
+
+    # Create large test data (but within window size)
+    test_data = b"X" * 1024  # 1KB of data
+    stream_id = 5  # Client stream ID (odd number)
+
+    syn_header = struct.pack(
+        YAMUX_HEADER_FORMAT,
+        0,  # version
+        TYPE_DATA,  # type
+        FLAG_SYN,  # flags
+        stream_id,
+        len(test_data),
+    )
+
+    # Send SYN with large data
+    await client_yamux.secured_conn.write(syn_header + test_data)
+    logging.debug(f"Sent SYN with {len(test_data)} bytes of large data")
+
+    # Server should accept the stream and have all data buffered
+    server_stream = await server_yamux.accept_stream()
+    assert server_stream.stream_id == stream_id
+
+    # Verify all data was buffered correctly
+    received = await server_stream.read(len(test_data))
+    assert received == test_data
+    assert len(received) == 1024
+    logging.debug("test_yamux_syn_with_large_data complete")