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

Update examples to use dynamic host IP instead of hardcoded localhost

Browse Source
This commit is contained in:
yashksaini-coder
2025-09-18 02:09:51 +05:30
parent b01f2bd105
commit 4dd2454a46
15 changed files with 114 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
examples/advanced/network_discover.py
View File

@ -17,9 +17,22 @@ try:
get_wildcard_address,
)
except ImportError:
# Fallbacks if utilities are missing
# Fallbacks if utilities are missing - use minimal network discovery
import socket
def get_available_interfaces(port: int, protocol: str = "tcp"):
return [Multiaddr(f"/ip4/127.0.0.1/{protocol}/{port}")]
# Try to get local network interfaces, fallback to loopback
addrs = []
try:
# Get hostname IP (better than hardcoded localhost)
hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
if local_ip != "127.0.0.1":
addrs.append(Multiaddr(f"/ip4/{local_ip}/{protocol}/{port}"))
except exception:
pass
# Always include loopback as fallback
addrs.append(Multiaddr(f"/ip4/127.0.0.1/{protocol}/{port}"))
return addrs
def expand_wildcard_address(addr: Multiaddr, port: int | None = None):
if port is None:
@ -28,6 +41,12 @@ except ImportError:
return [Multiaddr(addr_str + f"/{port}")]
def get_optimal_binding_address(port: int, protocol: str = "tcp"):
# Try to get a non-loopback address first
interfaces = get_available_interfaces(port, protocol)
for addr in interfaces:
if "127.0.0.1" not in str(addr):
return addr
# Fallback to loopback if no other interfaces found
return Multiaddr(f"/ip4/127.0.0.1/{protocol}/{port}")
def get_wildcard_address(port: int, protocol: str = "tcp"):

2
examples/bootstrap/bootstrap.py
View File

@ -120,7 +120,7 @@ def main() -> None:
Usage:
python bootstrap.py -p 8000
python bootstrap.py -p 8001 --custom-bootstrap \\
"/ip4/127.0.0.1/tcp/8000/p2p/QmYourPeerID"
"/ip4/[HOST_IP]/tcp/8000/p2p/QmYourPeerID"
"""
parser = argparse.ArgumentParser(

2
examples/chat/chat.py
View File

@ -110,7 +110,7 @@ def main() -> None:
where <DESTINATION> is the multiaddress of the previous listener host.
"""
example_maddr = (
"/ip4/127.0.0.1/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
"/ip4/[HOST_IP]/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
)
parser = argparse.ArgumentParser(description=description)
parser.add_argument("-p", "--port", default=0, type=int, help="source port number")

18
examples/doc-examples/example_encryption_insecure.py
View File

