diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,32 @@ # ChangeLog for http2 +## 5.4.13 + +* `Config` gains `confSendAllWith` and `confReadNWith`, a send and a read + which take an `STM` action to give up on. `allocSimpleConfigWith` fills + them with network's `sendAllWith` and `recvWith`. With them, a write the + peer does not read is given up on once it has taken the whole timeout, + and the receiver is stopped after a GOAWAY without being interrupted. +* `runWith` and `runIOWith` take an `STM` action. When it fires, the server + sends GOAWAY(NO_ERROR) with the largest stream identifier and a PING, then, + once the PING is answered or a second has passed, GOAWAY with the last + stream taken. Streams above it are refused with REFUSED_STREAM, and the + connection is closed when those below it are done (RFC 9113, section 6.8). +* A stream is half-closed in the transaction which queues the end of its + body. A client which had read the whole response could find the stream + still open and reset it; enough such resets made the peer close the + connection with "too many rst_stream". +* Give up on a write which takes the whole timeout. + [#233](https://github.com/kazu-yamamoto/http2/pull/233) +* Stop the receiver in STM when its reads can give up. + [#234](https://github.com/kazu-yamamoto/http2/pull/234) +* Start no GOAWAY thread for run and runIO. + [#235](https://github.com/kazu-yamamoto/http2/pull/235) +* Stop the receiver rather than cancel it when the sender fails first. + [#238](https://github.com/kazu-yamamoto/http2/pull/238) +* Requires network 3.2.12. +* Allows iproute 1.8 and unix-time 0.6. + ## 5.4.12 * Close a connection's streams before waiting for its threads. diff --git a/Network/HTTP2/Client.hs b/Network/HTTP2/Client.hs --- a/Network/HTTP2/Client.hs +++ b/Network/HTTP2/Client.hs @@ -76,7 +76,9 @@ confWriteBuffer, confBufferSize, confSendAll, + confSendAllWith, confReadN, + confReadNWith, confPositionReadMaker, confTimeoutManager, confMySockAddr, diff --git a/Network/HTTP2/H2/Config.hs b/Network/HTTP2/H2/Config.hs --- a/Network/HTTP2/H2/Config.hs +++ b/Network/HTTP2/H2/Config.hs @@ -3,11 +3,12 @@ module Network.HTTP2.H2.Config where import Control.Concurrent.STM +import Control.Monad (void) import Data.IORef import Foreign.Marshal.Alloc (free, mallocBytes) import Network.HTTP.Semantics.Client import Network.Socket -import Network.Socket.ByteString (sendAll) +import Network.Socket.ByteString (sendAll, sendAllWith) import qualified System.TimeManager as T import Network.HPACK @@ -31,13 +32,19 @@ -- reader ends as it does at end of input. Nothing is closed and no -- thread is interrupted to make it return. -- +-- It also fills 'confSendAllWith' and 'confReadNWith' (since 5.4.13), so +-- that the library can give up on a send or a read in the same way. +-- -- @since 5.4.11 allocSimpleConfigWith :: Socket -> BufferSize -> Int -> STM a -> IO Config allocSimpleConfigWith s bufsiz usec giveUp = do confWriteBuffer <- mallocBytes bufsiz let confBufferSize = bufsiz let confSendAll = sendAll s - confReadN <- (\ref -> defaultReadNWith s ref giveUp) <$> newIORef Nothing + confSendAllWith = Just $ \giveUp' bs -> sendAllWith s bs giveUp' + ref <- newIORef Nothing + let confReadN = defaultReadNWith s ref giveUp + confReadNWith = Just $ \stop -> defaultReadNWith s ref (void giveUp `orElse` stop) let confPositionReadMaker = defaultPositionReadMaker confTimeoutManager <- T.initialize usec confMySockAddr <- getSocketName s diff --git a/Network/HTTP2/H2/Context.hs b/Network/HTTP2/H2/Context.hs --- a/Network/HTTP2/H2/Context.hs +++ b/Network/HTTP2/H2/Context.hs @@ -4,6 +4,7 @@ module Network.HTTP2.H2.Context where +import Control.Concurrent (ThreadId) import Control.Concurrent.STM import qualified Control.Exception as E import Data.IORef @@ -95,7 +96,7 @@ , continued :: IORef (Maybe HeaderContinuation) , myStreamId :: TVar StreamId , peerStreamId :: IORef StreamId - , peerLastStreamId :: IORef StreamId + , peerLastStreamId :: TVar StreamId , outputBufferLimit :: IORef Int , outputQ :: TQueue Output -- ^ Invariant: Each stream will only ever have at most one 'Output' @@ -118,6 +119,10 @@ -- ^ For the threads that are timed out: owning a thread and -- timing it out are separate, so the manager carries no timer. , receiverDone :: TVar (Maybe E.SomeException) + , stopReceiving :: TVar Bool + -- ^ Set to end the receiver at the next frame: see 'stopReceiver'. + , receiverStoppable :: Bool + -- ^ Whether the receiver's reads give up on 'stopReceiving'. , appDone :: STM Bool -- ^ Server only: set by 'sioDone' once a 'runIO' application has -- nothing more to send. Never for 'run' and clients. @@ -128,6 +133,11 @@ , peerGoAway :: TVar (Maybe StreamId) -- ^ The last stream identifier of the peer's GOAWAY(NO_ERROR), once -- one has come: see 'goingAway'. + , myGoAway :: TVar (Maybe StreamId) + -- ^ Server only: the last stream identifier of our GOAWAY(NO_ERROR), + -- once we have sent one: see 'admitPeerStream'. + , pingAcked :: TVar ByteString + -- ^ The payload of the last PING the peer acknowledged. , activeRequests :: TVar Int -- ^ Client only: requests whose 'processResponse' has not returned. -- A response can be complete, its stream gone from the table, and @@ -179,7 +189,7 @@ continued <- newIORef Nothing myStreamId <- newTVarIO sid0 peerStreamId <- newIORef 0 - peerLastStreamId <- newIORef 0 + peerLastStreamId <- newTVarIO 0 outputBufferLimit <- newIORef buflim outputQ <- newTQueueIO outputQStreamID <- newTVarIO sid0 @@ -206,9 +216,13 @@ threadManager <- T.newThreadManager let timeoutManager = timmgr receiverDone <- newTVarIO Nothing + stopReceiving <- newTVarIO False + let receiverStoppable = isJust confReadNWith let informationalCallback = confOnInformational let appDone = fromMaybe (return False) mdone peerGoAway <- newTVarIO Nothing + myGoAway <- newTVarIO Nothing + pingAcked <- newTVarIO "" activeRequests <- newTVarIO 0 resetByMe <- newIORef IntSet.empty return Context{..} @@ -227,6 +241,33 @@ ---------------------------------------------------------------- +-- | Ending the receiver as at end of input, once the streams it reads for +-- are done: in 'STM' when its reads can give up, by an exception to it +-- otherwise. +stopReceiver :: Context -> ThreadId -> IO () +stopReceiver Context{stopReceiving, receiverStoppable} receiver + | receiverStoppable = atomically $ writeTVar stopReceiving True + | otherwise = E.throwTo receiver ConnectionIsClosed + +---------------------------------------------------------------- + +-- | Fires once the receiver has been told to stop. +receiverStopped :: Context -> STM () +receiverStopped Context{stopReceiving} = readTVar stopReceiving >>= check + +---------------------------------------------------------------- + +-- | Fires once the receiver has ended with 'ConnectionIsTimeout': a +-- write still blocked then is given up on. +timedOut :: Context -> STM () +timedOut Context{receiverDone} = do + done <- readTVar receiverDone + case done >>= E.fromException of + Just ConnectionIsTimeout -> return () + _ -> retry + +---------------------------------------------------------------- + isClient :: Context -> Bool isClient ctx = role ctx == Client @@ -251,10 +292,19 @@ ---------------------------------------------------------------- getPeerLastStreamId :: Context -> IO StreamId -getPeerLastStreamId ctx = readIORef $ peerLastStreamId ctx +getPeerLastStreamId ctx = readTVarIO $ peerLastStreamId ctx -modifyPeerLastStreamId :: Context -> StreamId -> IO () -modifyPeerLastStreamId ctx sid = atomicModifyIORef' (peerLastStreamId ctx) $ \n -> if sid > n then (sid, ()) else (n, ()) +-- | Counting a stream of the peer's towards the last stream identifier, +-- unless our GOAWAY has gone out below it. Answers whether it is +-- admitted; one that is not must be refused, not processed. +admitPeerStream :: Context -> StreamId -> IO Bool +admitPeerStream Context{peerLastStreamId, myGoAway} sid = atomically $ do + mlast <- readTVar myGoAway + case mlast of + Just lastSid | sid > lastSid -> return False + _ -> do + modifyTVar' peerLastStreamId $ max sid + return True ---------------------------------------------------------------- @@ -323,9 +373,19 @@ open st = st halfClosedRemote :: Context -> Stream -> IO () -halfClosedRemote ctx stream@Stream{streamState} = do - closingCode <- atomically $ stateTVar streamState closeHalf +halfClosedRemote ctx stream = halfClosedRemoteWith ctx stream $ return () + +-- | 'halfClosedRemote' in the transaction which hands the end of the body +-- to its reader, so that a reader which has seen the end never finds the +-- stream open, and resets it as one it did not read to the end. +halfClosedRemoteWith :: Context -> Stream -> STM a -> IO a +halfClosedRemoteWith ctx stream@Stream{streamState} deliver = do + (a, closingCode) <- atomically $ do + a <- deliver + closingCode <- stateTVar streamState closeHalf + return (a, closingCode) traverse_ (closed ctx stream) closingCode + return a where closeHalf :: StreamState -> (Maybe ClosedCode, StreamState) closeHalf x@(Closed _) = (Nothing, x) @@ -588,3 +648,26 @@ | otherwise = (evens, odds) noneOfMine = maybe True ((> lastSid) . fst) $ IntMap.lookupMin mine check $ IntMap.null theirs && noneOfMine && active == 0 + +---------------------------------------------------------------- +-- GOAWAY of ours + +-- | Fixing the last stream identifier of our GOAWAY(NO_ERROR) at the +-- streams admitted so far. Answers it the first time only. +goingAwayMine :: Context -> STM (Maybe StreamId) +goingAwayMine Context{peerLastStreamId, myGoAway} = do + old <- readTVar myGoAway + case old of + Just _ -> return Nothing + Nothing -> do + lastSid <- readTVar peerLastStreamId + writeTVar myGoAway $ Just lastSid + return $ Just lastSid + +-- | Waiting until the streams below our GOAWAY are done. Those above it +-- are refused, so none of the peer's is left. +drainedMine :: Context -> STM () +drainedMine Context{myGoAway, oddStreamTable} = do + readTVar myGoAway >>= check . isJust + odds <- oddTable <$> readTVar oddStreamTable + check $ IntMap.null odds diff --git a/Network/HTTP2/H2/Receiver.hs b/Network/HTTP2/H2/Receiver.hs --- a/Network/HTTP2/H2/Receiver.hs +++ b/Network/HTTP2/H2/Receiver.hs @@ -23,7 +23,6 @@ import Data.Void import Network.Control import Network.HTTP.Semantics -import qualified System.IO.Error as E import qualified System.ThreadManager as T import qualified System.TimeManager as TM @@ -72,10 +71,19 @@ else TM.withHandle (timeoutManager ctx) (E.throwTo tid ConnectionIsTimeout) loop2 + -- Told to stop, the receiver ends between frames as at end of input. + stopped = receiverStopped ctx + waitOrStop = do + stop <- + atomically $ + (True <$ stopped) `orElse` (False <$ waitControlQ (controlQ ctx)) + when stop $ E.throwIO ConnectionIsClosed + readHeader = readNGivingUp conf stopped frameHeaderLength + loop1 :: IO Void loop1 = do - atomically $ waitControlQ $ controlQ ctx - hd <- confReadN frameHeaderLength -- throwing an exception on timeout + waitOrStop + hd <- readHeader -- throwing an exception on timeout when (BS.null hd) $ E.throwIO ConnectionIsClosed processFrame ctx conf $ decodeFrameHeader hd loop1 @@ -85,8 +93,8 @@ -- If 'confReadN' is timeouted, 'ConnectionIsTimeout' is thrown -- to destroy the thread trees. So it is if the peer does not -- read what we send, and we wait here for ever. - atomically $ waitControlQ $ controlQ ctx - hd <- confReadN frameHeaderLength + waitOrStop + hd <- readHeader TM.tickle th when (BS.null hd) $ E.throwIO ConnectionIsClosed processFrame ctx conf $ decodeFrameHeader hd @@ -105,13 +113,13 @@ | isServer ctx = E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "push promise is not allowed" -processFrame Context{..} conf (ftyp, FrameHeader{payloadLength, streamId}) +processFrame ctx@Context{..} conf (ftyp, FrameHeader{payloadLength, streamId}) | ftyp > maxFrameType = do mx <- readIORef continued case mx of Nothing -> do -- ignoring unknown frame - void $ readPayload conf payloadLength + void $ readPayload ctx conf payloadLength Just _ -> E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "unknown frame" processFrame ctx@Context{..} conf typhdr@(ftyp, header) = do -- My SETTINGS_MAX_FRAME_SIZE @@ -136,20 +144,20 @@ -- 'confReadN' answers with an empty string at end of input, so a payload that -- comes back short means the peer hung up in the middle of the frame. Saying -- so here keeps every decoder below from being handed fewer bytes than the --- frame header promised it. -readPayload :: Config -> Int -> IO ByteString -readPayload Config{..} len = do - bs <- confReadN len +-- frame header promised it. A receiver told to stop gives it up the same. +readPayload :: Context -> Config -> Int -> IO ByteString +readPayload ctx conf len = do + bs <- readNGivingUp conf (receiverStopped ctx) len when (BS.length bs /= len) $ E.throwIO ConnectionIsClosed return bs controlOrStream :: Context -> Config -> FrameType -> FrameHeader -> IO () controlOrStream ctx@Context{..} conf ftyp header@FrameHeader{flags, streamId, payloadLength} | isControl streamId = do - bs <- readPayload conf payloadLength + bs <- readPayload ctx conf payloadLength control ftyp header bs ctx | ftyp == FramePushPromise = do - bs <- readPayload conf payloadLength + bs <- readPayload ctx conf payloadLength -- A promised stream can be refused over concurrency too, and by the -- time 'push' gets that far it has decoded the field block, so the -- same reasoning as 'resettable' applies: reset the promised stream @@ -159,7 +167,7 @@ | otherwise = do mcont <- checkContinued mstrm <- getStream ctx ftyp streamId - bs <- readPayload conf payloadLength + bs <- readPayload ctx conf payloadLength case mcont of Just hc -> continuation hc bs mstrm Nothing -> @@ -262,7 +270,7 @@ state <- onResponseHeaders ctx streamId hcl (hcEndOfStream hc) tbl processState state ctx strm streamId Open _ (Body q mcl bodyLength tlr) -> do - state <- onTrailers ctx streamId blk q mcl bodyLength tlr + state <- onTrailers ctx strm streamId blk q mcl bodyLength tlr processState state ctx strm streamId _otherwise -> -- The block began on a stream that was open for @@ -370,9 +378,13 @@ -- that had been acted on. launchHandler :: Context -> Stream -> InpObj -> IO () launchHandler ctx@Context{roleInfo} strm inpObj = do - modifyPeerLastStreamId ctx $ streamNumber strm + -- One above our GOAWAY is refused: the peer retries it elsewhere. + admitted <- admitPeerStream ctx sid + unless admitted $ E.throwIO $ StreamErrorIsSent RefusedStream sid "going away" let ServerInfo{..} = toServerInfo roleInfo launch ctx strm inpObj + where + sid = streamNumber strm ---------------------------------------------------------------- @@ -496,8 +508,9 @@ setframe = CFrames (Just peerAlist) (frames ++ [ack]) writeIORef myFirstSettings True enqueueControl controlQ setframe -control FramePing FrameHeader{flags, streamId} bs ctx@Context{mySettings, pingRate} = - unless (testAck flags) $ do +control FramePing FrameHeader{flags, streamId} bs ctx@Context{mySettings, pingRate, pingAcked} + | testAck flags = atomically $ writeTVar pingAcked bs + | otherwise = do rate <- getRate pingRate if rate > pingRateLimit mySettings then E.throwIO $ ConnectionErrorIsSent EnhanceYourCalm streamId "too many ping" @@ -514,7 +527,7 @@ receiver <- myThreadId T.forkManaged (threadManager ctx) "H2 draining after GOAWAY" $ do atomically $ drained ctx - E.throwTo receiver ConnectionIsClosed + stopReceiver ctx receiver else E.throwIO $ ConnectionErrorIsReceived err sid $ Short.toShort msg control FrameWindowUpdate header bs ctx = do WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs @@ -635,14 +648,14 @@ return s -- Transition (stream2) -stream FrameHeaders header@FrameHeader{flags, streamId} bs ctx s@(Open _ (Body q mcl bodyLength tlr)) _ = do +stream FrameHeaders header@FrameHeader{flags, streamId} bs ctx s@(Open _ (Body q mcl bodyLength tlr)) strm = do HeadersFrame _ frag <- guardIt $ decodeHeadersFrame header bs let endOfStream = testEndStream flags -- checking frag == "" is not necessary if endOfStream then do if testEndHeader flags - then onTrailers ctx streamId frag q mcl bodyLength tlr + then onTrailers ctx strm streamId frag q mcl bodyLength tlr else do startHeaderBlock ctx streamId endOfStream frag return s @@ -714,12 +727,20 @@ ProtocolError streamId "actual body length is not the same as content-length" + -- The end goes in with the half-close: see 'halfClosedRemoteWith'. + let deliver + | endOfStream = + halfClosedRemoteWith ctx strm . (<* writeTQueue q (Right (mempty, True))) + | otherwise = atomically -- Empty Frame Flooding - CVE-2019-9518 if body == "" - then unless endOfStream $ do - rate <- getRate emptyFrameRate - when (rate > emptyFrameRateLimit mySettings) $ do - E.throwIO $ ConnectionErrorIsSent EnhanceYourCalm streamId "too many empty data" + then + if endOfStream + then deliver $ return () + else do + rate <- getRate emptyFrameRate + when (rate > emptyFrameRateLimit mySettings) $ do + E.throwIO $ ConnectionErrorIsSent EnhanceYourCalm streamId "too many empty data" else do writeIORef bodyLength len -- Not for a stream closed since its state was read, by @@ -732,7 +753,7 @@ -- the stream. Checked in the same transaction, so that -- a closer that has emptied the queue finds nothing put -- in after it. - outcome <- atomically $ do + outcome <- deliver $ do st <- readTVar streamState discard <- readTVar streamDiscardBody if @@ -748,12 +769,7 @@ unless (connectionCreditedOnArrival streamId) $ giveBackConnectionWindow ctx $ BS.length body - if endOfStream - then do - -- no trailers - atomically $ writeTQueue q $ Right (mempty, True) - return HalfClosedRemote - else return s + return $ if endOfStream then HalfClosedRemote else s -- (No state transition) stream FrameWindowUpdate header bs _ s strm = do @@ -938,9 +954,6 @@ sid <- getPeerLastStreamId ctx return $ goawayFrame sid err msg -sendGoaway :: Config -> ByteString -> IO () -sendGoaway Config{..} frame = confSendAll frame `E.catchIOError` \_ -> return () - ---------------------------------------------------------------- sendPing :: Context -> Bool -> ByteString -> IO () @@ -958,6 +971,7 @@ -- the block is decoded, for the dynamic table. onTrailers :: Context + -> Stream -> StreamId -> HeaderBlockFragment -> TQueue (Either E.SomeException (ByteString, Bool)) @@ -967,7 +981,7 @@ -- ^ The length of the body received -> IORef (Maybe TokenHeaderTable) -> IO StreamState -onTrailers ctx streamId blk q mcl bodyLength tlr = do +onTrailers ctx strm streamId blk q mcl bodyLength tlr = do tbl <- hpackDecodeTrailer blk streamId ctx len <- readIORef bodyLength when (just mcl (/= len)) $ @@ -977,7 +991,7 @@ streamId "actual body length is not the same as content-length" writeIORef tlr (Just tbl) - atomically $ writeTQueue q $ Right (mempty, True) + halfClosedRemoteWith ctx strm $ writeTQueue q $ Right (mempty, True) return HalfClosedRemote -- | Start accumulating a header block that does not fit in a single frame diff --git a/Network/HTTP2/H2/Sender.hs b/Network/HTTP2/H2/Sender.hs --- a/Network/HTTP2/H2/Sender.hs +++ b/Network/HTTP2/H2/Sender.hs @@ -19,6 +19,7 @@ import Network.HTTP.Semantics.Client import Network.HTTP.Semantics.IO import qualified System.ThreadManager as T +import qualified System.TimeManager as TM import Imports import Network.HPACK (setLimitForEncoding, toTokenHeaderTable) @@ -107,10 +108,27 @@ checkDone _ _ _ = return Nothing frameSender :: Context -> Config -> IO E.SomeException -frameSender +frameSender ctx conf@Config{confSendAllWith} = do + labelMe "H2 sender" + case confSendAllWith of + Nothing -> frameSenderWith ctx conf $ sendAllGivingUp conf retry + -- A write is given up on once it has taken the whole timeout: the + -- receiver's timeout no longer covers it once the receiver has ended. + Just _ -> do + stalled <- newTVarIO False + TM.withHandle (timeoutManager ctx) (atomically $ writeTVar stalled True) $ \th -> do + TM.pause th + frameSenderWith ctx conf $ \bs -> do + TM.resume th + let giveUp = timedOut ctx `orElse` (readTVar stalled >>= check) + sendAllGivingUp conf giveUp bs <* TM.pause th + +frameSenderWith + :: Context -> Config -> (ByteString -> IO (Either () ())) -> IO E.SomeException +frameSenderWith ctx@Context{outputQ, controlQ, encodeDynamicTable, outputBufferLimit} - Config{..} = do - labelMe "H2 sender" + Config{..} + send = do -- DATA that has to wait for the connection window, in the order it -- came: see 'dequeue'. parked <- newTVarIO [] @@ -139,7 +157,10 @@ -- the buffer are filled. flushN :: Offset -> IO () flushN 0 = return () - flushN n = bufferIO confWriteBuffer n confSendAll + flushN n = bufferIO confWriteBuffer n $ \bs -> + -- Given up on once the connection has timed out: closed with + -- GOAWAY like any other timeout. + send bs >>= either (\_ -> E.throwIO ConnectionIsTimeout) return flushIfNecessary :: Offset -> IO Offset flushIfNecessary off = do diff --git a/Network/HTTP2/H2/Types.hs b/Network/HTTP2/H2/Types.hs --- a/Network/HTTP2/H2/Types.hs +++ b/Network/HTTP2/H2/Types.hs @@ -15,6 +15,7 @@ import Network.HTTP.Semantics.Client import Network.HTTP.Semantics.IO import Network.Socket hiding (Stream) +import System.IO.Error (catchIOError) import System.IO.Unsafe import qualified System.TimeManager as T @@ -266,7 +267,28 @@ -- So, this value is announced via SETTINGS_MAX_FRAME_SIZE -- to the peer. , confSendAll :: ByteString -> IO () + -- ^ Not used when 'confSendAllWith' is 'Just'. When replacing it in + -- a config which has one, e.g. from 'allocSimpleConfig', set + -- 'confSendAllWith' to 'Nothing' or to a send of the same kind. + , confSendAllWith :: Maybe (STM () -> ByteString -> IO (Either () ())) + -- ^ 'confSendAll' which the library can give up on: 'Left' when the + -- 'STM' action fires first. With it, a write which the peer does + -- not read is ended by the timeout rather than by killing a thread. + -- Part of the data may have gone, so the connection is closed after. + -- + -- @since 5.4.13 , confReadN :: Int -> IO ByteString + -- ^ When 'confReadNWith' is 'Just', frames are read with it and only + -- the connection preface with this. When replacing it in a config + -- which has one, e.g. from 'allocSimpleConfig', set 'confReadNWith' + -- to 'Nothing' or to a read of the same kind sharing its buffer. + , confReadNWith :: Maybe (STM () -> Int -> IO ByteString) + -- ^ 'confReadN' which the library can give up on: it answers with an + -- empty string, as at end of input, when the 'STM' action fires + -- first. With it, the reader is stopped without being interrupted. + -- It must share its buffer with 'confReadN'. + -- + -- @since 5.4.13 , confPositionReadMaker :: PositionReadMaker , confTimeoutManager :: TM.Manager , confMySockAddr :: SockAddr @@ -290,7 +312,9 @@ { confWriteBuffer = nullPtr , confBufferSize = 0 , confSendAll = \_ -> return () + , confSendAllWith = Nothing , confReadN = \_ -> return "" + , confReadNWith = Nothing , confPositionReadMaker = defaultPositionReadMaker , confTimeoutManager = T.defaultManager , confMySockAddr = SockAddrInet 0 0 @@ -298,6 +322,31 @@ , confReadNTimeout = False , confOnInformational = \_ _ -> return () } + +-- | Sending with 'confSendAllWith' if there is one, giving up when the +-- 'STM' action fires, and with 'confSendAll' otherwise. +sendAllGivingUp :: Config -> STM () -> ByteString -> IO (Either () ()) +sendAllGivingUp Config{..} giveUp bs = case confSendAllWith of + Just sendWith -> sendWith giveUp bs + Nothing -> Right <$> confSendAll bs + +-- | Reading with 'confReadNWith' if there is one, giving up when the +-- 'STM' action fires, and with 'confReadN' otherwise. +readNGivingUp :: Config -> STM () -> Int -> IO ByteString +readNGivingUp Config{..} giveUp n = case confReadNWith of + Just readWith -> readWith giveUp n + Nothing -> confReadN n + +-- | Sending GOAWAY, as a last word: it waits one second at most, since a +-- peer which does not read would hold the connection open for good. +sendGoaway :: Config -> ByteString -> IO () +sendGoaway conf frame = send `catchIOError` \_ -> return () + where + send = case confSendAllWith conf of + Nothing -> confSendAll conf frame + Just _ -> do + late <- registerDelay 1000000 + void $ sendAllGivingUp conf (readTVar late >>= check) frame isAsyncException :: E.Exception e => e -> Bool isAsyncException e = diff --git a/Network/HTTP2/Server.hs b/Network/HTTP2/Server.hs --- a/Network/HTTP2/Server.hs +++ b/Network/HTTP2/Server.hs @@ -26,6 +26,7 @@ module Network.HTTP2.Server ( -- * Runner run, + runWith, -- * Server configuration ServerConfig, @@ -56,7 +57,9 @@ confWriteBuffer, confBufferSize, confSendAll, + confSendAllWith, confReadN, + confReadNWith, confPositionReadMaker, confTimeoutManager, confMySockAddr, @@ -77,4 +80,5 @@ ServerConfig (..), defaultServerConfig, run, + runWith, ) diff --git a/Network/HTTP2/Server/Internal.hs b/Network/HTTP2/Server/Internal.hs --- a/Network/HTTP2/Server/Internal.hs +++ b/Network/HTTP2/Server/Internal.hs @@ -9,6 +9,7 @@ Stream, ServerIO (..), runIO, + runIOWith, ) where import Network.HTTP.Semantics.Server diff --git a/Network/HTTP2/Server/Run.hs b/Network/HTTP2/Server/Run.hs --- a/Network/HTTP2/Server/Run.hs +++ b/Network/HTTP2/Server/Run.hs @@ -45,12 +45,24 @@ -- | Running HTTP/2 server. run :: ServerConfig -> Config -> Server -> IO () -run sconf conf server = do +run = runMaybe Nothing + +-- | Like 'run', but the connection goes away once the 'STM' action fires: +-- GOAWAY(NO_ERROR) at once, streams above it refused, and the +-- connection closed when those below it are done. +-- +-- @since 5.4.13 +runWith :: STM () -> ServerConfig -> Config -> Server -> IO () +runWith = runMaybe . Just + +-- Nothing for 'run': no thread waits on an action which never fires. +runMaybe :: Maybe (STM ()) -> ServerConfig -> Config -> Server -> IO () +runMaybe mGoAway sconf conf server = do ok <- checkPreface conf when ok $ do let lnch = runServer conf server ctx <- setup sconf conf lnch Nothing - runH2 conf ctx + runH2 conf ctx mGoAway ---------------------------------------------------------------- @@ -71,7 +83,27 @@ -> Config -> (ServerIO Stream -> IO (IO ())) -> IO () -runIO sconf conf@Config{..} action = do +runIO = runIOMaybe Nothing + +-- | Like 'runIO', but the connection goes away once the 'STM' action +-- fires, as with 'runWith'. +-- +-- @since 5.4.13 +runIOWith + :: STM () + -> ServerConfig + -> Config + -> (ServerIO Stream -> IO (IO ())) + -> IO () +runIOWith = runIOMaybe . Just + +runIOMaybe + :: Maybe (STM ()) + -> ServerConfig + -> Config + -> (ServerIO Stream -> IO (IO ())) + -> IO () +runIOMaybe mGoAway sconf conf@Config{..} action = do ok <- checkPreface conf when ok $ do inpQ <- newTQueueIO @@ -97,7 +129,7 @@ , sioDone = atomically $ writeTVar done True } io <- action serverIO - concurrently_ io $ runH2 conf ctx + concurrently_ io $ runH2 conf ctx mGoAway checkPreface :: Config -> IO Bool checkPreface conf@Config{..} = do @@ -120,32 +152,75 @@ confTimeoutManager mIsDone -runH2 :: Config -> Context -> IO () -runH2 conf ctx = do +runH2 :: Config -> Context -> Maybe (STM ()) -> IO () +runH2 conf ctx mGoAway = do let mgr = threadManager ctx runReceiver = frameReceiver ctx conf runSender = frameSender ctx conf + -- Two GOAWAYs, as RFC 9113 section 6.8 has it: the first, with + -- the largest identifier, stops the client opening streams; the + -- second fixes the last one taken once a PING has gone round, so + -- that the streams it opened meanwhile are not refused. The + -- receiver is stopped as it is after the peer's GOAWAY, once the + -- streams below ours are done; it reads until then. + sendGoAway :: STM () -> Async a -> IO () + sendGoAway goAway ar = do + atomically goAway + enqueueControl (controlQ ctx) $ + CFrames + Nothing + [ goawayFrame maxStreamId NoError "going away" + , pingFrame False goAwayPing + ] + roundTrip <- registerDelay 1000000 + atomically $ + (readTVar (pingAcked ctx) >>= check . (== goAwayPing)) + `orElse` (readTVar roundTrip >>= check) + mlast <- atomically $ goingAwayMine ctx + forM_ mlast $ \lastSid -> + enqueueControl (controlQ ctx) $ + CFrames Nothing [goawayFrame lastSid NoError "going away"] + atomically $ drainedMine ctx + stopReceiver ctx $ asyncThreadId ar + withGoAway ar body = case mGoAway of + Nothing -> body + Just goAway -> withAsync (sendGoAway goAway ar) $ \_ -> body runBackgroundThreads = withAsync runReceiver $ \ar -> - withAsync runSender $ \as -> do - r <- waitEither ar as - e <- case r of - -- The receiver is done; the sender finishes once it - -- has flushed what is queued. - Left _ -> wait as - -- The sender finished first. Either the receiver is - -- done too and not yet seen to be, and this is its - -- error, or the sender failed: nothing more would go - -- out, and leaving the receiver to run on left the - -- connection open and silent, with no GOAWAY. Both - -- are closed with it. - Right e -> return e - closureServer conf ctx e + withGoAway ar $ + withAsync runSender $ \as -> do + r <- waitEither ar as + e <- case r of + -- The receiver is done; the sender finishes once it + -- has flushed what is queued. + Left _ -> wait as + -- The sender finished first. Either the receiver is + -- done too and not yet seen to be, and this is its + -- error, or the sender failed: nothing more would go + -- out, and leaving the receiver to run on left the + -- connection open and silent, with no GOAWAY. Both + -- are closed with it: the receiver is stopped + -- rather than cancelled when its reads can give up. + Right e -> do + stopReceiver ctx $ asyncThreadId ar + late <- registerDelay 1000000 + atomically $ + void (waitCatchSTM ar) `orElse` (readTVar late >>= check) + return e + closureServer conf ctx e stopAfterAndWait mgr runBackgroundThreads $ \res -> closeAllStreams (oddStreamTable ctx) (evenStreamTable ctx) res +-- | The payload of the PING sent with the first GOAWAY. +goAwayPing :: ByteString +goAwayPing = "goaway\0\0" + +-- | The largest stream identifier. +maxStreamId :: StreamId +maxStreamId = 2147483647 + -- connClose must not be called here since Run:fork calls it goaway :: Config -> ErrorCode -> ByteString -> IO () -goaway Config{..} etype debugmsg = confSendAll bytestream +goaway conf etype debugmsg = sendGoaway conf bytestream where bytestream = goawayFrame 0 etype debugmsg diff --git a/http2.cabal b/http2.cabal --- a/http2.cabal +++ b/http2.cabal @@ -1,6 +1,6 @@ cabal-version: 2.0 name: http2 -version: 5.4.12 +version: 5.4.13 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto @@ -117,13 +117,13 @@ containers >=0.6, http-semantics >= 0.4.2 && <0.5, http-types >=0.12 && <0.13, - iproute >= 1.7 && < 1.8, - network >=3.1, + iproute >= 1.7 && < 1.9, + network >=3.2.12, network-byte-order >=0.1.7 && <0.2, network-control >=0.1 && <0.2, stm >=2.5 && <2.6, time-manager >=0.5 && <0.6, - unix-time >=0.4.11 && <0.6, + unix-time >=0.4.11 && <0.7, utf8-string >=1.0 && <1.1 executable h2c-client @@ -293,6 +293,7 @@ HPACK.IntegerSpec HTTP2.ClientSpec HTTP2.FrameSpec + HTTP2.GiveUpSpec HTTP2.ServerSpec default-language: Haskell2010 @@ -312,6 +313,7 @@ network-byte-order, network-run >= 0.7 && <0.8, random, + stm, time-manager >=0.5 && <0.6, typed-process diff --git a/test/HTTP2/GiveUpSpec.hs b/test/HTTP2/GiveUpSpec.hs new file mode 100644 --- /dev/null +++ b/test/HTTP2/GiveUpSpec.hs @@ -0,0 +1,149 @@ +{-# LANGUAGE OverloadedStrings #-} + +module HTTP2.GiveUpSpec (spec) where + +import Control.Concurrent +import Control.Concurrent.STM +import qualified Control.Exception as E +import Control.Monad (forever) +import Data.ByteString (ByteString) +import qualified Data.ByteString as B +import Data.ByteString.Builder (byteString) +import Data.Word (Word8) +import Network.HTTP.Types +import Network.Run.TCP +import Network.Socket ( + ShutdownCmd (..), + SocketOption (..), + setSocketOption, + shutdown, + ) +import Network.Socket.ByteString +import System.IO.Unsafe +import System.Random +import System.Timeout (timeout) +import Test.Hspec + +import Network.HTTP2.Frame (connectionPreface) +import Network.HTTP2.Server + +port :: String +port = show $ unsafePerformIO (randomPort <$> getStdGen) + where + randomPort = fst . randomR (45321 :: Int, 46320) + +host :: String +host = "127.0.0.1" + +-- | The timeout of the server, in microseconds. +serverTimeout :: Int +serverTimeout = 1000000 + +spec :: Spec +spec = describe "give-up" $ do + it "gives up on a write the peer does not read" $ do + ended <- newEmptyMVar + let server _req _aux sendResponse = + sendResponse (responseStreaming ok200 [] endless) [] + endless write flush = forever $ write (byteString chunk) >> flush + chunk = B.replicate 65536 0x41 + runServer = runTCPServer (Just host) port $ \s -> + E.bracket (allocSimpleConfig' s 32768 serverTimeout) freeSimpleConfig $ \conf -> + (run defaultServerConfig conf server `E.catch` ignore) + `E.finally` putMVar ended () + E.bracket (forkIO runServer) killThread $ \_ -> do + threadDelay 10000 + -- The client opens the windows wide, asks, and never reads, + -- so the server's writes stop in the socket, not on a window. + r <- runTCPClient host port $ \s -> do + -- A fixed small buffer: one that grows would take the data in. + setSocketOption s RecvBuffer 4096 + sendAll s connectionPreface + sendAll s $ frame 4 0 0 $ B.pack [0, 4, 0x7f, 0xff, 0xff, 0xff] + sendAll s $ frame 8 0 0 $ B.pack [0x7f, 0xff, 0, 0] + sendAll s $ frame 1 5 1 getHeaderBlock + timeout 6000000 $ takeMVar ended + r `shouldBe` Just () + it "gives up on a write the peer does not read after the receiver ends" $ do + ended <- newEmptyMVar + let server _req _aux sendResponse = + sendResponse (responseStreaming ok200 [] endless) [] + endless write flush = forever $ write (byteString chunk) >> flush + chunk = B.replicate 65536 0x41 + runServer = runTCPServer (Just host) port $ \s -> + E.bracket (allocSimpleConfig' s 32768 serverTimeout) freeSimpleConfig $ \conf -> + (run defaultServerConfig conf server `E.catch` ignore) + `E.finally` putMVar ended () + E.bracket (forkIO runServer) killThread $ \_ -> do + threadDelay 10000 + -- As above, but the client ends its side once it has asked, so + -- the receiver ends on EOF rather than by its timeout. + r <- runTCPClient host port $ \s -> do + setSocketOption s RecvBuffer 4096 + sendAll s connectionPreface + sendAll s $ frame 4 0 0 $ B.pack [0, 4, 0x7f, 0xff, 0xff, 0xff] + sendAll s $ frame 8 0 0 $ B.pack [0x7f, 0xff, 0, 0] + sendAll s $ frame 1 5 1 getHeaderBlock + threadDelay 100000 + shutdown s ShutdownSend + timeout 6000000 $ takeMVar ended + r `shouldBe` Just () + it "stops a receiver which cannot be interrupted once it has gone away" $ do + ended <- newEmptyMVar + goVar <- newTVarIO False + let server _req _aux sendResponse = sendResponse (responseNoBody ok200 []) [] + -- A read no exception reaches, as on a platform where a blocked + -- recv cannot be interrupted; the one which gives up still can. + uninterruptible conf = conf{confReadN = E.uninterruptibleMask_ . confReadN conf} + runServer = runTCPServer (Just host) port $ \s -> + E.bracket (allocSimpleConfig' s 32768 30000000) freeSimpleConfig $ \conf -> + (go (uninterruptible conf) `E.catch` ignore) `E.finally` putMVar ended () + go conf = runWith (readTVar goVar >>= check) defaultServerConfig conf server + E.bracket (forkIO runServer) killThread $ \_ -> do + threadDelay 10000 + -- The client says nothing after its SETTINGS and does not answer + -- the PING, so the receiver waits in a read when it is stopped. + r <- runTCPClient host port $ \s -> do + sendAll s connectionPreface + sendAll s $ frame 4 0 0 "" + threadDelay 100000 + atomically $ writeTVar goVar True + timeout 5000000 $ takeMVar ended + r `shouldBe` Just () + it "stops a receiver which cannot be interrupted once the sender has failed" $ do + ended <- newEmptyMVar + let server _req _aux sendResponse = sendResponse (responseNoBody ok200 []) [] + -- No exception reaches the reads, and every write fails: the + -- receiver used to be cancelled, which waited for good. + broken conf = + conf + { confReadN = E.uninterruptibleMask_ . confReadN conf + , confReadNWith = fmap (\r stop -> E.uninterruptibleMask_ . r stop) (confReadNWith conf) + , confSendAllWith = Just $ \_ _ -> E.throwIO $ userError "broken" + } + runServer = runTCPServer (Just host) port $ \s -> + E.bracket (allocSimpleConfig' s 32768 30000000) freeSimpleConfig $ \conf -> + (run defaultServerConfig (broken conf) server `E.catch` ignore) + `E.finally` putMVar ended () + E.bracket (forkIO runServer) killThread $ \_ -> do + threadDelay 10000 + r <- runTCPClient host port $ \s -> do + sendAll s connectionPreface + sendAll s $ frame 4 0 0 "" + timeout 5000000 $ takeMVar ended + r `shouldBe` Just () + where + ignore :: E.SomeException -> IO () + ignore _ = return () + +-- | A frame of this type, flags and stream with this payload. +frame :: Word8 -> Word8 -> Word8 -> ByteString -> ByteString +frame typ flags sid payload = + B.pack [0, fromIntegral (len `div` 256), fromIntegral (len `mod` 256), typ, flags, 0, 0, 0, sid] + <> payload + where + len = B.length payload + +-- | GET / for 127.0.0.1, from the static table and one literal. +getHeaderBlock :: ByteString +getHeaderBlock = B.pack [0x82, 0x86, 0x84, 0x01, 0x09] <> "127.0.0.1" diff --git a/test/HTTP2/ServerSpec.hs b/test/HTTP2/ServerSpec.hs --- a/test/HTTP2/ServerSpec.hs +++ b/test/HTTP2/ServerSpec.hs @@ -18,6 +18,7 @@ import Control.Concurrent import Control.Concurrent.Async +import Control.Concurrent.STM (TVar, atomically, check, newTVarIO, readTVar, writeTVar) import qualified Control.Exception as E import Control.Monad import Crypto.Hash (Context, SHA1) @@ -27,6 +28,7 @@ import Data.ByteString.Builder (Builder, byteString) import qualified Data.ByteString.Char8 as C8 import Data.IORef +import Data.List (sort) import Data.Maybe (isJust, isNothing) import Network.HTTP.Semantics import Network.HTTP.Types @@ -445,6 +447,56 @@ Just (Left e) -> expectationFailure $ show (e :: C.HTTP2Error) Nothing -> expectationFailure "timed out" + it "answers the streams it has taken and refuses the rest once it goes away" $ do + -- RFC 9113, section 6.8: GOAWAY with the largest identifier, a + -- PING, and once the PING is answered GOAWAY(NO_ERROR) with the + -- last stream taken; one opened after it is refused, and the + -- connection is closed once stream 1 is answered. + goVar <- newTVarIO False + withServer (runServerGoingAway id goVar) $ do + threadDelay 10000 + timeout 5000000 (goAwayFromServer False goVar) + `shouldReturn` Just + [ "GOAWAY 2147483647 NoError" + , "GOAWAY 1 NoError" + , "RST_STREAM 3 RefusedStream" + , "DATA 1 END_STREAM" + , "GOAWAY 1 NoError" + ] + + it "goes away the same with reads which cannot give up" $ do + -- Without 'confReadNWith', as for warp and http2-tls, the + -- receiver is stopped by an exception instead. + goVar <- newTVarIO False + withServer (runServerGoingAway noReadNWith goVar) $ do + threadDelay 10000 + timeout 5000000 (goAwayFromServer False goVar) + `shouldReturn` Just + [ "GOAWAY 2147483647 NoError" + , "GOAWAY 1 NoError" + , "RST_STREAM 3 RefusedStream" + , "DATA 1 END_STREAM" + , "GOAWAY 1 NoError" + ] + + it "answers a stream opened between its two GOAWAYs" $ do + -- The second GOAWAY waits for the PING, so a stream already on + -- its way when the first came is taken, not refused. + goVar <- newTVarIO False + withServer (runServerGoingAway id goVar) $ do + threadDelay 10000 + r <- timeout 5000000 (goAwayFromServer True goVar) + sort <$> r + `shouldBe` Just + ( sort + [ "GOAWAY 2147483647 NoError" + , "GOAWAY 3 NoError" + , "DATA 3 END_STREAM" + , "DATA 1 END_STREAM" + , "GOAWAY 3 NoError" + ] + ) + it "closes the connection once a runIO application is done" $ -- 'sioDone' set a flag that nothing read, so the connection -- stayed open until the client closed it. @@ -478,6 +530,12 @@ timeout 5000000 goAwayFromClient `shouldReturn` Just ["HEADERS 1", "DATA 1 END_STREAM", "GOAWAY NoError"] + it "answers them the same with reads which cannot give up" $ + withServer (runServerWithConfig noReadNWith) $ do + threadDelay 10000 + timeout 5000000 goAwayFromClient + `shouldReturn` Just ["HEADERS 1", "DATA 1 END_STREAM", "GOAWAY NoError"] + it "resets a malformed request and goes on serving the connection" $ -- An upper-case field name makes the request malformed: a -- stream error (RFC 9113, section 8.1.1). It used to close the @@ -670,14 +728,22 @@ ignoreHTTP2Error _ = pure () runServer :: ServerSettings -> IO () -runServer set = runTCPServerWithSettings set (Just host) port runHTTP2Server +runServer = runServerWithConfig id + +-- | Like 'runServer', with the config changed. +runServerWithConfig :: (Config -> Config) -> ServerSettings -> IO () +runServerWithConfig modify set = runTCPServerWithSettings set (Just host) port runHTTP2Server where runHTTP2Server s = E.bracket (allocSimpleConfig s 32768) freeSimpleConfig - (\conf -> run defaultServerConfig conf server) + (\conf -> run defaultServerConfig (modify conf) server) +-- | A config whose reads the library cannot give up on. +noReadNWith :: Config -> Config +noReadNWith conf = conf{confReadNWith = Nothing} + -- | Like 'runServer', but announcing room for a single concurrent stream. -- | Uploading 100000 octets of a file through 'C.runIO', and what the server -- says it received. @@ -929,6 +995,75 @@ ] _ <- E.try (action s) :: IO (Either E.IOException ()) collectFrames s + +-- | Like 'runServerWithConfig', going away once the variable is set. +runServerGoingAway :: (Config -> Config) -> TVar Bool -> ServerSettings -> IO () +runServerGoingAway modify goVar set = runTCPServerWithSettings set (Just host) port runHTTP2Server + where + runHTTP2Server s = + E.bracket + (allocSimpleConfig s 32768) + freeSimpleConfig + (\conf -> runWith (readTVar goVar >>= check) defaultServerConfig (modify conf) server) + +-- | A request for /slow, then the server told to go away, and a request for +-- /: before the PING that came with the first GOAWAY is answered when +-- early, after the second GOAWAY otherwise. What the server sends from its +-- first GOAWAY on. +goAwayFromServer :: Bool -> TVar Bool -> IO [String] +goAwayFromServer early goVar = runTCPClient host port $ \s -> do + sendAll s connectionPreface + sendAll s $ encodeFrame (EncodeInfo defaultFlags 0 Nothing) $ SettingsFrame [] + request s 1 "/slow" + threadDelay 50000 + atomically $ writeTVar goVar True + first <- readUntil s isGoAway + when early $ request s 3 "/" + ping <- readUntil s isPing + second <- + if early + then return [] + else readUntil s isGoAway <* request s 3 "/" + rest <- collect s + return $ first ++ ping ++ second ++ rest + where + request s sid path = + sendAll s $ + encodeFrame (EncodeInfo (setEndStream $ setEndHeader defaultFlags) sid Nothing) $ + HeadersFrame Nothing $ + hpackEncode + [ (":scheme", "http") + , (":authority", "127.0.0.1") + , (":path", path) + , (":method", "GET") + ] + isGoAway (ftyp, _, _) = ftyp == FrameGoAway + isPing (ftyp, fh, _) = ftyp == FramePing && not (testAck (flags fh)) + -- Up to and including the first frame which satisfies the predicate; + -- a PING is answered as it comes. + readUntil s done = do + mf <- recvFrame s + case mf of + Nothing -> return [] + Just f -> do + answerPing s f + if done f + then return $ maybe [] pure $ frameText f + else maybe id (:) (frameText f) <$> readUntil s done + collect s = readUntil s (const False) + answerPing s (FramePing, fh, p) + | not (testAck (flags fh)) = + sendAll s $ encodeFrame (EncodeInfo (setAck defaultFlags) 0 Nothing) $ PingFrame p + answerPing _ _ = return () + frameText (FrameGoAway, fh, p) + | Right (GoAwayFrame sid err _) <- decodeGoAwayFrame fh p = + Just $ "GOAWAY " ++ show sid ++ " " ++ show err + frameText (FrameRSTStream, fh, p) + | Right (RSTStreamFrame err) <- decodeRSTStreamFrame fh p = + Just $ "RST_STREAM " ++ show (streamId fh) ++ " " ++ show err + frameText (FrameData, fh, _) + | testEndStream (flags fh) = Just $ "DATA " ++ show (streamId fh) ++ " END_STREAM" + frameText _ = Nothing -- | A request for /slow, then GOAWAY(NO_ERROR) at once. What the server -- sends from then on until it closes the connection.