py-libp2p/libp2p/peer/peerinfo.py

import multiaddr
import multiaddr.util

from .id import id_b58_decode
from .peerdata import PeerData


class PeerInfo:
    # pylint: disable=too-few-public-methods
    def __init__(self, peer_id, peer_data=None):
        self.peer_id = peer_id
        self.addrs = peer_data.get_addrs() if peer_data else None


def info_from_p2p_addr(addr):
    if not addr:
        raise InvalidAddrError()

    parts = multiaddr.util.split(addr)
    if not parts:
        raise InvalidAddrError()

    p2p_part = parts[-1]
    if p2p_part.protocols()[0].code != multiaddr.protocols.P_P2P:
        raise InvalidAddrError()

    # make sure the /p2p value parses as a peer.ID
    peer_id_str = p2p_part.value_for_protocol(multiaddr.protocols.P_P2P)
    peer_id = id_b58_decode(peer_id_str)

    # we might have received just an / p2p part, which means there's no addr.
    if len(parts) > 1:
        addr = multiaddr.util.join(parts[:-1])

    peer_data = PeerData()
    peer_data.addrs = [addr]
    peer_data.protocols = [p.code for p in addr.protocols()]

    return PeerInfo(peer_id, peer_data)


class InvalidAddrError(ValueError):
    pass