Add NAT traversal via UPnP port mapping

examples/upnp/README.md

@@ -0,0 +1,24 @@
+# UPnP Example
+
+This example demonstrates how to use the integrated UPnP behaviour to automatically create a port mapping on your local network's gateway (e.g., your home router).
+
+This allows peers from the public internet to directly dial your node, which is a crucial step for NAT traversal.
+
+## Usage
+
+First, ensure you have installed the necessary dependencies from the root of the repository:
+
+```sh
+pip install -e .[upnp]
+```
+
+Then, run the script in a terminal:
+
+```sh
+python examples/upnp/upnp_demo.py
+```
+
+The script will start a libp2p host and immediately try to discover a UPnP-enabled gateway on the network.
+
+- If it **succeeds**, it will print the new external address that has been mapped.
+- If it **fails** (e.g., your router doesn't have UPnP enabled or you're behind a double NAT), it will print a descriptive error message.

examples/upnp/upnp_demo.py

@@ -0,0 +1,68 @@
+import argparse
+import logging
+
+import multiaddr
+import trio
+
+from libp2p import new_host
+from libp2p.discovery.upnp import UpnpManager
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("upnp-demo")
+
+
+async def run(port: int) -> None:
+    listen_maddr = multiaddr.Multiaddr(f"/ip4/0.0.0.0/tcp/{port}")
+    host = new_host()
+
+    upnp = UpnpManager()
+    if not await upnp.discover():
+        logger.error(
+            "❌ Could not find a UPnP-enabled gateway. "
+            "The host will start, but may not be dialable from the public internet."
+        )
+    else:
+        logger.info(f"✅ UPnP gateway found! External IP: {upnp.get_external_ip()}")
+
+    mapped_port = None
+    async with host.run(listen_addrs=[listen_maddr]):
+        try:
+            actual_listen_maddr = host.get_addrs()[0]
+            mapped_port = actual_listen_maddr.value_for_protocol("tcp")
+
+            # If discovery was successful, now we can map the port
+            if upnp.get_external_ip():
+                if await upnp.add_port_mapping(mapped_port):
+                    logger.info(f"✅ Port {mapped_port} successfully mapped!")
+                else:
+                    logger.error(f"❌ Failed to map port {mapped_port}.")
+
+            logger.info(f"Host started with ID: {host.get_id().pretty()}")
+            logger.info(f"Listening on: {actual_listen_maddr}")
+            logger.info("Host is running. Press Ctrl+C to shut down.")
+            await trio.sleep_forever()
+
+        except KeyboardInterrupt:
+            logger.info("Shutting down...")
+        finally:
+            if mapped_port and upnp.get_external_ip():
+                logger.info("Removing UPnP port mapping...")
+                await upnp.remove_port_mapping(mapped_port)
+            logger.info("Shutdown complete.")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="UPnP example for py-libp2p")
+    parser.add_argument(
+        "-p", "--port", type=int, default=0, help="Local TCP port (0 for random)"
+    )
+    args = parser.parse_args()
+
+    try:
+        trio.run(run, args.port)
+    except KeyboardInterrupt:
+        pass
+
+
+if __name__ == "__main__":
+    main()

libp2p/discovery/upnp.py

@@ -0,0 +1,114 @@
+import ipaddress
+import logging
+
+import miniupnpc  # type: ignore[import-error]
+import trio
+
+logger = logging.getLogger("libp2p.discovery.upnp")
+
+
+class UpnpManager:
+    """
+    A simple, self-contained manager for UPnP port mapping that can be used
+    alongside a libp2p Host.
+    """
+
+    def __init__(self) -> None:
+        self._gateway = miniupnpc.UPnP()
+        self._lan_addr: str | None = None
+        self._external_ip: str | None = None
+
+    async def discover(self) -> bool:
+        """
+        Discover the UPnP IGD on the network.
+
+        :return: True if a gateway is found, False otherwise.
+        """
+        logger.debug("Discovering UPnP gateway...")
+        try:
+            try:
+                num_devices = await trio.to_thread.run_sync(self._gateway.discover)
+            except Exception as e:
+                if str(e) == "Success":
+                    num_devices = 1
+                else:
+                    logger.exception("UPnP discovery exception")
+                    return False
+
+            if num_devices > 0:
+                await trio.to_thread.run_sync(self._gateway.selectigd)
+                self._lan_addr = self._gateway.lanaddr
+                self._external_ip = await trio.to_thread.run_sync(
+                    self._gateway.externalipaddress
+                )
+                logger.debug(f"UPnP gateway found: {self._external_ip}")
+
+                if self._external_ip is None:
+                    logger.error("Gateway did not return an external IP address")
+                    return False
+
+                ip_obj = ipaddress.ip_address(self._external_ip)
+                if ip_obj.is_private:
+                    logger.warning(
+                        "UPnP gateway has a private IP; you may be behind a double NAT."
+                    )
+                    return False
+                return True
+            else:
+                logger.debug("No UPnP devices found")
+                return False
+        except Exception:
+            logger.exception("UPnP discovery failed")
+            return False
+
+    async def add_port_mapping(self, port: int, protocol: str = "TCP") -> bool:
+        """
+        Request a new port mapping from the gateway.
+
+        :param port: the internal port to map
+        :param protocol: the protocol to map (TCP or UDP)
+        :return: True on success, False otherwise
+        """
+        if not self._lan_addr:
+            logger.error(
+                "Cannot add port mapping: discovery has not been run successfully."
+            )
+            return False
+
+        logger.debug(f"Requesting UPnP mapping for {protocol} port {port}...")
+        try:
+            await trio.to_thread.run_sync(
+                lambda: self._gateway.addportmapping(
+                    port, protocol, self._lan_addr, port, "py-libp2p", ""
+                )
+            )
+            logger.info(
+                f"Successfully mapped external port {self._external_ip}:{port} "
+                f"to internal port {self._lan_addr}:{port}"
+            )
+            return True
+        except Exception:
+            logger.exception(f"Failed to map port {port}")
+        return False
+
+    async def remove_port_mapping(self, port: int, protocol: str = "TCP") -> bool:
+        """
+        Remove an existing port mapping.
+
+        :param port: the external port to unmap
+        :param protocol: the protocol (TCP or UDP)
+        :return: True on success, False otherwise
+        """
+        logger.debug(f"Removing UPnP mapping for {protocol} port {port}...")
+        try:
+            await trio.to_thread.run_sync(
+                lambda: self._gateway.deleteportmapping(port, protocol)
+            )
+            logger.info(f"Successfully removed mapping for port {port}")
+            return True
+        except Exception:
+            logger.exception(f"Failed to remove mapping for port {port}")
+        return False
+
+    def get_external_ip(self) -> str | None:
+        return self._external_ip

pyproject.toml

@@ -92,6 +92,8 @@ test = [
     "factory-boy>=2.12.0,<3.0.0",
 ]
 
+upnp = ["miniupnpc>=2.0"]
+
 [tool.setuptools]
 include-package-data = true