Examples¶
This section provides practical examples of how to use the multiaddr library for various use cases.
DNS Resolution Examples¶
The examples/dns/ directory contains comprehensive examples demonstrating DNS-based address resolution.
This example shows: - Standard DNS resolution for both IPv4 and IPv6 - Protocol-specific DNS resolution (dns4/dns6) - Peer ID preservation during resolution - Bootstrap node resolution using real libp2p nodes - Error handling and timeout management
examples/dns/dns_examples.py¶
#!/usr/bin/env python3
"""
DNS resolution examples for py-multiaddr.
This script demonstrates how to use DNS resolution functionality in py-multiaddr,
showing how to resolve bootstrap node addresses like those used in js-libp2p.
## Overview
This script shows various examples of DNS resolution using py-multiaddr:
1. **Basic DNS Resolution**: Simple resolution of DNS addresses to IP addresses
2. **Bootstrap Node Resolution**: Resolving real bootstrap node addresses with peer IDs
3. **DNS Protocol Comparison**: Testing different DNS protocols (/dns/, /dns4/, /dns6/, /dnsaddr/)
4. **Peer ID Preservation**: Ensuring peer IDs are maintained during resolution
5. **Sequential Resolution**: Processing multiple addresses sequentially
6. **py-libp2p Integration**: Example of how to use resolved addresses with py-libp2p
## Expected Output
When you run this script, you should see output similar to:
```
DNS Resolution Examples
==================================================
=== Basic DNS Resolution ===
Original address: /dns/example.com
Protocols: ['dns']
Resolved to 12 addresses:
1. /ip4/23.192.228.84
2. /ip4/23.215.0.136
... (more IPv4 and IPv6 addresses)
=== Bootstrap Node Resolution ===
Resolving: /dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Peer ID: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Resolved to 6 addresses:
1. /ip4/139.178.91.71/tcp/4001/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
2. /ip4/139.178.91.71/udp/4001/quic-v1/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
... (more addresses)
=== DNS Protocol Comparison ===
Testing DNS (both IPv4 and IPv6): /dns/example.com
Resolved to 12 addresses:
1. /ip4/23.215.0.136
2. /ip4/23.215.0.138
... (6 IPv4 + 6 IPv6 addresses)
=== Peer ID Preservation Test ===
Original address: /dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Original peer ID: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Resolved to 6 addresses:
1. /ip4/139.178.91.71/tcp/4001/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Peer ID: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Peer ID preserved: True
=== Sequential DNS Resolution ===
/dns/example.com:
Resolved to 12 addresses:
- /ip4/23.215.0.136
- /ip4/23.215.0.138
... (more addresses)
=== py-libp2p Integration Example ===
Processing bootstrap node: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Resolved: 139.178.91.71:4001 (peer: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN)
Resolved 1 bootstrap peers:
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN -> 139.178.91.71:4001
==================================================
All examples completed!
```
## Key Features Demonstrated
- **DNS Resolution**: Converting DNS addresses to IP addresses
- **Peer ID Preservation**: Maintaining peer IDs during resolution
- **Multiple Protocol Support**: /dns/, /dns4/, /dns6/, /dnsaddr/
- **Error Handling**: Graceful handling of resolution failures
- **Real Bootstrap Node Examples**: Using actual libp2p bootstrap nodes that resolve to IP addresses
## Requirements
- Python 3.10+
- py-multiaddr library
- trio library
- Internet connection for DNS resolution
## Usage
```bash
python trio_dns_examples.py
```
## Notes
- This script uses real bootstrap nodes from bootstrap.libp2p.io that actually resolve
- The DNS resolver requires trio for async operations
- Peer IDs are preserved during resolution for bootstrap node functionality
- All examples demonstrate working DNS resolution to actual IP addresses
"""
import trio
from multiaddr import Multiaddr
from multiaddr.resolvers import DNSResolver
async def basic_dns_resolution():
"""
Basic DNS resolution example.
This function demonstrates the simplest form of DNS resolution:
- Creates a DNS multiaddr for example.com
- Resolves it to actual IP addresses (both IPv4 and IPv6)
- Shows the original and resolved addresses
Expected output:
=== Basic DNS Resolution ===
Original address: /dns/example.com
Protocols: ['dns']
Resolved to 12 addresses:
1. /ip4/23.192.228.84
2. /ip4/23.215.0.136
3. /ip4/23.215.0.138
... (more IPv4 and IPv6 addresses)
"""
print("=== Basic DNS Resolution ===")
# Test with a simple DNS address
test_addr = "/dns/example.com"
ma = Multiaddr(test_addr)
print(f"Original address: {ma}")
print(f"Protocols: {[p.name for p in ma.protocols()]}")
try:
resolved = await ma.resolve()
print(f"Resolved to {len(resolved)} addresses:")
for i, addr in enumerate(resolved, 1):
print(f" {i}. {addr}")
except Exception as e:
print(f"Error resolving {test_addr}: {e}")
async def bootstrap_node_resolution():
"""
Resolve bootstrap node addresses.
This function demonstrates resolving real bootstrap node addresses that include peer IDs:
- Uses real domains from bootstrap.libp2p.io that actually resolve
- Shows how peer IDs are preserved during resolution
- Extracts connection information (IP addresses) from resolved addresses
Expected output:
=== Bootstrap Node Resolution ===
Resolving: /dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Peer ID: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Resolved to 6 addresses:
1. /ip4/139.178.91.71/tcp/4001/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
2. /ip4/139.178.91.71/udp/4001/quic-v1/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
3. /ip6/2604:1380:45e3:6e00::1/tcp/4001/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
4. /ip6/2604:1380:45e3:6e00::1/udp/4001/quic-v1/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
5. /ip4/139.178.91.71/tcp/443/wss/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
6. /ip6/2604:1380:45e3:6e00::1/tcp/443/wss/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
"""
print("\n=== Bootstrap Node Resolution ===")
# Use only one bootstrap address to reduce repetition
bootstrap_addresses = [
"/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
]
resolver = DNSResolver()
for addr_str in bootstrap_addresses:
print(f"\nResolving: {addr_str}")
try:
ma = Multiaddr(addr_str)
peer_id = ma.get_peer_id()
print(f" Peer ID: {peer_id}")
# Resolve the address
resolved = await resolver.resolve(ma)
print(f" Resolved to {len(resolved)} addresses:")
for i, resolved_ma in enumerate(resolved, 1):
print(f" {i}. {resolved_ma}")
# Extract IP and port information for TCP connections only
ip_addr = None
port = None
for proto, value in resolved_ma.items():
if proto.name in ("ip4", "ip6"):
ip_addr = value
elif proto.name == "tcp":
port = value
if ip_addr and port:
print(f" Connection: {ip_addr}:{port}")
except Exception as e:
print(f" Error: {e}")
async def dns_protocol_comparison():
"""
Compare different DNS protocols.
This function demonstrates the differences between various DNS protocols:
- /dns/ - Resolves to both IPv4 and IPv6 addresses
- /dns4/ - Resolves to IPv4 addresses only
- /dns6/ - Resolves to IPv6 addresses only
- /dnsaddr/ - Resolves to both IPv4 and IPv6 addresses (same as /dns/)
Expected output:
=== DNS Protocol Comparison ===
Testing DNS (both IPv4 and IPv6): /dns/example.com
Resolved to 12 addresses:
1. /ip4/23.215.0.136
2. /ip4/23.215.0.138
... (6 IPv4 + 6 IPv6 addresses)
Testing DNS4 (IPv4 only): /dns4/example.com
Resolved to 6 addresses:
1. /ip4/23.215.0.138
2. /ip4/96.7.128.175
... (6 IPv4 addresses only)
Testing DNS6 (IPv6 only): /dns6/example.com
Resolved to 6 addresses:
1. /ip6/2600:1408:ec00:36::1736:7f31
2. /ip6/2600:1406:3a00:21::173e:2e65
... (6 IPv6 addresses only)
"""
print("\n=== DNS Protocol Comparison ===")
dns_tests = [
("/dns/example.com", "DNS (both IPv4 and IPv6)"),
("/dns4/example.com", "DNS4 (IPv4 only)"),
("/dns6/example.com", "DNS6 (IPv6 only)"),
("/dnsaddr/bootstrap.libp2p.io", "DNSADDR (both IPv4 and IPv6)"),
]
resolver = DNSResolver()
for addr_str, description in dns_tests:
print(f"\nTesting {description}: {addr_str}")
try:
ma = Multiaddr(addr_str)
resolved = await resolver.resolve(ma)
print(f" Resolved to {len(resolved)} addresses:")
# Show only first 3 addresses to reduce repetition
for i, addr in enumerate(resolved[:3], 1):
print(f" {i}. {addr}")
if len(resolved) > 3:
print(f" ... and {len(resolved) - 3} more addresses")
except Exception as e:
print(f" Error: {e}")
async def peer_id_preservation_test():
"""
Test that peer IDs are preserved during resolution.
This function verifies that peer IDs are maintained when resolving DNS addresses:
- Creates a multiaddr with both DNS and peer ID components
- Resolves the DNS part to IP addresses
- Confirms the peer ID remains unchanged in the resolved addresses
Expected output:
=== Peer ID Preservation Test ===
Original address: /dnsaddr/bootstrap.libp2p.io/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Original peer ID: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Resolved to 6 addresses:
1. /ip4/139.178.91.71/tcp/4001/p2p/
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Peer ID: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Peer ID preserved: True
"""
print("\n=== Peer ID Preservation Test ===")
test_addr = "/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN"
try:
ma = Multiaddr(test_addr)
original_peer_id = ma.get_peer_id()
print(f"Original address: {ma}")
print(f"Original peer ID: {original_peer_id}")
resolved = await ma.resolve()
print(f"Resolved to {len(resolved)} addresses:")
# Show only first 2 addresses to reduce repetition
for i, resolved_ma in enumerate(resolved[:2], 1):
resolved_peer_id = resolved_ma.get_peer_id()
print(f" {i}. {resolved_ma}")
print(f" Peer ID: {resolved_peer_id}")
print(f" Peer ID preserved: {original_peer_id == resolved_peer_id}")
if len(resolved) > 2:
print(f" ... and {len(resolved) - 2} more addresses (all with preserved peer ID)")
except Exception as e:
print(f"Error: {e}")
async def concurrent_resolution():
"""
Demonstrate sequential DNS resolution.
This function shows how to process multiple DNS addresses sequentially:
- Resolves multiple addresses one by one
- Handles errors gracefully for each address
- Shows the results for each resolution attempt
Expected output:
=== Sequential DNS Resolution ===
/dns/example.com:
Resolved to 12 addresses:
- /ip4/23.215.0.136
- /ip4/23.215.0.138
... (more addresses)
/dns4/example.com:
Resolved to 6 addresses:
- /ip4/23.192.228.84
- /ip4/23.215.0.136
... (IPv4 addresses only)
/dns6/example.com:
Resolved to 6 addresses:
- /ip6/2600:1408:ec00:36::1736:7f31
- /ip6/2600:1406:3a00:21::173e:2e65
... (IPv6 addresses only)
"""
print("\n=== Sequential DNS Resolution ===")
addresses = [
"/dns/example.com",
"/dns4/example.com",
"/dns6/example.com",
"/dnsaddr/bootstrap.libp2p.io",
]
async def resolve_single(addr_str):
"""Resolve a single address."""
try:
ma = Multiaddr(addr_str)
resolved = await ma.resolve()
return addr_str, resolved, None
except Exception as e:
return addr_str, [], str(e)
# Resolve all addresses sequentially
results = []
for addr in addresses:
result = await resolve_single(addr)
results.append(result)
for addr_str, resolved, error in results:
print(f"\n{addr_str}:")
if error:
print(f" Error: {error}")
else:
print(f" Resolved to {len(resolved)} addresses:")
# Show only first 3 addresses to reduce repetition
for addr in resolved[:3]:
print(f" - {addr}")
if len(resolved) > 3:
print(f" ... and {len(resolved) - 3} more addresses")
async def py_libp2p_integration_example():
"""
Example of how to integrate with py-libp2p.
This function demonstrates a practical integration pattern for py-libp2p:
- Resolves real bootstrap node addresses to IP addresses
- Extracts connection information (peer ID, IP, port)
- Prepares the data structure needed for py-libp2p connections
- Shows how to handle the resolved peer information
Expected output:
=== py-libp2p Integration Example ===
Processing bootstrap node: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
Resolved: 139.178.91.71:4001 (peer: QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN)
Resolved 1 bootstrap peers:
QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN -> 139.178.91.71:4001
"""
print("\n=== py-libp2p Integration Example ===")
resolver = DNSResolver()
# Use only one bootstrap address to reduce repetition
bootstrap_addresses = [
"/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
]
resolved_peers = []
for addr_str in bootstrap_addresses:
ma = Multiaddr(addr_str)
peer_id = ma.get_peer_id()
print(f"\nProcessing bootstrap node: {peer_id}")
try:
# Resolve DNS addresses to IP addresses
resolved_addrs = await resolver.resolve(ma)
for resolved_ma in resolved_addrs:
# Extract connection information
ip_addr = None
port = None
for proto, value in resolved_ma.items():
if proto.name == "ip4":
ip_addr = value
elif proto.name == "ip6":
ip_addr = value
elif proto.name == "tcp":
port = value
if ip_addr and port and peer_id:
peer_info = {
"peer_id": peer_id,
"ip_addr": ip_addr,
"port": port,
"original_addr": str(ma),
"resolved_addr": str(resolved_ma),
}
resolved_peers.append(peer_info)
print(f" Resolved: {ip_addr}:{port} (peer: {peer_id})")
except Exception as e:
print(f" Error resolving {addr_str}: {e}")
print(f"\nResolved {len(resolved_peers)} bootstrap peers:")
for peer in resolved_peers:
print(f" {peer['peer_id']} -> {peer['ip_addr']}:{peer['port']}")
return resolved_peers
async def main():
"""
Run all examples.
This function orchestrates all the DNS resolution examples:
1. Basic DNS resolution
2. Bootstrap node resolution
3. DNS protocol comparison
4. Peer ID preservation test
5. Sequential resolution
6. py-libp2p integration example
Each example demonstrates different aspects of DNS resolution functionality
and shows how to use it with py-multiaddr.
"""
print("DNS Resolution Examples")
print("=" * 50)
try:
await basic_dns_resolution()
await bootstrap_node_resolution()
await dns_protocol_comparison()
await peer_id_preservation_test()
await concurrent_resolution()
await py_libp2p_integration_example()
print("\n" + "=" * 50)
print("All examples completed!")
print("\nSummary:")
print("- DNS resolution is working correctly")
print("- Real domains are being resolved to IP addresses")
print("- Peer IDs are preserved during resolution")
print("- All DNS protocols (/dns/, /dns4/, /dns6/, /dnsaddr/) are functional")
print("- Ready for integration with py-libp2p")
except KeyboardInterrupt:
print("\nExamples interrupted by user")
except Exception as e:
print(f"\nUnexpected error: {e}")
if __name__ == "__main__":
trio.run(main)
DNSADDR Examples¶
The examples/dnsaddr/ directory shows how to work with DNSADDR records for libp2p bootstrap nodes.
This example demonstrates: - DNSADDR record parsing and resolution - Peer ID extraction and preservation - Bootstrap node discovery - TXT record processing
examples/dnsaddr/dnsaddr.py¶
import trio
from multiaddr import Multiaddr
from multiaddr.resolvers import DNSResolver
# Example DNSADDR resolution script
#
# Expected output:
# - Successful resolutions show resolved multiaddrs (e.g., /ip4/139.178.91.71/tcp/4001/p2p/...)
# - Failed resolutions show "(No resolution results)" (empty list returned)
# - This matches the JS implementation and libp2p spec behavior
#
# Sample output:
# Resolving: /dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
# -> /ip4/139.178.91.71/tcp/4001/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
# -> /ip4/139.178.91.71/udp/4001/quic-v1/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
# ...
# Resolving: /dnsaddr/github.com/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN
# (No resolution results)
ADDRESSES = [
"/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
"/dnsaddr/bootstrap.libp2p.io/p2p/QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
"/dnsaddr/bootstrap.libp2p.io/p2p/QmZa1sAxajnQjVM8WjWXoMbmPd7NsWhfKsPkErzpm9wGkp",
"/dnsaddr/bootstrap.libp2p.io/p2p/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
"/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
"/dnsaddr/github.com/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
"/dnsaddr/cloudflare.com/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
"/dnsaddr/google.com/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
]
async def main():
resolver = DNSResolver()
for addr in ADDRESSES:
print(f"\nResolving: {addr}")
ma = Multiaddr(addr)
try:
resolved = await resolver.resolve(ma)
if resolved:
for r in resolved:
print(f" -> {r}")
else:
# If DNSADDR resolution fails (no TXT records or no matching entries),
# the result is an empty list (no resolution results), matching the JS
# implementation and spec.
print(" (No resolution results)")
except Exception as e:
print(f" (Error: {e})")
if __name__ == "__main__":
trio.run(main)
QUIC Protocol Examples¶
The examples/quic/ directory demonstrates how to work with QUIC and QUIC-v1 protocols in multiaddr addresses.
This example shows: - Basic QUIC and QUIC-v1 protocol usage - QUIC with peer IDs for P2P networking - Protocol stack analysis and manipulation - Address encapsulation and decapsulation - Binary representation and validation - IPv4 and IPv6 support with QUIC - Validation of valid and invalid QUIC address combinations
examples/quic/simple_quic_usage.py¶
#!/usr/bin/env python3
"""
Simple QUIC Protocol Usage Examples
This shows the most common ways to use QUIC and QUIC-v1 protocols
in the py-multiaddr implementation.
"""
from multiaddr import Multiaddr
from multiaddr.protocols import P_QUIC1
def main():
print("Simple QUIC Protocol Usage Examples")
print("=" * 40)
# 1. Basic QUIC addresses
print("\n1. Basic QUIC Addresses:")
# QUIC over UDP
quic_addr = Multiaddr("/ip4/127.0.0.1/udp/4001/quic")
print(f" QUIC: {quic_addr}")
# QUIC-v1 over UDP (newer version)
quic_v1_addr = Multiaddr("/ip4/127.0.0.1/udp/4001/quic-v1")
print(f" QUIC-v1: {quic_v1_addr}")
# IPv6 with QUIC
quic_ipv6 = Multiaddr("/ip6/::1/udp/4001/quic-v1")
print(f" IPv6 QUIC: {quic_ipv6}")
# 2. QUIC with peer IDs (libp2p style)
print("\n2. QUIC with Peer IDs (libp2p):")
quic_with_peer = Multiaddr(
"/ip4/127.0.0.1/udp/4001/quic-v1/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN"
)
print(f" Address: {quic_with_peer}")
print(f" Peer ID: {quic_with_peer.get_peer_id()}")
# 3. Protocol analysis
print("\n3. Protocol Analysis:")
protocols = list(quic_v1_addr.protocols())
print(f" Protocol stack for {quic_v1_addr}:")
for i, proto in enumerate(protocols):
print(f" {i + 1}. {proto.name} (code: {proto.code})")
# 4. Address manipulation
print("\n4. Address Manipulation:")
# Encapsulation: add QUIC layer to existing address
base = Multiaddr("/ip4/192.168.1.100/udp/4001")
quic_layer = Multiaddr("/quic-v1")
combined = base.encapsulate(quic_layer)
print(f" Base: {base}")
print(f" + QUIC: {quic_layer}")
print(f" = Result: {combined}")
# Decapsulation: remove QUIC layer
without_quic = combined.decapsulate_code(P_QUIC1)
print(f" Without QUIC: {without_quic}")
# 5. Binary representation
print("\n5. Binary Representation:")
binary = quic_v1_addr.to_bytes()
print(f" String: {quic_v1_addr}")
print(f" Binary: {binary.hex()}")
print(f" Size: {len(binary)} bytes")
# 6. Validation examples
print("\n6. Validation Examples:")
valid_examples = [
"/ip4/127.0.0.1/udp/4001/quic",
"/ip4/127.0.0.1/udp/4001/quic-v1",
"/ip6/::1/udp/4001/quic-v1",
]
print(" Valid QUIC addresses:")
for addr_str in valid_examples:
try:
ma = Multiaddr(addr_str)
print(f" ✅ {ma}")
except Exception as e:
print(f" ❌ {addr_str} - {e}")
invalid_examples = [
"/ip4/127.0.0.1/tcp/4001/quic", # QUIC over TCP (invalid)
"/ip4/127.0.0.1/udp/4001/quic/1234", # QUIC with port (invalid)
]
print(" Invalid QUIC addresses:")
for addr_str in invalid_examples:
try:
ma = Multiaddr(addr_str)
print(f" ⚠️ {ma} (unexpectedly valid)")
except Exception as e:
print(f" ❌ {addr_str} - {e}")
print("\n" + "=" * 40)
print("Key Points:")
print("- QUIC protocols are flag protocols (no additional data)")
print("- Must be used with UDP transport (not TCP)")
print("- QUIC-v1 is the newer, recommended version")
print("- Common in libp2p networks for secure communication")
print("- Can be combined with peer IDs for P2P networking")
if __name__ == "__main__":
main()
Thin Waist Address Examples¶
The examples/thin_waist/ directory demonstrates thin waist address validation and network interface discovery.
This example shows: - Network interface discovery - Wildcard address expansion - IPv4 and IPv6 support - Port management - Server binding scenarios
examples/thin_waist/thin_waist_example.py¶
#!/usr/bin/env python3
"""
Thin Waist Address Validation Example
This example demonstrates how to use the get_thin_waist_addresses function
to process multiaddrs and expand wildcard addresses to all available interfaces.
Usage:
python examples/thin_waist/thin_waist_example.py [--detailed]
"""
import sys
from multiaddr import Multiaddr
from multiaddr.utils import get_network_addrs, get_thin_waist_addresses
def show_network_info():
"""Display available network interfaces."""
print("=== Network Interface Information ===")
# Get all IPv4 addresses
ipv4_addrs = get_network_addrs(4)
print(f"Available IPv4 addresses: {ipv4_addrs}")
# Get all IPv6 addresses
ipv6_addrs = get_network_addrs(6)
print(f"Available IPv6 addresses: {ipv6_addrs}")
print()
def basic_examples():
"""Show basic thin waist address validation examples."""
print("=== Basic Examples ===")
# Example 1: Specific address (no expansion)
print("\n1. Specific IP address:")
addr = Multiaddr("/ip4/192.168.1.100/tcp/8080")
print(f" Input: {addr}")
result = get_thin_waist_addresses(addr)
print(f" Output: {result}")
# Example 2: IPv4 wildcard expansion
print("\n2. IPv4 wildcard expansion:")
addr = Multiaddr("/ip4/0.0.0.0/tcp/8080")
print(f" Input: {addr}")
result = get_thin_waist_addresses(addr)
print(" Output:")
for i, expanded in enumerate(result, 1):
print(f" {i}. {expanded}")
# Example 3: IPv6 wildcard expansion
print("\n3. IPv6 wildcard expansion:")
addr = Multiaddr("/ip6/::/tcp/8080")
print(f" Input: {addr}")
result = get_thin_waist_addresses(addr)
print(" Output:")
for i, expanded in enumerate(result, 1):
print(f" {i}. {expanded}")
# Example 4: Port override
print("\n4. Port override:")
addr = Multiaddr("/ip4/0.0.0.0/tcp/8080")
print(f" Input: {addr} (override port to 9000)")
result = get_thin_waist_addresses(addr, port=9000)
print(" Output:")
for i, expanded in enumerate(result, 1):
print(f" {i}. {expanded}")
# Example 5: No input
print("\n5. No input:")
result = get_thin_waist_addresses()
print(f" Output: {result}")
def detailed_examples():
"""Show detailed examples with more edge cases and explanations."""
print("\n=== Detailed Examples ===")
# Example: UDP transport
print("\n6. UDP transport:")
addr = Multiaddr("/ip4/0.0.0.0/udp/1234")
print(f" Input: {addr}")
result = get_thin_waist_addresses(addr)
print(" Output:")
for i, expanded in enumerate(result, 1):
print(f" {i}. {expanded}")
# Example: Port override on specific address
print("\n7. Port override on specific address:")
addr = Multiaddr("/ip4/192.168.1.100/tcp/8080")
print(f" Input: {addr}")
result = get_thin_waist_addresses(addr, port=9000)
print(f" Output: {result}")
# Example: Error handling
print("\n8. Error handling:")
try:
# Try to create an invalid multiaddr
addr = Multiaddr("/ip4/192.168.1.100/udp/1234/webrtc")
print(f" Created address: {addr}")
# This should return empty list for non-thin-waist addresses
addrs = get_thin_waist_addresses(addr)
print(f" Thin waist addresses: {addrs}")
except Exception as e:
print(f" Error creating invalid address: {e}")
def usage_examples():
"""Show practical usage examples."""
print("\n=== Practical Usage Examples ===")
print("\n9. Server binding scenario:")
print(" When you want to bind a server to all interfaces:")
wildcard = Multiaddr("/ip4/0.0.0.0/tcp/8080")
interfaces = get_thin_waist_addresses(wildcard)
print(f" Wildcard: {wildcard}")
print(" Available interfaces:")
for i, interface in enumerate(interfaces, 1):
print(f" {i}. {interface}")
print("\n10. Port configuration scenario:")
print(" When you need to change the port dynamically:")
original = Multiaddr("/ip4/0.0.0.0/tcp/8080")
new_port = 9000
updated = get_thin_waist_addresses(original, port=new_port)
print(f" Original: {original}")
print(f" New port: {new_port}")
print(" Updated interfaces:")
for i, interface in enumerate(updated, 1):
print(f" {i}. {interface}")
def main():
"""Run the thin waist address validation examples."""
print("Thin Waist Address Validation Example")
print("=" * 50)
# Check for detailed mode
detailed = "--detailed" in sys.argv
# Show network information
show_network_info()
# Show basic examples
basic_examples()
# Show detailed examples if requested
if detailed:
detailed_examples()
usage_examples()
print("\n" + "=" * 50)
print("Example completed!")
if not detailed:
print("Run with --detailed flag for more examples and edge cases.")
if __name__ == "__main__":
main()
Decapsulate Code Examples¶
The examples/decapsulate/ directory demonstrates how to use the decapsulate_code method for protocol layer manipulation.
This example demonstrates: - Protocol code-based layer removal - Protocol stack analysis - Address transformation - Error handling for edge cases - Practical network configuration scenarios
examples/decapsulate/decapsulate_example.py¶
#!/usr/bin/env python3
"""
Decapsulate Code Examples
This example demonstrates how to use the decapsulate_code method to remove
specific protocol layers from multiaddrs by their protocol codes.
The decapsulate_code method is useful for:
- Network protocol analysis
- Protocol stack manipulation
- Address transformation
- Debugging multiaddr structures
"""
import sys
from multiaddr import Multiaddr
from multiaddr.protocols import P_IP4, P_IP6, P_TCP, P_TLS, P_UDP
def print_separator(title):
"""Print a formatted separator with title."""
print(f"\n{'=' * 60}")
print(f" {title}")
print(f"{'=' * 60}")
def print_multiaddr_info(ma, description=""):
"""Print multiaddr information in a formatted way."""
if description:
print(f"\n{description}:")
print(f" String: {ma}")
print(f" Protocols: {[p.name for p in ma.protocols()]}")
print(f" Protocol codes: {[p.code for p in ma.protocols()]}")
def basic_decapsulate_examples():
"""Demonstrate basic decapsulate_code usage."""
print_separator("Basic Decapsulate Code Examples")
# Example 1: Remove TCP layer
ma1 = Multiaddr("/ip4/192.168.1.1/tcp/8080/udp/1234")
print_multiaddr_info(ma1, "Original multiaddr")
# Remove TCP (protocol code 6) and everything after it
result1 = ma1.decapsulate_code(P_TCP)
print_multiaddr_info(result1, "After decapsulating TCP (code 6)")
# Example 2: Remove UDP layer
ma2 = Multiaddr("/ip4/10.0.0.1/udp/1234/tcp/8080")
print_multiaddr_info(ma2, "Original multiaddr")
result2 = ma2.decapsulate_code(P_UDP)
print_multiaddr_info(result2, "After decapsulating UDP (code 17)")
# Example 3: Remove IP layer
ma3 = Multiaddr(
"/ip4/172.16.0.1/tcp/443/tls/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN"
)
print_multiaddr_info(ma3, "Original multiaddr")
result3 = ma3.decapsulate_code(P_IP4)
print_multiaddr_info(result3, "After decapsulating IP4 (code 4)")
def protocol_stack_analysis():
"""Demonstrate protocol stack analysis using decapsulate_code."""
print_separator("Protocol Stack Analysis")
# Complex multiaddr with multiple layers
ma = Multiaddr(
"/ip4/192.168.1.100/tcp/8080/tls/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN"
)
print_multiaddr_info(ma, "Complex multiaddr")
print("\nProtocol stack analysis:")
current = ma
layer = 1
while str(current) != "/":
print(f" Layer {layer}: {current}")
protocols = list(current.protocols())
if protocols:
# Remove the last protocol layer
last_protocol = protocols[-1]
current = current.decapsulate_code(last_protocol.code)
layer += 1
else:
break
def address_transformation():
"""Demonstrate address transformation scenarios."""
print_separator("Address Transformation Examples")
# Example 1: Convert secure to insecure
secure_addr = Multiaddr(
"/ip4/10.0.0.1/tcp/443/tls/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN"
)
print_multiaddr_info(secure_addr, "Secure address")
# Remove TLS layer to get insecure version
insecure_addr = secure_addr.decapsulate_code(P_TLS)
print_multiaddr_info(insecure_addr, "Insecure version (TLS removed)")
# Example 2: Extract base network address
full_addr = Multiaddr("/ip6/2001:db8::1/tcp/8080/udp/1234/sctp/5678")
print_multiaddr_info(full_addr, "Full address")
# Remove all transport layers to get just the IP address
base_addr = full_addr.decapsulate_code(P_TCP)
print_multiaddr_info(base_addr, "Base network address (transport removed)")
# Example 3: IPv4 to IPv6 conversion simulation
ipv4_addr = Multiaddr("/ip4/192.168.1.1/tcp/8080")
print_multiaddr_info(ipv4_addr, "IPv4 address")
# Simulate removing IPv4 to prepare for IPv6 replacement
transport_only = ipv4_addr.decapsulate_code(P_IP4)
print_multiaddr_info(transport_only, "Transport layer only (IP removed)")
def error_handling_examples():
"""Demonstrate error handling with decapsulate_code."""
print_separator("Error Handling Examples")
# Example 1: Protocol not found
ma = Multiaddr("/ip4/192.168.1.1/tcp/8080")
print_multiaddr_info(ma, "Original address")
try:
# Try to remove a protocol that doesn't exist
result = ma.decapsulate_code(P_UDP) # UDP not in this address
print_multiaddr_info(result, "After removing non-existent UDP")
except Exception as e:
print(f" Error: {e}")
# Example 2: Empty multiaddr
empty_ma = Multiaddr("/")
print_multiaddr_info(empty_ma, "Empty multiaddr")
try:
result = empty_ma.decapsulate_code(P_TCP)
print_multiaddr_info(result, "After decapsulating from empty")
except Exception as e:
print(f" Error: {e}")
def practical_use_cases():
"""Demonstrate practical use cases for decapsulate_code."""
print_separator("Practical Use Cases")
# Use case 1: Network configuration analysis
print("Use Case 1: Network Configuration Analysis")
configs = [
"/ip4/0.0.0.0/tcp/8080",
"/ip4/0.0.0.0/tcp/8080/tls",
"/ip6/::/tcp/8080/tls/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
"/ip4/192.168.1.1/udp/1234",
]
for config in configs:
ma = Multiaddr(config)
print(f"\n Config: {ma}")
# Check if it's a server config (has wildcard IP)
if "0.0.0.0" in str(ma) or "::" in str(ma):
print(" Type: Server binding address")
# Extract transport info
transport = ma.decapsulate_code(
P_IP4 if P_IP4 in [p.code for p in ma.protocols()] else P_IP6
)
print(f" Transport: {transport}")
else:
print(" Type: Client address")
# Use case 2: Protocol compatibility checking
print("\nUse Case 2: Protocol Compatibility Checking")
addresses = [
"/ip4/192.168.1.1/tcp/8080",
"/ip4/192.168.1.1/tcp/8080/tls",
"/ip6/2001:db8::1/udp/1234",
]
for addr_str in addresses:
ma = Multiaddr(addr_str)
print(f"\n Address: {ma}")
# Check if supports TLS
if P_TLS in [p.code for p in ma.protocols()]:
print(" TLS: Supported")
insecure = ma.decapsulate_code(P_TLS)
print(f" Insecure version: {insecure}")
else:
print(" TLS: Not supported")
# Check transport protocol
if P_TCP in [p.code for p in ma.protocols()]:
print(" Transport: TCP")
elif P_UDP in [p.code for p in ma.protocols()]:
print(" Transport: UDP")
def main():
"""Run all decapsulate_code examples."""
print("Decapsulate Code Examples")
print("Demonstrating multiaddr protocol layer manipulation")
try:
basic_decapsulate_examples()
protocol_stack_analysis()
address_transformation()
error_handling_examples()
practical_use_cases()
print_separator("Summary")
print("The decapsulate_code method provides fine-grained control")
print("over multiaddr protocol layers, enabling:")
print("- Protocol stack analysis")
print("- Address transformation")
print("- Network configuration management")
print("- Protocol compatibility checking")
except KeyboardInterrupt:
print("\n\nExamples interrupted by user.")
sys.exit(1)
except Exception as e:
print(f"\n\nError running examples: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
Running the Examples¶
All examples can be run directly with Python:
# DNS examples
python examples/dns/dns_examples.py
# DNSADDR examples
python examples/dnsaddr/dnsaddr.py
# QUIC examples
python examples/quic/simple_quic_usage.py
# Thin waist examples
python examples/thin_waist/thin_waist_example.py
# Decapsulate examples
python examples/decapsulate/decapsulate_example.py
Note: Some examples require network connectivity and may take a few seconds to complete due to DNS resolution.