network-transport-tcp 0.6.0 → 0.6.1

raw patch · 5 files changed

+154/−39 lines, 5 filesdep +asyncdep +uuiddep ~containersdep ~networkPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: async, uuid

Dependency ranges changed: containers, network

API changes (from Hackage documentation)

Files

• ChangeLog +17/−1
• network-transport-tcp.cabal +8/−6
• src/Network/Transport/TCP.hs +83/−20
• src/Network/Transport/TCP/Internal.hs +21/−7
• tests/TestTCP.hs +25/−5

@@ -1,8 +1,24 @@+next release++*++2019-12-31 FacundoDominguez <facundo.dominguez@tweag.io> 0.7.0++* Added support for unaddressable endpoints. (#61)++2019-12-31 FacundoDominguez <facundo.dominguez@tweag.io> 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 <facundo.dominguez@tweag.io> 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)

@@ -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 <facundo.dominguez@tweag.io> 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

@@ -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                                                         --

@@ -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

@@ -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) <$>