warp 3.1.3.1 → 3.1.4
raw patch · 19 files changed
+783/−523 lines, 19 filesdep ~http2dep ~wai
Dependency ranges changed: http2, wai
Files
- ChangeLog.md +6/−0
- Network/Wai/Handler/Warp.hs +11/−0
- Network/Wai/Handler/Warp/HTTP2.hs +4/−2
- Network/Wai/Handler/Warp/HTTP2/HPACK.hs +35/−13
- Network/Wai/Handler/Warp/HTTP2/Receiver.hs +7/−7
- Network/Wai/Handler/Warp/HTTP2/Request.hs +1/−1
- Network/Wai/Handler/Warp/HTTP2/Sender.hs +230/−237
- Network/Wai/Handler/Warp/HTTP2/Types.hs +78/−26
- Network/Wai/Handler/Warp/HTTP2/Worker.hs +152/−62
- Network/Wai/Handler/Warp/Internal.hs +11/−0
- Network/Wai/Handler/Warp/Recv.hs +1/−1
- Network/Wai/Handler/Warp/RequestHeader.hs +2/−28
- Network/Wai/Handler/Warp/Response.hs +41/−75
- Network/Wai/Handler/Warp/Run.hs +115/−31
- test/ExceptionSpec.hs +3/−18
- test/HTTP.hs +34/−0
- test/ResponseSpec.hs +37/−7
- test/RunSpec.hs +7/−8
- warp.cabal +8/−7
ChangeLog.md view
@@ -1,3 +1,9 @@+## 3.1.4++* Using newer http2 library to prevent change table size attacks.+* API for HTTP/2 server push and trailers. [#426](https://github.com/yesodweb/wai/pull/426)+* Preventing response splitting attacks. [#435](https://github.com/yesodweb/wai/pull/435)+ ## 3.1.3 * Warp now supports blaze-builder v0.4 or later only.
Network/Wai/Handler/Warp.hs view
@@ -38,10 +38,21 @@ -- module Network.Wai.Handler.Warp ( -- * Run a Warp server+ -- | All of these automatically serve the same 'Application' over HTTP\/1,+ -- HTTP\/1.1, and HTTP\/2. run , runEnv , runSettings , runSettingsSocket+ -- * Run an HTTP\/2-aware server+ -- | Each of these takes an HTTP\/2-aware application as well as a backup+ -- 'Application' to be used for HTTP\/1.1 and HTTP\/1 connections. These+ -- are only needed if your application needs access to HTTP\/2-specific+ -- features such as trailers or pushed streams.+ , runHTTP2+ , runHTTP2Env+ , runHTTP2Settings+ , runHTTP2SettingsSocket -- * Settings , Settings , defaultSettings
Network/Wai/Handler/Warp/HTTP2.hs view
@@ -7,9 +7,11 @@ import qualified Control.Exception as E import Control.Monad (when, unless, replicateM_) import Data.ByteString (ByteString)+ import Network.HTTP2 import Network.Socket (SockAddr)-import Network.Wai++import Network.Wai.HTTP2 (HTTP2Application) import Network.Wai.Handler.Warp.HTTP2.EncodeFrame import Network.Wai.Handler.Warp.HTTP2.Manager import Network.Wai.Handler.Warp.HTTP2.Receiver@@ -22,7 +24,7 @@ ---------------------------------------------------------------- -http2 :: Connection -> InternalInfo -> SockAddr -> Transport -> S.Settings -> (BufSize -> IO ByteString) -> Application -> IO ()+http2 :: Connection -> InternalInfo -> SockAddr -> Transport -> S.Settings -> (BufSize -> IO ByteString) -> HTTP2Application -> IO () http2 conn ii addr transport settings readN app = do checkTLS ok <- checkPreface
Network/Wai/Handler/Warp/HTTP2/HPACK.hs view
@@ -5,38 +5,48 @@ import Control.Arrow (first) import qualified Control.Exception as E+import qualified Data.ByteString as B import Data.ByteString.Builder (Builder) import qualified Data.ByteString.Char8 as B8 import Data.CaseInsensitive (foldedCase) import Data.IORef (readIORef, writeIORef)+import qualified Data.List as L import Network.HPACK import qualified Network.HTTP.Types as H import Network.HTTP2-import Network.Wai import Network.Wai.Handler.Warp.HTTP2.Types import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.Response import qualified Network.Wai.Handler.Warp.Settings as S import Network.Wai.Handler.Warp.Types -hpackEncodeHeader :: Context -> InternalInfo -> S.Settings -> Response- -> IO Builder-hpackEncodeHeader Context{encodeDynamicTable} ii settings rsp = do- hdr1 <- addServerAndDate hdr0+-- $setup+-- >>> :set -XOverloadedStrings++-- fixme: should we split "Set-Cookie:" headers?+hpackEncodeHeader :: Context -> InternalInfo -> S.Settings -> H.Status+ -> H.ResponseHeaders -> IO Builder+hpackEncodeHeader ctx ii settings s h = do+ hdr1 <- addServerAndDate h let hdr2 = (":status", status) : map (first foldedCase) hdr1- ehdrtbl <- readIORef encodeDynamicTable- (ehdrtbl', builder) <- encodeHeaderBuilder defaultEncodeStrategy ehdrtbl hdr2- writeIORef encodeDynamicTable ehdrtbl'- return builder+ hpackEncodeRawHeaders ctx hdr2 where- hdr0 = responseHeaders rsp- status = B8.pack $ show $ H.statusCode $ responseStatus rsp+ status = B8.pack $ show $ H.statusCode $ s dc = dateCacher ii- rspidxhdr = indexResponseHeader hdr0+ rspidxhdr = indexResponseHeader h defServer = S.settingsServerName settings addServerAndDate = addDate dc rspidxhdr . addServer defServer rspidxhdr +hpackEncodeCIHeaders :: Context -> [H.Header] -> IO Builder+hpackEncodeCIHeaders ctx = hpackEncodeRawHeaders ctx . map (first foldedCase) +hpackEncodeRawHeaders :: Context -> [(B.ByteString, B.ByteString)] -> IO Builder+hpackEncodeRawHeaders Context{encodeDynamicTable} hdr = do+ ehdrtbl <- readIORef encodeDynamicTable+ (ehdrtbl', builder) <- encodeHeaderBuilder defaultEncodeStrategy ehdrtbl hdr+ writeIORef encodeDynamicTable ehdrtbl'+ return builder+ ---------------------------------------------------------------- hpackDecodeHeader :: HeaderBlockFragment -> Context -> IO HeaderList@@ -44,6 +54,18 @@ hdrtbl <- readIORef decodeDynamicTable (hdrtbl', hdr) <- decodeHeader hdrtbl hdrblk `E.onException` cleanup writeIORef decodeDynamicTable hdrtbl'- return hdr+ return $ concatCookie hdr where cleanup = E.throwIO $ ConnectionError CompressionError "cannot decompress the header"++-- |+--+-- >>> concatCookie [("foo","bar")]+-- [("foo","bar")]+-- >>> concatCookie [("cookie","a=b"),("foo","bar"),("cookie","c=d"),("cookie","e=f")]+-- [("cookie","a=b; c=d; e=f"),("foo","bar")]+concatCookie :: HeaderList -> HeaderList+concatCookie hdr = case L.partition (\x -> fst x == "cookie") hdr of+ ([],_) -> hdr+ (cookies,others) -> let cookieValue = B.intercalate "; " (map snd cookies)+ in ("cookie",cookieValue) : others
Network/Wai/Handler/Warp/HTTP2/Receiver.hs view
@@ -56,7 +56,7 @@ cont <- processStreamGuardingError $ decodeFrameHeader hd when cont loop - processStreamGuardingError (_, FrameHeader{streamId})+ processStreamGuardingError (FrameHeaders, FrameHeader{streamId}) | isResponse streamId = E.throwIO $ ConnectionError ProtocolError "stream id should be odd" processStreamGuardingError (FrameUnknown _, FrameHeader{payloadLength}) = do mx <- readIORef continued@@ -161,15 +161,15 @@ -- idle, so that receiving frames on it is only a -- stream error. consume payloadLength- strm <- newStream streamId 0+ strm <- newStream concurrency streamId 0 writeIORef (streamState strm) $ Closed $ ResetByMe $ E.toException $ StreamError RefusedStream streamId insert streamTable streamId strm E.throwIO $ StreamError RefusedStream streamId ws <- initialWindowSize <$> readIORef http2settings- newstrm <- newStream streamId (fromIntegral ws)- when (ftyp == FrameHeaders) $ opened ctx newstrm+ newstrm <- newStream concurrency streamId (fromIntegral ws)+ when (ftyp == FrameHeaders) $ opened newstrm insert streamTable streamId newstrm return newstrm @@ -186,7 +186,7 @@ unless (testAck flags) $ do modifyIORef http2settings $ \old -> updateSettings old alist let frame = settingsFrame setAck []- enqueue outputQ (OFrame frame) highestPriority+ enqueue outputQ (OSettings frame alist) highestPriority return True control FramePing FrameHeader{flags} bs Context{outputQ} =@@ -310,10 +310,10 @@ atomically $ writeTVar streamWindow w return s -stream FrameRSTStream header bs ctx _ strm = do+stream FrameRSTStream header bs _ _ strm = do RSTStreamFrame e <- guardIt $ decoderstStreamFrame header bs let cc = Reset e- closed ctx strm cc+ closed strm cc return $ Closed cc -- will be written to streamState again stream FramePriority header bs Context{outputQ} s Stream{streamNumber,streamPriority} = do
Network/Wai/Handler/Warp/HTTP2/Request.hs view
@@ -37,7 +37,7 @@ vhMethod :: ByteString , vhPath :: ByteString , vhAuth :: Maybe ByteString- , vhCL :: Maybe Int+ , vhCL :: Maybe Int -- ^ Content-Length , vhHeader :: RequestHeaders }
Network/Wai/Handler/Warp/HTTP2/Sender.hs view
@@ -7,55 +7,71 @@ #if __GLASGOW_HASKELL__ < 709 import Control.Applicative #endif+import Control.Concurrent.MVar (putMVar) import Control.Concurrent.STM import qualified Control.Exception as E-import Control.Monad (unless, void, when)+import Control.Monad (void, when) import qualified Data.ByteString as BS+import qualified Data.ByteString.Builder as B (int32BE) import qualified Data.ByteString.Builder.Extra as B+import Data.Monoid ((<>)) import Foreign.Ptr+import qualified Network.HTTP.Types as H+import Network.HPACK (setLimitForEncoding) import Network.HTTP2 import Network.HTTP2.Priority-import Network.Wai+import Network.Wai (FilePart(..))+import Network.Wai.HTTP2 (Trailers, promiseHeaders) import Network.Wai.Handler.Warp.Buffer import Network.Wai.Handler.Warp.HTTP2.EncodeFrame import Network.Wai.Handler.Warp.HTTP2.HPACK import Network.Wai.Handler.Warp.HTTP2.Types import Network.Wai.Handler.Warp.IORef import qualified Network.Wai.Handler.Warp.Settings as S-import qualified Network.Wai.Handler.Warp.Timeout as T import Network.Wai.Handler.Warp.Types-import Network.Wai.Internal (Response(..))-import qualified System.PosixCompat.Files as P #ifdef WINDOWS import qualified System.IO as IO #else-import Network.Wai.Handler.Warp.FdCache+import Network.Wai.Handler.Warp.FdCache (getFd) import Network.Wai.Handler.Warp.SendFile (positionRead)+import qualified Network.Wai.Handler.Warp.Timeout as T import System.Posix.IO (openFd, OpenFileFlags(..), defaultFileFlags, OpenMode(ReadOnly), closeFd)-import System.Posix.Types+import System.Posix.Types (Fd) #endif ---------------------------------------------------------------- +-- | The platform-specific type of an open file to stream from. On Windows we+-- don't have pread, so this is just a Handle; on Unix platforms with pread,+-- this is a file descriptor supplied by the fd cache.+#ifdef WINDOWS+type OpenFile = IO.Handle+#else+type OpenFile = Fd+#endif+ data Leftover = LZero | LOne B.BufferWriter | LTwo BS.ByteString B.BufferWriter+ | LFile OpenFile Integer Integer (IO ()) ---------------------------------------------------------------- -unlessClosed :: Context -> Connection -> Stream -> IO () -> IO ()-unlessClosed ctx- Connection{connSendAll}+-- | Run the given action if the stream is not closed; handle any exceptions by+-- resetting the stream.+unlessClosed :: Connection -> Stream -> IO () -> IO Bool+unlessClosed Connection{connSendAll} strm@Stream{streamState,streamNumber} body = E.handle resetStream $ do state <- readIORef streamState- unless (isClosed state) body+ if (isClosed state) then return False else body >> return True where resetStream e = do- closed ctx strm (ResetByMe e)+ closed strm (ResetByMe e) let rst = resetFrame InternalError streamNumber connSendAll rst+ return False getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize getWindowSize connWindow strmWindow = do@@ -69,7 +85,7 @@ return $ min cw sw frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()-frameSender ctx@Context{outputQ,connectionWindow}+frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable} conn@Connection{connWriteBuffer,connBufferSize,connSendAll} ii settings = go `E.catch` ignore where@@ -90,67 +106,109 @@ switch OFinish = return () switch (OGoaway frame) = connSendAll frame+ switch (OSettings frame alist) = do+ connSendAll frame+ case lookup SettingsHeaderTableSize alist of+ Nothing -> return ()+ Just siz -> do+ dyntbl <- readIORef encodeDynamicTable+ setLimitForEncoding siz dyntbl+ loop switch (OFrame frame) = do connSendAll frame loop- switch (OResponse strm rsp aux) = do- unlessClosed ctx conn strm $ do- lim <- getWindowSize connectionWindow (streamWindow strm)- -- Header frame and Continuation frame- let sid = streamNumber strm- endOfStream = case aux of- Persist{} -> False- Oneshot hb -> not hb- len <- headerContinue sid rsp endOfStream- let total = len + frameHeaderLength- case aux of- Oneshot True -> do -- hasBody- -- Data frame payload- (off, _) <- sendHeadersIfNecessary total- let payloadOff = off + frameHeaderLength- Next datPayloadLen mnext <-- fillResponseBodyGetNext conn ii payloadOff lim rsp- fillDataHeaderSend strm total datPayloadLen mnext- maybeEnqueueNext strm mnext- Oneshot False -> do- -- "closed" must be before "connSendAll". If not,- -- the context would be switched to the receiver,- -- resulting the inconsistency of concurrency.- closed ctx strm Finished- flushN total- Persist sq tvar -> do- (off, needSend) <- sendHeadersIfNecessary total- let payloadOff = off + frameHeaderLength- Next datPayloadLen mnext <-- fillStreamBodyGetNext conn payloadOff lim sq tvar strm- -- If no data was immediately available, avoid sending an- -- empty data frame.- if datPayloadLen > 0 then- fillDataHeaderSend strm total datPayloadLen mnext- else- when needSend $ flushN off- maybeEnqueueNext strm mnext+ switch (OResponse strm s h aux) = do+ _ <- unlessClosed conn strm $+ getWindowSize connectionWindow (streamWindow strm) >>=+ sendResponse strm s h aux loop switch (ONext strm curr) = do- unlessClosed ctx conn strm $ do+ _ <- unlessClosed conn strm $ do lim <- getWindowSize connectionWindow (streamWindow strm) -- Data frame payload Next datPayloadLen mnext <- curr lim- fillDataHeaderSend strm 0 datPayloadLen mnext- maybeEnqueueNext strm mnext+ fillDataHeaderSend strm 0 datPayloadLen+ dispatchNext strm mnext loop+ switch (OPush oldStrm push mvar strm s h aux) = do+ pushed <- unlessClosed conn oldStrm $ do+ lim <- getWindowSize connectionWindow (streamWindow strm)+ -- Write and send the promise.+ builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push+ off <- pushContinue (streamNumber oldStrm) (streamNumber strm) builder+ flushN $ off + frameHeaderLength+ -- TODO(awpr): refactor sendResponse to be able to handle non-zero+ -- initial offsets and use that to potentially avoid the extra syscall.+ sendResponse strm s h aux lim+ putMVar mvar pushed+ loop + -- Send the response headers and as much of the response as is immediately+ -- available; shared by normal responses and pushed streams.+ sendResponse :: Stream -> H.Status -> H.ResponseHeaders -> Aux -> WindowSize -> IO ()+ sendResponse strm s h (Persist sq tvar) lim = do+ -- Header frame and Continuation frame+ let sid = streamNumber strm+ builder <- hpackEncodeHeader ctx ii settings s h+ len <- headerContinue sid builder False+ let total = len + frameHeaderLength+ (off, needSend) <- sendHeadersIfNecessary total+ let payloadOff = off + frameHeaderLength+ Next datPayloadLen mnext <-+ fillStreamBodyGetNext ii conn payloadOff lim sq tvar strm+ -- If no data was immediately available, avoid sending an+ -- empty data frame.+ if datPayloadLen > 0 then+ fillDataHeaderSend strm total datPayloadLen+ else+ when needSend $ flushN off+ dispatchNext strm mnext++ -- Send the stream's trailers and close the stream.+ sendTrailers :: Stream -> Trailers -> IO ()+ sendTrailers strm trailers = do+ -- Trailers always indicate the end of a stream; send them in+ -- consecutive header+continuation frames and end the stream. Some+ -- clients dislike empty headers frames, so end the stream with an+ -- empty data frame instead, as recommended by the spec.+ toFlush <- case trailers of+ [] -> frameHeaderLength <$ fillFrameHeader FrameData 0+ (streamNumber strm)+ (setEndStream defaultFlags)+ connWriteBuffer+ _ -> do+ builder <- hpackEncodeCIHeaders ctx trailers+ off <- headerContinue (streamNumber strm) builder True+ return (off + frameHeaderLength)+ -- 'closed' must be before 'flushN'. If not, the context would be+ -- switched to the receiver, resulting in the inconsistency of+ -- concurrency.+ closed strm Finished+ flushN toFlush+ -- Flush the connection buffer to the socket, where the first 'n' bytes of -- the buffer are filled. flushN :: Int -> IO () flushN n = bufferIO connWriteBuffer n connSendAll - headerContinue sid rsp endOfStream = do- builder <- hpackEncodeHeader ctx ii settings rsp+ -- A flags value with the end-header flag set iff the argument is B.Done.+ maybeEndHeaders B.Done = setEndHeader defaultFlags+ maybeEndHeaders _ = defaultFlags++ -- Write PUSH_PROMISE and possibly CONTINUATION frames into the connection+ -- buffer, using the given builder as their contents; flush them to the+ -- socket as necessary.+ pushContinue sid newSid builder = do+ let builder' = B.int32BE (fromIntegral newSid) <> builder+ (len, signal) <- B.runBuilder builder' bufHeaderPayload headerPayloadLim+ let flag = maybeEndHeaders signal+ fillFrameHeader FramePushPromise len sid flag connWriteBuffer+ continue sid len signal++ -- Write HEADER and possibly CONTINUATION frames.+ headerContinue sid builder endOfStream = do (len, signal) <- B.runBuilder builder bufHeaderPayload headerPayloadLim- let flag0 = case signal of- B.Done -> setEndHeader defaultFlags- _ -> defaultFlags+ let flag0 = maybeEndHeaders signal flag = if endOfStream then setEndStream flag0 else flag0 fillFrameHeader FrameHeaders len sid flag connWriteBuffer continue sid len signal@@ -159,9 +217,7 @@ continue sid len (B.More _ writer) = do flushN $ len + frameHeaderLength (len', signal') <- writer bufHeaderPayload headerPayloadLim- let flag = case signal' of- B.Done -> setEndHeader defaultFlags- _ -> defaultFlags+ let flag = maybeEndHeaders signal' fillFrameHeader FrameContinuation len' sid flag connWriteBuffer continue sid len' signal' continue sid len (B.Chunk bs writer) = do@@ -178,16 +234,22 @@ -- True if the connection buffer has room for a 1-byte data frame. canFitDataFrame total = total + frameHeaderLength < connBufferSize - -- Re-enqueue the stream in the output queue if more output is immediately- -- available; do nothing otherwise.- maybeEnqueueNext :: Stream -> Control DynaNext -> IO ()- maybeEnqueueNext strm (CNext next) = do+ -- Take the appropriate action based on the given 'Control':+ -- - If more output is immediately available, re-enqueue the stream in the+ -- output queue.+ -- - If the output is over and trailers are available, send them now and+ -- end the stream.+ -- - If we've drained the queue and handed the stream back to its waiter,+ -- do nothing.+ --+ -- This is done after sending any part of the stream body, so it's shared+ -- by 'sendResponse' and @switch (ONext ...)@.+ dispatchNext :: Stream -> Control DynaNext -> IO ()+ dispatchNext _ CNone = return ()+ dispatchNext strm (CFinish trailers) = sendTrailers strm trailers+ dispatchNext strm (CNext next) = do let out = ONext strm next enqueueOrSpawnTemporaryWaiter strm outputQ out- -- If the streaming is not finished, it must already have been- -- written to the 'TVar' owned by 'waiter', which will- -- put it back into the queue when more output becomes available.- maybeEnqueueNext _ _ = return () -- Send headers if there is not room for a 1-byte data frame, and return@@ -199,21 +261,12 @@ flushN total return (0, False) - fillDataHeaderSend strm otherLen datPayloadLen mnext = do+ fillDataHeaderSend strm otherLen datPayloadLen = do -- Data frame header let sid = streamNumber strm buf = connWriteBuffer `plusPtr` otherLen total = otherLen + frameHeaderLength + datPayloadLen- flag = case mnext of- CFinish -> setEndStream defaultFlags- _ -> defaultFlags- fillFrameHeader FrameData datPayloadLen sid flag buf- -- "closed" must be before "connSendAll". If not,- -- the context would be switched to the receiver,- -- resulting the inconsistency of concurrency.- case mnext of- CFinish -> closed ctx strm Finished- _ -> return ()+ fillFrameHeader FrameData datPayloadLen sid defaultFlags buf flushN total atomically $ do modifyTVar' connectionWindow (subtract datPayloadLen)@@ -225,130 +278,106 @@ ---------------------------------------------------------------- -{--ResponseFile Status ResponseHeaders FilePath (Maybe FilePart)-ResponseBuilder Status ResponseHeaders Builder-ResponseStream Status ResponseHeaders StreamingBody-ResponseRaw (IO ByteString -> (ByteString -> IO ()) -> IO ()) Response--}--fillResponseBodyGetNext :: Connection -> InternalInfo -> Int -> WindowSize -> Response -> IO Next-fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}- _ off lim (ResponseBuilder _ _ bb) = do+fillStreamBodyGetNext :: InternalInfo -> Connection -> Int -> WindowSize+ -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next+fillStreamBodyGetNext ii Connection{connWriteBuffer,connBufferSize}+ off lim sq tvar strm = do let datBuf = connWriteBuffer `plusPtr` off room = min (connBufferSize - off) lim- (len, signal) <- B.runBuilder bb datBuf room- return $ nextForBuilder connWriteBuffer connBufferSize len signal+ (leftover, cont, len) <- runStreamBuilder ii datBuf room sq+ nextForStream ii connWriteBuffer connBufferSize sq tvar strm leftover cont len +----------------------------------------------------------------++runStreamBuilder :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence+ -> IO (Leftover, Maybe Trailers, BytesFilled)+runStreamBuilder ii buf0 room0 sq = loop buf0 room0 0+ where+ loop !buf !room !total = do+ mbuilder <- atomically $ tryReadTBQueue sq+ case mbuilder of+ Nothing -> return (LZero, Nothing, total)+ Just (SBuilder builder) -> do+ (len, signal) <- B.runBuilder builder buf room+ let !total' = total + len+ case signal of+ B.Done -> loop (buf `plusPtr` len) (room - len) total'+ B.More _ writer -> return (LOne writer, Nothing, total')+ B.Chunk bs writer -> return (LTwo bs writer, Nothing, total')+ Just (SFile path part) -> do+ (leftover, len) <- runStreamFile ii buf room path part+ let !total' = total + len+ return (leftover, Nothing, total')+ -- TODO if file part is done, go back to loop+ Just SFlush -> return (LZero, Nothing, total)+ Just (SFinish trailers) -> return (LZero, Just trailers, total)++-- | Open the file and start reading into the send buffer.+runStreamFile :: InternalInfo -> Buffer -> BufSize -> FilePath -> FilePart+ -> IO (Leftover, BytesFilled)++-- | Read the given (OS-specific) file representation into the buffer. On+-- non-Windows systems this uses pread; on Windows this ignores the position+-- because we use the Handle's internal read position instead (because it's not+-- potentially shared with other readers).+readOpenFile :: OpenFile -> Buffer -> BufSize -> Integer -> IO Int+ #ifdef WINDOWS-fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}- _ off lim (ResponseFile _ _ path mpart) = do- let datBuf = connWriteBuffer `plusPtr` off- room = min (connBufferSize - off) lim- (start, bytes) <- fileStartEnd path mpart+runStreamFile _ buf room path part = do+ let start = filePartOffset part+ bytes = filePartByteCount part -- fixme: how to close Handle? GC does it at this moment. h <- IO.openBinaryFile path IO.ReadMode IO.hSeek h IO.AbsoluteSeek start- len <- IO.hGetBufSome h datBuf (mini room bytes)- let bytes' = bytes - fromIntegral len- return $ nextForFile len connWriteBuffer connBufferSize h bytes' (return ())+ fillBufFile buf room h start bytes (return ())++readOpenFile h buf room _ = IO.hGetBufSome h buf room #else-fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}- ii off lim (ResponseFile _ _ path mpart) = do+runStreamFile ii buf room path part = do+ let start = filePartOffset part+ bytes = filePartByteCount part (fd, refresh) <- case fdCacher ii of Just fdcache -> getFd fdcache path Nothing -> do fd' <- openFd path ReadOnly Nothing defaultFileFlags{nonBlock=True} th <- T.register (timeoutManager ii) (closeFd fd') return (fd', T.tickle th)- let datBuf = connWriteBuffer `plusPtr` off- room = min (connBufferSize - off) lim- (start, bytes) <- fileStartEnd path mpart- len <- positionRead fd datBuf (mini room bytes) start- refresh- let len' = fromIntegral len- return $ nextForFile len connWriteBuffer connBufferSize fd (start + len') (bytes - len') refresh-#endif--fillResponseBodyGetNext _ _ _ _ _ = error "fillResponseBodyGetNext"--fileStartEnd :: FilePath -> Maybe FilePart -> IO (Integer, Integer)-fileStartEnd path Nothing = do- end <- fromIntegral . P.fileSize <$> P.getFileStatus path- return (0, end)-fileStartEnd _ (Just part) =- return (filePartOffset part, filePartByteCount part)--------------------------------------------------------------------fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next-fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}- off lim sq tvar strm = do- let datBuf = connWriteBuffer `plusPtr` off- room = min (connBufferSize - off) lim- (leftover, cont, len) <- runStreamBuilder datBuf room sq- nextForStream connWriteBuffer connBufferSize sq tvar strm leftover cont len--------------------------------------------------------------------fillBufBuilder :: Buffer -> BufSize -> Leftover -> DynaNext-fillBufBuilder buf0 siz0 leftover lim = do- let payloadBuf = buf0 `plusPtr` frameHeaderLength- room = min (siz0 - frameHeaderLength) lim- case leftover of- LZero -> error "fillBufBuilder: LZero"- LOne writer -> do- (len, signal) <- writer payloadBuf room- getNext len signal- LTwo bs writer- | BS.length bs <= room -> do- buf1 <- copy payloadBuf bs- let len1 = BS.length bs- (len2, signal) <- writer buf1 (room - len1)- getNext (len1 + len2) signal- | otherwise -> do- let (bs1,bs2) = BS.splitAt room bs- void $ copy payloadBuf bs1- getNext room (B.Chunk bs2 writer)- where- getNext l s = return $ nextForBuilder buf0 siz0 l s+ fillBufFile buf room fd start bytes refresh -nextForBuilder :: Buffer -> BufSize -> BytesFilled -> B.Next -> Next-nextForBuilder _ _ len B.Done- = Next len CFinish-nextForBuilder buf siz len (B.More _ writer)- = Next len (CNext (fillBufBuilder buf siz (LOne writer)))-nextForBuilder buf siz len (B.Chunk bs writer)- = Next len (CNext (fillBufBuilder buf siz (LTwo bs writer)))+readOpenFile = positionRead+#endif -----------------------------------------------------------------+-- | Read as much of the file as is currently available into the buffer, then+-- return a 'Leftover' to indicate whether this file chunk has more data to+-- send. If this read hit the end of the file range, return 'LZero'; otherwise+-- return 'LFile' so this stream will continue reading from the file the next+-- time it's pulled from the queue.+fillBufFile :: Buffer -> BufSize -> OpenFile -> Integer -> Integer -> (IO ())+ -> IO (Leftover, BytesFilled)+fillBufFile buf room f start bytes refresh = do+ len <- readOpenFile f buf (mini room bytes) start+ refresh+ let len' = fromIntegral len+ leftover = if bytes > len' then+ LFile f (start + len') (bytes - len') refresh+ else+ LZero+ return (leftover, len) -runStreamBuilder :: Buffer -> BufSize -> TBQueue Sequence- -> IO (Leftover, Bool, BytesFilled)-runStreamBuilder buf0 room0 sq = loop buf0 room0 0- where- loop !buf !room !total = do- mbuilder <- atomically $ tryReadTBQueue sq- case mbuilder of- Nothing -> return (LZero, True, total)- Just (SBuilder builder) -> do- (len, signal) <- B.runBuilder builder buf room- let !total' = total + len- case signal of- B.Done -> loop (buf `plusPtr` len) (room - len) total'- B.More _ writer -> return (LOne writer, True, total')- B.Chunk bs writer -> return (LTwo bs writer, True, total')- Just SFlush -> return (LZero, True, total)- Just SFinish -> return (LZero, False, total)+mini :: Int -> Integer -> Int+mini i n+ | fromIntegral i < n = i+ | otherwise = fromIntegral n -fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext-fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do+fillBufStream :: InternalInfo -> Buffer -> BufSize -> Leftover+ -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext+fillBufStream ii buf0 siz0 leftover0 sq tvar strm lim0 = do let payloadBuf = buf0 `plusPtr` frameHeaderLength room0 = min (siz0 - frameHeaderLength) lim0 case leftover0 of LZero -> do- (leftover, cont, len) <- runStreamBuilder payloadBuf room0 sq- getNext leftover cont len+ (leftover, end, len) <- runStreamBuilder ii payloadBuf room0 sq+ getNext leftover end len LOne writer -> write writer payloadBuf room0 0 LTwo bs writer | BS.length bs <= room0 -> do@@ -358,71 +387,35 @@ | otherwise -> do let (bs1,bs2) = BS.splitAt room0 bs void $ copy payloadBuf bs1- getNext (LTwo bs2 writer) True room0+ getNext (LTwo bs2 writer) Nothing room0+ LFile fd start bytes refresh -> do+ (leftover, len) <- fillBufFile payloadBuf room0 fd start bytes refresh+ getNext leftover Nothing len where- getNext = nextForStream buf0 siz0 sq tvar strm+ getNext = nextForStream ii buf0 siz0 sq tvar strm write writer1 buf room sofar = do (len, signal) <- writer1 buf room case signal of B.Done -> do- (leftover, cont, extra) <- runStreamBuilder (buf `plusPtr` len) (room - len) sq+ (leftover, end, extra) <- runStreamBuilder ii (buf `plusPtr` len) (room - len) sq let !total = sofar + len + extra- getNext leftover cont total+ getNext leftover end total B.More _ writer -> do let !total = sofar + len- getNext (LOne writer) True total+ getNext (LOne writer) Nothing total B.Chunk bs writer -> do let !total = sofar + len- getNext (LTwo bs writer) True total+ getNext (LTwo bs writer) Nothing total -nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync -> Stream- -> Leftover -> Bool -> BytesFilled- -> IO Next-nextForStream _ _ _ tvar _ _ False len = do- atomically $ writeTVar tvar SyncFinish- return $ Next len CFinish-nextForStream buf siz sq tvar strm LZero True len = do- let out = ONext strm (fillBufStream buf siz LZero sq tvar strm)+nextForStream :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence+ -> TVar Sync -> Stream -> Leftover -> Maybe Trailers+ -> BytesFilled -> IO Next+nextForStream _ _ _ _ tvar _ _ (Just trailers) len = do+ atomically $ writeTVar tvar $ SyncFinish+ return $ Next len $ CFinish trailers+nextForStream ii buf siz sq tvar strm LZero Nothing len = do+ let out = ONext strm (fillBufStream ii buf siz LZero sq tvar strm) atomically $ writeTVar tvar $ SyncNext out return $ Next len CNone-nextForStream buf siz sq tvar strm leftover True len =- return $ Next len (CNext (fillBufStream buf siz leftover sq tvar strm))--------------------------------------------------------------------#ifdef WINDOWS-fillBufFile :: Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> DynaNext-fillBufFile buf siz h bytes refresh lim = do- let payloadBuf = buf `plusPtr` frameHeaderLength- room = min (siz - frameHeaderLength) lim- len <- IO.hGetBufSome h payloadBuf room- refresh- let bytes' = bytes - fromIntegral len- return $ nextForFile len buf siz h bytes' refresh--nextForFile :: BytesFilled -> Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> Next-nextForFile 0 _ _ _ _ _ = Next 0 CFinish-nextForFile len _ _ _ 0 _ = Next len CFinish-nextForFile len buf siz h bytes refresh =- Next len (CNext (fillBufFile buf siz h bytes refresh))-#else-fillBufFile :: Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> DynaNext-fillBufFile buf siz fd start bytes refresh lim = do- let payloadBuf = buf `plusPtr` frameHeaderLength- room = min (siz - frameHeaderLength) lim- len <- positionRead fd payloadBuf (mini room bytes) start- let len' = fromIntegral len- refresh- return $ nextForFile len buf siz fd (start + len') (bytes - len') refresh--nextForFile :: BytesFilled -> Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> Next-nextForFile 0 _ _ _ _ _ _ = Next 0 CFinish-nextForFile len _ _ _ _ 0 _ = Next len CFinish-nextForFile len buf siz fd start bytes refresh =- Next len (CNext (fillBufFile buf siz fd start bytes refresh))-#endif--mini :: Int -> Integer -> Int-mini i n- | fromIntegral i < n = i- | otherwise = fromIntegral n+nextForStream ii buf siz sq tvar strm leftover Nothing len =+ return $ Next len (CNext (fillBufStream ii buf siz leftover sq tvar strm))
Network/Wai/Handler/Warp/HTTP2/Types.hs view
@@ -5,9 +5,10 @@ import Data.ByteString.Builder (Builder) #if __GLASGOW_HASKELL__ < 709-import Control.Applicative ((<$>),(<*>))+import Control.Applicative ((<$>), (<*>), pure) #endif import Control.Concurrent (forkIO)+import Control.Concurrent.MVar (MVar) import Control.Concurrent.STM import Control.Exception (SomeException) import Control.Monad (void)@@ -17,7 +18,8 @@ import Data.IntMap.Strict (IntMap, IntMap) import qualified Data.IntMap.Strict as M import qualified Network.HTTP.Types as H-import Network.Wai (Request, Response)+import Network.Wai (Request, FilePart)+import Network.Wai.HTTP2 (PushPromise, Trailers) import Network.Wai.Handler.Warp.IORef import Network.Wai.Handler.Warp.Types @@ -36,7 +38,7 @@ where useHTTP2 = case tlsNegotiatedProtocol tls of Nothing -> False- Just proto -> "h2-" `BS.isPrefixOf` proto+ Just proto -> "h2-" `BS.isPrefixOf` proto || proto == "h2" ---------------------------------------------------------------- @@ -44,14 +46,21 @@ ---------------------------------------------------------------- -data Control a = CFinish+-- | The result of writing data from a stream's queue into the buffer.+data Control a = CFinish Trailers+ -- ^ The stream has ended, and the trailers should be sent. | CNext a+ -- ^ The stream has more data immediately available, and we+ -- should re-enqueue it when the stream window becomes open. | CNone+ -- ^ The stream queue has been drained and we've handed it off+ -- to its dedicated waiter thread, which will re-enqueue it when+ -- more data is available. instance Show (Control a) where- show CFinish = "CFinish"- show (CNext _) = "CNext"- show CNone = "CNone"+ show (CFinish _) = "CFinish"+ show (CNext _) = "CNext"+ show CNone = "CNone" type DynaNext = WindowSize -> IO Next @@ -60,28 +69,54 @@ data Next = Next BytesFilled (Control DynaNext) data Output = OFinish+ -- ^ Terminate the connection. | OGoaway ByteString+ -- ^ Send a goaway frame and terminate the connection.+ | OSettings ByteString SettingsList+ -- ^ Update settings and send an ack settings frame. | OFrame ByteString- | OResponse Stream Response Aux+ -- ^ Send an entire pre-encoded frame.+ | OResponse Stream H.Status H.ResponseHeaders Aux+ -- ^ Send the headers and as much of the response as is immediately+ -- available.+ | OPush Stream PushPromise (MVar Bool) Stream H.Status H.ResponseHeaders Aux+ -- ^ Send a PUSH_PROMISE frame, then act like OResponse; signal the+ -- MVar whether the promise has been sent. | ONext Stream DynaNext+ -- ^ Send a chunk of the response. outputStream :: Output -> Stream-outputStream (OResponse strm _ _) = strm-outputStream (ONext strm _) = strm-outputStream _ = error "outputStream"+outputStream (OResponse strm _ _ _) = strm+outputStream (ONext strm _) = strm+outputStream (OPush strm _ _ _ _ _ _) = strm+outputStream _ = error "outputStream" ---------------------------------------------------------------- -data Sequence = SFinish+-- | An element on the queue between a running stream and the sender; the order+-- should consist of any number of 'SFile', 'SBuilder', and 'SFlush', followed+-- by a single 'SFinish'.+data Sequence = SFinish Trailers+ -- ^ The stream is over; its trailers are provided. | SFlush+ -- ^ Any buffered data should be sent immediately. | SBuilder Builder+ -- ^ Append a chunk of data to the stream.+ | SFile FilePath FilePart+ -- ^ Append a chunk of a file's contents to the stream. +-- | A message from the sender to a stream's dedicated waiter thread. data Sync = SyncNone+ -- ^ Nothing interesting has happened. Go back to sleep. | SyncFinish+ -- ^ The stream has ended. | SyncNext Output+ -- ^ The stream's queue has been drained; wait for more to be+ -- available and re-enqueue the given 'Output'. -data Aux = Oneshot Bool- | Persist (TBQueue Sequence) (TVar Sync)+-- | Auxiliary information needed to communicate with a running stream: a queue+-- of stream elements ('Sequence') and a 'TVar' connected to its waiter thread.+data Aux = Persist (TBQueue Sequence) (TVar Sync) ---------------------------------------------------------------- @@ -89,13 +124,21 @@ data Context = Context { http2settings :: IORef Settings , streamTable :: StreamTable+ -- | Number of active streams initiated by the client; for enforcing our own+ -- max concurrency setting. , concurrency :: IORef Int+ -- | Number of active streams initiated by the server; for respecting the+ -- client's max concurrency setting.+ , pushConcurrency :: IORef Int -- | RFC 7540 says "Other frames (from any stream) MUST NOT -- occur between the HEADERS frame and any CONTINUATION -- frames that might follow". This field is used to implement -- this requirement. , continued :: IORef (Maybe StreamId) , currentStreamId :: IORef StreamId+ -- ^ Last client-initiated stream ID we've handled.+ , nextPushStreamId :: IORef StreamId+ -- ^ Next available server-initiated stream ID. , inputQ :: TQueue Input , outputQ :: PriorityTree Output , encodeDynamicTable :: IORef DynamicTable@@ -109,8 +152,10 @@ newContext = Context <$> newIORef defaultSettings <*> initialize 10 -- fixme: hard coding: 10 <*> newIORef 0+ <*> newIORef 0 <*> newIORef Nothing <*> newIORef 0+ <*> newIORef 2 -- first server push stream; 0 is reserved <*> newTQueueIO <*> newPriorityTree <*> (newDynamicTableForEncoding defaultDynamicTableSize >>= newIORef)@@ -177,28 +222,35 @@ , streamBodyLength :: IORef Int , streamWindow :: TVar WindowSize , streamPriority :: IORef Priority+ -- | The concurrency IORef in which this stream has been counted. The client+ -- and server each have separate concurrency values to respect, so pushed+ -- streams need to decrement a different count when they're closed. This+ -- should be either @concurrency ctx@ or @pushConcurrency ctx@.+ , concurrencyRef :: IORef Int } instance Show Stream where show s = show (streamNumber s) -newStream :: StreamId -> WindowSize -> IO Stream-newStream sid win = Stream sid <$> newIORef Idle- <*> newIORef Nothing- <*> newIORef 0- <*> newTVarIO win- <*> newIORef defaultPriority+newStream :: IORef Int -> StreamId -> WindowSize -> IO Stream+newStream ref sid win =+ Stream sid <$> newIORef Idle+ <*> newIORef Nothing+ <*> newIORef 0+ <*> newTVarIO win+ <*> newIORef defaultPriority+ <*> pure ref ---------------------------------------------------------------- -opened :: Context -> Stream -> IO ()-opened Context{concurrency} Stream{streamState} = do- atomicModifyIORef' concurrency (\x -> (x+1,()))+opened :: Stream -> IO ()+opened Stream{concurrencyRef,streamState} = do+ atomicModifyIORef' concurrencyRef (\x -> (x+1,())) writeIORef streamState (Open JustOpened) -closed :: Context -> Stream -> ClosedCode -> IO ()-closed Context{concurrency} Stream{streamState} cc = do- atomicModifyIORef' concurrency (\x -> (x-1,()))+closed :: Stream -> ClosedCode -> IO ()+closed Stream{concurrencyRef,streamState} cc = do+ atomicModifyIORef' concurrencyRef (\x -> (x-1,())) writeIORef streamState (Closed cc) ----------------------------------------------------------------
Network/Wai/Handler/Warp/HTTP2/Worker.hs view
@@ -1,10 +1,11 @@ {-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE RankNTypes #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE PatternGuards #-} {-# LANGUAGE CPP #-} module Network.Wai.Handler.Warp.HTTP2.Worker (- Responder+ Respond , response , worker ) where@@ -26,65 +27,152 @@ import Network.Wai.Handler.Warp.HTTP2.Manager import Network.Wai.Handler.Warp.HTTP2.Types import Network.Wai.Handler.Warp.IORef-import qualified Network.Wai.Handler.Warp.Response as R+import Network.Wai.HTTP2+ ( Chunk(..)+ , HTTP2Application+ , PushPromise+ , Responder(runResponder)+ , RespondFunc+ ) import qualified Network.Wai.Handler.Warp.Settings as S import qualified Network.Wai.Handler.Warp.Timeout as T-import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceived(..)) ---------------------------------------------------------------- --- | The wai definition is 'type Application = Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived'.--- This type implements the second argument (Response -> IO ResponseReceived)--- with extra arguments.-type Responder = ThreadContinue -> T.Handle -> Stream -> Request ->- Response -> IO ResponseReceived+-- | An 'HTTP2Application' takes a function of status, headers, trailers, and+-- body; this type implements that by currying some internal arguments.+--+-- The token type of the RespondFunc is set to be (). This is a bit+-- anti-climactic, but the real benefit of the token type is that the+-- application is forced to call the responder, and making it a boring type+-- doesn't break that property.+--+-- This is the argument to a 'Responder'.+type Respond = IO () -> Stream -> RespondFunc () --- | This function is passed to workers.--- They also pass 'Response's from 'Application's to this function.--- This function enqueues commands for the HTTP/2 sender.-response :: Context -> Manager -> Responder-response Context{outputQ} mgr tconf th strm req rsp = do- case rsp of- ResponseStream _ _ strmbdy -> do- -- We must not exit this WAI application.- -- If the application exits, streaming would be also closed.- -- So, this work occupies this thread.- --- -- We need to increase the number of workers.- myThreadId >>= replaceWithAction mgr- -- After this work, this thread stops to decease- -- the number of workers.- setThreadContinue tconf False- -- Since 'StreamingBody' is loop, we cannot control it.- -- So, let's serialize 'Builder' with a designated queue.- sq <- newTBQueueIO 10 -- fixme: hard coding: 10- tvar <- newTVarIO SyncNone- let out = OResponse strm rsp (Persist sq tvar)- -- Since we must not enqueue an empty queue to the priority- -- queue, we spawn a thread to ensure that the designated- -- queue is not empty.- void $ forkIO $ waiter tvar sq outputQ- atomically $ writeTVar tvar (SyncNext out)- let push b = do- atomically $ writeTBQueue sq (SBuilder b)- T.tickle th- flush = atomically $ writeTBQueue sq SFlush- strmbdy push flush- atomically $ writeTBQueue sq SFinish- _ -> do- setThreadContinue tconf True- let hasBody = requestMethod req /= H.methodHead- && R.hasBody (responseStatus rsp)- out = OResponse strm rsp (Oneshot hasBody)- enqueueOrSpawnTemporaryWaiter strm outputQ out- return ResponseReceived+-- | This function is passed to workers. They also pass responses from+-- 'HTTP2Application's to this function. This function enqueues commands for+-- the HTTP/2 sender.+response :: Context -> Manager -> ThreadContinue -> Respond+response ctx mgr tconf tickle strm s h strmbdy = do+ -- TODO(awpr) HEAD requests will still stream. + -- We must not exit this WAI application.+ -- If the application exits, streaming would be also closed.+ -- So, this work occupies this thread.+ --+ -- We need to increase the number of workers.+ myThreadId >>= replaceWithAction mgr+ -- After this work, this thread stops to decrease the number of workers.+ setThreadContinue tconf False++ runStream ctx OResponse tickle strm s h strmbdy++-- | Set up a waiter thread and run the stream body with functions to enqueue+-- 'Sequence's on the stream's queue.+runStream :: Context+ -> (Stream -> H.Status -> H.ResponseHeaders -> Aux -> Output)+ -> Respond+runStream Context{outputQ} mkOutput tickle strm s h strmbdy = do+ -- Since 'Body' is loop, we cannot control it.+ -- So, let's serialize 'Builder' with a designated queue.+ sq <- newTBQueueIO 10 -- fixme: hard coding: 10+ tvar <- newTVarIO SyncNone+ let out = mkOutput strm s h (Persist sq tvar)+ -- Since we must not enqueue an empty queue to the priority+ -- queue, we spawn a thread to ensure that the designated+ -- queue is not empty.+ void $ forkIO $ waiter tvar sq strm outputQ+ atomically $ writeTVar tvar (SyncNext out)+ let write chunk = do+ atomically $ writeTBQueue sq $ case chunk of+ BuilderChunk b -> SBuilder b+ FileChunk path part -> SFile path part+ tickle+ flush = atomically $ writeTBQueue sq SFlush+ trailers <- strmbdy write flush+ atomically $ writeTBQueue sq $ SFinish trailers++-- | Handle abnormal termination of a stream: mark it as closed, send a reset+-- frame, and call the user's 'settingsOnException' handler if applicable.+cleanupStream :: Context -> S.Settings -> Stream -> Maybe Request -> Maybe SomeException -> IO ()+cleanupStream Context{outputQ} set strm req me = do+ closed strm Killed+ let frame = resetFrame InternalError (streamNumber strm)+ enqueue outputQ (OFrame frame) highestPriority+ case me of+ Nothing -> return ()+ Just e -> S.settingsOnException set req e++-- | Push the given 'Responder' to the client if the settings allow it+-- (specifically 'enablePush' and 'maxConcurrentStreams'). Returns 'True' if+-- the stream was actually pushed.+--+-- This is the push function given to an 'HTTP2Application'.+pushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool+pushResponder ctx set strm promise responder = do+ let Context{ http2settings+ , pushConcurrency+ } = ctx+ cnt <- readIORef pushConcurrency+ settings <- readIORef http2settings+ let enabled = enablePush settings+ fits = maybe True (cnt <) $ maxConcurrentStreams settings+ canPush = fits && enabled+ if canPush then+ actuallyPushResponder ctx set strm promise responder+ else+ return False++-- | Set up a pushed stream and run the 'Responder' in its own thread. Waits+-- for the sender thread to handle the push request. This can fail to push the+-- stream and return 'False' if the sender dequeued the push request after the+-- associated stream was closed.+actuallyPushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool+actuallyPushResponder ctx set strm promise responder = do+ let Context{ http2settings+ , nextPushStreamId+ , pushConcurrency+ , streamTable+ } = ctx+ -- Claim the next outgoing stream.+ newSid <- atomicModifyIORef nextPushStreamId $ \sid -> (sid+2, sid)+ ws <- initialWindowSize <$> readIORef http2settings++ newStrm <- newStream pushConcurrency newSid ws+ -- Section 5.3.5 of RFC 7540 defines the weight of push promise is 16.+ -- But we need not to follow the spec. So, this value would change+ -- if necessary.+ writeIORef (streamPriority newStrm) $+ defaultPriority { streamDependency = streamNumber strm }+ opened newStrm+ insert streamTable newSid newStrm++ -- Set up a channel for the sender to report back whether it pushed the+ -- stream.+ mvar <- newEmptyMVar++ let mkOutput = OPush strm promise mvar+ tickle = return ()+ respond = runStream ctx mkOutput++ -- TODO(awpr): synthesize a Request for 'settingsOnException'?+ _ <- forkIO $ runResponder responder (respond tickle newStrm) `E.catch`+ (cleanupStream ctx set strm Nothing . Just)++ takeMVar mvar+ data Break = Break deriving (Show, Typeable) instance Exception Break -worker :: Context -> S.Settings -> T.Manager -> Application -> Responder -> IO ()-worker ctx@Context{inputQ,outputQ} set tm app responder = do+worker :: Context+ -> S.Settings+ -> T.Manager+ -> HTTP2Application+ -> (ThreadContinue -> Respond)+ -> IO ()+worker ctx@Context{inputQ} set tm app respond = do tid <- myThreadId sinfo <- newStreamInfo tcont <- newThreadContinue@@ -93,15 +181,17 @@ where go sinfo tcont th = do setThreadContinue tcont True+ ex <- E.try $ do T.pause th Input strm req <- atomically $ readTQueue inputQ setStreamInfo sinfo strm req T.resume th T.tickle th- app req $ responder tcont th strm req+ let responder = app req $ pushResponder ctx set strm+ runResponder responder $ respond tcont (T.tickle th) strm cont1 <- case ex of- Right ResponseReceived -> return True+ Right () -> return True Left e@(SomeException _) | Just Break <- E.fromException e -> do cleanup sinfo Nothing@@ -120,16 +210,16 @@ case m of Nothing -> return () Just (strm,req) -> do- closed ctx strm Killed- let frame = resetFrame InternalError (streamNumber strm)- enqueue outputQ (OFrame frame) highestPriority- case me of- Nothing -> return ()- Just e -> S.settingsOnException set (Just req) e+ cleanupStream ctx set strm (Just req) me clearStreamInfo sinfo -waiter :: TVar Sync -> TBQueue Sequence -> PriorityTree Output -> IO ()-waiter tvar sq outQ = do+-- | A dedicated waiter thread to re-enqueue the stream in the priority tree+-- whenever output becomes available. When the sender drains the queue and+-- moves on to another stream, it drops a message in the 'TVar', and this+-- thread wakes up, waits for more output to become available, and re-enqueues+-- the stream.+waiter :: TVar Sync -> TBQueue Sequence -> Stream -> PriorityTree Output -> IO ()+waiter tvar sq strm outQ = do -- waiting for actions other than SyncNone mx <- atomically $ do mout <- readTVar tvar@@ -140,7 +230,7 @@ return $ Just out SyncFinish -> return Nothing case mx of- Nothing -> return ()+ Nothing -> return () Just out -> do -- ensuring that the streaming queue is not empty. atomically $ do@@ -148,14 +238,14 @@ when isEmpty retry -- ensuring that stream window is greater than 0. enqueueWhenWindowIsOpen outQ out- waiter tvar sq outQ+ waiter tvar sq strm outQ ---------------------------------------------------------------- -- | It would nice if responders could return values to workers. -- Unfortunately, 'ResponseReceived' is already defined in WAI 2.0. -- It is not wise to change this type.--- So, a reference is shared by a responder and its worker.+-- So, a reference is shared by a 'Respond' and its worker. -- The reference refers a value of this type as a return value. -- If 'True', the worker continue to serve requests. -- Otherwise, the worker get finished.
Network/Wai/Handler/Warp/Internal.hs view
@@ -8,7 +8,18 @@ , runSettingsConnection , runSettingsConnectionMaker , runSettingsConnectionMakerSecure+ , runServe+ , runServeEnv+ , runServeSettings+ , runServeSettingsSocket+ , runServeSettingsConnection+ , runServeSettingsConnectionMaker+ , runServeSettingsConnectionMakerSecure , Transport (..)+ -- * ServeConnection+ , ServeConnection+ , serveDefault+ , serveHTTP2 -- * Connection , Connection (..) , socketConnection
Network/Wai/Handler/Warp/Recv.hs view
@@ -114,7 +114,7 @@ receiveloop :: CInt -> Ptr Word8 -> CSize -> IO CInt receiveloop sock ptr size = do #ifdef mingw32_HOST_OS- bytes <- windowsThreadBlockHack $ fmap fromIntegral $ readRawBufferPtr "recv" (FD sock 1) (castPtr ptr) 0 size+ bytes <- windowsThreadBlockHack . fromIntegral <$> readRawBufferPtr "recv" (FD sock 1) (castPtr ptr) 0 size #else bytes <- c_recv sock (castPtr ptr) size 0 #endif
Network/Wai/Handler/Warp/RequestHeader.hs view
@@ -9,7 +9,7 @@ import Control.Exception (throwIO) import Control.Monad (when) import qualified Data.ByteString as S-import qualified Data.ByteString.Char8 as B (unpack, readInteger)+import qualified Data.ByteString.Char8 as B (unpack) import Data.ByteString.Internal (ByteString(..), memchr) import qualified Data.CaseInsensitive as CI import Data.Word (Word8)@@ -18,7 +18,7 @@ import Foreign.Storable (peek) import qualified Network.HTTP.Types as H import Network.Wai.Handler.Warp.Types-import qualified Network.HTTP.Types.Header as HH+import Network.Wai.Internal (parseByteRanges) -- $setup -- >>> :set -XOverloadedStrings @@ -133,29 +133,3 @@ let (k, rest) = S.break (== 58) s -- ':' rest' = S.dropWhile (\c -> c == 32 || c == 9) $ S.drop 1 rest in (CI.mk k, rest')--parseByteRanges :: S.ByteString -> Maybe HH.ByteRanges-parseByteRanges bs1 = do- bs2 <- stripPrefix "bytes=" bs1- (r, bs3) <- range bs2- ranges (r:) bs3- where- range bs2 = do- (i, bs3) <- B.readInteger bs2- if i < 0 -- has prefix "-" ("-0" is not valid, but here treated as "0-")- then Just (HH.ByteRangeSuffix (negate i), bs3)- else do- bs4 <- stripPrefix "-" bs3- case B.readInteger bs4 of- Just (j, bs5) | j >= i -> Just (HH.ByteRangeFromTo i j, bs5)- _ -> Just (HH.ByteRangeFrom i, bs4)- ranges front bs3- | S.null bs3 = Just (front [])- | otherwise = do- bs4 <- stripPrefix "," bs3- (r, bs5) <- range bs4- ranges (front . (r:)) bs5-- stripPrefix x y- | x `S.isPrefixOf` y = Just (S.drop (S.length x) y)- | otherwise = Nothing
Network/Wai/Handler/Warp/Response.hs view
@@ -6,6 +6,7 @@ module Network.Wai.Handler.Warp.Response ( sendResponse+ , sanitizeHeaderValue -- for testing , fileRange -- for testing , warpVersion , defaultServerValue@@ -27,13 +28,14 @@ import Data.Array ((!)) import Data.ByteString (ByteString) import qualified Data.ByteString as S+import qualified Data.ByteString.Char8 as S8 import Data.ByteString.Builder (byteString, Builder) import Data.ByteString.Builder.Extra (flush) import qualified Data.ByteString.Char8 as B (pack) import qualified Data.CaseInsensitive as CI import Data.Function (on) import Data.List (deleteBy)-import Data.Maybe (isJust, listToMaybe)+import Data.Maybe #if MIN_VERSION_base(4,5,0) # if __GLASGOW_HASKELL__ < 709 import Data.Monoid (mempty)@@ -44,6 +46,7 @@ #endif import Data.Streaming.Blaze (newBlazeRecv, reuseBufferStrategy) import Data.Version (showVersion)+import Data.Word8 (_cr, _lf) import qualified Network.HTTP.Types as H import Network.Wai import Network.Wai.Handler.Warp.Buffer (toBuilderBuffer)@@ -51,14 +54,11 @@ import qualified Network.Wai.Handler.Warp.FdCache as F import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.IO (toBufIOWith)-import Network.Wai.Handler.Warp.RequestHeader (parseByteRanges) import Network.Wai.Handler.Warp.ResponseHeader import qualified Network.Wai.Handler.Warp.Timeout as T import Network.Wai.Handler.Warp.Types-import Network.Wai.Internal (Response (ResponseBuilder, ResponseStream, ResponseFile, ResponseRaw))-import Numeric (showInt)+import Network.Wai.Internal import qualified Paths_warp-import qualified System.PosixCompat.Files as P #if !MIN_VERSION_base(4,5,0) (<>) :: Monoid m => m -> m -> m@@ -68,11 +68,6 @@ -- $setup -- >>> :set -XOverloadedStrings -mapRight :: (b -> c) -> Either a b -> Either a c-mapRight f eith = case eith of- Right x -> Right (f x)- Left l -> Left l- ---------------------------------------------------------------- fileRange :: H.Status -> H.ResponseHeaders -> FilePath@@ -80,8 +75,7 @@ -> IO (Either IOException (H.Status, H.ResponseHeaders, Integer, Integer)) fileRange s0 hs0 path Nothing mRange =- mapRight (fileRangeSized s0 hs0 Nothing mRange . fromIntegral . P.fileSize) <$>- try (P.getFileStatus path)+ fmap (fileRangeSized s0 hs0 Nothing mRange) <$> tryGetFileSize path fileRange s0 hs0 _ mPart@(Just part) mRange = return . Right $ fileRangeSized s0 hs0 mPart mRange size where@@ -92,40 +86,10 @@ -> (H.Status, H.ResponseHeaders, Integer, Integer) fileRangeSized s0 hs0 mPart mRange fileSize = (s, hs, beg, len) where- (beg, end, len, isEntire) = checkPartRange fileSize mPart mRange- hs1 = addContentLength len hs0- hs | isEntire = hs1- | otherwise = addContentRange beg end fileSize hs1- s | isEntire = s0- | otherwise = H.status206--checkPartRange :: Integer -> Maybe FilePart -> Maybe HeaderValue- -> (Integer, Integer, Integer, Bool)-checkPartRange fileSize = checkPart- where- checkPart Nothing Nothing = (0, fileSize - 1, fileSize, True)- checkPart Nothing (Just range) = case parseByteRanges range >>= listToMaybe of- -- Range is broken- Nothing -> (0, fileSize - 1, fileSize, True)- Just hrange -> checkRange hrange- -- Ignore Range if FilePart is specified.- -- We assume that an application handled Range and specified- -- FilePart.- checkPart (Just part) _ = (beg, end, len, isEntire)- where- beg = filePartOffset part- len = filePartByteCount part- end = beg + len - 1- isEntire = beg == 0 && len == fileSize-- checkRange (H.ByteRangeFrom beg) = fromRange beg (fileSize - 1)- checkRange (H.ByteRangeFromTo beg end) = fromRange beg (min (fileSize - 1) end)- checkRange (H.ByteRangeSuffix count) = fromRange (max 0 (fileSize - count)) (fileSize - 1)-- fromRange beg end = (beg, end, len, isEntire)- where- len = end - beg + 1- isEntire = beg == 0 && len == fileSize+ part = fromMaybe (chooseFilePart fileSize mRange) mPart+ beg = filePartOffset part+ len = filePartByteCount part+ (s, hs) = adjustForFilePart s0 hs0 $ FilePart beg len fileSize ---------------------------------------------------------------- @@ -196,7 +160,7 @@ where ver = httpVersion req s = responseStatus response- hs0 = responseHeaders response+ hs0 = sanitizeHeaders $ responseHeaders response rspidxhdr = indexResponseHeader hs0 th = threadHandle ii dc = dateCacher ii@@ -220,6 +184,31 @@ ---------------------------------------------------------------- +sanitizeHeaders :: H.ResponseHeaders -> H.ResponseHeaders+sanitizeHeaders = map (sanitize <$>)+ where+ sanitize v+ | containsNewlines v = sanitizeHeaderValue v -- slow path+ | otherwise = v -- fast path++{-# INLINE containsNewlines #-}+containsNewlines :: ByteString -> Bool+containsNewlines = S.any (\w -> w == _cr || w == _lf)++{-# INLINE sanitizeHeaderValue #-}+sanitizeHeaderValue :: ByteString -> ByteString+sanitizeHeaderValue v = case S8.lines $ S.filter (/= _cr) v of+ [] -> ""+ x : xs -> S8.intercalate "\r\n" (x : mapMaybe addSpaceIfMissing xs)+ where+ addSpaceIfMissing line = case S8.uncons line of+ Nothing -> Nothing+ Just (first, _)+ | first == ' ' || first == '\t' -> Just line+ | otherwise -> Just $ " " <> line++----------------------------------------------------------------+ data Rsp = RspFile FilePath (Maybe FilePart) (Maybe HeaderValue) Bool (IO ()) | RspBuilder Builder Bool | RspStream StreamingBody Bool T.Handle@@ -235,19 +224,19 @@ -> Rsp -> IO () sendRsp conn mfdc ver s0 hs0 (RspFile path mPart mRange isHead hook) = do- ex <- fileRange s0 hs path mPart mRange+ ex <- fileRange s0 hs0 path mPart mRange case ex of Left _ex -> #ifdef WARP_DEBUG print _ex >> #endif sendRsp conn mfdc ver s2 hs2 (RspBuilder body True)- Right (s, hs1, beg, len)+ Right (s, hs, beg, len) | len >= 0 -> if isHead then- sendRsp conn mfdc ver s hs1 (RspBuilder mempty False)+ sendRsp conn mfdc ver s hs (RspBuilder mempty False) else do- lheader <- composeHeader ver s hs1+ lheader <- composeHeader ver s hs #ifdef WINDOWS let fid = FileId path Nothing hook' = hook@@ -263,10 +252,9 @@ connSendFile conn fid beg len hook' [lheader] | otherwise -> sendRsp conn mfdc ver H.status416- (filter (\(k, _) -> k /= "content-length") hs1)+ (filter (\(k, _) -> k /= "content-length") hs) (RspBuilder mempty True) where- hs = addAcceptRanges hs0 s2 = H.status404 hs2 = replaceHeader H.hContentType "text/plain; charset=utf-8" hs0 body = byteString "File not found"@@ -390,30 +378,8 @@ ---------------------------------------------------------------- -addAcceptRanges :: H.ResponseHeaders -> H.ResponseHeaders-addAcceptRanges hdrs = (hAcceptRanges, "bytes") : hdrs- addTransferEncoding :: H.ResponseHeaders -> H.ResponseHeaders addTransferEncoding hdrs = (hTransferEncoding, "chunked") : hdrs--addContentLength :: Integer -> H.ResponseHeaders -> H.ResponseHeaders-addContentLength cl hdrs = (H.hContentLength, len) : hdrs- where- len = B.pack $ show cl--addContentRange :: Integer -> Integer -> Integer- -> H.ResponseHeaders -> H.ResponseHeaders-addContentRange beg end total hdrs = (hContentRange, range) : hdrs- where- range = B.pack- -- building with ShowS- $ 'b' : 'y': 't' : 'e' : 's' : ' '- : (if beg > end then ('*':) else- showInt beg- . ('-' :)- . showInt end)- ( '/'- : showInt total "") addDate :: D.DateCache -> IndexedHeader -> H.ResponseHeaders -> IO H.ResponseHeaders addDate dc rspidxhdr hdrs = case rspidxhdr ! idxDate of
Network/Wai/Handler/Warp/Run.hs view
@@ -6,6 +6,9 @@ module Network.Wai.Handler.Warp.Run where +#if __GLASGOW_HASKELL__ < 709+import Control.Applicative ((<$>))+#endif import Control.Arrow (first) import Control.Concurrent (threadDelay) import qualified Control.Concurrent as Conc (yield)@@ -21,11 +24,12 @@ import Network.Socket (accept, withSocketsDo, SockAddr(SockAddrInet, SockAddrInet6)) import qualified Network.Socket.ByteString as Sock import Network.Wai+import Network.Wai.HTTP2 (HTTP2Application, promoteApplication) import Network.Wai.Handler.Warp.Buffer import Network.Wai.Handler.Warp.Counter import qualified Network.Wai.Handler.Warp.Date as D import qualified Network.Wai.Handler.Warp.FdCache as F-import Network.Wai.Handler.Warp.HTTP2+import Network.Wai.Handler.Warp.HTTP2 (http2, isHTTP2) import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.ReadInt import Network.Wai.Handler.Warp.Recv@@ -68,39 +72,68 @@ allowInterrupt = unblock $ return () #endif +-- Composition over two arguments at once; used for runHTTP2\*.+(.:) :: (c -> d) -> (a -> b -> c) -> a -> b -> d+f .: g = curry $ f . uncurry g+ -- | Run an 'Application' on the given port. -- This calls 'runSettings' with 'defaultSettings'. run :: Port -> Application -> IO ()-run p = runSettings defaultSettings { settingsPort = p }+run port = runServe port . serveDefault +-- | Serve an 'HTTP2Application' and an 'Application' together on the given+-- port.+runHTTP2 :: Port -> HTTP2Application -> Application -> IO ()+runHTTP2 port = runServe port .: serveHTTP2++-- | The generalized form of 'run'.+runServe :: Port -> ServeConnection -> IO ()+runServe p = runServeSettings defaultSettings { settingsPort = p }+ -- | Run an 'Application' on the port present in the @PORT@ -- environment variable. Uses the 'Port' given when the variable is unset. -- This calls 'runSettings' with 'defaultSettings'. -- -- Since 3.0.9 runEnv :: Port -> Application -> IO ()-runEnv p app = do- mp <- fmap (lookup "PORT") getEnvironment+runEnv port = runServeEnv port . serveDefault - maybe (run p app) runReadPort mp+-- | The HTTP\/2-aware form of 'runEnv'.+runHTTP2Env :: Port -> HTTP2Application -> Application -> IO ()+runHTTP2Env port = runServeEnv port .: serveHTTP2 +-- | The generalized form of 'runEnv'.+runServeEnv :: Port -> ServeConnection -> IO ()+runServeEnv p serveConn = do+ mp <- lookup "PORT" <$> getEnvironment++ maybe (runServe p serveConn) runReadPort mp+ where runReadPort :: String -> IO () runReadPort sp = case reads sp of- ((p', _):_) -> run p' app+ ((p', _):_) -> runServe p' serveConn _ -> fail $ "Invalid value in $PORT: " ++ sp -- | Run an 'Application' with the given 'Settings'. -- This opens a listen socket on the port defined in 'Settings' and -- calls 'runSettingsSocket'. runSettings :: Settings -> Application -> IO ()-runSettings set app = withSocketsDo $+runSettings set = runServeSettings set . serveDefault++-- | The HTTP\/2-aware form of 'runSettings'.+runHTTP2Settings :: Settings -> HTTP2Application -> Application -> IO ()+runHTTP2Settings set = runServeSettings set .: serveHTTP2++-- | The generalized form of 'runSettings'.+runServeSettings :: Settings -> ServeConnection -> IO ()+runServeSettings set serveConn = withSocketsDo $ bracket (bindPortTCP (settingsPort set) (settingsHost set)) sClose (\socket -> do setSocketCloseOnExec socket- runSettingsSocket set socket app)+ runServeSettingsSocket set socket serveConn) -- | This installs a shutdown handler for the given socket and -- calls 'runSettingsConnection' with the default connection setup action@@ -114,9 +147,22 @@ -- Note that the 'settingsPort' will still be passed to 'Application's via the -- 'serverPort' record. runSettingsSocket :: Settings -> Socket -> Application -> IO ()-runSettingsSocket set socket app = do+runSettingsSocket set socket = runServeSettingsSocket set socket . serveDefault++-- | The HTTP\/2-aware form of 'runSettingsSocket'.+runHTTP2SettingsSocket :: Settings+ -> Socket+ -> HTTP2Application+ -> Application+ -> IO ()+runHTTP2SettingsSocket set socket =+ runServeSettingsSocket set socket .: serveHTTP2++-- | The generalized form of 'runSettingsSocket'.+runServeSettingsSocket :: Settings -> Socket -> ServeConnection -> IO ()+runServeSettingsSocket set socket serveConn = do settingsInstallShutdownHandler set closeListenSocket- runSettingsConnection set getConn app+ runServeSettingsConnection set getConn serveConn where getConn = do #if WINDOWS@@ -140,7 +186,16 @@ -- -- Since 1.3.5 runSettingsConnection :: Settings -> IO (Connection, SockAddr) -> Application -> IO ()-runSettingsConnection set getConn app = runSettingsConnectionMaker set getConnMaker app+runSettingsConnection set getConn =+ runServeSettingsConnection set getConn . serveDefault++-- | The generalized form of 'runSettingsConnection'.+runServeSettingsConnection :: Settings+ -> IO (Connection, SockAddr)+ -> ServeConnection+ -> IO ()+runServeSettingsConnection set getConn serveConn =+ runServeSettingsConnectionMaker set getConnMaker serveConn where getConnMaker = do (conn, sa) <- getConn@@ -149,10 +204,18 @@ -- | This modifies the connection maker so that it returns 'TCP' for 'Transport' -- (i.e. plain HTTP) then calls 'runSettingsConnectionMakerSecure'. runSettingsConnectionMaker :: Settings -> IO (IO Connection, SockAddr) -> Application -> IO ()-runSettingsConnectionMaker x y =- runSettingsConnectionMakerSecure x (go y)+runSettingsConnectionMaker set getConnMaker =+ runServeSettingsConnectionMaker set getConnMaker . serveDefault++-- | The generalized form of 'runSettingsConnectionMaker'.+runServeSettingsConnectionMaker :: Settings+ -> IO (IO Connection, SockAddr)+ -> ServeConnection+ -> IO ()+runServeSettingsConnectionMaker x y =+ runServeSettingsConnectionMakerSecure x (toTCP <$> y) where- go = fmap (first (fmap (, TCP)))+ toTCP = first ((, TCP) <$>) ---------------------------------------------------------------- @@ -163,14 +226,22 @@ -- -- Since 2.1.4 runSettingsConnectionMakerSecure :: Settings -> IO (IO (Connection, Transport), SockAddr) -> Application -> IO ()-runSettingsConnectionMakerSecure set getConnMaker app = do+runSettingsConnectionMakerSecure set getConnMaker =+ runServeSettingsConnectionMakerSecure set getConnMaker . serveDefault++-- | The generalized form of 'runSettingsConnectionMakerSecure'.+runServeSettingsConnectionMakerSecure :: Settings+ -> IO (IO (Connection, Transport), SockAddr)+ -> ServeConnection+ -> IO ()+runServeSettingsConnectionMakerSecure set getConnMaker serveConn = do settingsBeforeMainLoop set counter <- newCounter D.withDateCache $ \dc -> F.withFdCache fdCacheDurationInSeconds $ \fc -> withTimeoutManager $ \tm ->- acceptConnection set getConnMaker app dc fc tm counter+ acceptConnection set getConnMaker serveConn dc fc tm counter where fdCacheDurationInSeconds = settingsFdCacheDuration set * 1000000 withTimeoutManager f = case settingsManager set of@@ -195,13 +266,13 @@ -- Our approach is explained in the comments below. acceptConnection :: Settings -> IO (IO (Connection, Transport), SockAddr)- -> Application+ -> ServeConnection -> D.DateCache -> Maybe F.MutableFdCache -> T.Manager -> Counter -> IO ()-acceptConnection set getConnMaker app dc fc tm counter = do+acceptConnection set getConnMaker serveConn dc fc tm counter = do -- First mask all exceptions in acceptLoop. This is necessary to -- ensure that no async exception is throw between the call to -- acceptNewConnection and the registering of connClose.@@ -223,7 +294,7 @@ case mx of Nothing -> return () Just (mkConn, addr) -> do- fork set mkConn addr app dc fc tm counter+ fork set mkConn addr serveConn dc fc tm counter acceptLoop acceptNewConnection = do@@ -247,13 +318,13 @@ fork :: Settings -> IO (Connection, Transport) -> SockAddr- -> Application+ -> ServeConnection -> D.DateCache -> Maybe F.MutableFdCache -> T.Manager -> Counter -> IO ()-fork set mkConn addr app dc fc tm counter = settingsFork set $ \ unmask ->+fork set mkConn addr serveConn dc fc tm counter = settingsFork set $ \ unmask -> -- Run the connection maker to get a new connection, and ensure -- that the connection is closed. If the mkConn call throws an -- exception, we will leak the connection. If the mkConn call is@@ -284,21 +355,34 @@ -- Actually serve this connection. -- bracket with closeConn above ensures the connection is closed.- when goingon $ serveConnection conn ii addr transport set app+ when goingon $ serveConn conn ii addr transport set where closeConn (conn, _transport) = connClose conn onOpen adr = increase counter >> settingsOnOpen set adr onClose adr _ = decrease counter >> settingsOnClose set adr -serveConnection :: Connection- -> InternalInfo- -> SockAddr- -> Transport- -> Settings- -> Application- -> IO ()-serveConnection conn ii origAddr transport settings app = do+-- The type of a function to serve a fully-prepared connection.+type ServeConnection = Connection+ -> InternalInfo+ -> SockAddr+ -> Transport+ -> Settings+ -> IO ()++-- Serve an HTTP\/2-aware application, rejecting clients that attempt to use an+-- older protocol version.+serveHTTP2Only :: HTTP2Application -> ServeConnection+serveHTTP2Only = error "serveHTTP2Only not implemented"++-- Serve an HTTP\/2-unaware Application to a connection over any HTTP version.+serveDefault :: Application -> ServeConnection+serveDefault app = serveHTTP2 (promoteApplication app) app++-- Serve an HTTP\/2-aware application over HTTP\/2 or a backup 'Application'+-- over HTTP\/1.1 or HTTP\/1.+serveHTTP2 :: HTTP2Application -> Application -> ServeConnection+serveHTTP2 app2 app conn ii origAddr transport settings = do -- fixme: Upgrading to HTTP/2 should be supported. (h2,bs) <- if isHTTP2 transport then return (True, "")@@ -311,7 +395,7 @@ if h2 then do recvN <- makeReceiveN bs (connRecv conn) (connRecvBuf conn) -- fixme: origAddr- http2 conn ii origAddr transport settings recvN app+ http2 conn ii origAddr transport settings recvN app2 else do istatus <- newIORef False src <- mkSource (wrappedRecv conn th istatus (settingsSlowlorisSize settings))
test/ExceptionSpec.hs view
@@ -6,8 +6,6 @@ import Control.Applicative #endif import Control.Monad-import Network.HTTP-import Network.Stream import Network.HTTP.Types hiding (Header) import Network.Wai hiding (Response) import Network.Wai.Internal (Request(..))@@ -18,6 +16,8 @@ import Control.Concurrent.Async (withAsync) import Network.Socket (sClose) +import HTTP+ main :: IO () main = hspec spec @@ -62,20 +62,5 @@ sc `shouldBe` (5,0,0) -} it "ioException" $ withTestServer $ \prt -> do- sc <- rspCode <$> sendGET (concat $ ["http://127.0.0.1:", show prt, "/ioException"])+ sc <- rspCode <$> sendGET ("http://127.0.0.1:" ++ show prt ++ "/ioException") sc `shouldBe` (5,0,0)--------------------------------------------------------------------sendGET :: String -> IO (Response String)-sendGET url = sendGETwH url []--sendGETwH :: String -> [Header] -> IO (Response String)-sendGETwH url hdr = unResult $ simpleHTTP $ (getRequest url) { rqHeaders = hdr }--unResult :: IO (Result (Response String)) -> IO (Response String)-unResult action = do- res <- action- case res of- Right rsp -> return rsp- Left _ -> error "Connection error"
+ test/HTTP.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE StandaloneDeriving #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+module HTTP (+ sendGET+ , sendGETwH+ , rspBody+ , rspCode+ , rspHeaders+ , getHeaderValue+ , HeaderName(..)+ ) where++import Network.HTTP+import Network.Stream++sendGET :: String -> IO (Response String)+sendGET url = sendGETwH url []++sendGETwH :: String -> [Header] -> IO (Response String)+sendGETwH url hdr = unResult $ simpleHTTP $ (getRequest url) { rqHeaders = hdr }++unResult :: IO (Result (Response String)) -> IO (Response String)+unResult action = do+ res <- action+ case res of+ Right rsp -> return rsp+ Left _ -> error "Connection error"++getHeaderValue :: HasHeaders a => HeaderName -> a -> Maybe String+getHeaderValue key r = case retrieveHeaders key r of+ [] -> Nothing+ x:_ -> Just $ hdrValue x++deriving instance Eq Header
test/ResponseSpec.hs view
@@ -14,6 +14,8 @@ import System.IO (hClose, hFlush) import Test.Hspec +import HTTP+ main :: IO () main = hspec spec @@ -82,6 +84,34 @@ spec :: Spec spec = do+ describe "preventing response splitting attack" $ do+ it "sanitizes header values" $ do+ let app _ respond = respond $ responseLBS status200 [("foo", "foo\r\nbar")] "Hello"+ withApp defaultSettings app $ \port -> do+ res <- sendGET $ "http://127.0.0.1:" ++ show port+ getHeaderValue (HdrCustom "foo") res `shouldBe`+ Just "foo bar" -- HTTP inserts two spaces for \r\n.++ describe "sanitizeHeaderValue" $ do+ it "doesn't alter valid multiline header values" $ do+ sanitizeHeaderValue "foo\r\n bar" `shouldBe` "foo\r\n bar"++ it "adds missing spaces after \r\n" $ do+ sanitizeHeaderValue "foo\r\nbar" `shouldBe` "foo\r\n bar"++ it "discards empty lines" $ do+ sanitizeHeaderValue "foo\r\n\r\nbar" `shouldBe` "foo\r\n bar"++ context "when sanitizing single occurences of \n" $ do+ it "replaces \n with \r\n" $ do+ sanitizeHeaderValue "foo\n bar" `shouldBe` "foo\r\n bar"++ it "adds missing spaces after \n" $ do+ sanitizeHeaderValue "foo\nbar" `shouldBe` "foo\r\n bar"++ it "discards single occurrences of \r" $ do+ sanitizeHeaderValue "foo\rbar" `shouldBe` "foobar"+ describe "range requests" $ do testRange "2-3" "23" $ Just "2-3/16" testRange "5-" "56789abcdef" $ Just "5-15/16"@@ -99,7 +129,7 @@ testFileRange "gets a file size from file system" status200 [] "attic/hex" Nothing Nothing- $ Right (status200,[("Content-Length","16")],0,16)+ $ Right (status200,[("Content-Length","16"), ("Accept-Ranges","bytes")],0,16) testFileRange "gets an error if a file does not exist" status200 [] "attic/nonexist" Nothing Nothing@@ -107,24 +137,24 @@ testFileRange "changes status if FileParts is specified" status200 [] "attic/hex" (Just (FilePart 2 10 16)) Nothing- $ Right (status206,[("Content-Range", "bytes 2-11/16"),("Content-Length","10")],2,10)+ $ Right (status206,[("Content-Range", "bytes 2-11/16"),("Content-Length","10"),("Accept-Ranges","bytes")],2,10) testFileRange "does not change status and does not add Content-Range if FileParts means the entire" status200 [] "attic/hex" (Just (FilePart 0 16 16)) Nothing- $ Right (status200,[("Content-Length","16")],0,16)+ $ Right (status200,[("Content-Length","16"), ("Accept-Ranges","bytes")],0,16) testFileRange "gets a file size from file system and handles Range and returns Partical Content" status200 [] "attic/hex" Nothing (Just "bytes=2-14")- $ Right (status206,[("Content-Range","bytes 2-14/16"),("Content-Length","13")],2,13)+ $ Right (status206,[("Content-Range","bytes 2-14/16"),("Content-Length","13"),("Accept-Ranges","bytes")],2,13) testFileRange "truncates end point of range to file size" status200 [] "attic/hex" Nothing (Just "bytes=10-20")- $ Right (status206,[("Content-Range","bytes 10-15/16"),("Content-Length","6")],10,6)+ $ Right (status206,[("Content-Range","bytes 10-15/16"),("Content-Length","6"),("Accept-Ranges","bytes")],10,6) testFileRange "gets a file size from file system and handles Range and returns OK if Range means the entire" status200 [] "attic/hex" Nothing (Just "bytes=0-15")- $ Right (status200,[("Content-Length","16")],0,16)+ $ Right (status200,[("Content-Length","16"),("Accept-Ranges","bytes")],0,16) testFileRange "igores Range if FilePart is specified" status200 [] "attic/hex" (Just (FilePart 2 10 16)) (Just "bytes=8-9")- $ Right (status206,[("Content-Range", "bytes 2-11/16"),("Content-Length","10")],2,10)+ $ Right (status206,[("Content-Range", "bytes 2-11/16"),("Content-Length","10"),("Accept-Ranges","bytes")],2,10)
test/RunSpec.hs view
@@ -18,7 +18,6 @@ import qualified Data.IORef as I import Data.Streaming.Network (bindPortTCP, getSocketTCP, safeRecv) import Network (connectTo, PortID (PortNumber))-import qualified Network.HTTP as HTTP import Network.HTTP.Types import Network.Socket (sClose) import Network.Socket.ByteString (sendAll)@@ -29,6 +28,8 @@ import System.Timeout (timeout) import Test.Hspec +import HTTP+ main :: IO () main = hspec spec @@ -349,11 +350,9 @@ , ("date", "date") ] "" withApp defaultSettings app $ \port -> do- Right res <- HTTP.simpleHTTP (HTTP.getRequest $ "http://127.0.0.1:" ++ show port)- map HTTP.hdrValue (HTTP.retrieveHeaders HTTP.HdrServer res)- `shouldBe` ["server"]- map HTTP.hdrValue (HTTP.retrieveHeaders HTTP.HdrDate res)- `shouldBe` ["date"]+ res <- sendGET $ "http://127.0.0.1:" ++ show port+ getHeaderValue HdrServer res `shouldBe` Just "server"+ getHeaderValue HdrDate res `shouldBe` Just "date" it "streaming echo #249" $ do let app req f = f $ responseStream status200 [] $ \write _ -> do@@ -375,8 +374,8 @@ let app _ f = f $ responseStream status200 [("content-length", "20")] $ \write _ -> do replicateM_ 4 $ write $ byteString "Hello" withApp defaultSettings app $ \port -> do- Right res <- HTTP.simpleHTTP (HTTP.getRequest $ "http://127.0.0.1:" ++ show port)- HTTP.rspBody res `shouldBe` "HelloHelloHelloHello"+ res <- sendGET $ "http://127.0.0.1:" ++ show port+ rspBody res `shouldBe` "HelloHelloHelloHello" consumeBody :: IO ByteString -> IO [ByteString] consumeBody body =
warp.cabal view
@@ -1,5 +1,5 @@ Name: warp-Version: 3.1.3.1+Version: 3.1.4 Synopsis: A fast, light-weight web server for WAI applications. License: MIT License-file: LICENSE@@ -42,15 +42,16 @@ , ghc-prim , http-types >= 0.8.5 , iproute >= 1.3.1- , http2 >= 1.0.2+ , http2 >= 1.1 , simple-sendfile >= 0.2.7 && < 0.3 , unix-compat >= 0.2- , wai >= 3.0 && < 3.1+ , wai >= 3.0.4 && < 3.1 , text , streaming-commons >= 0.1.10 , vault >= 0.3 , stm >= 2.3 , word8+ , hashable if flag(network-bytestring) Build-Depends: network >= 2.2.1.5 && < 2.2.3 , network-bytestring >= 0.1.3 && < 0.1.4@@ -94,7 +95,6 @@ Cpp-Options: -DWARP_DEBUG if (os(linux) || os(freebsd) || os(darwin)) && flag(allow-sendfilefd) Cpp-Options: -DSENDFILEFD- Build-Depends: hashable Other-modules: Network.Wai.Handler.Warp.MultiMap if os(windows) Cpp-Options: -DWINDOWS@@ -125,6 +125,7 @@ ResponseSpec RunSpec SendFileSpec+ HTTP Hs-Source-Dirs: test, . Type: exitcode-stdio-1.0 @@ -144,7 +145,7 @@ , simple-sendfile >= 0.2.4 && < 0.3 , transformers >= 0.2.2 , unix-compat >= 0.2- , wai+ , wai >= 3.0.4 && < 3.1 , network , HUnit , QuickCheck@@ -159,13 +160,13 @@ , directory , process , containers- , http2 >= 1.0.2+ , http2 >= 1.1 , word8+ , hashable if (os(linux) || os(freebsd) || os(darwin)) && flag(allow-sendfilefd) Cpp-Options: -DSENDFILEFD Build-Depends: unix- , hashable , http-date if os(windows) Cpp-Options: -DWINDOWS