Return session error if a request ID is reused in a SUBSCRIBE, or has the wrong parity. Other cases would require unbounded state to detect. Also, reconnect request ID validation with where request IDs are processed (in the bidi streams). Part of draft-18 update. PiperOrigin-RevId: 987077562
diff --git a/quiche/quic/moqt/moqt_bidi_stream.h b/quiche/quic/moqt/moqt_bidi_stream.h index dd6ae15..2c154c6 100644 --- a/quiche/quic/moqt/moqt_bidi_stream.h +++ b/quiche/quic/moqt/moqt_bidi_stream.h
@@ -36,6 +36,10 @@ using SessionErrorCallback = quiche::SingleUseCallback<void(MoqtError, absl::string_view)>; +// If it returns !ok, the session will immediately terminate and the caller +// must return immediately. +using ValidateRequestIdCallback = + quiche::MultiUseCallback<absl::Status(uint64_t)>; // MoqtBidiStreamBase is the base class for bidirectional streams in MoQT. It // contains basic methods for handling and dispatching messages. An instance of
diff --git a/quiche/quic/moqt/moqt_fetch_stream.cc b/quiche/quic/moqt/moqt_fetch_stream.cc index 5f1e9bd..78b97d4 100644 --- a/quiche/quic/moqt/moqt_fetch_stream.cc +++ b/quiche/quic/moqt/moqt_fetch_stream.cc
@@ -32,6 +32,7 @@ #include "quiche/quic/moqt/moqt_types.h" #include "quiche/quic/moqt/moqt_uni_stream.h" #include "quiche/common/platform/api/quiche_bug_tracker.h" +#include "quiche/common/quiche_status_utils.h" #include "quiche/web_transport/web_transport.h" namespace moqt { @@ -224,12 +225,14 @@ const MoqtControlMessageParser& message_parser, MoqtPublisher* absl_nonnull application, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, OpenStreamCallback open_stream_callback, GetSubscriptionCallback get_subscription_callback) : MoqtBidiStreamBase(framer, message_parser, std::move(session_error_callback)), data_stream_(nullptr), application_(application), + validate_request_id_(std::move(validate_request_id)), open_stream_callback_(std::move(open_stream_callback)), get_subscription_callback_(std::move(get_subscription_callback)), weak_ptr_factory_(this) {} @@ -246,6 +249,7 @@ return absl::InvalidArgumentError( "FETCH received on stream that already has a fetch"); } + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); request_id_ = message.request_id; parameters_ = message.parameters; MoqtDeliveryOrder delivery_order = @@ -383,6 +387,7 @@ absl::Status MoqtFetchResponseStream::OnControlMessage( const MoqtRequestUpdate& message) { + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); if (data_stream_ != nullptr && message.parameters.subscriber_priority.has_value()) { data_stream_->UpdatePriority(*message.parameters.subscriber_priority);
diff --git a/quiche/quic/moqt/moqt_fetch_stream.h b/quiche/quic/moqt/moqt_fetch_stream.h index 5024212..6f54cea 100644 --- a/quiche/quic/moqt/moqt_fetch_stream.h +++ b/quiche/quic/moqt/moqt_fetch_stream.h
@@ -107,6 +107,7 @@ const MoqtControlMessageParser& message_parser, MoqtPublisher* absl_nonnull application, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, OpenStreamCallback open_stream_callback, GetSubscriptionCallback get_subscription_callback); ~MoqtFetchResponseStream() { @@ -135,6 +136,7 @@ MoqtPriority default_publisher_priority_ = kDefaultPublisherPriority; std::unique_ptr<MoqtFetchTask> fetch_; MoqtPublisher* absl_nonnull application_; + ValidateRequestIdCallback validate_request_id_; OpenStreamCallback open_stream_callback_; GetSubscriptionCallback get_subscription_callback_; quiche::QuicheWeakPtrFactory<MoqtFetchResponseStream> weak_ptr_factory_;
diff --git a/quiche/quic/moqt/moqt_fetch_stream_test.cc b/quiche/quic/moqt/moqt_fetch_stream_test.cc index 4701907..be1d855 100644 --- a/quiche/quic/moqt/moqt_fetch_stream_test.cc +++ b/quiche/quic/moqt/moqt_fetch_stream_test.cc
@@ -439,6 +439,7 @@ EXPECT_CALL(session_error_callback_, Call).Times(testing::AnyNumber()); ON_CALL(mock_data_stream_, CanWrite).WillByDefault(Return(true)); ON_CALL(mock_stream_, CanWrite).WillByDefault(Return(true)); + ON_CALL(validate_request_id_, Call).WillByDefault(Return(absl::OkStatus())); } std::unique_ptr<MoqtFetchResponseStream> CreateAndBindStream( @@ -450,6 +451,7 @@ auto stream = std::make_unique<MoqtFetchResponseStream>( &framer_, message_parser_, &mock_publisher_, session_error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), open_stream_callback_.AsStdFunction(), std::move(subscription_cb)); stream->BindStream(&mock_stream_); return stream; @@ -463,6 +465,7 @@ MockFetchTask task_; StrictMock<testing::MockFunction<void(MoqtError, absl::string_view)>> session_error_callback_; + testing::MockFunction<absl::Status(uint64_t)> validate_request_id_; StrictMock< testing::MockFunction<void(webtransport::StreamId, MoqtTrackPriority)>> open_stream_callback_;
diff --git a/quiche/quic/moqt/moqt_namespace_stream.cc b/quiche/quic/moqt/moqt_namespace_stream.cc index 20cc605..bbf63ef 100644 --- a/quiche/quic/moqt/moqt_namespace_stream.cc +++ b/quiche/quic/moqt/moqt_namespace_stream.cc
@@ -24,6 +24,7 @@ #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" #include "quiche/common/platform/api/quiche_logging.h" +#include "quiche/common/quiche_status_utils.h" #include "quiche/web_transport/stream_helpers.h" namespace moqt { @@ -227,12 +228,14 @@ MoqtFramer* framer, const MoqtControlMessageParser& message_parser, AddPrefixCallback add_callback, RemovePrefixCallback remove_callback, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, MoqtIncomingSubscribeNamespaceCallback& application) // No stream_deleted_callback because there's no state yet. : MoqtBidiStreamBase(framer, message_parser, std::move(session_error_callback)), add_callback_(std::move(add_callback)), remove_callback_(std::move(remove_callback)), + validate_request_id_(std::move(validate_request_id)), application_(application) {} absl::Status MoqtSubscribeNamespaceResponseStream::OnRawControlMessage( @@ -243,10 +246,11 @@ absl::Status MoqtSubscribeNamespaceResponseStream::OnControlMessage( const MoqtSubscribeNamespace& message) { - request_id_ = message.request_id; if (add_callback_ == nullptr) { return absl::InvalidArgumentError("Two SUBSCRIBE_NAMESPACE on one stream"); } + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); + request_id_ = message.request_id; if (message.track_namespace_prefix.DoesNotExist()) { add_callback_ = nullptr; remove_callback_ = nullptr; @@ -270,6 +274,7 @@ absl::Status MoqtSubscribeNamespaceResponseStream::OnControlMessage( const MoqtRequestUpdate& message) { + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); if (task_ == nullptr) { // This stream is dying. return absl::OkStatus();
diff --git a/quiche/quic/moqt/moqt_namespace_stream.h b/quiche/quic/moqt/moqt_namespace_stream.h index 28972b5..5935220 100644 --- a/quiche/quic/moqt/moqt_namespace_stream.h +++ b/quiche/quic/moqt/moqt_namespace_stream.h
@@ -149,6 +149,7 @@ MoqtFramer* framer, const MoqtControlMessageParser& message_parser, AddPrefixCallback add_callback, RemovePrefixCallback remove_callback, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, MoqtIncomingSubscribeNamespaceCallback& application); ~MoqtSubscribeNamespaceResponseStream() { Detach(); } @@ -170,6 +171,7 @@ TrackNamespace prefix_; AddPrefixCallback add_callback_; RemovePrefixCallback remove_callback_; + ValidateRequestIdCallback validate_request_id_; MoqtIncomingSubscribeNamespaceCallback& application_; std::unique_ptr<MoqtNamespaceTask> task_; absl::flat_hash_set<TrackNamespace> published_suffixes_;
diff --git a/quiche/quic/moqt/moqt_namespace_stream_test.cc b/quiche/quic/moqt/moqt_namespace_stream_test.cc index 372f79c..8de9273 100644 --- a/quiche/quic/moqt/moqt_namespace_stream_test.cc +++ b/quiche/quic/moqt/moqt_namespace_stream_test.cc
@@ -287,7 +287,9 @@ application_callback_(mock_application_.AsStdFunction()), stream_(&framer_, ControlMessageParser(), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), - error_callback_.AsStdFunction(), application_callback_) { + error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), application_callback_) { + ON_CALL(validate_request_id_, Call).WillByDefault(Return(absl::OkStatus())); stream_.BindStream(&mock_stream_); EXPECT_CALL(mock_stream_, CanWrite()).WillRepeatedly(Return(true)); } @@ -299,6 +301,7 @@ MoqtFramer framer_; testing::MockFunction<void(MoqtError, absl::string_view)> error_callback_; + testing::MockFunction<absl::Status(uint64_t)> validate_request_id_; webtransport::test::MockStream mock_stream_; testing::MockFunction<bool(const TrackNamespace&)> add_callback_; testing::MockFunction<void(const TrackNamespace&)> remove_callback_; @@ -585,7 +588,7 @@ MoqtSubscribeNamespaceResponseStream stream2( &framer_, ControlMessageParser(), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), error_callback2.AsStdFunction(), - application_callback_); + validate_request_id_.AsStdFunction(), application_callback_); stream2.BindStream(&mock_stream2); EXPECT_CALL(mock_stream2, CanWrite()).WillRepeatedly(Return(true));
diff --git a/quiche/quic/moqt/moqt_publish_namespace_stream.cc b/quiche/quic/moqt/moqt_publish_namespace_stream.cc index b1c485d..9300ff5 100644 --- a/quiche/quic/moqt/moqt_publish_namespace_stream.cc +++ b/quiche/quic/moqt/moqt_publish_namespace_stream.cc
@@ -15,6 +15,7 @@ #include "quiche/quic/moqt/moqt_key_value_pair.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_parser.h" +#include "quiche/common/quiche_status_utils.h" namespace moqt { @@ -95,6 +96,7 @@ if (add_callback_ == nullptr) { return absl::InvalidArgumentError("Two PUBLISH_NAMESPACE on one stream"); } + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); if (message.track_namespace.DoesNotExist()) { add_callback_ = nullptr; remove_callback_ = nullptr; @@ -131,6 +133,7 @@ absl::Status MoqtPublishNamespaceResponseStream::OnControlMessage( const MoqtRequestUpdate& message) { + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); if (!prefix_.has_value()) { return absl::InvalidArgumentError( "REQUEST_UPDATE before PUBLISH_NAMESPACE on a PN stream");
diff --git a/quiche/quic/moqt/moqt_publish_namespace_stream.h b/quiche/quic/moqt/moqt_publish_namespace_stream.h index 1ad38d6..47b6a15 100644 --- a/quiche/quic/moqt/moqt_publish_namespace_stream.h +++ b/quiche/quic/moqt/moqt_publish_namespace_stream.h
@@ -72,11 +72,13 @@ AddPublishNamespaceCallback add_callback, RemovePublishNamespaceCallback remove_callback, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, MoqtIncomingPublishNamespaceCallback application) : MoqtBidiStreamBase(framer, message_parser, std::move(session_error_callback)), add_callback_(std::move(add_callback)), remove_callback_(std::move(remove_callback)), + validate_request_id_(std::move(validate_request_id)), application_(std::move(application)), weak_ptr_factory_(this) {} ~MoqtPublishNamespaceResponseStream() { Detach(); } @@ -96,6 +98,7 @@ std::optional<TrackNamespace> prefix_; AddPublishNamespaceCallback add_callback_; RemovePublishNamespaceCallback remove_callback_; + ValidateRequestIdCallback validate_request_id_; MoqtIncomingPublishNamespaceCallback application_; quiche::QuicheWeakPtrFactory<MoqtPublishNamespaceResponseStream> weak_ptr_factory_;
diff --git a/quiche/quic/moqt/moqt_publish_namespace_stream_test.cc b/quiche/quic/moqt/moqt_publish_namespace_stream_test.cc index c29c445..af228db 100644 --- a/quiche/quic/moqt/moqt_publish_namespace_stream_test.cc +++ b/quiche/quic/moqt/moqt_publish_namespace_stream_test.cc
@@ -152,6 +152,7 @@ add_callback_(), remove_callback_(), application_() { + ON_CALL(validate_request_id_, Call).WillByDefault(Return(absl::OkStatus())); EXPECT_CALL(remove_callback_, Call(_)).Times(testing::AnyNumber()); EXPECT_CALL(application_, Call(_, nullptr, _)).Times(testing::AnyNumber()); } @@ -159,6 +160,7 @@ MoqtFramer framer_; StrictMock<testing::MockFunction<void(MoqtError, absl::string_view)>> session_error_callback_; + testing::MockFunction<absl::Status(uint64_t)> validate_request_id_; StrictMock< testing::MockFunction<bool(const TrackNamespace&, MoqtBidiStreamBase*)>> add_callback_; @@ -177,7 +179,8 @@ MoqtControlMessageParser(kDefaultMoqtVersion, /*webtransport=*/true, quic::Perspective::IS_SERVER), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), - session_error_callback_.AsStdFunction(), application_.AsStdFunction()); + session_error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), application_.AsStdFunction()); response_stream->BindStream(&mock_stream_); MoqtPublishNamespace message; @@ -200,7 +203,8 @@ MoqtControlMessageParser(kDefaultMoqtVersion, /*webtransport=*/true, quic::Perspective::IS_SERVER), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), - session_error_callback_.AsStdFunction(), application_.AsStdFunction()); + session_error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), application_.AsStdFunction()); response_stream->BindStream(&mock_stream_); MoqtPublishNamespace message(5, TrackNamespace({"."})); @@ -221,7 +225,8 @@ MoqtControlMessageParser(kDefaultMoqtVersion, /*webtransport=*/true, quic::Perspective::IS_SERVER), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), - session_error_callback_.AsStdFunction(), application_.AsStdFunction()); + session_error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), application_.AsStdFunction()); response_stream->BindStream(&mock_stream_); MoqtPublishNamespace message; @@ -256,7 +261,8 @@ MoqtControlMessageParser(kDefaultMoqtVersion, /*webtransport=*/true, quic::Perspective::IS_SERVER), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), - session_error_callback_.AsStdFunction(), application_.AsStdFunction()); + session_error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), application_.AsStdFunction()); response_stream->BindStream(&mock_stream_); MoqtPublishNamespace message; @@ -297,7 +303,8 @@ MoqtControlMessageParser(kDefaultMoqtVersion, /*webtransport=*/true, quic::Perspective::IS_SERVER), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), - session_error_callback_.AsStdFunction(), application_.AsStdFunction()); + session_error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), application_.AsStdFunction()); response_stream->BindStream(&mock_stream_); MoqtPublishNamespace message; @@ -323,7 +330,8 @@ MoqtControlMessageParser(kDefaultMoqtVersion, /*webtransport=*/true, quic::Perspective::IS_SERVER), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), - session_error_callback_.AsStdFunction(), application_.AsStdFunction()); + session_error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), application_.AsStdFunction()); response_stream->BindStream(&mock_stream_); MoqtPublishNamespace message; @@ -377,7 +385,8 @@ MoqtControlMessageParser(kDefaultMoqtVersion, /*webtransport=*/true, quic::Perspective::IS_SERVER), add_callback_.AsStdFunction(), remove_callback_.AsStdFunction(), - session_error_callback_.AsStdFunction(), application_.AsStdFunction()); + session_error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), application_.AsStdFunction()); response_stream->BindStream(&mock_stream_); MoqtPublishNamespace message;
diff --git a/quiche/quic/moqt/moqt_publish_stream.cc b/quiche/quic/moqt/moqt_publish_stream.cc index 36293b0..d85dea5 100644 --- a/quiche/quic/moqt/moqt_publish_stream.cc +++ b/quiche/quic/moqt/moqt_publish_stream.cc
@@ -24,6 +24,7 @@ #include "quiche/quic/moqt/moqt_object_subscriber.h" #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" +#include "quiche/common/quiche_status_utils.h" namespace moqt { @@ -32,11 +33,13 @@ const MoqtControlMessageParser& message_parser, LivePublisher::RemoveCallback stream_deleted_callback, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, MoqtResponseCallback response_callback) : MoqtBidiStreamBase(framer, message_parser, std::move(session_error_callback)), response_callback_(std::move(response_callback)), - stream_deleted_callback_(std::move(stream_deleted_callback)) {} + stream_deleted_callback_(std::move(stream_deleted_callback)), + validate_request_id_(std::move(validate_request_id)) {} MoqtPublishRequestStream::~MoqtPublishRequestStream() { if (publisher_ != nullptr) { @@ -85,6 +88,7 @@ absl::Status MoqtPublishRequestStream::OnControlMessage( const MoqtRequestUpdate& message) { + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); MessageParameters in_parameters = message.parameters, out_parameters; out_parameters.largest_object = publisher_->publisher().largest_location(); if (in_parameters.subscription_filter.has_value()) { @@ -102,6 +106,7 @@ const quic::QuicClock* absl_nonnull clock, quic::QuicAlarmFactory* absl_nonnull alarm_factory, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, const MoqtIncomingPublishCallback* absl_nonnull incoming_publish_callback, LiveSubscriber::AddCallback add_callback, LiveSubscriber::RemoveCallback remove_callback) @@ -109,6 +114,7 @@ std::move(session_error_callback)), clock_(clock), alarm_factory_(alarm_factory), + validate_request_id_(std::move(validate_request_id)), incoming_publish_callback_(incoming_publish_callback), add_callback_(std::move(add_callback)), remove_callback_(std::move(remove_callback)), @@ -126,6 +132,7 @@ // Two PUBLISH messages for the same stream. return absl::InvalidArgumentError("Multiple PUBLISH on the same stream"); } + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); if (message.full_track_name.DoesNotExist()) { add_callback_ = nullptr; remove_callback_ = nullptr; @@ -199,6 +206,7 @@ absl::Status MoqtPublishResponseStream::OnControlMessage( const MoqtRequestUpdate& message) { + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); if (subscriber_ == nullptr) { // Stream is already closing. return absl::OkStatus();
diff --git a/quiche/quic/moqt/moqt_publish_stream.h b/quiche/quic/moqt/moqt_publish_stream.h index bb0202d..4beb0c7 100644 --- a/quiche/quic/moqt/moqt_publish_stream.h +++ b/quiche/quic/moqt/moqt_publish_stream.h
@@ -38,6 +38,7 @@ const MoqtControlMessageParser& message_parser, LivePublisher::RemoveCallback stream_deleted_callback, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, MoqtResponseCallback response_callback); ~MoqtPublishRequestStream(); @@ -74,6 +75,7 @@ std::unique_ptr<LivePublisher> publisher_; absl::flat_hash_map<uint64_t, MoqtResponseCallback> pending_updates_; LivePublisher::RemoveCallback stream_deleted_callback_; + ValidateRequestIdCallback validate_request_id_; }; class MoqtPublishResponseStream : public MoqtBidiStreamBase { @@ -84,6 +86,7 @@ const quic::QuicClock* absl_nonnull clock, quic::QuicAlarmFactory* absl_nonnull alarm_factory, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, const MoqtIncomingPublishCallback* absl_nonnull incoming_publish_callback, LiveSubscriber::AddCallback add_callback, LiveSubscriber::RemoveCallback remove_callback); @@ -105,7 +108,7 @@ LiveSubscriber* track() { return subscriber_.get(); } void Detach() override { - if (remove_callback_ != nullptr) { + if (remove_callback_ != nullptr && subscriber_ != nullptr) { LiveSubscriber::RemoveCallback callback = std::move(remove_callback_); remove_callback_ = nullptr; std::move(callback)(subscriber_.get()); @@ -121,6 +124,7 @@ absl::flat_hash_map<uint64_t, MoqtResponseCallback> pending_updates_; const quic::QuicClock* clock_; quic::QuicAlarmFactory* alarm_factory_; + ValidateRequestIdCallback validate_request_id_; const MoqtIncomingPublishCallback* incoming_publish_callback_; LiveSubscriber::AddCallback add_callback_; LiveSubscriber::RemoveCallback remove_callback_;
diff --git a/quiche/quic/moqt/moqt_publish_stream_test.cc b/quiche/quic/moqt/moqt_publish_stream_test.cc index 0e7076d..d339823 100644 --- a/quiche/quic/moqt/moqt_publish_stream_test.cc +++ b/quiche/quic/moqt/moqt_publish_stream_test.cc
@@ -75,10 +75,11 @@ message_parser_(kDefaultMoqtVersion, /*uses_web_transport=*/true, quic::Perspective::IS_CLIENT), track_publisher_(std::make_shared<TestTrackPublisher>(kTrackName)) { + ON_CALL(validate_request_id_, Call).WillByDefault(Return(absl::OkStatus())); // Construct the stream visitor. stream_ = std::make_unique<MoqtPublishRequestStream>( &framer_, message_parser_, deleted_callback_.AsStdFunction(), - error_callback_.AsStdFunction(), + error_callback_.AsStdFunction(), validate_request_id_.AsStdFunction(), [this](std::variant<MessageParameters, MoqtRequestErrorInfo> response) { response_ = response; }); @@ -105,6 +106,7 @@ testing::MockFunction<void(LivePublisher*)> deleted_callback_; testing::StrictMock<testing::MockFunction<void(MoqtError, absl::string_view)>> error_callback_; + testing::MockFunction<absl::Status(uint64_t)> validate_request_id_; MockSessionToPublisherInterface visitor_; webtransport::test::MockSession webtrans_; quic::MockClock mock_clock_; @@ -239,10 +241,11 @@ quic::Perspective::IS_SERVER), incoming_publish_callback_( incoming_publish_callback_mock_.AsStdFunction()) { + ON_CALL(validate_request_id_, Call).WillByDefault(Return(absl::OkStatus())); stream_ = std::make_unique<MoqtPublishResponseStream>( &framer_, message_parser_, &mock_clock_, &mock_alarm_factory_, - error_callback_.AsStdFunction(), &incoming_publish_callback_, - mock_add_callback_.AsStdFunction(), + error_callback_.AsStdFunction(), validate_request_id_.AsStdFunction(), + &incoming_publish_callback_, mock_add_callback_.AsStdFunction(), mock_remove_callback_.AsStdFunction()); stream_->BindStream(&mock_stream_); EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true)); @@ -259,6 +262,7 @@ quic::test::MockAlarmFactory mock_alarm_factory_; testing::StrictMock<testing::MockFunction<void(MoqtError, absl::string_view)>> error_callback_; + testing::MockFunction<absl::Status(uint64_t)> validate_request_id_; testing::MockFunction<SubscribeVisitor*( const FullTrackName&, const MessageParameters&, const TrackProperties&, @@ -607,8 +611,8 @@ webtransport::test::MockStream mock_stream2; MoqtPublishResponseStream stream2( &framer_, message_parser_, &mock_clock_, &mock_alarm_factory_, - error_callback2.AsStdFunction(), &incoming_publish_callback2, - mock_add_callback2.AsStdFunction(), + error_callback2.AsStdFunction(), validate_request_id_.AsStdFunction(), + &incoming_publish_callback2, mock_add_callback2.AsStdFunction(), mock_remove_callback2.AsStdFunction()); stream2.BindStream(&mock_stream2); EXPECT_CALL(mock_stream2, CanWrite).WillRepeatedly(Return(true));
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index 4999460..4fa267c 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -107,6 +107,7 @@ } if (parameters_.perspective == Perspective::IS_SERVER) { next_request_id_ = 1; + next_incoming_request_id_ = 0; } QUICHE_DCHECK(parameters_.moqt_implementation.empty()); parameters_.moqt_implementation = kImplementationName; @@ -630,6 +631,13 @@ } session->Error(code, reason); }, + [weak_session = GetWeakPtr()](uint64_t request_id) { + MoqtSession* session = MoqtSessionFromWeakPtr(weak_session); + if (session == nullptr) { + return absl::NotFoundError("Session is gone"); + } + return session->ValidateNewIncomingRequestId(request_id); + }, std::move(response_callback)); auto publish_state = std::make_unique<LivePublisher>( framer_, publisher, stream_visitor.get(), next_request_id_, @@ -903,16 +911,23 @@ } } -bool MoqtSession::ValidateRequestId(uint64_t request_id) { +absl::Status MoqtSession::ValidateNewIncomingRequestId(uint64_t request_id) { if ((request_id % 2 == 0) != (parameters_.perspective == Perspective::IS_SERVER)) { - QUICHE_DLOG(INFO) << ENDPOINT << "Request ID evenness incorrect"; + QUIC_DLOG(INFO) << parameters_.perspective + << "Request ID evenness incorrect"; Error(MoqtError::kInvalidRequestId, "Request ID evenness incorrect"); - return false; + return absl::InvalidArgumentError("Request ID evenness incorrect"); } - // TODO(martinduke): Write new checks for duplicate request IDs. It's - // probably best to track the largest observed plus a set of holes. - return true; + // Check the one request ID map we have for the value. Most active request IDs + // are not tracked at the session level. + if (published_subscriptions_.contains(request_id)) { + QUIC_DLOG(INFO) << parameters_.perspective << "Duplicate Request ID"; + Error(MoqtError::kInvalidRequestId, "Duplicate Request ID"); + return absl::InvalidArgumentError("Duplicate Request ID"); + } + next_incoming_request_id_ = request_id + 2; + return absl::OkStatus(); } void MoqtSession::UnknownBidiStream::OnCanRead() { @@ -959,6 +974,13 @@ session->Error(code, reason); } }, + [weakptr = session_->GetWeakPtr()](uint64_t request_id) { + MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); + if (session == nullptr) { + return absl::NotFoundError("Session is gone"); + } + return session->ValidateNewIncomingRequestId(request_id); + }, session_->callbacks_.incoming_subscribe_namespace_callback); namespace_stream->BindStream(std::move(parser_)); MoqtSubscribeNamespaceResponseStream* temp_stream = @@ -1011,6 +1033,13 @@ session->Error(code, reason); } }, + [weakptr = session_->GetWeakPtr()](uint64_t request_id) { + MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); + if (session == nullptr) { + return absl::NotFoundError("Session is gone"); + } + return session->ValidateNewIncomingRequestId(request_id); + }, [weakptr = session_->GetWeakPtr()]( const TrackNamespace& track_namespace, const MessageParameters* absl_nullable parameters, @@ -1042,6 +1071,13 @@ session->Error(code, reason); } }, + [weakptr = session_->GetWeakPtr()](uint64_t request_id) { + MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); + if (session == nullptr) { + return absl::NotFoundError("Session is gone"); + } + return session->ValidateNewIncomingRequestId(request_id); + }, &session_->callbacks_.incoming_publish_callback, [weakptr = session_->GetWeakPtr()](LiveSubscriber* track) { MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); @@ -1127,6 +1163,13 @@ session->Error(code, reason); } }, + [weakptr = session_->GetWeakPtr()](uint64_t request_id) { + MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); + if (session == nullptr) { + return absl::NotFoundError("Session is gone"); + } + return session->ValidateNewIncomingRequestId(request_id); + }, session_->weak_ptr_factory_for_publishers_.Create()); subscribe_stream->BindStream(std::move(parser_)); MoqtSubscribeResponseStream* temp_stream = subscribe_stream.get(); @@ -1147,6 +1190,13 @@ session->Error(code, reason); } }, + [weakptr = session_->GetWeakPtr()](uint64_t request_id) { + MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); + if (session == nullptr) { + return absl::NotFoundError("Session is gone"); + } + return session->ValidateNewIncomingRequestId(request_id); + }, session_->weak_ptr_factory_for_publishers_.Create()); track_status_stream->BindStream(std::move(parser_)); MoqtTrackStatusResponseStream* temp_stream = track_status_stream.get(); @@ -1165,6 +1215,13 @@ session->Error(code, reason); } }, + [weakptr = session_->GetWeakPtr()](uint64_t request_id) { + MoqtSession* session = MoqtSessionFromWeakPtr(weakptr); + if (session == nullptr) { + return absl::NotFoundError("Session is gone"); + } + return session->ValidateNewIncomingRequestId(request_id); + }, // OpenStreamCallback [weakptr = session_->GetWeakPtr()](webtransport::StreamId stream_id, MoqtTrackPriority priority) {
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h index a545325..7aba415 100644 --- a/quiche/quic/moqt/moqt_session.h +++ b/quiche/quic/moqt/moqt_session.h
@@ -5,6 +5,7 @@ #ifndef QUICHE_QUIC_MOQT_MOQT_SESSION_H_ #define QUICHE_QUIC_MOQT_MOQT_SESSION_H_ +#include <cstddef> #include <cstdint> #include <memory> #include <optional> @@ -14,14 +15,15 @@ #include "absl/base/casts.h" #include "absl/base/nullability.h" -#include "absl/cleanup/cleanup.h" #include "absl/container/btree_map.h" +#include "absl/container/btree_set.h" #include "absl/container/flat_hash_map.h" #include "absl/container/flat_hash_set.h" #include "absl/status/status.h" #include "absl/strings/string_view.h" #include "quiche/quic/core/quic_alarm.h" #include "quiche/quic/core/quic_alarm_factory.h" +#include "quiche/quic/core/quic_constants.h" #include "quiche/quic/core/quic_time.h" #include "quiche/quic/core/quic_types.h" #include "quiche/quic/moqt/moqt_bidi_stream.h" @@ -374,9 +376,9 @@ LiveSubscriber* SubscribeByName(const FullTrackName& track_name); MoqtFetchRequestStream* FetchById(uint64_t request_id); - // Checks that a subscribe ID from a SUBSCRIBE or FETCH is valid, and throws - // a session error if is not. - bool ValidateRequestId(uint64_t request_id); + // Checks that an incoming request ID is valid, and throws a session error if + // is not. + absl::Status ValidateNewIncomingRequestId(uint64_t request_id); // Sends an OBJECT_ACK message for a specific subscribe ID. void SendObjectAck(FullTrackName track_name, uint64_t group_id, @@ -448,8 +450,12 @@ // All outgoing SUBSCRIBE and incoming PUBLISH, indexed by track name. absl::flat_hash_map<FullTrackName, LiveSubscriber*> subscribe_by_name_; - // The next subscribe ID that the local endpoint can send. + // REQUEST_ID state. + // The next request ID that the local endpoint can send. Use NextRequestId() + // to obtain a value and automatically increment the next value. uint64_t next_request_id_ = 0; + // The next expected incoming request ID, used to compose draft-18 GOAWAY. + uint64_t next_incoming_request_id_ = 1; // All open incoming subscriptions, indexed by track name, used to check for // duplicates. @@ -466,8 +472,6 @@ absl::btree_multimap<MoqtTrackPriority, std::variant<FullTrackName, webtransport::StreamId>> requests_with_queued_streams_; - // This is only used to check for track_alias collisions. - absl::flat_hash_set<uint64_t> used_track_aliases_; uint64_t next_local_track_alias_ = 0; // Monitoring interfaces for expected incoming subscriptions.
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index e1ea4e6..8bc0162 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -49,6 +49,7 @@ #include "quiche/common/quiche_data_reader.h" #include "quiche/common/quiche_mem_slice.h" #include "quiche/common/quiche_weak_ptr.h" +#include "quiche/common/test_tools/quiche_test_utils.h" #include "quiche/web_transport/test_tools/in_memory_stream.h" #include "quiche/web_transport/test_tools/mock_web_transport.h" #include "quiche/web_transport/web_transport.h" @@ -873,10 +874,28 @@ } TEST_F(MoqtSessionTest, RequestIdWrongLsb) { - // TODO(martinduke): Implement this test. + // Client session expects odd request IDs from the server. + EXPECT_CALL(mock_session_, + CloseSession(static_cast<uint64_t>(MoqtError::kInvalidRequestId), + "Request ID evenness incorrect")); + EXPECT_FALSE( + MoqtSessionPeer::ValidateNewIncomingRequestId(&session_, 0).ok()); + + // Server session expects even request IDs from the client. + MoqtSessionParameters server_parameters(quic::Perspective::IS_SERVER); + MoqtSession server_session(&mock_session_, server_parameters, + std::make_unique<quic::test::TestAlarmFactory>(), + session_callbacks_.AsSessionCallbacks()); + EXPECT_CALL(mock_session_, + CloseSession(static_cast<uint64_t>(MoqtError::kInvalidRequestId), + "Request ID evenness incorrect")); + EXPECT_FALSE( + MoqtSessionPeer::ValidateNewIncomingRequestId(&server_session, 1).ok()); + QUICHE_EXPECT_OK( + MoqtSessionPeer::ValidateNewIncomingRequestId(&server_session, 0)); } -TEST_F(MoqtSessionTest, SubscribeIdNotIncreasing) { +TEST_F(MoqtSessionTest, RequestIdNotIncreasing) { MoqtSubscribe request = DefaultSubscribe(); bidi_wrapper_ = std::make_unique<MoqtBidiStreamTestWrapper>( ResponseStream(kSubscribeByte)); @@ -892,8 +911,9 @@ webtransport::test::MockStream bidi_stream_2; auto bidi_wrapper_2 = std::make_unique<MoqtBidiStreamTestWrapper>( ResponseStream(kSubscribeByte, &bidi_stream_2)); - EXPECT_CALL(bidi_stream_2, - Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _)); + EXPECT_CALL(mock_session_, + CloseSession(static_cast<uint64_t>(MoqtError::kInvalidRequestId), + "Duplicate Request ID")); bidi_wrapper_2->ReceiveMessage(request); } @@ -2427,7 +2447,7 @@ webtransport::test::InMemoryStreamWithWriteBuffer bidi_stream(0); MoqtFramer client_framer(session_parameters.using_webtrans, quic::Perspective::IS_CLIENT); - MoqtTrackStatus track_status = DefaultSubscribe(); + MoqtTrackStatus track_status = DefaultLocalSubscribe(); quiche::QuicheBuffer serialized_track_status = client_framer.SerializeTrackStatus(track_status); bidi_stream.Receive(serialized_track_status.AsStringView(),
diff --git a/quiche/quic/moqt/moqt_subscribe_stream.cc b/quiche/quic/moqt/moqt_subscribe_stream.cc index d0ccd90..cddee8c 100644 --- a/quiche/quic/moqt/moqt_subscribe_stream.cc +++ b/quiche/quic/moqt/moqt_subscribe_stream.cc
@@ -25,6 +25,7 @@ #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_publisher.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" +#include "quiche/common/quiche_status_utils.h" #include "quiche/common/quiche_weak_ptr.h" namespace moqt { @@ -179,12 +180,14 @@ LivePublisher::AddCallback add_callback, LivePublisher::RemoveCallback remove_callback, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, quiche::QuicheWeakPtr<SessionToPublisherInterface> session) : MoqtBidiStreamBase(framer, message_parser, std::move(session_error_callback)), track_alias_(track_alias), add_callback_(std::move(add_callback)), remove_callback_(std::move(remove_callback)), + validate_request_id_(std::move(validate_request_id)), session_(std::move(session)) {} absl::Status MoqtSubscribeResponseStream::OnRawControlMessage( @@ -199,6 +202,7 @@ return absl::InvalidArgumentError( "SUBSCRIBE received on stream that already has a subscription"); } + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); QUIC_DLOG(INFO) << "Received a SUBSCRIBE for " << message.full_track_name; if (session() == nullptr) { return absl::OkStatus(); @@ -235,6 +239,7 @@ absl::Status MoqtSubscribeResponseStream::OnControlMessage( const MoqtRequestUpdate& message) { + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); if (subscription_ == nullptr) { QUICHE_BUG(INFO) << "Received REQUEST_UPDATE, no subscription state"; return SendRequestError(RequestErrorCode::kInternalError, std::nullopt,
diff --git a/quiche/quic/moqt/moqt_subscribe_stream.h b/quiche/quic/moqt/moqt_subscribe_stream.h index 65e2954..a2406ae 100644 --- a/quiche/quic/moqt/moqt_subscribe_stream.h +++ b/quiche/quic/moqt/moqt_subscribe_stream.h
@@ -67,6 +67,7 @@ LivePublisher::AddCallback add_callback, LivePublisher::RemoveCallback remove_callback, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, quiche::QuicheWeakPtr<SessionToPublisherInterface> session); ~MoqtSubscribeResponseStream() { if (subscription_ != nullptr) { @@ -98,6 +99,7 @@ std::unique_ptr<LivePublisher> subscription_; LivePublisher::AddCallback add_callback_; LivePublisher::RemoveCallback remove_callback_; + ValidateRequestIdCallback validate_request_id_; quiche::QuicheWeakPtr<SessionToPublisherInterface> session_; };
diff --git a/quiche/quic/moqt/moqt_subscribe_stream_test.cc b/quiche/quic/moqt/moqt_subscribe_stream_test.cc index dc110d8..ceb6200 100644 --- a/quiche/quic/moqt/moqt_subscribe_stream_test.cc +++ b/quiche/quic/moqt/moqt_subscribe_stream_test.cc
@@ -244,10 +244,13 @@ message_parser_(kDefaultMoqtVersion, /*uses_web_transport=*/true, quic::Perspective::IS_SERVER), track_publisher_(std::make_shared<TestTrackPublisher>(kTrackName)) { + ON_CALL(mock_validate_request_id_, Call) + .WillByDefault(Return(absl::OkStatus())); stream_ = std::make_unique<MoqtSubscribeResponseStream>( &framer_, message_parser_, kTrackAlias, mock_add_callback_.AsStdFunction(), mock_remove_callback_.AsStdFunction(), error_callback_.AsStdFunction(), + mock_validate_request_id_.AsStdFunction(), visitor_.weak_ptr_factory_.Create()); stream_->BindStream(&mock_stream_); EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(testing::Return(true)); @@ -260,6 +263,7 @@ FullTrackName kTrackName{"foo", "bar"}; std::shared_ptr<TestTrackPublisher> track_publisher_; testing::MockFunction<void(MoqtError, absl::string_view)> error_callback_; + testing::MockFunction<absl::Status(uint64_t)> mock_validate_request_id_; testing::MockFunction<bool(LivePublisher*)> mock_add_callback_; testing::MockFunction<void(LivePublisher*)> mock_remove_callback_; MockSessionToPublisherInterface visitor_;
diff --git a/quiche/quic/moqt/moqt_track_status_stream.cc b/quiche/quic/moqt/moqt_track_status_stream.cc index 64514f3..44be988 100644 --- a/quiche/quic/moqt/moqt_track_status_stream.cc +++ b/quiche/quic/moqt/moqt_track_status_stream.cc
@@ -22,6 +22,7 @@ #include "quiche/quic/moqt/moqt_parser.h" #include "quiche/quic/moqt/moqt_session_callbacks.h" #include "quiche/common/platform/api/quiche_logging.h" +#include "quiche/common/quiche_status_utils.h" #include "quiche/common/quiche_weak_ptr.h" namespace moqt { @@ -94,9 +95,11 @@ MoqtFramer* absl_nonnull framer, const MoqtControlMessageParser& message_parser, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, quiche::QuicheWeakPtr<SessionToPublisherInterface> session) : MoqtBidiStreamBase(framer, message_parser, std::move(session_error_callback)), + validate_request_id_(std::move(validate_request_id)), session_(session) {} absl::Status MoqtTrackStatusResponseStream::OnRawControlMessage( @@ -110,6 +113,7 @@ if (request_id_.has_value()) { return absl::InvalidArgumentError("Duplicate TRACK_STATUS received"); } + QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id)); if (session() == nullptr) { return absl::InternalError("Session unavailable"); }
diff --git a/quiche/quic/moqt/moqt_track_status_stream.h b/quiche/quic/moqt/moqt_track_status_stream.h index d2bc178..4cc8e67 100644 --- a/quiche/quic/moqt/moqt_track_status_stream.h +++ b/quiche/quic/moqt/moqt_track_status_stream.h
@@ -64,6 +64,7 @@ MoqtFramer* absl_nonnull framer, const MoqtControlMessageParser& message_parser, SessionErrorCallback session_error_callback, + ValidateRequestIdCallback validate_request_id, quiche::QuicheWeakPtr<SessionToPublisherInterface> session); ~MoqtTrackStatusResponseStream() { Detach(); } @@ -95,6 +96,7 @@ } std::optional<uint64_t> request_id_; + ValidateRequestIdCallback validate_request_id_; const quiche::QuicheWeakPtr<SessionToPublisherInterface> session_; std::shared_ptr<MoqtTrackPublisher> publisher_ = nullptr; };
diff --git a/quiche/quic/moqt/moqt_track_status_stream_test.cc b/quiche/quic/moqt/moqt_track_status_stream_test.cc index 6e2e281..d527775 100644 --- a/quiche/quic/moqt/moqt_track_status_stream_test.cc +++ b/quiche/quic/moqt/moqt_track_status_stream_test.cc
@@ -199,11 +199,14 @@ message_parser_(kDefaultMoqtVersion, /*uses_web_transport=*/true, quic::Perspective::IS_SERVER), track_name_("foo", "bar"), - mock_publisher_(track_name_) {} + mock_publisher_(track_name_) { + ON_CALL(validate_request_id_, Call).WillByDefault(Return(absl::OkStatus())); + } MoqtTrackStatusResponseStream CreateStream() { return MoqtTrackStatusResponseStream( &framer_, message_parser_, session_error_callback_.AsStdFunction(), + validate_request_id_.AsStdFunction(), session_.weak_ptr_factory_.Create()); } @@ -214,6 +217,7 @@ FullTrackName track_name_; StrictMock<testing::MockFunction<void(MoqtError, absl::string_view)>> session_error_callback_; + testing::MockFunction<absl::Status(uint64_t)> validate_request_id_; MockSessionToPublisherInterface session_; MockTrackPublisher mock_publisher_; StrictMock<webtransport::test::MockStream> mock_stream_;
diff --git a/quiche/quic/moqt/test_tools/moqt_session_peer.h b/quiche/quic/moqt/test_tools/moqt_session_peer.h index 8f07ab2..6f7d673 100644 --- a/quiche/quic/moqt/test_tools/moqt_session_peer.h +++ b/quiche/quic/moqt/test_tools/moqt_session_peer.h
@@ -160,8 +160,9 @@ session->peer_setup_received_ = value; } - static void ValidateRequestId(MoqtSession* session, uint64_t id) { - session->ValidateRequestId(id); + static absl::Status ValidateNewIncomingRequestId(MoqtSession* session, + uint64_t id) { + return session->ValidateNewIncomingRequestId(id); } static quic::QuicAlarmFactory* GetAlarmFactory(MoqtSession* session) {