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.

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.

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.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