Merge branch 'main' into noise-arch-change

fix: message id type inconsistency in handle ihave and message id parsing improvement in handle iwant

examples/doc-examples/example_encryption_noise.py

@@ -24,13 +24,8 @@ async def main():
     noise_transport = NoiseTransport(
         # local_key_pair: The key pair used for libp2p identity and authentication
         libp2p_keypair=key_pair,
-        # noise_privkey: The private key used for Noise protocol encryption
         noise_privkey=key_pair.private_key,
-        # early_data: Optional data to send during the handshake
+        # TODO: add early data
-        # (None means no early data)
-        early_data=None,
-        # with_noise_pipes: Whether to use Noise pipes for additional security features
-        with_noise_pipes=False,
     )
 
     # Create a security options dictionary mapping protocol ID to transport

examples/doc-examples/example_multiplexer.py

@@ -28,9 +28,7 @@ async def main():
         noise_privkey=key_pair.private_key,
         # early_data: Optional data to send during the handshake
         # (None means no early data)
-        early_data=None,
+        # TODO: add early data
-        # with_noise_pipes: Whether to use Noise pipes for additional security features
-        with_noise_pipes=False,
     )
 
     # Create a security options dictionary mapping protocol ID to transport

examples/doc-examples/example_peer_discovery.py

@@ -31,9 +31,7 @@ async def main():
         noise_privkey=key_pair.private_key,
         # early_data: Optional data to send during the handshake
         # (None means no early data)
-        early_data=None,
+        # TODO: add early data
-        # with_noise_pipes: Whether to use Noise pipes for additional security features
-        with_noise_pipes=False,
     )
 
     # Create a security options dictionary mapping protocol ID to transport

examples/doc-examples/example_running.py

@@ -28,9 +28,7 @@ async def main():
         noise_privkey=key_pair.private_key,
         # early_data: Optional data to send during the handshake
         # (None means no early data)
-        early_data=None,
+        # TODO: add early data
-        # with_noise_pipes: Whether to use Noise pipes for additional security features
-        with_noise_pipes=False,
     )
 
     # Create a security options dictionary mapping protocol ID to transport

examples/kademlia/kademlia.py

