added WebSocket transport support

Signed-off-by: GautamBytes <manchandanigautam@gmail.com>

libp2p/transport/__init__.py

@@ -0,0 +1,7 @@
+from .tcp.tcp import TCP
+from .websocket.transport import WebsocketTransport
+
+__all__ = [
+    "TCP",
+    "WebsocketTransport",
+]

libp2p/transport/websocket/connection.py

@@ -0,0 +1,49 @@
+from trio.abc import Stream
+
+from libp2p.io.abc import ReadWriteCloser
+from libp2p.io.exceptions import IOException
+
+
+class P2PWebSocketConnection(ReadWriteCloser):
+    """
+    Wraps a raw trio.abc.Stream from an established websocket connection.
+    This bypasses message-framing issues and provides the raw stream
+    that libp2p protocols expect.
+    """
+
+    _stream: Stream
+
+    def __init__(self, stream: Stream):
+        self._stream = stream
+
+    async def write(self, data: bytes) -> None:
+        try:
+            await self._stream.send_all(data)
+        except Exception as e:
+            raise IOException from e
+
+    async def read(self, n: int | None = None) -> bytes:
+        """
+        Read up to n bytes (if n is given), else read up to 64KiB.
+        """
+        try:
+            if n is None:
+                # read a reasonable chunk
+                return await self._stream.receive_some(2**16)
+            return await self._stream.receive_some(n)
+        except Exception as e:
+            raise IOException from e
+
+    async def close(self) -> None:
+        await self._stream.aclose()
+
+    def get_remote_address(self) -> tuple[str, int] | None:
+        sock = getattr(self._stream, "socket", None)
+        if sock:
+            try:
+                addr = sock.getpeername()
+                if isinstance(addr, tuple) and len(addr) >= 2:
+                    return str(addr[0]), int(addr[1])
+            except OSError:
+                return None
+        return None

libp2p/transport/websocket/listener.py

@@ -0,0 +1,81 @@
+import logging
+import socket
+from typing import Any
+
+from multiaddr import Multiaddr
+import trio
+from trio_typing import TaskStatus
+from trio_websocket import serve_websocket
+
+from libp2p.abc import IListener
+from libp2p.custom_types import THandler
+from libp2p.network.connection.raw_connection import RawConnection
+
+from .connection import P2PWebSocketConnection
+
+logger = logging.getLogger("libp2p.transport.websocket.listener")
+
+
+class WebsocketListener(IListener):
+    """
+    Listen on /ip4/.../tcp/.../ws addresses, handshake WS, wrap into RawConnection.
+    """
+
+    def __init__(self, handler: THandler) -> None:
+        self._handler = handler
+        self._server = None
+
+    async def listen(self, maddr: Multiaddr, nursery: trio.Nursery) -> bool:
+        addr_str = str(maddr)
+        if addr_str.endswith("/wss"):
+            raise NotImplementedError("/wss (TLS) not yet supported")
+
+        host = (
+            maddr.value_for_protocol("ip4")
+            or maddr.value_for_protocol("ip6")
+            or maddr.value_for_protocol("dns")
+            or maddr.value_for_protocol("dns4")
+            or maddr.value_for_protocol("dns6")
+            or "0.0.0.0"
+        )
+        port = int(maddr.value_for_protocol("tcp"))
+
+        async def serve(
+            task_status: TaskStatus[Any] = trio.TASK_STATUS_IGNORED,
+        ) -> None:
+            # positional ssl_context=None
+            self._server = await serve_websocket(
+                self._handle_connection, host, port, None
+            )
+            task_status.started()
+            await self._server.wait_closed()
+
+        await nursery.start(serve)
+        return True
+
+    async def _handle_connection(self, websocket: Any) -> None:
+        try:
+            # use raw transport_stream
+            conn = P2PWebSocketConnection(websocket.stream)
+            raw = RawConnection(conn, initiator=False)
+            await self._handler(raw)
+        except Exception as e:
+            logger.debug("WebSocket connection error: %s", e)
+
+    def get_addrs(self) -> tuple[Multiaddr, ...]:
+        if not self._server or not self._server.sockets:
+            return ()
+        addrs = []
+        for sock in self._server.sockets:
+            host, port = sock.getsockname()[:2]
+            if sock.family == socket.AF_INET6:
+                addr = Multiaddr(f"/ip6/{host}/tcp/{port}/ws")
+            else:
+                addr = Multiaddr(f"/ip4/{host}/tcp/{port}/ws")
+            addrs.append(addr)
+        return tuple(addrs)
+
+    async def close(self) -> None:
+        if self._server:
+            self._server.close()
+            await self._server.wait_closed()

libp2p/transport/websocket/transport.py

@@ -0,0 +1,49 @@
+from multiaddr import Multiaddr
+from trio_websocket import open_websocket_url
+
+from libp2p.abc import IListener, ITransport
+from libp2p.custom_types import THandler
+from libp2p.network.connection.raw_connection import RawConnection
+from libp2p.transport.exceptions import OpenConnectionError
+
+from .connection import P2PWebSocketConnection
+from .listener import WebsocketListener
+
+
+class WebsocketTransport(ITransport):
+    """
+    Libp2p WebSocket transport: dial and listen on /ip4/.../tcp/.../ws
+    """
+
+    async def dial(self, maddr: Multiaddr) -> RawConnection:
+        text = str(maddr)
+        if text.endswith("/wss"):
+            raise NotImplementedError("/wss (TLS) not yet supported")
+        if not text.endswith("/ws"):
+            raise ValueError(f"WebsocketTransport only supports /ws, got {maddr}")
+
+        host = (
+            maddr.value_for_protocol("ip4")
+            or maddr.value_for_protocol("ip6")
+            or maddr.value_for_protocol("dns")
+            or maddr.value_for_protocol("dns4")
+            or maddr.value_for_protocol("dns6")
+        )
+        if host is None:
+            raise ValueError(f"No host protocol found in {maddr}")
+
+        port = int(maddr.value_for_protocol("tcp"))
+        uri = f"ws://{host}:{port}"
+
+        try:
+            async with open_websocket_url(uri, ssl_context=None) as ws:
+                conn = P2PWebSocketConnection(ws.stream)  # type: ignore[attr-defined]
+                return RawConnection(conn, initiator=True)
+        except Exception as e:
+            raise OpenConnectionError(f"Failed to dial WebSocket {maddr}: {e}") from e
+
+    def create_listener(self, handler: THandler) -> IListener:  # type: ignore[override]
+        """
+        The type checker is incorrectly reporting this as an inconsistent override.
+        """
+        return WebsocketListener(handler)