feat: add from_net_addr/to_net_addr helpers (fixes #115)

README.rst

@@ -263,6 +263,26 @@ for ``socket.connect()``:
 
 See ``examples/dial_args/dial_args_example.py`` for a printable demo.
 
+Socket address conversion
+-------------------------
+
+Convert between Python socket address tuples and multiaddrs:
+
+
+.. code-block:: python
+
+    from multiaddr import from_net_addr, to_net_addr
+
+    ma = from_net_addr(("1.2.3.4", 80))
+    print(ma)
+    # /ip4/1.2.3.4/tcp/80
+    print(to_net_addr(ma))
+    # ('1.2.3.4', 80)
+    print(from_net_addr(("::1", 53), transport="udp"))
+    # /ip6/::1/udp/53
+
+See ``examples/net_addr/net_addr_example.py`` for a printable demo.
+
 Thin Waist Address Validation
 -----------------------------
 

docs/examples.rst

@@ -156,6 +156,15 @@ socket-ready ``(network, address)`` pairs.
    :language: python
    :caption: examples/dial_args/dial_args_example.py
 
+Socket address conversion
+-------------------------
+
+The `examples/net_addr/` directory demonstrates ``from_net_addr`` / ``to_net_addr``.
+
+.. literalinclude:: ../examples/net_addr/net_addr_example.py
+   :language: python
+   :caption: examples/net_addr/net_addr_example.py
+
 Resolver Utility Examples
 --------------------------
 

examples/net_addr/net_addr_example.py

@@ -0,0 +1,32 @@
+"""
+from_net_addr / to_net_addr demo.
+
+Usage:
+    python examples/net_addr/net_addr_example.py
+"""
+
+from multiaddr import from_net_addr, to_net_addr
+
+
+def main() -> None:
+    print("=== from_net_addr() ===")
+    samples = [
+        (("1.2.3.4", 80), "tcp"),
+        (("::1", 443), "tcp"),
+        (("8.8.8.8", 53), "udp"),
+    ]
+    for addr, transport in samples:
+        ma = from_net_addr(addr, transport=transport)
+        print(f"{addr} transport={transport} -> {ma}")
+
+    print()
+    print("=== to_net_addr() ===")
+    for ma in (
+        from_net_addr(("1.2.3.4", 80)),
+        from_net_addr(("::1", 443)),
+    ):
+        print(f"{ma} -> {to_net_addr(ma)}")
+
+
+if __name__ == "__main__":
+    main()

multiaddr/__init__.py

@@ -35,6 +35,7 @@
     PRIVATE4,
     PRIVATE6,
     dial_args,
+    from_net_addr,
     get_multiaddr_options,
     get_network_addrs,
     get_thin_waist_addresses,
@@ -48,6 +49,7 @@
     is_public_addr,
     is_thin_waist,
     is_wildcard,
+    to_net_addr,
 )
 
 __author__ = "Steven Buss"
@@ -86,6 +88,7 @@
     "ResolutionError",
     "StringParseError",
     "dial_args",
+    "from_net_addr",
     "get_multiaddr_options",
     "get_network_addrs",
     "get_thin_waist_addresses",
@@ -101,4 +104,5 @@
     "is_wildcard",
     "protocol_with_code",
     "protocol_with_name",
+    "to_net_addr",
 ]

multiaddr/utils.py

@@ -265,3 +265,50 @@ def dial_args(ma: Multiaddr) -> tuple[str, str]:
         address = f"{opts['host']}:{opts['port']}"
 
     return (network, address)
