fix(yamux): add concurrency lock to YamuxStream to prevent data corruption

libp2p/host/basic_host.py

@@ -132,8 +132,20 @@ def get_peerstore(self) -> IPeerStore:
 
     def get_mux(self) -> Multiselect:
         """
-        :return: mux instance of host
+        Retrieve the muxer instance for the host.
+
+        Returns
+        -------
+        Multiselect
+            The muxer instance of the host. Never returns None.
+
+        Raises
+        ------
+        RuntimeError
+            If the multiselect instance is not initialized.
         """
+        if not hasattr(self, "multiselect") or self.multiselect is None:
+            raise RuntimeError("Multiselect instance not initialized")
         return self.multiselect
 
     def get_addrs(self) -> list[multiaddr.Multiaddr]:

libp2p/identity/identify/identify.py

@@ -57,19 +57,43 @@ def _remote_address_to_multiaddr(
 def _mk_identify_protobuf(
     host: IHost, observed_multiaddr: Multiaddr | None
 ) -> Identify:
-    public_key = host.get_public_key()
-    laddrs = host.get_addrs()
-    protocols = host.get_mux().get_protocols()
-
-    observed_addr = observed_multiaddr.to_bytes() if observed_multiaddr else b""
-    return Identify(
-        protocol_version=PROTOCOL_VERSION,
-        agent_version=AGENT_VERSION,
-        public_key=public_key.serialize(),
-        listen_addrs=map(_multiaddr_to_bytes, laddrs),
-        observed_addr=observed_addr,
-        protocols=protocols,
-    )
+    """
+    Create an Identify protobuf message.
+
+    Parameters
+    ----------
+    host : IHost
+        The host to create the identify message for.
+    observed_multiaddr : Multiaddr | None
+        The observed multiaddr of the peer, if any.
+
+    Returns
+    -------
+    Identify
+        The identify protobuf message.
+
+    Raises
+    ------
+    RuntimeError
+        If required host information is not available.
+    """
+    try:
+        public_key = host.get_public_key()
+        laddrs = host.get_addrs()
+        mux = host.get_mux()
+        protocols = tuple(str(p) for p in mux.get_protocols())  # get_protocols() now excludes None
+
+        observed_addr = observed_multiaddr.to_bytes() if observed_multiaddr else b""
+        return Identify(
+            protocol_version=PROTOCOL_VERSION,
+            agent_version=AGENT_VERSION,
+            public_key=public_key.serialize(),
+            listen_addrs=map(_multiaddr_to_bytes, laddrs),
+            observed_addr=observed_addr,
+            protocols=protocols,
+        )
+    except Exception as e:
+        raise RuntimeError(f"Failed to create identify protobuf: {str(e)}")
 
 
 def identify_handler_for(host: IHost) -> StreamHandlerFn:

libp2p/protocol_muxer/multiselect.py

@@ -101,6 +101,17 @@ async def negotiate(
         except trio.TooSlowError:
             raise MultiselectError("handshake read timeout")
 
+    def get_protocols(self) -> tuple[TProtocol, ...]:
+        """
+        Retrieve the protocols for which handlers have been registered.
+
+        Returns
+        -------
+        tuple[TProtocol, ...]
+            A tuple of registered protocol names, excluding None values.
+        """
+        return tuple(p for p in self.handlers.keys() if p is not None)
+
     async def handshake(self, communicator: IMultiselectCommunicator) -> None:
         """
         Perform handshake to agree on multiselect protocol.