feat: pluggable transport capabilities (Phase 1, 2, 3)

Author: asmit27rai

libp2p/bitswap/cid.py

@@ -24,8 +24,19 @@
 import hashlib
 from typing import TypeAlias
 
-from cid import CIDv0, CIDv1, V0Builder, V1Builder, from_string, make_cid
-from cid.prefix import Prefix
+try:
+    from cid import CIDv0, CIDv1, V0Builder, V1Builder, from_string, make_cid
+    from cid.prefix import Prefix
+    _HAS_CID_BUILDERS = True
+except Exception:
+    # Older/newer py-cid variations may not expose builder/prefix helpers.
+    from cid import CIDv0, CIDv1, from_string, make_cid
+    Prefix = None  # type: ignore
+    V0Builder = None  # type: ignore
+    V1Builder = None  # type: ignore
+    _HAS_CID_BUILDERS = False
+
+import multihash as _multihash
 from multicodec import Code, is_codec
 from multicodec.code_table import DAG_PB, RAW, SHA2_256
 
@@ -61,7 +72,12 @@ def _normalise_codec(codec: Code | str | int) -> Code:
 
 def compute_cid_v0_obj(data: bytes) -> CIDv0:
     """Compute a CIDv0 object for data."""
-    return V0Builder().sum(data)
+    if _HAS_CID_BUILDERS and V0Builder is not None:
+        return V0Builder().sum(data)
+
+    # Fallback: compute sha2-256 multihash and construct CIDv0
+    mh = _multihash.digest(data, "sha2-256")
+    return CIDv0(mh.encode())
 
 
 def compute_cid_v0(data: bytes) -> bytes:
@@ -84,7 +100,13 @@ def compute_cid_v0(data: bytes) -> bytes:
 def compute_cid_v1_obj(data: bytes, codec: Code | str | int = CODEC_RAW) -> CIDv1:
     """Compute a CIDv1 object for data and codec."""
     code_obj = _normalise_codec(codec)
-    return V1Builder(codec=str(code_obj), mh_type=str(HASH_SHA256)).sum(data)
+    if _HAS_CID_BUILDERS and V1Builder is not None:
+        return V1Builder(codec=str(code_obj), mh_type=str(HASH_SHA256)).sum(data)
+
+    # Fallback: compute multihash and construct CIDv1 using codec name
+    codec_name = getattr(code_obj, "name", str(code_obj))
+    mh = _multihash.digest(data, "sha2-256")
+    return CIDv1(codec_name, mh.encode())
 
 
 def compute_cid_v1(data: bytes, codec: Code | str | int = CODEC_RAW) -> bytes:
@@ -132,7 +154,26 @@ def get_cid_prefix(cid: bytes) -> bytes:
     if cid_obj.version != CID_V1:
         return b""
 
-    return cid_obj.prefix().to_bytes()
+    # Prefer high-level Prefix helper when available
+    if Prefix is not None and hasattr(cid_obj, "prefix"):
+        try:
+            return cid_obj.prefix().to_bytes()
+        except Exception:
+            pass
+
+    # Fallback: reconstruct prefix by removing the digest bytes from the
+    # raw CID buffer using py-multihash to determine digest length.
+    try:
+        mh_bytes = getattr(cid_obj, "multihash", None)
+        if mh_bytes is None:
+            # Some CID implementations expose different attributes
+            # Fall back to parsing the tail of the buffer.
+            return b""
+        mh = _multihash.decode(mh_bytes)
+        digest_len = mh.length
+        return cid_obj.buffer[:-digest_len]
+    except Exception:
+        return b""
 
 
 def reconstruct_cid_from_prefix_and_data(prefix: bytes, data: bytes) -> bytes:
@@ -153,12 +194,16 @@ def reconstruct_cid_from_prefix_and_data(prefix: bytes, data: bytes) -> bytes:
         # No prefix means CIDv0
         return compute_cid_v0(data)
 
