Rewrite MoqtDataParser to only consume one thing at a time. Also fix the way we handle errors when parsing datagrams. PiperOrigin-RevId: 703205925
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc index a5e0da0..81ab2d8 100644 --- a/quiche/quic/moqt/moqt_parser.cc +++ b/quiche/quic/moqt/moqt_parser.cc
@@ -4,22 +4,28 @@ #include "quiche/quic/moqt/moqt_parser.h" +#include <algorithm> #include <array> #include <cstddef> #include <cstdint> #include <cstring> #include <optional> #include <string> +#include <tuple> +#include "absl/base/casts.h" #include "absl/cleanup/cleanup.h" #include "absl/strings/str_cat.h" #include "absl/strings/string_view.h" +#include "absl/types/span.h" #include "quiche/quic/core/quic_data_reader.h" #include "quiche/quic/core/quic_time.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_priority.h" #include "quiche/common/platform/api/quiche_bug_tracker.h" #include "quiche/common/platform/api/quiche_logging.h" +#include "quiche/common/quiche_data_reader.h" +#include "quiche/common/quiche_stream.h" namespace moqt { @@ -61,81 +67,6 @@ return false; } -size_t ParseObjectHeader(quic::QuicDataReader& reader, MoqtObject& object, - MoqtDataStreamType type) { - if (!reader.ReadVarInt62(&object.track_alias)) { - return 0; - } - if (type != MoqtDataStreamType::kStreamHeaderFetch && - !reader.ReadVarInt62(&object.group_id)) { - return 0; - } - if (type == MoqtDataStreamType::kStreamHeaderSubgroup) { - uint64_t subgroup_id; - if (!reader.ReadVarInt62(&subgroup_id)) { - return 0; - } - object.subgroup_id = subgroup_id; - } - if (type == MoqtDataStreamType::kObjectDatagram && - !reader.ReadVarInt62(&object.object_id)) { - return 0; - } - if (type != MoqtDataStreamType::kStreamHeaderFetch && - !reader.ReadUInt8(&object.publisher_priority)) { - return 0; - } - uint64_t status = static_cast<uint64_t>(MoqtObjectStatus::kNormal); - if (type == MoqtDataStreamType::kObjectDatagram && - (!reader.ReadVarInt62(&object.payload_length) || - (object.payload_length == 0 && !reader.ReadVarInt62(&status)))) { - return 0; - } - object.object_status = IntegerToObjectStatus(status); - return reader.PreviouslyReadPayload().size(); -} - -size_t ParseObjectSubheader(quic::QuicDataReader& reader, MoqtObject& object, - MoqtDataStreamType type) { - switch (type) { - case MoqtDataStreamType::kStreamHeaderFetch: - if (!reader.ReadVarInt62(&object.group_id)) { - return 0; - } - if (type == MoqtDataStreamType::kStreamHeaderFetch) { - uint64_t value; - if (!reader.ReadVarInt62(&value)) { - return 0; - } - object.subgroup_id = value; - } - [[fallthrough]]; - - case MoqtDataStreamType::kStreamHeaderSubgroup: { - if (!reader.ReadVarInt62(&object.object_id)) { - return 0; - } - if (type == MoqtDataStreamType::kStreamHeaderFetch && - !reader.ReadUInt8(&object.publisher_priority)) { - return 0; - } - if (!reader.ReadVarInt62(&object.payload_length)) { - return 0; - } - uint64_t status = static_cast<uint64_t>(MoqtObjectStatus::kNormal); - if (object.payload_length == 0 && !reader.ReadVarInt62(&status)) { - return 0; - } - object.object_status = IntegerToObjectStatus(status); - return reader.PreviouslyReadPayload().size(); - } - - default: - QUICHE_NOTREACHED(); - return 0; - } -} - } // namespace // The buffering philosophy is complicated, to minimize copying. Here is an @@ -1057,31 +988,44 @@ if (parsing_error_) { return; // Don't send multiple parse errors. } + next_input_ = kFailed; no_more_data_ = true; parsing_error_ = true; visitor_.OnParsingError(MoqtError::kProtocolViolation, reason); } -absl::string_view ParseDatagram(absl::string_view data, - MoqtObject& object_metadata) { - uint64_t value; +std::optional<absl::string_view> ParseDatagram(absl::string_view data, + MoqtObject& object_metadata) { + uint64_t type_raw, object_status_raw; quic::QuicDataReader reader(data); - if (!reader.ReadVarInt62(&value)) { - return absl::string_view(); + if (!reader.ReadVarInt62(&type_raw) || + type_raw != static_cast<uint64_t>(MoqtDataStreamType::kObjectDatagram) || + !reader.ReadVarInt62(&object_metadata.track_alias) || + !reader.ReadVarInt62(&object_metadata.group_id) || + !reader.ReadVarInt62(&object_metadata.object_id) || + !reader.ReadUInt8(&object_metadata.publisher_priority) || + !reader.ReadVarInt62(&object_metadata.payload_length)) { + return std::nullopt; } - if (static_cast<MoqtDataStreamType>(value) != - MoqtDataStreamType::kObjectDatagram) { - return absl::string_view(); + if (object_metadata.payload_length > 0) { + object_metadata.object_status = MoqtObjectStatus::kNormal; + } else { + if (!reader.ReadVarInt62(&object_status_raw)) { + return std::nullopt; + } + object_metadata.object_status = IntegerToObjectStatus(object_status_raw); + if (object_metadata.object_status == + MoqtObjectStatus::kInvalidObjectStatus) { + return std::nullopt; + } } - size_t processed_data = ParseObjectHeader( - reader, object_metadata, MoqtDataStreamType::kObjectDatagram); - if (processed_data == 0) { // Incomplete header - return absl::string_view(); + if (reader.PeekRemainingPayload().size() != object_metadata.payload_length) { + return std::nullopt; } return reader.PeekRemainingPayload(); } -void MoqtDataParser::ProcessData(absl::string_view data, bool fin) { +void MoqtDataParser::ReadAllData() { if (processing_) { QUICHE_BUG(MoqtDataParser_reentry) << "Calling ProcessData() when ProcessData() is already in progress."; @@ -1090,104 +1034,253 @@ processing_ = true; auto on_return = absl::MakeCleanup([&] { processing_ = false; }); - if (no_more_data_) { - ParseError("Data after end of stream"); - return; - } - - // Sad path: there is already data buffered. Attempt to transfer a small - // chunk from `data` into the buffer, in hope that it will make the contents - // of the buffer parsable without any leftover data. This is a reasonable - // expectation, since object headers are small, and are often followed by - // large blobs of data. - while (!buffered_message_.empty() && !data.empty()) { - absl::string_view chunk = data.substr(0, chunk_size_); - absl::StrAppend(&buffered_message_, chunk); - absl::string_view unprocessed = ProcessDataInner(buffered_message_); - if (unprocessed.size() >= chunk.size()) { - // chunk didn't allow any processing at all. - data.remove_prefix(chunk.size()); - } else { - buffered_message_.clear(); - data.remove_prefix(chunk.size() - unprocessed.size()); + State last_state = state(); + for (;;) { + ParseNextItemFromStream(); + if (state() == last_state || no_more_data_) { + break; } - } - - // Happy path: there is no buffered data. - if (buffered_message_.empty() && !data.empty()) { - buffered_message_.assign(ProcessDataInner(data)); - } - - if (fin) { - if (!buffered_message_.empty() || !metadata_.has_value() || - payload_length_remaining_ > 0) { - ParseError("FIN received at an unexpected point in the stream"); - return; - } - no_more_data_ = true; + last_state = state(); } } -absl::string_view MoqtDataParser::ProcessDataInner(absl::string_view data) { - quic::QuicDataReader reader(data); - while (!reader.IsDoneReading()) { - absl::string_view remainder = reader.PeekRemainingPayload(); - switch (GetNextInput()) { - case kStreamType: { - uint64_t value; - if (!reader.ReadVarInt62(&value)) { - return remainder; - } - if (!IsAllowedStreamType(value)) { - ParseError(absl::StrCat("Unknown stream type: ", value)); - return ""; - } - type_ = static_cast<MoqtDataStreamType>(value); - continue; - } +std::optional<uint64_t> MoqtDataParser::ReadVarInt62(bool& fin_read) { + fin_read = false; - case kHeader: { - MoqtObject header; - size_t bytes_read = ParseObjectHeader(reader, header, *type_); - if (bytes_read == 0) { - return remainder; - } - metadata_ = header; - continue; - } - - case kSubheader: { - size_t bytes_read = ParseObjectSubheader(reader, *metadata_, *type_); - if (bytes_read == 0) { - return remainder; - } - if (metadata_->object_status == - MoqtObjectStatus::kInvalidObjectStatus) { - ParseError("Invalid object status provided"); - return ""; - } - payload_length_remaining_ = metadata_->payload_length; - if (payload_length_remaining_ == 0) { - visitor_.OnObjectMessage(*metadata_, "", true); - } - continue; - } - - case kData: { - absl::string_view payload = - reader.ReadAtMost(payload_length_remaining_); - visitor_.OnObjectMessage(*metadata_, payload, - payload.size() == payload_length_remaining_); - payload_length_remaining_ -= payload.size(); - - continue; - } - - case kPadding: - return ""; + quiche::ReadStream::PeekResult peek_result = stream_.PeekNextReadableRegion(); + if (!peek_result.has_data()) { + if (peek_result.fin_next) { + fin_read = stream_.SkipBytes(0); + QUICHE_DCHECK(fin_read); } + return std::nullopt; } - return ""; + char first_byte = peek_result.peeked_data[0]; + size_t varint_size = + 1 << ((absl::bit_cast<uint8_t>(first_byte) & 0b11000000) >> 6); + if (stream_.ReadableBytes() < varint_size) { + return std::nullopt; + } + + char buffer[8]; + absl::Span<char> bytes_to_read = + absl::MakeSpan(buffer).subspan(0, varint_size); + quiche::ReadStream::ReadResult read_result = stream_.Read(bytes_to_read); + QUICHE_DCHECK_EQ(read_result.bytes_read, varint_size); + fin_read = read_result.fin; + + quiche::QuicheDataReader reader(buffer, read_result.bytes_read); + uint64_t result; + bool success = reader.ReadVarInt62(&result); + QUICHE_DCHECK(success); + QUICHE_DCHECK(reader.IsDoneReading()); + return result; +} + +std::optional<uint64_t> MoqtDataParser::ReadVarInt62NoFin() { + bool fin_read = false; + std::optional<uint64_t> result = ReadVarInt62(fin_read); + if (fin_read) { + ParseError("Unexpected FIN received in the middle of a header"); + return std::nullopt; + } + return result; +} + +std::optional<uint8_t> MoqtDataParser::ReadUint8NoFin() { + char buffer[1]; + quiche::ReadStream::ReadResult read_result = + stream_.Read(absl::MakeSpan(buffer)); + if (read_result.fin) { + ParseError("Unexpected FIN received in the middle of a header"); + return std::nullopt; + } + if (read_result.bytes_read == 0) { + return std::nullopt; + } + return absl::bit_cast<uint8_t>(buffer[0]); +} + +void MoqtDataParser::AdvanceParserState() { + QUICHE_DCHECK(type_ == MoqtDataStreamType::kStreamHeaderSubgroup || + type_ == MoqtDataStreamType::kStreamHeaderFetch); + const bool is_fetch = type_ == MoqtDataStreamType::kStreamHeaderFetch; + switch (next_input_) { + // The state table is factored into a separate function (rather than + // inlined) in order to separate the order of elements from the way they are + // parsed. + case kStreamType: + next_input_ = kTrackAlias; + break; + case kTrackAlias: + next_input_ = kGroupId; + break; + case kGroupId: + next_input_ = kSubgroupId; + break; + case kSubgroupId: + next_input_ = is_fetch ? kObjectId : kPublisherPriority; + break; + case kPublisherPriority: + next_input_ = is_fetch ? kObjectPayloadLength : kObjectId; + break; + case kObjectId: + next_input_ = is_fetch ? kPublisherPriority : kObjectPayloadLength; + break; + case kStatus: + case kData: + next_input_ = is_fetch ? kGroupId : kObjectId; + break; + + case kObjectPayloadLength: // Either kStatus or kData depending on length. + case kPadding: // Handled separately. + case kFailed: // Should cause parsing to cease. + QUICHE_NOTREACHED(); + break; + } +} + +void MoqtDataParser::ParseNextItemFromStream() { + switch (next_input_) { + case kStreamType: { + std::optional<uint64_t> value_read = ReadVarInt62NoFin(); + if (value_read.has_value()) { + if (!IsAllowedStreamType(*value_read)) { + ParseError("Invalid stream type supplied"); + return; + } + type_ = static_cast<MoqtDataStreamType>(*value_read); + switch (*type_) { + case MoqtDataStreamType::kStreamHeaderSubgroup: + case MoqtDataStreamType::kStreamHeaderFetch: + AdvanceParserState(); + break; + case MoqtDataStreamType::kPadding: + next_input_ = kPadding; + break; + case MoqtDataStreamType::kObjectDatagram: + QUICHE_BUG(ParseDataFromStream_kStreamType_unexpected); + return; + } + } + return; + } + + case kTrackAlias: { + std::optional<uint64_t> value_read = ReadVarInt62NoFin(); + if (value_read.has_value()) { + metadata_.track_alias = *value_read; + AdvanceParserState(); + } + return; + } + + case kGroupId: { + std::optional<uint64_t> value_read = ReadVarInt62NoFin(); + if (value_read.has_value()) { + metadata_.group_id = *value_read; + AdvanceParserState(); + } + return; + } + + case kSubgroupId: { + std::optional<uint64_t> value_read = ReadVarInt62NoFin(); + if (value_read.has_value()) { + metadata_.subgroup_id = *value_read; + AdvanceParserState(); + } + return; + } + + case kPublisherPriority: { + std::optional<uint8_t> value_read = ReadUint8NoFin(); + if (value_read.has_value()) { + metadata_.publisher_priority = *value_read; + AdvanceParserState(); + } + return; + } + + case kObjectId: { + std::optional<uint64_t> value_read = ReadVarInt62NoFin(); + if (value_read.has_value()) { + metadata_.object_id = *value_read; + AdvanceParserState(); + } + return; + } + + case kObjectPayloadLength: { + std::optional<uint64_t> value_read = ReadVarInt62NoFin(); + if (value_read.has_value()) { + metadata_.payload_length = *value_read; + payload_length_remaining_ = *value_read; + if (metadata_.payload_length > 0) { + metadata_.object_status = MoqtObjectStatus::kNormal; + next_input_ = kData; + } else { + next_input_ = kStatus; + } + } + return; + } + + case kStatus: { + bool fin_read = false; + std::optional<uint64_t> value_read = ReadVarInt62(fin_read); + if (value_read.has_value()) { + metadata_.object_status = IntegerToObjectStatus(*value_read); + if (metadata_.object_status == MoqtObjectStatus::kInvalidObjectStatus) { + ParseError("Invalid object status provided"); + return; + } + + visitor_.OnObjectMessage(metadata_, "", /*end_of_message=*/true); + AdvanceParserState(); + } + if (fin_read) { + no_more_data_ = true; + return; + } + return; + } + + case kData: { + while (payload_length_remaining_ > 0) { + quiche::ReadStream::PeekResult peek_result = + stream_.PeekNextReadableRegion(); + if (peek_result.peeked_data.empty() && !peek_result.fin_next) { + return; + } + if (peek_result.fin_next && + peek_result.peeked_data.size() < payload_length_remaining_) { + ParseError("FIN received at an unexpected point in the stream"); + return; + } + + size_t chunk_size = + std::min(payload_length_remaining_, peek_result.peeked_data.size()); + payload_length_remaining_ -= chunk_size; + bool done = payload_length_remaining_ == 0; + visitor_.OnObjectMessage( + metadata_, peek_result.peeked_data.substr(0, chunk_size), done); + const bool fin = stream_.SkipBytes(chunk_size); + if (done) { + no_more_data_ |= fin; + AdvanceParserState(); + } + } + return; + } + + case kPadding: + no_more_data_ |= stream_.SkipBytes(stream_.ReadableBytes()); + return; + + case kFailed: + return; + } } } // namespace moqt
diff --git a/quiche/quic/moqt/moqt_parser.h b/quiche/quic/moqt/moqt_parser.h index 34953e4..a96a70d 100644 --- a/quiche/quic/moqt/moqt_parser.h +++ b/quiche/quic/moqt/moqt_parser.h
@@ -17,6 +17,7 @@ #include "quiche/quic/core/quic_data_reader.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/common/platform/api/quiche_export.h" +#include "quiche/common/quiche_stream.h" namespace moqt { @@ -174,90 +175,80 @@ // re-entrancy. }; -// Parses an MoQT datagram. Returns the payload bytes, or empty string_view on -// error. The caller provides the whole datagram in `data`. The function puts -// the object metadata in `object_metadata`. -absl::string_view ParseDatagram(absl::string_view data, - MoqtObject& object_metadata); +// Parses an MoQT datagram. Returns the payload bytes, or std::nullopt on error. +// The caller provides the whole datagram in `data`. The function puts the +// object metadata in `object_metadata`. +std::optional<absl::string_view> ParseDatagram(absl::string_view data, + MoqtObject& object_metadata); // Parser for MoQT unidirectional data stream. class QUICHE_EXPORT MoqtDataParser { public: - explicit MoqtDataParser(MoqtDataParserVisitor* visitor) - : visitor_(*visitor) {} - ~MoqtDataParser() = default; + // `stream` must outlive the parser. The parser does not configure itself as + // a listener for the read events of the stream; it is responsibility of the + // caller to do so via one of the read methods below. + explicit MoqtDataParser(quiche::ReadStream* stream, + MoqtDataParserVisitor* visitor) + : stream_(*stream), visitor_(*visitor) {} - // Take a buffer from the transport in |data|. Parse each complete message and - // call the appropriate visitor function. If |fin| is true, there - // is no more data arriving on the stream, so the parser will deliver any - // message encoded as to run to the end of the stream. - // All bytes can be freed. Calls OnParsingError() when there is a parsing - // error. - void ProcessData(absl::string_view data, bool fin); + // Reads all of the available objects on the stream. + void ReadAllData(); - // Alters `chunk_size_` value (see discussion below). Primarily intended to - // be used for testing. - void set_chunk_size(size_t size) { chunk_size_ = size; } - + // Returns the type of the unidirectional stream, if already known. std::optional<MoqtDataStreamType> stream_type() const { return type_; } private: friend class test::MoqtDataParserPeer; - // If there is buffered data from the previous attempt at parsing it, new data - // will be added in `chunk_size_`-sized chunks. - constexpr static size_t kDefaultChunkSize = 64; - // Current state of the parser. enum NextInput { - // Nothing has been read yet; the next thing to be read is the stream type - // varint. kStreamType, - // The next thing to be read is the stream header. - kHeader, - // The next thing to be read is the stream subheader for the given object. - kSubheader, - // The next thing to be read is the object payload. + kTrackAlias, + kGroupId, + kSubgroupId, + kPublisherPriority, + kObjectId, + kObjectPayloadLength, + kStatus, kData, - // The next thing to be read (and ignored) is padding. kPadding, + kFailed, }; + struct State { + NextInput next_input; + uint64_t payload_remaining; - // Infers the current state of the parser. - NextInput GetNextInput() const { - if (!type_.has_value()) { - return kStreamType; - } - if (type_ == MoqtDataStreamType::kPadding) { - return kPadding; - } - if (!metadata_.has_value()) { - return kHeader; - } - if (payload_length_remaining_ > 0) { - return kData; - } - return kSubheader; - } + bool operator==(const State&) const = default; + }; + State state() const { return State{next_input_, payload_length_remaining_}; } - // Processes all that can be entirely processed, and returns the view for the - // data that needs to be buffered. - absl::string_view ProcessDataInner(absl::string_view data); + // Reads a single varint from the underlying stream. + std::optional<uint64_t> ReadVarInt62(bool& fin_read); + // Reads a single varint from the underlying stream. Triggers a parse error if + // a FIN has been encountered. + std::optional<uint64_t> ReadVarInt62NoFin(); + // Reads a single uint8 from the underlying stream. Triggers a parse error if + // a FIN has been encountered. + std::optional<uint8_t> ReadUint8NoFin(); + + // Advances the state machine of the parser to the next expected state. + void AdvanceParserState(); + // Reads the next available item from the stream. + void ParseNextItemFromStream(); void ParseError(absl::string_view reason); + quiche::ReadStream& stream_; MoqtDataParserVisitor& visitor_; - size_t chunk_size_ = kDefaultChunkSize; bool no_more_data_ = false; // Fatal error or fin. No more parsing. bool parsing_error_ = false; std::string buffered_message_; - // The three variables below implicitly drive the state machine; see - // `GetNextInput()` for how the state is derived. std::optional<MoqtDataStreamType> type_ = std::nullopt; - std::optional<MoqtObject> metadata_ = std::nullopt; + NextInput next_input_ = kStreamType; + MoqtObject metadata_; size_t payload_length_remaining_ = 0; bool processing_ = false; // True if currently in ProcessData(), to prevent
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc index 1c77875..a4b38bb 100644 --- a/quiche/quic/moqt/moqt_parser_test.cc +++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -21,6 +21,7 @@ #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/test_tools/moqt_test_message.h" #include "quiche/quic/platform/api/quic_test.h" +#include "quiche/web_transport/test_tools/in_memory_stream.h" namespace moqt::test { @@ -237,7 +238,8 @@ : message_type_(GetParam().message_type), webtrans_(GetParam().uses_web_transport), control_parser_(GetParam().uses_web_transport, visitor_), - data_parser_(&visitor_) {} + data_stream_(/*stream_id=*/0), + data_parser_(&data_stream_, &visitor_) {} bool IsDataStream() { return absl::holds_alternative<MoqtDataStreamType>(message_type_); @@ -254,7 +256,8 @@ void ProcessData(absl::string_view data, bool fin) { if (IsDataStream()) { - data_parser_.ProcessData(data, fin); + data_stream_.Receive(data, fin); + data_parser_.ReadAllData(); } else { control_parser_.ProcessData(data, fin); } @@ -265,6 +268,7 @@ GeneralizedMessageType message_type_; bool webtrans_; MoqtControlParser control_parser_; + webtransport::test::InMemoryStream data_stream_; MoqtDataParser data_parser_; }; @@ -354,7 +358,6 @@ TEST_P(MoqtParserTest, TwoBytesAtATime) { std::unique_ptr<TestMessageBase> message = MakeMessage(); - data_parser_.set_chunk_size(1); for (size_t i = 0; i < message->total_message_size(); i += 3) { EXPECT_EQ(visitor_.messages_received_, 0); EXPECT_FALSE(visitor_.end_of_message_); @@ -432,24 +435,28 @@ // Send the header + some payload, pure payload, then pure payload to end the // message. TEST_F(MoqtMessageSpecificTest, ThreePartObject) { - MoqtDataParser parser(&visitor_); + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtDataParser parser(&stream, &visitor_); auto message = std::make_unique<StreamHeaderSubgroupMessage>(); EXPECT_TRUE(message->SetPayloadLength(14)); - parser.ProcessData(message->PacketSample(), false); + stream.Receive(message->PacketSample(), false); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 0); EXPECT_TRUE(message->EqualFieldValues(*visitor_.last_message_)); EXPECT_FALSE(visitor_.end_of_message_); EXPECT_EQ(visitor_.object_payload(), "foo"); // second part - parser.ProcessData("bar", false); + stream.Receive("bar", false); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 0); EXPECT_TRUE(message->EqualFieldValues(*visitor_.last_message_)); EXPECT_FALSE(visitor_.end_of_message_); EXPECT_EQ(visitor_.object_payload(), "foobar"); // third part includes FIN - parser.ProcessData("deadbeef", true); + stream.Receive("deadbeef", true); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 1); EXPECT_TRUE(message->EqualFieldValues(*visitor_.last_message_)); EXPECT_TRUE(visitor_.end_of_message_); @@ -459,19 +466,22 @@ // Send the part of header, rest of header + payload, plus payload. TEST_F(MoqtMessageSpecificTest, ThreePartObjectFirstIncomplete) { - MoqtDataParser parser(&visitor_); + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtDataParser parser(&stream, &visitor_); auto message = std::make_unique<StreamHeaderSubgroupMessage>(); EXPECT_TRUE(message->SetPayloadLength(51)); // first part - parser.ProcessData(message->PacketSample().substr(0, 4), false); + stream.Receive(message->PacketSample().substr(0, 4), false); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 0); // second part. Add padding to it. message->set_wire_image_size(55); - parser.ProcessData( + stream.Receive( message->PacketSample().substr(4, message->total_message_size() - 4), false); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 0); EXPECT_TRUE(message->EqualFieldValues(*visitor_.last_message_)); EXPECT_FALSE(visitor_.end_of_message_); @@ -480,7 +490,8 @@ EXPECT_EQ(visitor_.object_payload().length(), 48); // third part includes FIN - parser.ProcessData("bar", true); + stream.Receive("bar", true); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 1); EXPECT_TRUE(message->EqualFieldValues(*visitor_.last_message_)); EXPECT_TRUE(visitor_.end_of_message_); @@ -489,10 +500,12 @@ } TEST_F(MoqtMessageSpecificTest, StreamHeaderSubgroupFollowOn) { - MoqtDataParser parser(&visitor_); + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtDataParser parser(&stream, &visitor_); // first part auto message1 = std::make_unique<StreamHeaderSubgroupMessage>(); - parser.ProcessData(message1->PacketSample(), false); + stream.Receive(message1->PacketSample(), false); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 1); EXPECT_TRUE(message1->EqualFieldValues(*visitor_.last_message_)); EXPECT_TRUE(visitor_.end_of_message_); @@ -501,7 +514,8 @@ // second part visitor_.object_payloads_.clear(); auto message2 = std::make_unique<StreamMiddlerSubgroupMessage>(); - parser.ProcessData(message2->PacketSample(), false); + stream.Receive(message2->PacketSample(), false); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 2); EXPECT_TRUE(message2->EqualFieldValues(*visitor_.last_message_)); EXPECT_TRUE(visitor_.end_of_message_); @@ -857,11 +871,13 @@ } TEST_F(MoqtMessageSpecificTest, FinMidPayload) { - MoqtDataParser parser(&visitor_); + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtDataParser parser(&stream, &visitor_); auto message = std::make_unique<StreamHeaderSubgroupMessage>(); - parser.ProcessData( + stream.Receive( message->PacketSample().substr(0, message->total_message_size() - 1), true); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 0); EXPECT_EQ(visitor_.parsing_error_, "FIN received at an unexpected point in the stream"); @@ -869,12 +885,15 @@ } TEST_F(MoqtMessageSpecificTest, PartialPayloadThenFin) { - MoqtDataParser parser(&visitor_); + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtDataParser parser(&stream, &visitor_); auto message = std::make_unique<StreamHeaderSubgroupMessage>(); - parser.ProcessData( + stream.Receive( message->PacketSample().substr(0, message->total_message_size() - 1), false); - parser.ProcessData(absl::string_view(), true); + parser.ReadAllData(); + stream.Receive(absl::string_view(), true); + parser.ReadAllData(); EXPECT_EQ(visitor_.messages_received_, 0); EXPECT_EQ(visitor_.parsing_error_, "FIN received at an unexpected point in the stream"); @@ -890,16 +909,18 @@ } TEST_F(MoqtMessageSpecificTest, InvalidObjectStatus) { - MoqtDataParser parser(&visitor_); + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtDataParser parser(&stream, &visitor_); char stream_header_subgroup[] = { 0x04, // type field 0x04, 0x05, 0x08, // varints 0x07, // publisher priority 0x06, 0x00, 0x0f, // object middler; status = 0x0f }; - parser.ProcessData( + stream.Receive( absl::string_view(stream_header_subgroup, sizeof(stream_header_subgroup)), false); + parser.ReadAllData(); EXPECT_EQ(visitor_.parsing_error_, "Invalid object status provided"); EXPECT_EQ(visitor_.parsing_error_code_, MoqtError::kProtocolViolation); } @@ -1243,7 +1264,9 @@ TEST_F(MoqtMessageSpecificTest, DatagramSuccessful) { ObjectDatagramMessage message; MoqtObject object; - absl::string_view payload = ParseDatagram(message.PacketSample(), object); + std::optional<absl::string_view> payload = + ParseDatagram(message.PacketSample(), object); + ASSERT_TRUE(payload.has_value()); TestMessageBase::MessageStructuredData object_metadata = TestMessageBase::MessageStructuredData(object); EXPECT_TRUE(message.EqualFieldValues(object_metadata)); @@ -1253,24 +1276,26 @@ TEST_F(MoqtMessageSpecificTest, WrongMessageInDatagram) { StreamHeaderSubgroupMessage message; MoqtObject object; - absl::string_view payload = ParseDatagram(message.PacketSample(), object); - EXPECT_TRUE(payload.empty()); + std::optional<absl::string_view> payload = + ParseDatagram(message.PacketSample(), object); + EXPECT_EQ(payload, std::nullopt); } TEST_F(MoqtMessageSpecificTest, TruncatedDatagram) { ObjectDatagramMessage message; message.set_wire_image_size(4); MoqtObject object; - absl::string_view payload = ParseDatagram(message.PacketSample(), object); - EXPECT_TRUE(payload.empty()); + std::optional<absl::string_view> payload = + ParseDatagram(message.PacketSample(), object); + EXPECT_EQ(payload, std::nullopt); } TEST_F(MoqtMessageSpecificTest, VeryTruncatedDatagram) { char message = 0x40; MoqtObject object; - absl::string_view payload = + std::optional<absl::string_view> payload = ParseDatagram(absl::string_view(&message, sizeof(message)), object); - EXPECT_TRUE(payload.empty()); + EXPECT_EQ(payload, std::nullopt); } TEST_F(MoqtMessageSpecificTest, SubscribeOkInvalidContentExists) { @@ -1340,13 +1365,15 @@ } TEST_F(MoqtMessageSpecificTest, PaddingStream) { - MoqtDataParser parser(&visitor_); + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtDataParser parser(&stream, &visitor_); std::string buffer(32, '\0'); quic::QuicDataWriter writer(buffer.size(), buffer.data()); ASSERT_TRUE(writer.WriteVarInt62( static_cast<uint64_t>(MoqtDataStreamType::kPadding))); for (int i = 0; i < 100; ++i) { - parser.ProcessData(buffer, false); + stream.Receive(buffer, false); + parser.ReadAllData(); ASSERT_EQ(visitor_.messages_received_, 0); ASSERT_EQ(visitor_.parsing_error_, std::nullopt); }
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index 772dbff..6a06e00 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -190,14 +190,18 @@ void MoqtSession::OnDatagramReceived(absl::string_view datagram) { MoqtObject message; - absl::string_view payload = ParseDatagram(datagram, message); + std::optional<absl::string_view> payload = ParseDatagram(datagram, message); + if (!payload.has_value()) { + Error(MoqtError::kProtocolViolation, "Malformed datagram received"); + return; + } QUICHE_DLOG(INFO) << ENDPOINT << "Received OBJECT message in datagram for subscribe_id " << " for track alias " << message.track_alias << " with sequence " << message.group_id << ":" << message.object_id << " priority " << message.publisher_priority << " length " - << payload.size(); + << payload->size(); SubscribeRemoteTrack* track = RemoteTrackByAlias(message.track_alias); if (track == nullptr) { return; @@ -219,7 +223,7 @@ visitor->OnObjectFragment( track->full_track_name(), FullSequence{message.group_id, 0, message.object_id}, - message.publisher_priority, message.object_status, payload, true); + message.publisher_priority, message.object_status, *payload, true); } } @@ -1086,9 +1090,9 @@ << stream_->GetStreamId() << " for track alias " << message.track_alias << " with sequence " << message.group_id << ":" << message.object_id - << " priority " << message.publisher_priority - << " length " << payload.size() << " length " - << message.payload_length << (end_of_message ? "F" : ""); + << " priority " << message.publisher_priority << " length " + << payload.size() << " length " << message.payload_length + << (end_of_message ? "F" : ""); if (!session_->parameters_.deliver_partial_objects) { if (!end_of_message) { // Buffer partial object. if (partial_object_.empty()) { @@ -1143,9 +1147,7 @@ partial_object_.clear(); } -void MoqtSession::IncomingDataStream::OnCanRead() { - ForwardStreamDataToParser(*stream_, parser_); -} +void MoqtSession::IncomingDataStream::OnCanRead() { parser_.ReadAllData(); } void MoqtSession::IncomingDataStream::OnControlMessageReceived() { session_->Error(MoqtError::kProtocolViolation,
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h index 33063ba..7908c9e 100644 --- a/quiche/quic/moqt/moqt_session.h +++ b/quiche/quic/moqt/moqt_session.h
@@ -262,7 +262,7 @@ public MoqtDataParserVisitor { public: IncomingDataStream(MoqtSession* session, webtransport::Stream* stream) - : session_(session), stream_(stream), parser_(this) {} + : session_(session), stream_(stream), parser_(stream, this) {} // webtransport::StreamVisitor implementation. void OnCanRead() override;