diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,8 +1,24 @@ +next release + +* + +2019-12-31 FacundoDominguez 0.7.0 + +* Added support for unaddressable endpoints. (#61) + +2019-12-31 FacundoDominguez 0.6.1 + +* Relax dependency bounds to build with ghc-8.6.5 +* apiSend RELY violation is removed for closed remote endpoints (#70) +* The server no longer needs crash if accept throws an exception. +* Check peer-reported host against socket host (#54) +* Fix possible endless waiting on the 'crossed' MVar (#74) +* Fix possible msg corruption on a busy network (#85) + 2017-08-21 FacundoDominguez 0.6.0 * Implemented protocol versioning (#55) * Extend interface so queue policies and lengths can be configured. -* Check peer-reported host against socket host (#54) * Test improvements * Fix races when an EndPoint is closed or failed (#60) * Fix timeout socket connections (#53) diff --git a/network-transport-tcp.cabal b/network-transport-tcp.cabal --- a/network-transport-tcp.cabal +++ b/network-transport-tcp.cabal @@ -1,5 +1,5 @@ Name: network-transport-tcp -Version: 0.6.0 +Version: 0.6.1 Cabal-Version: >=1.10 Build-Type: Simple License: BSD3 @@ -9,7 +9,7 @@ maintainer: Facundo Domínguez Stability: experimental Homepage: http://haskell-distributed.github.com -Bug-Reports: https://cloud-haskell.atlassian.net/browse/NTTCP +Bug-Reports: https://github.com/haskell-distributed/network-transport-tcp/issues Synopsis: TCP instantiation of Network.Transport Description: TCP instantiation of Network.Transport Tested-With: GHC==7.6.3 GHC==7.8.4 GHC==7.10.3 @@ -26,11 +26,13 @@ Library Build-Depends: base >= 4.3 && < 5, + async >= 2.2 && < 2.3, network-transport >= 0.5 && < 0.6, data-accessor >= 0.2 && < 0.3, - containers >= 0.4 && < 0.6, + containers >= 0.4 && < 0.7, bytestring >= 0.9 && < 0.11, - network >= 2.6.2 && < 2.7 + network >= 2.6.2 && < 2.9, + uuid >= 1.3 && < 1.4 Exposed-modules: Network.Transport.TCP, Network.Transport.TCP.Internal Default-Extensions: CPP @@ -43,13 +45,13 @@ Exposed-modules: Network.Transport.TCP.Mock.Socket Network.Transport.TCP.Mock.Socket.ByteString -Test-Suite TestTCP +Test-Suite TestTCP Type: exitcode-stdio-1.0 Main-Is: TestTCP.hs Build-Depends: base >= 4.3 && < 5, bytestring >= 0.9 && < 0.11, network-transport-tests >= 0.2.1.0 && < 0.3, - network >= 2.3 && < 2.7, + network >= 2.3 && < 2.9, network-transport, network-transport-tcp ghc-options: -threaded -rtsopts -with-rtsopts=-N diff --git a/src/Network/Transport/TCP.hs b/src/Network/Transport/TCP.hs --- a/src/Network/Transport/TCP.hs +++ b/src/Network/Transport/TCP.hs @@ -14,6 +14,7 @@ {-# LANGUAGE RecursiveDo #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE RankNTypes #-} +{-# LANGUAGE LambdaCase #-} module Network.Transport.TCP ( -- * Main API @@ -59,7 +60,7 @@ , encodeWord32 , tryCloseSocket , tryShutdownSocketBoth - , decodeSockAddr + , resolveSockAddr , EndPointId , encodeEndPointAddress , decodeEndPointAddress @@ -119,11 +120,14 @@ , modifyMVar , modifyMVar_ , readMVar + , tryReadMVar , takeMVar , putMVar + , tryPutMVar , newEmptyMVar , withMVar ) +import Control.Concurrent.Async (async, wait) import Control.Category ((>>>)) import Control.Applicative ((<$>)) import Control.Monad (when, unless, join, mplus, (<=<)) @@ -141,6 +145,7 @@ , finally , catch , bracket + , mask , mask_ ) import Data.IORef (IORef, newIORef, writeIORef, readIORef, writeIORef) @@ -148,7 +153,7 @@ import qualified Data.ByteString as BS (concat) import qualified Data.ByteString.Char8 as BSC (pack, unpack) import Data.Bits (shiftL, (.|.)) -import Data.Maybe (isJust) +import Data.Maybe (isJust, isNothing, fromJust) import Data.Word (Word32) import Data.Set (Set) import qualified Data.Set as Set @@ -441,7 +446,12 @@ -- | When the connection is being probed, yields an IO action that can be -- used to release any resources dedicated to the probing. , remoteProbing :: Maybe (IO ()) - , remoteSendLock :: !(MVar ()) + -- | MVar protects the socket usage by the concurrent threads and + -- prohibits its usage after SomeException. + -- + -- Nothing allows the socket usage. @Just e@ is set on an + -- exception after which the socket should not be used (see 'sendOn'). + , remoteSendLock :: !(MVar (Maybe SomeException)) -- | An IO which returns when the socket (remoteSocket) has been closed. -- The program/thread which created the socket is always responsible -- for closing it, but sometimes other threads need to know when this @@ -509,6 +519,10 @@ -- could otherwise deny service to some victim by claiming the victim's -- address. , tcpCheckPeerHost :: Bool + -- | What to do if there's an exception when accepting a new TCP + -- connection. Throwing an exception here will cause the server to + -- terminate. + , tcpServerExceptionHandler :: SomeException -> IO () } -- | Internal functionality we expose for unit testing @@ -608,7 +622,7 @@ ) errorHandler :: TCPTransport -> SomeException -> IO () - errorHandler _ = throwIO + errorHandler _ = tcpServerExceptionHandler params terminationHandler :: TCPTransport -> SomeException -> IO () terminationHandler transport ex = do @@ -631,6 +645,7 @@ , tcpMaxAddressLength = maxBound , tcpMaxReceiveLength = maxBound , tcpCheckPeerHost = False + , tcpServerExceptionHandler = throwIO } -------------------------------------------------------------------------------- @@ -814,12 +829,18 @@ RemoteEndPointClosing _ _ -> do alive <- readIORef connAlive if alive - then relyViolation (ourEndPoint, theirEndPoint) "apiSend" + -- RemoteEndPointClosing is only entered by 'closeIfUnused', + -- which guarantees that there are no alive connections. + then relyViolation (ourEndPoint, theirEndPoint) "apiSend RemoteEndPointClosing" else throwIO $ TransportError SendClosed "Connection closed" RemoteEndPointClosed -> do alive <- readIORef connAlive if alive - then relyViolation (ourEndPoint, theirEndPoint) "apiSend" + -- This is normal. If the remote endpoint closes up while we have + -- an outgoing connection (CloseEndPoint or CloseSocket message), + -- we'll post the connection lost event but we won't update these + -- 'connAlive' IORefs. + then throwIO $ TransportError SendFailed "Remote endpoint closed" else throwIO $ TransportError SendClosed "Connection closed" RemoteEndPointFailed err -> do alive <- readIORef connAlive @@ -959,9 +980,10 @@ handleConnectionRequestV0 :: (N.Socket, N.SockAddr) -> IO (Maybe (IO ())) handleConnectionRequestV0 (sock, sockAddr) = do -- Get the OS-determined host and port. - (actualHost, actualPort) <- - decodeSockAddr sockAddr >>= + (numericHost, resolvedHost, actualPort) <- + resolveSockAddr sockAddr >>= maybe (throwIO (userError "handleConnectionRequest: invalid socket address")) return + (ourEndPointId, theirAddress) <- do ourEndPointId <- recvWord32 sock let maxAddressLength = tcpMaxAddressLength $ transportParams transport @@ -976,7 +998,7 @@ return (decodeEndPointAddress theirAddress) let checkPeerHost = tcpCheckPeerHost (transportParams transport) - if checkPeerHost && (theirHost /= actualHost) + if checkPeerHost && (theirHost /= resolvedHost) && (theirHost /= numericHost) then do -- If the OS-determined host doesn't match the host that the peer gave us, -- then we have no choice but to reject the connection. It's because we @@ -985,7 +1007,8 @@ -- claiming to have their host and port. sendMany sock $ encodeWord32 (encodeConnectionRequestResponse ConnectionRequestHostMismatch) - : prependLength [BSC.pack actualHost] + : (prependLength [BSC.pack theirHost] ++ prependLength [BSC.pack numericHost] ++ prependLength [BSC.pack resolvedHost]) + return Nothing else do ourEndPoint <- withMVar (transportState transport) $ \st -> case st of @@ -1014,7 +1037,7 @@ [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestCrossed)] probeIfValid theirEndPoint else do - sendLock <- newMVar () + sendLock <- newMVar Nothing let vst = ValidRemoteEndPointState { remoteSocket = sock , remoteSocketClosed = socketClosed @@ -1497,7 +1520,7 @@ -- (readMVar socketClosedVar), and we'll take care of closing it up -- once handleIncomingMessages has finished. Right (socketClosedVar, sock, ConnectionRequestAccepted) -> do - sendLock <- newMVar () + sendLock <- newMVar Nothing let vst = ValidRemoteEndPointState { remoteSocket = sock , remoteSocketClosed = readMVar socketClosedVar @@ -1536,8 +1559,19 @@ let handler :: SomeException -> IO (TransportError ConnectErrorCode) handler err = return (TransportError ConnectFailed (show err)) err <- handle handler $ do - actualHost <- recvWithLength (tcpMaxReceiveLength params) sock - return (TransportError ConnectFailed ("setupRemoteEndPoint: Host mismatch " ++ BSC.unpack (BS.concat actualHost))) + claimedHost <- recvWithLength (tcpMaxReceiveLength params) sock + actualNumericHost <- recvWithLength (tcpMaxReceiveLength params) sock + actualResolvedHost <- recvWithLength (tcpMaxReceiveLength params) sock + let reason = concat [ + "setupRemoteEndPoint: Host mismatch" + , ". Claimed: " + , BSC.unpack (BS.concat claimedHost) + , "; Numeric: " + , BSC.unpack (BS.concat actualNumericHost) + , "; Resolved: " + , BSC.unpack (BS.concat actualResolvedHost) + ] + return (TransportError ConnectFailed reason) resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err) tryCloseSocket sock `finally` putMVar socketClosedVar () return Nothing @@ -1655,8 +1689,11 @@ resolveInit :: EndPointPair -> RemoteState -> IO () resolveInit (ourEndPoint, theirEndPoint) newState = modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of - RemoteEndPointInit resolved _ _ -> do + RemoteEndPointInit resolved crossed _ -> do putMVar resolved () + -- Unblock the reader (if any) if the ConnectionRequestCrossed + -- message did not come within the connection timeout. + tryPutMVar crossed () case newState of RemoteEndPointClosed -> removeRemoteEndPoint (ourEndPoint, theirEndPoint) @@ -1822,9 +1859,12 @@ (RequestedByThem, RequestedByUs) -> if ourAddress > theirAddress then do - -- Wait for the Crossed message - readMVarTimeout mtimer crossed - return (theirEndPoint, True) + -- Wait for the Crossed message and recheck the state + -- of the remote endpoint after this (it may well be + -- invalid already in case of a timeout). + tryReadMVar crossed >>= \case + Nothing -> readMVarTimeout mtimer crossed >> go + _ -> return (theirEndPoint, True) else return (theirEndPoint, False) (RequestedByThem, RequestedByThem) -> @@ -1854,9 +1894,32 @@ const $ readMVar mv -- | Send a payload over a heavyweight connection (thread safe) +-- +-- The socket cannot be used for sending after the non-atomic 'sendMany' +-- is interrupted - otherwise, the other side may get the msg corrupted. +-- +-- There are two types of possible exceptions here: +-- 1) Outer asynchronous exceptions (like 'ProcessLinkException'). +-- 2) Synchronous exceptions (inner or outer). +-- On a synchronous exception the remote endpoint is failed (see 'runScheduledAction', +-- for example) and its socket is not supposed to be used again. +-- +-- With 'async' the code is run in a new thread which is not-targeted (and +-- thus, not interrupted) by the 1st type of exceptions. With 'remoteSendLock' +-- we protect the socket usage by the concurrent threads, as well as prevent +-- that usage after SomeException. sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO () -sendOn vst bs = withMVar (remoteSendLock vst) $ \() -> - sendMany (remoteSocket vst) bs +sendOn vst bs = (wait =<<) $ async $ + mask $ \restore -> do + let lock = remoteSendLock vst + maybeException <- takeMVar lock + when (isNothing maybeException) $ + restore (sendMany (remoteSocket vst) bs) `catch` \ex -> do + putMVar lock (Just ex) + throwIO ex + putMVar lock maybeException + forM_ maybeException $ \e -> + throwIO $ userError $ "sendOn failed earlier with: " ++ show e -------------------------------------------------------------------------------- -- Scheduling actions -- diff --git a/src/Network/Transport/TCP/Internal.hs b/src/Network/Transport/TCP/Internal.hs --- a/src/Network/Transport/TCP/Internal.hs +++ b/src/Network/Transport/TCP/Internal.hs @@ -13,7 +13,7 @@ , encodeWord32 , tryCloseSocket , tryShutdownSocketBoth - , decodeSockAddr + , resolveSockAddr , EndPointId , encodeEndPointAddress , decodeEndPointAddress @@ -61,6 +61,7 @@ , ShutdownCmd(ShutdownBoth) , SockAddr(..) , inet_ntoa + , getNameInfo ) #ifdef USE_MOCK_NETWORK @@ -305,13 +306,26 @@ then throwIO (userError "recvExact: Socket closed") else go (bs : acc) (l - fromIntegral (BS.length bs)) --- | Produce a HostName and ServiceName from a SockAddr. Only gives 'Just' for --- IPv4 addresses. -decodeSockAddr :: N.SockAddr -> IO (Maybe (N.HostName, N.ServiceName)) -decodeSockAddr sockAddr = case sockAddr of +-- | Get the numeric host, resolved host (via getNameInfo), and port from a +-- SockAddr. The numeric host is first, then resolved host (which may be the +-- same as the numeric host). +-- Will only give 'Just' for IPv4 addresses. +resolveSockAddr :: N.SockAddr -> IO (Maybe (N.HostName, N.HostName, N.ServiceName)) +resolveSockAddr sockAddr = case sockAddr of N.SockAddrInet port host -> do - hostString <- N.inet_ntoa host - return $ Just (hostString, show port) + (mResolvedHost, mResolvedPort) <- N.getNameInfo [] True False sockAddr + case (mResolvedHost, mResolvedPort) of + (Just resolvedHost, Nothing) -> do + numericHost <- N.inet_ntoa host + return $ Just (numericHost, resolvedHost, show port) + _ -> error $ concat [ + "decodeSockAddr: unexpected resolution " + , show sockAddr + , " -> " + , show mResolvedHost + , ", " + , show mResolvedPort + ] _ -> return Nothing -- | Encode end point address diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs --- a/tests/TestTCP.hs +++ b/tests/TestTCP.hs @@ -1015,8 +1015,8 @@ -- | Ensure that if the peer's claimed host doesn't match its actual host, -- the connection is rejected (when tcpCheckPeerHost is enabled). -testCheckPeerHost :: IO () -testCheckPeerHost = do +testCheckPeerHostReject :: IO () +testCheckPeerHostReject = do let params = defaultTCPParameters { tcpCheckPeerHost = True } Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") params @@ -1029,11 +1029,30 @@ Left err <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints - TransportError ConnectFailed "setupRemoteEndPoint: Host mismatch 127.0.0.1" - <- return err + TransportError ConnectFailed _ <- return err return () +-- | Ensure that if peer host checking works through name resolution: if the +-- peer claims "localhost", and connects to a transport also on localhost, +-- it should be accepted. +testCheckPeerHostResolve :: IO () +testCheckPeerHostResolve = do + + let params = defaultTCPParameters { tcpCheckPeerHost = True } + Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") params + -- EndPoints on this transport have addresses with "localhost" host part. + Right transport2 <- createTransport "127.0.0.1" "0" ((,) "localhost") defaultTCPParameters + + Right ep1 <- newEndPoint transport1 + Right ep2 <- newEndPoint transport2 + + Right conn <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints + + close conn + + return () + main :: IO () main = do tcpResult <- tryIO $ runTests @@ -1052,7 +1071,8 @@ , ("InvalidCloseConnection", testInvalidCloseConnection) , ("MaxLength", testMaxLength) , ("CloseEndPoint", testCloseEndPoint) - , ("CheckPeerHost", testCheckPeerHost) + , ("CheckPeerHostReject", testCheckPeerHostReject) + , ("CheckPeerHostResolve", testCheckPeerHostResolve) ] -- Run the generic tests even if the TCP specific tests failed.. testTransport (either (Left . show) (Right) <$>