varun-r-mallya/py-libp2p
1
0
Fork 0
mirror of https://github.com/varun-r-mallya/py-libp2p.git synced 2025-12-31 20:36:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'acul71/feat/804-add-thin-waist-address' into feat/804-add-thin-waist-address

Browse Source
This commit is contained in:
yashksaini-coder
2025-08-19 19:56:20 +05:30
parent 9378490dcb 05b372b1eb
commit 7f6469d5d4
7 changed files with 311 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
libp2p/utils/address_validation.py
View File

@ -1,9 +1,13 @@
from __future__ import annotations
from typing import List, Optional
from multiaddr import Multiaddr
try:
from multiaddr.utils import get_thin_waist_addresses, get_network_addrs # type: ignore
from multiaddr.utils import ( # type: ignore
get_network_addrs,
get_thin_waist_addresses,
)
_HAS_THIN_WAIST = True
except ImportError: # pragma: no cover - only executed in older environments
_HAS_THIN_WAIST = False
@ -11,7 +15,7 @@ except ImportError: # pragma: no cover - only executed in older environments
get_network_addrs = None # type: ignore
def _safe_get_network_addrs(ip_version: int) -> List[str]:
def _safe_get_network_addrs(ip_version: int) -> list[str]:
"""
Internal safe wrapper. Returns a list of IP addresses for the requested IP version.
Falls back to minimal defaults when Thin Waist helpers are missing.
@ -31,7 +35,7 @@ def _safe_get_network_addrs(ip_version: int) -> List[str]:
return []
def _safe_expand(addr: Multiaddr, port: Optional[int] = None) -> List[Multiaddr]:
def _safe_expand(addr: Multiaddr, port: int | None = None) -> list[Multiaddr]:
"""
Internal safe expansion wrapper. Returns a list of Multiaddr objects.
If Thin Waist isn't available, returns [addr] (identity).
@ -46,7 +50,7 @@ def _safe_expand(addr: Multiaddr, port: Optional[int] = None) -> List[Multiaddr]
return [addr]
def get_available_interfaces(port: int, protocol: str = "tcp") -> List[Multiaddr]:
def get_available_interfaces(port: int, protocol: str = "tcp") -> list[Multiaddr]:
"""
Discover available network interfaces (IPv4 + IPv6 if supported) for binding.
@ -54,7 +58,7 @@ def get_available_interfaces(port: int, protocol: str = "tcp") -> List[Multiaddr
:param protocol: Transport protocol (e.g., "tcp" or "udp").
:return: List of Multiaddr objects representing candidate interface addresses.
"""
addrs: List[Multiaddr] = []
addrs: list[Multiaddr] = []
# IPv4 enumeration
seen_v4: set[str] = set()
@ -62,14 +66,11 @@ def get_available_interfaces(port: int, protocol: str = "tcp") -> List[Multiaddr
seen_v4.add(ip)
addrs.append(Multiaddr(f"/ip4/{ip}/{protocol}/{port}"))
# Ensure loopback IPv4 explicitly present (JS echo parity) even if not returned
if "127.0.0.1" not in seen_v4:
addrs.append(Multiaddr(f"/ip4/127.0.0.1/{protocol}/{port}"))
# IPv6 enumeration (optional: only include if we have at least one global or loopback)
seen_v6: set[str] = set()
# IPv6 enumeration (optional: only include if we have at least one global or
# loopback)
for ip in _safe_get_network_addrs(6):
seen_v6.add(ip)
# Avoid returning unusable wildcard expansions if the environment does not
# support IPv6
addrs.append(Multiaddr(f"/ip6/{ip}/{protocol}/{port}"))
# Optionally ensure IPv6 loopback when any IPv6 present but loopback missing
if seen_v6 and "::1" not in seen_v6:
@ -82,7 +83,9 @@ def get_available_interfaces(port: int, protocol: str = "tcp") -> List[Multiaddr
return addrs
def expand_wildcard_address(addr: Multiaddr, port: Optional[int] = None) -> List[Multiaddr]:
def expand_wildcard_address(
addr: Multiaddr, port: int | None = None
) -> list[Multiaddr]:
"""
Expand a wildcard (e.g. /ip4/0.0.0.0/tcp/0) into all concrete interfaces.
@ -132,4 +135,4 @@ __all__ = [
"get_available_interfaces",
"get_optimal_binding_address",
"expand_wildcard_address",
]
]