-    try:
-        return Prefix.from_bytes(prefix).sum(data).buffer
-    except ValueError:
-        # Preserve previous permissive behavior for malformed prefixes.
-        digest = hashlib.sha256(data).digest()
-        return prefix + digest
+    if Prefix is not None:
+        try:
+            return Prefix.from_bytes(prefix).sum(data).buffer
+        except ValueError:
+            pass
+
+    # Fallback: when Prefix helper isn't available, try to conservatively
+    # append the raw digest to the prefix (preserves prior permissive behavior).
+    digest = hashlib.sha256(data).digest()
+    return prefix + digest
 
 
 def verify_cid(cid: bytes, data: bytes) -> bool:
@@ -187,7 +232,11 @@ def verify_cid(cid: bytes, data: bytes) -> bool:
         return False
 
     try:
-        recomputed = cid_obj.prefix().sum(data).buffer
+        # Prefer using compute helpers which work across py-cid variants.
+        if cid_obj.version == CID_V1:
+            recomputed = compute_cid_v1(data, codec=getattr(cid_obj, "codec", None))
+        else:
+            recomputed = compute_cid_v0(data)
     except ValueError:
         logger.debug("        Failed to recompute CID from parsed prefix")
         return False

libp2p/capabilities.py

@@ -0,0 +1,89 @@
+"""
+Transport and connection capability declarations.
+
+Instead of the core checking ``isinstance(transport, QUICTransport)`` to decide
+whether to skip security or muxing upgrades, transports and connections *declare*
+what they already provide via simple boolean properties.
+
+Any transport or connection that satisfies the structural protocol (duck typing)
+is automatically recognised — no base-class inheritance required.
+
+Example — a hypothetical WebRTC transport that bundles its own DTLS + SCTP::
+
+    class WebRTCTransport(ITransport):
+        @property
+        def provides_security(self) -> bool:
+            return True
+
+        @property
+        def provides_muxing(self) -> bool:
+            return True
+
+See Also
+--------
+:pep:`544` — Structural subtyping (static duck typing) via ``typing.Protocol``.
+"""
+
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class TransportCapabilities(Protocol):
+    """Structural protocol implemented by transports that bundle their own
+    security and/or multiplexing layers.
+
+    A transport that does **not** implement these properties is assumed to
+    provide neither (``getattr(t, 'provides_security', False)`` → ``False``).
+    """
+
+    @property
+    def provides_security(self) -> bool:
+        """Return ``True`` if this transport includes built-in encryption /
+        authentication (e.g. QUIC's integrated TLS 1.3)."""
+        ...
+
+    @property
+    def provides_muxing(self) -> bool:
+        """Return ``True`` if this transport includes built-in stream
+        multiplexing (e.g. QUIC's native streams)."""
+        ...
+
+
+@runtime_checkable
+class ConnectionCapabilities(Protocol):
+    """Structural protocol implemented by connections that are already
+    secured and/or multiplexed at creation time.
+
+    A connection that does **not** implement these properties is assumed to
+    be neither secure nor muxed.
+    """
+
+    @property
+    def is_secure(self) -> bool:
+        """Return ``True`` if the connection is already encrypted and the
+        remote peer's identity has been verified."""
+        ...
+
+    @property
+    def is_muxed(self) -> bool:
+        """Return ``True`` if the connection already supports opening
+        multiple independent streams."""
+        ...
+
+
+@runtime_checkable
+class NeedsSetup(Protocol):
+    """Structural protocol for transports that require lifecycle hooks
+    from the swarm (e.g. a background nursery or a back-reference to
+    the swarm itself).
+
+    Transports that do **not** need these hooks simply omit the methods.
+    """
+
+    def set_background_nursery(self, nursery: object) -> None:
+        """Receive the long-lived nursery managed by the swarm."""
+        ...
+
+    def set_swarm(self, swarm: object) -> None:
+        """Receive a back-reference to the owning swarm."""
+        ...

libp2p/custom_types.py

@@ -17,6 +17,24 @@
     IMuxedStream = cast(type, object)
     QUICConnection = cast(type, object)
 
+from libp2p.io.abc import (
+    ReadWriteCloser,
+)
+from collections.abc import (
+    Awaitable,
+    Callable,
+    Mapping,
+)
+from typing import TYPE_CHECKING, NewType, Union, cast
+
+if TYPE_CHECKING:
+    from libp2p.abc import IMuxedConn, IMuxedStream, INetStream, ISecureTransport
+else:
+    IMuxedConn = cast(type, object)
+    INetStream = cast(type, object)
+    ISecureTransport = cast(type, object)
+    IMuxedStream = cast(type, object)
+
 from libp2p.io.abc import (
     ReadWriteCloser,
 )
