Merge pull request #90 from kazu-yamamoto/half-closed-local

Half closed local

Network/HTTP2/Arch/Context.hs

@@ -160,17 +160,17 @@ setStreamState _ Stream{streamState} val = writeIORef streamState val
 opened :: Context -> Stream -> IO ()
 opened ctx@Context{concurrency} strm = do
     atomicModifyIORef' concurrency (\x -> (x+1,()))
-    setStreamState ctx strm (Open JustOpened)
+    setStreamState ctx strm (Open Nothing JustOpened)
 
 halfClosedRemote :: Context -> Stream -> IO ()
 halfClosedRemote ctx stream@Stream{streamState} = do
     closingCode <- atomicModifyIORef streamState closeHalf
     traverse_ (closed ctx stream) closingCode
   where
     closeHalf :: StreamState -> (StreamState, Maybe ClosedCode)
-    closeHalf x@(Closed _)         = (x, Nothing)
-    closeHalf (HalfClosedLocal cc) = (Closed cc, Just cc)
-    closeHalf _                    = (HalfClosedRemote, Nothing)
+    closeHalf x@(Closed _)       = (x, Nothing)
+    closeHalf (Open (Just cc) _) = (Closed cc, Just cc)
+    closeHalf _                  = (HalfClosedRemote, Nothing)
 
 halfClosedLocal :: Context -> Stream -> ClosedCode -> IO ()
 halfClosedLocal ctx stream@Stream{streamState} cc = do
@@ -181,7 +181,8 @@ halfClosedLocal ctx stream@Stream{streamState} cc = do
     closeHalf :: StreamState -> (StreamState, Bool)
     closeHalf x@(Closed _)     = (x, False)
     closeHalf HalfClosedRemote = (Closed cc, True)
-    closeHalf _                = (HalfClosedLocal cc, False)
+    closeHalf (Open Nothing o) = (Open (Just cc) o, False)
+    closeHalf _                = (Open (Just cc) JustOpened, False)
 
 closed :: Context -> Stream -> ClosedCode -> IO ()
 closed ctx@Context{concurrency,streamTable} strm@Stream{streamNumber} cc = do

Network/HTTP2/Arch/Receiver.hs

