py-libp2p/libp2p/peer/peerinfo.py

from typing import (
    List,
)

import multiaddr

from .id import (
    ID,
    id_b58_decode,
)
from .peerdata import PeerData


class PeerInfo:
    # pylint: disable=too-few-public-methods

    peer_id: ID
    addrs: List[multiaddr.Multiaddr]

    def __init__(self, peer_id: ID, peer_data: PeerData = None) -> None:
        self.peer_id = peer_id
        self.addrs = peer_data.get_addrs() if peer_data else None


def info_from_p2p_addr(addr: multiaddr.Multiaddr) -> PeerInfo:
    if not addr:
        raise InvalidAddrError("`addr` should not be `None`")

    if not isinstance(addr, multiaddr.Multiaddr):
        raise InvalidAddrError(f"`addr`={addr} should be of type `Multiaddr`")

    parts = addr.split()
    if not parts:
        raise InvalidAddrError(f"`parts`={parts} should at least have a protocol `P_P2P`")

    p2p_part = parts[-1]
    last_protocol_code = p2p_part.protocols()[0].code
    if last_protocol_code != multiaddr.protocols.P_P2P:
        raise InvalidAddrError(
            f"The last protocol should be `P_P2P` instead of `{last_protocol_code}`"
        )

    # make sure the /p2p value parses as a peer.ID
    peer_id_str: str = p2p_part.value_for_protocol(multiaddr.protocols.P_P2P)
    peer_id: 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.Multiaddr.join(*parts[:-1])

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

    return PeerInfo(peer_id, peer_data)


class InvalidAddrError(ValueError):
    pass