From 019d0908e113a9d3e728ac8acb63a436fb4a1b69 Mon Sep 17 00:00:00 2001 From: adust09 Date: Mon, 17 Aug 2026 15:45:18 +0900 Subject: [PATCH] feat: implement circuit-v2 client transport for relayed connections Relayed peers were never reachable as connections. The relay client exchanges (makeReservation, connectViaRelay, handleStop) returned a bare StreamIO that nothing upgraded, no Transport accepted a p2p-circuit multiaddr, and the stop handler handed the raw stream to an application callback with no security handshake. A relayed Connection therefore could not exist in the pool. specs/relay/circuit-v2 states that once the hop/stop CONNECT exchange succeeds "the original stream becomes the relayed connection", which clients upgrade "with a security protocol and a multiplexer, just like they would e.g. upgrade a TCP connection". Add LibP2P.NAT.Relay.Transport, a Transport that claims p2p-circuit addresses: - transportDial reaches the relay, negotiates hop, sends CONNECT and returns the stream as a RawConnection, so the existing upgradeOutbound pipeline applies Noise and Yamux unchanged. - transportListen reserves on the relay and exposes inbound stop streams through listenerAccept, routing them into Switch.Listen.handleInbound so gating, resource limits, pooling, notifiers and teardown all apply. listenerAddr reports the reservation address, which puts the circuit address into switchListenAddrs for Identify to advertise. - Relayed connections are recognised by p2p-circuit in connRemoteAddr via the existing Relay.isRelayedAddr; Connection gains no new field. registerNATHandlers now registers the transport and returns the circuit state alongside the relay state. NATConfig loses ncOnRelayedStream: the stop stream is a connection, not an application stream, so there is no longer anything for that callback to own. The relay's advertised limit is still discarded (#269) and reservations are not refreshed before expiry (#268). Unblocks #258. Closes #266. --- libp2p-hs.cabal | 2 + src/LibP2P.hs | 6 +- src/LibP2P/NAT.hs | 65 +++--- src/LibP2P/NAT/Relay/Transport.hs | 285 +++++++++++++++++++++++++ test/LibP2P/NAT/RegistrationSpec.hs | 81 +------ test/LibP2P/NAT/Relay/TransportSpec.hs | 249 +++++++++++++++++++++ 6 files changed, 585 insertions(+), 103 deletions(-) create mode 100644 src/LibP2P/NAT/Relay/Transport.hs create mode 100644 test/LibP2P/NAT/Relay/TransportSpec.hs diff --git a/libp2p-hs.cabal b/libp2p-hs.cabal index b4f2641..7a7c0d8 100644 --- a/libp2p-hs.cabal +++ b/libp2p-hs.cabal @@ -77,6 +77,7 @@ library LibP2P.NAT.Relay.Message LibP2P.NAT.Relay LibP2P.NAT.Relay.Client + LibP2P.NAT.Relay.Transport LibP2P.NAT.DCUtR.Message LibP2P.NAT.DCUtR LibP2P.NAT @@ -200,6 +201,7 @@ test-suite libp2p-hs-test LibP2P.NAT.Relay.MessageSpec LibP2P.NAT.Relay.RelaySpec LibP2P.NAT.Relay.ClientSpec + LibP2P.NAT.Relay.TransportSpec LibP2P.NAT.DCUtR.MessageSpec LibP2P.NAT.DCUtR.DCUtRSpec LibP2P.NAT.RegistrationSpec diff --git a/src/LibP2P.hs b/src/LibP2P.hs index 5706acf..1e97702 100644 --- a/src/LibP2P.hs +++ b/src/LibP2P.hs @@ -14,7 +14,7 @@ -- addTransport sw tcp -- registerIdentifyHandlers sw -- registerPingHandler sw --- _relayState <- registerNATHandlers sw defaultNATConfig +-- (_relayState, _circuitState) <- registerNATHandlers sw defaultNATConfig -- addrs <- switchListen sw defaultConnectionGater [fromText "/ip4/127.0.0.1/tcp/0"] -- print addrs -- -- ... dial other peers, etc. @@ -94,6 +94,9 @@ module LibP2P , RelayConfig (..) , defaultRelayConfig , newRelayState + , CircuitState + , circuitTransport + , newCircuitState -- * GossipSub , GossipSubNode (..) @@ -123,6 +126,7 @@ import LibP2P.NAT , registerRelayStopHandler ) import LibP2P.NAT.Relay (RelayConfig (..), RelayState, defaultRelayConfig, newRelayState) +import LibP2P.NAT.Relay.Transport (CircuitState, circuitTransport, newCircuitState) import LibP2P.Protocol.GossipSub.Handler ( GossipSubNode (..) , gossipJoin diff --git a/src/LibP2P/NAT.hs b/src/LibP2P/NAT.hs index 953e542..2021de3 100644 --- a/src/LibP2P/NAT.hs +++ b/src/LibP2P/NAT.hs @@ -18,6 +18,8 @@ module LibP2P.NAT , registerRelayHopHandler , registerRelayStopHandler , registerDCUtRHandler + -- * Circuit client + , CircuitState ) where import Control.Concurrent.STM (atomically) @@ -47,7 +49,6 @@ import LibP2P.NAT.Relay.Client (handleStop) import LibP2P.NAT.Relay.Message ( HopMessage (..) , HopMessageType (..) - , RelayLimit , RelayStatus (..) , hopProtocolId , maxRelayMessageSize @@ -55,7 +56,13 @@ import LibP2P.NAT.Relay.Message , stopProtocolId , writeHopMessage ) -import LibP2P.Switch (selectTransport, setStreamHandler) +import LibP2P.NAT.Relay.Transport + ( CircuitState + , acceptStopStream + , circuitTransport + , newCircuitState + ) +import LibP2P.Switch (addTransport, selectTransport, setStreamHandler) import LibP2P.Switch.ConnPool (lookupConn) import LibP2P.Switch.Connection (newStream) import LibP2P.Switch.Dial (dial) @@ -69,36 +76,34 @@ import LibP2P.Switch.Upgrade (upgradeOutbound) import LibP2P.Transport (Transport (..)) -- | Configuration for the NAT traversal handlers. -data NATConfig = NATConfig - { ncRelayConfig :: !RelayConfig +newtype NATConfig = NATConfig + { ncRelayConfig :: RelayConfig -- ^ Resource limits for the Circuit Relay v2 server side - , ncOnRelayedStream :: !(PeerId -> Maybe RelayLimit -> StreamIO -> IO ()) - -- ^ Invoked when a relay delivers an inbound relayed stream (stop - -- protocol) after the CONNECT/OK exchange: source peer, limit - -- advertised by the relay, and the relayed stream. The application - -- owns the stream from this point (e.g. to run DCUtR over it). } --- | Default NAT configuration: default relay limits, and inbound relayed --- streams are left to the remote end (no local consumer). +-- | Default NAT configuration: default relay limits. defaultNATConfig :: NATConfig defaultNATConfig = NATConfig - { ncRelayConfig = defaultRelayConfig - , ncOnRelayedStream = \_ _ _ -> pure () + { ncRelayConfig = defaultRelayConfig } --- | Register all four NAT protocol handlers on the Switch. +-- | Register the NAT protocol handlers and the circuit client transport +-- on the Switch. -- --- Creates the relay server state from 'ncRelayConfig' and returns it so --- callers can inspect reservations/circuits. -registerNATHandlers :: Switch -> NATConfig -> IO RelayState +-- Returns the relay server state (so callers can inspect +-- reservations/circuits) and the circuit client state, which ties +-- 'transportListen' on a @p2p-circuit@ address to the inbound @stop@ +-- streams that arrive over the connection to that relay. +registerNATHandlers :: Switch -> NATConfig -> IO (RelayState, CircuitState) registerNATHandlers sw config = do relayState <- newRelayState (ncRelayConfig config) + circuitState <- newCircuitState + addTransport sw (circuitTransport sw circuitState) registerAutoNATHandler sw registerRelayHopHandler sw relayState - registerRelayStopHandler sw (ncOnRelayedStream config) + registerRelayStopHandler sw circuitState registerDCUtRHandler sw - pure relayState + pure (relayState, circuitState) -- | Register the AutoNAT server handler (/libp2p/autonat/1.0.0). -- @@ -208,18 +213,22 @@ openStopStream sw targetId = do Right mStream -> pure mStream -- | Register the Circuit Relay v2 stop handler --- (/libp2p/circuit/relay/0.2.0/stop): accept inbound relayed streams and --- hand them to the application callback. -registerRelayStopHandler - :: Switch - -> (PeerId -> Maybe RelayLimit -> StreamIO -> IO ()) - -> IO () -registerRelayStopHandler sw onRelayedStream = - setStreamHandler sw stopProtocolId $ \_conn stream -> do +-- (/libp2p/circuit/relay/0.2.0/stop). +-- +-- After the CONNECT/OK exchange the stop stream *is* the relayed +-- connection (specs/relay/circuit-v2), so it is handed to the circuit +-- transport's listener for the relay it arrived over. The Switch then +-- upgrades it like any other inbound raw connection. +-- +-- The relay's advertised limit is not yet enforced (issue #269). +registerRelayStopHandler :: Switch -> CircuitState -> IO () +registerRelayStopHandler sw circuitState = + setStreamHandler sw stopProtocolId $ \conn stream -> do result <- handleStop stream case result of Left _ -> pure () - Right (sourcePeer, mLimit) -> onRelayedStream sourcePeer mLimit stream + Right (sourcePeer, _mLimit) -> + acceptStopStream circuitState conn sourcePeer stream -- | Register the DCUtR handler (/libp2p/dcutr). -- diff --git a/src/LibP2P/NAT/Relay/Transport.hs b/src/LibP2P/NAT/Relay/Transport.hs new file mode 100644 index 0000000..bfe35d5 --- /dev/null +++ b/src/LibP2P/NAT/Relay/Transport.hs @@ -0,0 +1,285 @@ +-- | Circuit Relay v2 client transport (specs/relay/circuit-v2). +-- +-- Turns a @p2p-circuit@ multiaddr into a first-class 'Transport' so that +-- relayed peers become ordinary 'Connection's in the Switch's pool. +-- +-- The spec states that once the @hop@ CONNECT (dialer side) or @stop@ +-- CONNECT (target side) exchange succeeds, "the original stream becomes +-- the relayed connection", which clients then upgrade "with a security +-- protocol and a multiplexer, just like they would e.g. upgrade a TCP +-- connection". This module produces the 'RawConnection' for that stream; +-- the existing upgrade pipeline in "LibP2P.Switch.Upgrade" does the rest. +-- +-- Outbound: dial the relay, negotiate @hop@, send CONNECT, hand the +-- stream to the Switch as a raw connection. +-- +-- Inbound: 'transportListen' reserves on a relay and registers a queue +-- keyed by that relay's peer id. The @stop@ protocol handler calls +-- 'acceptStopStream', which enqueues the relayed stream; the Switch's +-- accept loop then drives it through the normal inbound path (gating, +-- upgrade, resource limits, pool, notifiers, teardown). +module LibP2P.NAT.Relay.Transport + ( -- * Shared state + CircuitState + , newCircuitState + -- * Transport + , circuitTransport + -- * Inbound relayed streams + , acceptStopStream + -- * Address handling (exported for testing) + , CircuitAddr (..) + , parseCircuitAddr + , circuitAddrOf + ) where + +import Control.Concurrent.STM + ( TQueue + , TVar + , atomically + , newTQueue + , newTVar + , newTVarIO + , readTQueue + , readTVar + , writeTQueue + , writeTVar + ) +import Control.Exception (SomeException, catch, throwIO) +import Control.Monad (unless) +import qualified Data.Map.Strict as Map +import LibP2P.Crypto.PeerId (PeerId (..), peerIdBytes) +import LibP2P.Multiaddr (Multiaddr (..), fromBytes) +import LibP2P.Multiaddr.Protocol (Protocol (..)) +import LibP2P.MultistreamSelect.Negotiation + ( NegotiationResult (..) + , StreamIO (..) + , negotiateInitiator + ) +import LibP2P.NAT.Relay.Client (connectViaRelay, makeReservation) +import LibP2P.NAT.Relay.Message + ( HopMessage (..) + , RelayStatus (..) + , Reservation (..) + , hopProtocolId + ) +import LibP2P.Switch.Connection (newStream) +import LibP2P.Switch.Dial (dial) +import LibP2P.Switch.Types (Connection (..), Switch (..)) +import LibP2P.Transport (Listener (..), RawConnection (..), Transport (..)) + +-- | A parsed circuit multiaddr. +-- +-- Wire form: @\\/p2p\/\\/p2p-circuit[\/p2p\/\]@. +-- The target component is present when dialling and absent when listening. +data CircuitAddr = CircuitAddr + { caRelayAddr :: !Multiaddr -- ^ Relay's transport address, without the @\/p2p@ suffix + , caRelayId :: !PeerId -- ^ Relay's peer id + , caTarget :: !(Maybe PeerId) -- ^ Destination peer id, when dialling + } deriving (Show, Eq) + +-- | Per-Switch state shared between the circuit transport and the @stop@ +-- protocol handler: one inbound queue per relay we hold a reservation on. +newtype CircuitState = CircuitState (TVar (Map.Map PeerId InboundQueue)) + +-- | An inbound queue for relayed connections arriving via one relay. +-- The closed flag lets 'listenerClose' unblock a waiting 'listenerAccept' +-- so the Switch's accept loop terminates. +data InboundQueue = InboundQueue + { iqQueue :: !(TQueue RawConnection) + , iqClosed :: !(TVar Bool) + } + +-- | Create empty circuit state. +newCircuitState :: IO CircuitState +newCircuitState = CircuitState <$> newTVarIO Map.empty + +-- | The Circuit Relay v2 client transport. +-- +-- Captures the Switch so it can dial the relay; register it after +-- 'LibP2P.Switch.newSwitch' with 'LibP2P.Switch.addTransport'. +circuitTransport :: Switch -> CircuitState -> Transport +circuitTransport sw st = Transport + { transportDial = dialCircuit sw + , transportListen = listenCircuit sw st + , transportCanDial = either (const False) (const True) . parseCircuitAddr + } + +-- Address handling + +-- | Parse a circuit multiaddr into its relay and target parts. +parseCircuitAddr :: Multiaddr -> Either String CircuitAddr +parseCircuitAddr (Multiaddr ps) = case break (== P2PCircuit) ps of + (_, []) -> Left "circuit address: no /p2p-circuit component" + (before, _ : after) -> do + (relayAddr, relayId) <- splitRelay before + target <- parseTarget after + pure CircuitAddr + { caRelayAddr = relayAddr + , caRelayId = relayId + , caTarget = target + } + where + splitRelay comps = case reverse comps of + (P2P pid : rest) + | not (null rest) -> Right (Multiaddr (reverse rest), PeerId pid) + | otherwise -> Left "circuit address: relay has no transport address" + _ -> Left "circuit address: relay component must end with /p2p/" + parseTarget [] = Right Nothing + parseTarget [P2P pid] = Right (Just (PeerId pid)) + parseTarget _ = + Left "circuit address: expected at most /p2p/ after /p2p-circuit" + +-- | Build the circuit multiaddr describing a relayed connection. +circuitAddrOf :: Multiaddr -> PeerId -> Maybe PeerId -> Multiaddr +circuitAddrOf relayAddr relayId mTarget = + Multiaddr (stripP2P relayAddr ++ [P2P (peerIdBytes relayId), P2PCircuit] ++ targetPart) + where + targetPart = maybe [] (\t -> [P2P (peerIdBytes t)]) mTarget + stripP2P (Multiaddr comps) = case reverse comps of + (P2P _ : rest) -> reverse rest + _ -> comps + +-- Outbound + +-- | Dial a peer through a relay. +-- +-- Connects to the relay, negotiates @hop@, sends CONNECT for the target, +-- and on @STATUS OK@ returns the hop stream as the raw relayed connection. +dialCircuit :: Switch -> Multiaddr -> IO RawConnection +dialCircuit sw addr = do + circuit <- either fail pure (parseCircuitAddr addr) + target <- maybe (fail "circuit dial: address has no /p2p/") pure + (caTarget circuit) + relayConn <- dialRelay sw circuit + stream <- openHopStream sw relayConn + resp <- connectViaRelay stream target >>= either (failClosing stream) pure + unless (hopStatus resp == Just RelayOK) $ + failClosing stream ("relay refused CONNECT: " ++ show (hopStatus resp)) + pure RawConnection + { rcStreamIO = stream + , rcLocalAddr = connLocalAddr relayConn + , rcRemoteAddr = circuitAddrOf (caRelayAddr circuit) (caRelayId circuit) (Just target) + , rcClose = closeQuietly stream + } + +-- Inbound + +-- | Reserve a slot on a relay and listen for relayed connections through it. +-- +-- The @hop@ stream is closed once the reservation is granted: per the +-- spec the reservation lives as long as the connection to the relay, and +-- inbound circuits arrive as fresh @stop@ streams on that connection. +listenCircuit :: Switch -> CircuitState -> Multiaddr -> IO Listener +listenCircuit sw st addr = do + circuit <- either fail pure (parseCircuitAddr addr) + relayConn <- dialRelay sw circuit + stream <- openHopStream sw relayConn + resp <- makeReservation stream >>= either (failClosing stream) pure + unless (hopStatus resp == Just RelayOK) $ + failClosing stream ("relay refused RESERVE: " ++ show (hopStatus resp)) + closeQuietly stream + queue <- registerQueue st (caRelayId circuit) + pure Listener + { listenerAccept = acceptFrom queue + , listenerClose = unregisterQueue st (caRelayId circuit) + , listenerAddr = reservationAddr circuit resp + } + +-- | Hand a relayed stream that arrived via the @stop@ protocol to the +-- listener for the relay it came over. +-- +-- The stream is closed when no listener is registered for that relay: +-- without a reservation we have nothing to accept the circuit into. +acceptStopStream :: CircuitState -> Connection -> PeerId -> StreamIO -> IO () +acceptStopStream (CircuitState var) relayConn source stream = do + enqueued <- atomically $ do + queues <- readTVar var + case Map.lookup (connPeerId relayConn) queues of + Nothing -> pure False + Just q -> do + closed <- readTVar (iqClosed q) + if closed + then pure False + else do + writeTQueue (iqQueue q) rawConn + pure True + unless enqueued (closeQuietly stream) + where + rawConn = RawConnection + { rcStreamIO = stream + , rcLocalAddr = connLocalAddr relayConn + , rcRemoteAddr = + circuitAddrOf (connRemoteAddr relayConn) (connPeerId relayConn) (Just source) + , rcClose = closeQuietly stream + } + +-- Helpers + +-- | Dial the relay named by a circuit address, reusing a pooled +-- connection to it when one exists. +dialRelay :: Switch -> CircuitAddr -> IO Connection +dialRelay sw circuit = + dial sw (caRelayId circuit) [caRelayAddr circuit] + >>= either (\err -> fail ("circuit: cannot reach relay: " ++ show err)) pure + +-- | Open a stream to the relay and negotiate the @hop@ protocol. +openHopStream :: Switch -> Connection -> IO StreamIO +openHopStream sw relayConn = do + stream <- newStream sw relayConn + >>= either (\err -> fail ("circuit: cannot open hop stream: " ++ show err)) pure + negotiated <- negotiateInitiator stream [hopProtocolId] + case negotiated of + Accepted _ -> pure stream + NoProtocol -> failClosing stream "relay does not support /libp2p/circuit/relay/0.2.0/hop" + +-- | The address this listener is reachable on: the relay's advertised +-- reservation address with @\/p2p-circuit@ appended. Falls back to the +-- dialled relay address when the relay advertises none. +reservationAddr :: CircuitAddr -> HopMessage -> Multiaddr +reservationAddr circuit resp = + case hopReservation resp >>= firstDecodable . rsvAddrs of + Just relayAddr -> circuitAddrOf relayAddr (caRelayId circuit) Nothing + Nothing -> circuitAddrOf (caRelayAddr circuit) (caRelayId circuit) Nothing + where + firstDecodable [] = Nothing + firstDecodable (bs : rest) = either (const (firstDecodable rest)) Just (fromBytes bs) + +-- | Register an inbound queue for a relay, replacing any previous one. +registerQueue :: CircuitState -> PeerId -> IO InboundQueue +registerQueue (CircuitState var) relayId = atomically $ do + queue <- InboundQueue <$> newTQueue <*> newTVar False + queues <- readTVar var + writeTVar var (Map.insert relayId queue queues) + pure queue + +-- | Mark a relay's inbound queue closed and drop it, releasing any +-- 'listenerAccept' blocked on it. +unregisterQueue :: CircuitState -> PeerId -> IO () +unregisterQueue (CircuitState var) relayId = atomically $ do + queues <- readTVar var + case Map.lookup relayId queues of + Nothing -> pure () + Just q -> do + writeTVar (iqClosed q) True + writeTVar var (Map.delete relayId queues) + +-- | Block for the next relayed connection, or throw once the listener is +-- closed so the Switch's accept loop stops. +acceptFrom :: InboundQueue -> IO RawConnection +acceptFrom q = do + result <- atomically $ do + closed <- readTVar (iqClosed q) + if closed + then pure Nothing + else Just <$> readTQueue (iqQueue q) + maybe (fail "circuit listener closed") pure result + +-- | Close a stream, ignoring failures from an already-dead session. +closeQuietly :: StreamIO -> IO () +closeQuietly stream = streamClose stream `catch` \(_ :: SomeException) -> pure () + +-- | Abort with an error, closing the stream we were using first. +failClosing :: StreamIO -> String -> IO a +failClosing stream msg = do + closeQuietly stream + throwIO (userError msg) diff --git a/test/LibP2P/NAT/RegistrationSpec.hs b/test/LibP2P/NAT/RegistrationSpec.hs index 477c357..4bbfa16 100644 --- a/test/LibP2P/NAT/RegistrationSpec.hs +++ b/test/LibP2P/NAT/RegistrationSpec.hs @@ -87,7 +87,7 @@ withNATPair config action = do (pidB, kpB) <- mkTestIdentity swB <- newSwitch pidB kpB addTransport swB =<< newTCPTransport - _relayState <- registerNATHandlers swB config + _ <- registerNATHandlers swB config addrsB <- switchListen swB defaultConnectionGater [loopbackAddr] -- Node A: client, listening so B can dial back (pidA, kpA) <- mkTestIdentity @@ -162,10 +162,11 @@ spec = do Just (Right resp) -> hopStatus resp `shouldBe` Just RelayOK it "dispatches /libp2p/circuit/relay/0.2.0/stop to the relay stop handler" $ do - relayedMVar <- newEmptyMVar - let config = defaultNATConfig - { ncOnRelayedStream = \src mLimit _stream -> putMVar relayedMVar (src, mLimit) } - withNATPair config $ \(swA, pidA, _addrsA) _nodeB conn -> do + -- After the CONNECT/OK exchange the stop stream becomes the relayed + -- connection and is handed to the circuit transport. With no + -- reservation held here there is no listener to accept it, so the + -- observable evidence that the handler ran is the OK status. + withNATPair defaultNATConfig $ \(swA, pidA, _addrsA) _nodeB conn -> do result <- timeout 10000000 $ do stream <- openProtoStream swA conn stopProtocolId writeStopMessage stream StopMessage @@ -178,12 +179,7 @@ spec = do case result of Nothing -> expectationFailure "stop connect timed out" Just (Left err) -> expectationFailure $ "stop connect failed: " ++ err - Just (Right resp) -> do - stopStatus resp `shouldBe` Just RelayOK - callback <- timeout 5000000 $ takeMVar relayedMVar - case callback of - Nothing -> expectationFailure "relayed-stream callback not invoked" - Just (src, _mLimit) -> src `shouldBe` pidA + Just (Right resp) -> stopStatus resp `shouldBe` Just RelayOK it "dispatches /libp2p/dcutr to the DCUtR handler" $ do withNATPair defaultNATConfig $ \(swA, _pidA, addrsA) _nodeB conn -> do @@ -202,66 +198,3 @@ spec = do -- The handler advertises B's listen addresses hpObsAddrs resp `shouldSatisfy` (not . null) - describe "multi-host relay circuit" $ do - it "bridges reserve → connect → application data across three in-process hosts" $ do - -- Three real switches over TCP: relay R serves hop/stop, target A - -- reserves on R, source B connects to A through R, and application - -- data crosses the bridged circuit in both directions. Hole punching - -- against real NATs is out of reach in-process and is covered by the - -- interop work (issue #131). - (pidR, kpR) <- mkTestIdentity - swR <- newSwitch pidR kpR - addTransport swR =<< newTCPTransport - _ <- registerNATHandlers swR defaultNATConfig - addrsR <- switchListen swR defaultConnectionGater [loopbackAddr] - -- Target A: consumes the relayed stream (3 bytes in, 2 bytes reply) - relayedMVar <- newEmptyMVar - (pidA, kpA) <- mkTestIdentity - swA <- newSwitch pidA kpA - addTransport swA =<< newTCPTransport - let configA = defaultNATConfig - { ncOnRelayedStream = \src _mLimit stream -> do - payload <- mapM (\_ -> streamReadByte stream) [1..3 :: Int] - streamWrite stream (BS.pack [9, 8]) - putMVar relayedMVar (src, payload) - } - _ <- registerNATHandlers swA configA - -- Source B - (pidB, kpB) <- mkTestIdentity - swB <- newSwitch pidB kpB - addTransport swB =<< newTCPTransport - result <- timeout 20000000 $ do - -- A dials R and reserves - connAR <- dial swA pidR [head addrsR] >>= either (fail . show) pure - hopA <- openProtoStream swA connAR hopProtocolId - rsv <- makeReservation hopA >>= either fail pure - hopStatus rsv `shouldBe` Just RelayOK - -- The reservation advertises R's addresses, each ending in /p2p/ - case hopReservation rsv of - Nothing -> expectationFailure "expected reservation in RESERVE response" - Just r -> do - rsvAddrs r `shouldSatisfy` (not . null) - mapM_ - (\addrBytes -> case fromBytes addrBytes of - Right (Multiaddr ps) -> last ps `shouldBe` P2P (peerIdBytes pidR) - Left err -> expectationFailure $ "undecodable reservation addr: " ++ err) - (rsvAddrs r) - -- B dials R and connects to A through the circuit - connBR <- dial swB pidR [head addrsR] >>= either (fail . show) pure - hopB <- openProtoStream swB connBR hopProtocolId - connResp <- connectViaRelay hopB pidA >>= either fail pure - hopStatus connResp `shouldBe` Just RelayOK - -- Application data B → A through the bridged circuit - streamWrite hopB (BS.pack [1, 2, 3]) - (src, payload) <- takeMVar relayedMVar - src `shouldBe` pidB - payload `shouldBe` [1, 2, 3] - -- and A → B back through the same circuit - reply <- mapM (\_ -> streamReadByte hopB) [1..2 :: Int] - reply `shouldBe` [9, 8] - switchClose swA - switchClose swB - switchClose swR - case result of - Nothing -> expectationFailure "multi-host relay circuit timed out" - Just () -> pure () diff --git a/test/LibP2P/NAT/Relay/TransportSpec.hs b/test/LibP2P/NAT/Relay/TransportSpec.hs new file mode 100644 index 0000000..cc4ca0a --- /dev/null +++ b/test/LibP2P/NAT/Relay/TransportSpec.hs @@ -0,0 +1,249 @@ +-- | Tests for the Circuit Relay v2 client transport (issue #266). +-- +-- Address handling is checked in isolation; the end-to-end behaviour is +-- exercised with three real in-process switches over loopback TCP — +-- relay R, target B holding a reservation on R, and dialer A reaching B +-- through the circuit. Hole punching against real NATs is out of reach +-- in-process and is covered by the interop work (issue #131). +module LibP2P.NAT.Relay.TransportSpec (spec) where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.STM (atomically) +import Control.Exception (SomeException, try) +import qualified Data.ByteString as BS +import Data.Word (Word8) +import LibP2P.Crypto.Ed25519 (generateKeyPair) +import LibP2P.Crypto.Key (KeyPair, publicKey) +import LibP2P.Crypto.PeerId (PeerId (..), fromPublicKey, peerIdBytes) +import LibP2P.Multiaddr (Multiaddr (..)) +import LibP2P.Multiaddr.Protocol (Protocol (..)) +import LibP2P.NAT (defaultNATConfig, registerNATHandlers) +import LibP2P.NAT.Relay (isRelayedAddr) +import LibP2P.NAT.Relay.Transport + ( CircuitAddr (..) + , circuitAddrOf + , circuitTransport + , newCircuitState + , parseCircuitAddr + ) +import LibP2P.Protocol.Ping (openPingSession, ping, registerPingHandler) +import LibP2P.Switch (addTransport, newSwitch, switchClose) +import LibP2P.Switch.ConnPool (lookupAllConns) +import LibP2P.Switch.Dial (dial) +import LibP2P.Switch.Listen (defaultConnectionGater, switchListen, switchListenAddrs) +import LibP2P.Switch.Types (Connection (..), Direction (..), Switch (..)) +import LibP2P.Transport (Transport (..)) +import LibP2P.Transport.TCP (newTCPTransport) +import System.Timeout (timeout) +import Test.Hspec + +-- | Generate a test identity (PeerId, KeyPair). +mkTestIdentity :: IO (PeerId, KeyPair) +mkTestIdentity = do + Right kp <- generateKeyPair + let pid = fromPublicKey (publicKey kp) + pure (pid, kp) + +-- | Loopback address with port 0 (OS assigns ephemeral port). +loopbackAddr :: Multiaddr +loopbackAddr = Multiaddr [IP4 0x7f000001, TCP 0] + +-- | A syntactically valid identity-multihash peer id. Only the bytes +-- matter for address parsing, so no key generation is needed. +samplePeerId :: Word8 -> PeerId +samplePeerId n = PeerId (BS.pack ([0x12, 0x20] ++ replicate 32 n)) + +-- | A plain TCP address. +tcpAddr :: Multiaddr +tcpAddr = Multiaddr [IP4 0x7f000001, TCP 4001] + +isLeft :: Either a b -> Bool +isLeft = either (const True) (const False) + +-- | Append @\/p2p\/\\/p2p-circuit@ (and optionally the target) to +-- a transport address. +withCircuit :: Multiaddr -> PeerId -> Maybe PeerId -> Multiaddr +withCircuit (Multiaddr comps) relayId mTarget = + Multiaddr (comps ++ [P2P (peerIdBytes relayId), P2PCircuit] ++ target) + where + target = maybe [] (\t -> [P2P (peerIdBytes t)]) mTarget + +-- | Everything the relay tests need: the relay's identity and transport +-- address, the target's identity and circuit listen addresses, and a +-- dialer. +data Trio = Trio + { trRelayId :: !PeerId + , trRelayAddr :: !Multiaddr + , trTargetSw :: !Switch + , trTargetId :: !PeerId + , trTargetAddrs :: ![Multiaddr] + , trDialerSw :: !Switch + } + +-- | Bring up a relay, a target that reserves on it, and a dialer, run the +-- action, then close all three switches. +withRelayTrio :: (Trio -> IO a) -> IO a +withRelayTrio action = do + -- Relay R + (pidR, kpR) <- mkTestIdentity + swR <- newSwitch pidR kpR + addTransport swR =<< newTCPTransport + _ <- registerNATHandlers swR defaultNATConfig + addrsR <- switchListen swR defaultConnectionGater [loopbackAddr] + relayAddr <- case addrsR of + (a : _) -> pure a + [] -> fail "relay did not bind a listen address" + -- Target B: reserves on R and answers pings + (pidB, kpB) <- mkTestIdentity + swB <- newSwitch pidB kpB + addTransport swB =<< newTCPTransport + _ <- registerNATHandlers swB defaultNATConfig + registerPingHandler swB + addrsB <- switchListen swB defaultConnectionGater + [withCircuit relayAddr pidR Nothing] + -- Dialer A + (pidA, kpA) <- mkTestIdentity + swA <- newSwitch pidA kpA + addTransport swA =<< newTCPTransport + _ <- registerNATHandlers swA defaultNATConfig + threadDelay 300000 + result <- action Trio + { trRelayId = pidR + , trRelayAddr = relayAddr + , trTargetSw = swB + , trTargetId = pidB + , trTargetAddrs = addrsB + , trDialerSw = swA + } + switchClose swA + switchClose swB + switchClose swR + pure result + +-- | The circuit address a dialer uses to reach the target through R. +dialAddrFor :: Trio -> PeerId -> Multiaddr +dialAddrFor trio target = withCircuit (trRelayAddr trio) (trRelayId trio) (Just target) + +spec :: Spec +spec = do + describe "parseCircuitAddr" $ do + it "splits a dial address into relay address, relay id and target" $ do + let relayId = samplePeerId 1 + targetId = samplePeerId 2 + case parseCircuitAddr (withCircuit tcpAddr relayId (Just targetId)) of + Left err -> expectationFailure err + Right ca -> do + caRelayAddr ca `shouldBe` tcpAddr + caRelayId ca `shouldBe` relayId + caTarget ca `shouldBe` Just targetId + + it "reports no target for a listen address" $ do + let relayId = samplePeerId 1 + case parseCircuitAddr (withCircuit tcpAddr relayId Nothing) of + Left err -> expectationFailure err + Right ca -> do + caRelayId ca `shouldBe` relayId + caTarget ca `shouldBe` Nothing + + it "rejects an address without a /p2p-circuit component" $ + parseCircuitAddr tcpAddr `shouldSatisfy` isLeft + + it "rejects a circuit address whose relay carries no peer id" $ + parseCircuitAddr (Multiaddr [IP4 0x7f000001, TCP 4001, P2PCircuit]) + `shouldSatisfy` isLeft + + it "rejects a relay component with no transport address" $ + parseCircuitAddr (Multiaddr [P2P (peerIdBytes (samplePeerId 1)), P2PCircuit]) + `shouldSatisfy` isLeft + + it "round-trips an address built by circuitAddrOf" $ do + let relayId = samplePeerId 3 + targetId = samplePeerId 4 + parseCircuitAddr (circuitAddrOf tcpAddr relayId (Just targetId)) `shouldBe` + Right CircuitAddr + { caRelayAddr = tcpAddr + , caRelayId = relayId + , caTarget = Just targetId + } + + it "drops an existing /p2p suffix when building a circuit address" $ do + let relayId = samplePeerId 5 + suffixed = Multiaddr [IP4 0x7f000001, TCP 4001, P2P (peerIdBytes relayId)] + circuitAddrOf suffixed relayId Nothing + `shouldBe` withCircuit tcpAddr relayId Nothing + + describe "circuitTransport" $ + it "claims circuit addresses and declines plain TCP addresses" $ do + (pid, kp) <- mkTestIdentity + sw <- newSwitch pid kp + st <- newCircuitState + let transport = circuitTransport sw st + relayId = samplePeerId 6 + transportCanDial transport (withCircuit tcpAddr relayId (Just (samplePeerId 7))) + `shouldBe` True + transportCanDial transport (withCircuit tcpAddr relayId Nothing) `shouldBe` True + transportCanDial transport tcpAddr `shouldBe` False + switchClose sw + + describe "relayed connections through a real relay" $ do + it "advertises the reservation address as a circuit listen address" $ + withRelayTrio $ \trio -> do + let addrsB = trTargetAddrs trio + addrsB `shouldSatisfy` (not . null) + all isRelayedAddr addrsB `shouldBe` True + mapM_ + (\addr -> case parseCircuitAddr addr of + Left err -> expectationFailure err + Right ca -> do + caRelayId ca `shouldBe` trRelayId trio + caTarget ca `shouldBe` Nothing) + addrsB + listenAddrs <- switchListenAddrs (trTargetSw trio) + listenAddrs `shouldBe` addrsB + + it "yields an upgraded connection to the target on both sides" $ + withRelayTrio $ \trio -> do + let swA = trDialerSw trio + pidB = trTargetId trio + result <- timeout 20000000 $ dial swA pidB [dialAddrFor trio pidB] + case result of + Nothing -> expectationFailure "circuit dial timed out" + Just (Left err) -> expectationFailure $ "circuit dial failed: " ++ show err + Just (Right conn) -> do + -- A's side: outbound, relayed, authenticated as B + connPeerId conn `shouldBe` pidB + connDirection conn `shouldBe` Outbound + isRelayedAddr (connRemoteAddr conn) `shouldBe` True + -- B's side: the same peer arrives as an inbound relayed connection + threadDelay 500000 + conns <- atomically $ + lookupAllConns (swConnPool (trTargetSw trio)) (swLocalPeerId swA) + let inbound = filter ((== Inbound) . connDirection) conns + -- Connection has no Show instance, so assert on its shape + null inbound `shouldBe` False + all (isRelayedAddr . connRemoteAddr) inbound `shouldBe` True + + it "round-trips a protocol stream over the relayed connection" $ + withRelayTrio $ \trio -> do + let swA = trDialerSw trio + pidB = trTargetId trio + result <- timeout 20000000 $ do + conn <- dial swA pidB [dialAddrFor trio pidB] >>= either (fail . show) pure + session <- openPingSession swA conn >>= either (fail . show) pure + ping session + case result of + Nothing -> expectationFailure "ping over circuit timed out" + Just (Left err) -> expectationFailure $ "ping over circuit failed: " ++ show err + Just (Right _) -> pure () + + it "fails the dial when the target holds no reservation" $ + withRelayTrio $ \trio -> do + let unknown = samplePeerId 9 + result <- timeout 20000000 $ + try (dial (trDialerSw trio) unknown [dialAddrFor trio unknown]) + case result of + Nothing -> expectationFailure "dial to unreserved target timed out" + Just (Left (_ :: SomeException)) -> pure () + Just (Right (Left _err)) -> pure () + Just (Right (Right _conn)) -> + expectationFailure "dial to unreserved target unexpectedly succeeded"