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"