diff --git a/proxygen/lib/http/codec/HTTPBinaryCodec.cpp b/proxygen/lib/http/codec/HTTPBinaryCodec.cpp index a9d031b59..1fc9db618 100644 --- a/proxygen/lib/http/codec/HTTPBinaryCodec.cpp +++ b/proxygen/lib/http/codec/HTTPBinaryCodec.cpp @@ -19,6 +19,7 @@ break; \ } else if ((parseResult).parseResultState_ == \ ParseResultState::WAITING_FOR_MORE_DATA) { \ + parsed += (parseResult).bytesParsed_; \ parserWaitingForMoreData_ = true; \ break; \ } @@ -221,17 +222,20 @@ ParseResult HTTPBinaryCodec::parseSingleHeaderHelper( size_t& remaining, uint64_t maxLength, size_t& numHeaders) { - std::string headerName; - auto headerNameRes = parseKnownLengthString( - cursor, remaining, maxLength, "headerName", headerName); - if (headerNameRes.parseResultState_ == ParseResultState::ERROR || - headerNameRes.parseResultState_ == - ParseResultState::WAITING_FOR_MORE_DATA) { - return headerNameRes; + if (!pendingFieldName_) { + std::string headerName; + auto headerNameRes = parseKnownLengthString( + cursor, remaining, maxLength, "headerName", headerName); + if (headerNameRes.parseResultState_ == ParseResultState::ERROR || + headerNameRes.parseResultState_ == + ParseResultState::WAITING_FOR_MORE_DATA) { + return headerNameRes; + } + parsed += headerNameRes.bytesParsed_; + remaining -= headerNameRes.bytesParsed_; + maxLength -= std::min(maxLength, headerNameRes.bytesParsed_); + pendingFieldName_.emplace(std::move(headerName)); } - parsed += headerNameRes.bytesParsed_; - remaining -= headerNameRes.bytesParsed_; - maxLength -= std::min(maxLength, headerNameRes.bytesParsed_); std::string headerValue; auto headerValueRes = parseKnownLengthString( @@ -244,6 +248,9 @@ ParseResult HTTPBinaryCodec::parseSingleHeaderHelper( parsed += headerValueRes.bytesParsed_; remaining -= headerValueRes.bytesParsed_; + CHECK(pendingFieldName_); + auto headerName = std::move(*pendingFieldName_); + pendingFieldName_.reset(); if (!decodeInfo.onHeader(proxygen::HPACKHeaderName(headerName), headerValue) || !decodeInfo.parsingError.empty()) { @@ -287,20 +294,31 @@ ParseResult HTTPBinaryCodec::parseKnownLengthHeadersHelper( return ParseResult(ParseResultState::WAITING_FOR_MORE_DATA); } + const auto fieldSectionLength = static_cast(lengthOfHeaders->first); + folly::io::Cursor fieldSectionCursor(cursor, fieldSectionLength); + size_t fieldSectionParsed = 0; + size_t fieldSectionRemaining = fieldSectionLength; size_t numHeaders = 0; - while (parsed < lengthOfHeaders->first) { - auto result = parseSingleHeaderHelper(cursor, + while (fieldSectionParsed < fieldSectionLength) { + auto result = parseSingleHeaderHelper(fieldSectionCursor, decodeInfo, - parsed, - remaining, - maxFieldSectionSize_, + fieldSectionParsed, + fieldSectionRemaining, + fieldSectionRemaining, numHeaders); - if (result.parseResultState_ == ParseResultState::ERROR || - result.parseResultState_ == ParseResultState::WAITING_FOR_MORE_DATA) { + if (result.parseResultState_ == ParseResultState::ERROR) { return result; } + if (result.parseResultState_ == ParseResultState::WAITING_FOR_MORE_DATA) { + return ParseResult( + fmt::format("{} field exceeds declared section length of {}", + isTrailers ? "Trailer" : "Header", + lengthOfHeaders->first)); + } } + cursor.skip(fieldSectionLength); + parsed += fieldSectionParsed; return ParseResult(parsed); } @@ -314,21 +332,28 @@ ParseResult HTTPBinaryCodec::parseIndeterminateLengthHeadersHelper( auto currentByte = cursor.peek().data(); size_t numHeaders = 0; // Continue parsing headers until we reach the Content Terminator field (0) - while (currentByte != nullptr && *currentByte != 0x00) { + while (currentByte != nullptr && + (pendingFieldName_ || *currentByte != 0x00)) { // A field whose declared length cannot fit in what is left of the section // has to be rejected on sight: waiting for it to arrive would retain the // fields already parsed plus the whole declaration const uint64_t sectionBudget = - maxFieldSectionSize_ - std::min(parsed, maxFieldSectionSize_); + maxFieldSectionSize_ - + std::min(indeterminateFieldSectionBytesParsed_ + parsed, + maxFieldSectionSize_); auto result = parseSingleHeaderHelper( cursor, decodeInfo, parsed, remaining, sectionBudget, numHeaders); - if (result.parseResultState_ == ParseResultState::ERROR || - result.parseResultState_ == ParseResultState::WAITING_FOR_MORE_DATA) { + if (result.parseResultState_ == ParseResultState::ERROR) { return result; } + if (result.parseResultState_ == ParseResultState::WAITING_FOR_MORE_DATA) { + indeterminateFieldSectionBytesParsed_ += parsed; + return ParseResult(parsed, ParseResultState::WAITING_FOR_MORE_DATA); + } // There is no declared length to check against, so bound the section by // what has actually been seen - if (parsed > maxFieldSectionSize_) { + if (indeterminateFieldSectionBytesParsed_ > maxFieldSectionSize_ || + parsed > maxFieldSectionSize_ - indeterminateFieldSectionBytesParsed_) { return ParseResult( fmt::format("Unterminated {} section exceeds the maximum field " "section size of {}", @@ -339,7 +364,8 @@ ParseResult HTTPBinaryCodec::parseIndeterminateLengthHeadersHelper( // waiting for more data since we haven't seen a Content Terminator field // yet if (cursor.isAtEnd()) { - return ParseResult(ParseResultState::WAITING_FOR_MORE_DATA); + indeterminateFieldSectionBytesParsed_ += parsed; + return ParseResult(parsed, ParseResultState::WAITING_FOR_MORE_DATA); } currentByte = cursor.peek().data(); } @@ -347,6 +373,7 @@ ParseResult HTTPBinaryCodec::parseIndeterminateLengthHeadersHelper( parsed++; remaining--; cursor.skip(1); + indeterminateFieldSectionBytesParsed_ = 0; return ParseResult(parsed); } @@ -613,7 +640,8 @@ size_t HTTPBinaryCodec::onIngress(const folly::IOBuf& buf) { void HTTPBinaryCodec::onIngressEOF() { if (!parseError_ && - (!bufferedIngress_.empty() || remainingContentLength_.value_or(0) > 0)) { + (!bufferedIngress_.empty() || remainingContentLength_.value_or(0) > 0 || + indeterminateFieldSectionBytesParsed_ > 0)) { // Case where the ingress EOF is received before the entire message is // parsed callback_->onError(ingressTxnID_, @@ -632,7 +660,7 @@ void HTTPBinaryCodec::onIngressEOF() { "contains framing indicator")); return; } - if (state_ == ParseState::HEADERS_SECTION) { + if (!parseError_ && state_ == ParseState::HEADERS_SECTION) { // Case where the sent message only contains control data and no headers // nor body callback_->onHeadersComplete(ingressTxnID_, std::move(decodeInfo_.msg)); diff --git a/proxygen/lib/http/codec/HTTPBinaryCodec.h b/proxygen/lib/http/codec/HTTPBinaryCodec.h index c1f061504..ea5106216 100644 --- a/proxygen/lib/http/codec/HTTPBinaryCodec.h +++ b/proxygen/lib/http/codec/HTTPBinaryCodec.h @@ -236,6 +236,9 @@ class HTTPBinaryCodec : public HTTPCodec { bool parserWaitingForMoreData_{false}; bool parserPaused_; folly::Optional parseError_{folly::none}; + // Complete field bytes trimmed from the current indeterminate section. + size_t indeterminateFieldSectionBytesParsed_{0}; + folly::Optional pendingFieldName_{folly::none}; enum class FramingIndicator : uint8_t { REQUEST_KNOWN_LENGTH = 0, diff --git a/proxygen/lib/http/codec/test/HTTPBinaryCodecTest.cpp b/proxygen/lib/http/codec/test/HTTPBinaryCodecTest.cpp index ffb157ed4..3ba9dce6a 100644 --- a/proxygen/lib/http/codec/test/HTTPBinaryCodecTest.cpp +++ b/proxygen/lib/http/codec/test/HTTPBinaryCodecTest.cpp @@ -58,9 +58,17 @@ class HTTPBinaryCodecForTest : public HTTPBinaryCodec { return *msgBody_; } + size_t getPendingFieldCount() const { + return decodeInfo_.msg ? decodeInfo_.msg->getHeaders().size() : 0; + } + size_t getBufferedIngressSize() const { return bufferedIngress_.chainLength(); } + + const std::string* getPendingFieldName() const { + return pendingFieldName_ ? &*pendingFieldName_ : nullptr; + } }; namespace { @@ -89,6 +97,19 @@ void writeRequestPreamble(folly::io::QueueAppender& appender, writeVarintString(appender, "/"); } +void writeOverrunningKnownLengthFieldSection( + folly::io::QueueAppender& appender) { + folly::IOBufQueue fieldSection{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender fieldAppender(&fieldSection, 64); + writeVarintString(fieldAppender, "x-first"); + writeVarintString(fieldAppender, "first-value"); + writeVarintString(fieldAppender, "x-last"); + writeVarint(fieldAppender, 64); + + writeVarint(appender, fieldSection.chainLength()); + appender.insert(fieldSection.move()); +} + std::unique_ptr makeFiller(size_t size) { auto buf = folly::IOBuf::create(size); buf->append(size); @@ -359,11 +380,12 @@ TEST_F(HttpBinaryUpstreamCodecTest, testParseKnownLengthHeadersEmpty) { .bytesParsed_, 1); } -TEST_F(HttpBinaryUpstreamCodecTest, testParseKnownLengthHeadersWaiting) { +TEST_F(HttpBinaryUpstreamCodecTest, + testParseKnownLengthHeadersOverrunsDeclaredSection) { // Format is `**user-agent*curl/7.16.3 libcurl/7.16.3 OpenSSL/0.9.7l // zlib/1.2.3*host*www.example.com*accept-language.en, mi` where the `*` after - // accept-language is value 7 instead of 6 which should cause parsing to wait - // to get more data + // accept-language is value 7 instead of 6, which exceeds the declared + // section boundary. const std::vector invalidHeadersLength{ 0x40, 0x6c, 0x0a, 0x75, 0x73, 0x65, 0x72, 0x2d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x34, 0x63, 0x75, 0x72, 0x6c, 0x2f, 0x37, 0x2e, 0x31, @@ -391,7 +413,7 @@ TEST_F(HttpBinaryUpstreamCodecTest, testParseKnownLengthHeadersWaiting) { decodeInfo, /*knownLength=*/true) .parseResultState_, - ParseResultState::WAITING_FOR_MORE_DATA); + ParseResultState::ERROR); // Format is `**a*b` where the first `*` represents a too long length const std::vector invalidHeadersUnderflow{ @@ -796,7 +818,9 @@ TEST_F(HttpBinaryDownstreamCodecTest, testOnIngressFailureMalformedMessage) { // Check onError was called with the correct error EXPECT_EQ(std::string(callback.lastParseError.get()->what()), - "Incomplete message received"); + "Invalid Message: Header field exceeds declared section length of " + "108"); + EXPECT_EQ(callback.headersComplete, 0); } TEST_F(HttpBinaryDownstreamCodecTest, testOnIngressFailureIncompleteMessage) { @@ -1240,6 +1264,245 @@ TEST_F(HttpBinaryDownstreamCodecTest, testPartialChunkIsNotOverCounted) { EXPECT_EQ(callback.data_.move()->to(), first + second); } +TEST_F(HttpBinaryDownstreamCodecTest, testSplitHeaderFieldIsDecodedOnce) { + const std::string splitValue = "split-value"; + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 256); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, "x-first"); + writeVarintString(appender, "first-value"); + writeVarintString(appender, "x-split"); + writeVarint(appender, splitValue.size()); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + + for (size_t i = 0; i < splitValue.size(); i++) { + auto chunk = folly::IOBuf::copyBuffer(splitValue.data() + i, 1); + binaryCodecIndeterminateLength_->onIngress(*chunk); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), + i + 1 == splitValue.size() ? 2 : 1); + } + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + folly::IOBufQueue terminators{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender terminatorAppender(&terminators, 4); + writeVarint(terminatorAppender, 0 /* field section terminator */); + writeVarint(terminatorAppender, 0 /* empty content field */); + writeVarint(terminatorAppender, 0 /* content terminator */); + writeVarint(terminatorAppender, 0 /* trailer section terminator */); + binaryCodecIndeterminateLength_->onIngress(*terminators.front()); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().size(), 2); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty("x-first"), + "first-value"); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty("x-split"), splitValue); +} + +TEST_F(HttpBinaryDownstreamCodecTest, testSplitTrailerFieldIsDecodedOnce) { + const std::string splitValue = "split-value"; + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 256); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarint(appender, 0 /* field section terminator */); + writeVarint(appender, 0 /* empty content field */); + writeVarint(appender, 0 /* content terminator */); + writeVarintString(appender, "x-first"); + writeVarintString(appender, "first-value"); + writeVarintString(appender, "x-split"); + writeVarint(appender, splitValue.size()); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + + for (size_t i = 0; i < splitValue.size(); i++) { + auto chunk = folly::IOBuf::copyBuffer(splitValue.data() + i, 1); + binaryCodecIndeterminateLength_->onIngress(*chunk); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), + i + 1 == splitValue.size() ? 2 : 1); + } + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + auto terminator = folly::IOBuf::copyBuffer("\0", 1); + binaryCodecIndeterminateLength_->onIngress(*terminator); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + ASSERT_NE(callback.msg->getTrailers(), nullptr); + EXPECT_EQ(callback.msg->getTrailers()->size(), 2); + EXPECT_EQ(callback.msg->getTrailers()->getSingleOrEmpty("x-first"), + "first-value"); + EXPECT_EQ(callback.msg->getTrailers()->getSingleOrEmpty("x-split"), + splitValue); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testSplitHeaderLengthVarintIsDecodedOnce) { + const std::string splitValue(64, 'a'); + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 256); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, "x-first"); + writeVarintString(appender, "first-value"); + writeVarintString(appender, "x-split"); + writeVarint(appender, splitValue.size()); + + auto head = message.split(message.chainLength() - 1); + auto lengthSuffix = message.move(); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*head); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + + binaryCodecIndeterminateLength_->onIngress(*lengthSuffix); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + + auto value = folly::IOBuf::copyBuffer(splitValue.data(), splitValue.size()); + binaryCodecIndeterminateLength_->onIngress(*value); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 2); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + auto terminators = folly::IOBuf::copyBuffer("\0\0\0\0", 4); + binaryCodecIndeterminateLength_->onIngress(*terminators); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty("x-split"), splitValue); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testBytePacedLargeFieldNameIsTrimmedOnce) { + constexpr size_t kOhttpFieldSectionSize = size_t{64} * 1024; + constexpr size_t kFieldPartSize = kOhttpFieldSectionSize / 2 - 16; + const std::string headerName(kFieldPartSize, 'x'); + const std::string headerValue(kFieldPartSize, 'v'); + + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 256); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarint(appender, headerName.size()); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setMaxFieldSectionSize( + kOhttpFieldSectionSize); + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + const auto sendByte = [&](const std::string& input, size_t offset) { + auto chunk = folly::IOBuf::copyBuffer(input.data() + offset, 1); + binaryCodecIndeterminateLength_->onIngress(*chunk); + }; + for (size_t i = 0; i < headerName.size(); i++) { + sendByte(headerName, i); + } + + const auto* pendingFieldName = + binaryCodecIndeterminateLength_->getPendingFieldName(); + ASSERT_NE(pendingFieldName, nullptr); + EXPECT_EQ(*pendingFieldName, headerName); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 0); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + folly::IOBufQueue valueLength{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender valueLengthAppender(&valueLength, 8); + writeVarint(valueLengthAppender, headerValue.size()); + const auto valueLengthSize = valueLength.chainLength(); + binaryCodecIndeterminateLength_->onIngress(*valueLength.front()); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), + valueLengthSize); + + for (size_t i = 0; i + 1 < headerValue.size(); i++) { + sendByte(headerValue, i); + } + pendingFieldName = binaryCodecIndeterminateLength_->getPendingFieldName(); + ASSERT_NE(pendingFieldName, nullptr); + EXPECT_EQ(*pendingFieldName, headerName); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), + valueLengthSize + headerValue.size() - 1); + sendByte(headerValue, headerValue.size() - 1); + + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldName(), nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + auto terminators = folly::IOBuf::copyBuffer("\0\0\0\0", 4); + binaryCodecIndeterminateLength_->onIngress(*terminators); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty(headerName), + headerValue); +} + +TEST_F(HttpBinaryDownstreamCodecTest, testCachedFieldNameAllowsEmptyValue) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, "x-empty"); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + ASSERT_NE(binaryCodecIndeterminateLength_->getPendingFieldName(), nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + auto emptyValueAndTerminators = folly::IOBuf::copyBuffer("\0\0\0\0\0", 5); + binaryCodecIndeterminateLength_->onIngress(*emptyValueAndTerminators); + + EXPECT_EQ(callback.lastParseError, nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldName(), nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().size(), 1); + EXPECT_TRUE(callback.msg->getHeaders().exists("x-empty")); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty("x-empty"), ""); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testKnownLengthHeaderFieldCannotExceedDeclaredSection) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 0 /* request, known length */); + writeOverrunningKnownLengthFieldSection(appender); + + FakeHTTPCodecCallback callback; + binaryCodecKnownLength_->setCallback(&callback); + binaryCodecKnownLength_->onIngress(*message.front()); + + ASSERT_NE(callback.lastParseError, nullptr); + EXPECT_EQ(std::string(callback.lastParseError->what()), + "Invalid Message: Failure to parse: headerValue of declared length " + "64 exceeds the remaining field section budget of 2"); + EXPECT_EQ(binaryCodecKnownLength_->getPendingFieldCount(), 1); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testKnownLengthTrailerFieldCannotExceedDeclaredSection) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 0 /* request, known length */); + writeVarint(appender, 0 /* empty header section */); + writeVarint(appender, 0 /* empty content */); + writeOverrunningKnownLengthFieldSection(appender); + + FakeHTTPCodecCallback callback; + binaryCodecKnownLength_->setCallback(&callback); + binaryCodecKnownLength_->onIngress(*message.front()); + + ASSERT_NE(callback.lastParseError, nullptr); + EXPECT_EQ(std::string(callback.lastParseError->what()), + "Invalid Message: Failure to parse: headerValue of declared length " + "64 exceeds the remaining field section budget of 2"); + EXPECT_EQ(binaryCodecKnownLength_->getPendingFieldCount(), 1); +} + // Field section lengths are QUIC varints, so a peer can declare up to 2^62-1 // bytes. The declaration must be rejected up front instead of buffering // ingress until the section is complete. @@ -1306,15 +1569,54 @@ TEST_F(HttpBinaryDownstreamCodecTest, writeVarintString(fieldsAppender, "x-pad"); writeVarintString(fieldsAppender, padValue); } + constexpr size_t kSuccessfulBatchCount = 2; + binaryCodecIndeterminateLength_->setMaxFieldSectionSize( + fields.chainLength() * kSuccessfulBatchCount); auto fieldsBuf = fields.move(); - for (size_t i = 0; i < kFillerChunkCount && !callback.lastParseError; i++) { + for (size_t i = 0; i < kSuccessfulBatchCount; i++) { binaryCodecIndeterminateLength_->onIngress(*fieldsBuf); + ASSERT_EQ(callback.lastParseError, nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); } + binaryCodecIndeterminateLength_->onIngress(*fieldsBuf); EXPECT_NE(callback.lastParseError, nullptr); } +TEST_F(HttpBinaryDownstreamCodecTest, + testIndeterminateFieldSectionLimitResetsForTrailers) { + const std::string fieldName = "x-near-cap"; + const std::string fieldValue(128, 'a'); + const auto fieldSize = + quic::getQuicIntegerSize(fieldName.size()).value() + fieldName.size() + + quic::getQuicIntegerSize(fieldValue.size()).value() + fieldValue.size(); + + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 512); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, fieldName); + writeVarintString(appender, fieldValue); + writeVarint(appender, 0 /* field section terminator */); + writeVarint(appender, 0 /* empty content field */); + writeVarint(appender, 0 /* content terminator */); + writeVarintString(appender, fieldName); + writeVarintString(appender, fieldValue); + writeVarint(appender, 0 /* trailer section terminator */); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setMaxFieldSectionSize(fieldSize); + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + EXPECT_EQ(callback.lastParseError, nullptr); + ASSERT_NE(callback.msg, nullptr); + EXPECT_EQ(callback.msg->getHeaders().getSingleOrEmpty(fieldName), fieldValue); + ASSERT_NE(callback.msg->getTrailers(), nullptr); + EXPECT_EQ(callback.msg->getTrailers()->getSingleOrEmpty(fieldName), + fieldValue); +} + TEST_F(HttpBinaryDownstreamCodecTest, testMaxFieldSectionSizeIsConfigurable) { const std::string headerValue(256, 'a'); folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; @@ -1432,4 +1734,53 @@ TEST_F(HttpBinaryDownstreamCodecTest, testTruncatedContentIsIncompleteAtEOF) { "Incomplete message received"); } +TEST_F(HttpBinaryDownstreamCodecTest, + testTrimmedFieldWithoutTerminatorIsIncompleteAtEOF) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarintString(appender, "x-complete"); + writeVarintString(appender, "value"); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + EXPECT_EQ(callback.lastParseError, nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + binaryCodecIndeterminateLength_->onIngressEOF(); + + ASSERT_NE(callback.lastParseError, nullptr); + EXPECT_EQ(std::string(callback.lastParseError->what()), + "Incomplete message received"); +} + +TEST_F(HttpBinaryDownstreamCodecTest, + testTrimmedTrailerWithoutTerminatorIsIncompleteAtEOF) { + folly::IOBufQueue message{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender(&message, 128); + writeRequestPreamble(appender, 2 /* request, indeterminate length */); + writeVarint(appender, 0 /* field section terminator */); + writeVarint(appender, 0 /* empty content field */); + writeVarint(appender, 0 /* content terminator */); + writeVarintString(appender, "x-complete"); + writeVarintString(appender, "value"); + + FakeHTTPCodecCallback callback; + binaryCodecIndeterminateLength_->setCallback(&callback); + binaryCodecIndeterminateLength_->onIngress(*message.front()); + + EXPECT_EQ(callback.lastParseError, nullptr); + EXPECT_EQ(binaryCodecIndeterminateLength_->getPendingFieldCount(), 1); + EXPECT_EQ(binaryCodecIndeterminateLength_->getBufferedIngressSize(), 0); + + binaryCodecIndeterminateLength_->onIngressEOF(); + + ASSERT_NE(callback.lastParseError, nullptr); + EXPECT_EQ(std::string(callback.lastParseError->what()), + "Incomplete message received"); +} + } // namespace proxygen::test diff --git a/proxygen/lib/http/session/HQSession.cpp b/proxygen/lib/http/session/HQSession.cpp index a067f40a6..18fcca211 100644 --- a/proxygen/lib/http/session/HQSession.cpp +++ b/proxygen/lib/http/session/HQSession.cpp @@ -3594,12 +3594,19 @@ void HQSession::deliverWTStream(HQStreamTransportBase* connectStream, return; } connectStream->txn_.getWebTransport(); // lazy-init webTransportImpl_ + // The handler runs below and can end the txn, freeing the handle. + HTTPTransaction::DestructorGuard dg(&connectStream->txn_); + WebTransportImpl::StreamReadHandle* readHandle = nullptr; if (sock_->isBidirectionalStream(streamID)) { - auto handle = connectStream->txn_.onWebTransportBidiStream(streamID); - sock_->setReadCallback(streamID, handle.readHandle); + readHandle = + connectStream->txn_.onWebTransportBidiStream(streamID).readHandle; } else { - auto* handle = connectStream->txn_.onWebTransportUniStream(streamID); - sock_->setReadCallback(streamID, handle); + readHandle = connectStream->txn_.onWebTransportUniStream(streamID); + } + // No handle means the handler ended the session; an unregistered one means + // it called stopSending. + if (readHandle && readHandle->readCallbackRegistered()) { + sock_->setReadCallback(streamID, readHandle); } } @@ -3676,7 +3683,16 @@ void HQSession::deliverPendingWTStreams(quic::StreamId sessionID) noexcept { VLOG(4) << "delivering " << entry->streams.size() << " buffered WT streams wt-sess-id=" << sessionID << " sess=" << *this; + // The handler may end the CONNECT partway through the loop. The guard keeps + // connectStream alive so the streams behind it can be rejected. + HTTPTransaction::DestructorGuard dg(&connectStream->txn_); for (auto id : entry->streams) { + if (connectStream->txn_.isIngressComplete()) { + VLOG(4) << "WT connect stream ended mid-drain, rejecting id=" << id + << " wt-sess-id=" << sessionID << " sess=" << *this; + rejectStream(id, kBufferedRejectedErr); + continue; + } deliverWTStream(connectStream, id); } } @@ -4037,7 +4053,15 @@ HQSession::HQStreamTransport::stopReadingWebTransportIngress( } auto res = session_.sock_->setReadCallback(id, nullptr, quicErrorCode); if (res.hasError()) { - return folly::makeUnexpected(WebTransport::ErrorCode::GENERIC_ERROR); + // setReadCallback only sends STOP_SENDING if a read callback is + // installed, which it isn't until deliverWTStream runs. + if (!quicErrorCode) { + return folly::makeUnexpected(WebTransport::ErrorCode::GENERIC_ERROR); + } + auto ssRes = session_.sock_->stopSending(id, *quicErrorCode); + if (ssRes.hasError()) { + return folly::makeUnexpected(WebTransport::ErrorCode::GENERIC_ERROR); + } } } return folly::unit; diff --git a/proxygen/lib/http/session/HTTPTransaction.cpp b/proxygen/lib/http/session/HTTPTransaction.cpp index d20e9fcb5..542310d7b 100644 --- a/proxygen/lib/http/session/HTTPTransaction.cpp +++ b/proxygen/lib/http/session/HTTPTransaction.cpp @@ -1825,12 +1825,19 @@ WebTransportImpl::BidiStreamHandle HTTPTransaction::onWebTransportBidiStream( id, WebTransport::kInternalError); return {.readHandle = nullptr, .writeHandle = nullptr}; } + // The handler may close the session or abort the txn, freeing the handles + // and possibly this. + DestructorGuard dg(this); refreshTimeout(); auto handle = webTransportImpl_->onWebTransportBidiStream(id); handler_->onWebTransportBidiStream( id, {.readHandle = handle.readHandle, .writeHandle = handle.writeHandle}); - // what if the handler killed handle (stop sending/fin/rst)? - return handle; + if (!handler_ || isIngressComplete()) { + // The handler ended the txn; dg's release frees the handles with it. + return {.readHandle = nullptr, .writeHandle = nullptr}; + } + return {.readHandle = webTransportImpl_->getReadHandle(id), + .writeHandle = webTransportImpl_->getWriteHandle(id)}; } WebTransportImpl::StreamReadHandle* HTTPTransaction::onWebTransportUniStream( @@ -1840,12 +1847,16 @@ WebTransportImpl::StreamReadHandle* HTTPTransaction::onWebTransportUniStream( id, WebTransport::kInternalError); return nullptr; } + // See onWebTransportBidiStream. + DestructorGuard dg(this); refreshTimeout(); auto handle = webTransportImpl_->onWebTransportUniStream(id); handler_->onWebTransportUniStream(id, handle); - // what if the handler killed handle (stop sending)? - return handle; + if (!handler_ || isIngressComplete()) { + return nullptr; + } + return webTransportImpl_->getReadHandle(id); } bool HTTPTransaction::onWebTransportStopSending(HTTPCodec::StreamID id, diff --git a/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp b/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp index 995f17e1e..8e1646f22 100644 --- a/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp +++ b/proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp @@ -3940,6 +3940,99 @@ TEST_P(HQDownstreamSessionTestWebTransport, BufferedWTStreamRejectedOnClose) { EXPECT_EQ(*socketDriver_->streams_[wtStreamId].error, bufferedRejectedErr); } +// The handler aborts the CONNECT on the first buffered stream, which would +// destroy the stream the drain loop is walking. The rest get rejected. +TEST_P(HQDownstreamSessionTestWebTransport, + BufferedWTStreamConnectAbortedDuringDrain) { + const quic::StreamId sessionId = 0; + const quic::StreamId wtStreamId1 = 14; + const quic::StreamId wtStreamId2 = 18; + + sendSettings(); + queuePeerWTUniStream(socketDriver_.get(), wtStreamId1, sessionId); + queuePeerWTUniStream(socketDriver_.get(), wtStreamId2, sessionId); + flushRequestsAndLoopN(1); + EXPECT_FALSE(socketDriver_->streams_[wtStreamId1].error.has_value()); + EXPECT_FALSE(socketDriver_->streams_[wtStreamId2].error.has_value()); + + HTTPMessage req; + req.setHTTPVersion(1, 1); + req.setUpgradeProtocol("webtransport"); + req.setMethod(HTTPMethod::CONNECT); + req.setURL("/webtransport"); + req.getHeaders().set(HTTP_HEADER_HOST, "www.facebook.com"); + auto actualSessionId = sendRequest(req, /*eom=*/false); + ASSERT_EQ(actualSessionId, sessionId); + + auto handler = addSimpleStrictHandler(); + handler->expectHeaders([&]() { + HTTPMessage resp; + resp.setStatusCode(200); + handler->txn_->sendHeaders(resp); + EXPECT_NE(handler->txn_->getWebTransport(), nullptr); + }); + // Only the first buffered stream reaches the handler. + EXPECT_CALL(*handler, onWebTransportUniStream(_, _)) + .Times(1) + .WillOnce([&](HTTPCodec::StreamID, WebTransport::StreamReadHandle*) { + handler->txn_->sendAbort(); + }); + handler->expectDetachTransaction(); + flushRequestsAndLoopN(3); + + const uint64_t bufferedRejectedErr = + WebTransport::toHTTPErrorCode(WebTransport::kBufferedStreamRejected); + ASSERT_TRUE(socketDriver_->streams_[wtStreamId2].error.has_value()); + EXPECT_EQ(*socketDriver_->streams_[wtStreamId2].error, bufferedRejectedErr); + + hqSession_->dropConnection(); + flushRequestsAndLoop(); +} + +// The handler stops reading from inside onWebTransportUniStream, so +// deliverWTStream must not turn around and register the handle with quic. +TEST_P(HQDownstreamSessionTestWebTransport, StopSendingFromOnNewUniStream) { + const quic::StreamId sessionId = 0; + const quic::StreamId wtStreamId = 14; + + sendSettings(); + HTTPMessage req; + req.setHTTPVersion(1, 1); + req.setUpgradeProtocol("webtransport"); + req.setMethod(HTTPMethod::CONNECT); + req.setURL("/webtransport"); + req.getHeaders().set(HTTP_HEADER_HOST, "www.facebook.com"); + auto actualSessionId = sendRequest(req, /*eom=*/false); + ASSERT_EQ(actualSessionId, sessionId); + + auto handler = addSimpleStrictHandler(); + handler->expectHeaders([&]() { + HTTPMessage resp; + resp.setStatusCode(200); + handler->txn_->sendHeaders(resp); + EXPECT_NE(handler->txn_->getWebTransport(), nullptr); + }); + flushRequestsAndLoopN(3); + + EXPECT_CALL(*handler, onWebTransportUniStream(wtStreamId, _)) + .WillOnce( + [](HTTPCodec::StreamID, WebTransport::StreamReadHandle* readHandle) { + readHandle->stopSending(WebTransport::kInternalError); + }); + queuePeerWTUniStream(socketDriver_.get(), wtStreamId, sessionId); + flushRequestsAndLoopN(2); + + // The handle dies with the session, so it must not be left registered. The + // STOP_SENDING still has to go out, via QuicSocket::stopSending(). + EXPECT_EQ(socketDriver_->streams_[wtStreamId].readCB, nullptr); + EXPECT_TRUE(socketDriver_->streams_[wtStreamId].error.has_value()); + + handler->expectDetachTransaction(); + handler->txn_->sendAbort(); + hqSession_->dropConnection(); + flushRequestsAndLoop(); +} + INSTANTIATE_TEST_SUITE_P(HQDownstreamSessionTest, HQDownstreamSessionTestWebTransport, Values([] { diff --git a/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp b/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp index b9125b615..55c51c908 100644 --- a/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp +++ b/proxygen/lib/http/session/test/HTTPTransactionWebTransportTest.cpp @@ -512,7 +512,8 @@ TEST_F(HTTPTransactionWebTransportTest, BidiStreamEdgeCases) { // it gets stopReadingWebTransportIngress when the EOF is read out EXPECT_CALL(transport_, - stopReadingWebTransportIngress(0, folly::Optional())); + stopReadingWebTransportIngress(0, folly::Optional())) + .WillOnce(Return(folly::unit)); EXPECT_CALL(transport_, sendWTMaxData(kDefaultWTReceiveWindow)).Times(0); auto fut = streamHandle.readHandle->readStreamData() @@ -1393,4 +1394,112 @@ TEST_F(HTTPTransactionWebTransportTest, FrequentFlowControlGrants) { .WillRepeatedly(Return(folly::unit)); } +// The handler closes the session, freeing the handle we just created. +// Returning it would leave deliverWTStream registering freed memory with quic. +TEST_F(HTTPTransactionWebTransportTest, CloseSessionFromOnNewUniStream) { + auto* wtImpl = dynamic_cast(wt_); + ASSERT_NE(wtImpl, nullptr); + auto* txn = txn_.get(); + // closeSession() may detach the txn, so keep it alive for the assertions. + HTTPTransaction::DestructorGuard dg(txn); + + EXPECT_CALL(handler_, onWebTransportUniStream(_, _)) + .WillOnce( + [this](HTTPCodec::StreamID, WebTransport::StreamReadHandle* handle) { + EXPECT_NE(handle, nullptr); + EXPECT_CALL(transport_, sendEOM(txn_.get(), nullptr)); + EXPECT_CALL(transport_, sendAbort(txn_.get(), _)); + wt_->closeSession(); + }); + EXPECT_CALL( + transport_, + stopReadingWebTransportIngress(2, makeOpt(WebTransport::kSessionGone))) + .WillRepeatedly(Return(folly::unit)); + + auto* handle = txn->onWebTransportUniStream(2); + + EXPECT_EQ(handle, nullptr); + EXPECT_EQ(wtImpl->getReadHandle(2), nullptr); + EXPECT_EQ(wtImpl->readStreamData(2).error(), + WebTransport::ErrorCode::INVALID_STREAM_ID); +} + +// Same hole on the bidi path. +TEST_F(HTTPTransactionWebTransportTest, CloseSessionFromOnNewBidiStream) { + auto* wtImpl = dynamic_cast(wt_); + ASSERT_NE(wtImpl, nullptr); + auto* txn = txn_.get(); + HTTPTransaction::DestructorGuard dg(txn); + + EXPECT_CALL(handler_, onWebTransportBidiStream(_, _)) + .WillOnce( + [this](HTTPCodec::StreamID, WebTransport::BidiStreamHandle bidi) { + EXPECT_NE(bidi.readHandle, nullptr); + EXPECT_CALL(transport_, sendEOM(txn_.get(), nullptr)); + EXPECT_CALL(transport_, sendAbort(txn_.get(), _)); + wt_->closeSession(); + }); + EXPECT_CALL( + transport_, + stopReadingWebTransportIngress(1, makeOpt(WebTransport::kSessionGone))) + .WillRepeatedly(Return(folly::unit)); + EXPECT_CALL(transport_, + resetWebTransportEgress(1, WebTransport::kSessionGone)) + .WillRepeatedly(Return(folly::unit)); + + auto bidi = txn->onWebTransportBidiStream(1); + + EXPECT_EQ(bidi.readHandle, nullptr); + EXPECT_EQ(bidi.writeHandle, nullptr); + EXPECT_EQ(wtImpl->getReadHandle(1), nullptr); +} + +// A stream that buffered data + FIN is no longer open(), but its read callback +// is still registered. Teardown has to unregister it before freeing it. +TEST_F(HTTPTransactionWebTransportTest, TerminateSessionUnregistersEofStream) { + WebTransport::StreamReadHandle* readHandle{nullptr}; + EXPECT_CALL(handler_, onWebTransportUniStream(_, _)) + .WillOnce(SaveArg<1>(&readHandle)); + + auto* implHandle = txn_->onWebTransportUniStream(2); + ASSERT_NE(implHandle, nullptr); + ASSERT_NE(readHandle, nullptr); + + // Data + FIN arrive with no readStreamData() outstanding. + EXPECT_CALL(transport_, readWebTransportData(2, 65535)) + .WillOnce([](auto, auto) { return std::make_pair(makeBuf(10), true); }); + implHandle->readAvailable(2); + EXPECT_FALSE(implHandle->open()); + + EXPECT_CALL(transport_, stopReadingWebTransportIngress(2, _)) + .Times(1) + .WillOnce(Return(folly::unit)); + EXPECT_CALL(transport_, sendEOM(txn_.get(), nullptr)); + EXPECT_CALL(transport_, sendAbort(txn_.get(), _)); + wt_->closeSession(); +} + +// Aborting the txn leaves the stream in wtIngressStreams_, so the re-lookup +// finds a handle that releasing the guard frees a moment later. +TEST_F(HTTPTransactionWebTransportTest, AbortTxnFromOnNewUniStream) { + auto* txn = txn_.get(); + + EXPECT_CALL(handler_, onWebTransportUniStream(_, _)) + .WillOnce( + [this](HTTPCodec::StreamID, WebTransport::StreamReadHandle* handle) { + EXPECT_NE(handle, nullptr); + // Completes ingress and egress: the txn dies with the last guard. + EXPECT_CALL(transport_, sendAbort(txn_.get(), _)); + txn_->sendAbort(); + }); + EXPECT_CALL(transport_, stopReadingWebTransportIngress(2, _)) + .WillRepeatedly(Return(folly::unit)); + + auto* handle = txn->onWebTransportUniStream(2); + + // The txn, webTransportImpl_ and the handle are all gone by now. + EXPECT_EQ(txn_, nullptr); + EXPECT_EQ(handle, nullptr); +} + } // namespace proxygen::test diff --git a/proxygen/lib/http/webtransport/WebTransportImpl.cpp b/proxygen/lib/http/webtransport/WebTransportImpl.cpp index 7c75d39c6..00a80154d 100644 --- a/proxygen/lib/http/webtransport/WebTransportImpl.cpp +++ b/proxygen/lib/http/webtransport/WebTransportImpl.cpp @@ -40,12 +40,19 @@ void WebTransportImpl::terminateSessionStreams(uint32_t errorCode, for (auto& [id, stream] : wtIngressStreams) { // Deliver an error to the application if needed VLOG(4) << "aborting wt ingress id=" << id << " err=" << errorCode - << "; open=" << int(stream.open()); + << "; open=" << int(stream.open()) + << "; readCb=" << int(stream.readCallbackRegistered()); if (stream.open()) { stream.deliverReadError(WebTransport::Exception(errorCode, reason)); - stopReadingWebTransportIngress(id, errorCode); + } + // Not the same as open(): a handle that saw a FIN with no read + // outstanding is closed but still registered, and about to be destroyed. + if (stream.readCallbackRegistered()) { + stream.onReadCallbackUnregistered(); + tp_.stopReadingWebTransportIngress(id, errorCode); } } + sp_.refreshTimeout(); auto wtEgressStreams = std::move(wtEgressStreams_); for (auto& [id, stream] : wtEgressStreams) { @@ -217,6 +224,14 @@ folly::Expected WebTransportImpl::stopReadingWebTransportIngress( HTTPCodec::StreamID id, folly::Optional errorCode) { auto res = tp_.stopReadingWebTransportIngress(id, errorCode); + if (!res.hasError()) { + // On failure the transport may still hold the handle, so leave it marked + // registered and let teardown unregister it. + auto it = wtIngressStreams_.find(id); + if (it != wtIngressStreams_.end()) { + it->second.onReadCallbackUnregistered(); + } + } sp_.refreshTimeout(); return res; } @@ -429,16 +444,15 @@ WebTransportImpl::StreamReadHandle::readStreamData() { VLOG(4) << __func__ << " waiting for data"; auto contract = folly::makePromiseContract(); readPromise_ = std::move(contract.promise); - readPromise_.setInterruptHandler( - [this](const folly::exception_wrapper& ex) { - VLOG(4) << "Exception from interrupt handler ex=" << ex.what(); - CHECK(ex.with_exception([this](const folly::FutureCancellation& ex) { - // TODO: allow app to configure the reset code on cancellation? - impl_.tp_.stopReadingWebTransportIngress( - id_, WebTransport::kInternalError); - deliverReadError(ex); - })) << "Unexpected exception type"; - }); + readPromise_.setInterruptHandler([this]( + const folly::exception_wrapper& ex) { + VLOG(4) << "Exception from interrupt handler ex=" << ex.what(); + CHECK(ex.with_exception([this](const folly::FutureCancellation& ex) { + // TODO: allow app to configure the reset code on cancellation? + impl_.stopReadingWebTransportIngress(id_, WebTransport::kInternalError); + deliverReadError(ex); + })) << "Unexpected exception type"; + }); return std::move(contract.future); } else { VLOG(4) << __func__ << " returning data len=" << buf_.chainLength(); @@ -515,7 +529,8 @@ WebTransport::FCState WebTransportImpl::StreamReadHandle::dataAvailable( void WebTransportImpl::StreamReadHandle::readError( quic::StreamId id, quic::QuicError error) noexcept { - // Do I need to setReadCallback(id, nullptr); + // The transport delivers readError() and drops the read callback itself. + readCallbackRegistered_ = false; impl_.sp_.refreshTimeout(); auto quicAppErrorCode = error.code.asApplicationErrorCode(); if (quicAppErrorCode) { diff --git a/proxygen/lib/http/webtransport/WebTransportImpl.h b/proxygen/lib/http/webtransport/WebTransportImpl.h index 2ee848f6f..da1169e71 100644 --- a/proxygen/lib/http/webtransport/WebTransportImpl.h +++ b/proxygen/lib/http/webtransport/WebTransportImpl.h @@ -324,6 +324,15 @@ class WebTransportImpl : public WebTransport { return !eof_ && !ex_; } + // True while the transport holds this handle as the stream's read + // callback. Destroying the handle while set leaves a dangling callback. + [[nodiscard]] bool readCallbackRegistered() const { + return readCallbackRegistered_; + } + void onReadCallbackUnregistered() { + readCallbackRegistered_ = false; + } + // quic::StreamReadCallback overrides void readAvailable(quic::StreamId id) noexcept override; void readError(quic::StreamId id, quic::QuicError error) noexcept override; @@ -333,6 +342,7 @@ class WebTransportImpl : public WebTransport { folly::Promise readPromise_; folly::IOBufQueue buf_{folly::IOBufQueue::cacheChainLength()}; bool eof_{false}; + bool readCallbackRegistered_{true}; }; private: @@ -430,6 +440,15 @@ class WebTransportImpl : public WebTransport { void closeEgressStream(HTTPCodec::StreamID id); void closeIngressStream(HTTPCodec::StreamID id); + [[nodiscard]] StreamReadHandle* getReadHandle(HTTPCodec::StreamID id) { + auto it = wtIngressStreams_.find(id); + return it == wtIngressStreams_.end() ? nullptr : &it->second; + } + [[nodiscard]] StreamWriteHandle* getWriteHandle(HTTPCodec::StreamID id) { + auto it = wtEgressStreams_.find(id); + return it == wtEgressStreams_.end() ? nullptr : &it->second; + } + [[nodiscard]] bool isBidirectional(HTTPCodec::StreamID id) const { return (id & 0b10) == 0; }