snap: cache early dtls packets in race condition

If SCTP is negotiated after the DTLS handshake has been completed there exists a race condition with SNAP wherein the answerer can start opening  channels (and send data) before the answer with the sctp-init arrives at the offerer.

Cache a limited amount of decrypted and authenticated packets and replay them after receiving the answer. This problem also exists for non-snap but is mitigated by the symmetric open on both sides so the loss in one direction does not matter much.

Note: doing a "half-open" on the offerer socket does not work as it can not send acknowledgements of received packets due to the missing peer verification tag.

Bug: webrtc:426480601
Change-Id: Ic8ee193b6ca8774943e19e37073814af144ce7d4
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/478160
Reviewed-by: Harald Alvestrand <hta@webrtc.org>
Commit-Queue: Philipp Hancke <philipp.hancke@googlemail.com>
Reviewed-by: Victor Boivie <boivie@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#48041}
diff --git a/media/BUILD.gn b/media/BUILD.gn
index db6d02b..98ab48a 100644
--- a/media/BUILD.gn
+++ b/media/BUILD.gn
@@ -755,6 +755,7 @@
       "../p2p:dtls_transport_internal",
       "../p2p:packet_transport_internal",
       "../rtc_base:async_packet_socket",
+      "../rtc_base:buffer",
       "../rtc_base:checks",
       "../rtc_base:copy_on_write_buffer",
       "../rtc_base:event_tracer",
diff --git a/media/sctp/dcsctp_transport.cc b/media/sctp/dcsctp_transport.cc
index 907140c..2bdfe26 100644
--- a/media/sctp/dcsctp_transport.cc
+++ b/media/sctp/dcsctp_transport.cc
@@ -44,6 +44,7 @@
 #include "p2p/base/packet_transport_internal.h"
 #include "p2p/dtls/dtls_transport_internal.h"
 #include "rtc_base/async_packet_socket.h"
+#include "rtc_base/buffer.h"
 #include "rtc_base/checks.h"
 #include "rtc_base/copy_on_write_buffer.h"
 #include "rtc_base/logging.h"
@@ -751,13 +752,22 @@
   }
 
   RTC_DLOG(LS_VERBOSE) << debug_name_ << "->OnTransportReadPacket(), length="
-                       << packet.payload().size();
+                       << packet.payload().size() << " socket=" << !!socket_;
   if (socket_) {
     socket_->ReceivePacket(packet.payload());
+    return;
   }
+
+  // Buffering decrypted packets is only required in an edge case of SNAP.
+  if (early_received_packets_.size() >= kMaxEarlyReceivedPackets) {
+    early_received_packets_.erase(early_received_packets_.begin());
+  }
+  early_received_packets_.emplace_back(packet.payload().data(),
+                                       packet.payload().size());
 }
 
 void DcSctpTransport::MaybeConnectSocket() {
+  RTC_DCHECK_RUN_ON(network_thread_);
   RTC_DLOG(LS_VERBOSE)
       << debug_name_ << "->MaybeConnectSocket(), writable="
       << (transport_ ? std::to_string(transport_->writable()) : "UNSET")
@@ -767,9 +777,17 @@
   if (transport_ && transport_->writable() && socket_ &&
       socket_->state() == dcsctp::SocketState::kClosed) {
     if (!(local_init_.has_value() && remote_init_.has_value())) {
-      return socket_->Connect();
+      socket_->Connect();
+      return;
     }
     socket_->ConnectWithConnectionToken(*local_init_, *remote_init_);
+    // Replay any datachannel packets that arrived before the socket existed.
+    std::vector<webrtc::Buffer> packets = std::move(early_received_packets_);
+    early_received_packets_.clear();
+    for (const webrtc::Buffer& packet : packets) {
+      socket_->ReceivePacket(
+          std::span<const uint8_t>(packet.data(), packet.size()));
+    }
   }
 }
 
@@ -806,4 +824,9 @@
       });
 }
 
+size_t DcSctpTransport::EarlyReceivedPacketCountForTesting() const {
+  RTC_DCHECK_RUN_ON(network_thread_);
+  return early_received_packets_.size();
+}
+
 }  // namespace webrtc
diff --git a/media/sctp/dcsctp_transport.h b/media/sctp/dcsctp_transport.h
index bf945c9..26dd14c 100644
--- a/media/sctp/dcsctp_transport.h
+++ b/media/sctp/dcsctp_transport.h
@@ -39,6 +39,7 @@
 #include "net/dcsctp/timer/task_queue_timeout.h"
 #include "p2p/base/packet_transport_internal.h"
 #include "p2p/dtls/dtls_transport_internal.h"
+#include "rtc_base/buffer.h"
 #include "rtc_base/containers/flat_map.h"
 #include "rtc_base/copy_on_write_buffer.h"
 #include "rtc_base/network/received_packet.h"
@@ -80,6 +81,10 @@
 
   static std::vector<uint8_t> GenerateConnectionToken(const Environment& env);
 
