libp2p.utils package

Submodules

libp2p.utils.logging module

libp2p.utils.logging.cleanup_logging() → None: Clean up logging resources on exit.

libp2p.utils.logging.setup_logging() → None

Set up logging configuration based on environment variables.

Environment Variables:

LIBP2P_DEBUG: Controls logging levels. Examples: - “DEBUG” (all modules at DEBUG level) - “libp2p.identity.identify:DEBUG” (only identify module at DEBUG) - “identity.identify:DEBUG” (same as above, libp2p prefix optional) - “libp2p.identity:DEBUG,libp2p.transport:INFO” (multiple modules)
LIBP2P_DEBUG_FILE: If set, specifies the file path for log output. When this variable is set, logs will only be written to the specified file. If not set, logs will be written to both a default file (in the system’s temp directory) and to stderr (console output).

The logging system uses Python’s native hierarchical logging:

Loggers are organized in a hierarchy using dots (e.g., libp2p.identity.identify)
Child loggers inherit their parent’s level unless explicitly set
The root libp2p logger controls the default level

libp2p.utils.multiaddr_utils module

Multiaddr utility functions for IPv4/IPv6 handling.

This module provides helper functions to extract IP addresses from multiaddrs in a version-agnostic way, supporting both IPv4 and IPv6. Uses py-multiaddr utilities (e.g. get_multiaddr_options) when available.

libp2p.utils.multiaddr_utils.extract_host_from_multiaddr(maddr: Multiaddr) → str | None

Extract host (IP or DNS name) from multiaddr.

Tries ip4, ip6, dns, dns4, dns6 in order. Returns the first found value, or None if no host protocol is present. Handles ProtocolLookupError gracefully for each protocol.

Parameters:: maddr – Multiaddr to extract host from
Returns:: Host string or None if not found

libp2p.utils.multiaddr_utils.extract_ip_from_multiaddr(maddr: Multiaddr) → str | None

Extract IP address (IPv4 or IPv6) from multiaddr using py-multiaddr utilities.

Prefers get_multiaddr_options for consistent parsing; falls back to value_for_protocol for ip4/ip6 when options do not provide a host.

Parameters:: maddr – Multiaddr to extract from
Returns:: IP address string or None if not found

libp2p.utils.multiaddr_utils.format_host_for_url(host: str) → str

Format host for use in URLs.

IPv6 addresses (detected by presence of :) are wrapped in square brackets per RFC 3986.

Examples:

>>> format_host_for_url("127.0.0.1")
'127.0.0.1'
>>> format_host_for_url("::1")
'[::1]'
>>> format_host_for_url("example.com")
'example.com'

Parameters:: host – Hostname or IP address string
Returns:: Host formatted for URL inclusion

libp2p.utils.multiaddr_utils.get_ip_protocol_from_multiaddr(maddr: Multiaddr) → str | None

Get the IP protocol name (ip4 or ip6) from multiaddr.

Parameters:: maddr – Multiaddr to check
Returns:: Protocol name (“ip4” or “ip6”) or None

libp2p.utils.multiaddr_utils.get_protocol_layers(maddr: Multiaddr, maxsplit: int = -1) → list[Multiaddr]

Return protocol stack as per-layer multiaddrs (Section 3.2 multiaddr integration).

Uses Multiaddr.split() for protocol stack analysis and debugging.

Parameters:

maddr – Multiaddr to analyze
maxsplit – Optional max number of splits (-1 for no limit)

Returns:

List of Multiaddr, one per protocol layer

libp2p.utils.multiaddr_utils.join_multiaddrs(*addrs: str | bytes | Multiaddr) → Multiaddr

Concatenate multiple multiaddrs (Section 7.1).

Uses Multiaddr.join() when available (py-multiaddr 0.1.1+), otherwise chains encapsulate() for compatibility with older versions.

Parameters:: addrs – One or more multiaddr strings, bytes, or Multiaddr instances
Returns:: Single Multiaddr combining all parts in order

libp2p.utils.multiaddr_utils.multiaddr_from_socket(socket_obj: socket | Any) → Multiaddr

