test_transforms.py

"""Tests for multiaddr resolvers."""

import io

import pytest

import multiaddr.protocols
from multiaddr.codecs import CODEC_CACHE, CodecBase, codec_by_name
from multiaddr.exceptions import BinaryParseError, StringParseError
from multiaddr.multiaddr import Multiaddr
from multiaddr.protocols import REGISTRY, Protocol
from multiaddr.transforms import bytes_iter, bytes_to_string, size_for_addr, string_to_bytes

# These test values were generated by running them through the go implementation
# of multiaddr (https://github.com/multiformats/go-multiaddr)
#
# All values are bijective.
ADDR_BYTES_STR_TEST_DATA = [
    (REGISTRY.find("ip4"), b"\x0a\x0b\x0c\x0d", "10.11.12.13"),
    (
        REGISTRY.find("ip6"),
        b"\x1a\xa1\x2b\xb2\x3c\xc3\x4d\xd4\x5e\xe5\x6f\xf6\x7a\xb7\x8a\xc8",
        "1aa1:2bb2:3cc3:4dd4:5ee5:6ff6:7ab7:8ac8",
    ),
    (REGISTRY.find("tcp"), b"\xab\xcd", "43981"),
    (
        REGISTRY.find("onion"),
        b"\x9a\x18\x08\x73\x06\x36\x90\x43\x09\x1f\x04\xd2",
        "timaq4ygg2iegci7:1234",
    ),
    (
        REGISTRY.find("p2p"),
        b"\x12\x20\xd5\x2e\xbb\x89\xd8\x5b\x02\xa2\x84\x94\x82\x03\xa6\x2f"
        b"\xf2\x83\x89\xc5\x7c\x9f\x42\xbe\xec\x4e\xc2\x0d\xb7\x6a\x68\x91\x1c\x0b",
        "QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSupNKC",
    ),
    # Additional test data
    (
        REGISTRY.find("dns4"),
        b"\xd9\x85\xd9\x88\xd9\x82\xd8\xb9.\xd9\x88\xd8\xb2\xd8\xa7\xd8\xb1\xd8\xa9"
        b"-\xd8\xa7\xd9\x84\xd8\xa7\xd8\xaa\xd8\xb5\xd8\xa7\xd9\x84\xd8\xa7\xd8\xaa"
        b".\xd9\x85\xd8\xb5\xd8\xb1",
        # Explicitly mark this as unicode, as the "u" forces the text to be displayed LTR in editors
        "موقع.وزارة-الاتصالات.مصر",
    ),
    (
        REGISTRY.find("dns4"),
        b"fu\xc3\x9fball.example",
        "fußball.example",
    ),  # This will fail if IDNA-2003/NamePrep is used
]

ADDR_BYTES_FROM_STR_TEST_DATA = [
    # New CIDv1 string to new CIDv1 binary format mapping (non-bijective)
    (
        REGISTRY.find("p2p"),
        b"\x01\x72\x12\x20\xd5\x2e\xbb\x89\xd8\x5b\x02\xa2\x84\x94\x82\x03\xa6\x2f"
        b"\xf2\x83\x89\xc5\x7c\x9f\x42\xbe\xec\x4e\xc2\x0d\xb7\x6a\x68\x91\x1c\x0b",
        "bafzbeigvf25ytwc3akrijfecaotc74udrhcxzh2cx3we5qqnw5vgrei4bm",
    ),
]

ADDR_BYTES_TO_STR_TEST_DATA = [
    # Old CIDv0 binary to old CIDv0 string format mapping (non-bijective)
    (
        REGISTRY.find("p2p"),
        b"\x12\x20\xd5\x2e\xbb\x89\xd8\x5b\x02\xa2\x84\x94\x82\x03\xa6\x2f\xf2"
        b"\x83\x89\xc5\x7c\x9f\x42\xbe\xec\x4e\xc2\x0d\xb7\x6a\x68\x91\x1c\x0b",
        "QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSupNKC",
    ),
]

BYTES_MAP_STR_TEST_DATA = [
    ("/ip4/127.0.0.1/udp/1234", b"\x04\x7f\x00\x00\x01\x91\x02\x04\xd2"),
    ("/ip4/127.0.0.1/tcp/4321", b"\x04\x7f\x00\x00\x01\x06\x10\xe1"),
    (
        "/ip4/127.0.0.1/udp/1234/ip4/127.0.0.1/tcp/4321",
        b"\x04\x7f\x00\x00\x01\x91\x02\x04\xd2\x04\x7f\x00\x00\x01\x06\x10\xe1",
    ),
]


@pytest.mark.parametrize(
    "codec_name, buf, expected",
    [
        (None, b"\x01\x02\x03", (0, 0)),
        ("ip4", b"\x01\x02\x03", (4, 0)),
        ("cid", b"\x40\x50\x60\x51", (64, 1)),
    ],
)
def test_size_for_addr(codec_name, buf, expected):
    buf_io = io.BytesIO(buf)
    assert (size_for_addr(codec_by_name(codec_name), buf_io), buf_io.tell()) == expected


@pytest.mark.parametrize(
    "buf, expected",
    [
        # "/ip4/127.0.0.1/udp/1234/ip4/127.0.0.1/tcp/4321"
        (
            b"\x04\x7f\x00\x00\x01\x91\x02\x04\xd2\x04\x7f\x00\x00\x01\x06\x10\xe1",
            [
                (REGISTRY.find("ip4"), b"\x7f\x00\x00\x01"),
                (REGISTRY.find("udp"), b"\x04\xd2"),
                (REGISTRY.find("ip4"), b"\x7f\x00\x00\x01"),
                (REGISTRY.find("tcp"), b"\x10\xe1"),
            ],
        ),
    ],
)
def test_bytes_iter(buf, expected):
    assert list((proto, val) for _, proto, _, val in bytes_iter(buf)) == expected