@ -1,6 +1,5 @@
import secrets
import multiaddr
import trio
from libp2p import (
@ -9,9 +8,10 @@ from libp2p import (
from libp2p.crypto.secp256k1 import (
create_new_key_pair,
)
from libp2p.security.insecure.transport import (
PLAINTEXT_PROTOCOL_ID,
InsecureTransport,
from libp2p.security.insecure.transport import PLAINTEXT_PROTOCOL_ID, Transport
from libp2p.utils.address_validation import (
get_available_interfaces,
get_optimal_binding_address,
)
@ -21,7 +21,7 @@ async def main():
key_pair = create_new_key_pair(secret)
# Create an insecure transport (not recommended for production)
insecure_transport = InsecureTransport(
insecure_transport = Transport(
# local_key_pair: The key pair used for libp2p identity
local_key_pair=key_pair,
# secure_bytes_provider: Optional function to generate secure random bytes
@ -38,17 +38,19 @@ async def main():
# Create a host with the key pair and insecure transport
host = new_host(key_pair=key_pair, sec_opt=security_options)
# Configure the listening address
# Configure the listening address using the new paradigm
port = 8000
listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}")
listen_addrs = get_available_interfaces(port)
optimal_addr = get_optimal_binding_address(port)
# Start the host
async with host.run(listen_addrs=[listen_addr]):
async with host.run(listen_addrs=listen_addrs):
print(
"libp2p has started with insecure transport "
"(not recommended for production)"
)
print("libp2p is listening on:", host.get_addrs())
print(f"Optimal address: {optimal_addr}")
# Keep the host running
await trio.sleep_forever()

13
examples/doc-examples/example_encryption_noise.py
View File

@ -1,6 +1,5 @@
import secrets
import multiaddr
import trio
from libp2p import (
@ -13,6 +12,10 @@ from libp2p.security.noise.transport import (
PROTOCOL_ID as NOISE_PROTOCOL_ID,
Transport as NoiseTransport,
)
from libp2p.utils.address_validation import (
get_available_interfaces,
get_optimal_binding_address,
)
async def main():
@ -39,14 +42,16 @@ async def main():
# Create a host with the key pair and Noise security
host = new_host(key_pair=key_pair, sec_opt=security_options)
# Configure the listening address
# Configure the listening address using the new paradigm
port = 8000
listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}")
listen_addrs = get_available_interfaces(port)
optimal_addr = get_optimal_binding_address(port)
# Start the host
async with host.run(listen_addrs=[listen_addr]):
async with host.run(listen_addrs=listen_addrs):
print("libp2p has started with Noise encryption")
print("libp2p is listening on:", host.get_addrs())
print(f"Optimal address: {optimal_addr}")
# Keep the host running
await trio.sleep_forever()

13
examples/doc-examples/example_encryption_secio.py
View File

@ -1,6 +1,5 @@
import secrets
import multiaddr
import trio
from libp2p import (
@ -13,6 +12,10 @@ from libp2p.security.secio.transport import (
ID as SECIO_PROTOCOL_ID,
Transport as SecioTransport,
)
from libp2p.utils.address_validation import (
get_available_interfaces,
get_optimal_binding_address,
)
async def main():
@ -32,14 +35,16 @@ async def main():
# Create a host with the key pair and SECIO security
host = new_host(key_pair=key_pair, sec_opt=security_options)
# Configure the listening address
# Configure the listening address using the new paradigm
port = 8000
listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}")
listen_addrs = get_available_interfaces(port)
optimal_addr = get_optimal_binding_address(port)
# Start the host
async with host.run(listen_addrs=[listen_addr]):
async with host.run(listen_addrs=listen_addrs):
print("libp2p has started with SECIO encryption")
print("libp2p is listening on:", host.get_addrs())
print(f"Optimal address: {optimal_addr}")
# Keep the host running
await trio.sleep_forever()

2
examples/doc-examples/example_net_stream.py
View File

@ -234,7 +234,7 @@ async def run_enhanced_demo(
def main() -> None:
example_maddr = (
"/ip4/127.0.0.1/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
"/ip4/[HOST_IP]/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
)
parser = argparse.ArgumentParser(

13
examples/doc-examples/example_peer_discovery.py
View File

@ -1,6 +1,5 @@
import secrets
import multiaddr
import trio
from libp2p import (
@ -16,6 +15,10 @@ from libp2p.security.noise.transport import (
PROTOCOL_ID as NOISE_PROTOCOL_ID,
Transport as NoiseTransport,
)
from libp2p.utils.address_validation import (
get_available_interfaces,
get_optimal_binding_address,
)
async def main():
@ -42,14 +45,16 @@ async def main():
# Create a host with the key pair, Noise security, and mplex multiplexer
host = new_host(key_pair=key_pair, sec_opt=security_options)
# Configure the listening address
# Configure the listening address using the new paradigm
port = 8000
listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}")
listen_addrs = get_available_interfaces(port)
optimal_addr = get_optimal_binding_address(port)
# Start the host
async with host.run(listen_addrs=[listen_addr]):
async with host.run(listen_addrs=listen_addrs):
print("libp2p has started")
print("libp2p is listening on:", host.get_addrs())
print(f"Optimal address: {optimal_addr}")
# Connect to bootstrap peers manually
bootstrap_list = [

21
examples/doc-examples/example_quic_transport.py
View File

@ -1,6 +1,5 @@
import secrets
import multiaddr
import trio
from libp2p import (
@ -9,6 +8,10 @@ from libp2p import (
from libp2p.crypto.secp256k1 import (
create_new_key_pair,
)
from libp2p.utils.address_validation import (
get_available_interfaces,
get_optimal_binding_address,
)
async def main():
@ -19,14 +22,24 @@ async def main():
# Create a host with the key pair
host = new_host(key_pair=key_pair, enable_quic=True)
# Configure the listening address
# Configure the listening address using the new paradigm
port = 8000
listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/udp/{port}/quic-v1")
listen_addrs = get_available_interfaces(port, protocol="udp")
# Convert TCP addresses to QUIC-v1 addresses
quic_addrs = []
for addr in listen_addrs:
addr_str = str(addr).replace("/tcp/", "/udp/") + "/quic-v1"
from multiaddr import Multiaddr
quic_addrs.append(Multiaddr(addr_str))
optimal_addr = get_optimal_binding_address(port, protocol="udp")
optimal_quic_str = str(optimal_addr).replace("/tcp/", "/udp/") + "/quic-v1"
# Start the host
async with host.run(listen_addrs=[listen_addr]):
async with host.run(listen_addrs=quic_addrs):
print("libp2p has started with QUIC transport")
print("libp2p is listening on:", host.get_addrs())
print(f"Optimal address: {optimal_quic_str}")
# Keep the host running
await trio.sleep_forever()

2
examples/echo/echo.py
View File

@ -125,7 +125,7 @@ def main() -> None:
where <DESTINATION> is the multiaddress of the previous listener host.
"""
example_maddr = (
"/ip4/127.0.0.1/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
"/ip4/[HOST_IP]/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
)
parser = argparse.ArgumentParser(description=description)
parser.add_argument("-p", "--port", default=0, type=int, help="source port number")

27
examples/echo/echo_quic.py
View File

@ -43,13 +43,22 @@ async def _echo_stream_handler(stream: INetStream) -> None:
async def run_server(port: int, seed: int | None = None) -> None:
"""Run echo server with QUIC transport."""
from libp2p.utils.address_validation import find_free_port
from libp2p.utils.address_validation import (
find_free_port,
get_available_interfaces,
get_optimal_binding_address,
)
if port <= 0:
port = find_free_port()
# For QUIC, we need to use UDP addresses - use loopback for security
listen_addr = Multiaddr(f"/ip4/127.0.0.1/udp/{port}/quic")
# For QUIC, we need UDP addresses - use the new address paradigm
tcp_addrs = get_available_interfaces(port)
# Convert TCP addresses to QUIC addresses
quic_addrs = []
for addr in tcp_addrs:
addr_str = str(addr).replace("/tcp/", "/udp/") + "/quic"
quic_addrs.append(Multiaddr(addr_str))
if seed:
import random
@ -69,7 +78,7 @@ async def run_server(port: int, seed: int | None = None) -> None:
)
# Server mode: start listener
async with host.run(listen_addrs=[listen_addr]):
async with host.run(listen_addrs=quic_addrs):
try:
print(f"I am {host.get_id().to_string()}")
host.set_stream_handler(PROTOCOL_ID, _echo_stream_handler)
@ -81,11 +90,13 @@ async def run_server(port: int, seed: int | None = None) -> None:
for addr in all_addrs:
print(f"{addr}")
# Use the first address as the default for the client command
default_addr = all_addrs[0]
# Use optimal address for the client command
optimal_tcp = get_optimal_binding_address(port)
optimal_quic_str = str(optimal_tcp).replace("/tcp/", "/udp/") + "/quic"
optimal_quic_with_peer = f"{optimal_quic_str}/p2p/{host.get_id().to_string()}"
print(
f"\nRun this from the same folder in another console:\n\n"
f"python3 ./examples/echo/echo_quic.py -d {default_addr}\n"
f"python3 ./examples/echo/echo_quic.py -d {optimal_quic_with_peer}\n"
)
print("Waiting for incoming QUIC connections...")
await trio.sleep_forever()
@ -167,7 +178,7 @@ def main() -> None:
where <DESTINATION> is the QUIC multiaddress of the previous listener host.
"""
example_maddr = "/ip4/127.0.0.1/udp/8000/quic/p2p/QmQn4SwGkDZKkUEpBRBv"
example_maddr = "/ip4/[HOST_IP]/udp/8000/quic/p2p/QmQn4SwGkDZKkUEpBRBv"
parser = argparse.ArgumentParser(description=description)
parser.add_argument("-p", "--port", default=0, type=int, help="UDP port number")

2
examples/identify/identify.py
View File

@ -262,7 +262,7 @@ def main() -> None:
"""
example_maddr = (
"/ip4/127.0.0.1/tcp/8888/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
"/ip4/[HOST_IP]/tcp/8888/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
)
parser = argparse.ArgumentParser(description=description)

13
examples/identify_push/identify_push_demo.py
View File

@ -36,6 +36,9 @@ from libp2p.identity.identify_push import (
from libp2p.peer.peerinfo import (
info_from_p2p_addr,
)
from libp2p.utils.address_validation import (
get_available_interfaces,
)
# Configure logging
logger = logging.getLogger(__name__)
@ -207,13 +210,13 @@ async def main() -> None:
ID_PUSH, create_custom_identify_push_handler(host_2, "Host 2")
)
# Start listening on random ports using the run context manager
listen_addr_1 = multiaddr.Multiaddr("/ip4/127.0.0.1/tcp/0")
listen_addr_2 = multiaddr.Multiaddr("/ip4/127.0.0.1/tcp/0")
# Start listening on available interfaces using random ports
listen_addrs_1 = get_available_interfaces(0) # 0 for random port
listen_addrs_2 = get_available_interfaces(0) # 0 for random port
async with (
host_1.run([listen_addr_1]),
host_2.run([listen_addr_2]),
host_1.run(listen_addrs_1),
host_2.run(listen_addrs_2),
trio.open_nursery() as nursery,
):
# Start the peer-store cleanup task

10
examples/identify_push/identify_push_listener_dialer.py
View File

@ -14,7 +14,7 @@ Usage:
python identify_push_listener_dialer.py
# Then in another console, run as a dialer (default port 8889):
python identify_push_listener_dialer.py -d /ip4/127.0.0.1/tcp/8888/p2p/PEER_ID
python identify_push_listener_dialer.py -d /ip4/[HOST_IP]/tcp/8888/p2p/PEER_ID
(where PEER_ID is the peer ID displayed by the listener)
"""
@ -291,10 +291,12 @@ async def run_dialer(
identify_push_handler_for(host, use_varint_format=use_varint_format),
)
# Start listening on a different port
listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}")
# Start listening on available interfaces
from libp2p.utils.address_validation import get_available_interfaces
async with host.run([listen_addr]):
listen_addrs = get_available_interfaces(port)
async with host.run(listen_addrs):
logger.info("Dialer host ready!")
print("Dialer host ready!")

2
examples/ping/ping.py
View File

@ -118,7 +118,7 @@ def main() -> None:
"""
example_maddr = (
"/ip4/127.0.0.1/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
"/ip4/[HOST_IP]/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q"
)
parser = argparse.ArgumentParser(description=description)