diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index b98deeeefa..174a74e111 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/pkgs/http2/CHANGELOG.md @@ -1,6 +1,14 @@ ## 3.0.1-wip - Gracefully handle receiving headers on a stream that the client has canceled. (#1799) +- Bound peer-initiated concurrent streams and reject peers that ignore the + advertised `SETTINGS_MAX_CONCURRENT_STREAMS` value. Reserved peer streams + are also bounded locally, and disabled server push is enforced on receipt. + The default limit is 100. +- Bound inbound compressed field blocks, decoded field sections, CONTINUATION + frame counts, and incomplete field-block duration. HPACK decoding continues + in discard mode after the decoded limit is exceeded so the connection-wide + compression context remains synchronized. ## 3.0.0 diff --git a/pkgs/http2/lib/src/connection.dart b/pkgs/http2/lib/src/connection.dart index 6a9f3fc9de..46a2569a0b 100644 --- a/pkgs/http2/lib/src/connection.dart +++ b/pkgs/http2/lib/src/connection.dart @@ -118,7 +118,11 @@ abstract class Connection { final Window _localWindow = Window(); /// Used for defragmenting PushPromise/Header frames. - final FrameDefragmenter _defragmenter = FrameDefragmenter(); + late final FrameDefragmenter _defragmenter; + + late final int? _maxInboundHeaderListSize; + late final Duration? _inboundHeaderBlockTimeout; + Timer? _inboundHeaderBlockTimer; /// The outgoing frames of this connection; late FrameWriter _frameWriter; @@ -164,9 +168,22 @@ abstract class Connection { StreamSink> outgoing, Settings settingsObject, ) { + _validateSettings(settingsObject); + _maxInboundHeaderListSize = settingsObject.maxInboundHeaderListSize; + _inboundHeaderBlockTimeout = settingsObject.inboundHeaderBlockTimeout; + _defragmenter = FrameDefragmenter( + maxHeaderBlockSize: settingsObject.maxInboundHeaderBlockSize, + maxContinuationFrames: settingsObject.maxContinuationFramesPerBlock, + ); + // Setup frame reading. var incomingFrames = - FrameReader(incoming, acknowledgedSettings).startDecoding(); + FrameReader( + incoming, + acknowledgedSettings, + onFieldBlockStart: _startInboundHeaderBlockTimer, + onFieldBlockEnd: _cancelInboundHeaderBlockTimer, + ).startDecoding(); _frameReaderSubscription = incomingFrames.listen( (Frame frame) { _catchProtocolErrors(() => _handleFrameImpl(frame)); @@ -245,6 +262,10 @@ abstract class Connection { _settingsHandler.peerSettings, _settingsHandler.acknowledgedSettings, _activeStateHandler, + settingsObject.concurrentStreamLimit, + settingsObject.maxPeerStreamLimitViolations, + _terminateForPeerStreamLimitAbuse, + settingsObject is ClientSettings && settingsObject.allowServerPushes, ); } else { _streams = StreamHandler.server( @@ -254,6 +275,10 @@ abstract class Connection { _settingsHandler.peerSettings, _settingsHandler.acknowledgedSettings, _activeStateHandler, + settingsObject.concurrentStreamLimit, + settingsObject.maxPeerStreamLimitViolations, + _terminateForPeerStreamLimitAbuse, + false, ); } @@ -282,6 +307,16 @@ abstract class Connection { ); } + final maxInboundHeaderListSize = settings.maxInboundHeaderListSize; + if (maxInboundHeaderListSize != null) { + settingsList.add( + Setting( + Setting.SETTINGS_MAX_HEADER_LIST_SIZE, + maxInboundHeaderListSize, + ), + ); + } + if (settings is ClientSettings) { // By default the server is allowed to do server pushes. if (!settings.allowServerPushes) { @@ -336,8 +371,10 @@ abstract class Connection { _terminate(ErrorCode.FLOW_CONTROL_ERROR, message: '$error'); } on FrameSizeException catch (error) { _terminate(ErrorCode.FRAME_SIZE_ERROR, message: '$error'); + } on HeaderBlockProcessingException catch (error) { + _terminate(ErrorCode.ENHANCE_YOUR_CALM, message: '$error'); } on HPackDecodingException catch (error) { - _terminate(ErrorCode.PROTOCOL_ERROR, message: '$error'); + _terminate(ErrorCode.COMPRESSION_ERROR, message: '$error'); } on TerminatedException { // We tried to perform an action even though the connection was already // terminated. @@ -371,13 +408,21 @@ abstract class Connection { // [This needs to be done even if the frames get ignored, since the entire // connection shares one HPack compression context.] if (frame is HeadersFrame) { - frame.decodedHeaders = _hpackContext.decoder.decode( + final result = _hpackContext.decoder.decodeWithLimit( frame.headerBlockFragment, + maxHeaderListSize: _maxInboundHeaderListSize, ); + frame.decodedHeaders = result.headers; + frame.decodedHeaderListSize = result.headerListSize; + frame.headerListSizeExceeded = result.headerListSizeExceeded; } else if (frame is PushPromiseFrame) { - frame.decodedHeaders = _hpackContext.decoder.decode( + final result = _hpackContext.decoder.decodeWithLimit( frame.headerBlockFragment, + maxHeaderListSize: _maxInboundHeaderListSize, ); + frame.decodedHeaders = result.headers; + frame.decodedHeaderListSize = result.headerListSize; + frame.headerListSizeExceeded = result.headerListSizeExceeded; } if (_frameReceived.hasListener) { _frameReceived.add(null); @@ -406,6 +451,61 @@ abstract class Connection { } } + void _startInboundHeaderBlockTimer() { + final timeout = _inboundHeaderBlockTimeout; + if (timeout == null || _inboundHeaderBlockTimer != null) return; + _inboundHeaderBlockTimer = Timer(timeout, () { + _terminate( + ErrorCode.ENHANCE_YOUR_CALM, + message: 'Inbound field block was not completed within $timeout.', + ); + }); + } + + void _cancelInboundHeaderBlockTimer() { + _inboundHeaderBlockTimer?.cancel(); + _inboundHeaderBlockTimer = null; + } + + void _terminateForPeerStreamLimitAbuse(String message) { + _terminate(ErrorCode.ENHANCE_YOUR_CALM, message: message); + } + + static void _validateSettings(Settings settings) { + void nonNegative(int? value, String name) { + if (value != null && value < 0) { + throw ArgumentError.value(value, name, 'must be >= 0'); + } + } + + nonNegative(settings.concurrentStreamLimit, 'concurrentStreamLimit'); + nonNegative( + settings.maxInboundHeaderBlockSize, + 'maxInboundHeaderBlockSize', + ); + nonNegative(settings.maxInboundHeaderListSize, 'maxInboundHeaderListSize'); + nonNegative( + settings.maxContinuationFramesPerBlock, + 'maxContinuationFramesPerBlock', + ); + final timeout = settings.inboundHeaderBlockTimeout; + if (timeout != null && timeout.isNegative) { + throw ArgumentError.value( + timeout, + 'inboundHeaderBlockTimeout', + 'must not be negative', + ); + } + final violations = settings.maxPeerStreamLimitViolations; + if (violations != null && violations < 1) { + throw ArgumentError.value( + violations, + 'maxPeerStreamLimitViolations', + 'must be >= 1', + ); + } + } + void _finishing({bool active = true, String? message}) { // If this connection is already dead, we return. if (_state.isTerminated) return; @@ -454,6 +554,7 @@ abstract class Connection { // TODO: When do we complete here? if (_state.state != ConnectionState.Terminated) { _state.state = ConnectionState.Terminated; + _cancelInboundHeaderBlockTimer(); var cancelFuture = Future.sync(_frameReaderSubscription.cancel); if (!causedByTransportError) { diff --git a/pkgs/http2/lib/src/frames/frame_defragmenter.dart b/pkgs/http2/lib/src/frames/frame_defragmenter.dart index f17fee04ae..7ea7759659 100644 --- a/pkgs/http2/lib/src/frames/frame_defragmenter.dart +++ b/pkgs/http2/lib/src/frames/frame_defragmenter.dart @@ -2,94 +2,161 @@ // for details. All rights reserved. Use of this source code is governed by a // BSD-style license that can be found in the LICENSE file. -import '../sync_errors.dart'; +import 'dart:typed_data'; +import '../sync_errors.dart'; import 'frames.dart'; -/// Class used for defragmenting [HeadersFrame]s and [PushPromiseFrame]s. -// TODO: Somehow emit an error if too many continuation frames have been sent -// (since we're buffering all of them). +/// Defragments field blocks from [HeadersFrame]s and [PushPromiseFrame]s with +/// bounded buffering. class FrameDefragmenter { - /// The current incomplete [HeadersFrame] fragment. - HeadersFrame? _headersFrame; + final int? maxHeaderBlockSize; + final int? maxContinuationFrames; - /// The current incomplete [PushPromiseFrame] fragment. - PushPromiseFrame? _pushPromiseFrame; + Frame? _initialFrame; + BytesBuilder? _fragments; + int _fieldBlockSize = 0; + int _combinedPayloadLength = 0; + int _continuationFrames = 0; + + FrameDefragmenter({this.maxHeaderBlockSize, this.maxContinuationFrames}) { + if (maxHeaderBlockSize case final limit? when limit < 0) { + throw ArgumentError.value(limit, 'maxHeaderBlockSize', 'must be >= 0'); + } + if (maxContinuationFrames case final limit? when limit < 0) { + throw ArgumentError.value(limit, 'maxContinuationFrames', 'must be >= 0'); + } + } + + /// Whether an incomplete field block is currently buffered. + bool get isDefragmenting => _initialFrame != null; /// Tries to defragment [frame]. /// - /// If the given [frame] is a [HeadersFrame] or a [PushPromiseFrame] which - /// needs de-fragmentation, it will be saved and `null` will be returned. - /// - /// If there is currently an incomplete [HeadersFrame] or [PushPromiseFrame] - /// saved, [frame] needs to be a [ContinuationFrame]. It will be added to the - /// saved frame. In case the defragmentation is complete, the defragmented - /// [HeadersFrame] or [PushPromiseFrame] will be returned. - /// - /// All other [Frame] types will be returned. - // TODO: Consider handling continuation frames without preceding - // headers/push-promise frame here instead of the call site? + /// Incomplete field blocks return `null`. A completed field block is returned + /// as one [HeadersFrame] or [PushPromiseFrame]. Fragment bytes are retained + /// as chunks and combined only once, avoiding repeated whole-block copies. Frame? tryDefragmentFrame(Frame? frame) { - if (_headersFrame != null) { - if (frame is ContinuationFrame) { - if (_headersFrame!.header.streamId != frame.header.streamId) { - throw ProtocolException( - 'Defragmentation: frames have different stream ids.', - ); - } - _headersFrame = _headersFrame!.addBlockContinuation(frame); - - if (frame.hasEndHeadersFlag) { - var frame = _headersFrame; - _headersFrame = null; - return frame; - } else { - return null; - } - } else { - throw ProtocolException( + final initialFrame = _initialFrame; + if (initialFrame != null) { + if (frame is! ContinuationFrame) { + _failProtocol( 'Defragmentation: Incomplete frame must be followed by ' 'continuation frame.', ); } - } else if (_pushPromiseFrame != null) { - if (frame is ContinuationFrame) { - if (_pushPromiseFrame!.header.streamId != frame.header.streamId) { - throw ProtocolException( - 'Defragmentation: frames have different stream ids.', - ); - } - _pushPromiseFrame = _pushPromiseFrame!.addBlockContinuation(frame); - - if (frame.hasEndHeadersFlag) { - var frame = _pushPromiseFrame; - _pushPromiseFrame = null; - return frame; - } else { - return null; - } - } else { - throw ProtocolException( - 'Defragmentation: Incomplete frame must be followed by ' - 'continuation frame.', + if (initialFrame.header.streamId != frame.header.streamId) { + _failProtocol('Defragmentation: frames have different stream ids.'); + } + + _continuationFrames++; + final continuationLimit = maxContinuationFrames; + if (continuationLimit != null && + _continuationFrames > continuationLimit) { + _failResourceLimit( + 'Inbound field block used $_continuationFrames CONTINUATION frames, ' + 'exceeding the limit of $continuationLimit.', ); } - } else { - if (frame is HeadersFrame) { - if (!frame.hasEndHeadersFlag) { - _headersFrame = frame; - return null; - } - } else if (frame is PushPromiseFrame) { - if (!frame.hasEndHeadersFlag) { - _pushPromiseFrame = frame; - return null; - } + + _append(frame.headerBlockFragment); + _combinedPayloadLength += frame.header.length; + if (!frame.hasEndHeadersFlag) return null; + + final bytes = _fragments!.takeBytes(); + final combined = _complete(initialFrame, frame, bytes); + _reset(); + return combined; + } + + if (frame is HeadersFrame || frame is PushPromiseFrame) { + final fragment = _headerBlockFragment(frame!); + _checkHeaderBlockSize(fragment.length); + if (!_hasEndHeadersFlag(frame)) { + _initialFrame = frame; + _fragments = BytesBuilder(copy: false)..add(fragment); + _fieldBlockSize = fragment.length; + _combinedPayloadLength = frame.header.length; + return null; } } - // If this frame is not relevant for header defragmentation, we pass it to - // the next stage. return frame; } + + void _append(List fragment) { + _checkHeaderBlockSize(_fieldBlockSize + fragment.length); + _fragments!.add(fragment); + _fieldBlockSize += fragment.length; + } + + void _checkHeaderBlockSize(int size) { + final limit = maxHeaderBlockSize; + if (limit != null && size > limit) { + _failResourceLimit( + 'Inbound compressed field block exceeds the limit ' + '($size bytes > $limit bytes).', + ); + } + } + + Frame _complete( + Frame initialFrame, + ContinuationFrame finalFrame, + Uint8List bytes, + ) { + final header = FrameHeader( + _combinedPayloadLength, + initialFrame.header.type, + initialFrame.header.flags | finalFrame.header.flags, + initialFrame.header.streamId, + ); + if (initialFrame is HeadersFrame) { + return HeadersFrame( + header, + initialFrame.padLength, + initialFrame.exclusiveDependency, + initialFrame.streamDependency, + initialFrame.weight, + bytes, + ); + } + final pushPromise = initialFrame as PushPromiseFrame; + return PushPromiseFrame( + header, + pushPromise.padLength, + pushPromise.promisedStreamId, + bytes, + ); + } + + static List _headerBlockFragment(Frame frame) => switch (frame) { + HeadersFrame frame => frame.headerBlockFragment, + PushPromiseFrame frame => frame.headerBlockFragment, + _ => throw StateError('Expected HEADERS or PUSH_PROMISE frame.'), + }; + + static bool _hasEndHeadersFlag(Frame frame) => switch (frame) { + HeadersFrame frame => frame.hasEndHeadersFlag, + PushPromiseFrame frame => frame.hasEndHeadersFlag, + _ => throw StateError('Expected HEADERS or PUSH_PROMISE frame.'), + }; + + Never _failProtocol(String message) { + _reset(); + throw ProtocolException(message); + } + + Never _failResourceLimit(String message) { + _reset(); + throw HeaderBlockProcessingException(message); + } + + void _reset() { + _initialFrame = null; + _fragments = null; + _fieldBlockSize = 0; + _combinedPayloadLength = 0; + _continuationFrames = 0; + } } diff --git a/pkgs/http2/lib/src/frames/frame_reader.dart b/pkgs/http2/lib/src/frames/frame_reader.dart index 336e38f05b..393e1a6f59 100644 --- a/pkgs/http2/lib/src/frames/frame_reader.dart +++ b/pkgs/http2/lib/src/frames/frame_reader.dart @@ -12,9 +12,22 @@ class FrameReader { /// complying with. final ActiveSettings _localSettings; + /// Called after an inbound HEADERS or PUSH_PROMISE frame header is parsed, + /// before waiting for its payload bytes. + final void Function()? _onFieldBlockStart; + + /// Called after the final frame of an inbound field block is fully read. + final void Function()? _onFieldBlockEnd; + final _framesController = StreamController(); - FrameReader(this._inputStream, this._localSettings); + FrameReader( + this._inputStream, + this._localSettings, { + void Function()? onFieldBlockStart, + void Function()? onFieldBlockEnd, + }) : _onFieldBlockStart = onFieldBlockStart, + _onFieldBlockEnd = onFieldBlockEnd; /// Starts to listen on the input stream and decodes HTTP/2 transport frames. Stream startDecoding() { @@ -27,7 +40,14 @@ class FrameReader { _mergeLists(bufferedData, FRAME_HEADER_SIZE); // Read the frame header from the first byte array. - return _readFrameHeader(bufferedData[0], 0); + final header = _readFrameHeader(bufferedData[0], 0); + final isFieldBlockStart = + header.type == FrameType.HEADERS || + header.type == FrameType.PUSH_PROMISE; + if (isFieldBlockStart) { + _onFieldBlockStart?.call(); + } + return header; } return null; } @@ -40,6 +60,15 @@ class FrameReader { // Read the frame. var frame = _readFrame(header, bufferedData[0], FRAME_HEADER_SIZE); + final completesFieldBlock = switch (frame) { + HeadersFrame frame => frame.hasEndHeadersFlag, + PushPromiseFrame frame => frame.hasEndHeadersFlag, + ContinuationFrame frame => frame.hasEndHeadersFlag, + _ => false, + }; + if (completesFieldBlock) { + _onFieldBlockEnd?.call(); + } // Update bufferedData/bufferedLength var firstChunkLen = bufferedData[0].length; diff --git a/pkgs/http2/lib/src/frames/frame_types.dart b/pkgs/http2/lib/src/frames/frame_types.dart index 9d8d3c4d55..84a811824f 100644 --- a/pkgs/http2/lib/src/frames/frame_types.dart +++ b/pkgs/http2/lib/src/frames/frame_types.dart @@ -115,47 +115,17 @@ class HeadersFrame extends Frame { /// This will be set from the outside after decoding. late List
decodedHeaders; + /// Decoded field-section size using name + value + 32 bytes per field. + int decodedHeaderListSize = 0; + + /// Whether the decoded field section exceeded the local retention limit. + bool headerListSizeExceeded = false; + bool get hasEndStreamFlag => _isFlagSet(header.flags, FLAG_END_STREAM); bool get hasEndHeadersFlag => _isFlagSet(header.flags, FLAG_END_HEADERS); bool get hasPaddedFlag => _isFlagSet(header.flags, FLAG_PADDED); bool get hasPriorityFlag => _isFlagSet(header.flags, FLAG_PRIORITY); - HeadersFrame addBlockContinuation(ContinuationFrame frame) { - var fragment = frame.headerBlockFragment; - var flags = header.flags | frame.header.flags; - var fh = FrameHeader( - header.length + fragment.length, - header.type, - flags, - header.streamId, - ); - - var mergedHeaderBlockFragment = Uint8List( - headerBlockFragment.length + fragment.length, - ); - - mergedHeaderBlockFragment.setRange( - 0, - headerBlockFragment.length, - headerBlockFragment, - ); - - mergedHeaderBlockFragment.setRange( - headerBlockFragment.length, - mergedHeaderBlockFragment.length, - fragment, - ); - - return HeadersFrame( - fh, - padLength, - exclusiveDependency, - streamDependency, - weight, - mergedHeaderBlockFragment, - ); - } - @override Map toJson() => super.toJson()..addAll({ @@ -250,6 +220,12 @@ class PushPromiseFrame extends Frame { /// This will be set from the outside after decoding. late List
decodedHeaders; + /// Decoded field-section size using name + value + 32 bytes per field. + int decodedHeaderListSize = 0; + + /// Whether the decoded field section exceeded the local retention limit. + bool headerListSizeExceeded = false; + PushPromiseFrame( super.header, this.padLength, @@ -260,40 +236,6 @@ class PushPromiseFrame extends Frame { bool get hasEndHeadersFlag => _isFlagSet(header.flags, FLAG_END_HEADERS); bool get hasPaddedFlag => _isFlagSet(header.flags, FLAG_PADDED); - PushPromiseFrame addBlockContinuation(ContinuationFrame frame) { - var fragment = frame.headerBlockFragment; - var flags = header.flags | frame.header.flags; - var fh = FrameHeader( - header.length + fragment.length, - header.type, - flags, - header.streamId, - ); - - var mergedHeaderBlockFragment = Uint8List( - headerBlockFragment.length + fragment.length, - ); - - mergedHeaderBlockFragment.setRange( - 0, - headerBlockFragment.length, - headerBlockFragment, - ); - - mergedHeaderBlockFragment.setRange( - headerBlockFragment.length, - mergedHeaderBlockFragment.length, - fragment, - ); - - return PushPromiseFrame( - fh, - padLength, - promisedStreamId, - mergedHeaderBlockFragment, - ); - } - @override Map toJson() => super.toJson()..addAll({ diff --git a/pkgs/http2/lib/src/hpack/hpack.dart b/pkgs/http2/lib/src/hpack/hpack.dart index 6ce7bf0513..88eee9f42e 100644 --- a/pkgs/http2/lib/src/hpack/hpack.dart +++ b/pkgs/http2/lib/src/hpack/hpack.dart @@ -56,6 +56,20 @@ class Header { } } +/// Result of decoding one HPACK field block with an optional field-section +/// size limit. +class HPackDecodingResult { + final List
headers; + final int headerListSize; + final bool headerListSizeExceeded; + + const HPackDecodingResult( + this.headers, + this.headerListSize, + this.headerListSizeExceeded, + ); +} + /// A stateful HPACK decoder. class HPackDecoder { late int _maxHeaderTableSize; @@ -66,7 +80,21 @@ class HPackDecoder { _maxHeaderTableSize = newMaximumSize; } - List
decode(List data) { + List
decode(List data) => decodeWithLimit(data).headers; + + /// Decodes [data] while bounding retained application-visible headers. + /// + /// Once [maxHeaderListSize] is exceeded, already retained headers are + /// released and subsequent fields are parsed without being added to the + /// result. Parsing continues so incremental-indexing updates are applied to + /// the connection-wide dynamic table and later field blocks remain in sync. + HPackDecodingResult decodeWithLimit( + List data, { + int? maxHeaderListSize, + }) { + if (maxHeaderListSize case final limit? when limit < 0) { + throw ArgumentError.value(limit, 'maxHeaderListSize', 'must be >= 0'); + } var offset = 0; int readInteger(int prefixBits) { @@ -121,6 +149,20 @@ class HPackDecoder { try { var headers =
[]; + var headerListSize = 0; + var headerListSizeExceeded = false; + + void processHeader(Header header) { + headerListSize += header.name.length + header.value.length + 32; + if (headerListSizeExceeded) return; + if (maxHeaderListSize != null && headerListSize > maxHeaderListSize) { + headers.clear(); + headerListSizeExceeded = true; + } else { + headers.add(header); + } + } + while (offset < data.length) { var byte = data[offset]; var isIndexedField = (byte & 0x80) != 0; @@ -133,15 +175,15 @@ class HPackDecoder { if (isIndexedField) { var index = readInteger(7); var field = _table.lookup(index); - headers.add(field); + processHeader(field); } else if (isIncrementalIndexing) { var field = readHeaderFieldInternal(readInteger(6)); _table.addHeaderField(field); - headers.add(field); + processHeader(field); } else if (isWithoutIndexing) { - headers.add(readHeaderFieldInternal(readInteger(4))); + processHeader(readHeaderFieldInternal(readInteger(4))); } else if (isNeverIndexing) { - headers.add( + processHeader( readHeaderFieldInternal(readInteger(4), neverIndexed: true), ); } else if (isDynamicTableSizeUpdate) { @@ -159,7 +201,11 @@ class HPackDecoder { throw HPackDecodingException('Invalid encoding of headers.'); } } - return headers; + return HPackDecodingResult( + headers, + headerListSize, + headerListSizeExceeded, + ); // ignore: avoid_catching_errors } on RangeError catch (e) { throw HPackDecodingException('$e'); diff --git a/pkgs/http2/lib/src/streams/stream_handler.dart b/pkgs/http2/lib/src/streams/stream_handler.dart index 7db1310658..04d91326e2 100644 --- a/pkgs/http2/lib/src/streams/stream_handler.dart +++ b/pkgs/http2/lib/src/streams/stream_handler.dart @@ -141,6 +141,12 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { final ActiveSettings _peerSettings; final ActiveSettings _localSettings; + final int? _peerInitiatedStreamLimit; + final int? _maxPeerStreamLimitViolations; + final void Function(String)? _onPeerStreamLimitAbuse; + final bool _allowPeerPushes; + int _consecutivePeerStreamLimitViolations = 0; + final Map _openStreams = {}; int nextStreamId; int lastRemoteStreamId; @@ -170,6 +176,10 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { this._onActiveStateChanged, this.nextStreamId, this.lastRemoteStreamId, + this._peerInitiatedStreamLimit, + this._maxPeerStreamLimitViolations, + this._onPeerStreamLimitAbuse, + this._allowPeerPushes, ); factory StreamHandler.client( @@ -178,8 +188,12 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { ConnectionMessageQueueOut outgoingQueue, ActiveSettings peerSettings, ActiveSettings localSettings, - ActiveStateHandler onActiveStateChanged, - ) { + ActiveStateHandler onActiveStateChanged, [ + int? peerInitiatedStreamLimit, + int? maxPeerStreamLimitViolations, + void Function(String)? onPeerStreamLimitAbuse, + bool allowPeerPushes = false, + ]) { return StreamHandler._( writer, incomingQueue, @@ -189,6 +203,10 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { onActiveStateChanged, 1, 0, + peerInitiatedStreamLimit, + maxPeerStreamLimitViolations, + onPeerStreamLimitAbuse, + allowPeerPushes, ); } @@ -198,8 +216,12 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { ConnectionMessageQueueOut outgoingQueue, ActiveSettings peerSettings, ActiveSettings localSettings, - ActiveStateHandler onActiveStateChanged, - ) { + ActiveStateHandler onActiveStateChanged, [ + int? peerInitiatedStreamLimit, + int? maxPeerStreamLimitViolations, + void Function(String)? onPeerStreamLimitAbuse, + bool allowPeerPushes = false, + ]) { return StreamHandler._( writer, incomingQueue, @@ -209,6 +231,10 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { onActiveStateChanged, 2, -1, + peerInitiatedStreamLimit, + maxPeerStreamLimitViolations, + onPeerStreamLimitAbuse, + allowPeerPushes, ); } @@ -275,7 +301,9 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { Http2StreamImpl newLocalStream() { return ensureNotTerminatedSync(() { - assert(_canCreateNewStream()); + if (!_canCreateNewStream()) { + throw StateError('Maximum number of concurrent streams reached.'); + } if (MAX_STREAM_ID < nextStreamId) { throw StateError( @@ -288,36 +316,85 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { }); } - Http2StreamImpl newRemoteStream(int remoteStreamId) { + Http2StreamImpl? newRemoteStream(int remoteStreamId) { return ensureNotTerminatedSync(() { - assert(remoteStreamId <= MAX_STREAM_ID); - // NOTE: We cannot enforce that a new stream id is 2 higher than the last - // used stream id. Meaning there can be "holes" in the sense that stream - // ids are not used: - // - // http/2 spec: - // The first use of a new stream identifier implicitly closes all - // streams in the "idle" state that might have been initiated by that - // peer with a lower-valued stream identifier. For example, if a client - // sends a HEADERS frame on stream 7 without ever sending a frame on - // stream 5, then stream 5 transitions to the "closed" state when the - // first frame for stream 7 is sent or received. - - if (remoteStreamId <= lastRemoteStreamId) { - throw ProtocolException( - 'Remote tried to open new stream which is ' - 'not in "idle" state.', - ); + _registerNewRemoteStreamId(remoteStreamId); + if (!_canAcceptPeerInitiatedStream()) { + _rejectPeerStreamLimitViolation(remoteStreamId); + return null; } - - var sameDirection = (nextStreamId + remoteStreamId).isEven; - assert(!sameDirection); - - lastRemoteStreamId = remoteStreamId; + _consecutivePeerStreamLimitViolations = 0; return _newStreamInternal(remoteStreamId); }); } + void _registerNewRemoteStreamId(int remoteStreamId) { + if (remoteStreamId > MAX_STREAM_ID) { + throw ProtocolException('Remote stream id exceeds the HTTP/2 limit.'); + } + + // A new higher stream id implicitly closes lower unused stream ids, so + // holes in the sequence are valid but reuse and reordering are not. + if (remoteStreamId <= lastRemoteStreamId) { + throw ProtocolException( + 'Remote tried to open new stream which is not in "idle" state.', + ); + } + + var sameDirection = (nextStreamId + remoteStreamId).isEven; + if (sameDirection) { + throw ProtocolException( + 'Remote tried to open a stream with a locally initiated stream id.', + ); + } + lastRemoteStreamId = remoteStreamId; + } + + bool _canAcceptPeerInitiatedStream() { + final limit = _peerInitiatedStreamLimit; + return limit == null || + _numberOfActivePeerStreams + _numberOfReservedPeerStreams < limit; + } + + void _rejectPeerStreamLimitViolation(int streamId) { + _frameWriter.writeRstStreamFrame(streamId, ErrorCode.REFUSED_STREAM); + // Enforce local admission immediately, but do not treat the peer as abusive + // until it has acknowledged the advertised limit and can be expected to + // obey it. + if (_localSettings.maxConcurrentStreams == null) return; + _consecutivePeerStreamLimitViolations++; + final violationLimit = _maxPeerStreamLimitViolations; + if (violationLimit != null && + _consecutivePeerStreamLimitViolations >= violationLimit) { + _onPeerStreamLimitAbuse?.call( + 'Peer repeatedly exceeded SETTINGS_MAX_CONCURRENT_STREAMS ' + '($_consecutivePeerStreamLimitViolations violations, ' + 'configured stream limit: $_peerInitiatedStreamLimit).', + ); + } + } + + void _rejectOversizedNewRemoteStream(HeadersFrame frame) { + _registerNewRemoteStreamId(frame.header.streamId); + _frameWriter.writeRstStreamFrame( + frame.header.streamId, + ErrorCode.ENHANCE_YOUR_CALM, + ); + } + + void _rejectOversizedHeaders(Http2StreamImpl stream, HeadersFrame frame) { + _frameWriter.writeRstStreamFrame(stream.id, ErrorCode.ENHANCE_YOUR_CALM); + _closeStreamAbnormally( + stream, + StreamException( + stream.id, + 'Decoded field section exceeded the local limit ' + '(${frame.decodedHeaderListSize} bytes).', + ), + propagateException: true, + ); + } + Http2StreamImpl _newStreamInternal(int streamId) { // For each new stream we must: // - setup sending/receiving [Window]s with correct initial size @@ -607,7 +684,12 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { if (frame is HeadersFrame) { if (isServer) { + if (frame.headerListSizeExceeded) { + _rejectOversizedNewRemoteStream(frame); + return; + } var newStream = newRemoteStream(frame.header.streamId); + if (newStream == null) return; _changeState(newStream, StreamState.Open); _handleHeadersFrame(newStream, frame); @@ -688,6 +770,10 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { throw _throwStreamClosedException(frame.header.streamId); } } else { + if (frame is HeadersFrame && frame.headerListSizeExceeded) { + _rejectOversizedHeaders(stream, frame); + return; + } if (frame is HeadersFrame) { _handleHeadersFrame(stream, frame); } else if (frame is DataFrame) { @@ -740,12 +826,28 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { } void _handlePushPromiseFrame(Http2StreamImpl stream, PushPromiseFrame frame) { + if (!_allowPeerPushes) { + throw ProtocolException( + 'Received PUSH_PROMISE while server push is disabled.', + ); + } + if (stream.state != StreamState.Open && stream.state != StreamState.HalfClosedLocal) { throw ProtocolException('Expected open state (was: ${stream.state}).'); } + if (frame.headerListSizeExceeded) { + _registerNewRemoteStreamId(frame.promisedStreamId); + _frameWriter.writeRstStreamFrame( + frame.promisedStreamId, + ErrorCode.ENHANCE_YOUR_CALM, + ); + return; + } + var pushedStream = newRemoteStream(frame.promisedStreamId); + if (pushedStream == null) return; _changeState(pushedStream, StreamState.ReservedRemote); incomingQueue.processPushPromiseFrame(frame, pushedStream); @@ -915,6 +1017,14 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { /// [StreamState.HalfClosedRemote]) int _numberOfActiveStreams = 0; + /// The number of active streams initiated by the peer. + int _numberOfActivePeerStreams = 0; + + /// Peer-initiated streams in ReservedRemote state. Although these do not + /// count toward SETTINGS_MAX_CONCURRENT_STREAMS, they allocate the same + /// stream machinery and are included in the local admission guard. + int _numberOfReservedPeerStreams = 0; + bool _canCreateNewStream() { var limit = _peerSettings.maxConcurrentStreams; return limit == null || _numberOfActiveStreams < limit; @@ -946,34 +1056,57 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { (from != StreamState.Terminated && to == StreamState.Terminated), ); - // If we initiated the stream and it became "open" or "closed" we need to - // update the [_numberOfActiveStreams] counter. - if (_didInitiateStream(stream)) { - // NOTE: We wait until the stream is completely done. - // (If we waited only until `StreamState.Closed` then we might still have - // the endStream header/data message buffered, but not yet sent out). - switch (stream.state) { + // Locally initiated streams remain counted until buffered output is done. + // Peer-initiated streams follow the RFC active-state definition and stop + // counting when they enter Closed. + int updateActiveCount(int count, {required bool countClosedAsActive}) { + switch (from) { case StreamState.ReservedLocal: case StreamState.ReservedRemote: case StreamState.Idle: if (to == StreamState.Open || to == StreamState.HalfClosedLocal || to == StreamState.HalfClosedRemote) { - _numberOfActiveStreams++; + return count + 1; } break; case StreamState.Open: case StreamState.HalfClosedLocal: case StreamState.HalfClosedRemote: + if (to == StreamState.Terminated || + (!countClosedAsActive && to == StreamState.Closed)) { + return count - 1; + } + break; case StreamState.Closed: - if (to == StreamState.Terminated) { - _numberOfActiveStreams--; + if (countClosedAsActive && to == StreamState.Terminated) { + return count - 1; } break; case StreamState.Terminated: // There is nothing to do here. break; } + return count; + } + + if (_didInitiateStream(stream)) { + _numberOfActiveStreams = updateActiveCount( + _numberOfActiveStreams, + countClosedAsActive: true, + ); + } else { + if (from == StreamState.Idle && to == StreamState.ReservedRemote) { + _numberOfReservedPeerStreams++; + } else if (from == StreamState.ReservedRemote && + to != StreamState.ReservedRemote) { + _numberOfReservedPeerStreams--; + assert(_numberOfReservedPeerStreams >= 0); + } + _numberOfActivePeerStreams = updateActiveCount( + _numberOfActivePeerStreams, + countClosedAsActive: false, + ); } stream.state = to; } diff --git a/pkgs/http2/lib/src/sync_errors.dart b/pkgs/http2/lib/src/sync_errors.dart index 3d11616ad1..8ff88f598a 100644 --- a/pkgs/http2/lib/src/sync_errors.dart +++ b/pkgs/http2/lib/src/sync_errors.dart @@ -29,6 +29,16 @@ class FrameSizeException implements Exception { String toString() => 'FrameSizeException: $_message'; } +/// An inbound field block exceeded a local resource limit. +class HeaderBlockProcessingException implements Exception { + final String _message; + + HeaderBlockProcessingException(this._message); + + @override + String toString() => 'HeaderBlockProcessingException: $_message'; +} + class TerminatedException implements Exception { @override String toString() => 'TerminatedException: The object has been terminated.'; diff --git a/pkgs/http2/lib/transport.dart b/pkgs/http2/lib/transport.dart index 4584e71bc6..e9a6bca900 100644 --- a/pkgs/http2/lib/transport.dart +++ b/pkgs/http2/lib/transport.dart @@ -13,22 +13,124 @@ export 'src/hpack/hpack.dart' show Header; typedef ActiveStateHandler = void Function(bool isActive); +/// Default maximum number of peer-initiated streams per connection. +/// +/// A limit of 100 follows the lower bound recommended by +/// [RFC 9113 section 6.5.2](https://www.rfc-editor.org/rfc/rfc9113.html#section-6.5.2), +/// avoiding an unnecessarily restrictive default while bounding per-stream +/// state. +const int defaultMaxConcurrentStreams = 100; + +/// Default maximum compressed size of one inbound HTTP/2 field block. +/// +/// 16 KiB matches the protocol's initial maximum frame payload size from +/// [RFC 9113 section 4.1](https://www.rfc-editor.org/rfc/rfc9113.html#section-4.1). +/// This bounds fragmented field-block buffering to the byte budget of one +/// default-sized frame. +const int defaultMaxInboundHeaderBlockSize = 16 * 1024; + +/// Default maximum decoded size of one inbound HTTP/2 field section. +/// +/// 8 KiB is a conservative application-visible budget that bounds HPACK +/// amplification before headers reach application code. The field-size +/// accounting follows +/// [RFC 9113 section 6.5.2](https://www.rfc-editor.org/rfc/rfc9113.html#section-6.5.2). +const int defaultMaxInboundHeaderListSize = 8 * 1024; + +/// Default absolute time allowed to finish an inbound HTTP/2 field block. +/// +/// Ten seconds allows for ordinary network delay while bounding how long a +/// slow peer can retain field-block and HPACK processing state. +const Duration defaultInboundHeaderBlockTimeout = Duration(seconds: 10); + +/// Default maximum number of CONTINUATION frames in one field block. +/// +/// The byte limit alone does not bound floods of empty or very small frames. +/// 16 still permits a 16 KiB block to be split into fragments averaging 1 KiB +/// while placing a finite bound on per-frame processing, addressing the class +/// of issue described by [CERT VU#421644](https://kb.cert.org/vuls/id/421644). +const int defaultMaxContinuationFramesPerBlock = 16; + +/// Default number of consecutive peer stream-limit violations before GOAWAY. +/// +/// Once the peer has acknowledged the advertised stream limit, eight tolerates +/// a short burst of racing streams while still terminating a peer that +/// repeatedly ignores the limit. Accepting a peer stream resets the violation +/// count. +const int defaultMaxPeerStreamLimitViolations = 8; + /// Settings for a [TransportConnection]. abstract class Settings { /// The maximum number of concurrent streams the remote end can open - /// (defaults to being unlimited). + /// (defaults to [defaultMaxConcurrentStreams]). + /// + /// This limit is advertised and enforced locally. Set to `null` to make the + /// number of peer-initiated streams unlimited. Reserved peer streams are + /// included in the local allocation guard even though the HTTP/2 setting + /// counts only open and half-closed streams. final int? concurrentStreamLimit; /// The default stream window size the remote peer can use when creating new /// streams (defaults to 65535 bytes). final int? streamWindowSize; - const Settings({this.concurrentStreamLimit, this.streamWindowSize}); + /// Maximum compressed bytes retained for one inbound field block. + /// + /// This is a local enforcement limit and is not advertised to the peer. + /// Set to `null` to disable the limit. + final int? maxInboundHeaderBlockSize; + + /// Maximum decoded size of one inbound field section. + /// + /// The size is the sum of `name.length + value.length + 32` for every field. + /// This value is both advertised with SETTINGS_MAX_HEADER_LIST_SIZE and + /// enforced locally. Set to `null` to disable both behaviors. + final int? maxInboundHeaderListSize; + + /// Absolute time allowed to receive a complete inbound field block. + /// + /// The timer starts as soon as the initial HEADERS or PUSH_PROMISE frame + /// header is parsed, before its payload is buffered. It is not extended by + /// CONTINUATION frames and ends only after the final frame is fully read. + /// Set to `null` to disable the timeout. + final Duration? inboundHeaderBlockTimeout; + + /// Maximum number of CONTINUATION frames accepted for one field block. + /// + /// Set to `null` to disable the frame-count limit. + final int? maxContinuationFramesPerBlock; + + /// Consecutive peer stream-limit violations allowed before terminating the + /// connection with ENHANCE_YOUR_CALM. + /// + /// Individual violations are rejected with REFUSED_STREAM. The counter is + /// incremented only after the peer acknowledges the advertised stream limit + /// and is reset after a peer-initiated stream is accepted. Set to `null` to + /// never terminate the connection based only on this violation count. + final int? maxPeerStreamLimitViolations; + + const Settings({ + this.concurrentStreamLimit, + this.streamWindowSize, + this.maxInboundHeaderBlockSize, + this.maxInboundHeaderListSize, + this.inboundHeaderBlockTimeout, + this.maxContinuationFramesPerBlock, + this.maxPeerStreamLimitViolations, + }); } /// Settings for a [TransportConnection] a server can make. class ServerSettings extends Settings { - const ServerSettings({super.concurrentStreamLimit, super.streamWindowSize}); + const ServerSettings({ + super.concurrentStreamLimit = defaultMaxConcurrentStreams, + super.streamWindowSize, + super.maxInboundHeaderBlockSize = defaultMaxInboundHeaderBlockSize, + super.maxInboundHeaderListSize = defaultMaxInboundHeaderListSize, + super.inboundHeaderBlockTimeout = defaultInboundHeaderBlockTimeout, + super.maxContinuationFramesPerBlock = defaultMaxContinuationFramesPerBlock, + super.maxPeerStreamLimitViolations = defaultMaxPeerStreamLimitViolations, + }); } /// Settings for a [TransportConnection] a client can make. @@ -37,8 +139,13 @@ class ClientSettings extends Settings { final bool allowServerPushes; const ClientSettings({ - super.concurrentStreamLimit, + super.concurrentStreamLimit = defaultMaxConcurrentStreams, super.streamWindowSize, + super.maxInboundHeaderBlockSize = defaultMaxInboundHeaderBlockSize, + super.maxInboundHeaderListSize = defaultMaxInboundHeaderListSize, + super.inboundHeaderBlockTimeout = defaultInboundHeaderBlockTimeout, + super.maxContinuationFramesPerBlock = defaultMaxContinuationFramesPerBlock, + super.maxPeerStreamLimitViolations = defaultMaxPeerStreamLimitViolations, this.allowServerPushes = false, }); } @@ -119,9 +226,7 @@ abstract class ServerTransportConnection extends TransportConnection { factory ServerTransportConnection.viaStreams( Stream> incoming, StreamSink> outgoing, { - ServerSettings? settings = const ServerSettings( - concurrentStreamLimit: 1000, - ), + ServerSettings? settings = const ServerSettings(), }) { settings ??= const ServerSettings(); return ServerConnection(incoming, outgoing, settings); diff --git a/pkgs/http2/test/client_test.dart b/pkgs/http2/test/client_test.dart index 9f1a98a50d..e31ada0639 100644 --- a/pkgs/http2/test/client_test.dart +++ b/pkgs/http2/test/client_test.dart @@ -66,6 +66,174 @@ void main() { }); }); + group('resource limits', () { + clientTest('rejects PUSH_PROMISE when server push is disabled', ( + ClientTransportConnection client, + FrameWriter serverWriter, + StreamIterator serverReader, + Future Function() nextFrame, + ) async { + final handshakeDone = Completer(); + + Future serverFun() async { + serverWriter.writeSettingsFrame([]); + final clientSettings = await nextFrame() as SettingsFrame; + expect( + clientSettings.settings, + contains( + isA() + .having( + (s) => s.identifier, + 'identifier', + Setting.SETTINGS_ENABLE_PUSH, + ) + .having((s) => s.value, 'value', 0), + ), + ); + serverWriter.writeSettingsAckFrame(); + expect(await nextFrame(), isA()); + handshakeDone.complete(); + + final request = await nextFrame() as HeadersFrame; + serverWriter.writePushPromiseFrame(request.header.streamId, 2, [ + Header.ascii('a', 'b'), + ]); + + expect( + await nextFrame(), + isA() + .having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.PROTOCOL_ERROR, + ) + .having( + (f) => ascii.decode(f.debugData), + 'debugData', + contains('server push is disabled'), + ), + ); + expect(await serverReader.moveNext(), isFalse); + await serverWriter.close(); + } + + Future clientFun() async { + await handshakeDone.future; + final stream = client.makeRequest([ + Header.ascii('a', 'b'), + ], endStream: true); + final incoming = stream.incomingMessages.drain().then( + (_) {}, + onError: (_) {}, + ); + expect(await stream.peerPushes.toList(), isEmpty); + await incoming; + } + + await Future.wait([serverFun(), clientFun()]); + }); + + clientTest( + 'bounds reserved peer streams and recovers after reset', + ( + ClientTransportConnection client, + FrameWriter serverWriter, + StreamIterator serverReader, + Future Function() nextFrame, + ) async { + final handshakeDone = Completer(); + final receivedTwoPushes = Completer(); + final receivedThreePushes = Completer(); + final pushedStreamIds = []; + final pushedStreamDone = >[]; + + Future serverFun() async { + serverWriter.writeSettingsFrame([]); + expect(await nextFrame(), isA()); + serverWriter.writeSettingsAckFrame(); + expect(await nextFrame(), isA()); + handshakeDone.complete(); + + final request = await nextFrame() as HeadersFrame; + final parentStreamId = request.header.streamId; + serverWriter.writePushPromiseFrame(parentStreamId, 2, [ + Header.ascii('a', 'b'), + ]); + serverWriter.writePushPromiseFrame(parentStreamId, 4, [ + Header.ascii('a', 'b'), + ]); + await receivedTwoPushes.future; + + serverWriter.writePushPromiseFrame(parentStreamId, 6, [ + Header.ascii('a', 'b'), + ]); + expect( + await nextFrame(), + isA() + .having((f) => f.header.streamId, 'header.streamId', 6) + .having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.REFUSED_STREAM, + ), + ); + + serverWriter.writeRstStreamFrame(2, ErrorCode.CANCEL); + serverWriter.writePushPromiseFrame(parentStreamId, 8, [ + Header.ascii('a', 'b'), + ]); + await receivedThreePushes.future; + + serverWriter.writeRstStreamFrame(4, ErrorCode.CANCEL); + serverWriter.writeRstStreamFrame(8, ErrorCode.CANCEL); + serverWriter.writeHeadersFrame(parentStreamId, [ + Header.ascii('status', 'ok'), + ], endStream: true); + + expect( + await nextFrame(), + isA().having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.NO_ERROR, + ), + ); + expect(await serverReader.moveNext(), isFalse); + await serverWriter.close(); + } + + Future clientFun() async { + await handshakeDone.future; + final stream = client.makeRequest([ + Header.ascii('request', 'ok'), + ], endStream: true); + stream.peerPushes.listen((push) { + pushedStreamIds.add(push.stream.id); + pushedStreamDone.add( + push.stream.incomingMessages.drain().then( + (_) {}, + onError: (_) {}, + ), + ); + if (pushedStreamIds.length == 2) receivedTwoPushes.complete(); + if (pushedStreamIds.length == 3) receivedThreePushes.complete(); + }); + + expect(await stream.incomingMessages.toList(), hasLength(1)); + await Future.wait(pushedStreamDone); + expect(pushedStreamIds, [2, 4, 8]); + await client.finish(); + } + + await Future.wait([serverFun(), clientFun()]); + }, + clientSettings: const ClientSettings( + allowServerPushes: true, + concurrentStreamLimit: 2, + ), + ); + }); + group('connection-operational', () { clientTest('on-connection-operational-fires', ( ClientTransportConnection client, @@ -881,7 +1049,7 @@ void main() { } await Future.wait([serverFun(), clientFun()]); - }); + }, clientSettings: const ClientSettings(allowServerPushes: true)); clientTest('client-reports-flowcontrol-error-on-negative-window', ( ClientTransportConnection client, @@ -1095,10 +1263,11 @@ void clientTest( StreamIterator frameReader, Future Function() readNext, ) - func, -) { + func, { + ClientSettings? clientSettings, +}) { return test(name, () { - var streams = ClientStreams(); + var streams = ClientStreams(clientSettings); var serverReader = streams.serverConnectionFrameReader; Future readNext() async { @@ -1116,11 +1285,14 @@ void clientTest( } class ClientStreams { + final ClientSettings? clientSettings; final StreamController> writeA = StreamController(); final StreamController> writeB = StreamController(); Stream> get readA => writeA.stream; Stream> get readB => writeB.stream; + ClientStreams([this.clientSettings]); + StreamIterator get serverConnectionFrameReader { var localSettings = ActiveSettings(); var streamAfterConnectionPreface = readConnectionPreface(readA); @@ -1136,5 +1308,9 @@ class ClientStreams { } ClientTransportConnection get clientConnection => - ClientTransportConnection.viaStreams(readB, writeA); + ClientTransportConnection.viaStreams( + readB, + writeA, + settings: clientSettings, + ); } diff --git a/pkgs/http2/test/server_test.dart b/pkgs/http2/test/server_test.dart index 2aac5fe2e2..d3a1a0d176 100644 --- a/pkgs/http2/test/server_test.dart +++ b/pkgs/http2/test/server_test.dart @@ -3,11 +3,13 @@ // BSD-style license that can be found in the LICENSE file. import 'dart:async'; +import 'dart:convert'; import 'package:http2/src/connection_preface.dart'; import 'package:http2/src/frames/frames.dart'; import 'package:http2/src/hpack/hpack.dart'; import 'package:http2/src/settings/settings.dart'; +import 'package:http2/src/sync_errors.dart'; import 'package:http2/transport.dart'; import 'package:test/test.dart'; @@ -42,6 +44,488 @@ void main() { }); }); + group('resource limits', () { + serverTest('advertises secure defaults', ( + ServerTransportConnection server, + FrameWriter clientWriter, + StreamIterator clientReader, + Future Function() nextFrame, + ) async { + final settingsFrame = await nextFrame() as SettingsFrame; + final settings = { + for (final setting in settingsFrame.settings) + setting.identifier: setting.value, + }; + + expect( + settings[Setting.SETTINGS_MAX_CONCURRENT_STREAMS], + defaultMaxConcurrentStreams, + ); + expect( + settings[Setting.SETTINGS_MAX_HEADER_LIST_SIZE], + defaultMaxInboundHeaderListSize, + ); + + clientWriter.writeSettingsAckFrame(); + clientWriter.writeSettingsFrame([]); + expect(await nextFrame(), isA()); + final termination = server.terminate(); + expect( + await nextFrame(), + isA().having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.NO_ERROR, + ), + ); + expect(await clientReader.moveNext(), isFalse); + await termination; + }); + + serverTest( + 'enforces peer stream limit and recovers after close', + ( + ServerTransportConnection server, + FrameWriter clientWriter, + StreamIterator clientReader, + Future Function() nextFrame, + ) async { + final acceptedTwo = Completer(); + final refusalObserved = Completer(); + + Future serverFun() async { + final streams = StreamIterator(server.incomingStreams); + expect(await streams.moveNext(), isTrue); + final first = streams.current; + expect(first.id, 1); + expect(await streams.moveNext(), isTrue); + expect(streams.current.id, 3); + acceptedTwo.complete(); + + await refusalObserved.future; + first.terminate(); + + expect(await streams.moveNext(), isTrue); + expect(streams.current.id, 7); + await server.terminate(); + } + + Future clientFun() async { + expect(await nextFrame(), isA()); + // Deliberately withhold the ACK for the server's advertised + // limits. Local admission must not depend on peer compliance. + clientWriter.writeSettingsFrame([]); + expect(await nextFrame(), isA()); + + clientWriter.writeHeadersFrame(1, [Header.ascii('a', 'b')]); + clientWriter.writeHeadersFrame(3, [Header.ascii('a', 'b')]); + await acceptedTwo.future; + + clientWriter.writeHeadersFrame(5, [Header.ascii('a', 'b')]); + expect( + await nextFrame(), + isA() + .having((f) => f.header.streamId, 'header.streamId', 5) + .having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.REFUSED_STREAM, + ), + ); + refusalObserved.complete(); + + expect( + await nextFrame(), + isA().having( + (f) => f.header.streamId, + 'header.streamId', + 1, + ), + ); + clientWriter.writeHeadersFrame(7, [Header.ascii('a', 'b')]); + expect( + await nextFrame(), + isA().having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.NO_ERROR, + ), + ); + expect(await clientReader.moveNext(), isFalse); + } + + await Future.wait([serverFun(), clientFun()]); + }, + serverSettings: const ServerSettings( + concurrentStreamLimit: 2, + maxPeerStreamLimitViolations: 8, + ), + ); + + serverTest( + 'terminates repeated peer stream-limit violations', + ( + ServerTransportConnection server, + FrameWriter clientWriter, + StreamIterator clientReader, + Future Function() nextFrame, + ) async { + Future serverFun() async { + final streams = StreamIterator(server.incomingStreams); + expect(await streams.moveNext(), isTrue); + expect(streams.current.id, 1); + expect(await streams.moveNext(), isFalse); + } + + Future clientFun() async { + expect(await nextFrame(), isA()); + clientWriter.writeSettingsAckFrame(); + clientWriter.writeSettingsFrame([]); + expect(await nextFrame(), isA()); + + clientWriter.writeHeadersFrame(1, [Header.ascii('a', 'b')]); + for (final streamId in [3, 5, 7]) { + clientWriter.writeHeadersFrame(streamId, [ + Header.ascii('a', 'b'), + ]); + expect( + await nextFrame(), + isA() + .having( + (f) => f.header.streamId, + 'header.streamId', + streamId, + ) + .having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.REFUSED_STREAM, + ), + ); + } + expect( + await nextFrame(), + isA().having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.ENHANCE_YOUR_CALM, + ), + ); + expect(await clientReader.moveNext(), isFalse); + } + + await Future.wait([serverFun(), clientFun()]); + }, + serverSettings: const ServerSettings( + concurrentStreamLimit: 1, + maxPeerStreamLimitViolations: 3, + ), + ); + + serverTest( + 'counts stream-limit violations only after settings ACK', + ( + ServerTransportConnection server, + FrameWriter clientWriter, + StreamIterator clientReader, + Future Function() nextFrame, + ) async { + Future serverFun() async { + final streams = StreamIterator(server.incomingStreams); + expect(await streams.moveNext(), isTrue); + expect(streams.current.id, 1); + expect(await streams.moveNext(), isFalse); + } + + Future expectRefused(int streamId) async { + clientWriter.writeHeadersFrame(streamId, [Header.ascii('a', 'b')]); + expect( + await nextFrame(), + isA() + .having((f) => f.header.streamId, 'header.streamId', streamId) + .having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.REFUSED_STREAM, + ), + ); + } + + Future clientFun() async { + expect(await nextFrame(), isA()); + clientWriter.writeSettingsFrame([]); + expect(await nextFrame(), isA()); + + clientWriter.writeHeadersFrame(1, [Header.ascii('a', 'b')]); + for (final streamId in [3, 5, 7]) { + await expectRefused(streamId); + } + + clientWriter.writeSettingsAckFrame(); + for (final streamId in [9, 11, 13]) { + await expectRefused(streamId); + } + expect( + await nextFrame(), + isA().having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.ENHANCE_YOUR_CALM, + ), + ); + expect(await clientReader.moveNext(), isFalse); + } + + await Future.wait([serverFun(), clientFun()]); + }, + serverSettings: const ServerSettings( + concurrentStreamLimit: 1, + maxPeerStreamLimitViolations: 3, + ), + ); + + serverTest( + 'rejects an oversized decoded field section before publication', + ( + ServerTransportConnection server, + FrameWriter clientWriter, + StreamIterator clientReader, + Future Function() nextFrame, + ) async { + Future serverFun() async { + final streams = StreamIterator(server.incomingStreams); + expect(await streams.moveNext(), isTrue); + expect(streams.current.id, 3); + await server.terminate(); + } + + Future clientFun() async { + expect(await nextFrame(), isA()); + clientWriter.writeSettingsAckFrame(); + clientWriter.writeSettingsFrame([]); + expect(await nextFrame(), isA()); + + clientWriter.writeHeadersFrame(1, [Header.ascii('a', 'x' * 40)]); + expect( + await nextFrame(), + isA() + .having((f) => f.header.streamId, 'header.streamId', 1) + .having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.ENHANCE_YOUR_CALM, + ), + ); + + clientWriter.writeHeadersFrame(3, [Header.ascii('a', 'b')]); + expect(await nextFrame(), isA()); + expect(await clientReader.moveNext(), isFalse); + } + + await Future.wait([serverFun(), clientFun()]); + }, + serverSettings: const ServerSettings(maxInboundHeaderListSize: 64), + ); + + serverTest('rejects oversized trailers without publishing them', ( + ServerTransportConnection server, + FrameWriter clientWriter, + StreamIterator clientReader, + Future Function() nextFrame, + ) async { + Future serverFun() async { + final streams = StreamIterator(server.incomingStreams); + expect(await streams.moveNext(), isTrue); + final messages = StreamIterator(streams.current.incomingMessages); + expect(await messages.moveNext(), isTrue); + final initial = messages.current as HeadersStreamMessage; + expect(initial.headers, hasLength(1)); + expect(initial.headers.single.value, ascii.encode('b')); + await expectLater( + messages.moveNext(), + throwsA(isA()), + ); + await server.terminate(); + } + + Future clientFun() async { + expect(await nextFrame(), isA()); + clientWriter.writeSettingsAckFrame(); + clientWriter.writeSettingsFrame([]); + expect(await nextFrame(), isA()); + + clientWriter.writeHeadersFrame(1, [ + Header.ascii('a', 'b'), + ], endStream: false); + clientWriter.writeHeadersFrame(1, [ + Header.ascii('trailer', 'x' * 40), + ]); + expect( + await nextFrame(), + isA() + .having((f) => f.header.streamId, 'header.streamId', 1) + .having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.ENHANCE_YOUR_CALM, + ), + ); + expect(await nextFrame(), isA()); + expect(await clientReader.moveNext(), isFalse); + } + + await Future.wait([serverFun(), clientFun()]); + }, serverSettings: const ServerSettings(maxInboundHeaderListSize: 64)); + + test('terminates an oversized compressed field block', () async { + final streams = ClientErrorStreams( + const ServerSettings(maxInboundHeaderBlockSize: 2), + ); + final server = streams.serverConnection; + final incoming = server.incomingStreams.toList(); + final clientReader = streams.clientConnectionFrameReader; + + streams.writeConnectionPreface(); + + expect(await clientReader.moveNext(), isTrue); + expect(clientReader.current, isA()); + streams.writeRawFrame( + type: FrameType.SETTINGS, + flags: SettingsFrame.FLAG_ACK, + streamId: 0, + payload: const [], + ); + streams.writeRawFrame( + type: FrameType.SETTINGS, + flags: 0, + streamId: 0, + payload: const [], + ); + expect(await clientReader.moveNext(), isTrue); + expect(clientReader.current, isA()); + + streams.writeRawFrame( + type: FrameType.HEADERS, + flags: HeadersFrame.FLAG_END_HEADERS, + streamId: 1, + payload: [0x84, 0x84, 0x84], + ); + + expect(await clientReader.moveNext(), isTrue); + expect( + clientReader.current, + isA().having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.ENHANCE_YOUR_CALM, + ), + ); + expect(await clientReader.moveNext(), isFalse); + expect(await incoming, isEmpty); + }); + + test('terminates an incomplete field block after the timeout', () async { + final streams = ClientErrorStreams( + const ServerSettings( + inboundHeaderBlockTimeout: Duration(milliseconds: 50), + ), + ); + final server = streams.serverConnection; + final incoming = server.incomingStreams.toList(); + final clientReader = streams.clientConnectionFrameReader; + + streams.writeConnectionPreface(); + + expect(await clientReader.moveNext(), isTrue); + expect(clientReader.current, isA()); + streams.writeRawFrame( + type: FrameType.SETTINGS, + flags: SettingsFrame.FLAG_ACK, + streamId: 0, + payload: const [], + ); + streams.writeRawFrame( + type: FrameType.SETTINGS, + flags: 0, + streamId: 0, + payload: const [], + ); + expect(await clientReader.moveNext(), isTrue); + expect(clientReader.current, isA()); + + streams.writeRawFrame( + type: FrameType.HEADERS, + flags: 0, + streamId: 1, + payload: [0x84], + ); + + expect(await clientReader.moveNext(), isTrue); + expect( + clientReader.current, + isA().having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.ENHANCE_YOUR_CALM, + ), + ); + expect(await clientReader.moveNext(), isFalse); + expect(await incoming, isEmpty); + }); + + test( + 'starts the field-block timeout before payload completion', + () async { + final streams = ClientErrorStreams( + const ServerSettings( + inboundHeaderBlockTimeout: Duration(milliseconds: 50), + ), + ); + final server = streams.serverConnection; + final incoming = server.incomingStreams.toList(); + final clientReader = streams.clientConnectionFrameReader; + + streams.writeConnectionPreface(); + + expect(await clientReader.moveNext(), isTrue); + expect(clientReader.current, isA()); + streams.writeRawFrame( + type: FrameType.SETTINGS, + flags: SettingsFrame.FLAG_ACK, + streamId: 0, + payload: const [], + ); + streams.writeRawFrame( + type: FrameType.SETTINGS, + flags: 0, + streamId: 0, + payload: const [], + ); + expect(await clientReader.moveNext(), isTrue); + expect(clientReader.current, isA()); + + streams.writeRawFrameHeader( + length: 16, + type: FrameType.HEADERS, + flags: HeadersFrame.FLAG_END_HEADERS, + streamId: 1, + ); + + expect(await clientReader.moveNext(), isTrue); + expect( + clientReader.current, + isA().having( + (f) => f.errorCode, + 'errorCode', + ErrorCode.ENHANCE_YOUR_CALM, + ), + ); + expect(await clientReader.moveNext(), isFalse); + expect(await incoming, isEmpty); + }, + ); + }); + group('client-errors', () { serverTest('no-settings-frame-at-beginning', ( ServerTransportConnection server, @@ -228,10 +712,11 @@ void serverTest( StreamIterator frameReader, Future Function() readNext, ) - func, -) { + func, { + ServerSettings? serverSettings, +}) { return test(name, () { - var streams = ClientErrorStreams(); + var streams = ClientErrorStreams(serverSettings); var clientReader = streams.clientConnectionFrameReader; Future readNext() async { @@ -249,11 +734,58 @@ void serverTest( } class ClientErrorStreams { + final ServerSettings? serverSettings; final StreamController> writeA = StreamController(); final StreamController> writeB = StreamController(); Stream> get readA => writeA.stream; Stream> get readB => writeB.stream; + ClientErrorStreams([this.serverSettings]); + + void writeConnectionPreface() { + writeB.add(CONNECTION_PREFACE); + } + + void writeRawFrame({ + required int type, + required int flags, + required int streamId, + required List payload, + }) { + final length = payload.length; + writeB.add([ + (length >> 16) & 0xff, + (length >> 8) & 0xff, + length & 0xff, + type, + flags, + (streamId >> 24) & 0x7f, + (streamId >> 16) & 0xff, + (streamId >> 8) & 0xff, + streamId & 0xff, + ...payload, + ]); + } + + void writeRawFrameHeader({ + required int length, + required int type, + required int flags, + required int streamId, + }) { + writeB.add([ + (length >> 16) & 0xff, + (length >> 8) & 0xff, + length & 0xff, + type, + flags, + (streamId >> 24) & 0x7f, + (streamId >> 16) & 0xff, + (streamId >> 8) & 0xff, + streamId & 0xff, + ]); + } + StreamIterator get clientConnectionFrameReader { var localSettings = ActiveSettings(); return StreamIterator(FrameReader(readA, localSettings).startDecoding()); @@ -267,5 +799,9 @@ class ClientErrorStreams { } ServerTransportConnection get serverConnection => - ServerTransportConnection.viaStreams(readB, writeA); + ServerTransportConnection.viaStreams( + readB, + writeA, + settings: serverSettings, + ); } diff --git a/pkgs/http2/test/src/frames/frame_defragmenter_test.dart b/pkgs/http2/test/src/frames/frame_defragmenter_test.dart index 8d012eb842..9b6b911fb3 100644 --- a/pkgs/http2/test/src/frames/frame_defragmenter_test.dart +++ b/pkgs/http2/test/src/frames/frame_defragmenter_test.dart @@ -4,6 +4,7 @@ import 'package:http2/src/frames/frame_defragmenter.dart'; import 'package:http2/src/frames/frames.dart'; +import 'package:http2/src/sync_errors.dart'; import 'package:test/test.dart'; import '../error_matchers.dart'; @@ -156,6 +157,86 @@ void main() { var f1 = continuationFrame([4, 5, 6], fragmented: true, streamId: 1); expect(defrag.tryDefragmentFrame(f1), equals(f1)); }); + + test('allows a compressed field block at the exact byte limit', () { + var defrag = FrameDefragmenter(maxHeaderBlockSize: 6); + + expect( + defrag.tryDefragmentFrame(headersFrame([1, 2, 3, 4, 5, 6])), + isA(), + ); + }); + + test('rejects a compressed field block one byte over the limit', () { + var defrag = FrameDefragmenter(maxHeaderBlockSize: 5); + + expect( + () => defrag.tryDefragmentFrame(headersFrame([1, 2, 3, 4, 5, 6])), + throwsA(isA()), + ); + }); + + test('enforces the byte limit while receiving continuations', () { + var defrag = FrameDefragmenter(maxHeaderBlockSize: 6); + + expect( + defrag.tryDefragmentFrame(headersFrame([1, 2, 3], fragmented: true)), + isNull, + ); + expect( + defrag.tryDefragmentFrame( + continuationFrame([4, 5, 6], fragmented: true), + ), + isNull, + ); + expect( + () => defrag.tryDefragmentFrame(continuationFrame([7])), + throwsA(isA()), + ); + expect(defrag.isDefragmenting, isFalse); + }); + + test('enforces the continuation frame count', () { + var defrag = FrameDefragmenter(maxContinuationFrames: 2); + + expect( + defrag.tryDefragmentFrame(headersFrame([1], fragmented: true)), + isNull, + ); + expect( + defrag.tryDefragmentFrame(continuationFrame([2], fragmented: true)), + isNull, + ); + expect( + defrag.tryDefragmentFrame(continuationFrame([3], fragmented: true)), + isNull, + ); + expect( + () => defrag.tryDefragmentFrame(continuationFrame([4])), + throwsA(isA()), + ); + }); + + test('combines many small continuations once at completion', () { + var defrag = FrameDefragmenter( + maxHeaderBlockSize: 65, + maxContinuationFrames: 64, + ); + expect( + defrag.tryDefragmentFrame(headersFrame([0], fragmented: true)), + isNull, + ); + for (var i = 1; i < 64; i++) { + expect( + defrag.tryDefragmentFrame(continuationFrame([i], fragmented: true)), + isNull, + ); + } + final result = + defrag.tryDefragmentFrame(continuationFrame([64])) as HeadersFrame; + + expect(result.headerBlockFragment, List.generate(65, (i) => i)); + }); }); }); } diff --git a/pkgs/http2/test/src/hpack/hpack_test.dart b/pkgs/http2/test/src/hpack/hpack_test.dart index 4882562338..bab1056e6f 100644 --- a/pkgs/http2/test/src/hpack/hpack_test.dart +++ b/pkgs/http2/test/src/hpack/hpack_test.dart @@ -590,6 +590,100 @@ void main() { }); }); + group('bounded decoder tests', () { + test('allows the decoded field section at the exact limit', () { + var context = HPackContext(); + + final result = context.decoder.decodeWithLimit([ + 0x84, + ], maxHeaderListSize: 38); + + expect(result.headerListSize, 38); + expect(result.headerListSizeExceeded, isFalse); + expect(result.headers, [isHeader(':path', '/')]); + }); + + test('discards retained headers one byte over the decoded limit', () { + var context = HPackContext(); + + final result = context.decoder.decodeWithLimit([ + 0x84, + ], maxHeaderListSize: 37); + + expect(result.headerListSize, 38); + expect(result.headerListSizeExceeded, isTrue); + expect(result.headers, isEmpty); + }); + + test('bounds repeated indexed-field amplification', () { + var context = HPackContext(); + + final result = context.decoder.decodeWithLimit( + List.filled(300, 0x84), + maxHeaderListSize: 8 * 1024, + ); + + expect(result.headerListSize, 300 * 38); + expect(result.headerListSizeExceeded, isTrue); + expect(result.headers, isEmpty); + }); + + test('counts Huffman-expanded values against the decoded limit', () { + var context = HPackContext(); + const huffmanAuthority = [ + 0x41, + 0x8c, + 0xf1, + 0xe3, + 0xc2, + 0xe5, + 0xf2, + 0x3a, + 0x6b, + 0xa0, + 0xab, + 0x90, + 0xf4, + 0xff, + ]; + + final result = context.decoder.decodeWithLimit( + huffmanAuthority, + maxHeaderListSize: 56, + ); + + expect(result.headerListSize, 57); + expect(result.headerListSizeExceeded, isTrue); + expect(result.headers, isEmpty); + }); + + test('preserves dynamic-table synchronization after discarding', () { + const charA = 0x61; + const charB = 0x62; + var context = HPackContext(); + final oversizedBlock = [ + ...TestHelper.insertIntoDynamicTable(40, charA, charB), + 0x84, + ]; + + final rejected = context.decoder.decodeWithLimit( + oversizedBlock, + maxHeaderListSize: 40, + ); + expect(rejected.headerListSize, 78); + expect(rejected.headerListSizeExceeded, isTrue); + expect(rejected.headers, isEmpty); + + final next = context.decoder.decodeWithLimit( + TestHelper.dynamicTableLookup(0), + maxHeaderListSize: 40, + ); + expect(next.headerListSizeExceeded, isFalse); + expect(next.headers, hasLength(1)); + TestHelper.expectHeader(next.headers.single, 40, charA, charB); + }); + }); + group('negative-decoder-tests', () { test('invalid-integer-encoding', () { var context = HPackContext(); diff --git a/pkgs/http2/test/transport_test.dart b/pkgs/http2/test/transport_test.dart index 72aedf9ae7..f4477156fc 100644 --- a/pkgs/http2/test/transport_test.dart +++ b/pkgs/http2/test/transport_test.dart @@ -154,6 +154,50 @@ void main() { expect(await stream.peerPushes.toList(), isEmpty); }); + transportTest( + 'rejects an oversized push promise without publishing it', + ( + ClientTransportConnection client, + ServerTransportConnection server, + ) async { + Future serverFun() async { + await for (final stream in server.incomingStreams) { + expect(await stream.incomingMessages.toList(), hasLength(1)); + + final reset = Completer(); + final pushed = stream.push([Header.ascii('a', 'x' * 40)]); + pushed.onTerminated = (errorCode) { + expect(errorCode, ErrorCode.ENHANCE_YOUR_CALM); + reset.complete(); + }; + final pushedMessages = pushed.incomingMessages.drain().then( + (_) {}, + onError: (_) {}, + ); + + stream.sendHeaders([Header.ascii('status', 'ok')], endStream: true); + await Future.wait([reset.future, pushedMessages]); + } + await server.finish(); + } + + Future clientFun() async { + final stream = client.makeRequest([ + Header.ascii('request', 'ok'), + ], endStream: true); + expect(await stream.incomingMessages.toList(), hasLength(1)); + expect(await stream.peerPushes.toList(), isEmpty); + await client.finish(); + } + + await Future.wait([serverFun(), clientFun()]); + }, + clientSettings: const ClientSettings( + allowServerPushes: true, + maxInboundHeaderListSize: 64, + ), + ); + // By default, the stream concurrency level is set to this limit. const kDefaultStreamLimit = 100; transportTest( @@ -577,16 +621,12 @@ void main() { await testWindowSize(client, server, Window().size); }); - transportTest( - 'fast-sender-receiver-paused--10kb-window-size', - ( - ClientTransportConnection client, - ServerTransportConnection server, - ) async { - await testWindowSize(client, server, 8096); - }, - clientSettings: const ClientSettings(streamWindowSize: 8096), - ); + transportTest('fast-sender-receiver-paused--10kb-window-size', ( + ClientTransportConnection client, + ServerTransportConnection server, + ) async { + await testWindowSize(client, server, 8096); + }, clientSettings: const ClientSettings(streamWindowSize: 8096)); }); }); }