diff --git a/Network/Wai/Handler/Warp.hs b/Network/Wai/Handler/Warp.hs --- a/Network/Wai/Handler/Warp.hs +++ b/Network/Wai/Handler/Warp.hs @@ -5,6 +5,7 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE MagicHash #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# OPTIONS_GHC -fno-warn-warnings-deprecations #-} --------------------------------------------------------- -- -- Module : Network.Wai.Handler.Warp @@ -59,7 +60,7 @@ #endif ) where -import Prelude hiding (catch, lines) +import Prelude hiding (lines) import Network.Wai import Data.ByteString (ByteString) @@ -71,18 +72,30 @@ import Network.Socket (accept, SockAddr) import qualified Network.Socket.ByteString as Sock import Control.Exception - ( bracket, finally, Exception, SomeException, catch + ( mask, handle, onException, bracket + , Exception, SomeException , fromException, AsyncException (ThreadKilled) , try +#if GLASGOW_HASKELL >= 721 + , allowInterrupt +#else + , unblock +#endif +#if WINDOWS + , finally +#endif ) import Control.Concurrent (forkIO) import Data.Maybe (fromMaybe, isJust) +import Data.Char (toLower, isHexDigit) +import Data.Word (Word) import Data.Typeable (Typeable) import Data.Conduit (ResourceT, runResourceT) import qualified Data.Conduit as C import qualified Data.Conduit.List as CL +import qualified Data.Conduit.Binary as CB import Data.Conduit.Blaze (builderToByteString) import Control.Exception.Lifted (throwIO) import Blaze.ByteString.Builder.HTTP @@ -96,11 +109,12 @@ import qualified System.PosixCompat.Files as P import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Trans.Class (lift) import qualified Timeout as T import Timeout (Manager, registerKillThread, pause, resume) import Data.Word (Word8) import Data.List (foldl') -import Control.Monad (forever, when) +import Control.Monad (forever, when, void) import qualified Network.HTTP.Types as H import qualified Data.CaseInsensitive as CI import System.IO (hPrint, stderr) @@ -120,6 +134,11 @@ warpVersion :: String warpVersion = showVersion Paths_warp.version +#if GLASGOW_HASKELL < 721 +allowInterrupt :: IO () +allowInterrupt = unblock $ return () +#endif + -- | -- -- In order to provide slowloris protection, Warp provides timeout handlers. We @@ -202,13 +221,14 @@ port = settingsPort set tm <- maybe (T.initialize $ settingsTimeout set * 1000000) return $ settingsManager set - forever $ do + mask $ \restore -> forever $ do + allowInterrupt (conn, addr) <- getConn - _ <- forkIO $ do + void $ forkIO $ do th <- T.registerKillThread tm - serveConnection set th onE port app conn addr - T.cancel th - return () + handle onE $ (do restore $ serveConnection set th port app conn addr + connClose conn >> T.cancel th + ) `onException` (T.cancel th >> connClose conn) -- | Contains a @Source@ and a byte count that is still to be read in. newtype IsolatedBSSource = IsolatedBSSource (I.IORef (Int, C.Source (ResourceT IO) ByteString)) @@ -271,14 +291,9 @@ serveConnection :: Settings -> T.Handle - -> (SomeException -> IO ()) - -> Port -> Application -> Connection -> SockAddr -> IO () -serveConnection settings th onException port app conn remoteHost' = - catch - (finally - (runResourceT serveConnection') - (connClose conn)) - onException + -> Port -> Application -> Connection -> SockAddr-> IO () +serveConnection settings th port app conn remoteHost' = + runResourceT serveConnection' where serveConnection' :: ResourceT IO () serveConnection' = do @@ -286,7 +301,7 @@ serveConnection'' fromClient serveConnection'' fromClient = do - (env, ibs) <- parseRequest conn port remoteHost' fromClient + (env, getSource) <- parseRequest conn port remoteHost' fromClient case settingsIntercept settings env of Nothing -> do -- Let the application run for as long as it wants @@ -294,20 +309,20 @@ res <- app env -- flush the rest of the request body - ibsIsolate ibs C.$$ CL.sinkNull - fromClient' <- liftIO $ ibsDone ibs + requestBody env C.$$ CL.sinkNull + fromClient' <- liftIO getSource liftIO $ T.resume th keepAlive <- sendResponse th env conn res when keepAlive $ serveConnection'' fromClient' Just intercept -> do liftIO $ T.pause th - fromClient' <- liftIO $ ibsDone ibs + fromClient' <- liftIO getSource intercept fromClient' conn parseRequest :: Connection -> Port -> SockAddr -> C.Source (ResourceT IO) S.ByteString - -> ResourceT IO (Request, IsolatedBSSource) + -> ResourceT IO (Request, IO (C.Source (ResourceT IO) ByteString)) parseRequest conn port remoteHost' src1 = do (src2, headers') <- src1 C.$$+ takeHeaders parseRequest' conn port headers' remoteHost' src2 @@ -347,7 +362,7 @@ -> [ByteString] -> SockAddr -> C.Source (ResourceT IO) S.ByteString -- FIXME was buffered - -> ResourceT IO (Request, IsolatedBSSource) + -> ResourceT IO (Request, IO (C.Source (ResourceT IO) ByteString)) parseRequest' _ _ [] _ _ = throwIO $ NotEnoughLines [] parseRequest' conn port (firstLine:otherLines) remoteHost' src = do (method, rpath', gets, httpversion) <- parseFirst firstLine @@ -364,9 +379,17 @@ Nothing -> 0 Just bs -> readInt bs let serverName' = takeUntil 58 host -- ':' - (rbody, ibs) <- liftIO $ do - ibs <- fmap IsolatedBSSource $ I.newIORef (len0, src) - return (ibsIsolate ibs, ibs) + let chunked = maybe False ((== "chunked") . B.map toLower) + $ lookup "transfer-encoding" heads + (rbody, getSource) <- liftIO $ + if chunked + then do + ref <- I.newIORef (src, NeedLen) + return (chunkedSource ref, fmap fst $ I.readIORef ref) + else do + ibs <- fmap IsolatedBSSource $ I.newIORef (len0, src) + return (ibsIsolate ibs, ibsDone ibs) + return (Request { requestMethod = method , httpVersion = httpversion @@ -381,8 +404,73 @@ , remoteHost = remoteHost' , requestBody = rbody , vault = mempty - }, ibs) + }, getSource) +data ChunkState = NeedLen + | NeedLenNewline + | HaveLen Word + +chunkedSource :: MonadIO m + => I.IORef (C.Source m ByteString, ChunkState) + -> C.Source m ByteString +chunkedSource ipair = do + (src, mlen) <- liftIO $ I.readIORef ipair + go src mlen + where + go' src front = do + (src', (len, bs)) <- lift $ src C.$$+ front getLen + let src'' + | S.null bs = src' + | otherwise = C.yield bs >> src' + go src'' $ HaveLen len + + go src NeedLen = go' src id + go src NeedLenNewline = go' src (CB.take 2 >>) + go src (HaveLen 0) = liftIO $ I.writeIORef ipair (src, HaveLen 0) + go src (HaveLen len) = do + (src', mbs) <- lift $ src C.$$+ CL.head + case mbs of + Nothing -> liftIO $ I.writeIORef ipair (src', HaveLen 0) + Just bs -> + case S.length bs `compare` fromIntegral len of + EQ -> yield' src' NeedLenNewline bs + LT -> do + let mlen = HaveLen $ len - fromIntegral (S.length bs) + yield' src' mlen bs + GT -> do + let (x, y) = S.splitAt (fromIntegral len) bs + let src'' = C.yield y >> src' + yield' src'' NeedLenNewline x + + yield' src mlen bs = do + liftIO $ I.writeIORef ipair (src, mlen) + C.yield bs + go src mlen + + getLen :: Monad m => C.Sink ByteString m (Word, ByteString) + getLen = do + mbs <- CL.head + case mbs of + Nothing -> return (0, S.empty) + Just bs -> do + (x, y) <- + case S.breakByte 10 bs of + (x, y) + | S.null y -> do + mbs2 <- CL.head + case mbs2 of + Nothing -> return (x, y) + Just bs2 -> return $ S.breakByte 10 $ bs `S.append` bs2 + | otherwise -> return (x, y) + let w = + S.foldl' (\i c -> i * 16 + fromIntegral (hexToWord c)) 0 + $ B.takeWhile isHexDigit x + return (w, S.drop 1 y) + + hexToWord w + | w < 58 = w - 48 + | w < 71 = w - 55 + | otherwise = w - 87 takeUntil :: Word8 -> ByteString -> ByteString takeUntil c bs = diff --git a/test/main.hs b/test/main.hs --- a/test/main.hs +++ b/test/main.hs @@ -146,3 +146,73 @@ describe "connection termination" $ do it "ConnectionClosedByPeer" $ runTerminateTest ConnectionClosedByPeer "GET / HTTP/1.1\r\ncontent-length: 10\r\n\r\nhello" it "IncompleteHeaders" $ runTerminateTest IncompleteHeaders "GET / HTTP/1.1\r\ncontent-length: 10\r\n" + + describe "chunked bodies" $ do + it "works" $ do + ifront <- I.newIORef id + port <- getPort + tid <- forkIO $ run port $ \req -> do + bss <- requestBody req $$ Data.Conduit.List.consume + liftIO $ I.atomicModifyIORef ifront $ \front -> (front . (S.concat bss:), ()) + return $ responseLBS status200 [] "" + threadDelay 1000 + handle <- connectTo "127.0.0.1" $ PortNumber $ fromIntegral port + let input = S.concat + [ "POST / HTTP/1.1\r\nTransfer-Encoding: Chunked\r\n\r\n" + , "c\r\nHello World\n\r\n3\r\nBye\r\n0\r\n" + , "POST / HTTP/1.1\r\nTransfer-Encoding: Chunked\r\n\r\n" + , "b\r\nHello World\r\n0\r\n" + ] + hPutStr handle input + hFlush handle + hClose handle + threadDelay 1000 + killThread tid + front <- I.readIORef ifront + front [] @?= + [ "Hello World\nBye" + , "Hello World" + ] + it "lots of chunks" $ do + ifront <- I.newIORef id + port <- getPort + tid <- forkIO $ run port $ \req -> do + bss <- requestBody req $$ Data.Conduit.List.consume + liftIO $ I.atomicModifyIORef ifront $ \front -> (front . (S.concat bss:), ()) + return $ responseLBS status200 [] "" + threadDelay 1000 + handle <- connectTo "127.0.0.1" $ PortNumber $ fromIntegral port + let input = concat $ replicate 2 $ + ["POST / HTTP/1.1\r\nTransfer-Encoding: Chunked\r\n\r\n"] ++ + (replicate 50 "5\r\n12345\r\n") ++ + ["0\r\n"] + mapM_ (\bs -> hPutStr handle bs >> hFlush handle) input + hClose handle + threadDelay 1000 + killThread tid + front <- I.readIORef ifront + front [] @?= replicate 2 (S.concat $ replicate 50 "12345") + it "in chunks" $ do + ifront <- I.newIORef id + port <- getPort + tid <- forkIO $ run port $ \req -> do + bss <- requestBody req $$ Data.Conduit.List.consume + liftIO $ I.atomicModifyIORef ifront $ \front -> (front . (S.concat bss:), ()) + return $ responseLBS status200 [] "" + threadDelay 1000 + handle <- connectTo "127.0.0.1" $ PortNumber $ fromIntegral port + let input = S.concat + [ "POST / HTTP/1.1\r\nTransfer-Encoding: Chunked\r\n\r\n" + , "c\r\nHello World\n\r\n3\r\nBye\r\n0\r\n" + , "POST / HTTP/1.1\r\nTransfer-Encoding: Chunked\r\n\r\n" + , "b\r\nHello World\r\n0\r\n" + ] + mapM_ (\bs -> hPutStr handle bs >> hFlush handle) $ map S.singleton $ S.unpack input + hClose handle + threadDelay 1000 + killThread tid + front <- I.readIORef ifront + front [] @?= + [ "Hello World\nBye" + , "Hello World" + ] diff --git a/warp.cabal b/warp.cabal --- a/warp.cabal +++ b/warp.cabal @@ -1,5 +1,5 @@ Name: warp -Version: 1.2.0.2 +Version: 1.2.1 Synopsis: A fast, light-weight web server for WAI applications. License: MIT License-file: LICENSE