Create multiaddr from socket, detecting IPv4 or IPv6.

Parameters:: socket_obj – Socket to get address from (socket.socket or trio.SocketType)
Returns:: Multiaddr with appropriate IP protocol

libp2p.utils.varint module

libp2p.utils.varint.decode_uvarint(data: bytes) → int: Decode a varint from bytes.

async libp2p.utils.varint.decode_uvarint_from_stream(reader: Reader) → int: https://en.wikipedia.org/wiki/LEB128.

libp2p.utils.varint.decode_varint_from_bytes(data: bytes) → int: Decode a varint from bytes (alias for decode_uvarint for backward comp).

libp2p.utils.varint.decode_varint_with_size(data: bytes) → tuple[int, int]

Decode a varint from bytes and return both the value and the number of bytes consumed.

Returns:: Tuple[int, int]: (value, bytes_consumed)

libp2p.utils.varint.encode_delim(msg: bytes) → bytes

libp2p.utils.varint.encode_uvarint(value: int) → bytes: Encode an unsigned integer as a varint.

libp2p.utils.varint.encode_varint_prefixed(data: bytes) → bytes: Encode data with a varint length prefix.

async libp2p.utils.varint.read_delim(reader: Reader) → bytes

async libp2p.utils.varint.read_length_prefixed_protobuf(stream: INetStream, use_varint_format: bool = True, max_length: int = 1048576) → bytes: Read a protobuf message from a stream, handling both formats.

async libp2p.utils.varint.read_varint_prefixed_bytes(reader: Reader) → bytes

libp2p.utils.varint.read_varint_prefixed_bytes_sync(stream: BinaryIO, max_length: int = 1048576) → bytes

Read varint-prefixed bytes from a stream.

Args:: stream: A stream-like object with a read() method max_length: Maximum allowed data length to prevent memory exhaustion
Returns:: bytes: The data without the length prefix
Raises:: ValueError: If the length prefix is invalid or too large EOFError: If the stream ends unexpectedly

libp2p.utils.version module

libp2p.utils.version.get_agent_version() → str

Return the version of libp2p.

If the version cannot be determined due to an exception, return “py-libp2p/unknown”.

Returns:: The version of libp2p.
Return type:: str

libp2p.utils.paths module

Cross-platform path utilities for py-libp2p.

This module provides standardized path operations to ensure consistent behavior across Windows, macOS, and Linux platforms.

Key functions include:

join_paths() – join path components using the OS-correct separator.
get_temp_dir() – get the system temporary directory.
get_script_dir() – get the directory of the calling script.
create_temp_file() – create a unique temporary file.
ensure_dir_exists() – create a directory (and parents) if needed.
resolve_relative_path() – resolve a relative path from a base.

Use these instead of os.path helpers or hard-coded separators to keep the codebase portable across Windows, macOS, and Linux.

libp2p.utils.paths.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

libp2p.utils.paths.ensure_dir_exists(path: str | Path) → Path

Ensure directory exists, create if needed.

Args:: path: Directory path to ensure exists
Returns:: Path: Path object for the directory

libp2p.utils.paths.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

libp2p.utils.paths.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

libp2p.utils.paths.get_config_dir() → Path

Get user config directory (cross-platform).

Returns:: Path: Platform-specific config directory

libp2p.utils.paths.get_project_root() → Path

Get the project root directory.

Returns:: Path: Path to the py-libp2p project root

libp2p.utils.paths.get_python_executable() → Path

Get current Python executable path.

Returns:: Path: Path to the current Python executable

libp2p.utils.paths.get_script_binary_path() → Path

Get path to script’s binary directory.

Returns:: Path: Directory containing the script’s binary

