py-libp2p/libp2p/peer/peerinfo.py

from collections.abc import (
    Sequence,
)
from typing import (
    Any,
)

import multiaddr

from .id import (
    ID,
)


class PeerInfo:
    peer_id: ID
    addrs: list[multiaddr.Multiaddr]

    def __init__(self, peer_id: ID, addrs: Sequence[multiaddr.Multiaddr]) -> None:
        self.peer_id = peer_id
        self.addrs = list(addrs)

    def __eq__(self, other: Any) -> bool:
        return (
            isinstance(other, PeerInfo)
            and self.peer_id == other.peer_id
            and self.addrs == other.addrs
        )


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

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

    p2p_part = parts[-1]
    p2p_protocols = p2p_part.protocols()
    if not p2p_protocols:
        raise InvalidAddrError("The last part of the address has no protocols")
    last_protocol = p2p_protocols[0]
    if last_protocol is None:
        raise InvalidAddrError("The last protocol is None")

    last_protocol_code = last_protocol.code
    if last_protocol_code != multiaddr.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 = p2p_part.value_for_protocol(multiaddr.multiaddr.protocols.P_P2P)
    if peer_id_str is None:
        raise InvalidAddrError("Missing value for /p2p protocol in multiaddr")

    peer_id: ID = ID.from_base58(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])

    return PeerInfo(peer_id, [addr])


class InvalidAddrError(ValueError):
    pass