@@ -37,6 +55,11 @@
 AsyncValidatorFn = Callable[[ID, rpc_pb2.Message], Awaitable[bool]]
 ValidatorFn = Union[SyncValidatorFn, AsyncValidatorFn]
 UnsubscribeFn = Callable[[], Awaitable[None]]
-TQUICStreamHandlerFn = Callable[[QUICStream], Awaitable[None]]
-TQUICConnHandlerFn = Callable[[QUICConnection], Awaitable[None]]
 MessageID = NewType("MessageID", str)
+
+# Re-export QUIC-specific types for backward compatibility.
+# New code should import directly from libp2p.transport.quic.types.
+from libp2p.transport.quic.types import (  # noqa: E402, F401
+    TQUICConnHandlerFn as TQUICConnHandlerFn,
+    TQUICStreamHandlerFn as TQUICStreamHandlerFn,
+)

libp2p/host/basic_host.py

@@ -92,7 +92,6 @@
 from libp2p.tools.async_service import (
     background_trio_service,
 )
-from libp2p.transport.quic.connection import QUICConnection
 import libp2p.utils.paths
 from libp2p.utils.varint import (
     read_length_prefixed_protobuf,
@@ -954,7 +953,7 @@ async def _on_notifee_connected(self, conn: INetConn) -> None:
         if not is_initiator:
             # Only the dialer (initiator) needs to actively run identify.
             return
-        if not self._is_quic_muxer(muxed_conn):
+        if not self._is_native_muxer(muxed_conn):
             return
         event_started = getattr(conn, "event_started", None)
         if event_started is not None and not event_started.is_set():
@@ -976,15 +975,16 @@ def _get_first_connection(self, peer_id: ID) -> INetConn | None:
             return connections[0]
         return None
 
-    def _is_quic_muxer(self, muxed_conn: IMuxedConn | None) -> bool:
-        return isinstance(muxed_conn, QUICConnection)
+    def _is_native_muxer(self, muxed_conn: IMuxedConn | None) -> bool:
+        """Return True if the muxed connection is natively muxed (e.g. QUIC)."""
+        return getattr(muxed_conn, 'is_muxed', False)
 
     def _should_identify_peer(self, peer_id: ID) -> bool:
         connection = self._get_first_connection(peer_id)
         if connection is None:
             return False
         muxed_conn = getattr(connection, "muxed_conn", None)
-        return self._is_quic_muxer(muxed_conn)
+        return self._is_native_muxer(muxed_conn)
 
     # Reference: `BasicHost.newStreamHandler` in Go.
     async def _swarm_stream_handler(self, net_stream: INetStream) -> None:
@@ -1061,6 +1061,18 @@ async def _swarm_stream_handler(self, net_stream: INetStream) -> None:
             await net_stream.reset()
             return
 
+        # Phase 2: Check handler connection requirements (if declared)
+        from libp2p.requirements import check_connection_requirements
+
+        underlying_conn = getattr(net_stream, "muxed_conn", None)
+        if not check_connection_requirements(handler, underlying_conn):
+            logger.warning(
+                "Handler for protocol %s has unmet connection requirements "
+                "on stream from peer %s — proceeding anyway",
+                protocol,
+                net_stream.muxed_conn.peer_id,
+            )
+
         await handler(net_stream)
 
     def get_live_peers(self) -> list[ID]:

libp2p/network/swarm.py

@@ -38,6 +38,7 @@
 )
 from libp2p.network.auto_connector import AutoConnector
 from libp2p.network.config import ConnectionConfig, RetryConfig
+from libp2p.transport.quic.config import QUICTransportConfig
 from libp2p.network.connection_gate import ConnectionGate
 from libp2p.network.connection_pruner import ConnectionPruner
 from libp2p.network.tag_store import TagInfo, TagStore
@@ -57,9 +58,6 @@
     OpenConnectionError,
     SecurityUpgradeFailure,
 )