@pytest.mark.parametrize(
    "proto, buf, expected", ADDR_BYTES_STR_TEST_DATA + ADDR_BYTES_TO_STR_TEST_DATA
)
def test_codec_to_string(proto, buf, expected):
    assert codec_by_name(proto.codec).to_string(proto, buf) == expected


@pytest.mark.parametrize(
    "proto, expected, string", ADDR_BYTES_STR_TEST_DATA + ADDR_BYTES_FROM_STR_TEST_DATA
)
def test_codec_to_bytes(proto, string, expected):
    assert codec_by_name(proto.codec).to_bytes(proto, string) == expected


@pytest.mark.parametrize("string, buf", BYTES_MAP_STR_TEST_DATA)
def test_string_to_bytes(string, buf):
    assert string_to_bytes(string) == buf
    assert Multiaddr(string).to_bytes() == buf


@pytest.mark.parametrize("string, buf", BYTES_MAP_STR_TEST_DATA)
def test_bytes_to_string(string, buf):
    assert bytes_to_string(buf) == string


class DummyProtocol(Protocol):
    def __init__(self, code, name, codec=None):
        self.code = code
        self.name = name
        self.codec = codec


class UnparsableProtocol(DummyProtocol):
    def __init__(self):
        super().__init__(
            333, "unparsable", "unparsable_codec"
        )  # Use a custom codec that will cause BinaryParseError


# Add a custom codec for UnparsableProtocol
class UnparsableCodec(CodecBase):
    def to_bytes(self, proto, string):
        raise BinaryParseError("Invalid bytes for unparsable protocol", b"", "unparsable")

    def to_string(self, proto, buf):
        raise BinaryParseError("Invalid bytes for unparsable protocol", buf, "unparsable")


# Register the custom codec
CODEC_CACHE["unparsable_codec"] = UnparsableCodec()


@pytest.fixture
def protocol_extension(monkeypatch):
    # "Add" additional non-parsable protocol to protocols from code list
    registry = multiaddr.protocols.REGISTRY.copy(unlock=True)
    registry.add(UnparsableProtocol())
    monkeypatch.setattr(multiaddr.protocols, "REGISTRY", registry)


@pytest.mark.parametrize("string", ["test", "/ip4/", "/unparsable/5"])
def test_string_to_bytes_value_error(protocol_extension, string):
    with pytest.raises(StringParseError):
        string_to_bytes(string)


@pytest.mark.parametrize("bytes", [b"\xcd\x02\x0c\x0d", b"\x35\x03a:b"])
def test_bytes_to_string_value_error(protocol_extension, bytes):
    with pytest.raises(BinaryParseError):
        bytes_to_string(bytes)


@pytest.mark.parametrize(
    "proto, address",
    [
        (REGISTRY.find("ip4"), "1124.2.3"),
        (REGISTRY.find("ip6"), "123.123.123.123"),
        (REGISTRY.find("tcp"), "a"),
        (REGISTRY.find("tcp"), "100000"),
        (REGISTRY.find("onion"), "100000"),
        (REGISTRY.find("onion"), "1234567890123456:0"),
        (REGISTRY.find("onion"), "timaq4ygg2iegci7:a"),
        (REGISTRY.find("onion"), "timaq4ygg2iegci7:0"),
        (REGISTRY.find("onion"), "timaq4ygg2iegci7:71234"),
        (REGISTRY.find("p2p"), "15230d52ebb89d85b02a284948203a"),
        (REGISTRY.find("ip6zone"), ""),
    ],
)
def test_codec_to_bytes_value_error(proto, address):
    # Codecs themselves may raise any exception type - it will then be converted
    # to `StringParseError` by a higher level
    with pytest.raises(Exception):
        codec_by_name(proto.codec).to_bytes(proto, address)


@pytest.mark.parametrize(
    "proto, buf",
    [
        (REGISTRY.find("tcp"), b"\xff\xff\xff\xff"),
        (REGISTRY.find("ip6zone"), b""),
    ],
)
def test_codec_to_string_value_error(proto, buf):
    # Codecs themselves may raise any exception type - it will then be converted
    # to `BinaryParseError` by a higher level
    with pytest.raises(Exception):
        codec_by_name(proto.codec).to_string(proto, buf)


@pytest.mark.parametrize(
    "proto, buf, expected",
    [
        (
            REGISTRY.find("p2p"),  # This one gets autoconverted to CIDv0
            b"\x01\x72\x00\x24\x08\x01\x12\x20\xc6\x35\xed\x47\x2d\x58\xed\xd8\xb9\xee\x46\x41"
            b"\xec\x75\x57\xdc\x80\x6b\x48\x86\xe5\x68\x49\x1a\xb0\x0b\xff\x12\x69\xa7\xbe\x65",
            "12D3KooWPA6ax6t3jqTyGq73Zm1RmwppYqxaXzrtarfcTWGp5Wzx",
        ),
    ],
)
def test_cid_autoconvert_to_string(proto, buf, expected):
    assert codec_by_name("cid").to_string(proto, buf) == expected


def test_string_iter_unix_path():
    from multiaddr.transforms import string_iter

    parts = list(string_iter("/unix/var/run/docker.sock"))
    assert len(parts) == 1
    proto, _, value = parts[0]
    assert proto.name == "unix"
    assert value == "var/run/docker.sock"