feat/606-enable-nat-traversal-via-hole-punching (#668)

* feat: base implementation of dcutr for hole-punching

* chore: removed circuit-relay imports from __init__

* feat: implemented dcutr protocol

* added test suite with mock setup

* Fix pre-commit hook issues in DCUtR implementation

* usages of CONNECT_TYPE and SYNC_TYPE have been replaced with HolePunch.Type.CONNECT and HolePunch.Type.SYNC

* added unit tests for dcutr and nat module and

* added multiaddr.get_peer_id() with proper DNS address handling and fixed method signature inconsistencies

* added assertions to verify DCUtR hole punch result in integration test

---------

Co-authored-by: Manu Sheel Gupta <manusheel.edu@gmail.com>

libp2p/relay/circuit_v2/nat.py

@@ -0,0 +1,300 @@
+"""
+NAT traversal utilities for libp2p.
+
+This module provides utilities for NAT traversal and reachability detection.
+"""
+
+import ipaddress
+import logging
+
+from multiaddr import (
+    Multiaddr,
+)
+
+from libp2p.abc import (
+    IHost,
+    INetConn,
+)
+from libp2p.peer.id import (
+    ID,
+)
+
+logger = logging.getLogger("libp2p.relay.circuit_v2.nat")
+
+# Timeout for reachability checks
+REACHABILITY_TIMEOUT = 10  # seconds
+
+# Define private IP ranges
+PRIVATE_IP_RANGES = [
+    ("10.0.0.0", "10.255.255.255"),  # Class A private network: 10.0.0.0/8
+    ("172.16.0.0", "172.31.255.255"),  # Class B private network: 172.16.0.0/12
+    ("192.168.0.0", "192.168.255.255"),  # Class C private network: 192.168.0.0/16
+]
+
+# Link-local address range: 169.254.0.0/16
+LINK_LOCAL_RANGE = ("169.254.0.0", "169.254.255.255")
+
+# Loopback address range: 127.0.0.0/8
+LOOPBACK_RANGE = ("127.0.0.0", "127.255.255.255")
+
+
+def ip_to_int(ip: str) -> int:
+    """
+    Convert an IP address to an integer.
+
+    Parameters
+    ----------
+    ip : str
+        IP address to convert
+
+    Returns
+    -------
+    int
+        Integer representation of the IP
+
+    """
+    try:
+        return int(ipaddress.IPv4Address(ip))
+    except ipaddress.AddressValueError:
+        # Handle IPv6 addresses
+        return int(ipaddress.IPv6Address(ip))
+
+
+def is_ip_in_range(ip: str, start_range: str, end_range: str) -> bool:
+    """
+    Check if an IP address is within a range.
+
+    Parameters
+    ----------
+    ip : str
+        IP address to check
+    start_range : str
+        Start of the range
+    end_range : str
+        End of the range
+
+    Returns
+    -------
+    bool
+        True if the IP is in the range
+
+    """
+    try:
+        ip_int = ip_to_int(ip)
+        start_int = ip_to_int(start_range)
+        end_int = ip_to_int(end_range)
+        return start_int <= ip_int <= end_int
+    except Exception:
+        return False
+
+
+def is_private_ip(ip: str) -> bool:
+    """
+    Check if an IP address is private.
+
+    Parameters
+    ----------
+    ip : str
+        IP address to check
+
+    Returns
+    -------
+    bool
+        True if IP is private
+
+    """
+    for start_range, end_range in PRIVATE_IP_RANGES:
+        if is_ip_in_range(ip, start_range, end_range):
+            return True
+
+    # Check for link-local addresses
+    if is_ip_in_range(ip, *LINK_LOCAL_RANGE):
+        return True
+
+    # Check for loopback addresses
+    if is_ip_in_range(ip, *LOOPBACK_RANGE):
+        return True
+
+    return False
+
+
+def extract_ip_from_multiaddr(addr: Multiaddr) -> str | None:
+    """
+    Extract the IP address from a multiaddr.
+
+    Parameters
+    ----------
+    addr : Multiaddr
+        Multiaddr to extract from
+
+    Returns
+    -------
+    Optional[str]
+        IP address or None if not found
+
+    """
+    # Convert to string representation
+    addr_str = str(addr)
+
+    # Look for IPv4 address
+    ipv4_start = addr_str.find("/ip4/")
+    if ipv4_start != -1:
+        # Extract the IPv4 address
+        ipv4_end = addr_str.find("/", ipv4_start + 5)
+        if ipv4_end != -1:
+            return addr_str[ipv4_start + 5 : ipv4_end]
+
+    # Look for IPv6 address
+    ipv6_start = addr_str.find("/ip6/")
+    if ipv6_start != -1:
+        # Extract the IPv6 address
+        ipv6_end = addr_str.find("/", ipv6_start + 5)
+        if ipv6_end != -1:
+            return addr_str[ipv6_start + 5 : ipv6_end]
+
+    return None
+
+
+class ReachabilityChecker:
+    """
+    Utility class for checking peer reachability.
+
+    This class assesses whether a peer's addresses are likely
+    to be directly reachable or behind NAT.
+    """
+
+    def __init__(self, host: IHost):
+        """
+        Initialize the reachability checker.
+
+        Parameters
+        ----------
+        host : IHost
+            The libp2p host
+
+        """
+        self.host = host
+        self._peer_reachability: dict[ID, bool] = {}
+        self._known_public_peers: set[ID] = set()
+
+    def is_addr_public(self, addr: Multiaddr) -> bool:
+        """
+        Check if an address is likely to be publicly reachable.
+
+        Parameters
+        ----------
+        addr : Multiaddr
+            The multiaddr to check
+
+        Returns
+        -------
+        bool
+            True if address is likely public
+
+        """
+        # Extract the IP address
+        ip = extract_ip_from_multiaddr(addr)
+        if not ip:
+            return False
+
+        # Check if it's a private IP
+        return not is_private_ip(ip)
+
+    def get_public_addrs(self, addrs: list[Multiaddr]) -> list[Multiaddr]:
+        """
+        Filter a list of addresses to only include likely public ones.
+
+        Parameters
+        ----------
+        addrs : List[Multiaddr]
+            List of addresses to filter
+
+        Returns
+        -------
+        List[Multiaddr]
+            List of likely public addresses
+
+        """
+        return [addr for addr in addrs if self.is_addr_public(addr)]
+
+    async def check_peer_reachability(self, peer_id: ID) -> bool:
+        """
+        Check if a peer is directly reachable.
+
+        Parameters
+        ----------
+        peer_id : ID
+            The peer ID to check
+
+        Returns
+        -------
+        bool
+            True if peer is likely directly reachable
+
+        """
+        # Check if we already know
+        if peer_id in self._peer_reachability:
+            return self._peer_reachability[peer_id]
+
+        # Check if the peer is connected
+        network = self.host.get_network()
+        connections: INetConn | list[INetConn] | None = network.connections.get(peer_id)
+        if not connections:
+            # Not connected, can't determine reachability
+            return False
+
+        # Check if any connection is direct (not relayed)
+        if isinstance(connections, list):
+            for conn in connections:
+                # Get the transport addresses
+                addrs = conn.get_transport_addresses()
+
+                # If any address doesn't start with /p2p-circuit,
+                # it's a direct connection
+                if any(not str(addr).startswith("/p2p-circuit") for addr in addrs):
+                    self._peer_reachability[peer_id] = True
+                    return True
+        else:
+            # Handle single connection case
+            addrs = connections.get_transport_addresses()
+            if any(not str(addr).startswith("/p2p-circuit") for addr in addrs):
+                self._peer_reachability[peer_id] = True
+                return True
+
+        # Get the peer's addresses from peerstore
+        try:
+            addrs = self.host.get_peerstore().addrs(peer_id)
+            # Check if peer has any public addresses
+            public_addrs = self.get_public_addrs(addrs)
+            if public_addrs:
+                self._peer_reachability[peer_id] = True
+                return True
+        except Exception as e:
+            logger.debug("Error getting peer addresses: %s", str(e))
+
+        # Default to not directly reachable
+        self._peer_reachability[peer_id] = False
+        return False
+
+    async def check_self_reachability(self) -> tuple[bool, list[Multiaddr]]:
+        """
+        Check if this host is likely directly reachable.
+
+        Returns
+        -------
+        Tuple[bool, List[Multiaddr]]
+            Tuple of (is_reachable, public_addresses)
+
+        """
+        # Get all host addresses
+        addrs = self.host.get_addrs()
+
+        # Filter for public addresses
+        public_addrs = self.get_public_addrs(addrs)
+
+        # If we have public addresses, assume we're reachable
+        # This is a simplified assumption - real reachability would need
+        # external checking
+        is_reachable = len(public_addrs) > 0
+
+        return is_reachable, public_addrs