libp2p.utils.paths.get_script_dir(script_path: str | Path | 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

libp2p.utils.paths.get_temp_dir() → Path

Get cross-platform temporary directory.

Returns:: Path: Platform-specific temporary directory path

libp2p.utils.paths.get_venv_path() → Path | None

Get virtual environment path if active.

Returns:: Path: Virtual environment path if active, None otherwise

libp2p.utils.paths.join_paths(*parts: str | Path) → Path

Cross-platform path joining.

Args:: parts: Path components to join
Returns:: Path: Joined path using platform-appropriate separator

libp2p.utils.paths.normalize_path(path: str | Path) → Path

Normalize a path, resolving any symbolic links and relative components.

Args:: path: Path to normalize
Returns:: Path: Normalized absolute path

libp2p.utils.paths.resolve_relative_path(base_path: str | Path, relative_path: str | Path) → 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

libp2p.utils.address_validation module

libp2p.utils.address_validation.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.

Parameters:

addr – Multiaddr to expand.
port – Optional override for port selection.

Returns:

List of concrete Multiaddr instances.

libp2p.utils.address_validation.find_free_port(ip_version: int = 4) → int

Find a free port on localhost.

Parameters:: ip_version – IP version (4 for IPv4, 6 for IPv6)
Returns:: Available port number

libp2p.utils.address_validation.get_available_interfaces(port: int, protocol: str = 'tcp') → list[Multiaddr]

Discover available network interfaces (IPv4 + IPv6 if supported) for binding.

Parameters:

port – Port number to bind to.
protocol – Transport protocol (e.g., “tcp” or “udp”).

Returns:

List of Multiaddr objects representing candidate interface addresses.

libp2p.utils.address_validation.get_optimal_binding_address(port: int, protocol: str = 'tcp') → Multiaddr

Choose an optimal address for an example to bind to: - Prefer non-loopback IPv6 - Then non-loopback IPv4 - Then loopback IPv6 - Then loopback IPv4 - Fallback to wildcard IPv4

Parameters:

port – Port number.
protocol – Transport protocol.

Returns:

A single Multiaddr chosen heuristically.

libp2p.utils.address_validation.get_wildcard_address(port: int, protocol: str = 'tcp', ip_version: int = 4) → Multiaddr

Get wildcard address (0.0.0.0 or ::) when explicitly needed.

This function provides access to wildcard binding as a feature when explicitly required, preserving the ability to bind to all interfaces.

Parameters:

port – Port number.
protocol – Transport protocol.
ip_version – IP version (4 for 0.0.0.0, 6 for ::).

Returns:

A Multiaddr with wildcard binding (0.0.0.0 or ::).

libp2p.utils.dns_utils module

DNS resolution utilities with retry, exponential backoff, and optional timeout.

Used by bootstrap discovery and transports (TCP, WebSocket) for consistent DNS resolution behavior and optional metrics.

class libp2p.utils.dns_utils.DNSResolutionMetrics(success_count: int = 0, failure_count: int = 0)

Bases: object

Simple metrics for DNS resolution (success/failure counts).

failure_count: int = 0

record_failure() → None

record_success() → None

success_count: int = 0

libp2p.utils.dns_utils.get_default_dns_metrics() → DNSResolutionMetrics | None: Return the default DNS metrics instance if set.

async libp2p.utils.dns_utils.resolve_multiaddr_with_retry(maddr: Multiaddr, resolver: DNSResolver | None = None, *, max_retries: int = 3, base_delay: float = 0.5, max_delay: float = 10.0, timeout_seconds: float | None = 10.0, metrics: DNSResolutionMetrics | None = None) → list[Multiaddr] | None

Resolve a DNS multiaddr with retries and exponential backoff.

Parameters:

maddr – Multiaddr to resolve (dns/dns4/dns6/dnsaddr as first protocol).
resolver – DNSResolver instance; if None, a new one is created.
max_retries – Maximum number of resolution attempts (including first).
base_delay – Initial delay in seconds before first retry.
max_delay – Maximum delay in seconds between retries.
timeout_seconds – Per-attempt timeout in seconds; None for no timeout.
metrics – Optional metrics to record success/failure.

Returns:

List of resolved Multiaddrs, or None if all attempts failed.

libp2p.utils.dns_utils.set_default_dns_metrics(metrics: DNSResolutionMetrics | None) → None: Set the default DNS metrics instance (e.g. for bootstrap or host).