+  // Returns the number of packets currently buffered while waiting for the
+  // SCTP socket to be created. See `early_received_packets_`.
+  size_t EarlyReceivedPacketCountForTesting() const override;
+
  private:
   // dcsctp::DcSctpSocketCallbacks
   dcsctp::SendPacketStatus SendPacketWithStatus(
@@ -155,6 +160,15 @@
   static dcsctp::DcSctpOptions CreateDcSctpOptions(
       const SctpOptions& options,
       const FieldTrialsView& field_trials);
+
+  // With SNAP the answerer can (if datachannels are negotiated after the DTLS
+  // handshake) start sending datachannel packets as soon as it has processed
+  // the offer. This causes a race condition where those packets arrive before
+  // the answer. Buffering (a limited amount of) them avoids a resend after
+  // timeout.
+  static constexpr size_t kMaxEarlyReceivedPackets = 32;
+  std::vector<webrtc::Buffer> early_received_packets_
+      RTC_GUARDED_BY(network_thread_);
 };
 
 }  // namespace webrtc
diff --git a/media/sctp/sctp_transport_internal.h b/media/sctp/sctp_transport_internal.h
index 8170aea..d535f99 100644
--- a/media/sctp/sctp_transport_internal.h
+++ b/media/sctp/sctp_transport_internal.h
@@ -88,6 +88,8 @@
   virtual size_t buffered_amount(int sid) const = 0;
   virtual size_t buffered_amount_low_threshold(int sid) const = 0;
   virtual void SetBufferedAmountLowThreshold(int sid, size_t bytes) = 0;
+
+  virtual size_t EarlyReceivedPacketCountForTesting() const = 0;
 };
 
 }  //  namespace webrtc
diff --git a/pc/BUILD.gn b/pc/BUILD.gn
index 8f0e8f3..ea46273 100644
--- a/pc/BUILD.gn
+++ b/pc/BUILD.gn
@@ -3831,6 +3831,7 @@
       ":integration_test_helpers",
       ":media_session",
       ":pc_test_utils",
+      ":sctp_transport",
       ":session_description",
       "../api:data_channel_interface",
       "../api:dtls_transport_interface",
diff --git a/pc/data_channel_integrationtest.cc b/pc/data_channel_integrationtest.cc
index d178e3c..043a352 100644
--- a/pc/data_channel_integrationtest.cc
+++ b/pc/data_channel_integrationtest.cc
@@ -36,6 +36,7 @@
 #include "p2p/base/transport_info.h"
 #include "p2p/test/test_turn_server.h"
 #include "pc/media_session.h"
+#include "pc/sctp_transport.h"
 #include "pc/session_description.h"
 #include "pc/test/fake_rtc_certificate_generator.h"
 #include "pc/test/integration_test_helpers.h"
@@ -59,9 +60,12 @@
 namespace {
 
 using ::testing::Eq;
+using ::testing::IsEmpty;
 using ::testing::IsTrue;
 using ::testing::Ne;
+using ::testing::Not;
 using ::testing::NotNull;
+using ::testing::SizeIs;
 using ::testing::ValuesIn;
 
 // All tests in this file require SCTP support.
@@ -881,6 +885,112 @@
       IsRtcOk());
 }
 