@@ -174,7 +174,7 @@ controlOrStream ctx@Context{..} conf@Config{..} ftyp header@FrameHeader{streamId
 processState :: StreamState -> Context -> Stream -> StreamId -> IO Bool
 
 -- Transition (process1)
-processState (Open (NoBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} streamId = do
+processState (Open _ (NoBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} streamId = do
     let mcl = fst <$> (getHeaderValue tokenContentLength reqvt >>= C8.readInt)
     when (just mcl (/= (0 :: Int))) $ E.throwIO $ StreamErrorIsSent ProtocolError streamId "no body but content-length is not zero"
     halfClosedRemote ctx strm
@@ -188,12 +188,12 @@ processState (Open (NoBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInp
     return False
 
 -- Transition (process2)
-processState (Open (HasBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} _streamId = do
+processState (Open hcl (HasBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} _streamId = do
     let mcl = fst <$> (getHeaderValue tokenContentLength reqvt >>= C8.readInt)
     bodyLength <- newIORef 0
     tlr <- newIORef Nothing
     q <- newTQueueIO
-    setStreamState ctx strm $ Open (Body q mcl bodyLength tlr)
+    setStreamState ctx strm $ Open hcl (Body q mcl bodyLength tlr)
     incref <- newIORef 0
     bodySource <- mkSource q $ informWindowUpdate ctx strm incref
     let inpObj = InpObj tbl mcl (readSource bodySource) tlr
@@ -205,7 +205,7 @@ processState (Open (HasBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamIn
     return False
 
 -- Transition (process3)
-processState s@(Open Continued{}) ctx strm _streamId = do
+processState s@(Open _ Continued{}) ctx strm _streamId = do
     setStreamState ctx strm s
     return True
 
@@ -343,7 +343,7 @@ checkPriority p me
 stream :: FrameType -> FrameHeader -> ByteString -> Context -> StreamState -> Stream -> IO StreamState
 
 -- Transition (stream1)
-stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx s@(Open JustOpened) Stream{streamNumber} = do
+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
@@ -361,15 +361,16 @@ stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx s@(Open JustOpened
         if endOfHeader then do
             tbl <- hpackDecodeHeader frag streamId ctx
             return $ if endOfStream then
-                        Open (NoBody tbl)
+                       -- turned into HalfClosedRemote in processState
+                        Open hcl (NoBody tbl)
                        else
-                        Open (HasBody tbl)
+                        Open hcl (HasBody tbl)
           else do
             let siz = BS.length frag
-            return $ Open $ Continued [frag] siz 1 endOfStream
+            return $ Open hcl $ Continued [frag] siz 1 endOfStream
 
 -- Transition (stream2)
-stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx (Open (Body q _ _ tlr)) _ = do
+stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx (Open _ (Body q _ _ tlr)) _ = do
     HeadersFrame _ frag <- guardIt $ decodeHeadersFrame header bs
     let endOfStream = testEndStream flags
     -- checking frag == "" is not necessary
@@ -382,23 +383,11 @@ stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx (Open (Body q _ _
         -- we don't support continuation here.
         E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "continuation in trailer is not supported"
 
--- ignore data-frame except for flow-control when we're done locally
-stream FrameData
-       FrameHeader{flags}
-       _bs
-       _ctx s@(HalfClosedLocal _)
-       _ = do
-    let endOfStream = testEndStream flags
-    if endOfStream then do
-        return HalfClosedRemote
-      else
-        return s
-
 -- Transition (stream4)
 stream FrameData
        header@FrameHeader{flags,payloadLength,streamId}
        bs
-       Context{emptyFrameRate} s@(Open (Body q mcl bodyLength _))
+       Context{emptyFrameRate} s@(Open _ (Body q mcl bodyLength _))
        _ = do
     DataFrame body <- guardIt $ decodeDataFrame header bs
     len0 <- readIORef bodyLength
@@ -424,7 +413,7 @@ stream FrameData
         return s
 
 -- Transition (stream5)
-stream FrameContinuation FrameHeader{flags,streamId} frag ctx s@(Open (Continued rfrags siz n endOfStream)) _ = do
+stream FrameContinuation FrameHeader{flags,streamId} frag ctx s@(Open hcl (Continued rfrags siz n endOfStream)) _ = do
     let endOfHeader = testEndHeader flags
     if frag == "" && not endOfHeader then do
         -- Empty Frame Flooding - CVE-2019-9518
@@ -445,11 +434,12 @@ stream FrameContinuation FrameHeader{flags,streamId} frag ctx s@(Open (Continued
             let hdrblk = BS.concat $ reverse rfrags'
             tbl <- hpackDecodeHeader hdrblk streamId ctx
             return $ if endOfStream then
-                        Open (NoBody tbl)
+                        -- turned into HalfClosedRemote in processState
+                        Open hcl (NoBody tbl)
                        else
-                        Open (HasBody tbl)
+                        Open hcl (HasBody tbl)
           else
-            return $ Open $ Continued rfrags' siz' n' endOfStream
+            return $ Open hcl $ Continued rfrags' siz' n' endOfStream
 
 -- (No state transition)
 stream FrameWindowUpdate header bs _ s strm = do
@@ -488,7 +478,7 @@ stream FramePriority header bs _ s Stream{streamNumber} = do
 
 -- this ordering is important
 stream FrameContinuation FrameHeader{streamId} _ _ _ _ = E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "continue frame cannot come here"
-stream _ FrameHeader{streamId} _ _ (Open Continued{}) _ = E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "an illegal frame follows header/continuation frames"
+stream _ FrameHeader{streamId} _ _ (Open _ Continued{}) _ = E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "an illegal frame follows header/continuation frames"
 -- Ignore frames to streams we have just reset, per section 5.1.
 stream _ _ _ _ st@(Closed (ResetByMe _)) _ = return st
 stream FrameData FrameHeader{streamId} _ _ _ _ = E.throwIO $ StreamErrorIsSent StreamClosed streamId $ fromString ("illegal data frame for " ++ show streamId)

Network/HTTP2/Arch/Sender.hs

@@ -148,13 +148,12 @@ frameSender ctx@Context{outputQ,controlQ,encodeDynamicTable,outputBufferLimit}
                 _           -> False
         (ths,_) <- toHeaderTable $ fixHeaders hdr
         off' <- headerContinue sid ths endOfStream off0
+        -- halfClosedLocal calls closed which removes
+        -- the stream from stream table.
+        when endOfStream $ halfClosedLocal ctx strm Finished
         off <- flushIfNecessary off'
         case body of
-            OutBodyNone -> do
-                -- halfClosedLocal calls closed which removes
-                -- the stream from stream table.
-                when (isServer ctx) $ halfClosedLocal ctx strm Finished
-                return off
+            OutBodyNone -> return off
             OutBodyFile (FileSpec path fileoff bytecount) -> do
                 (pread, sentinel') <- confPositionReadMaker path
                 refresh <- case sentinel' of
@@ -288,7 +287,7 @@ frameSender ctx@Context{outputQ,controlQ,encodeDynamicTable,outputBufferLimit}
         fillFrameHeader FrameData datPayloadLen streamNumber flag buf
         off'' <- handleTrailers mtrailers off'
         void tell
-        when (isServer ctx) $ halfClosedLocal ctx strm Finished
+        halfClosedLocal ctx strm Finished
         decreaseWindowSize ctx strm datPayloadLen
         if reqflush then do
             flushN off''
@@ -299,7 +298,7 @@ frameSender ctx@Context{outputQ,controlQ,encodeDynamicTable,outputBufferLimit}
         handleTrailers Nothing off0 = return off0
         handleTrailers (Just trailers) off0 = do
             (ths,_) <- toHeaderTable trailers
-            headerContinue streamNumber ths True off0
+            headerContinue streamNumber ths True {- endOfStream -} off0
 
     fillDataHeaderEnqueueNext _
                    off 0 (Just next) tlrmkr _ out reqflush = do

Network/HTTP2/Arch/Stream.hs

@@ -28,9 +28,9 @@ isHalfClosedRemote (Closed _)       = True
 isHalfClosedRemote _                = False
 
 isHalfClosedLocal :: StreamState -> Bool
-isHalfClosedLocal (HalfClosedLocal _) = True
-isHalfClosedLocal (Closed _)       = True
-isHalfClosedLocal _                = False
+isHalfClosedLocal (Open (Just _) _) = True
+isHalfClosedLocal (Closed _)        = True
+isHalfClosedLocal _                 = False
 
 isClosed :: StreamState -> Bool
 isClosed Closed{} = True
@@ -83,7 +83,7 @@ closeAllStreams (StreamTable ref) mErr' = do
     forM_ strms $ \strm -> do
       st <- readStreamState strm
       case st of
-        Open (Body q _ _ _) ->
+        Open _ (Body q _ _ _) ->
           atomically $ writeTQueue q $ maybe (Right mempty) Left mErr
         _otherwise ->
           return ()

Network/HTTP2/Arch/Types.hs

@@ -220,17 +220,16 @@ data ClosedCode = Finished
 
 data StreamState =
     Idle
-  | Open OpenState
+  | Open (Maybe ClosedCode) OpenState -- HalfClosedLocal if Just
   | HalfClosedRemote
-  | HalfClosedLocal ClosedCode
   | Closed ClosedCode
   | Reserved
 
 instance Show StreamState where
     show Idle                = "Idle"
-    show Open{}              = "Open"
+    show (Open Nothing _)    = "Open"
+    show (Open (Just e) _)   = "HalfClosedLocal: " ++ show e
     show HalfClosedRemote    = "HalfClosedRemote"
-    show (HalfClosedLocal e) = "HalfClosedLocal: " ++ show e
     show (Closed e)          = "Closed: " ++ show e
     show Reserved            = "Reserved"