-from libp2p.transport.quic.config import QUICTransportConfig
-from libp2p.transport.quic.connection import QUICConnection
-from libp2p.transport.quic.transport import QUICTransport
 from libp2p.transport.upgrader import (
     TransportUpgrader,
 )
@@ -200,13 +198,10 @@ async def run(self) -> None:
 
             # Set background nursery BEFORE setting the event
             # This ensures transports have the nursery when they check
-            if isinstance(self.transport, QUICTransport):
-                self.transport.set_background_nursery(nursery)
-                self.transport.set_swarm(self)
-            elif hasattr(self.transport, "set_background_nursery"):
-                # WebSocket transport also needs background nursery
-                # for connection management
+            if hasattr(self.transport, "set_background_nursery"):
                 self.transport.set_background_nursery(nursery)  # type: ignore[attr-defined]
+            if hasattr(self.transport, "set_swarm"):
+                self.transport.set_swarm(self)  # type: ignore[attr-defined]
 
             # Set event after background nursery is configured
             # This ensures transports have the nursery when they check the event
@@ -647,11 +642,11 @@ async def _dial_addr_single_attempt(self, addr: Multiaddr, peer_id: ID) -> INetC
                 pass
             raise SwarmException(f"Unexpected error dialing peer {peer_id}") from e
 
-        if isinstance(self.transport, QUICTransport) and isinstance(
-            raw_conn, IMuxedConn
+        if getattr(self.transport, 'provides_muxing', False) and getattr(
+            raw_conn, 'is_muxed', False
         ):
             logger.info(
-                "Skipping upgrade for QUIC, QUIC connections are already multiplexed"
+                "Skipping upgrade — transport already provides security + muxing"
             )
             try:
                 swarm_conn = await self.add_conn(raw_conn, direction="outbound")
@@ -886,7 +881,7 @@ async def new_stream(self, peer_id: ID) -> INetStream:
                     f"Failed to get a valid connection for peer {peer_id}"
                 )
 
-        if isinstance(self.transport, QUICTransport) and connection is not None:
+        if getattr(self.transport, 'provides_muxing', False) and connection is not None:
             conn = cast("SwarmConn", connection)
             try:
                 stream = await conn.new_stream()
@@ -1115,14 +1110,16 @@ async def conn_handler(
                             pass
                         return
 
-                # No need to upgrade QUIC Connection
-                if isinstance(self.transport, QUICTransport):
+                # No need to upgrade connections from transports with built-in muxing
+                if getattr(self.transport, 'provides_muxing', False):
                     try:
-                        quic_conn = cast(QUICConnection, read_write_closer)
-                        await self.add_conn(quic_conn, direction="inbound")
-                        peer_id = quic_conn.peer_id
+                        # The connection is already muxed; add it directly.
+                        muxed_conn = cast(IMuxedConn, read_write_closer)
+                        await self.add_conn(muxed_conn, direction="inbound")
+                        peer_id = muxed_conn.peer_id
                         logger.debug(
-                            f"successfully opened quic connection to peer {peer_id}"
+                            "successfully opened native-muxed connection "
+                            f"to peer {peer_id}"
                         )
                         # NOTE: This is a intentional barrier to prevent from the
                         # handler exiting and closing the connection.
@@ -1511,10 +1508,13 @@ async def add_conn(
         logger.debug("Swarm::add_conn | starting muxed connection")
         self.manager.run_task(muxed_conn.start)
         await muxed_conn.event_started.wait()
-        # For QUIC connections, also verify connection is established
-        if isinstance(muxed_conn, QUICConnection):
-            if not muxed_conn.is_established:
-                await muxed_conn._connected_event.wait()
+        # For connections that need an explicit handshake-completion step,
+        # wait until the connection reports itself as established.
+        if hasattr(muxed_conn, 'is_established') and hasattr(
+            muxed_conn, '_connected_event'
+        ):
+            if not muxed_conn.is_established:  # type: ignore[attr-defined]
+                await muxed_conn._connected_event.wait()  # type: ignore[attr-defined]
         logger.debug("Swarm::add_conn | starting swarm connection")
         self.manager.run_task(swarm_conn.start)
         await swarm_conn.event_started.wait()