+// Fixture for tests of draft-hancke-tsvwg-snap, which carries the SCTP-init
+// cookie in the SDP (a=sctp-init) and is gated behind the WebRTC-Sctp-Snap
+// field trial.
+class DataChannelIntegrationTestWithSctpSnap
+    : public PeerConnectionIntegrationBaseTest {
+ protected:
+  DataChannelIntegrationTestWithSctpSnap()
+      : PeerConnectionIntegrationBaseTest(SdpSemantics::kUnifiedPlan) {
+    // Must be set before the PeerConnectionWrappers are created.
+    SetFieldTrials("WebRTC-Sctp-Snap/Enabled/");
+  }
+};
+
+TEST_F(DataChannelIntegrationTestWithSctpSnap,
+       EarlyDataChannelPacketsAreBufferedUntilAnswerApplied) {
+  ASSERT_TRUE(CreatePeerConnectionWrappers());
+  ConnectFakeSignaling();
+
+  // Phase 1: establish an audio/video connection (no data channel yet).
+  caller()->AddAudioVideoTracks();
+  callee()->AddAudioVideoTracks();
+  caller()->CreateAndSetAndSignalOffer();
+  ASSERT_TRUE(WaitUntil([&] { return SignalingStateStable(); }));
+  MediaExpectations media_expectations;
+  media_expectations.ExpectBidirectionalAudioAndVideo();
+  ASSERT_TRUE(ExpectNewFrames(media_expectations));
+
+  // Phase 2: add a data channel and send an offer with an sctp-init, but
+  // capture the answer and never apply it on the caller.
+  caller()->CreateDataChannel();
+  callee()->CreateDataChannel();
+  std::string captured_answer;
+  caller()->SetReceivedSdpMunger(
+      [&](std::unique_ptr<SessionDescriptionInterface>& sdp) {
+        sdp->ToString(&captured_answer);
+        sdp = nullptr;
+      });
+  caller()->CreateAndSetAndSignalOffer();
+  EXPECT_EQ(caller()->pc()->signaling_state(),
+            PeerConnectionInterface::kHaveLocalOffer);
+  EXPECT_THAT(captured_answer, Not(IsEmpty()));
+
+  // Caller has no SCTP socket yet, callee has and is sending data
+  // which must be cached by the caller.
+  ASSERT_TRUE(WaitUntil([&] {
+    auto transport = callee()->pc()->GetSctpTransport();
+    return transport &&
+           transport->Information().state() == SctpTransportState::kConnected;
+  }));
+  EXPECT_FALSE(caller()->data_observer()->IsOpen());
+  EXPECT_TRUE(callee()->data_observer()->IsOpen());
+
+  auto caller_cached_packet_count = [&]() -> size_t {
+    return network_thread()->BlockingCall([&]() -> size_t {
+      auto* sctp_transport =
+          static_cast<SctpTransport*>(caller()->pc()->GetSctpTransport().get());
+      if (!sctp_transport) {
+        return 0;
+      }
+      return sctp_transport->internal()->EarlyReceivedPacketCountForTesting();
+    });
+  };
+
+  // Before any application data is sent, the caller has cached exactly one
+  // packet: the DCEP "open" message for the data channel.
+  EXPECT_TRUE(WaitUntil([&] { return caller_cached_packet_count() == 1u; }));
+
+  // Send many small numbered messages. The caller has no SCTP socket yet, so it
+  // never acknowledges them; the callee keeps retransmitting and the caller
+  // caches every (re)transmitted packet. The early-packet buffer therefore
+  // fills to its cap and is bounded there, never growing beyond it.
+  constexpr int kNumMessages = 64;
+  for (int i = 0; i < kNumMessages; ++i) {
+    callee()->data_channel()->Send(DataBuffer(std::to_string(i)));
+  }
+  EXPECT_TRUE(WaitUntil([&] { return caller_cached_packet_count() == 32u; }));
+
+  // Phase 3: apply the captured answer. We expect two open data channels
+  // on each side.
+  caller()->SetReceivedSdpMunger(nullptr);
+  caller()->ReceiveSdpMessage(SdpType::kAnswer, captured_answer);
+  ASSERT_TRUE(WaitUntil([&] { return SignalingStateStable(); }));
+
+  ASSERT_THAT(WaitUntil([&] { return caller()->data_channels(); }, SizeIs(2)),
+              IsRtcOk());
+  for (const auto& observer : caller()->data_observers()) {
+    EXPECT_TRUE(WaitUntil([&] { return observer->IsOpen(); }));
+  }
+  ASSERT_THAT(WaitUntil([&] { return callee()->data_channels(); }, SizeIs(2)),
+              IsRtcOk());
+  for (const auto& observer : callee()->data_observers()) {
+    EXPECT_TRUE(WaitUntil([&] { return observer->IsOpen(); }));
+  }
+
+  // The caller must receive all kNumMessages messages, in order, on the
+  // channel negotiated in-band from the callee. Some of them buffered,
+  // some as resends.
+  MockDataChannelObserver* receiver = caller()->data_observers().back().get();
+  ASSERT_TRUE(WaitUntil([&] {
+    return static_cast<int>(receiver->received_message_count()) == kNumMessages;
+  }));
+  for (int i = 0; i < kNumMessages; ++i) {
+    EXPECT_EQ(receiver->messages()[i].data, std::to_string(i));
+  }
+}
+
 // Set up a connection initially just using SCTP data channels, later
 // upgrading to audio/video, ensuring frames are received end-to-end.
 // Effectively the inverse of the test above. This was broken in M57; see
diff --git a/pc/sctp_transport_unittest.cc b/pc/sctp_transport_unittest.cc
index 5349538..b3be484 100644
--- a/pc/sctp_transport_unittest.cc
+++ b/pc/sctp_transport_unittest.cc
@@ -76,6 +76,7 @@
   size_t buffered_amount(int sid) const override { return 0; }
   size_t buffered_amount_low_threshold(int sid) const override { return 0; }
   void SetBufferedAmountLowThreshold(int sid, size_t bytes) override {}
+  size_t EarlyReceivedPacketCountForTesting() const override { return 0; }
 
   void SendSignalAssociationChangeCommunicationUp() {
     ASSERT_TRUE(on_connected_callback_);
diff --git a/test/pc/sctp/fake_sctp_transport.h b/test/pc/sctp/fake_sctp_transport.h
index cb3cbcd..da11cdd 100644
--- a/test/pc/sctp/fake_sctp_transport.h
+++ b/test/pc/sctp/fake_sctp_transport.h
@@ -68,6 +68,7 @@
   size_t buffered_amount(int sid) const override { return 0; }
   size_t buffered_amount_low_threshold(int sid) const override { return 0; }
   void SetBufferedAmountLowThreshold(int sid, size_t bytes) override {}
+  size_t EarlyReceivedPacketCountForTesting() const override { return 0; }
   int local_port() const {
     RTC_DCHECK(local_port_);
     return *local_port_;