@@ -41,6 +41,7 @@ from libp2p.tools.async_service import (
 from libp2p.tools.utils import (
     info_from_p2p_addr,
 )
+from libp2p.utils.paths import get_script_dir, join_paths
 
 # Configure logging
 logging.basicConfig(
@@ -53,8 +54,8 @@ logger = logging.getLogger("kademlia-example")
 # Configure DHT module loggers to inherit from the parent logger
 # This ensures all kademlia-example.* loggers use the same configuration
 # Get the directory where this script is located
-SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+SCRIPT_DIR = get_script_dir(__file__)
-SERVER_ADDR_LOG = os.path.join(SCRIPT_DIR, "server_node_addr.txt")
+SERVER_ADDR_LOG = join_paths(SCRIPT_DIR, "server_node_addr.txt")
 
 # Set the level for all child loggers
 for module in [

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/discovery/bootstrap/bootstrap.py

@@ -2,15 +2,20 @@ import logging
 
 from multiaddr import Multiaddr
 from multiaddr.resolvers import DNSResolver
+import trio
 
 from libp2p.abc import ID, INetworkService, PeerInfo
 from libp2p.discovery.bootstrap.utils import validate_bootstrap_addresses
 from libp2p.discovery.events.peerDiscovery import peerDiscovery
+from libp2p.network.exceptions import SwarmException
 from libp2p.peer.peerinfo import info_from_p2p_addr
+from libp2p.peer.peerstore import PERMANENT_ADDR_TTL
 
 logger = logging.getLogger("libp2p.discovery.bootstrap")
 resolver = DNSResolver()
 
+DEFAULT_CONNECTION_TIMEOUT = 10
+
 
 class BootstrapDiscovery:
     """
@@ -19,68 +24,147 @@ class BootstrapDiscovery:
     """
 
     def __init__(self, swarm: INetworkService, bootstrap_addrs: list[str]):
+        """
+        Initialize BootstrapDiscovery.
+
+        Args:
+            swarm: The network service (swarm) instance
+            bootstrap_addrs: List of bootstrap peer multiaddresses
+
+        """
         self.swarm = swarm
         self.peerstore = swarm.peerstore
         self.bootstrap_addrs = bootstrap_addrs or []
         self.discovered_peers: set[str] = set()
+        self.connection_timeout: int = DEFAULT_CONNECTION_TIMEOUT
 
     async def start(self) -> None:
-        """Process bootstrap addresses and emit peer discovery events."""
+        """Process bootstrap addresses and emit peer discovery events in parallel."""
-        logger.debug(
+        logger.info(
             f"Starting bootstrap discovery with "
             f"{len(self.bootstrap_addrs)} bootstrap addresses"
         )
 
+        # Show all bootstrap addresses being processed
+        for i, addr in enumerate(self.bootstrap_addrs):
+            logger.debug(f"{i + 1}. {addr}")
+
         # Validate and filter bootstrap addresses
         self.bootstrap_addrs = validate_bootstrap_addresses(self.bootstrap_addrs)
+        logger.info(f"Valid addresses after validation: {len(self.bootstrap_addrs)}")
 
-        for addr_str in self.bootstrap_addrs:
+        # Use Trio nursery for PARALLEL address processing
-            try:
+        try:
-                await self._process_bootstrap_addr(addr_str)
+            async with trio.open_nursery() as nursery:
-            except Exception as e:
+                logger.debug(
-                logger.debug(f"Failed to process bootstrap address {addr_str}: {e}")
+                    f"Starting {len(self.bootstrap_addrs)} parallel address "
+                    f"processing tasks"
+                )
+
+                # Start all bootstrap address processing tasks in parallel
+                for addr_str in self.bootstrap_addrs:
+                    logger.debug(f"Starting parallel task for: {addr_str}")
+                    nursery.start_soon(self._process_bootstrap_addr, addr_str)
+
+                # The nursery will wait for all address processing tasks to complete
+                logger.debug(
+                    "Nursery active - waiting for address processing tasks to complete"
+                )
+
+        except trio.Cancelled:
+            logger.debug("Bootstrap address processing cancelled - cleaning up tasks")
+            raise
+        except Exception as e:
+            logger.error(f"Bootstrap address processing failed: {e}")
+            raise
+
+        logger.info("Bootstrap discovery startup complete - all tasks finished")
 
     def stop(self) -> None:
         """Clean up bootstrap discovery resources."""
-        logger.debug("Stopping bootstrap discovery")
+        logger.info("Stopping bootstrap discovery and cleaning up tasks")
+
+        # Clear discovered peers
         self.discovered_peers.clear()
 
+        logger.debug("Bootstrap discovery cleanup completed")
+
     async def _process_bootstrap_addr(self, addr_str: str) -> None:
         """Convert string address to PeerInfo and add to peerstore."""
         try:
-            multiaddr = Multiaddr(addr_str)
+            try:
+                multiaddr = Multiaddr(addr_str)
+            except Exception as e:
+                logger.debug(f"Invalid multiaddr format '{addr_str}': {e}")
+                return
+
+            if self.is_dns_addr(multiaddr):
+                resolved_addrs = await resolver.resolve(multiaddr)
+                if resolved_addrs is None:
+                    logger.warning(f"DNS resolution returned None for: {addr_str}")
+                    return
+
+                peer_id_str = multiaddr.get_peer_id()
+                if peer_id_str is None:
+                    logger.warning(f"Missing peer ID in DNS address: {addr_str}")
+                    return
+                peer_id = ID.from_base58(peer_id_str)
+                addrs = [addr for addr in resolved_addrs]
+                if not addrs:
+                    logger.warning(f"No addresses resolved for DNS address: {addr_str}")
+                    return
+                peer_info = PeerInfo(peer_id, addrs)
+                await self.add_addr(peer_info)
+            else:
+                peer_info = info_from_p2p_addr(multiaddr)
+                await self.add_addr(peer_info)
         except Exception as e:
-            logger.debug(f"Invalid multiaddr format '{addr_str}': {e}")
+            logger.warning(f"Failed to process bootstrap address {addr_str}: {e}")
-            return
-        if self.is_dns_addr(multiaddr):
-            resolved_addrs = await resolver.resolve(multiaddr)
-            peer_id_str = multiaddr.get_peer_id()
-            if peer_id_str is None:
-                logger.warning(f"Missing peer ID in DNS address: {addr_str}")
-                return
-            peer_id = ID.from_base58(peer_id_str)
-            addrs = [addr for addr in resolved_addrs]
-            if not addrs:
-                logger.warning(f"No addresses resolved for DNS address: {addr_str}")
-                return
-            peer_info = PeerInfo(peer_id, addrs)
-            self.add_addr(peer_info)
-        else:
-            self.add_addr(info_from_p2p_addr(multiaddr))
 
     def is_dns_addr(self, addr: Multiaddr) -> bool:
         """Check if the address is a DNS address."""
         return any(protocol.name == "dnsaddr" for protocol in addr.protocols())
 
-    def add_addr(self, peer_info: PeerInfo) -> None:
+    async def add_addr(self, peer_info: PeerInfo) -> None:
-        """Add a peer to the peerstore and emit discovery event."""
+        """
+        Add a peer to the peerstore, emit discovery event,
+        and attempt connection in parallel.
+        """
+        logger.debug(
+            f"Adding peer {peer_info.peer_id} with {len(peer_info.addrs)} addresses"
+        )
+
         # Skip if it's our own peer
         if peer_info.peer_id == self.swarm.get_peer_id():
             logger.debug(f"Skipping own peer ID: {peer_info.peer_id}")
             return
 
-        # Always add addresses to peerstore (allows multiple addresses for same peer)
+        # Filter addresses to only include IPv4+TCP (only supported protocol)
-        self.peerstore.add_addrs(peer_info.peer_id, peer_info.addrs, 10)
+        ipv4_tcp_addrs = []
+        filtered_out_addrs = []
+
+        for addr in peer_info.addrs:
+            if self._is_ipv4_tcp_addr(addr):
+                ipv4_tcp_addrs.append(addr)
+            else:
+                filtered_out_addrs.append(addr)
+
+        # Log filtering results
+        logger.debug(
+            f"Address filtering for {peer_info.peer_id}: "
+            f"{len(ipv4_tcp_addrs)} IPv4+TCP, {len(filtered_out_addrs)} filtered"
+        )
+
+        # Skip peer if no IPv4+TCP addresses available
+        if not ipv4_tcp_addrs:
+            logger.warning(
+                f"❌ No IPv4+TCP addresses for {peer_info.peer_id} - "
+                f"skipping connection attempts"
+            )
+            return
+
+        # Add only IPv4+TCP addresses to peerstore
+        self.peerstore.add_addrs(peer_info.peer_id, ipv4_tcp_addrs, PERMANENT_ADDR_TTL)
 
         # Only emit discovery event if this is the first time we see this peer
         peer_id_str = str(peer_info.peer_id)
@@ -89,6 +173,140 @@ class BootstrapDiscovery:
             self.discovered_peers.add(peer_id_str)
             # Emit peer discovery event
             peerDiscovery.emit_peer_discovered(peer_info)
-            logger.debug(f"Peer discovered: {peer_info.peer_id}")
+            logger.info(f"Peer discovered: {peer_info.peer_id}")
+
+            # Connect to peer (parallel across different bootstrap addresses)
+            logger.debug("Connecting to discovered peer...")
+            await self._connect_to_peer(peer_info.peer_id)
+
         else:
-            logger.debug(f"Additional addresses added for peer: {peer_info.peer_id}")
+            logger.debug(
+                f"Additional addresses added for existing peer: {peer_info.peer_id}"
+            )
+            # Even for existing peers, try to connect if not already connected
+            if peer_info.peer_id not in self.swarm.connections:
+                logger.debug("Connecting to existing peer...")
+                await self._connect_to_peer(peer_info.peer_id)
+
+    async def _connect_to_peer(self, peer_id: ID) -> None:
+        """
+        Attempt to establish a connection to a peer with timeout.
+
+        Uses swarm.dial_peer to connect using addresses stored in peerstore.
+        Times out after self.connection_timeout seconds to prevent hanging.
+        """
+        logger.debug(f"Connection attempt for peer: {peer_id}")
+
+        # Pre-connection validation: Check if already connected
+        if peer_id in self.swarm.connections:
+            logger.debug(
+                f"Already connected to {peer_id} - skipping connection attempt"
+            )
+            return
+
+        # Check available addresses before attempting connection
+        available_addrs = self.peerstore.addrs(peer_id)
+        logger.debug(f"Connecting to {peer_id} ({len(available_addrs)} addresses)")
+
+        if not available_addrs:
+            logger.error(f"❌ No addresses available for {peer_id} - cannot connect")
+            return
+
+        # Record start time for connection attempt monitoring
+        connection_start_time = trio.current_time()
+
+        try:
+            with trio.move_on_after(self.connection_timeout):
+                # Log connection attempt
+                logger.debug(
+                    f"Attempting connection to {peer_id} using "
+                    f"{len(available_addrs)} addresses"
+                )
+
+                # Use swarm.dial_peer to connect using stored addresses
+                await self.swarm.dial_peer(peer_id)
+
+                # Calculate connection time
+                connection_time = trio.current_time() - connection_start_time
+
+                # Post-connection validation: Verify connection was actually established
+                if peer_id in self.swarm.connections:
+                    logger.info(
+                        f"✅ Connected to {peer_id} (took {connection_time:.2f}s)"
+                    )
+
+                else:
+                    logger.warning(
+                        f"Dial succeeded but connection not found for {peer_id}"
+                    )
+        except trio.TooSlowError:
+            logger.warning(
+                f"❌ Connection to {peer_id} timed out after {self.connection_timeout}s"
+            )
+        except SwarmException as e:
+            # Calculate failed connection time
+            failed_connection_time = trio.current_time() - connection_start_time
+
+            # Enhanced error logging
+            error_msg = str(e)
+            if "no addresses established a successful connection" in error_msg:
+                logger.warning(
+                    f"❌ Failed to connect to {peer_id} after trying all "
+                    f"{len(available_addrs)} addresses "
+                    f"(took {failed_connection_time:.2f}s)"
+                )
+                # Log individual address failures if this is a MultiError
+                if (
+                    e.__cause__ is not None
+                    and hasattr(e.__cause__, "exceptions")
+                    and getattr(e.__cause__, "exceptions", None) is not None
+                ):
+                    exceptions_list = getattr(e.__cause__, "exceptions")
+                    logger.debug("📋 Individual address failure details:")
+                    for i, addr_exception in enumerate(exceptions_list, 1):
+                        logger.debug(f"Address {i}: {addr_exception}")
+                        # Also log the actual address that failed
+                        if i <= len(available_addrs):
+                            logger.debug(f"Failed address: {available_addrs[i - 1]}")
+                else:
+                    logger.warning("No detailed exception information available")
+            else:
+                logger.warning(
+                    f"❌ Failed to connect to {peer_id}: {e} "
+                    f"(took {failed_connection_time:.2f}s)"
+                )
+
+        except Exception as e:
+            # Handle unexpected errors that aren't swarm-specific
+            failed_connection_time = trio.current_time() - connection_start_time
+            logger.error(
+                f"❌ Unexpected error connecting to {peer_id}: "
+                f"{e} (took {failed_connection_time:.2f}s)"
+            )
+            # Don't re-raise to prevent killing the nursery and other parallel tasks
+
+    def _is_ipv4_tcp_addr(self, addr: Multiaddr) -> bool:
+        """
+        Check if address is IPv4 with TCP protocol only.
+
+        Filters out IPv6, UDP, QUIC, WebSocket, and other unsupported protocols.
+        Only IPv4+TCP addresses are supported by the current transport.
+        """
+        try:
+            protocols = addr.protocols()
+
+            # Must have IPv4 protocol
+            has_ipv4 = any(p.name == "ip4" for p in protocols)
+            if not has_ipv4:
+                return False
+
+            # Must have TCP protocol
+            has_tcp = any(p.name == "tcp" for p in protocols)
+            if not has_tcp:
+                return False
+
+            return True
+
+        except Exception:
+            # If we can't parse the address, don't use it
+            return False

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_ids_wanted: list[MessageID] = [
-            msg_id
+            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
+        msg_ids: list[tuple[bytes, bytes]] = [
-        # FIXME: Find a better way to parse the msg ids
+            safe_parse_message_id(msg) for msg in iwant_msg.messageIDs
-        msg_ids: list[Any] = [literal_eval(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}")

libp2p/security/noise/early_data.py

@@ -0,0 +1,68 @@
+from abc import ABC, abstractmethod
+
+from libp2p.abc import IRawConnection
+from libp2p.custom_types import TProtocol
+from libp2p.peer.id import ID
+
+from .pb import noise_pb2 as noise_pb
+
+
+class EarlyDataHandler(ABC):
+    """Interface for handling early data during Noise handshake"""
+
+    @abstractmethod
+    async def send(
+        self, conn: IRawConnection, peer_id: ID
+    ) -> noise_pb.NoiseExtensions | None:
+        """Called to generate early data to send during handshake"""
+        pass
+
+    @abstractmethod
+    async def received(
+        self, conn: IRawConnection, extensions: noise_pb.NoiseExtensions | None
+    ) -> None:
+        """Called when early data is received during handshake"""
+        pass
+
+
+class TransportEarlyDataHandler(EarlyDataHandler):
+    """Default early data handler for muxer negotiation"""
+
+    def __init__(self, supported_muxers: list[TProtocol]):
+        self.supported_muxers = supported_muxers
+        self.received_muxers: list[TProtocol] = []
+
+    async def send(
+        self, conn: IRawConnection, peer_id: ID
+    ) -> noise_pb.NoiseExtensions | None:
+        """Send our supported muxers list"""
+        if not self.supported_muxers:
+            return None
+
+        extensions = noise_pb.NoiseExtensions()
+        # Convert TProtocol to string for serialization
+        extensions.stream_muxers[:] = [str(muxer) for muxer in self.supported_muxers]
+        return extensions
+
+    async def received(
+        self, conn: IRawConnection, extensions: noise_pb.NoiseExtensions | None
+    ) -> None:
+        """Store received muxers list"""
+        if extensions and extensions.stream_muxers:
+            self.received_muxers = [
+                TProtocol(muxer) for muxer in extensions.stream_muxers
+            ]
+
+    def match_muxers(self, is_initiator: bool) -> TProtocol | None:
+        """Find first common muxer between local and remote"""
+        if is_initiator:
+            # Initiator: find first local muxer that remote supports
+            for local_muxer in self.supported_muxers:
+                if local_muxer in self.received_muxers:
+                    return local_muxer
+        else:
+            # Responder: find first remote muxer that we support
+            for remote_muxer in self.received_muxers:
+                if remote_muxer in self.supported_muxers:
+                    return remote_muxer
+        return None

libp2p/security/noise/patterns.py

@@ -30,6 +30,9 @@ from libp2p.security.secure_session import (
     SecureSession,
 )
 
+from .early_data import (
+    EarlyDataHandler,
+)
 from .exceptions import (
     HandshakeHasNotFinished,
     InvalidSignature,
@@ -45,6 +48,7 @@ from .messages import (
     make_handshake_payload_sig,
     verify_handshake_payload_sig,
 )
+from .pb import noise_pb2 as noise_pb
 
 
 class IPattern(ABC):
@@ -62,7 +66,8 @@ class BasePattern(IPattern):
     noise_static_key: PrivateKey
     local_peer: ID
     libp2p_privkey: PrivateKey
-    early_data: bytes | None
+    initiator_early_data_handler: EarlyDataHandler | None
+    responder_early_data_handler: EarlyDataHandler | None
 
     def create_noise_state(self) -> NoiseState:
         noise_state = NoiseState.from_name(self.protocol_name)
@@ -73,11 +78,50 @@ class BasePattern(IPattern):
             raise NoiseStateError("noise_protocol is not initialized")
         return noise_state
 
-    def make_handshake_payload(self) -> NoiseHandshakePayload:
+    async def make_handshake_payload(
+        self, conn: IRawConnection, peer_id: ID, is_initiator: bool
+    ) -> NoiseHandshakePayload:
         signature = make_handshake_payload_sig(
             self.libp2p_privkey, self.noise_static_key.get_public_key()
         )
-        return NoiseHandshakePayload(self.libp2p_privkey.get_public_key(), signature)
+
+        # NEW: Get early data from appropriate handler
+        extensions = None
+        if is_initiator and self.initiator_early_data_handler:
+            extensions = await self.initiator_early_data_handler.send(conn, peer_id)
+        elif not is_initiator and self.responder_early_data_handler:
+            extensions = await self.responder_early_data_handler.send(conn, peer_id)
+
+        # NEW: Serialize extensions into early_data field
+        early_data = None
+        if extensions:
+            early_data = extensions.SerializeToString()
+
+        return NoiseHandshakePayload(
+            self.libp2p_privkey.get_public_key(),
+            signature,
+            early_data,  # ← This is the key addition
+        )
+
+    async def handle_received_payload(
+        self, conn: IRawConnection, payload: NoiseHandshakePayload, is_initiator: bool
+    ) -> None:
+        """Process early data from received payload"""
+        if not payload.early_data:
+            return
+
+        # Deserialize the NoiseExtensions from early_data field
+        try:
+            extensions = noise_pb.NoiseExtensions.FromString(payload.early_data)
+        except Exception:
+            # Invalid extensions, ignore silently
+            return
+
+        # Pass to appropriate handler
+        if is_initiator and self.initiator_early_data_handler:
+            await self.initiator_early_data_handler.received(conn, extensions)
+        elif not is_initiator and self.responder_early_data_handler:
+            await self.responder_early_data_handler.received(conn, extensions)
 
 
 class PatternXX(BasePattern):
@@ -86,13 +130,15 @@ class PatternXX(BasePattern):
         local_peer: ID,
         libp2p_privkey: PrivateKey,
         noise_static_key: PrivateKey,
-        early_data: bytes | None = None,
+        initiator_early_data_handler: EarlyDataHandler | None,
+        responder_early_data_handler: EarlyDataHandler | None,
     ) -> None:
         self.protocol_name = b"Noise_XX_25519_ChaChaPoly_SHA256"
         self.local_peer = local_peer
         self.libp2p_privkey = libp2p_privkey
         self.noise_static_key = noise_static_key
-        self.early_data = early_data
+        self.initiator_early_data_handler = initiator_early_data_handler
+        self.responder_early_data_handler = responder_early_data_handler
 
     async def handshake_inbound(self, conn: IRawConnection) -> ISecureConn:
         noise_state = self.create_noise_state()
@@ -106,18 +152,23 @@ class PatternXX(BasePattern):
 
         read_writer = NoiseHandshakeReadWriter(conn, noise_state)
 
-        # Consume msg#1.
+        # 1. Consume msg#1 (just empty bytes)
         await read_writer.read_msg()
 
-        # Send msg#2, which should include our handshake payload.
+        # 2. Send msg#2 with our payload INCLUDING EARLY DATA
-        our_payload = self.make_handshake_payload()
+        our_payload = await self.make_handshake_payload(
+            conn,
+            self.local_peer,  # We send our own peer ID in responder role
+            is_initiator=False,
+        )
         msg_2 = our_payload.serialize()
         await read_writer.write_msg(msg_2)
 
-        # Receive and consume msg#3.
+        # 3. Receive msg#3
         msg_3 = await read_writer.read_msg()
         peer_handshake_payload = NoiseHandshakePayload.deserialize(msg_3)
 
+        # Extract remote pubkey from noise handshake state
         if handshake_state.rs is None:
             raise NoiseStateError(
                 "something is wrong in the underlying noise `handshake_state`: "
@@ -126,14 +177,31 @@ class PatternXX(BasePattern):
             )
         remote_pubkey = self._get_pubkey_from_noise_keypair(handshake_state.rs)
 
+        # 4. Verify signature (unchanged)
         if not verify_handshake_payload_sig(peer_handshake_payload, remote_pubkey):
             raise InvalidSignature
+
+        # NEW: Process early data from msg#3 AFTER signature verification
+        await self.handle_received_payload(
+            conn, peer_handshake_payload, is_initiator=False
+        )
+
         remote_peer_id_from_pubkey = ID.from_pubkey(peer_handshake_payload.id_pubkey)
 
         if not noise_state.handshake_finished:
             raise HandshakeHasNotFinished(
                 "handshake is done but it is not marked as finished in `noise_state`"
             )
+
+        # NEW: Get negotiated muxer for connection state
+        # negotiated_muxer = None
+        if self.responder_early_data_handler and hasattr(
+            self.responder_early_data_handler, "match_muxers"
+        ):
+            # negotiated_muxer =
+            # self.responder_early_data_handler.match_muxers(is_initiator=False)
+            pass
+
         transport_read_writer = NoiseTransportReadWriter(conn, noise_state)
         return SecureSession(
             local_peer=self.local_peer,
@@ -142,6 +210,8 @@ class PatternXX(BasePattern):
             remote_permanent_pubkey=remote_pubkey,
             is_initiator=False,
             conn=transport_read_writer,
+            # NOTE: negotiated_muxer would need to be added to SecureSession constructor
+            # For now, store it in connection metadata or similar
         )
 
     async def handshake_outbound(
@@ -158,24 +228,27 @@ class PatternXX(BasePattern):
         if handshake_state is None:
             raise NoiseStateError("Handshake state is not initialized")
 
-        # Send msg#1, which is *not* encrypted.
+        # 1. Send msg#1 (empty) - no early data possible in XX pattern
         msg_1 = b""
         await read_writer.write_msg(msg_1)
 
-        # Read msg#2 from the remote, which contains the public key of the peer.
+        # 2. Read msg#2 from responder
         msg_2 = await read_writer.read_msg()
         peer_handshake_payload = NoiseHandshakePayload.deserialize(msg_2)
 
+        # Extract remote pubkey from noise handshake state
         if handshake_state.rs is None:
             raise NoiseStateError(
                 "something is wrong in the underlying noise `handshake_state`: "
-                "we received and consumed msg#3, which should have included the "
+                "we received and consumed msg#2, which should have included the "
                 "remote static public key, but it is not present in the handshake_state"
             )
         remote_pubkey = self._get_pubkey_from_noise_keypair(handshake_state.rs)
 
+        # Verify signature BEFORE processing early data (security)
         if not verify_handshake_payload_sig(peer_handshake_payload, remote_pubkey):
             raise InvalidSignature
+
         remote_peer_id_from_pubkey = ID.from_pubkey(peer_handshake_payload.id_pubkey)
         if remote_peer_id_from_pubkey != remote_peer:
             raise PeerIDMismatchesPubkey(
@@ -184,8 +257,15 @@ class PatternXX(BasePattern):
                 f"remote_peer_id_from_pubkey={remote_peer_id_from_pubkey}"
             )
 
-        # Send msg#3, which includes our encrypted payload and our noise static key.
+        # NEW: Process early data from msg#2 AFTER verification
-        our_payload = self.make_handshake_payload()
+        await self.handle_received_payload(
+            conn, peer_handshake_payload, is_initiator=True
+        )
+
+        # 3. Send msg#3 with our payload INCLUDING EARLY DATA
+        our_payload = await self.make_handshake_payload(
+            conn, remote_peer, is_initiator=True
+        )
         msg_3 = our_payload.serialize()
         await read_writer.write_msg(msg_3)
 
@@ -193,6 +273,16 @@ class PatternXX(BasePattern):
             raise HandshakeHasNotFinished(
                 "handshake is done but it is not marked as finished in `noise_state`"
             )
+
+        # NEW: Get negotiated muxer
+        # negotiated_muxer = None
+        if self.initiator_early_data_handler and hasattr(
+            self.initiator_early_data_handler, "match_muxers"
+        ):
+            pass
+            # negotiated_muxer =
+            # self.initiator_early_data_handler.match_muxers(is_initiator=True)
+
         transport_read_writer = NoiseTransportReadWriter(conn, noise_state)
         return SecureSession(
             local_peer=self.local_peer,
@@ -201,6 +291,8 @@ class PatternXX(BasePattern):
             remote_permanent_pubkey=remote_pubkey,
             is_initiator=True,
             conn=transport_read_writer,
+            # NOTE: negotiated_muxer would need to be added to SecureSession constructor
+            # For now, store it in connection metadata or similar
         )
 
     @staticmethod

libp2p/security/noise/pb/noise.proto

@@ -1,8 +1,13 @@
-syntax = "proto3";
+syntax = "proto2";
 package pb;
 
-message NoiseHandshakePayload {
+message NoiseExtensions {
-	bytes identity_key = 1;
+    repeated bytes webtransport_certhashes = 1;
-	bytes identity_sig = 2;
+    repeated string stream_muxers = 2;
-	bytes data = 3;
+}
+
+message NoiseHandshakePayload {
+	optional bytes identity_key = 1;
+	optional bytes identity_sig = 2;
+	optional bytes data = 3;
 }

libp2p/security/noise/pb/noise_pb2.py

@@ -13,13 +13,15 @@ _sym_db = _symbol_database.Default()
 
 
 
-DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n$libp2p/security/noise/pb/noise.proto\x12\x02pb\"Q\n\x15NoiseHandshakePayload\x12\x14\n\x0cidentity_key\x18\x01 \x01(\x0c\x12\x14\n\x0cidentity_sig\x18\x02 \x01(\x0c\x12\x0c\n\x04\x64\x61ta\x18\x03 \x01(\x0c\x62\x06proto3')
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n$libp2p/security/noise/pb/noise.proto\x12\x02pb\"I\n\x0fNoiseExtensions\x12\x1f\n\x17webtransport_certhashes\x18\x01 \x03(\x0c\x12\x15\n\rstream_muxers\x18\x02 \x03(\t\"Q\n\x15NoiseHandshakePayload\x12\x14\n\x0cidentity_key\x18\x01 \x01(\x0c\x12\x14\n\x0cidentity_sig\x18\x02 \x01(\x0c\x12\x0c\n\x04\x64\x61ta\x18\x03 \x01(\x0c')
 
 _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals())
 _builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'libp2p.security.noise.pb.noise_pb2', globals())
 if _descriptor._USE_C_DESCRIPTORS == False:
 
   DESCRIPTOR._options = None
-  _NOISEHANDSHAKEPAYLOAD._serialized_start=44
+  _NOISEEXTENSIONS._serialized_start=44
-  _NOISEHANDSHAKEPAYLOAD._serialized_end=125
+  _NOISEEXTENSIONS._serialized_end=117
+  _NOISEHANDSHAKEPAYLOAD._serialized_start=119
+  _NOISEHANDSHAKEPAYLOAD._serialized_end=200
 # @@protoc_insertion_point(module_scope)

libp2p/security/noise/pb/noise_pb2.pyi

@@ -4,12 +4,34 @@ isort:skip_file
 """
 
 import builtins
+import collections.abc
 import google.protobuf.descriptor
+import google.protobuf.internal.containers
 import google.protobuf.message
 import typing
 
 DESCRIPTOR: google.protobuf.descriptor.FileDescriptor
 
+@typing.final
+class NoiseExtensions(google.protobuf.message.Message):
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    WEBTRANSPORT_CERTHASHES_FIELD_NUMBER: builtins.int
+    STREAM_MUXERS_FIELD_NUMBER: builtins.int
+    @property
+    def webtransport_certhashes(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.bytes]: ...
+    @property
+    def stream_muxers(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...
+    def __init__(
+        self,
+        *,
+        webtransport_certhashes: collections.abc.Iterable[builtins.bytes] | None = ...,
+        stream_muxers: collections.abc.Iterable[builtins.str] | None = ...,
+    ) -> None: ...
+    def ClearField(self, field_name: typing.Literal["stream_muxers", b"stream_muxers", "webtransport_certhashes", b"webtransport_certhashes"]) -> None: ...
+
+global___NoiseExtensions = NoiseExtensions
+
 @typing.final
 class NoiseHandshakePayload(google.protobuf.message.Message):
     DESCRIPTOR: google.protobuf.descriptor.Descriptor
@@ -23,10 +45,11 @@ class NoiseHandshakePayload(google.protobuf.message.Message):
     def __init__(
         self,
         *,
-        identity_key: builtins.bytes = ...,
+        identity_key: builtins.bytes | None = ...,
-        identity_sig: builtins.bytes = ...,
+        identity_sig: builtins.bytes | None = ...,
-        data: builtins.bytes = ...,
+        data: builtins.bytes | None = ...,
     ) -> None: ...
+    def HasField(self, field_name: typing.Literal["data", b"data", "identity_key", b"identity_key", "identity_sig", b"identity_sig"]) -> builtins.bool: ...
     def ClearField(self, field_name: typing.Literal["data", b"data", "identity_key", b"identity_key", "identity_sig", b"identity_sig"]) -> None: ...
 
 global___NoiseHandshakePayload = NoiseHandshakePayload

libp2p/security/noise/transport.py

@@ -14,6 +14,7 @@ from libp2p.peer.id import (
     ID,
 )
 
+from .early_data import EarlyDataHandler, TransportEarlyDataHandler
 from .patterns import (
     IPattern,
     PatternXX,
@@ -26,35 +27,40 @@ class Transport(ISecureTransport):
     libp2p_privkey: PrivateKey
     noise_privkey: PrivateKey
     local_peer: ID
-    early_data: bytes | None
+    supported_muxers: list[TProtocol]
-    with_noise_pipes: bool
+    initiator_early_data_handler: EarlyDataHandler | None
+    responder_early_data_handler: EarlyDataHandler | None
 
     def __init__(
         self,
         libp2p_keypair: KeyPair,
         noise_privkey: PrivateKey,
-        early_data: bytes | None = None,
+        supported_muxers: list[TProtocol] | None = None,
-        with_noise_pipes: bool = False,
+        initiator_handler: EarlyDataHandler | None = None,
+        responder_handler: EarlyDataHandler | None = None,
     ) -> None:
         self.libp2p_privkey = libp2p_keypair.private_key
         self.noise_privkey = noise_privkey
         self.local_peer = ID.from_pubkey(libp2p_keypair.public_key)
-        self.early_data = early_data
+        self.supported_muxers = supported_muxers or []
-        self.with_noise_pipes = with_noise_pipes
 
-        if self.with_noise_pipes:
+        # Create default handlers for muxer negotiation if none provided
-            raise NotImplementedError
+        if initiator_handler is None and self.supported_muxers:
+            initiator_handler = TransportEarlyDataHandler(self.supported_muxers)
+        if responder_handler is None and self.supported_muxers:
+            responder_handler = TransportEarlyDataHandler(self.supported_muxers)
+
+        self.initiator_early_data_handler = initiator_handler
+        self.responder_early_data_handler = responder_handler
 
     def get_pattern(self) -> IPattern:
-        if self.with_noise_pipes:
+        return PatternXX(
-            raise NotImplementedError
+            self.local_peer,
-        else:
+            self.libp2p_privkey,
-            return PatternXX(
+            self.noise_privkey,
-                self.local_peer,
+            self.initiator_early_data_handler,
-                self.libp2p_privkey,
+            self.responder_early_data_handler,
-                self.noise_privkey,
+        )
-                self.early_data,
-            )
 
     async def secure_inbound(self, conn: IRawConnection) -> ISecureConn:
         pattern = self.get_pattern()

libp2p/utils/logging.py

@@ -1,7 +1,4 @@
 import atexit
-from datetime import (
-    datetime,
-)
 import logging
 import logging.handlers
 import os
@@ -21,6 +18,9 @@ log_queue: "queue.Queue[Any]" = queue.Queue()
 # Store the current listener to stop it on exit
 _current_listener: logging.handlers.QueueListener | None = None
 
+# Store the handlers for proper cleanup
+_current_handlers: list[logging.Handler] = []
+
 # Event to track when the listener is ready
 _listener_ready = threading.Event()
 
@@ -95,7 +95,7 @@ def setup_logging() -> None:
         - Child loggers inherit their parent's level unless explicitly set
         - The root libp2p logger controls the default level
     """
-    global _current_listener, _listener_ready
+    global _current_listener, _listener_ready, _current_handlers
 
     # Reset the event
     _listener_ready.clear()
@@ -105,6 +105,12 @@ def setup_logging() -> None:
         _current_listener.stop()
         _current_listener = None
 
+    # Close and clear existing handlers
+    for handler in _current_handlers:
+        if isinstance(handler, logging.FileHandler):
+            handler.close()
+    _current_handlers.clear()
+
     # Get the log level from environment variable
     debug_str = os.environ.get("LIBP2P_DEBUG", "")
 
@@ -148,13 +154,10 @@ def setup_logging() -> None:
         log_path = Path(log_file)
         log_path.parent.mkdir(parents=True, exist_ok=True)
     else:
-        # Default log file with timestamp and unique identifier
+        # Use cross-platform temp file creation
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+        from libp2p.utils.paths import create_temp_file
-        unique_id = os.urandom(4).hex()  # Add a unique identifier to prevent collisions
+
-        if os.name == "nt":  # Windows
+        log_file = str(create_temp_file(prefix="py-libp2p_", suffix=".log"))
-            log_file = f"C:\\Windows\\Temp\\py-libp2p_{timestamp}_{unique_id}.log"
-        else:  # Unix-like
-            log_file = f"/tmp/py-libp2p_{timestamp}_{unique_id}.log"
 
         # Print the log file path so users know where to find it
         print(f"Logging to: {log_file}", file=sys.stderr)
@@ -195,6 +198,9 @@ def setup_logging() -> None:
             logger.setLevel(level)
             logger.propagate = False  # Prevent message duplication
 
+    # Store handlers globally for cleanup
+    _current_handlers.extend(handlers)
+
     # Start the listener AFTER configuring all loggers
     _current_listener = logging.handlers.QueueListener(
         log_queue, *handlers, respect_handler_level=True
@@ -209,7 +215,13 @@ def setup_logging() -> None:
 @atexit.register
 def cleanup_logging() -> None:
     """Clean up logging resources on exit."""
-    global _current_listener
+    global _current_listener, _current_handlers
     if _current_listener is not None:
         _current_listener.stop()
         _current_listener = None
+
+    # Close all file handlers to ensure proper cleanup on Windows
+    for handler in _current_handlers:
+        if isinstance(handler, logging.FileHandler):
+            handler.close()
+    _current_handlers.clear()

libp2p/utils/paths.py

@@ -0,0 +1,267 @@
+"""
+Cross-platform path utilities for py-libp2p.
+
+This module provides standardized path operations to ensure consistent
+behavior across Windows, macOS, and Linux platforms.
+"""
+
+import os
+from pathlib import Path
+import sys
+import tempfile
+from typing import Union
+
+PathLike = Union[str, Path]
+
+
+def get_temp_dir() -> Path:
+    """
+    Get cross-platform temporary directory.
+
+    Returns:
+        Path: Platform-specific temporary directory path
+
+    """
+    return Path(tempfile.gettempdir())
+
+
+def get_project_root() -> Path:
+    """
+    Get the project root directory.
+
+    Returns:
+        Path: Path to the py-libp2p project root
+
+    """
+    # Navigate from libp2p/utils/paths.py to project root
+    return Path(__file__).parent.parent.parent
+
+
+def join_paths(*parts: PathLike) -> Path:
+    """
+    Cross-platform path joining.
+
+    Args:
+        *parts: Path components to join
+
+    Returns:
+        Path: Joined path using platform-appropriate separator
+
+    """
+    return Path(*parts)
+
+
+def ensure_dir_exists(path: PathLike) -> Path:
+    """
+    Ensure directory exists, create if needed.
+
+    Args:
+        path: Directory path to ensure exists
+
+    Returns:
+        Path: Path object for the directory
+
+    """
+    path_obj = Path(path)
+    path_obj.mkdir(parents=True, exist_ok=True)
+    return path_obj
+
+
+def get_config_dir() -> Path:
+    """
+    Get user config directory (cross-platform).
+
+    Returns:
+        Path: Platform-specific config directory
+
+    """
+    if os.name == "nt":  # Windows
+        appdata = os.environ.get("APPDATA", "")
+        if appdata:
+            return Path(appdata) / "py-libp2p"
+        else:
+            # Fallback to user home directory
+            return Path.home() / "AppData" / "Roaming" / "py-libp2p"
+    else:  # Unix-like (Linux, macOS)
+        return Path.home() / ".config" / "py-libp2p"
+
+
+def get_script_dir(script_path: PathLike | None = None) -> Path:
+    """
+    Get the directory containing a script file.
+
+    Args:
+        script_path: Path to the script file. If None, uses __file__
+
+    Returns:
+        Path: Directory containing the script
+
+    Raises:
+        RuntimeError: If script path cannot be determined
+
+    """
+    if script_path is None:
+        # This will be the directory of the calling script
+        import inspect
+
+        frame = inspect.currentframe()
+        if frame and frame.f_back:
+            script_path = frame.f_back.f_globals.get("__file__")
+        else:
+            raise RuntimeError("Could not determine script path")
+
+    if script_path is None:
+        raise RuntimeError("Script path is None")
+
+    return Path(script_path).parent.absolute()
+
+
+def create_temp_file(prefix: str = "py-libp2p_", suffix: str = ".log") -> Path:
+    """
+    Create a temporary file with a unique name.
+
+    Args:
+        prefix: File name prefix
+        suffix: File name suffix
+
+    Returns:
+        Path: Path to the created temporary file
+
+    """
+    temp_dir = get_temp_dir()
+    # Create a unique filename using timestamp and random bytes
+    import secrets
+    import time
+
+    timestamp = time.strftime("%Y%m%d_%H%M%S")
+    microseconds = f"{time.time() % 1:.6f}"[2:]  # Get microseconds as string
+    unique_id = secrets.token_hex(4)
+    filename = f"{prefix}{timestamp}_{microseconds}_{unique_id}{suffix}"
+
+    temp_file = temp_dir / filename
+    # Create the file by touching it
+    temp_file.touch()
+    return temp_file
+
+
+def resolve_relative_path(base_path: PathLike, relative_path: PathLike) -> Path:
+    """
+    Resolve a relative path from a base path.
+
+    Args:
+        base_path: Base directory path
+        relative_path: Relative path to resolve
+
+    Returns:
+        Path: Resolved absolute path
+
+    """
+    base = Path(base_path).resolve()
+    relative = Path(relative_path)
+
+    if relative.is_absolute():
+        return relative
+    else:
+        return (base / relative).resolve()
+
+
+def normalize_path(path: PathLike) -> Path:
+    """
+    Normalize a path, resolving any symbolic links and relative components.
+
+    Args:
+        path: Path to normalize
+
+    Returns:
+        Path: Normalized absolute path
+
+    """
+    return Path(path).resolve()
+
+
+def get_venv_path() -> Path | None:
+    """
+    Get virtual environment path if active.
+
+    Returns:
+        Path: Virtual environment path if active, None otherwise
+
+    """
+    venv_path = os.environ.get("VIRTUAL_ENV")
+    if venv_path:
+        return Path(venv_path)
+    return None
+
+
+def get_python_executable() -> Path:
+    """
+    Get current Python executable path.
+
+    Returns:
+        Path: Path to the current Python executable
+
+    """
+    return Path(sys.executable)
+
+
+def find_executable(name: str) -> Path | None:
+    """
+    Find executable in system PATH.
+
+    Args:
+        name: Name of the executable to find
+
+    Returns:
+        Path: Path to executable if found, None otherwise
+
+    """
+    # Check if name already contains path
+    if os.path.dirname(name):
+        path = Path(name)
+        if path.exists() and os.access(path, os.X_OK):
+            return path
+        return None
+
+    # Search in PATH
+    for path_dir in os.environ.get("PATH", "").split(os.pathsep):
+        if not path_dir:
+            continue
+        path = Path(path_dir) / name
+        if path.exists() and os.access(path, os.X_OK):
+            return path
+
+    return None
+
+
+def get_script_binary_path() -> Path:
+    """
+    Get path to script's binary directory.
+
+    Returns:
+        Path: Directory containing the script's binary
+
+    """
+    return get_python_executable().parent
+
+
+def get_binary_path(binary_name: str) -> Path | None:
+    """
+    Find binary in PATH or virtual environment.
+
+    Args:
+        binary_name: Name of the binary to find
+
+    Returns:
+        Path: Path to binary if found, None otherwise
+
+    """
+    # First check in virtual environment if active
+    venv_path = get_venv_path()
+    if venv_path:
+        venv_bin = venv_path / "bin" if os.name != "nt" else venv_path / "Scripts"
+        binary_path = venv_bin / binary_name
+        if binary_path.exists() and os.access(binary_path, os.X_OK):
+            return binary_path
+
+    # Fall back to system PATH
+    return find_executable(binary_name)

newsfragments/843.bugfix.rst

@@ -0,0 +1 @@
+Fixed message id type inconsistency in handle ihave and message id parsing improvement in handle iwant in pubsub module.

newsfragments/849.feature.rst

@@ -0,0 +1 @@
+Add automatic peer dialing in bootstrap module using trio.Nursery.

newsfragments/886.bugfix.rst

@@ -0,0 +1,2 @@
+Fixed cross-platform path handling by replacing hardcoded OS-specific
+paths with standardized utilities in core modules and examples.

scripts/audit_paths.py

@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""
+Audit script to identify path handling issues in the py-libp2p codebase.
+
+This script scans for patterns that should be migrated to use the new
+cross-platform path utilities.
+"""
+
+import argparse
+from pathlib import Path
+import re
+from typing import Any
+
+
+def scan_for_path_issues(directory: Path) -> dict[str, list[dict[str, Any]]]:
+    """
+    Scan for path handling issues in the codebase.
+
+    Args:
+        directory: Root directory to scan
+
+    Returns:
+        Dictionary mapping issue types to lists of found issues
+
+    """
+    issues = {
+        "hard_coded_slash": [],
+        "os_path_join": [],
+        "temp_hardcode": [],
+        "os_path_dirname": [],
+        "os_path_abspath": [],
+        "direct_path_concat": [],
+    }
+
+    # Patterns to search for
+    patterns = {
+        "hard_coded_slash": r'["\'][^"\']*\/[^"\']*["\']',
+        "os_path_join": r"os\.path\.join\(",
+        "temp_hardcode": r'["\']\/tmp\/|["\']C:\\\\',
+        "os_path_dirname": r"os\.path\.dirname\(",
+        "os_path_abspath": r"os\.path\.abspath\(",
+        "direct_path_concat": r'["\'][^"\']*["\']\s*\+\s*["\'][^"\']*["\']',
+    }
+
+    # Files to exclude
+    exclude_patterns = [
+        r"__pycache__",
+        r"\.git",
+        r"\.pytest_cache",
+        r"\.mypy_cache",
+        r"\.ruff_cache",
+        r"env/",
+        r"venv/",
+        r"\.venv/",
+    ]
+
+    for py_file in directory.rglob("*.py"):
+        # Skip excluded files
+        if any(re.search(pattern, str(py_file)) for pattern in exclude_patterns):
+            continue
+
+        try:
+            content = py_file.read_text(encoding="utf-8")
+        except UnicodeDecodeError:
+            print(f"Warning: Could not read {py_file} (encoding issue)")
+            continue
+
+        for issue_type, pattern in patterns.items():
+            matches = re.finditer(pattern, content, re.MULTILINE)
+            for match in matches:
+                line_num = content[: match.start()].count("\n") + 1
+                line_content = content.split("\n")[line_num - 1].strip()
+
+                issues[issue_type].append(
+                    {
+                        "file": py_file,
+                        "line": line_num,
+                        "content": match.group(),
+                        "full_line": line_content,
+                        "relative_path": py_file.relative_to(directory),
+                    }
+                )
+
+    return issues
+
+
+def generate_migration_suggestions(issues: dict[str, list[dict[str, Any]]]) -> str:
+    """
+    Generate migration suggestions for found issues.
+
+    Args:
+        issues: Dictionary of found issues
+
+    Returns:
+        Formatted string with migration suggestions
+
+    """
+    suggestions = []
+
+    for issue_type, issue_list in issues.items():
+        if not issue_list:
+            continue
+
+        suggestions.append(f"\n## {issue_type.replace('_', ' ').title()}")
+        suggestions.append(f"Found {len(issue_list)} instances:")
+
+        for issue in issue_list[:10]:  # Show first 10 examples
+            suggestions.append(f"\n### {issue['relative_path']}:{issue['line']}")
+            suggestions.append("```python")
+            suggestions.append("# Current code:")
+            suggestions.append(f"{issue['full_line']}")
+            suggestions.append("```")
+
+            # Add migration suggestion based on issue type
+            if issue_type == "os_path_join":
+                suggestions.append("```python")
+                suggestions.append("# Suggested fix:")
+                suggestions.append("from libp2p.utils.paths import join_paths")
+                suggestions.append(
+                    "# Replace os.path.join(a, b, c) with join_paths(a, b, c)"
+                )
+                suggestions.append("```")
+            elif issue_type == "temp_hardcode":
+                suggestions.append("```python")
+                suggestions.append("# Suggested fix:")
+                suggestions.append(
+                    "from libp2p.utils.paths import get_temp_dir, create_temp_file"
+                )
+                temp_fix_msg = (
+                    "# Replace hard-coded temp paths with get_temp_dir() or "
+                    "create_temp_file()"
+                )
+                suggestions.append(temp_fix_msg)
+                suggestions.append("```")
+            elif issue_type == "os_path_dirname":
+                suggestions.append("```python")
+                suggestions.append("# Suggested fix:")
+                suggestions.append("from libp2p.utils.paths import get_script_dir")
+                script_dir_fix_msg = (
+                    "# Replace os.path.dirname(os.path.abspath(__file__)) with "
+                    "get_script_dir(__file__)"
+                )
+                suggestions.append(script_dir_fix_msg)
+                suggestions.append("```")
+
+        if len(issue_list) > 10:
+            suggestions.append(f"\n... and {len(issue_list) - 10} more instances")
+
+    return "\n".join(suggestions)
+
+
+def generate_summary_report(issues: dict[str, list[dict[str, Any]]]) -> str:
+    """
+    Generate a summary report of all found issues.
+
+    Args:
+        issues: Dictionary of found issues
+
+    Returns:
+        Formatted summary report
+
+    """
+    total_issues = sum(len(issue_list) for issue_list in issues.values())
+
+    report = [
+        "# Cross-Platform Path Handling Audit Report",
+        "",
+        "## Summary",
+        f"Total issues found: {total_issues}",
+        "",
+        "## Issue Breakdown:",
+    ]
+
+    for issue_type, issue_list in issues.items():
+        if issue_list:
+            issue_title = issue_type.replace("_", " ").title()
+            instances_count = len(issue_list)
+            report.append(f"- **{issue_title}**: {instances_count} instances")
+
+    report.append("")
+    report.append("## Priority Matrix:")
+    report.append("")
+    report.append("| Priority | Issue Type | Risk Level | Impact |")
+    report.append("|----------|------------|------------|---------|")
+
+    priority_map = {
+        "temp_hardcode": (
+            "🔴 P0",
+            "HIGH",
+            "Core functionality fails on different platforms",
+        ),
+        "os_path_join": ("🟡 P1", "MEDIUM", "Examples and utilities may break"),
+        "os_path_dirname": ("🟡 P1", "MEDIUM", "Script location detection issues"),
+        "hard_coded_slash": ("🟢 P2", "LOW", "Future-proofing and consistency"),
+        "os_path_abspath": ("🟢 P2", "LOW", "Path resolution consistency"),
+        "direct_path_concat": ("🟢 P2", "LOW", "String concatenation issues"),
+    }
+
+    for issue_type, issue_list in issues.items():
+        if issue_list:
+            priority, risk, impact = priority_map.get(
+                issue_type, ("🟢 P2", "LOW", "General improvement")
+            )
+            issue_title = issue_type.replace("_", " ").title()
+            report.append(f"| {priority} | {issue_title} | {risk} | {impact} |")
+
+    return "\n".join(report)
+
+
+def main():
+    """Main function to run the audit."""
+    parser = argparse.ArgumentParser(
+        description="Audit py-libp2p codebase for path handling issues"
+    )
+    parser.add_argument(
+        "--directory",
+        default=".",
+        help="Directory to scan (default: current directory)",
+    )
+    parser.add_argument("--output", help="Output file for detailed report")
+    parser.add_argument(
+        "--summary-only", action="store_true", help="Only show summary report"
+    )
+
+    args = parser.parse_args()
+
+    directory = Path(args.directory)
+    if not directory.exists():
+        print(f"Error: Directory {directory} does not exist")
+        return 1
+
+    print("🔍 Scanning for path handling issues...")
+    issues = scan_for_path_issues(directory)
+
+    # Generate and display summary
+    summary = generate_summary_report(issues)
+    print(summary)
+
+    if not args.summary_only:
+        # Generate detailed suggestions
+        suggestions = generate_migration_suggestions(issues)
+
+        if args.output:
+            with open(args.output, "w", encoding="utf-8") as f:
+                f.write(summary)
+                f.write(suggestions)
+            print(f"\n📄 Detailed report saved to {args.output}")
+        else:
+            print(suggestions)
+
+    return 0
+
+
+if __name__ == "__main__":
+    exit(main())

tests/core/pubsub/test_gossipsub.py

@@ -1,4 +1,8 @@
 import random
+from unittest.mock import (
+    AsyncMock,
+    MagicMock,
+)
 
 import pytest
 import trio
@@ -7,6 +11,9 @@ from libp2p.pubsub.gossipsub import (
     PROTOCOL_ID,
     GossipSub,
 )
+from libp2p.pubsub.pb import (
+    rpc_pb2,
+)
 from libp2p.tools.utils import (
     connect,
 )
@@ -754,3 +761,173 @@ async def test_single_host():
         assert connected_peers == 0, (
             f"Single host has {connected_peers} connections, expected 0"
         )
+
+
+@pytest.mark.trio
+async def test_handle_ihave(monkeypatch):
+    async with PubsubFactory.create_batch_with_gossipsub(2) as pubsubs_gsub:
+        gossipsub_routers = []
+        for pubsub in pubsubs_gsub:
+            if isinstance(pubsub.router, GossipSub):
+                gossipsub_routers.append(pubsub.router)
+        gossipsubs = tuple(gossipsub_routers)
+
+        index_alice = 0
+        index_bob = 1
+        id_bob = pubsubs_gsub[index_bob].my_id
+
+        # Connect Alice and Bob
+        await connect(pubsubs_gsub[index_alice].host, pubsubs_gsub[index_bob].host)
+        await trio.sleep(0.1)  # Allow connections to establish
+
+        # Mock emit_iwant to capture calls
+        mock_emit_iwant = AsyncMock()
+        monkeypatch.setattr(gossipsubs[index_alice], "emit_iwant", mock_emit_iwant)
+
+        # Create a test message ID as a string representation of a (seqno, from) tuple
+        test_seqno = b"1234"
+        test_from = id_bob.to_bytes()
+        test_msg_id = f"(b'{test_seqno.hex()}', b'{test_from.hex()}')"
+        ihave_msg = rpc_pb2.ControlIHave(messageIDs=[test_msg_id])
+
+        # Mock seen_messages.cache to avoid false positives
+        monkeypatch.setattr(pubsubs_gsub[index_alice].seen_messages, "cache", {})
+
+        # Simulate Bob sending IHAVE to Alice
+        await gossipsubs[index_alice].handle_ihave(ihave_msg, id_bob)
+
+        # Check if emit_iwant was called with the correct message ID
+        mock_emit_iwant.assert_called_once()
+        called_args = mock_emit_iwant.call_args[0]
+        assert called_args[0] == [test_msg_id]  # Expected message IDs
+        assert called_args[1] == id_bob  # Sender peer ID
+
+
+@pytest.mark.trio
+async def test_handle_iwant(monkeypatch):
+    async with PubsubFactory.create_batch_with_gossipsub(2) as pubsubs_gsub:
+        gossipsub_routers = []
+        for pubsub in pubsubs_gsub:
+            if isinstance(pubsub.router, GossipSub):
+                gossipsub_routers.append(pubsub.router)
+        gossipsubs = tuple(gossipsub_routers)
+
+        index_alice = 0
+        index_bob = 1
+        id_alice = pubsubs_gsub[index_alice].my_id
+
+        # Connect Alice and Bob
+        await connect(pubsubs_gsub[index_alice].host, pubsubs_gsub[index_bob].host)
+        await trio.sleep(0.1)  # Allow connections to establish
+
+        # Mock mcache.get to return a message
+        test_message = rpc_pb2.Message(data=b"test_data")
+        test_seqno = b"1234"
+        test_from = id_alice.to_bytes()
+
+        # ✅ Correct: use raw tuple and str() to serialize, no hex()
+        test_msg_id = str((test_seqno, test_from))
+
+        mock_mcache_get = MagicMock(return_value=test_message)
+        monkeypatch.setattr(gossipsubs[index_bob].mcache, "get", mock_mcache_get)
+
+        # Mock write_msg to capture the sent packet
+        mock_write_msg = AsyncMock()
+        monkeypatch.setattr(gossipsubs[index_bob].pubsub, "write_msg", mock_write_msg)
+
+        # Simulate Alice sending IWANT to Bob
+        iwant_msg = rpc_pb2.ControlIWant(messageIDs=[test_msg_id])
+        await gossipsubs[index_bob].handle_iwant(iwant_msg, id_alice)
+
+        # Check if write_msg was called with the correct packet
+        mock_write_msg.assert_called_once()
+        packet = mock_write_msg.call_args[0][1]
+        assert isinstance(packet, rpc_pb2.RPC)
+        assert len(packet.publish) == 1
+        assert packet.publish[0] == test_message
+
+        # Verify that mcache.get was called with the correct parsed message ID
+        mock_mcache_get.assert_called_once()
+        called_msg_id = mock_mcache_get.call_args[0][0]
+        assert isinstance(called_msg_id, tuple)
+        assert called_msg_id == (test_seqno, test_from)
+
+
+@pytest.mark.trio
+async def test_handle_iwant_invalid_msg_id(monkeypatch):
+    """
+    Test that handle_iwant raises ValueError for malformed message IDs.
+    """
+    async with PubsubFactory.create_batch_with_gossipsub(2) as pubsubs_gsub:
+        gossipsub_routers = []
+        for pubsub in pubsubs_gsub:
+            if isinstance(pubsub.router, GossipSub):
+                gossipsub_routers.append(pubsub.router)
+        gossipsubs = tuple(gossipsub_routers)
+
+        index_alice = 0
+        index_bob = 1
+        id_alice = pubsubs_gsub[index_alice].my_id
+
+        await connect(pubsubs_gsub[index_alice].host, pubsubs_gsub[index_bob].host)
+        await trio.sleep(0.1)
+
+        # Malformed message ID (not a tuple string)
+        malformed_msg_id = "not_a_valid_msg_id"
+        iwant_msg = rpc_pb2.ControlIWant(messageIDs=[malformed_msg_id])
+
+        # Mock mcache.get and write_msg to ensure they are not called
+        mock_mcache_get = MagicMock()
+        monkeypatch.setattr(gossipsubs[index_bob].mcache, "get", mock_mcache_get)
+        mock_write_msg = AsyncMock()
+        monkeypatch.setattr(gossipsubs[index_bob].pubsub, "write_msg", mock_write_msg)
+
+        with pytest.raises(ValueError):
+            await gossipsubs[index_bob].handle_iwant(iwant_msg, id_alice)
+        mock_mcache_get.assert_not_called()
+        mock_write_msg.assert_not_called()
+
+        # Message ID that's a tuple string but not (bytes, bytes)
+        invalid_tuple_msg_id = "('abc', 123)"
+        iwant_msg = rpc_pb2.ControlIWant(messageIDs=[invalid_tuple_msg_id])
+        with pytest.raises(ValueError):
+            await gossipsubs[index_bob].handle_iwant(iwant_msg, id_alice)
+        mock_mcache_get.assert_not_called()
+        mock_write_msg.assert_not_called()
+
+
+@pytest.mark.trio
+async def test_handle_ihave_empty_message_ids(monkeypatch):
+    """
+    Test that handle_ihave with an empty messageIDs list does not call emit_iwant.
+    """
+    async with PubsubFactory.create_batch_with_gossipsub(2) as pubsubs_gsub:
+        gossipsub_routers = []
+        for pubsub in pubsubs_gsub:
+            if isinstance(pubsub.router, GossipSub):
+                gossipsub_routers.append(pubsub.router)
+        gossipsubs = tuple(gossipsub_routers)
+
+        index_alice = 0
+        index_bob = 1
+        id_bob = pubsubs_gsub[index_bob].my_id
+
+        # Connect Alice and Bob
+        await connect(pubsubs_gsub[index_alice].host, pubsubs_gsub[index_bob].host)
+        await trio.sleep(0.1)  # Allow connections to establish
+
+        # Mock emit_iwant to capture calls
+        mock_emit_iwant = AsyncMock()
+        monkeypatch.setattr(gossipsubs[index_alice], "emit_iwant", mock_emit_iwant)
+
+        # Empty messageIDs list
+        ihave_msg = rpc_pb2.ControlIHave(messageIDs=[])
+
+        # Mock seen_messages.cache to avoid false positives
+        monkeypatch.setattr(pubsubs_gsub[index_alice].seen_messages, "cache", {})
+
+        # Simulate Bob sending IHAVE to Alice
+        await gossipsubs[index_alice].handle_ihave(ihave_msg, id_bob)
+
+        # emit_iwant should not be called since there are no message IDs
+        mock_emit_iwant.assert_not_called()

tests/core/security/noise/test_noise_extension_protobuf.py

@@ -0,0 +1,13 @@
+from libp2p.security.noise.pb import noise_pb2 as noise_pb
+
+
+def test_noise_extensions_serialization():
+    # Test NoiseExtensions
+    ext = noise_pb.NoiseExtensions()
+    ext.stream_muxers.append("/mplex/6.7.0")
+    ext.stream_muxers.append("/yamux/1.0.0")
+
+    # Serialize and deserialize
+    data = ext.SerializeToString()
+    ext2 = noise_pb.NoiseExtensions.FromString(data)
+    assert list(ext2.stream_muxers) == ["/mplex/6.7.0", "/yamux/1.0.0"]

tests/utils/factories.py

@@ -173,8 +173,7 @@ def noise_transport_factory(key_pair: KeyPair) -> ISecureTransport:
     return NoiseTransport(
         libp2p_keypair=key_pair,
         noise_privkey=noise_static_key_factory(),
-        early_data=None,
+        # TODO: add early data
-        with_noise_pipes=False,
     )
 
 

tests/utils/test_logging.py

@@ -15,6 +15,7 @@ import pytest
 import trio
 
 from libp2p.utils.logging import (
+    _current_handlers,
     _current_listener,
     _listener_ready,
     log_queue,
@@ -24,13 +25,19 @@ from libp2p.utils.logging import (
 
 def _reset_logging():
     """Reset all logging state."""
-    global _current_listener, _listener_ready
+    global _current_listener, _listener_ready, _current_handlers
 
     # Stop existing listener if any
     if _current_listener is not None:
         _current_listener.stop()
         _current_listener = None
 
+    # Close all file handlers to ensure proper cleanup on Windows
+    for handler in _current_handlers:
+        if isinstance(handler, logging.FileHandler):
+            handler.close()
+    _current_handlers.clear()
+
     # Reset the event
     _listener_ready = threading.Event()
 
@@ -174,6 +181,15 @@ async def test_custom_log_file(clean_env):
         if _current_listener is not None:
             _current_listener.stop()
 
+        # Give a moment for the listener to fully stop
+        await trio.sleep(0.05)
+
+        # Close all file handlers to release the file
+        for handler in _current_handlers:
+            if isinstance(handler, logging.FileHandler):
+                handler.flush()  # Ensure all writes are flushed
+                handler.close()
+
         # Check if the file exists and contains our message
         assert log_file.exists()
         content = log_file.read_text()
@@ -185,16 +201,15 @@ async def test_default_log_file(clean_env):
     """Test logging to the default file path."""
     os.environ["LIBP2P_DEBUG"] = "INFO"
 
-    with patch("libp2p.utils.logging.datetime") as mock_datetime:
+    with patch("libp2p.utils.paths.create_temp_file") as mock_create_temp:
-        # Mock the timestamp to have a predictable filename
+        # Mock the temp file creation to return a predictable path
-        mock_datetime.now.return_value.strftime.return_value = "20240101_120000"
+        mock_temp_file = (
+            Path(tempfile.gettempdir()) / "test_py-libp2p_20240101_120000.log"
+        )
+        mock_create_temp.return_value = mock_temp_file
 
         # Remove the log file if it exists
-        if os.name == "nt":  # Windows
+        mock_temp_file.unlink(missing_ok=True)
-            log_file = Path("C:/Windows/Temp/20240101_120000_py-libp2p.log")
-        else:  # Unix-like
-            log_file = Path("/tmp/20240101_120000_py-libp2p.log")
-        log_file.unlink(missing_ok=True)
 
         setup_logging()
 
@@ -211,9 +226,18 @@ async def test_default_log_file(clean_env):
         if _current_listener is not None:
             _current_listener.stop()
 
-        # Check the default log file
+        # Give a moment for the listener to fully stop
-        if log_file.exists():  # Only check content if we have write permission
+        await trio.sleep(0.05)
-            content = log_file.read_text()
+
+        # Close all file handlers to release the file
+        for handler in _current_handlers:
+            if isinstance(handler, logging.FileHandler):
+                handler.flush()  # Ensure all writes are flushed
+                handler.close()
+
+        # Check the mocked temp file
+        if mock_temp_file.exists():
+            content = mock_temp_file.read_text()
             assert "Test message" in content
 
 

tests/utils/test_paths.py

@@ -0,0 +1,290 @@
+"""
+Tests for cross-platform path utilities.
+"""
+
+import os
+from pathlib import Path
+import tempfile
+
+import pytest
+
+from libp2p.utils.paths import (
+    create_temp_file,
+    ensure_dir_exists,
+    find_executable,
+    get_binary_path,
+    get_config_dir,
+    get_project_root,
+    get_python_executable,
+    get_script_binary_path,
+    get_script_dir,
+    get_temp_dir,
+    get_venv_path,
+    join_paths,
+    normalize_path,
+    resolve_relative_path,
+)
+
+
+class TestPathUtilities:
+    """Test cross-platform path utilities."""
+
+    def test_get_temp_dir(self):
+        """Test that temp directory is accessible and exists."""
+        temp_dir = get_temp_dir()
+        assert isinstance(temp_dir, Path)
+        assert temp_dir.exists()
+        assert temp_dir.is_dir()
+        # Should match system temp directory
+        assert temp_dir == Path(tempfile.gettempdir())
+
+    def test_get_project_root(self):
+        """Test that project root is correctly determined."""
+        project_root = get_project_root()
+        assert isinstance(project_root, Path)
+        assert project_root.exists()
+        # Should contain pyproject.toml
+        assert (project_root / "pyproject.toml").exists()
+        # Should contain libp2p directory
+        assert (project_root / "libp2p").exists()
+
+    def test_join_paths(self):
+        """Test cross-platform path joining."""
+        # Test with strings
+        result = join_paths("a", "b", "c")
+        expected = Path("a") / "b" / "c"
+        assert result == expected
+
+        # Test with mixed types
+        result = join_paths("a", Path("b"), "c")
+        expected = Path("a") / "b" / "c"
+        assert result == expected
+
+        # Test with absolute path
+        result = join_paths("/absolute", "path")
+        expected = Path("/absolute") / "path"
+        assert result == expected
+
+    def test_ensure_dir_exists(self, tmp_path):
+        """Test directory creation and existence checking."""
+        # Test creating new directory
+        new_dir = tmp_path / "new_dir"
+        result = ensure_dir_exists(new_dir)
+        assert result == new_dir
+        assert new_dir.exists()
+        assert new_dir.is_dir()
+
+        # Test creating nested directory
+        nested_dir = tmp_path / "parent" / "child" / "grandchild"
+        result = ensure_dir_exists(nested_dir)
+        assert result == nested_dir
+        assert nested_dir.exists()
+        assert nested_dir.is_dir()
+
+        # Test with existing directory
+        result = ensure_dir_exists(new_dir)
+        assert result == new_dir
+        assert new_dir.exists()
+
+    def test_get_config_dir(self):
+        """Test platform-specific config directory."""
+        config_dir = get_config_dir()
+        assert isinstance(config_dir, Path)
+
+        if os.name == "nt":  # Windows
+            # Should be in AppData/Roaming or user home
+            assert "AppData" in str(config_dir) or "py-libp2p" in str(config_dir)
+        else:  # Unix-like
+            # Should be in ~/.config
+            assert ".config" in str(config_dir)
+            assert "py-libp2p" in str(config_dir)
+
+    def test_get_script_dir(self):
+        """Test script directory detection."""
+        # Test with current file
+        script_dir = get_script_dir(__file__)
+        assert isinstance(script_dir, Path)
+        assert script_dir.exists()
+        assert script_dir.is_dir()
+        # Should contain this test file
+        assert (script_dir / "test_paths.py").exists()
+
+    def test_create_temp_file(self):
+        """Test temporary file creation."""
+        temp_file = create_temp_file()
+        assert isinstance(temp_file, Path)
+        assert temp_file.parent == get_temp_dir()
+        assert temp_file.name.startswith("py-libp2p_")
+        assert temp_file.name.endswith(".log")
+
+        # Test with custom prefix and suffix
+        temp_file = create_temp_file(prefix="test_", suffix=".txt")
+        assert temp_file.name.startswith("test_")
+        assert temp_file.name.endswith(".txt")
+
+    def test_resolve_relative_path(self, tmp_path):
+        """Test relative path resolution."""
+        base_path = tmp_path / "base"
+        base_path.mkdir()
+
+        # Test relative path
+        relative_path = "subdir/file.txt"
+        result = resolve_relative_path(base_path, relative_path)
+        expected = (base_path / "subdir" / "file.txt").resolve()
+        assert result == expected
+
+        # Test absolute path (platform-agnostic)
+        if os.name == "nt":  # Windows
+            absolute_path = "C:\\absolute\\path"
+        else:  # Unix-like
+            absolute_path = "/absolute/path"
+        result = resolve_relative_path(base_path, absolute_path)
+        assert result == Path(absolute_path)
+
+    def test_normalize_path(self, tmp_path):
+        """Test path normalization."""
+        # Test with relative path
+        relative_path = tmp_path / ".." / "normalize_test"
+        result = normalize_path(relative_path)
+        assert result.is_absolute()
+        assert "normalize_test" in str(result)
+
+        # Test with absolute path
+        absolute_path = tmp_path / "test_file"
+        result = normalize_path(absolute_path)
+        assert result.is_absolute()
+        assert result == absolute_path.resolve()
+
+    def test_get_venv_path(self, monkeypatch):
+        """Test virtual environment path detection."""
+        # Test when no virtual environment is active
+        # Temporarily clear VIRTUAL_ENV to test the "no venv" case
+        monkeypatch.delenv("VIRTUAL_ENV", raising=False)
+        result = get_venv_path()
+        assert result is None
+
+        # Test when virtual environment is active
+        test_venv_path = "/path/to/venv"
+        monkeypatch.setenv("VIRTUAL_ENV", test_venv_path)
+        result = get_venv_path()
+        assert result == Path(test_venv_path)
+
+    def test_get_python_executable(self):
+        """Test Python executable path detection."""
+        result = get_python_executable()
+        assert isinstance(result, Path)
+        assert result.exists()
+        assert result.name.startswith("python")
+
+    def test_find_executable(self):
+        """Test executable finding in PATH."""
+        # Test with non-existent executable
+        result = find_executable("nonexistent_executable")
+        assert result is None
+
+        # Test with existing executable (python should be available)
+        result = find_executable("python")
+        if result:
+            assert isinstance(result, Path)
+            assert result.exists()
+
+    def test_get_script_binary_path(self):
+        """Test script binary path detection."""
+        result = get_script_binary_path()
+        assert isinstance(result, Path)
+        assert result.exists()
+        assert result.is_dir()
+
+    def test_get_binary_path(self, monkeypatch):
+        """Test binary path resolution with virtual environment."""
+        # Test when no virtual environment is active
+        result = get_binary_path("python")
+        if result:
+            assert isinstance(result, Path)
+            assert result.exists()
+
+        # Test when virtual environment is active
+        test_venv_path = "/path/to/venv"
+        monkeypatch.setenv("VIRTUAL_ENV", test_venv_path)
+        # This test is more complex as it depends on the actual venv structure
+        # We'll just verify the function doesn't crash
+        result = get_binary_path("python")
+        # Result can be None if binary not found in venv
+        if result:
+            assert isinstance(result, Path)
+
+
+class TestCrossPlatformCompatibility:
+    """Test cross-platform compatibility."""
+
+    def test_config_dir_platform_specific_windows(self, monkeypatch):
+        """Test config directory respects Windows conventions."""
+        import platform
+
+        # Only run this test on Windows systems
+        if platform.system() != "Windows":
+            pytest.skip("This test only runs on Windows systems")
+
+        monkeypatch.setattr("os.name", "nt")
+        monkeypatch.setenv("APPDATA", "C:\\Users\\Test\\AppData\\Roaming")
+        config_dir = get_config_dir()
+        assert "AppData" in str(config_dir)
+        assert "py-libp2p" in str(config_dir)
+
+    def test_path_separators_consistent(self):
+        """Test that path separators are handled consistently."""
+        # Test that join_paths uses platform-appropriate separators
+        result = join_paths("dir1", "dir2", "file.txt")
+        expected = Path("dir1") / "dir2" / "file.txt"
+        assert result == expected
+
+        # Test that the result uses correct separators for the platform
+        if os.name == "nt":  # Windows
+            assert "\\" in str(result) or "/" in str(result)
+        else:  # Unix-like
+            assert "/" in str(result)
+
+    def test_temp_file_uniqueness(self):
+        """Test that temporary files have unique names."""
+        files = set()
+        for _ in range(10):
+            temp_file = create_temp_file()
+            assert temp_file not in files
+            files.add(temp_file)
+
+
+class TestBackwardCompatibility:
+    """Test backward compatibility with existing code patterns."""
+
+    def test_path_operations_equivalent(self):
+        """Test that new path operations are equivalent to old os.path operations."""
+        # Test join_paths vs os.path.join
+        parts = ["a", "b", "c"]
+        new_result = join_paths(*parts)
+        old_result = Path(os.path.join(*parts))
+        assert new_result == old_result
+
+        # Test get_script_dir vs os.path.dirname(os.path.abspath(__file__))
+        new_script_dir = get_script_dir(__file__)
+        old_script_dir = Path(os.path.dirname(os.path.abspath(__file__)))
+        assert new_script_dir == old_script_dir
+
+    def test_existing_functionality_preserved(self):
+        """Ensure no existing functionality is broken."""
+        # Test that all functions return Path objects
+        assert isinstance(get_temp_dir(), Path)
+        assert isinstance(get_project_root(), Path)
+        assert isinstance(join_paths("a", "b"), Path)
+        assert isinstance(ensure_dir_exists(tempfile.gettempdir()), Path)
+        assert isinstance(get_config_dir(), Path)
+        assert isinstance(get_script_dir(__file__), Path)
+        assert isinstance(create_temp_file(), Path)
+        assert isinstance(resolve_relative_path(".", "test"), Path)
+        assert isinstance(normalize_path("."), Path)
+        assert isinstance(get_python_executable(), Path)
+        assert isinstance(get_script_binary_path(), Path)
+
+        # Test optional return types
+        venv_path = get_venv_path()
+        if venv_path is not None:
+            assert isinstance(venv_path, Path)