diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,35 @@ # ChangeLog +## 0.3.23 + +Adds `scMaxConnections` to `ServerConfig` and `setMaxBidiStreams` and +`setMaxUniStreams` to `Network.QUIC`. In `Network.QUIC.Internal`, +`newRecvQ` takes a `RecvBudget`, `getMyStreamId` takes a `StreamId`, +`queueTx` and `checkBlocked` are in `STM`, the DATAGRAM queue is a +`DatagramQ`, and `setPeerCIDAndRetireCIDs` is replaced by +`addPeerCIDAndRetire`. + +* Take the flow control window in the transaction that finds it + [#224](https://github.com/kazu-yamamoto/quic/pull/224) +* Add a new peer CID along with the retirement it asks for + [#225](https://github.com/kazu-yamamoto/quic/pull/225) +* Drop a DATAGRAM when too many wait for the application + [#227](https://github.com/kazu-yamamoto/quic/pull/227) +* Share one byte budget among a server's receive queues + [#228](https://github.com/kazu-yamamoto/quic/pull/228) +* Refuse a connection over scMaxConnections with CONNECTION_REFUSED + [#229](https://github.com/kazu-yamamoto/quic/pull/229) +* Let the application set how many streams the peer may open + [#230](https://github.com/kazu-yamamoto/quic/pull/230) +* Require a Retry of a client whose token validates nothing + [#231](https://github.com/kazu-yamamoto/quic/pull/231) +* Check a unidirectional stream against the unidirectional streams we opened + [#232](https://github.com/kazu-yamamoto/quic/pull/232) +* Answer a frame received in 0-RTT in 1-RTT + [#233](https://github.com/kazu-yamamoto/quic/pull/233) +* Make no congestion event and send no PING for lost ACK-only packets + [#234](https://github.com/kazu-yamamoto/quic/pull/234) + ## 0.3.22 Requires crypto-token 0.2.1. The public API is unchanged. In diff --git a/Network/QUIC.hs b/Network/QUIC.hs --- a/Network/QUIC.hs +++ b/Network/QUIC.hs @@ -36,6 +36,10 @@ resetStream, stopStream, + -- ** Limits on the peer's streams + setMaxBidiStreams, + setMaxUniStreams, + -- * IO recvStream, resetReceived, diff --git a/Network/QUIC/Client/Run.hs b/Network/QUIC/Client/Run.hs --- a/Network/QUIC/Client/Run.hs +++ b/Network/QUIC/Client/Run.hs @@ -162,7 +162,7 @@ -- thread is the only one the socket has ever had. flip E.onException (NS.unsafeClose sock) $ do when ccSockConnected $ NS.connect sock peersa - q <- newRecvQ + q <- newRecvQ =<< newRecvBudget maxBound retire0 <- newTVarIO False sref <- newIORef (sock, retire0) pathInfo <- newPathInfo peersa @@ -191,7 +191,7 @@ let myAuthCIDs = defaultAuthCIDs{initSrcCID = Just myCID} peerAuthCIDs = defaultAuthCIDs{initSrcCID = Just peerCID, origDstCID = Just peerCID} genSRT <- makeGenStatelessReset - connRecvDatagramQ <- newTQueueIO + connRecvDatagramQ <- newDatagramQ conn <- clientConnection conf diff --git a/Network/QUIC/Config.hs b/Network/QUIC/Config.hs --- a/Network/QUIC/Config.hs +++ b/Network/QUIC/Config.hs @@ -194,6 +194,11 @@ -- ^ Request a certificate from clients. , scRequireRetry :: Bool -- ^ Requiring QUIC retry. + , scMaxConnections :: Int + -- ^ How many connections may be open at once. An Initial packet that + -- would open one more is answered with CONNECTION_REFUSED. 0 or less, + -- the default being 0, sets no limit; a server open to the Internet + -- should set one. , scSessionManager :: SessionManager -- ^ A session manager of TLS 1.3. , scDebugLog :: Maybe FilePath @@ -243,6 +248,7 @@ , scALPN = Nothing , scWantClientCert = False , scRequireRetry = False + , scMaxConnections = 0 , scSessionManager = noSessionManager , scDebugLog = Nothing , scTicketLifetime = 7200 diff --git a/Network/QUIC/Connection/Crypto.hs b/Network/QUIC/Connection/Crypto.hs --- a/Network/QUIC/Connection/Crypto.hs +++ b/Network/QUIC/Connection/Crypto.hs @@ -19,6 +19,7 @@ initializeCoder, initializeCoder1RTT, updateCoder1RTT, + genNiteCoder, getCoder, getProtector, -- diff --git a/Network/QUIC/Connection/Migration.hs b/Network/QUIC/Connection/Migration.hs --- a/Network/QUIC/Connection/Migration.hs +++ b/Network/QUIC/Connection/Migration.hs @@ -12,7 +12,7 @@ getNewMyCID, getMyCIDSeqNum, setMyCID, - setPeerCIDAndRetireCIDs, + addPeerCIDAndRetire, retirePeerCID, retireMyCID, isMyCIDSeqNumIssued, @@ -206,13 +206,26 @@ setPeerRetirePriorTo Connection{..} rpt = atomically $ modifyTVar' peerCIDDB $ \db -> db{retirePriorTo = rpt} --- | Receiving NewConnectionID -setPeerCIDAndRetireCIDs :: Connection -> Int -> IO [Int] -setPeerCIDAndRetireCIDs Connection{..} rpt = atomically $ do - db <- readTVar peerCIDDB - let (db', ns) = arrange rpt db - writeTVar peerCIDDB db' - return ns +-- | Receiving NewConnectionID: adding the CID and retiring those below +-- Retire Prior To in one go. Retired first, nothing might be left to use; +-- added first, the limit is checked only after the retirement +-- (RFC 9000 Sec 5.1.1). 'Nothing' if the limit is exceeded. +addPeerCIDAndRetire :: Connection -> CIDInfo -> Int -> IO (Maybe [Int]) +addPeerCIDAndRetire conn@Connection{..} cidInfo rpt = do + let lim = activeConnectionIdLimit $ getMyParameters conn + atomically $ do + db0 <- readTVar peerCIDDB + let db1 + | Map.member (cidInfoCID cidInfo) (revInfos db0) = db0 -- retransmitted + | otherwise = add cidInfo db0 + (db2, ns) + | rpt >= retirePriorTo db0 = arrange rpt db1 + | otherwise = (db1, []) + if Map.size (revInfos db2) > lim + then return Nothing + else do + writeTVar peerCIDDB db2 + return $ Just ns arrange :: Int -> CIDDB -> (CIDDB, [Int]) arrange rpt db@CIDDB{..} = (db', dropSeqnums) @@ -220,11 +233,10 @@ (toDrops, cidInfos') = IntMap.partitionWithKey (\k _ -> k < rpt) cidInfos dropSeqnums = IntMap.foldrWithKey (\k _ ks -> k : ks) [] toDrops dropCIDs = IntMap.foldr (\c r -> cidInfoCID c : r) [] toDrops - -- IntMap.findMin is a partial function. - -- But receiver guarantees that there is at least one cidinfo. + -- The caller adds a CID at or above rpt first, so one is left. usedCIDInfo' | cidInfoSeq usedCIDInfo >= rpt = usedCIDInfo - | otherwise = snd $ IntMap.findMin cidInfos' + | otherwise = maybe usedCIDInfo snd $ IntMap.lookupMin cidInfos' revInfos' = foldr Map.delete revInfos dropCIDs db' = db diff --git a/Network/QUIC/Connection/Stream.hs b/Network/QUIC/Connection/Stream.hs --- a/Network/QUIC/Connection/Stream.hs +++ b/Network/QUIC/Connection/Stream.hs @@ -14,6 +14,8 @@ updatePeerStreamId, checkStreamIdRoom, releasePeerStream, + setMaxBidiStreams, + setMaxUniStreams, ) where import Control.Concurrent.STM @@ -24,12 +26,15 @@ import Network.QUIC.Connection.Types import Network.QUIC.Connector import Network.QUIC.Imports -import Network.QUIC.Parameters import Network.QUIC.Types -getMyStreamId :: Connection -> IO Int -getMyStreamId Connection{..} = do - next <- currentStream <$> readTVarIO myStreamId +-- | The last stream we opened of the kind 'StreamId' is. +getMyStreamId :: Connection -> StreamId -> IO Int +getMyStreamId Connection{..} sid = do + let tvar + | isUnidirectional sid = myUniStreamId + | otherwise = myStreamId + next <- currentStream <$> readTVarIO tvar return $ next - 4 possibleMyStreams :: Connection -> IO Int @@ -149,16 +154,12 @@ | otherwise = peerUniStreamId conn atomicModifyIORef' ref checkConc where - params = getMyParameters conn - initialStreams - | dir == Bidirectional = initialMaxStreamsBidi params - | otherwise = initialMaxStreamsUni params checkConc conc@Concurrency{..} = let StreamIdBase base = maxStreams closed = closedStreams + 1 - base' = initialStreams + closed + base' = targetStreams + closed conc' = conc{closedStreams = closed} - in if base' - base >= max 1 (initialStreams `div` 2) + in if base' - base >= max 1 (targetStreams `div` 2) then (conc'{maxStreams = StreamIdBase base'}, Just base') else (conc', Nothing) @@ -182,3 +183,34 @@ dir | isUnidirectional sid = Unidirectional | otherwise = Bidirectional + +-- | Setting how many bidirectional streams the peer may have open at once, +-- 'initialMaxStreamsBidi' to begin with. A larger number is announced +-- at once. MAX_STREAMS cannot go down (RFC 9000 Sec 4.6), so a smaller +-- one takes effect as the peer's open streams are closed. +setMaxBidiStreams :: Connection -> Int -> IO () +setMaxBidiStreams conn = setMaxStreams conn Bidirectional + +-- | 'setMaxBidiStreams' for unidirectional streams. +setMaxUniStreams :: Connection -> Int -> IO () +setMaxUniStreams conn = setMaxStreams conn Unidirectional + +setMaxStreams :: Connection -> Direction -> Int -> IO () +setMaxStreams conn dir n0 = do + mx <- atomicModifyIORef' ref retarget + case mx of + Nothing -> return () + Just nms -> sendFrames conn RTT1Level [MaxStreams dir nms] + where + ref + | dir == Bidirectional = peerStreamId conn + | otherwise = peerUniStreamId conn + -- RFC 9000 Sec 4.6: no more than 2^60 streams. + n = max 0 $ min n0 (2 ^ (60 :: Int)) + retarget conc@Concurrency{..} = + let StreamIdBase base = maxStreams + base' = min (2 ^ (60 :: Int)) (n + closedStreams) + conc' = conc{targetStreams = n} + in if base' > base + then (conc'{maxStreams = StreamIdBase base'}, Just base') + else (conc', Nothing) diff --git a/Network/QUIC/Connection/Types.hs b/Network/QUIC/Connection/Types.hs --- a/Network/QUIC/Connection/Types.hs +++ b/Network/QUIC/Connection/Types.hs @@ -195,6 +195,8 @@ , maxStreams :: StreamIdBase , closedStreams :: Int -- ^ For the peer's streams: how many of them we are done with + , targetStreams :: Int + -- ^ For the peer's streams: how many may be open at once (#86) } deriving (Show) @@ -207,6 +209,7 @@ | otherwise = if bidi then 1 else 3 maxStreams = StreamIdBase n closedStreams = 0 + targetStreams = n -- | The peer-initiated streams of one type that have been opened so far. -- @@ -605,8 +608,6 @@ ---------------------------------------------------------------- type SendStreamQ = TQueue TxStreamData - -type DatagramQ = TQueue ByteString data Shared = Shared { sharedCloseSent :: IORef Bool diff --git a/Network/QUIC/IO.hs b/Network/QUIC/IO.hs --- a/Network/QUIC/IO.hs +++ b/Network/QUIC/IO.hs @@ -42,8 +42,8 @@ -- | Counting and queueing at once, so that the offset given is the -- stream's next one and the queue holds a stream's data in order. -queueTx :: Connection -> Stream -> [StreamData] -> Int -> IO () -queueTx conn s dats len = atomically $ do +queueTx :: Connection -> Stream -> [StreamData] -> Int -> STM () +queueTx conn s dats len = do off <- addTxStreamData s len addTxData conn len putSendStreamQSTM conn $ TxStreamData s off dats len False @@ -74,16 +74,22 @@ flowControl dats len wait = do -- FLOW CONTROL: MAX_STREAM_DATA: send: respecting peer's limit -- FLOW CONTROL: MAX_DATA: send: respecting peer's limit - eblocked <- checkBlocked s len wait - case eblocked of - Right n - -- Counted before queued, so that a reset in between takes - -- it into the final size. - | len == n -> queueTx conn s dats len - | otherwise -> do - let (dats1, dats2) = split n dats + -- The window is taken in the transaction that finds it. Two + -- streams sending at once used to see the same window and send it + -- twice over. + eblocked <- atomically $ do + r <- checkBlocked s len wait + case r of + Right n -> do + let (dats1, dats2) + | len == n = (dats, []) + | otherwise = split n dats queueTx conn s dats1 n - flowControl dats2 (len - n) False + return $ Right (n, dats2) + Left blocked -> return $ Left blocked + case eblocked of + Right (n, dats2) -> + when (n < len) $ flowControl dats2 (len - n) False Left blocked -> do -- Read each time round rather than once: the handshake may -- finish while we are waiting for the window, and a BLOCKED @@ -117,8 +123,8 @@ where len = BS.length bs -checkBlocked :: Stream -> Int -> Bool -> IO (Either Blocked Int) -checkBlocked s len wait = atomically $ do +checkBlocked :: Stream -> Int -> Bool -> STM (Either Blocked Int) +checkBlocked s len wait = do let conn = streamConnection s checkPeerClosedSTM conn strmFlow <- readStreamFlowTx s @@ -343,8 +349,8 @@ -- | Receive all available DATAGRAM frames. -- This function is non-blocking. recvDatagramMany :: Connection -> IO [ByteString] -recvDatagramMany = atomically . flushTQueue . connRecvDatagramQ +recvDatagramMany = atomically . flushDatagramQSTM . connRecvDatagramQ -- | Receiving a DATAGRAM frame in STM. recvDatagramSTM :: Connection -> STM ByteString -recvDatagramSTM conn = readTQueue (connRecvDatagramQ conn) `orElse` peerClosedSTM conn +recvDatagramSTM conn = readDatagramQSTM (connRecvDatagramQ conn) `orElse` peerClosedSTM conn diff --git a/Network/QUIC/Internal.hs b/Network/QUIC/Internal.hs --- a/Network/QUIC/Internal.hs +++ b/Network/QUIC/Internal.hs @@ -17,6 +17,7 @@ module Network.QUIC.Client.WatchDog, module Network.QUIC.Socket, sendOutput, + processFrame, ) where import Network.QUIC.Client.Reader ( @@ -35,6 +36,7 @@ import Network.QUIC.Parameters import Network.QUIC.QLogger import Network.QUIC.Qlog +import Network.QUIC.Receiver (processFrame) import Network.QUIC.Recovery import Network.QUIC.Sender (sendOutput) import Network.QUIC.Socket diff --git a/Network/QUIC/Packet.hs b/Network/QUIC/Packet.hs --- a/Network/QUIC/Packet.hs +++ b/Network/QUIC/Packet.hs @@ -2,6 +2,7 @@ -- * Encode encodeVersionNegotiationPacket, encodeRetryPacket, + encodeInitialRefusal, encodePlainPacket, makeStatelessReset, diff --git a/Network/QUIC/Packet/Encode.hs b/Network/QUIC/Packet/Encode.hs --- a/Network/QUIC/Packet/Encode.hs +++ b/Network/QUIC/Packet/Encode.hs @@ -5,10 +5,13 @@ encodeVersionNegotiationPacket, encodeRetryPacket, encodePlainPacket, + encodeInitialRefusal, makeStatelessReset, ) where import qualified Data.ByteString as BS +import qualified Data.ByteString.Short as Short +import Foreign.Marshal.Alloc (allocaBytes) import System.Random (getStdRandom, randomRIO) #if MIN_VERSION_random(1,3,0) import System.Random (uniformByteString) @@ -54,6 +57,36 @@ ---------------------------------------------------------------- +-- | An Initial packet closing with CONNECTION_REFUSED, made without a +-- connection: the keys are the ones the client's Destination Connection +-- ID gives (RFC 9001 Sec 5.2), and that ID is our Source Connection ID. +encodeInitialRefusal :: Version -> CID -> CID -> IO ByteString +encodeInitialRefusal ver peerCID origCID = + allocaBytes bufLen $ \buf -> allocaBytes bufLen $ \encBuf -> do + (coder, protector) <- + genNiteCoder False ver defaultCipher $ initialSecrets ver origCID + wbuf <- newWriteBuffer buf bufLen + (epn, epnLen) <- + encodeLongHeaderPP' True wbuf InitialPacketType ver peerCID origCID (Flags 0) 0 + encodeInt' wbuf 0 -- token + (len, _) <- + protectPayloadHeader' + encBuf + defaultCipher + coder + protector + wbuf + [ConnectionClose ConnectionRefused 0 Short.empty] + 0 + epn + epnLen + buf + Nothing + InitialLevel + if len < 0 then return BS.empty else BS.packCStringLen (castPtr buf, len) + where + bufLen = 1500 + encodeRetryPacket :: RetryPacket -> IO ByteString encodeRetryPacket (RetryPacket ver dCID sCID token (Left odCID)) = withWriteBuffer maximumQUICHeaderSize $ \wbuf -> do save wbuf @@ -204,9 +237,22 @@ -> PacketNumber -> IO (EncodedPacketNumber, Int) encodeLongHeaderPP conn wbuf pkttyp ver dCID sCID flags pn = do + quicBit <- greaseQuicBit <$> getPeerParameters conn + encodeLongHeaderPP' quicBit wbuf pkttyp ver dCID sCID flags pn + +encodeLongHeaderPP' + :: Bool + -> WriteBuffer + -> LongHeaderPacketType + -> Version + -> CID + -> CID + -> Flags Raw + -> PacketNumber + -> IO (EncodedPacketNumber, Int) +encodeLongHeaderPP' quicBit wbuf pkttyp ver dCID sCID flags pn = do let el@(_, pnLen) = encodePacketNumber 0 {- dummy -} pn pp = encodePktNumLength pnLen - quicBit <- greaseQuicBit <$> getPeerParameters conn Flags flags' <- encodeLongHeaderFlags ver pkttyp flags pp quicBit write8 wbuf flags' encodeLongHeader wbuf ver dCID sCID @@ -227,15 +273,46 @@ -> Bool -> IO (Int, Int) protectPayloadHeader conn wbuf frames pn epn epnLen headerBeg mlen lvl keyPhase = do + cipher <- getCipher conn lvl + coder <- getCoder conn lvl keyPhase + protector <- getProtector conn lvl + protectPayloadHeader' + (encodeBuf conn) + cipher + coder + protector + wbuf + frames + pn + epn + epnLen + headerBeg + mlen + lvl + +-- | 'protectPayloadHeader' with what it takes from a connection given. +protectPayloadHeader' + :: Buffer + -> Cipher + -> Coder + -> Protector + -> WriteBuffer + -> [Frame] + -> PacketNumber + -> EncodedPacketNumber + -> Int + -> Buffer + -> Maybe Int + -> EncryptionLevel + -> IO (Int, Int) +protectPayloadHeader' encBuf cipher coder protector wbuf frames pn epn epnLen headerBeg mlen lvl = do -- Real size is maximumUdpPayloadSize. But smaller is better. - let encBuf = encodeBuf conn - encBufLen = 1500 - 20 - 8 + let encBufLen = 1500 - 20 - 8 payloadWithoutPaddingSiz <- encodeFramesWithPadding encBuf encBufLen frames - cipher <- getCipher conn lvl -- before length or packer number lengthOrPNBeg <- currentOffset wbuf (packetLen, headerLen, plainLen, tagLen, padLen) <- - calcLen cipher lengthOrPNBeg payloadWithoutPaddingSiz + calcLen lengthOrPNBeg payloadWithoutPaddingSiz when (lvl /= RTT1Level) $ writeLen (epnLen + plainLen + tagLen) pnBeg <- currentOffset wbuf writeEpn epnLen @@ -245,8 +322,6 @@ header <- mkBS headerBeg headerLen -- let sampleBeg = pnBeg `plusPtr` 4 - coder <- getCoder conn lvl keyPhase - protector <- getProtector conn lvl setSample protector sampleBeg len <- encrypt coder cryptoBeg plaintext (AssDat header) pn -- @@ -259,7 +334,7 @@ ok <- withMask protector $ protectHeader headerBeg pnBeg epnLen return $ if ok then (packetLen, padLen) else (-1, -1) where - calcLen cipher lengthOrPNBeg payloadWithoutPaddingSiz = do + calcLen lengthOrPNBeg payloadWithoutPaddingSiz = do let headerLen = (lengthOrPNBeg `minusPtr` headerBeg) -- length: assuming 2byte length diff --git a/Network/QUIC/Receiver.hs b/Network/QUIC/Receiver.hs --- a/Network/QUIC/Receiver.hs +++ b/Network/QUIC/Receiver.hs @@ -3,6 +3,7 @@ module Network.QUIC.Receiver ( receiver, + processFrame, ) where import qualified Control.Exception as E @@ -11,7 +12,6 @@ import Network.TLS (AlertDescription (..)) import System.Log.FastLogger -import Control.Concurrent.STM import Network.QUIC.Config import Network.QUIC.Connection import Network.QUIC.Connector @@ -259,11 +259,10 @@ ) guardStream _ _ _ = return () --- fixme: what about unidirection stream? streamNotCreatedYet :: Connection -> StreamId -> ReasonPhrase -> IO () streamNotCreatedYet conn sid emsg | isInitiated conn sid = do - curSid <- getMyStreamId conn + curSid <- getMyStreamId conn sid when (sid > curSid) $ closeConnection conn StreamStateError emsg streamNotCreatedYet _ _ _ = return () @@ -464,11 +463,18 @@ Reassembled _ -> return () deliverStream conn found strm +-- | The level to answer a frame received at: 0-RTT goes from client to +-- server only, so a server answers 0-RTT in 1-RTT (RFC 9000 Sec 17.2.3). +replyLevel :: EncryptionLevel -> EncryptionLevel +replyLevel RTT0Level = RTT1Level +replyLevel lvl = lvl + processFrame :: Connection -> EncryptionLevel -> Frame -> IO () processFrame _ _ Padding{} = return () processFrame conn lvl Ping = do -- see ackEli above - when (lvl /= InitialLevel && lvl /= RTT1Level) $ sendFrames conn lvl [] + when (lvl /= InitialLevel && lvl /= RTT1Level) $ + sendFrames conn (replyLevel lvl) [] processFrame conn lvl (Ack ackInfo@(AckInfo largestAcked _ _) ackDelay) = do when (lvl == RTT0Level) $ closeConnection conn ProtocolViolation "ACK" -- RFC 9000 Sec 19.3.1: walking the ranges down from the largest @@ -576,7 +582,7 @@ -- No more STREAM frames may go out on it. Without this, -- 'sendStream' went on working after the RESET_STREAM. finalSize <- resetTxStream strm - sendFrames conn lvl [ResetStream sid err finalSize] + sendFrames conn (replyLevel lvl) [ResetStream sid err finalSize] processFrame _ _ (CryptoF _ "") = return () processFrame conn lvl (CryptoF off cdat) = do when (lvl == RTT0Level) $ @@ -704,29 +710,7 @@ consistent <- isPeerCIDConsistent conn cidInfo unless consistent $ closeConnection conn ProtocolViolation "NEW_CONNECTION_ID repeated differently" - -- Retiring CIDs first then add a new CID. - -- - -- RFC 9000 Sec 5.1.1 says: - -- An endpoint MAY send connection IDs that temporarily exceed a - -- peer's limit if the NEW_CONNECTION_ID frame also requires the - -- retirement of any excess, by including a sufficiently large - -- value in the Retire Prior To field. prevRetirePriorTo <- getPeerRetirePriorTo conn - -- RFC 900 Sec 19.15 says: - -- Once a sender indicates a Retire Prior To value, smaller values - -- sent in subsequent NEW_CONNECTION_ID frames have no effect. A - -- receiver MUST ignore any Retire Prior To fields that do not - -- increase the largest received Retire Prior To value. - when (retirePriorTo >= prevRetirePriorTo) $ do - -- RFC 9000 Sec 5.1.2 says: - -- Upon receipt of an increased Retire Prior To field, the - -- peer MUST stop using the corresponding connection IDs and - -- retire them with RETIRE_CONNECTION_ID frames before adding - -- the newly provided connection ID to the set of active - -- connection IDs. - seqNums <- setPeerCIDAndRetireCIDs conn retirePriorTo -- upadting RPT - sendFrames conn RTT1Level $ map RetireConnectionID seqNums - -- Adding a new CID if seqNum < prevRetirePriorTo then -- RFC 9000 Sec 19.15 says: @@ -736,11 +720,27 @@ -- MUST send a corresponding RETIRE_CONNECTION_ID frame -- that retires the newly received connection ID, unless -- it has already done so for that sequence number. + -- + -- Retire Prior To is not above seqNum here, so it does not + -- increase and is ignored (RFC 9000 Sec 19.15). sendFrames conn RTT1Level [RetireConnectionID seqNum] else do - ng <- not <$> addPeerCID conn cidInfo - when ng $ - closeConnection conn ConnectionIdLimitError "NEW_CONNECTION_ID limit error" + -- RFC 9000 Sec 5.1.1 says: + -- An endpoint MAY send connection IDs that temporarily exceed a + -- peer's limit if the NEW_CONNECTION_ID frame also requires the + -- retirement of any excess, by including a sufficiently large + -- value in the Retire Prior To field. + -- + -- The new CID is added along with the retirement, so that one + -- retiring all the others leaves it to use (#223). + mseqNums <- addPeerCIDAndRetire conn cidInfo retirePriorTo + case mseqNums of + Nothing -> + closeConnection conn ConnectionIdLimitError "NEW_CONNECTION_ID limit error" + Just seqNums -> + unless (null seqNums) $ + sendFrames conn RTT1Level $ + map RetireConnectionID seqNums processFrame conn RTT1Level (RetireConnectionID sn) = do -- RFC 9000 Sec 19.16: "Receipt of a RETIRE_CONNECTION_ID frame -- containing a sequence number greater than any previously sent to the @@ -805,7 +805,7 @@ closeConnection conn ProtocolViolation "DATAGRAM not supported" when (1 + lenSize + BS.length dat > limitBytes) $ closeConnection conn ProtocolViolation "DATAGRAM size violation" - atomically $ writeTQueue (connRecvDatagramQ conn) dat + writeDatagramQ (connRecvDatagramQ conn) dat processFrame conn _ _ = closeConnection conn ProtocolViolation "Frame is not allowed" -- Return value indicates duplication. diff --git a/Network/QUIC/Recovery.hs b/Network/QUIC/Recovery.hs --- a/Network/QUIC/Recovery.hs +++ b/Network/QUIC/Recovery.hs @@ -31,6 +31,8 @@ -- Release releaseByRetry, releaseOldest, + onPacketsLost, + retransmit, -- Timer beforeAntiAmp, ldccTimer, @@ -48,6 +50,9 @@ LDCC, newLDCC, lostCandidates, + recoveryCC, + ptoPing, + congestionRecoveryStartTime, SentPackets (..), qlogSent, ) where @@ -61,4 +66,4 @@ import Network.QUIC.Recovery.Release import Network.QUIC.Recovery.Timer import Network.QUIC.Recovery.Types -import Network.QUIC.Recovery.Utils (sendPing) +import Network.QUIC.Recovery.Utils (retransmit, sendPing) diff --git a/Network/QUIC/Recovery/Interface.hs b/Network/QUIC/Recovery/Interface.hs --- a/Network/QUIC/Recovery/Interface.hs +++ b/Network/QUIC/Recovery/Interface.hs @@ -46,6 +46,8 @@ unless (null packets) $ do onPacketsLost ldcc packets retransmit ldcc packets + -- Nothing to resend: a PING, to hurry the peer along all the same. + unless (any spAckEliciting packets) $ sendPing ldcc lvl setLossDetectionTimer ldcc resender :: LDCC -> IO () diff --git a/Network/QUIC/Recovery/Release.hs b/Network/QUIC/Recovery/Release.hs --- a/Network/QUIC/Recovery/Release.hs +++ b/Network/QUIC/Recovery/Release.hs @@ -63,16 +63,20 @@ ---------------------------------------------------------------- onPacketsLost :: LDCC -> Seq SentPacket -> IO () -onPacketsLost ldcc@LDCC{..} lostPackets = case Seq.viewr lostPackets of - EmptyR -> return () - _ :> lastPkt -> do - decreaseCC ldcc lostPackets - isRecovery <- - inCongestionRecovery (spTimeSent lastPkt) . congestionRecoveryStartTime - <$> readTVarIO recoveryCC - onCongestionEvent ldcc lostPackets isRecovery - mapM_ (qlogPacketLost ldcc . LostPacket) lostPackets +onPacketsLost ldcc@LDCC{..} lostPackets = do + decreaseCC ldcc lostPackets + -- RFC 9002 B.8: only packets in flight make a congestion event. An + -- ACK-only packet lost used to halve the window. + case Seq.viewr inflight of + EmptyR -> return () + _ :> lastPkt -> do + isRecovery <- + inCongestionRecovery (spTimeSent lastPkt) . congestionRecoveryStartTime + <$> readTVarIO recoveryCC + onCongestionEvent ldcc inflight isRecovery + mapM_ (qlogPacketLost ldcc . LostPacket) lostPackets where + inflight = Seq.filter ((> 0) . inFlightBytes) lostPackets onCongestionEvent = updateCC ---------------------------------------------------------------- diff --git a/Network/QUIC/Recovery/Utils.hs b/Network/QUIC/Recovery/Utils.hs --- a/Network/QUIC/Recovery/Utils.hs +++ b/Network/QUIC/Recovery/Utils.hs @@ -25,10 +25,11 @@ ---------------------------------------------------------------- +-- | Resending the ack-eliciting packets lost. Lost ACK-only packets need +-- nothing: a PING for them used to send a probe outside the congestion +-- window and resend a packet that was not lost. retransmit :: LDCC -> Seq SentPacket -> IO () -retransmit ldcc lostPackets - | null packetsToBeResent = getEncryptionLevel ldcc >>= sendPing ldcc - | otherwise = mapM_ put packetsToBeResent +retransmit ldcc lostPackets = mapM_ put packetsToBeResent where packetsToBeResent = Seq.filter spAckEliciting lostPackets put = putRetrans ldcc . spPlainPacket diff --git a/Network/QUIC/Server.hs b/Network/QUIC/Server.hs --- a/Network/QUIC/Server.hs +++ b/Network/QUIC/Server.hs @@ -12,6 +12,7 @@ scALPN, scWantClientCert, scRequireRetry, + scMaxConnections, scUse0RTT, scMaxDatagramFrameSize, scCiphers, diff --git a/Network/QUIC/Server/Reader.hs b/Network/QUIC/Server/Reader.hs --- a/Network/QUIC/Server/Reader.hs +++ b/Network/QUIC/Server/Reader.hs @@ -9,6 +9,7 @@ liveConnections, spawnConnection, waitNoConnection, + releaseConnection, spawnDispatcher, waitNoDispatcher, tokenMgr, @@ -70,6 +71,8 @@ , stopServerNow :: IO () -- ^ Stopping this server: what the shutdown handler is handed, and -- what a connection's 'stop' reaches it by. + , connectionCount :: TVar Int + -- ^ Connections forked and not yet ended, for 'scMaxConnections'. } newDispatch :: ServerConfig -> IO () -> IO Dispatch @@ -83,6 +86,7 @@ <*> T.newThreadManager <*> T.newThreadManager <*> pure stopNow + <*> newTVarIO 0 where conf = CT.defaultConfig @@ -128,6 +132,18 @@ spawnConnection :: Dispatch -> IO () -> IO () spawnConnection Dispatch{..} = T.forkManaged connectionThreads "QUIC connection" +-- | Counting a connection in, unless there are 'scMaxConnections' already. +reserveConnection :: Dispatch -> Int -> IO Bool +reserveConnection Dispatch{..} lim = atomically $ do + n <- readTVar connectionCount + if lim <= 0 || n < lim + then writeTVar connectionCount (n + 1) >> return True + else return False + +-- | Counting a connection out. +releaseConnection :: Dispatch -> IO () +releaseConnection Dispatch{..} = atomically $ modifyTVar' connectionCount (subtract 1) + -- | Waiting until no connection is running. waitNoConnection :: Dispatch -> IO () waitNoConnection Dispatch{..} = T.waitUntilAllGone connectionThreads @@ -146,20 +162,24 @@ -- Source CID -> RecvQ -- Initials and RTT0 are queued before Conneciton is created. -newtype RecvQDict = RecvQDict (LRUCacheRef CID RecvQ) +data RecvQDict = RecvQDict RecvBudget (LRUCacheRef CID RecvQ) recvQDictSize :: Int recvQDictSize = 100 +-- | The queues share one budget, the endpoint's (#170). newRecvQDict :: IO RecvQDict -newRecvQDict = RecvQDict <$> LRUCache.newLRUCacheRef recvQDictSize +newRecvQDict = + RecvQDict + <$> newRecvBudget recvBudgetLimit + <*> LRUCache.newLRUCacheRef recvQDictSize -- Looking up and insert a new RecvQ if not exist. lookupInsertRecvQDict :: RecvQDict -> CID -> IO (RecvQ, Bool) -lookupInsertRecvQDict (RecvQDict ref) dcid = LRUCache.cached ref dcid newRecvQ +lookupInsertRecvQDict (RecvQDict budget ref) dcid = LRUCache.cached ref dcid $ newRecvQ budget lookupRecvQDict :: RecvQDict -> CID -> IO (Maybe RecvQ) -lookupRecvQDict (RecvQDict ref) dcid = LRUCache.cached' ref dcid +lookupRecvQDict (RecvQDict _ ref) dcid = LRUCache.cached' ref dcid ---------------------------------------------------------------- @@ -277,7 +297,7 @@ -> (CryptPacket, EncryptionLevel, Int) -> IO () dispatch - Dispatch{..} + d@Dispatch{..} ServerConfig{..} forkConnection logAction @@ -330,7 +350,9 @@ -- Retry -- but the address it arrives from has -- proved nothing. fresh <- isTokenFresh ct - pushToAcceptFirst $ fresh && isTokenAddress ct peersa + if fresh && isTokenAddress ct peersa + then pushToAcceptFirst True + else unvalidated -- A token we cannot decrypt is not a token. RFC 9000 -- section 8.1.3: "If the token is invalid, then the -- server SHOULD proceed as if the client did not have @@ -339,14 +361,35 @@ -- rubbish than for sending nothing, and turned off -- the anti-amplification limit for an address that -- had proved nothing. - Nothing -> pushToAcceptFirst False + Nothing -> unvalidated Just conn -> writeRecvQ (connRecvQ conn) $ mkReceivedPacket cpkt tim siz lvl where myVersions = scVersions + -- An address no token has validated: a Retry if one is required, + -- as for an Initial with no token at all. + unvalidated + | scRequireRetry = sendRetry + | otherwise = pushToAcceptFirst False pushToAcceptQ myAuthCIDs peerAuthCIDs addrValid = do let key = nonZeroLengthCID sCID peersa + mq <- lookupRecvQDict srcTable key + case mq of + -- A retransmission of an Initial already accepted. + Just q -> writeRecvQ q $ mkReceivedPacket cpkt tim siz lvl + Nothing -> do + ok <- reserveConnection d scMaxConnections + if ok + then accept myAuthCIDs peerAuthCIDs addrValid key + else -- RFC 9000 Sec 5.2.2: "If a server refuses to + -- accept a new connection, it SHOULD send an Initial + -- packet containing a CONNECTION_CLOSE frame with + -- error code CONNECTION_REFUSED." + encodeInitialRefusal peerVer sCID dCID >>= send' + accept myAuthCIDs peerAuthCIDs addrValid key = do (q, exist) <- lookupInsertRecvQDict srcTable key writeRecvQ q $ mkReceivedPacket cpkt tim siz lvl + -- Another dispatcher got there first. + when exist $ releaseConnection d unless exist $ do let reg = registerConnectionDict dstTable unreg cid = diff --git a/Network/QUIC/Server/Run.hs b/Network/QUIC/Server/Run.hs --- a/Network/QUIC/Server/Run.hs +++ b/Network/QUIC/Server/Run.hs @@ -234,7 +234,7 @@ -> Dispatch -> Accept -> IO () -runServer conf server0 dispatch acc = do +runServer conf server0 dispatch acc = (`E.finally` done) $ do labelMe "QUIC runServer" E.bracket open clse $ \(ConnRes conn myAuthCIDs _reader) -> handleLogUnit (debugLog conn) $ do @@ -316,6 +316,9 @@ setConnectionClosed conn closure conn ldcc ex where + done = do + closeRecvQ $ accRecvQ acc + releaseConnection dispatch open = createServerConnection conf dispatch acc -- 'killReaders' stops this connection's own threads and sees them out, -- which has to happen before the buffers they use are freed. Nothing @@ -367,7 +370,7 @@ (debugLog, dclean) <- dirDebugLogger scDebugLog ocid `E.onException` qclean flip E.onException (dclean >> qclean) $ do debugLog $ "Original CID: " <> bhow ocid - connRecvDatagramQ <- newTQueueIO + connRecvDatagramQ <- newDatagramQ conn <- serverConnection conf diff --git a/Network/QUIC/Types/Queue.hs b/Network/QUIC/Types/Queue.hs --- a/Network/QUIC/Types/Queue.hs +++ b/Network/QUIC/Types/Queue.hs @@ -1,7 +1,8 @@ module Network.QUIC.Types.Queue where import Control.Concurrent.STM -import Control.Monad (when) +import Control.Monad (void, when) +import Data.ByteString (ByteString) import Network.QUIC.Types.Packet @@ -23,18 +24,44 @@ recvQLimit :: Int recvQLimit = 4096 -data RecvQ = RecvQ (TQueue ReceivedPacket) (TVar Int) +-- | How many bytes the packets waiting in all of a server's 'RecvQ's may +-- take together. 'recvQLimit' bounds one connection, and a server has as +-- many connections as it is asked for, so this is what bounds the server +-- (#170). Some ten connections' worth of full queues. +recvBudgetLimit :: Int +recvBudgetLimit = 64 * 1024 * 1024 -newRecvQ :: IO RecvQ -newRecvQ = RecvQ <$> newTQueueIO <*> newTVarIO 0 +-- | Bytes in use, shared by the queues of an endpoint, and its limit. +data RecvBudget = RecvBudget Int (TVar Int) +newRecvBudget :: Int -> IO RecvBudget +newRecvBudget lim = RecvBudget lim <$> newTVarIO 0 + +-- | Packets, how many, their bytes, whether closed, and the budget. +data RecvQ + = RecvQ (TQueue ReceivedPacket) (TVar Int) (TVar Int) (TVar Bool) RecvBudget + +newRecvQ :: RecvBudget -> IO RecvQ +newRecvQ budget = + RecvQ + <$> newTQueueIO + <*> newTVarIO 0 + <*> newTVarIO 0 + <*> newTVarIO False + <*> pure budget + readRecvQ :: RecvQ -> IO ReceivedPacket -readRecvQ (RecvQ q qlen) = atomically $ do +readRecvQ (RecvQ q qlen qbytes _ (RecvBudget _ used)) = atomically $ do x <- readTQueue q + let n = rpReceivedBytes x modifyTVar' qlen (subtract 1) + modifyTVar' qbytes (subtract n) + modifyTVar' used (subtract n) return x --- | Queue a packet, or drop it if 'recvQLimit' are already waiting. +-- | Queue a packet, or drop it if 'recvQLimit' are already waiting, if the +-- budget it shares with the other queues has no room for it, or if the +-- queue has been closed. -- -- What is dropped is the packet that has just arrived rather than one -- already queued, so that what the connection has taken on is kept and only @@ -45,18 +72,72 @@ -- is one sending faster than the receiver can refuse what it sends, and a -- line of log for each of those packets is the same attack by another road. writeRecvQ :: RecvQ -> ReceivedPacket -> IO () -writeRecvQ (RecvQ q qlen) x = atomically $ do +writeRecvQ (RecvQ q qlen qbytes closed (RecvBudget lim used)) x = atomically $ do + c <- readTVar closed n <- readTVar qlen - when (n < recvQLimit) $ do + u <- readTVar used + let b = rpReceivedBytes x + when (not c && n < recvQLimit && u + b <= lim) $ do writeTQueue q x writeTVar qlen $! n + 1 + modifyTVar' qbytes (+ b) + writeTVar used $! u + b -- | Put back a packet this connection had already taken. -- --- Not subject to the limit: these come from the packets put off for want of --- the keys to read them, which the handshake bounds, and dropping one would --- lose a packet the connection has already accounted for. +-- Not subject to the limits: these come from the packets put off for want +-- of the keys to read them, which the handshake bounds, and dropping one +-- would lose a packet the connection has already accounted for. prependRecvQ :: RecvQ -> ReceivedPacket -> STM () -prependRecvQ (RecvQ q qlen) x = do +prependRecvQ (RecvQ q qlen qbytes _ (RecvBudget _ used)) x = do + let b = rpReceivedBytes x unGetTQueue q x modifyTVar' qlen (+ 1) + modifyTVar' qbytes (+ b) + modifyTVar' used (+ b) + +-- | Closing a queue nobody will read again: what it holds goes back to the +-- budget, and what is written to it from now on is dropped. A queue the +-- dispatcher can still find, for a connection that has ended, would +-- otherwise hold its share of the budget for good. +closeRecvQ :: RecvQ -> IO () +closeRecvQ (RecvQ q qlen qbytes closed (RecvBudget _ used)) = atomically $ do + writeTVar closed True + b <- readTVar qbytes + modifyTVar' used (subtract b) + writeTVar qbytes 0 + writeTVar qlen 0 + void $ flushTQueue q + +---------------------------------------------------------------- + +-- | How many DATAGRAM frames a connection may have waiting for the +-- application. Each fits in a packet, so this is a megabyte and a half +-- or so (#226). +datagramQLimit :: Int +datagramQLimit = 1024 + +data DatagramQ = DatagramQ (TQueue ByteString) (TVar Int) + +newDatagramQ :: IO DatagramQ +newDatagramQ = DatagramQ <$> newTQueueIO <*> newTVarIO 0 + +readDatagramQSTM :: DatagramQ -> STM ByteString +readDatagramQSTM (DatagramQ q qlen) = do + x <- readTQueue q + modifyTVar' qlen (subtract 1) + return x + +flushDatagramQSTM :: DatagramQ -> STM [ByteString] +flushDatagramQSTM (DatagramQ q qlen) = do + writeTVar qlen 0 + flushTQueue q + +-- | Queue a DATAGRAM, or drop it if 'datagramQLimit' are already waiting. +-- RFC 9221 Sec 5 lets a receiver drop DATAGRAM frames. +writeDatagramQ :: DatagramQ -> ByteString -> IO () +writeDatagramQ (DatagramQ q qlen) x = atomically $ do + n <- readTVar qlen + when (n < datagramQLimit) $ do + writeTQueue q x + writeTVar qlen $! n + 1 diff --git a/quic.cabal b/quic.cabal --- a/quic.cabal +++ b/quic.cabal @@ -1,6 +1,6 @@ cabal-version: 2.0 name: quic -version: 0.3.22 +version: 0.3.23 license: BSD3 license-file: LICENSE maintainer: kazu@iij.ad.jp diff --git a/test/DatagramSpec.hs b/test/DatagramSpec.hs --- a/test/DatagramSpec.hs +++ b/test/DatagramSpec.hs @@ -3,6 +3,7 @@ module DatagramSpec where import Control.Concurrent +import Control.Concurrent.STM (atomically) import qualified Control.Exception as E import Control.Monad import Data.ByteString () @@ -138,6 +139,16 @@ sendDatagram conn (BS.replicate 50 0x41) threadDelay 500 + -- #226: the queue grew for as long as the peer sent faster than the + -- application read. + describe "the DATAGRAM receive queue" $ + it "drops what arrives past its limit" $ do + q <- newDatagramQ + forM_ [1 .. datagramQLimit + 10] $ \_ -> writeDatagramQ q "x" + length <$> atomically (flushDatagramQSTM q) + `shouldReturn` datagramQLimit + writeDatagramQ q "y" + atomically (flushDatagramQSTM q) `shouldReturn` ["y"] describe "Datagram SHOULD requirements" $ do -- "This frame SHOULD be sent as soon as possible... and MAY be coalesced with other frames." it "Can send and receive DATAGRAM in 1-RTT" $ do diff --git a/test/HandshakeSpec.hs b/test/HandshakeSpec.hs --- a/test/HandshakeSpec.hs +++ b/test/HandshakeSpec.hs @@ -4,6 +4,7 @@ module HandshakeSpec where import Control.Concurrent +import Control.Concurrent.Async (wait, withAsync) import qualified Control.Exception as E import Control.Monad import qualified Data.ByteString as BS @@ -78,6 +79,21 @@ let cc = testClientConfig sc = sc0{scRequireRetry = True} testHandshake cc sc waitS FullHandshake + -- A token that proves nothing does not stand in for the Retry. + it "requires a Retry of a client with an invalid token" $ do + let cc = + testClientConfig + { ccResumption = + (ccResumption testClientConfig) + { resumptionToken = "invalid token" + , resumptionVersion = ccVersion testClientConfig + } + } + sc = sc0{scRequireRetry = True} + E.bracket (forkIO $ S.run sc $ \_ -> threadDelay 2000000) killThread $ \_ -> do + waitS + C.run cc (\conn -> waitEstablished conn >> retry <$> getConnectionInfo conn) + `shouldReturn` True it "can handshake in the case of resumption" $ do let cc = testClientConfig sc = sc0 @@ -152,6 +168,31 @@ te == cryptoError TLS.HandshakeFailure | otherwise = False testHandshake3 cc1 cc2 sc waitS handshakeFailure + -- RFC 9000 Sec 5.2.2: a connection over the limit is refused with + -- CONNECTION_REFUSED, and one ending makes room for another (#170). + it "refuses a connection over scMaxConnections" $ do + let sc = sc0{scMaxConnections = 1} + first <- newEmptyMVar + done <- newEmptyMVar + let holder = C.run testClientConfig $ \conn -> do + waitEstablished conn + putMVar first () + takeMVar done + refused (TransportErrorIsReceived ConnectionRefused _) = True + refused _ = False + establish = C.run testClientConfig waitEstablished + again 0 = establish + again n = + establish `E.catch` \(_ :: QUICException) -> + threadDelay 200000 >> again (n - 1 :: Int) + E.bracket (forkIO $ S.run sc $ \_ -> threadDelay 10000000) killThread $ \_ -> do + waitS + withAsync holder $ \h -> do + takeMVar first + establish `shouldThrow` refused + putMVar done () + wait h + again 10 it "reconnects with the version a Version Negotiation offers" $ do let cc = testClientConfig{ccVersion = Version2, ccVersions = [Version2, Version1]} sc = sc0{scVersions = [Version1]} diff --git a/test/IOSpec.hs b/test/IOSpec.hs --- a/test/IOSpec.hs +++ b/test/IOSpec.hs @@ -6,7 +6,7 @@ import Control.Concurrent import Control.Concurrent.Async -import Control.Concurrent.STM (readTVarIO) +import Control.Concurrent.STM (atomically, readTVarIO) import qualified Control.Exception as E import Control.Monad import Data.Bits (shiftR) @@ -15,6 +15,7 @@ import Data.IORef import qualified Data.IntMap.Strict as IntMap import GHC.Conc (getUncaughtExceptionHandler, setUncaughtExceptionHandler) +import Network.Control (txfSent) import qualified Network.Socket as NS import qualified Network.Socket.ByteString as NSB import System.Directory (listDirectory) @@ -79,6 +80,24 @@ -- The first one taken and lost: the second keeps its offset. _ <- takeSendStreamQ clientConn offset <$> takeSendStreamQ clientConn `shouldReturn` (6, 16) + -- RFC 9000 Sec 4.1: two streams sending at once used to both see the + -- same connection window and queue twice what MAX_DATA allowed. + describe "the connection window" $ + it "is not given to two streams sending at once" $ do + scw <- makeTestServerConfig + (clientConn, _) <- makeConnections scw Version1 + overs <- forM [0 .. 999] $ \i -> do + s1 <- addStream clientConn (8 * i) + s2 <- addStream clientConn (8 * i + 4) + setTxMaxStreamData s1 100 + setTxMaxStreamData s2 100 + base <- txfSent <$> atomically (readConnectionFlowTx clientConn) + setTxMaxData clientConn $ base + 100 + let send s = Timeout.timeout 1000 $ sendStream s $ BS.replicate 100 0 + _ <- concurrently (send s1) (send s2) + sent <- txfSent <$> atomically (readConnectionFlowTx clientConn) + return $ sent - base + filter (> 100) overs `shouldBe` [] describe "a stateless reset to the client" $ it "ends the connection with ConnectionIsReset" $ testStatelessResetReceived cc sc waitS @@ -173,11 +192,48 @@ withPipe (DelayServerPacket 300) $ testLateCopy False cc sc waitS it "ignores a late copy of the data it received on a stream the peer opened" $ do withPipe (DelayClientPacket 300) $ testLateCopy True cc sc waitS + -- RFC 9000 Sec 17.2.3: 0-RTT goes from client to server only. + describe "a frame received in 0-RTT" $ do + let replied serverConn = do + out <- atomically $ takeOutputSTM serverConn + case out of + OutControl lvl frames -> return (lvl, frames) + _ -> fail "no control frames" + it "has its ACK sent in 1-RTT" $ do + scz <- makeTestServerConfig + (_, serverConn) <- makeConnections scz Version1 + processFrame serverConn RTT0Level Ping + fst <$> replied serverConn `shouldReturn` RTT1Level + it "has its RESET_STREAM sent in 1-RTT" $ do + scz <- makeTestServerConfig + (_, serverConn) <- makeConnections scz Version1 + _ <- addStream serverConn 0 + processFrame serverConn RTT0Level $ StopSending 0 $ ApplicationProtocolError 0 + (lvl, frames) <- replied serverConn + lvl `shouldBe` RTT1Level + frames `shouldSatisfy` any isResetStream describe "stream limits" $ do + -- #86: the number the peer may have open is no longer fixed. + it "announces a raised limit at once and holds a lowered one" $ do + scl <- makeTestServerConfig + (_, serverConn) <- makeConnections scl Version1 + let initial = initialMaxStreamsBidi $ getMyParameters serverConn + setMaxBidiStreams serverConn $ initial + 10 + out <- atomically $ takeOutputSTM serverConn + case out of + OutControl RTT1Level [MaxStreams Bidirectional n] -> n `shouldBe` initial + 10 + _ -> expectationFailure "no MAX_STREAMS" + -- MAX_STREAMS cannot go down, and closing one stream of many + -- under a lower target gives the peer no new one. + setMaxBidiStreams serverConn 1 + atomically (isEmptyOutputSTM serverConn) `shouldReturn` True + checkStreamIdRoom serverConn Bidirectional `shouldReturn` Nothing it "keeps the peer's open bidirectional streams within the limit" $ do withPipe (DropClientPacket []) $ testOpenStreams cc sc waitS it "raises the limit on unidirectional streams by their own number" $ do withPipe (DropClientPacket []) $ testUniStreams cc sc waitS + it "ignores STOP_SENDING for a unidirectional stream it has closed" $ do + withPipe (DropClientPacket []) $ testStopClosedUni cc sc waitS describe "a stream closed before the peer finishes it" $ do it "is forgotten once the peer answers STOP_SENDING" $ do withPipe (DropClientPacket []) $ testDepartedForgotten cc sc waitS @@ -659,6 +715,48 @@ forever $ do strm <- acceptStream conn recvStream strm 1 + +isResetStream :: Frame -> Bool +isResetStream ResetStream{} = True +isResetStream _ = False + +-- | A unidirectional stream used to be checked against the counter of +-- bidirectional ones, so a client that had opened none of those closed the +-- connection with STREAM_STATE_ERROR on a STOP_SENDING for stream 2. +testStopClosedUni :: C.ClientConfig -> ServerConfig -> IO () -> IO () +testStopClosedUni cc sc waitS = do + closed <- newEmptyMVar + stopped <- newEmptyMVar + withAsync (server closed stopped) $ \_ -> client closed stopped + where + client closed stopped = do + waitS + C.run cc $ \conn -> do + uni <- unidirectionalStream conn + sendStream uni "x" + closeStream uni + putMVar closed () + () <- takeMVar stopped + threadDelay 200000 + strm <- stream conn + sendStream strm "ping" + shutdownStream strm + recvStream strm 4 `shouldReturn` "pong" + server closed stopped = run sc $ \conn -> do + uni <- acceptStream conn + recvStream uni 1 `shouldReturn` "x" + () <- takeMVar closed + -- Sent as is: 'stopStream' sends nothing for a stream whose FIN + -- has arrived, and a peer's may cross it. + sendFrames + conn + RTT1Level + [StopSending (streamId uni) $ ApplicationProtocolError 0] + putMVar stopped () + strm <- acceptStream conn + _ <- recvStream strm 4 + sendStream strm "pong" + shutdownStream strm testRecvStreamClientStopFirst :: C.ClientConfig -> ServerConfig -> IO () -> IO () diff --git a/test/MigrationSpec.hs b/test/MigrationSpec.hs --- a/test/MigrationSpec.hs +++ b/test/MigrationSpec.hs @@ -5,7 +5,7 @@ import Control.Concurrent import Control.Concurrent.Async (concurrently, mapConcurrently, wait, withAsync) import Control.Concurrent.STM -import Control.Monad (forM, void) +import Control.Monad (forM, forM_, void) import qualified Data.ByteString as BS import qualified Data.ByteString.Short as Short import Network.Control (txfLimit) @@ -51,6 +51,27 @@ -- CID of its own on the new path, while the old path keeps the old -- one; the old one was given up at once, and the old path's challenge -- went out with the new. + -- #223: a NEW_CONNECTION_ID retiring every CID it does not bring + -- left none to use, and the connection ended with INTERNAL_ERROR. + describe "addPeerCIDAndRetire" $ do + let cid :: Int -> CID + cid i = toCID $ BS.replicate 8 $ fromIntegral i + srt i = StatelessResetToken $ Short.toShort $ BS.replicate 16 $ fromIntegral i + info i = newCIDInfo i (cid i) (srt i) + it "moves to the new CID when all the others are retired" $ do + (_, serverConn) <- makeConnections serverConf Version1 + addPeerCIDAndRetire serverConn (info 1) 1 `shouldReturn` Just [0] + getPeerCID serverConn `shouldReturn` cid 1 + it "lets a CID exceed the limit when the excess is retired" $ do + (_, serverConn) <- makeConnections serverConf Version1 + let lim = activeConnectionIdLimit $ getMyParameters serverConn + forM_ [1 .. lim - 1] $ \i -> + addPeerCIDAndRetire serverConn (info i) 0 `shouldReturn` Just [] + addPeerCIDAndRetire serverConn (info lim) 0 `shouldReturn` Nothing + addPeerCIDAndRetire serverConn (info lim) lim + `shouldReturn` Just [0 .. lim - 1] + getPeerCID serverConn `shouldReturn` cid lim + describe "validateMigration" $ do let next conn = do out <- T.timeout 4000000 $ atomically $ takeOutputSTM conn diff --git a/test/PacketSpec.hs b/test/PacketSpec.hs --- a/test/PacketSpec.hs +++ b/test/PacketSpec.hs @@ -180,7 +180,7 @@ -- dummy let clientConf = testClientConfig (sock, peersa) <- clientSocket "127.0.0.1" "2000" - q <- newRecvQ + q <- newRecvQ =<< newRecvBudget maxBound retire <- newTVarIO False sref <- newIORef (sock, retire) pathInfo <- newPathInfo peersa @@ -188,7 +188,7 @@ let ver = v verInfo = VersionInfo ver [ver] genSRT <- makeGenStatelessReset - dq <- newTQueueIO + dq <- newDatagramQ ---- clientConn <- clientConnection diff --git a/test/RecoverySpec.hs b/test/RecoverySpec.hs --- a/test/RecoverySpec.hs +++ b/test/RecoverySpec.hs @@ -4,6 +4,7 @@ import Control.Concurrent.STM import Data.ByteString.Short () +import Data.Maybe (isJust) import qualified Data.Sequence as Seq import Data.UnixTime import Test.Hspec @@ -51,6 +52,28 @@ atomically $ writeTVar (encryptionLevel cs) RTT1Level onAckReceived ldcc HandshakeLevel (AckInfo 0 0 []) 0 getPtoTimeAndSpace ldcc `shouldReturn` Nothing + -- RFC 9002 B.8: only packets in flight make a congestion event. + describe "a lost packet" $ do + let lostWith frames = do + cs <- newConnState Client + ldcc <- newLDCC cs (\_ -> return ()) (\_ -> return ()) + now <- getTimeMicrosecond + let lost = + (sp 1 (any ackEliciting frames) now) + { spEncryptionLevel = RTT1Level + , spSentBytes = 1200 + , spPlainPacket = PlainPacket (Short $ toCID "") (Plain (Flags 0) 1 frames 0) + } + onPacketsLost ldcc $ Seq.singleton lost + retransmit ldcc $ Seq.singleton lost + -- A congestion event starts a recovery period. + event <- isJust . congestionRecoveryStartTime <$> readTVarIO (recoveryCC ldcc) + ping <- readTVarIO $ ptoPing ldcc + return (event, ping) + it "is no congestion event and needs no PING if it carried only an ACK" $ + lostWith [Ack (AckInfo 0 0 []) noDelay] `shouldReturn` (False, Nothing) + it "is a congestion event if it was in flight" $ + fst <$> lostWith [Ping] `shouldReturn` True describe "persistent congestion" $ do it "does not find a pair" $ do findDuration (Seq.fromList [sp0]) 0 `shouldBe` Nothing diff --git a/test/TypesSpec.hs b/test/TypesSpec.hs --- a/test/TypesSpec.hs +++ b/test/TypesSpec.hs @@ -43,7 +43,7 @@ -- anyone who can see a connection ID can fill it. See #167. describe "RecvQ" $ do it "keeps what it has taken and drops what is past the limit" $ do - q <- newRecvQ + q <- newRecvQ =<< newRecvBudget maxBound mapM_ (writeRecvQ q . numbered) [1 .. recvQLimit + 100] ns <- replicateM recvQLimit (number <$> readRecvQ q) -- The packets kept are the ones that arrived first: what is @@ -52,7 +52,7 @@ ns `shouldBe` [1 .. recvQLimit] Timeout.timeout 100000 (readRecvQ q) `shouldReturn` Nothing it "takes back a packet it had already taken, full or not" $ do - q <- newRecvQ + q <- newRecvQ =<< newRecvBudget maxBound mapM_ (writeRecvQ q . numbered) [1 .. recvQLimit] -- Putting back what was put off for want of keys is not the -- same as taking something new on, and the limit does not apply @@ -60,6 +60,29 @@ -- for. atomically $ prependRecvQ q $ numbered 0 number <$> readRecvQ q `shouldReturn` 0 + -- #170: what one connection may queue times the number of + -- connections was a limit on nothing. + it "shares a budget with the other queues of the endpoint" $ do + budget <- newRecvBudget 3600 -- three packets of 1200 + q1 <- newRecvQ budget + q2 <- newRecvQ budget + mapM_ (writeRecvQ q1 . numbered) [1, 2] + mapM_ (writeRecvQ q2 . numbered) [3, 4] + number <$> readRecvQ q1 `shouldReturn` 1 + writeRecvQ q2 $ numbered 5 + number <$> readRecvQ q2 `shouldReturn` 3 + number <$> readRecvQ q2 `shouldReturn` 5 + it "gives back what a closed queue holds and takes no more" $ do + budget <- newRecvBudget 2400 + q1 <- newRecvQ budget + q2 <- newRecvQ budget + mapM_ (writeRecvQ q1 . numbered) [1, 2] + closeRecvQ q1 + writeRecvQ q1 $ numbered 3 + mapM_ (writeRecvQ q2 . numbered) [4, 5] + number <$> readRecvQ q2 `shouldReturn` 4 + number <$> readRecvQ q2 `shouldReturn` 5 + Timeout.timeout 100000 (readRecvQ q1) `shouldReturn` Nothing -- RFC 9000 Sec 19.3: the field counts units of 2 ^ ack_delay_exponent -- microseconds, and it is a varint, so up to 2 ^ 62 - 1 of them. describe "ackDelayMicro" $ do