+
+
+def from_net_addr(
+    addr: tuple[Any, ...],
+    *,
+    transport: str = "tcp",
+) -> Multiaddr:
+    """Convert a socket address tuple to a Multiaddr.
+
+    Args:
+        addr: A socket address tuple such as ``(host, port)`` or an IPv6
+            ``(host, port, flowinfo, scope_id)`` tuple.
+        transport: ``"tcp"`` or ``"udp"`` (default ``"tcp"``).
+
+    Examples:
+        >>> from_net_addr(("1.2.3.4", 80))
+        Multiaddr('/ip4/1.2.3.4/tcp/80')
+        >>> from_net_addr(("::1", 53), transport="udp")
+        Multiaddr('/ip6/::1/udp/53')
+    """
+    if transport not in ("tcp", "udp"):
+        raise ValueError(f"unsupported transport: {transport!r}")
+    if not addr or len(addr) < 2:
+        raise ValueError("addr must be a (host, port[, ...]) tuple")
+
+    host, port = addr[0], addr[1]
+    if not isinstance(host, str):
+        raise TypeError("host must be a string")
+    if not isinstance(port, int):
+        raise TypeError("port must be an integer")
+
+    ip = ipaddress.ip_address(host)
+    ip_proto = "ip4" if isinstance(ip, ipaddress.IPv4Address) else "ip6"
+    return Multiaddr(f"/{ip_proto}/{host}/{transport}/{port}")
+
+
+def to_net_addr(ma: Multiaddr) -> tuple[str, int]:
+    """Convert a thin-waist Multiaddr to a ``(host, port)`` socket address tuple.
+
+    Examples:
+        >>> to_net_addr(Multiaddr("/ip4/1.2.3.4/tcp/80"))
+        ('1.2.3.4', 80)
+    """
+    opts = get_multiaddr_options(ma)
+    if opts is None:
+        raise ValueError(f"{ma} is not a thin waist address")
+    return (opts["host"], opts["port"])

newsfragments/115.feature.rst

@@ -0,0 +1 @@
+Add ``from_net_addr()`` / ``to_net_addr()`` for socket address tuple conversion.

tests/test_net_addr.py

@@ -0,0 +1,31 @@
+import pytest
+
+from multiaddr import Multiaddr, from_net_addr, to_net_addr
+
+
+def test_from_net_addr_tcp4():
+    assert str(from_net_addr(("1.2.3.4", 80))) == "/ip4/1.2.3.4/tcp/80"
+
+
+def test_from_net_addr_udp6():
+    assert str(from_net_addr(("::1", 53), transport="udp")) == "/ip6/::1/udp/53"
+
+
+def test_from_net_addr_ipv6_tuple():
+    assert str(from_net_addr(("2001:db8::1", 443, 0, 0))) == "/ip6/2001:db8::1/tcp/443"
+
+
+def test_from_net_addr_rejects_bad_transport():
+    with pytest.raises(ValueError, match="unsupported transport"):
+        from_net_addr(("1.2.3.4", 80), transport="sctp")
+
+
+def test_to_net_addr_roundtrip():
+    ma = Multiaddr("/ip4/1.2.3.4/tcp/80")
+    assert to_net_addr(ma) == ("1.2.3.4", 80)
+    assert str(from_net_addr(to_net_addr(ma))) == str(ma)
+
+
+def test_to_net_addr_rejects_non_thin_waist():
+    with pytest.raises(ValueError, match="thin waist"):
+        to_net_addr(Multiaddr("/unix/tmp/socket"))

tests/test_package_exports.py

@@ -2,7 +2,15 @@
 
 EXPECTED_EXPORTS = {
     "Action",
+    "BinaryParseError",
     "Filters",
+    "IP4_LOOPBACK",
+    "IP4_UNSPECIFIED",
+    "IP6_LOOPBACK",
+    "IP6_UNSPECIFIED",
+    "Multiaddr",
+    "PRIVATE4",
+    "PRIVATE6",
     "PROTOCOLS",
     "P_DNS",
     "P_DNS4",
@@ -13,28 +21,21 @@
     "P_P2P",
     "P_TCP",
     "P_UDP",
-    "REGISTRY",
-    "IP4_LOOPBACK",
-    "IP4_UNSPECIFIED",
-    "IP6_LOOPBACK",
-    "IP6_UNSPECIFIED",
-    "PRIVATE4",
-    "PRIVATE6",
-    "BinaryParseError",
-    "Multiaddr",
     "ParseError",
     "Protocol",
     "ProtocolExistsError",
     "ProtocolLookupError",
     "ProtocolNotFoundError",
     "ProtocolRegistryLocked",
+    "REGISTRY",
     "RecursionLimitError",
     "ResolutionError",
     "StringParseError",
+    "dial_args",
+    "from_net_addr",
     "get_multiaddr_options",
     "get_network_addrs",
     "get_thin_waist_addresses",
-    "dial_args",
     "interface_multiaddrs",
     "is_ip6_link_local",
     "is_ip_loopback",
@@ -47,6 +48,7 @@
     "is_wildcard",
     "protocol_with_code",
     "protocol_with_name",
+    "to_net_addr",
 }