http2-5.4.14: Network/HTTP2/H2/Receiver.hs
{-# LANGUAGE MultiWayIf #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternGuards #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
module Network.HTTP2.H2.Receiver (
frameReceiver,
closureClient,
closureServer,
sendPing,
) where
import Control.Concurrent
import Control.Concurrent.STM
import qualified Control.Exception as E
import qualified Data.ByteString as BS
import qualified Data.ByteString.Char8 as C8
import qualified Data.ByteString.Short as Short
import qualified Data.ByteString.UTF8 as UTF8
import Data.IORef
import Data.Void
import GHC.IO.Exception (
IOErrorType (ResourceVanished),
IOException (ioe_errno),
)
import Network.Control
import Network.HTTP.Semantics
import System.IO.Error (ioeGetErrorType)
import qualified System.ThreadManager as T
import qualified System.TimeManager as TM
import Imports hiding (delete, insert)
import Network.HTTP2.Frame
import Network.HTTP2.H2.Context
import Network.HTTP2.H2.EncodeFrame
import Network.HTTP2.H2.HPACK
import Network.HTTP2.H2.Queue
import Network.HTTP2.H2.Settings
import Network.HTTP2.H2.Stream
import Network.HTTP2.H2.StreamTable
import Network.HTTP2.H2.Types
import Network.HTTP2.H2.Window
----------------------------------------------------------------
continuationLimit :: Int
continuationLimit = 10
headerFragmentLimit :: Int
headerFragmentLimit = 51200 -- 50K
----------------------------------------------------------------
frameReceiver :: Context -> Config -> IO E.SomeException
frameReceiver ctx@Context{receiverDone} conf@Config{..} =
E.mask $ \unmask -> do
-- This catches an asynchronous exception.
-- It is re-thrown by "runH2"
mErr <- E.try $ unmask switch
case mErr of
Left err -> do
atomically $ writeTVar receiverDone $ Just err
return err
Right x -> do
absurd x -- We only terminate due to exceptions
where
switch :: IO Void
switch = do
labelMe "H2 receiver"
tid <- myThreadId
if confReadNTimeout
then
loop1
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
waitOrStop
hd <- readHeader -- throwing an exception on timeout
when (BS.null hd) $ E.throwIO ConnectionIsClosed
processFrame ctx conf $ decodeFrameHeader hd
loop1
loop2 :: TM.Handle -> IO Void
loop2 th = do
-- 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.
waitOrStop
hd <- readHeader
TM.tickle th
when (BS.null hd) $ E.throwIO ConnectionIsClosed
processFrame ctx conf $ decodeFrameHeader hd
loop2 th
----------------------------------------------------------------
processFrame :: Context -> Config -> (FrameType, FrameHeader) -> IO ()
processFrame ctx _conf (fid, FrameHeader{streamId})
| isServer ctx
&& isServerInitiated streamId
&& (fid `notElem` [FramePriority, FrameRSTStream, FrameWindowUpdate]) =
E.throwIO $
ConnectionErrorIsSent ProtocolError streamId "stream id should be odd"
processFrame ctx _conf (FramePushPromise, FrameHeader{streamId})
| isServer ctx =
E.throwIO $
ConnectionErrorIsSent ProtocolError streamId "push promise is not allowed"
processFrame ctx@Context{..} conf (ftyp, FrameHeader{payloadLength, streamId})
| ftyp > maxFrameType = do
mx <- readIORef continued
case mx of
Nothing -> do
-- ignoring unknown frame
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
-- My SETTINGS_ENABLE_PUSH
case checkFrameHeader typhdr of
Left (FrameDecodeError ec sid msg) -> E.throwIO $ ConnectionErrorIsSent ec sid msg
Right _ -> do
let Settings{maxFrameSize, enablePush} = mySettings
sid = streamId header
when (payloadLength header > maxFrameSize) $
E.throwIO $
ConnectionErrorIsSent FrameSizeError sid "exceeds maximum frame size"
when (not enablePush && ftyp == FramePushPromise) $
E.throwIO $
ConnectionErrorIsSent ProtocolError sid "push not enabled"
controlOrStream ctx conf ftyp header
----------------------------------------------------------------
-- | Read a frame payload in full.
--
-- '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. 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 ctx conf payloadLength
control ftyp header bs ctx
| ftyp == FramePushPromise = do
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
-- and read on. There is no 'Stream' to close -- it was refused
-- before one was made -- so this resets by identifier alone.
push header bs ctx `E.catch` resetPromised
| otherwise = do
mcont <- checkContinued
mstrm <- getStream ctx ftyp streamId
bs <- readPayload ctx conf payloadLength
case mcont of
Just hc -> continuation hc bs mstrm
Nothing ->
case mstrm of
Just strm -> resettable strm $ do
state0 <- readStreamState strm
state <- stream ftyp header bs ctx state0 strm
processState state ctx strm streamId
Nothing
| ftyp == FramePriority -> do
-- for h2spec only
PriorityFrame newpri <- guardIt $ decodePriorityFrame header bs
checkPriority newpri streamId
| ftyp == FrameData ->
-- Dropped, but still paid for.
informIgnoredData ctx streamId payloadLength
| ftyp == FrameHeaders -> do
HeadersFrame _ frag <- guardIt $ decodeHeadersFrame header bs
if testEndHeader flags
then hpackDiscardHeader frag streamId ctx
else startHeaderBlock ctx streamId (testEndStream flags) frag
| otherwise -> return ()
where
resetContinued = writeIORef continued Nothing
resetPromised (StreamErrorIsSent err sid _msg) =
enqueueControl controlQ $ CFrames Nothing [resetFrame err sid]
resetPromised e = E.throwIO e
-- Answer a stream error by resetting that stream and reading on, which is
-- what RFC 9113 section 5.4.2 asks for: "an error related to a specific
-- stream that does not affect processing of other streams".
--
-- Safe only here, after the payload has been read and any field block in
-- it decoded, so that the connection sits at a frame boundary and the
-- HPACK tables still agree with the peer's. Where neither holds -- a
-- field block abandoned part-way, a stream refused before its payload was
-- read -- the error is raised as a connection error where it is detected,
-- and travels straight past this handler.
resettable strm act = act `E.catch` reset
where
reset e@(StreamErrorIsSent err sid _msg) = do
-- MadeYouReset: CVE-2025-8671. A reset we send because of
-- what the peer sent frees the stream's concurrency slot just as
-- one the peer sends does, while a handler already launched for
-- the stream runs on. Counted with the peer's own resets, or a
-- peer that never sends RST_STREAM -- a PRIORITY on the stream
-- depending on itself is enough -- could keep any number of
-- handlers running past SETTINGS_MAX_CONCURRENT_STREAMS.
--
-- REFUSED_STREAM is left out: it launches nothing, and it is the
-- answer section 8.7 means a peer to be able to retry.
when (err /= RefusedStream) $ do
rate <- getRate rstRate
when (rate > rstRateLimit mySettings) $
E.throwIO $
ConnectionErrorIsSent EnhanceYourCalm sid "too many stream errors"
resetContinued
-- 'closed' hands the exception to whoever is reading the stream
-- and takes it out of the stream table.
closed ctx strm $ ResetByMe $ E.toException e
enqueueControl controlQ $ CFrames Nothing [resetFrame err sid]
reset e = E.throwIO e
checkContinued :: IO (Maybe HeaderContinuation)
checkContinued = do
mx <- readIORef continued
case mx of
Nothing -> return ()
Just hc
| hcStreamId hc == streamId && ftyp == FrameContinuation -> return ()
| otherwise ->
E.throwIO $
ConnectionErrorIsSent ProtocolError streamId "continuation frame must follow"
return mx
continuation
:: HeaderContinuation -> HeaderBlockFragment -> Maybe Stream -> IO ()
continuation hc frag mstrm
| frag == "" && not (testEndHeader flags) = do
-- Empty Frame Flooding - CVE-2019-9518
rate <- getRate emptyFrameRate
when (rate > emptyFrameRateLimit mySettings) $
E.throwIO $
ConnectionErrorIsSent EnhanceYourCalm streamId "too many empty continuation"
| otherwise = do
phb' <- addFragment streamId frag (hcBlock hc)
if testEndHeader flags
then completeBlock hc (completeHeaderBlock phb') mstrm
else writeIORef continued $ Just hc{hcBlock = phb'}
completeBlock
:: HeaderContinuation -> HeaderBlockFragment -> Maybe Stream -> IO ()
completeBlock hc blk mstrm = do
resetContinued
case mstrm of
Just strm -> resettable strm $ do
state0 <- readStreamState strm
case state0 of
Open hcl JustOpened -> do
tbl <- hpackDecodeHeader blk streamId ctx
state <- onResponseHeaders ctx streamId hcl (hcEndOfStream hc) tbl
processState state ctx strm streamId
Open _ (Body q mcl bodyLength tlr) -> do
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
-- headers or trailers, and the sender has since
-- closed it -- a reset crossing the block. There is
-- nothing to deliver, but the block still has to go
-- through the decoder, since it may have changed the
-- dynamic table. Handing it to 'stream' as a
-- CONTINUATION would get it refused as one that
-- cannot come here, closing the connection.
hpackDiscardHeader blk streamId ctx
Nothing ->
hpackDiscardHeader blk streamId ctx
-- | Is this a response that is defined to have no content?
--
-- RFC 9113, section 8.1.1: "A response that is defined to have no content,
-- as described in Section 6.4.1 of [HTTP], can have a non-zero
-- content-length header field, even though no content is included in DATA
-- frames." Those are the responses to HEAD, 204 and 304, and 2xx to
-- CONNECT; the content-length of one to HEAD, in particular, is that of the
-- content a GET would have had. Checking it against the content that
-- arrived made every such response to HEAD a stream error.
hasNoContent :: Context -> Stream -> ValueTable -> IO Bool
hasNoContent ctx Stream{streamRequestMethod} vt
| isServer ctx = return False
| otherwise = do
mmethod <- readIORef streamRequestMethod
let status = getFieldValue tokenStatus vt
return $
mmethod == Just "HEAD"
|| status `elem` [Just "204", Just "304"]
|| (mmethod == Just "CONNECT" && maybe False ("2" `BS.isPrefixOf`) status)
----------------------------------------------------------------
processState :: StreamState -> Context -> Stream -> StreamId -> IO ()
-- Transition (process1)
processState (Open _ (NoBody tbl@(_, reqvt))) ctx strm@Stream{streamInput} streamId = do
-- My SETTINGS_MAX_CONCURRENT_STREAMS
when (isServer ctx) $ checkOddConcurrency ctx streamId
noContent <- hasNoContent ctx strm reqvt
let mcl = fst <$> (getFieldValue tokenContentLength reqvt >>= C8.readInt)
when (not noContent && just mcl (/= (0 :: Int))) $
E.throwIO $
StreamErrorIsSent
ProtocolError
streamId
"no body but content-length is not zero"
tlr <- newIORef Nothing
let inpObj = InpObj tbl (Just 0) (return (mempty, True)) tlr
if isServer ctx
then do
launchHandler ctx strm inpObj
else putMVar streamInput $ Right inpObj
halfClosedRemote ctx strm
-- Transition (process2)
processState (Open _ (HasBody tbl@(_, reqvt))) ctx strm@Stream{streamInput, streamRxQ} _streamId = do
-- My SETTINGS_MAX_CONCURRENT_STREAMS
when (isServer ctx) $ checkOddConcurrency ctx _streamId
noContent <- hasNoContent ctx strm reqvt
let mcl
-- Its content-length describes content it does not have.
| noContent = Just 0
| otherwise = fst <$> (getFieldValue tokenContentLength reqvt >>= C8.readInt)
bodyLength <- newIORef 0
tlr <- newIORef Nothing
q <- newTQueueIO
writeIORef streamRxQ $ Just q
setOpenState ctx strm $ Body q mcl bodyLength tlr
-- FLOW CONTROL: WINDOW_UPDATE 0: recv: announcing my limit properly
-- FLOW CONTROL: WINDOW_UPDATE: recv: announcing my limit properly
bodySource <- mkSource q $ informWindowUpdate ctx strm
let inpObj = InpObj tbl mcl (readSource bodySource) tlr
if isServer ctx
then do
launchHandler ctx strm inpObj
else putMVar streamInput $ Right inpObj
-- Transition (process4)
processState HalfClosedRemote ctx strm _streamId = do
halfClosedRemote ctx strm
-- Transition (process5)
processState (Closed cc) ctx strm _streamId = do
closed ctx strm cc
-- Transition (process6)
processState (Open _ o) ctx strm _streamId =
-- Open JustOpened, Open Body. Not the whole state: see 'setOpenState'.
setOpenState ctx strm o
processState s ctx strm _streamId = do
-- Idle
setStreamState ctx strm s
-- | Handing a request to the server's handler.
--
-- The stream counts towards the last stream identifier of our GOAWAY from
-- here on: from now its request "might have been processed" (RFC 9113,
-- section 6.8), and a client is free to retry, on another connection, any
-- request on a stream above it. It used to be counted once the handler
-- had returned, so a GOAWAY sent while handlers were running left their
-- streams out, and a client could send again a request -- a POST, say --
-- that had been acted on.
launchHandler :: Context -> Stream -> InpObj -> IO ()
launchHandler ctx@Context{roleInfo} strm inpObj = do
-- 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
----------------------------------------------------------------
{- FOURMOLU_DISABLE -}
getStream :: Context -> FrameType -> StreamId -> IO (Maybe Stream)
getStream ctx@Context{..} ftyp streamId
| isEven = lookupEven evenStreamTable streamId >>= getEvenStream ctx ftyp
| otherwise = lookupOdd oddStreamTable streamId >>= getOddStream ctx ftyp streamId
where
isEven = isServerInitiated streamId
{- FOURMOLU_ENABLE -}
getEvenStream :: Context -> FrameType -> Maybe Stream -> IO (Maybe Stream)
getEvenStream ctx ftyp js@(Just strm) = do
when (ftyp == FrameHeaders) $ do
st <- readStreamState strm
when (isReserved st) $ halfClosedLocal ctx strm Finished
return js
getEvenStream _ _ Nothing = return Nothing
getOddStream
:: Context -> FrameType -> StreamId -> Maybe Stream -> IO (Maybe Stream)
getOddStream ctx ftyp streamId js@(Just strm0) = do
when (ftyp == FrameHeaders) $ do
st <- readStreamState strm0
when (isHalfClosedRemote st) $
E.throwIO $
ConnectionErrorIsSent
StreamClosed
streamId
"header must not be sent to half or fully closed stream"
-- Priority made an idle stream
when (isIdle st) $ opened ctx strm0
return js
getOddStream ctx ftyp streamId Nothing
| isServer ctx = do
csid <- getPeerStreamID ctx
if streamId <= csid -- consider the stream closed
then do
-- RFC 9113 section 5.1: "An endpoint MUST ignore frames that
-- it receives on closed streams after it has sent a
-- RST_STREAM frame." DATA is in that list because resetting a
-- stream mid-body leaves whatever the peer already put on the
-- wire still to arrive. HEADERS is, for a stream we reset:
-- trailers the peer sent before our reset reached it. They
-- used to close the connection, every other stream on it with
-- it. Their field block is still decoded, for the dynamic
-- table ('hpackDiscardHeader'), CONTINUATION and all.
-- HEADERS on any other closed
-- stream is the reuse of a stream identifier, which section
-- 5.1.1 makes a connection error.
resetByUs <- wasResetByMe ctx streamId
if ftyp `elem` [FrameData, FrameWindowUpdate, FrameRSTStream, FramePriority]
|| (ftyp `elem` [FrameHeaders, FrameContinuation] && resetByUs)
then return Nothing -- will be ignored
else
E.throwIO $
ConnectionErrorIsSent
ProtocolError
streamId
"stream identifier must not decrease"
else do
-- consider the stream idle
when (ftyp `notElem` [FrameHeaders, FramePriority]) $ do
let errmsg =
Short.toShort
( "this frame is not allowed in an idle stream: "
`BS.append` C8.pack (show ftyp)
)
E.throwIO $ ConnectionErrorIsSent ProtocolError streamId errmsg
if ftyp == FramePriority
then
-- PRIORITY does not open a stream (RFC 9113, section
-- 5.1): it is checked and dropped like one for a
-- stream we do not have. It used to create the
-- stream, taking a concurrency slot that nothing ever
-- gave back, since no HEADERS need follow: a peer
-- could fill SETTINGS_MAX_CONCURRENT_STREAMS with
-- PRIORITY frames alone and have every request after
-- them refused.
return Nothing
else do
setPeerStreamID ctx streamId
-- FLOW CONTROL: SETTINGS_MAX_CONCURRENT_STREAMS: recv: rejecting if over my limit
Just <$> openOddStreamCheck ctx streamId ftyp
| otherwise =
-- We received a frame from the server on an unknown stream
-- (likely a previously created and then subsequently reset stream).
-- We just drop it.
return Nothing
----------------------------------------------------------------
type Payload = ByteString
control :: FrameType -> FrameHeader -> Payload -> Context -> IO ()
control FrameSettings header@FrameHeader{flags, streamId} bs Context{myFirstSettings, controlQ, settingsRate, mySettings, rxFlow} = do
SettingsFrame peerAlist <- guardIt $ decodeSettingsFrame header bs
traverse_ E.throwIO $ checkSettingsList peerAlist
if testAck flags
then do
when (peerAlist /= []) $
E.throwIO $
ConnectionErrorIsSent FrameSizeError streamId "ack settings has a body"
else do
-- Settings Flood - CVE-2019-9515
rate <- getRate settingsRate
when (rate > settingsRateLimit mySettings) $
E.throwIO $
ConnectionErrorIsSent EnhanceYourCalm streamId "too many settings"
let ack = settingsFrame setAck []
sent <- readIORef myFirstSettings
if sent
then do
let setframe = CFrames (Just peerAlist) [ack]
enqueueControl controlQ setframe
else do
-- Server side only
connRxWS <- rxfBufSize <$> readIORef rxFlow
let frames = makeNegotiationFrames mySettings connRxWS
setframe = CFrames (Just peerAlist) (frames ++ [ack])
writeIORef myFirstSettings True
enqueueControl controlQ setframe
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"
else sendPing ctx True bs
control FrameGoAway header bs ctx = do
GoAwayFrame sid err msg <- guardIt $ decodeGoAwayFrame header bs
if err == NoError
then do
first <- goingAway ctx sid
-- The receiver goes on reading for the streams left, and stops
-- the way it does when the peer closes the connection, once
-- they are done.
when first $ do
receiver <- myThreadId
T.forkManaged (threadManager ctx) "H2 draining after GOAWAY" $ do
atomically $ drained ctx
stopReceiver ctx receiver
else E.throwIO $ ConnectionErrorIsReceived err sid $ Short.toShort msg
control FrameWindowUpdate header bs ctx = do
WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs
increaseConnectionWindowSize ctx n
control _ _ _ _ =
-- must not reach here
return ()
----------------------------------------------------------------
-- Called in client only
push :: FrameHeader -> ByteString -> Context -> IO ()
push header@FrameHeader{streamId} bs ctx = do
PushPromiseFrame sid frag <- guardIt $ decodePushPromiseFrame header bs
unless (isServerInitiated sid) $
E.throwIO $
ConnectionErrorIsSent
ProtocolError
streamId
"push promise must specify an even stream identifier"
when (frag == "") $
E.throwIO $
ConnectionErrorIsSent
ProtocolError
streamId
"wrong header fragment for push promise"
-- A malformed promised request is an error on the promised stream,
-- which 'resetPromised' resets.
(_, vt) <- hpackDecodeHeader frag streamId ctx `E.catch` onPromised sid
let ClientInfo{..} = toClientInfo $ roleInfo ctx
when
( getFieldValue tokenAuthority vt == Just (UTF8.fromString authority)
&& getFieldValue tokenScheme vt == Just scheme
)
$ do
let mmethod = getFieldValue tokenMethod vt
mpath = getFieldValue tokenPath vt
case (mmethod, mpath) of
(Just method, Just path) ->
-- FLOW CONTROL: SETTINGS_MAX_CONCURRENT_STREAMS: recv: rejecting if over my limit
openEvenStreamCacheCheck ctx sid method path
_ -> return ()
----------------------------------------------------------------
onPromised :: StreamId -> HTTP2Error -> IO a
onPromised sid (StreamErrorIsSent err _ msg) = E.throwIO $ StreamErrorIsSent err sid msg
onPromised _ e = E.throwIO e
{-# INLINE guardIt #-}
guardIt :: Either FrameDecodeError a -> IO a
guardIt x = case x of
Left (FrameDecodeError ec sid msg) -> E.throwIO $ ConnectionErrorIsSent ec sid msg
Right frame -> return frame
{-# INLINE checkPriority #-}
checkPriority :: Priority -> StreamId -> IO ()
checkPriority p me
| dep == me =
E.throwIO $ StreamErrorIsSent ProtocolError me "priority depends on itself"
| otherwise = return ()
where
dep = streamDependency p
-- | Handle a decoded response HEADERS section. On the client, a 1xx
-- informational response (e.g. 103 Early Hints) is delivered to the
-- informational callback and the stream keeps waiting for the final response;
-- otherwise the headers become the (final) response.
onResponseHeaders
:: Context
-> StreamId
-> Maybe ClosedCode
-> Bool
-> TokenHeaderTable
-> IO StreamState
onResponseHeaders ctx streamId hcl endOfStream tbl
| endOfStream = return $ Open hcl (NoBody tbl)
| role ctx == Client && isInformational = do
informationalCallback ctx streamId tbl
return $ Open hcl JustOpened
| otherwise = return $ Open hcl (HasBody tbl)
where
isInformational =
maybe False ("1" `BS.isPrefixOf`) $ getFieldValue tokenStatus (snd tbl)
stream
:: FrameType
-> FrameHeader
-> ByteString
-> Context
-> StreamState
-> Stream
-> IO StreamState
-- Transition (stream1)
stream FrameHeaders header@FrameHeader{flags, streamId} bs ctx s@(Open hcl JustOpened) Stream{streamNumber} = do
HeadersFrame mp frag <- guardIt $ decodeHeadersFrame header bs
let endOfStream = testEndStream flags
endOfHeader = testEndHeader flags
if frag == "" && not endOfStream && not endOfHeader
then do
-- Empty Frame Flooding - CVE-2019-9518
rate <- getRate $ emptyFrameRate ctx
if rate > emptyFrameRateLimit (mySettings ctx)
then
E.throwIO $
ConnectionErrorIsSent EnhanceYourCalm streamId "too many empty headers"
else return s
else do
case mp of
Nothing -> return ()
Just p -> checkPriority p streamNumber
if endOfHeader
then do
tbl <- hpackDecodeHeader frag streamId ctx
onResponseHeaders ctx streamId hcl endOfStream tbl
else do
startHeaderBlock ctx streamId endOfStream frag
return s
-- Transition (stream2)
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 strm streamId frag q mcl bodyLength tlr
else do
startHeaderBlock ctx streamId endOfStream frag
return s
else
E.throwIO $
ConnectionErrorIsSent
ProtocolError
streamId
"trailers without END_STREAM"
-- Transition (stream4)
stream
FrameData
header@FrameHeader{flags, payloadLength, streamId}
bs
ctx@Context{emptyFrameRate, rxFlow, mySettings}
s@(Open _ (Body q mcl bodyLength _))
strm@Stream{..} = do
DataFrame body <- guardIt $ decodeDataFrame header bs
-- FLOW CONTROL: WINDOW_UPDATE 0: recv: rejecting if over my limit
okc <- atomicModifyIORef' rxFlow $ checkRxLimit payloadLength
unless okc $
E.throwIO $
ConnectionErrorIsSent
EnhanceYourCalm
streamId
"exceeds connection flow-control limit"
-- FLOW CONTROL: WINDOW_UPDATE: recv: rejecting if over my limit
oks <- atomicModifyIORef' streamRxFlow $ checkRxLimit payloadLength
unless oks $
E.throwIO $
ConnectionErrorIsSent
EnhanceYourCalm
streamId
"exceeds stream flow-control limit"
-- A push gives the connection window everything back now, padding
-- and all ('connectionCreditedOnArrival'); its stream window, and
-- any other stream's windows, as below.
when (connectionCreditedOnArrival streamId) $
giveBackConnectionWindow ctx payloadLength
-- The padding is charged to both windows, as it must be, but it
-- never reaches the reader, whose reading is what gives octets
-- back ('readSource'). Left there, each padded frame shrank the
-- peer's windows for good, until the connection stalled. It is
-- done with as soon as it arrives, so it goes straight back.
informWindowUpdate ctx strm $ payloadLength - BS.length body
len0 <- readIORef bodyLength
-- The content itself: 'payloadLength', which flow control goes by,
-- also counts the padding, and so made a padded body look longer
-- than its content-length.
let len = len0 + BS.length body
endOfStream = testEndStream flags
-- RFC 9113, section 8.1.1: a body that does not add up to its
-- content-length makes the message malformed. Checked before any
-- of this frame is delivered: it used to be checked once the last
-- of the body had been queued with END_STREAM, and the reader took
-- the body for complete before the reset reached it -- a server
-- handler, say, or a proxy that went by the content-length. Now
-- the reader gets the reset ('closed', by 'resettable') in place of
-- the end, and a body longer than its content-length is cut off as
-- soon as it is. What is not queued goes back to the connection
-- window.
when (just mcl (len >) || (endOfStream && just mcl (len <))) $ do
unless (connectionCreditedOnArrival streamId) $
giveBackConnectionWindow ctx $
BS.length body
E.throwIO $
StreamErrorIsSent
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
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
-- a reader that has done with it ('giveBackUnread') or a
-- reset of ours: nothing would ever read it, or give it
-- back to the connection window. Nor for one whose
-- handler has returned ('streamDiscardBody'): nobody
-- will read it either, and it is given back to both
-- windows, so that the peer can send the rest and end
-- the stream. Checked in the same transaction, so that
-- a closer that has emptied the queue finds nothing put
-- in after it.
outcome <- deliver $ do
st <- readTVar streamState
discard <- readTVar streamDiscardBody
if
| isClosed st -> return Dropped
| discard -> return Discarded
| otherwise -> do
writeTQueue q $ Right (body, endOfStream)
return Queued
case outcome of
Queued -> return ()
Discarded -> informWindowUpdate ctx strm $ BS.length body
Dropped ->
unless (connectionCreditedOnArrival streamId) $
giveBackConnectionWindow ctx $
BS.length body
return $ if endOfStream then HalfClosedRemote else s
-- (No state transition)
stream FrameWindowUpdate header bs _ s strm = do
WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs
increaseStreamWindowSize strm n
return s
-- Transition (stream6)
stream FrameRSTStream header@FrameHeader{streamId} bs ctx s strm = do
-- Rapid Rest: CVE-2023-44487
rate <- getRate $ rstRate ctx
when (rate > rstRateLimit (mySettings ctx)) $
E.throwIO $
ConnectionErrorIsSent EnhanceYourCalm streamId "too many rst_stream"
RSTStreamFrame err <- guardIt $ decodeRSTStreamFrame header bs
let cc = Reset err
closed ctx strm cc
-- HTTP2 spec, section 5.1, "Stream States":
--
-- > A stream in the "open" state may be used by both peers to send frames
-- > of any type. (..) From this state, either endpoint can send a frame
-- > with an END_STREAM flag set, which causes the stream to transition into
-- > one of the "half-closed" states. An endpoint sending an END_STREAM
-- > flag causes the stream state to become "half-closed (local)"; an
-- > endpoint receiving an END_STREAM flag causes the stream state to become
-- > "half-closed (remote)".
--
-- Crucially (for the specific case we're dealing with here), it continues:
--
-- > /Either endpoint/ can send a RST_STREAM frame from this state, causing
-- > it to transition immediately to "closed".
--
-- (emphasis not in original).
--
-- In addition, the spec states (about the open state):
--
-- > Either endpoint can send a RST_STREAM frame from this state, causing it
-- > to transition immediately to "closed".
--
-- This justifies the non-error cases, below. (Section 8.1 of the spec
-- is also relevant, but it is less explicit about the /either endpoint/
-- part.)
--
-- The error code the peer sent does not enter into it. Receiving a
-- RST_STREAM closes that stream and nothing else, whatever the reason
-- given; the code is for whoever is reading the stream, and reaches them
-- as 'StreamResetIsReceived' by way of 'closed' above. Ending the whole
-- connection over it would punish every other stream on the connection
-- for a peer's complaint about one.
case s of
-- Open /or/ half-closed (local)
Open _ _ -> return (Closed cc)
HalfClosedRemote -> return (Closed cc)
Reserved -> return (Closed cc)
Closed _ -> return (Closed cc)
-- Only an idle stream is left, which a PRIORITY frame can have
-- created. Section 5.1 again, on "idle": "Receiving any frame other
-- than HEADERS or PRIORITY on a stream in this state MUST be treated
-- as a connection error (Section 5.4.1) of type PROTOCOL_ERROR."
Idle ->
E.throwIO $
ConnectionErrorIsSent
ProtocolError
streamId
"rst_stream on an idle stream"
-- (No state transition)
stream FramePriority header bs _ s Stream{streamNumber} = do
-- ignore
-- Resource Loop - CVE-2019-9513
PriorityFrame newpri <- guardIt $ decodePriorityFrame header bs
checkPriority newpri streamNumber
return s
-- this ordering is important
stream FrameContinuation FrameHeader{streamId} _ _ _ _ =
E.throwIO $
ConnectionErrorIsSent ProtocolError streamId "continue frame cannot come here"
-- DATA that is not taken in, below, is still paid for. RFC 9113 section
-- 6.9: "A receiver that receives a flow-controlled frame MUST always account
-- for its contribution against the connection flow-control window, unless
-- the receiver treats this as a connection error." The peer charged it to
-- the connection window before sending it, and unless it is charged and
-- given back here as well, the peer's view of that window shrinks for good.
--
-- Ignore frames to streams we have just reset, per section 5.1.
stream FrameData FrameHeader{payloadLength, streamId} _ ctx st@(Closed (ResetByMe _)) _ = do
informIgnoredData ctx streamId payloadLength
return st
stream _ _ _ _ st@(Closed (ResetByMe _)) _ = return st
stream FrameData FrameHeader{payloadLength, streamId} _ ctx _ _ = do
informIgnoredData ctx streamId payloadLength
E.throwIO $
StreamErrorIsSent StreamClosed streamId $
fromString ("illegal data frame for " ++ show streamId)
stream x FrameHeader{streamId} _ _ _ _ =
E.throwIO $
StreamErrorIsSent ProtocolError streamId $
fromString ("illegal frame " ++ show x ++ " for " ++ show streamId)
----------------------------------------------------------------
-- | What became of the content of a DATA frame.
data DataOutcome
= -- | Put in the stream's queue for the reader
Queued
| -- | Thrown away, the reader gone ('streamDiscardBody')
Discarded
| -- | Thrown away, the stream closed
Dropped
----------------------------------------------------------------
-- | Type for input streaming.
data Source = Source RxQ (Int -> IO ()) (IORef Bool)
mkSource :: RxQ -> (Int -> IO ()) -> IO Source
mkSource q inform = Source q inform <$> newIORef False
readSource :: Source -> IO (ByteString, Bool)
readSource (Source q inform refEOF) = do
eof <- readIORef refEOF
if eof
then return (mempty, True)
else do
mBS <- atomically $ readTQueue q
case mBS of
Left err -> do
writeIORef refEOF True
E.throwIO err
Right (bs, isEOF) -> do
writeIORef refEOF isEOF
let len = BS.length bs
inform len
return (bs, isEOF)
----------------------------------------------------------------
closureClient :: Config -> Context -> Either E.SomeException a -> IO a
closureClient conf ctx (Right x) = do
frame <- goaway ctx NoError "no error"
sendGoaway conf frame
return x
closureClient conf ctx (Left se) = closureServer conf ctx se
closureServer :: Config -> Context -> E.SomeException -> IO a
closureServer conf ctx se
| isAsyncException se = do
frame <- goaway ctx NoError "maybe timeout by manager"
sendGoaway conf frame
E.throwIO se
| Just ConnectionIsClosed <- E.fromException se = do
frame <- goaway ctx NoError "no error"
sendGoaway conf frame
E.throwIO ConnectionIsClosed
| Just ConnectionIsTimeout <- E.fromException se = do
frame <- goaway ctx NoError "timeout"
sendGoaway conf frame
E.throwIO ConnectionIsTimeout
| Just e@(ConnectionErrorIsReceived _err _sid msg) <- E.fromException se = do
frame <- goaway ctx NoError $ Short.fromShort msg
sendGoaway conf frame
E.throwIO e
| Just e@(ConnectionErrorIsSent err _sid msg) <- E.fromException se = do
frame <- goaway ctx err $ Short.fromShort msg
sendGoaway conf frame
E.throwIO e
| Just e@(StreamErrorIsSent err _sid msg) <- E.fromException se = do
let frame = resetFrame err _sid
frame' <- goaway ctx err $ Short.fromShort msg
sendGoaway conf (frame <> frame')
E.throwIO e
| Just e@(StreamErrorIsReceived err _sid) <- E.fromException se = do
frame <- goaway ctx err "treat a stream error as a connection error"
sendGoaway conf frame
E.throwIO e
| Just (_ :: HTTP2Error) <- E.fromException se = E.throwIO se
-- Nothing can be sent once the peer has reset the connection.
| Just ioe <- E.fromException se
, isReset ioe =
E.throwIO ConnectionIsReset
| otherwise = E.throwIO $ BadThingHappen se
-- | A reset is 'ResourceVanished', except on Windows: there network
-- reports every error as another, WSAECONNRESET and WSAECONNABORTED
-- among them, and keeps the Winsock code in 'ioe_errno'.
isReset :: E.IOException -> Bool
isReset ioe =
ioeGetErrorType ioe == ResourceVanished
|| ioe_errno ioe `elem` [Just 10053, Just 10054]
goaway :: Context -> ErrorCode -> ByteString -> IO ByteString
goaway ctx err msg = do
sid <- getPeerLastStreamId ctx
return $ goawayFrame sid err msg
----------------------------------------------------------------
sendPing :: Context -> Bool -> ByteString -> IO ()
sendPing Context{..} ack bs = enqueueControl controlQ $ CFrames Nothing [frame]
where
frame = pingFrame ack bs
----------------------------------------------------------------
-- | Deliver a complete trailer block: the body ends with it.
--
-- So it has to add up to its content-length, as it does when it ends with
-- END_STREAM on DATA ('stream'). It was not checked here at all: a body
-- of any length went through as long as trailers ended it. Checked once
-- the block is decoded, for the dynamic table.
onTrailers
:: Context
-> Stream
-> StreamId
-> HeaderBlockFragment
-> TQueue (Either E.SomeException (ByteString, Bool))
-> Maybe Int
-- ^ content-length
-> IORef Int
-- ^ The length of the body received
-> IORef (Maybe TokenHeaderTable)
-> IO StreamState
onTrailers ctx strm streamId blk q mcl bodyLength tlr = do
tbl <- hpackDecodeTrailer blk streamId ctx
len <- readIORef bodyLength
when (just mcl (/= len)) $
E.throwIO $
StreamErrorIsSent
ProtocolError
streamId
"actual body length is not the same as content-length"
writeIORef tlr (Just tbl)
halfClosedRemoteWith ctx strm $ writeTQueue q $ Right (mempty, True)
return HalfClosedRemote
-- | Start accumulating a header block that does not fit in a single frame
startHeaderBlock
:: Context
-> StreamId
-> Bool
-- ^ END_STREAM, from the HEADERS frame
-> HeaderBlockFragment
-- ^ The fragment in the HEADERS frame
-> IO ()
startHeaderBlock Context{continued} streamId endOfStream frag =
writeIORef continued . Just $
HeaderContinuation
{ hcStreamId = streamId
, hcBlock = newPartialHeaderBlock frag
, hcEndOfStream = endOfStream
}
newPartialHeaderBlock :: HeaderBlockFragment -> PartialHeaderBlock
newPartialHeaderBlock frag =
PartialHeaderBlock
{ phbFragments = [frag]
, phbTotalSize = BS.length frag
, phbNumFrames = 1
}
addFragment
:: StreamId
-- ^ Used for error messages only
-> HeaderBlockFragment
-> PartialHeaderBlock
-> IO PartialHeaderBlock
addFragment streamId frag phb = do
when (phbTotalSize phb' > headerFragmentLimit) $
E.throwIO $
ConnectionErrorIsSent EnhanceYourCalm streamId "Header is too big"
when (phbNumFrames phb' > continuationLimit) $
E.throwIO $
ConnectionErrorIsSent EnhanceYourCalm streamId "Header is too fragmented"
return phb'
where
phb' =
PartialHeaderBlock
{ phbFragments = frag : phbFragments phb
, phbTotalSize = phbTotalSize phb + BS.length frag
, phbNumFrames = phbNumFrames phb + 1
}
completeHeaderBlock :: PartialHeaderBlock -> HeaderBlockFragment
completeHeaderBlock = BS.concat . reverse . phbFragments