* Copyright (c) 2021 The WebRTC project authors. All Rights Reserved.
*
* Use of this source code is governed by a BSD-style license
* that can be found in the LICENSE file in the root of the source
* tree. An additional intellectual property rights grant can be found
* in the file PATENTS. All contributing project authors may
* be found in the AUTHORS file in the root of the source tree.
*/
#include "net/dcsctp/socket/dcsctp_socket.h"
#include <algorithm>
#include <cstdint>
#include <limits>
#include <memory>
#include <string>
#include <utility>
#include <vector>
#include "absl/functional/bind_front.h"
#include "absl/memory/memory.h"
#include "absl/strings/string_view.h"
#include "absl/types/optional.h"
#include "api/array_view.h"
#include "api/task_queue/task_queue_base.h"
#include "net/dcsctp/packet/chunk/abort_chunk.h"
#include "net/dcsctp/packet/chunk/chunk.h"
#include "net/dcsctp/packet/chunk/cookie_ack_chunk.h"
#include "net/dcsctp/packet/chunk/cookie_echo_chunk.h"
#include "net/dcsctp/packet/chunk/data_chunk.h"
#include "net/dcsctp/packet/chunk/data_common.h"
#include "net/dcsctp/packet/chunk/error_chunk.h"
#include "net/dcsctp/packet/chunk/forward_tsn_chunk.h"
#include "net/dcsctp/packet/chunk/forward_tsn_common.h"
#include "net/dcsctp/packet/chunk/heartbeat_ack_chunk.h"
#include "net/dcsctp/packet/chunk/heartbeat_request_chunk.h"
#include "net/dcsctp/packet/chunk/idata_chunk.h"
#include "net/dcsctp/packet/chunk/iforward_tsn_chunk.h"
#include "net/dcsctp/packet/chunk/init_ack_chunk.h"
#include "net/dcsctp/packet/chunk/init_chunk.h"
#include "net/dcsctp/packet/chunk/reconfig_chunk.h"
#include "net/dcsctp/packet/chunk/sack_chunk.h"
#include "net/dcsctp/packet/chunk/shutdown_ack_chunk.h"
#include "net/dcsctp/packet/chunk/shutdown_chunk.h"
#include "net/dcsctp/packet/chunk/shutdown_complete_chunk.h"
#include "net/dcsctp/packet/chunk_validators.h"
#include "net/dcsctp/packet/data.h"
#include "net/dcsctp/packet/error_cause/cookie_received_while_shutting_down_cause.h"
#include "net/dcsctp/packet/error_cause/error_cause.h"
#include "net/dcsctp/packet/error_cause/no_user_data_cause.h"
#include "net/dcsctp/packet/error_cause/out_of_resource_error_cause.h"
#include "net/dcsctp/packet/error_cause/protocol_violation_cause.h"
#include "net/dcsctp/packet/error_cause/unrecognized_chunk_type_cause.h"
#include "net/dcsctp/packet/error_cause/user_initiated_abort_cause.h"
#include "net/dcsctp/packet/parameter/forward_tsn_supported_parameter.h"
#include "net/dcsctp/packet/parameter/parameter.h"
#include "net/dcsctp/packet/parameter/state_cookie_parameter.h"
#include "net/dcsctp/packet/parameter/supported_extensions_parameter.h"
#include "net/dcsctp/packet/parameter/zero_checksum_acceptable_chunk_parameter.h"
#include "net/dcsctp/packet/sctp_packet.h"
#include "net/dcsctp/packet/tlv_trait.h"
#include "net/dcsctp/public/dcsctp_message.h"
#include "net/dcsctp/public/dcsctp_options.h"
#include "net/dcsctp/public/dcsctp_socket.h"
#include "net/dcsctp/public/packet_observer.h"
#include "net/dcsctp/public/types.h"
#include "net/dcsctp/rx/data_tracker.h"
#include "net/dcsctp/rx/reassembly_queue.h"
#include "net/dcsctp/socket/callback_deferrer.h"
#include "net/dcsctp/socket/capabilities.h"
#include "net/dcsctp/socket/heartbeat_handler.h"
#include "net/dcsctp/socket/state_cookie.h"
#include "net/dcsctp/socket/stream_reset_handler.h"
#include "net/dcsctp/socket/transmission_control_block.h"
#include "net/dcsctp/timer/timer.h"
#include "net/dcsctp/tx/retransmission_queue.h"
#include "net/dcsctp/tx/send_queue.h"
#include "rtc_base/checks.h"
#include "rtc_base/logging.h"
#include "rtc_base/strings/string_builder.h"
#include "rtc_base/strings/string_format.h"
namespace dcsctp {
namespace {
constexpr uint32_t kMinVerificationTag = 1;
constexpr uint32_t kMaxVerificationTag = std::numeric_limits<uint32_t>::max();
constexpr uint32_t kMinInitialTsn = 0;
constexpr uint32_t kMaxInitialTsn = std::numeric_limits<uint32_t>::max();
Capabilities ComputeCapabilities(const DcSctpOptions& options,
uint16_t peer_nbr_outbound_streams,
uint16_t peer_nbr_inbound_streams,
const Parameters& parameters) {
Capabilities capabilities;
absl::optional<SupportedExtensionsParameter> supported_extensions =
parameters.get<SupportedExtensionsParameter>();
if (options.enable_partial_reliability) {
capabilities.partial_reliability =
parameters.get<ForwardTsnSupportedParameter>().has_value();
if (supported_extensions.has_value()) {
capabilities.partial_reliability |=
supported_extensions->supports(ForwardTsnChunk::kType);
}
}
if (options.enable_message_interleaving && supported_extensions.has_value()) {
capabilities.message_interleaving =
supported_extensions->supports(IDataChunk::kType) &&
supported_extensions->supports(IForwardTsnChunk::kType);
}
if (supported_extensions.has_value() &&
supported_extensions->supports(ReConfigChunk::kType)) {
capabilities.reconfig = true;
}
if (options.enable_zero_checksum &&
parameters.get<ZeroChecksumAcceptableChunkParameter>().has_value()) {
capabilities.zero_checksum = true;
}
capabilities.negotiated_maximum_incoming_streams = std::min(
options.announced_maximum_incoming_streams, peer_nbr_outbound_streams);
capabilities.negotiated_maximum_outgoing_streams = std::min(
options.announced_maximum_outgoing_streams, peer_nbr_inbound_streams);
return capabilities;
}
void AddCapabilityParameters(const DcSctpOptions& options,
Parameters::Builder& builder) {
std::vector<uint8_t> chunk_types = {ReConfigChunk::kType};
if (options.enable_partial_reliability) {
builder.Add(ForwardTsnSupportedParameter());
chunk_types.push_back(ForwardTsnChunk::kType);
}
if (options.enable_message_interleaving) {
chunk_types.push_back(IDataChunk::kType);
chunk_types.push_back(IForwardTsnChunk::kType);
}
if (options.enable_zero_checksum) {
builder.Add(ZeroChecksumAcceptableChunkParameter());
}
builder.Add(SupportedExtensionsParameter(std::move(chunk_types)));
}
TieTag MakeTieTag(DcSctpSocketCallbacks& cb) {
uint32_t tie_tag_upper =
cb.GetRandomInt(0, std::numeric_limits<uint32_t>::max());
uint32_t tie_tag_lower =
cb.GetRandomInt(1, std::numeric_limits<uint32_t>::max());
return TieTag(static_cast<uint64_t>(tie_tag_upper) << 32 |
static_cast<uint64_t>(tie_tag_lower));
}
SctpImplementation DeterminePeerImplementation(
rtc::ArrayView<const uint8_t> cookie) {
if (cookie.size() > 8) {
absl::string_view magic(reinterpret_cast<const char*>(cookie.data()), 8);
if (magic == "dcSCTP00") {
return SctpImplementation::kDcsctp;
}
if (magic == "KAME-BSD") {
return SctpImplementation::kUsrSctp;
}
}
return SctpImplementation::kOther;
}
}
DcSctpSocket::DcSctpSocket(absl::string_view log_prefix,
DcSctpSocketCallbacks& callbacks,
std::unique_ptr<PacketObserver> packet_observer,
const DcSctpOptions& options)
: log_prefix_(std::string(log_prefix) + ": "),
packet_observer_(std::move(packet_observer)),
options_(options),
callbacks_(callbacks),
timer_manager_([this](webrtc::TaskQueueBase::DelayPrecision precision) {
return callbacks_.CreateTimeout(precision);
}),
t1_init_(timer_manager_.CreateTimer(
"t1-init",
absl::bind_front(&DcSctpSocket::OnInitTimerExpiry, this),
TimerOptions(options.t1_init_timeout,
TimerBackoffAlgorithm::kExponential,
options.max_init_retransmits))),
t1_cookie_(timer_manager_.CreateTimer(
"t1-cookie",
absl::bind_front(&DcSctpSocket::OnCookieTimerExpiry, this),
TimerOptions(options.t1_cookie_timeout,
TimerBackoffAlgorithm::kExponential,
options.max_init_retransmits))),
t2_shutdown_(timer_manager_.CreateTimer(
"t2-shutdown",
absl::bind_front(&DcSctpSocket::OnShutdownTimerExpiry, this),
TimerOptions(options.t2_shutdown_timeout,
TimerBackoffAlgorithm::kExponential,
options.max_retransmissions))),
packet_sender_(callbacks_,
absl::bind_front(&DcSctpSocket::OnSentPacket, this)),
send_queue_(log_prefix_,
&callbacks_,
options_.max_send_buffer_size,
options_.mtu,
options_.default_stream_priority,
options_.total_buffered_amount_low_threshold) {}
std::string DcSctpSocket::log_prefix() const {
return log_prefix_ + "[" + std::string(ToString(state_)) + "] ";
}
bool DcSctpSocket::IsConsistent() const {
if (tcb_ != nullptr && tcb_->reassembly_queue().HasMessages()) {
return false;
}
switch (state_) {
case State::kClosed:
return (tcb_ == nullptr && !t1_init_->is_running() &&
!t1_cookie_->is_running() && !t2_shutdown_->is_running());
case State::kCookieWait:
return (tcb_ == nullptr && t1_init_->is_running() &&
!t1_cookie_->is_running() && !t2_shutdown_->is_running());
case State::kCookieEchoed:
return (tcb_ != nullptr && !t1_init_->is_running() &&
t1_cookie_->is_running() && !t2_shutdown_->is_running() &&
tcb_->has_cookie_echo_chunk());
case State::kEstablished:
return (tcb_ != nullptr && !t1_init_->is_running() &&
!t1_cookie_->is_running() && !t2_shutdown_->is_running());
case State::kShutdownPending:
return (tcb_ != nullptr && !t1_init_->is_running() &&
!t1_cookie_->is_running() && !t2_shutdown_->is_running());
case State::kShutdownSent:
return (tcb_ != nullptr && !t1_init_->is_running() &&
!t1_cookie_->is_running() && t2_shutdown_->is_running());
case State::kShutdownReceived:
return (tcb_ != nullptr && !t1_init_->is_running() &&
!t1_cookie_->is_running() && !t2_shutdown_->is_running());
case State::kShutdownAckSent:
return (tcb_ != nullptr && !t1_init_->is_running() &&
!t1_cookie_->is_running() && t2_shutdown_->is_running());
}
}
constexpr absl::string_view DcSctpSocket::ToString(DcSctpSocket::State state) {
switch (state) {
case DcSctpSocket::State::kClosed:
return "CLOSED";
case DcSctpSocket::State::kCookieWait:
return "COOKIE_WAIT";
case DcSctpSocket::State::kCookieEchoed:
return "COOKIE_ECHOED";
case DcSctpSocket::State::kEstablished:
return "ESTABLISHED";
case DcSctpSocket::State::kShutdownPending:
return "SHUTDOWN_PENDING";
case DcSctpSocket::State::kShutdownSent:
return "SHUTDOWN_SENT";
case DcSctpSocket::State::kShutdownReceived:
return "SHUTDOWN_RECEIVED";
case DcSctpSocket::State::kShutdownAckSent:
return "SHUTDOWN_ACK_SENT";
}
}
void DcSctpSocket::SetState(State state, absl::string_view reason) {
if (state_ != state) {
RTC_DLOG(LS_VERBOSE) << log_prefix_ << "Socket state changed from "
<< ToString(state_) << " to " << ToString(state)
<< " due to " << reason;
state_ = state;
}
}
void DcSctpSocket::SendInit() {
Parameters::Builder params_builder;
AddCapabilityParameters(options_, params_builder);
InitChunk init(connect_params_.verification_tag,
options_.max_receiver_window_buffer_size,
options_.announced_maximum_outgoing_streams,
options_.announced_maximum_incoming_streams,
connect_params_.initial_tsn, params_builder.Build());
SctpPacket::Builder b(VerificationTag(0), options_);
b.Add(init);
packet_sender_.Send(b, true);
}
void DcSctpSocket::MakeConnectionParameters() {
VerificationTag new_verification_tag(
callbacks_.GetRandomInt(kMinVerificationTag, kMaxVerificationTag));
TSN initial_tsn(callbacks_.GetRandomInt(kMinInitialTsn, kMaxInitialTsn));
connect_params_.initial_tsn = initial_tsn;
connect_params_.verification_tag = new_verification_tag;
}
void DcSctpSocket::Connect() {
RTC_DCHECK_RUN_ON(&thread_checker_);
CallbackDeferrer::ScopedDeferrer deferrer(callbacks_);
if (state_ == State::kClosed) {
MakeConnectionParameters();
RTC_DLOG(LS_INFO)
<< log_prefix()
<< rtc::StringFormat(
"Connecting. my_verification_tag=%08x, my_initial_tsn=%u",
*connect_params_.verification_tag, *connect_params_.initial_tsn);
SendInit();
t1_init_->Start();
SetState(State::kCookieWait, "Connect called");
} else {
RTC_DLOG(LS_WARNING) << log_prefix()
<< "Called Connect on a socket that is not closed";
}
RTC_DCHECK(IsConsistent());
}
void DcSctpSocket::CreateTransmissionControlBlock(
const Capabilities& capabilities,
VerificationTag my_verification_tag,
TSN my_initial_tsn,
VerificationTag peer_verification_tag,
TSN peer_initial_tsn,
size_t a_rwnd,
TieTag tie_tag) {
metrics_.uses_message_interleaving = capabilities.message_interleaving;
metrics_.uses_zero_checksum = capabilities.zero_checksum;
metrics_.negotiated_maximum_incoming_streams =
capabilities.negotiated_maximum_incoming_streams;
metrics_.negotiated_maximum_outgoing_streams =
capabilities.negotiated_maximum_outgoing_streams;
tcb_ = std::make_unique<TransmissionControlBlock>(
timer_manager_, log_prefix_, options_, capabilities, callbacks_,
send_queue_, my_verification_tag, my_initial_tsn, peer_verification_tag,
peer_initial_tsn, a_rwnd, tie_tag, packet_sender_,
[this]() { return state_ == State::kEstablished; });
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Created TCB: " << tcb_->ToString();
}
void DcSctpSocket::RestoreFromState(const DcSctpSocketHandoverState& state) {
RTC_DCHECK_RUN_ON(&thread_checker_);
CallbackDeferrer::ScopedDeferrer deferrer(callbacks_);
if (state_ != State::kClosed) {
callbacks_.OnError(ErrorKind::kUnsupportedOperation,
"Only closed socket can be restored from state");
} else {
if (state.socket_state ==
DcSctpSocketHandoverState::SocketState::kConnected) {
VerificationTag my_verification_tag =
VerificationTag(state.my_verification_tag);
connect_params_.verification_tag = my_verification_tag;
Capabilities capabilities;
capabilities.partial_reliability = state.capabilities.partial_reliability;
capabilities.message_interleaving =
state.capabilities.message_interleaving;
capabilities.reconfig = state.capabilities.reconfig;
capabilities.zero_checksum = state.capabilities.zero_checksum;
capabilities.negotiated_maximum_incoming_streams =
state.capabilities.negotiated_maximum_incoming_streams;
capabilities.negotiated_maximum_outgoing_streams =
state.capabilities.negotiated_maximum_outgoing_streams;
send_queue_.RestoreFromState(state);
CreateTransmissionControlBlock(
capabilities, my_verification_tag, TSN(state.my_initial_tsn),
VerificationTag(state.peer_verification_tag),
TSN(state.peer_initial_tsn), static_cast<size_t>(0),
TieTag(state.tie_tag));
tcb_->RestoreFromState(state);
SetState(State::kEstablished, "restored from handover state");
callbacks_.OnConnected();
}
}
RTC_DCHECK(IsConsistent());
}
void DcSctpSocket::Shutdown() {
RTC_DCHECK_RUN_ON(&thread_checker_);
CallbackDeferrer::ScopedDeferrer deferrer(callbacks_);
if (tcb_ != nullptr) {
if (state_ != State::kShutdownSent && state_ != State::kShutdownAckSent) {
SetState(State::kShutdownPending, "Shutdown called");
t1_init_->Stop();
t1_cookie_->Stop();
MaybeSendShutdownOrAck();
}
} else {
InternalClose(ErrorKind::kNoError, "");
}
RTC_DCHECK(IsConsistent());
}
void DcSctpSocket::Close() {
RTC_DCHECK_RUN_ON(&thread_checker_);
CallbackDeferrer::ScopedDeferrer deferrer(callbacks_);
if (state_ != State::kClosed) {
if (tcb_ != nullptr) {
SctpPacket::Builder b = tcb_->PacketBuilder();
b.Add(AbortChunk(true,
Parameters::Builder()
.Add(UserInitiatedAbortCause("Close called"))
.Build()));
packet_sender_.Send(b);
}
InternalClose(ErrorKind::kNoError, "");
} else {
RTC_DLOG(LS_INFO) << log_prefix() << "Called Close on a closed socket";
}
RTC_DCHECK(IsConsistent());
}
void DcSctpSocket::CloseConnectionBecauseOfTooManyTransmissionErrors() {
packet_sender_.Send(tcb_->PacketBuilder().Add(AbortChunk(
true, Parameters::Builder()
.Add(UserInitiatedAbortCause("Too many retransmissions"))
.Build())));
InternalClose(ErrorKind::kTooManyRetries, "Too many retransmissions");
}
void DcSctpSocket::InternalClose(ErrorKind error, absl::string_view message) {
if (state_ != State::kClosed) {
t1_init_->Stop();
t1_cookie_->Stop();
t2_shutdown_->Stop();
tcb_ = nullptr;
if (error == ErrorKind::kNoError) {
callbacks_.OnClosed();
} else {
callbacks_.OnAborted(error, message);
}
SetState(State::kClosed, message);
}
RTC_DCHECK(IsConsistent());
}
void DcSctpSocket::SetStreamPriority(StreamID stream_id,
StreamPriority priority) {
RTC_DCHECK_RUN_ON(&thread_checker_);
send_queue_.SetStreamPriority(stream_id, priority);
}
StreamPriority DcSctpSocket::GetStreamPriority(StreamID stream_id) const {
RTC_DCHECK_RUN_ON(&thread_checker_);
return send_queue_.GetStreamPriority(stream_id);
}
SendStatus DcSctpSocket::Send(DcSctpMessage message,
const SendOptions& send_options) {
RTC_DCHECK_RUN_ON(&thread_checker_);
CallbackDeferrer::ScopedDeferrer deferrer(callbacks_);
LifecycleId lifecycle_id = send_options.lifecycle_id;
if (message.payload().empty()) {
if (lifecycle_id.IsSet()) {
callbacks_.OnLifecycleEnd(lifecycle_id);
}
callbacks_.OnError(ErrorKind::kProtocolViolation,
"Unable to send empty message");
return SendStatus::kErrorMessageEmpty;
}
if (message.payload().size() > options_.max_message_size) {
if (lifecycle_id.IsSet()) {
callbacks_.OnLifecycleEnd(lifecycle_id);
}
callbacks_.OnError(ErrorKind::kProtocolViolation,
"Unable to send too large message");
return SendStatus::kErrorMessageTooLarge;
}
if (state_ == State::kShutdownPending || state_ == State::kShutdownSent ||
state_ == State::kShutdownReceived || state_ == State::kShutdownAckSent) {
if (lifecycle_id.IsSet()) {
callbacks_.OnLifecycleEnd(lifecycle_id);
}
callbacks_.OnError(ErrorKind::kWrongSequence,
"Unable to send message as the socket is shutting down");
return SendStatus::kErrorShuttingDown;
}
if (send_queue_.IsFull()) {
if (lifecycle_id.IsSet()) {
callbacks_.OnLifecycleEnd(lifecycle_id);
}
callbacks_.OnError(ErrorKind::kResourceExhaustion,
"Unable to send message as the send queue is full");
return SendStatus::kErrorResourceExhaustion;
}
TimeMs now = callbacks_.TimeMillis();
++metrics_.tx_messages_count;
send_queue_.Add(now, std::move(message), send_options);
if (tcb_ != nullptr) {
tcb_->SendBufferedPackets(now);
}
RTC_DCHECK(IsConsistent());
return SendStatus::kSuccess;
}
ResetStreamsStatus DcSctpSocket::ResetStreams(
rtc::ArrayView<const StreamID> outgoing_streams) {
RTC_DCHECK_RUN_ON(&thread_checker_);
CallbackDeferrer::ScopedDeferrer deferrer(callbacks_);
if (tcb_ == nullptr) {
callbacks_.OnError(ErrorKind::kWrongSequence,
"Can't reset streams as the socket is not connected");
return ResetStreamsStatus::kNotConnected;
}
if (!tcb_->capabilities().reconfig) {
callbacks_.OnError(ErrorKind::kUnsupportedOperation,
"Can't reset streams as the peer doesn't support it");
return ResetStreamsStatus::kNotSupported;
}
tcb_->stream_reset_handler().ResetStreams(outgoing_streams);
MaybeSendResetStreamsRequest();
RTC_DCHECK(IsConsistent());
return ResetStreamsStatus::kPerformed;
}
SocketState DcSctpSocket::state() const {
RTC_DCHECK_RUN_ON(&thread_checker_);
switch (state_) {
case State::kClosed:
return SocketState::kClosed;
case State::kCookieWait:
case State::kCookieEchoed:
return SocketState::kConnecting;
case State::kEstablished:
return SocketState::kConnected;
case State::kShutdownPending:
case State::kShutdownSent:
case State::kShutdownReceived:
case State::kShutdownAckSent:
return SocketState::kShuttingDown;
}
}
void DcSctpSocket::SetMaxMessageSize(size_t max_message_size) {
RTC_DCHECK_RUN_ON(&thread_checker_);
options_.max_message_size = max_message_size;
}
size_t DcSctpSocket::buffered_amount(StreamID stream_id) const {
RTC_DCHECK_RUN_ON(&thread_checker_);
return send_queue_.buffered_amount(stream_id);
}
size_t DcSctpSocket::buffered_amount_low_threshold(StreamID stream_id) const {
RTC_DCHECK_RUN_ON(&thread_checker_);
return send_queue_.buffered_amount_low_threshold(stream_id);
}
void DcSctpSocket::SetBufferedAmountLowThreshold(StreamID stream_id,
size_t bytes) {
RTC_DCHECK_RUN_ON(&thread_checker_);
send_queue_.SetBufferedAmountLowThreshold(stream_id, bytes);
}
absl::optional<Metrics> DcSctpSocket::GetMetrics() const {
RTC_DCHECK_RUN_ON(&thread_checker_);
if (tcb_ == nullptr) {
return absl::nullopt;
}
Metrics metrics = metrics_;
metrics.cwnd_bytes = tcb_->cwnd();
metrics.srtt_ms = tcb_->current_srtt().value();
size_t packet_payload_size =
options_.mtu - SctpPacket::kHeaderSize - DataChunk::kHeaderSize;
metrics.unack_data_count =
tcb_->retransmission_queue().outstanding_items() +
(send_queue_.total_buffered_amount() + packet_payload_size - 1) /
packet_payload_size;
metrics.peer_rwnd_bytes = tcb_->retransmission_queue().rwnd();
metrics.negotiated_maximum_incoming_streams =
tcb_->capabilities().negotiated_maximum_incoming_streams;
metrics.negotiated_maximum_incoming_streams =
tcb_->capabilities().negotiated_maximum_incoming_streams;
metrics.rtx_packets_count = tcb_->retransmission_queue().rtx_packets_count();
metrics.rtx_bytes_count = tcb_->retransmission_queue().rtx_bytes_count();
return metrics;
}
void DcSctpSocket::MaybeSendShutdownOnPacketReceived(const SctpPacket& packet) {
if (state_ == State::kShutdownSent) {
bool has_data_chunk =
std::find_if(packet.descriptors().begin(), packet.descriptors().end(),
[](const SctpPacket::ChunkDescriptor& descriptor) {
return descriptor.type == DataChunk::kType;
}) != packet.descriptors().end();
if (has_data_chunk) {
SendShutdown();
t2_shutdown_->set_duration(tcb_->current_rto());
t2_shutdown_->Start();
}
}
}
void DcSctpSocket::MaybeSendResetStreamsRequest() {
absl::optional<ReConfigChunk> reconfig =
tcb_->stream_reset_handler().MakeStreamResetRequest();
if (reconfig.has_value()) {
SctpPacket::Builder builder = tcb_->PacketBuilder();
builder.Add(*reconfig);
packet_sender_.Send(builder);
}
}
bool DcSctpSocket::ValidatePacket(const SctpPacket& packet) {
const CommonHeader& header = packet.common_header();
VerificationTag my_verification_tag =
tcb_ != nullptr ? tcb_->my_verification_tag() : VerificationTag(0);
if (header.verification_tag == VerificationTag(0)) {
if (packet.descriptors().size() == 1 &&
packet.descriptors()[0].type == InitChunk::kType) {
return true;
}
callbacks_.OnError(
ErrorKind::kParseFailed,
"Only a single INIT chunk can be present in packets sent on "
"verification_tag = 0");
return false;
}
if (packet.descriptors().size() == 1 &&
packet.descriptors()[0].type == AbortChunk::kType) {
bool t_bit = (packet.descriptors()[0].flags & 0x01) != 0;
if (t_bit && tcb_ == nullptr) {
return true;
}
if ((!t_bit && header.verification_tag == my_verification_tag) ||
(t_bit && header.verification_tag == tcb_->peer_verification_tag())) {
return true;
}
callbacks_.OnError(ErrorKind::kParseFailed,
"ABORT chunk verification tag was wrong");
return false;
}
if (packet.descriptors()[0].type == InitAckChunk::kType) {
if (header.verification_tag == connect_params_.verification_tag) {
return true;
}
callbacks_.OnError(
ErrorKind::kParseFailed,
rtc::StringFormat(
"Packet has invalid verification tag: %08x, expected %08x",
*header.verification_tag, *connect_params_.verification_tag));
return false;
}
if (packet.descriptors()[0].type == CookieEchoChunk::kType) {
return true;
}
if (packet.descriptors().size() == 1 &&
packet.descriptors()[0].type == ShutdownCompleteChunk::kType) {
bool t_bit = (packet.descriptors()[0].flags & 0x01) != 0;
if (t_bit && tcb_ == nullptr) {
return true;
}
if ((!t_bit && header.verification_tag == my_verification_tag) ||
(t_bit && header.verification_tag == tcb_->peer_verification_tag())) {
return true;
}
callbacks_.OnError(ErrorKind::kParseFailed,
"SHUTDOWN_COMPLETE chunk verification tag was wrong");
return false;
}
if (header.verification_tag == my_verification_tag) {
return true;
}
callbacks_.OnError(
ErrorKind::kParseFailed,
rtc::StringFormat(
"Packet has invalid verification tag: %08x, expected %08x",
*header.verification_tag, *my_verification_tag));
return false;
}
void DcSctpSocket::HandleTimeout(TimeoutID timeout_id) {
RTC_DCHECK_RUN_ON(&thread_checker_);
CallbackDeferrer::ScopedDeferrer deferrer(callbacks_);
timer_manager_.HandleTimeout(timeout_id);
if (tcb_ != nullptr && tcb_->HasTooManyTxErrors()) {
CloseConnectionBecauseOfTooManyTransmissionErrors();
}
RTC_DCHECK(IsConsistent());
}
void DcSctpSocket::ReceivePacket(rtc::ArrayView<const uint8_t> data) {
RTC_DCHECK_RUN_ON(&thread_checker_);
CallbackDeferrer::ScopedDeferrer deferrer(callbacks_);
++metrics_.rx_packets_count;
if (packet_observer_ != nullptr) {
packet_observer_->OnReceivedPacket(callbacks_.TimeMillis(), data);
}
absl::optional<SctpPacket> packet = SctpPacket::Parse(data, options_);
if (!packet.has_value()) {
callbacks_.OnError(ErrorKind::kParseFailed,
"Failed to parse received SCTP packet");
RTC_DCHECK(IsConsistent());
return;
}
if (RTC_DLOG_IS_ON) {
for (const auto& descriptor : packet->descriptors()) {
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Received "
<< DebugConvertChunkToString(descriptor.data);
}
}
if (!ValidatePacket(*packet)) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Packet failed verification tag check - dropping";
RTC_DCHECK(IsConsistent());
return;
}
MaybeSendShutdownOnPacketReceived(*packet);
for (const auto& descriptor : packet->descriptors()) {
if (!Dispatch(packet->common_header(), descriptor)) {
break;
}
}
if (tcb_ != nullptr) {
tcb_->data_tracker().ObservePacketEnd();
tcb_->MaybeSendSack();
}
RTC_DCHECK(IsConsistent());
}
void DcSctpSocket::DebugPrintOutgoing(rtc::ArrayView<const uint8_t> payload) {
auto packet = SctpPacket::Parse(payload, options_);
RTC_DCHECK(packet.has_value());
for (const auto& desc : packet->descriptors()) {
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Sent "
<< DebugConvertChunkToString(desc.data);
}
}
bool DcSctpSocket::Dispatch(const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
switch (descriptor.type) {
case DataChunk::kType:
HandleData(header, descriptor);
break;
case InitChunk::kType:
HandleInit(header, descriptor);
break;
case InitAckChunk::kType:
HandleInitAck(header, descriptor);
break;
case SackChunk::kType:
HandleSack(header, descriptor);
break;
case HeartbeatRequestChunk::kType:
HandleHeartbeatRequest(header, descriptor);
break;
case HeartbeatAckChunk::kType:
HandleHeartbeatAck(header, descriptor);
break;
case AbortChunk::kType:
HandleAbort(header, descriptor);
break;
case ErrorChunk::kType:
HandleError(header, descriptor);
break;
case CookieEchoChunk::kType:
HandleCookieEcho(header, descriptor);
break;
case CookieAckChunk::kType:
HandleCookieAck(header, descriptor);
break;
case ShutdownChunk::kType:
HandleShutdown(header, descriptor);
break;
case ShutdownAckChunk::kType:
HandleShutdownAck(header, descriptor);
break;
case ShutdownCompleteChunk::kType:
HandleShutdownComplete(header, descriptor);
break;
case ReConfigChunk::kType:
HandleReconfig(header, descriptor);
break;
case ForwardTsnChunk::kType:
HandleForwardTsn(header, descriptor);
break;
case IDataChunk::kType:
HandleIData(header, descriptor);
break;
case IForwardTsnChunk::kType:
HandleIForwardTsn(header, descriptor);
break;
default:
return HandleUnrecognizedChunk(descriptor);
}
return true;
}
bool DcSctpSocket::HandleUnrecognizedChunk(
const SctpPacket::ChunkDescriptor& descriptor) {
bool report_as_error = (descriptor.type & 0x40) != 0;
bool continue_processing = (descriptor.type & 0x80) != 0;
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Received unknown chunk: "
<< static_cast<int>(descriptor.type);
if (report_as_error) {
rtc::StringBuilder sb;
sb << "Received unknown chunk of type: "
<< static_cast<int>(descriptor.type) << " with report-error bit set";
callbacks_.OnError(ErrorKind::kParseFailed, sb.str());
RTC_DLOG(LS_VERBOSE)
<< log_prefix()
<< "Unknown chunk, with type indicating it should be reported.";
if (tcb_ != nullptr) {
packet_sender_.Send(tcb_->PacketBuilder().Add(
ErrorChunk(Parameters::Builder()
.Add(UnrecognizedChunkTypeCause(std::vector<uint8_t>(
descriptor.data.begin(), descriptor.data.end())))
.Build())));
}
}
if (!continue_processing) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Unknown chunk, with type indicating not to "
"process any further chunks";
}
return continue_processing;
}
absl::optional<DurationMs> DcSctpSocket::OnInitTimerExpiry() {
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Timer " << t1_init_->name()
<< " has expired: " << t1_init_->expiration_count()
<< "/" << t1_init_->options().max_restarts.value_or(-1);
RTC_DCHECK(state_ == State::kCookieWait);
if (t1_init_->is_running()) {
SendInit();
} else {
InternalClose(ErrorKind::kTooManyRetries, "No INIT_ACK received");
}
RTC_DCHECK(IsConsistent());
return absl::nullopt;
}
absl::optional<DurationMs> DcSctpSocket::OnCookieTimerExpiry() {
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Timer " << t1_cookie_->name()
<< " has expired: " << t1_cookie_->expiration_count()
<< "/"
<< t1_cookie_->options().max_restarts.value_or(-1);
RTC_DCHECK(state_ == State::kCookieEchoed);
if (t1_cookie_->is_running()) {
tcb_->SendBufferedPackets(callbacks_.TimeMillis());
} else {
InternalClose(ErrorKind::kTooManyRetries, "No COOKIE_ACK received");
}
RTC_DCHECK(IsConsistent());
return absl::nullopt;
}
absl::optional<DurationMs> DcSctpSocket::OnShutdownTimerExpiry() {
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Timer " << t2_shutdown_->name()
<< " has expired: " << t2_shutdown_->expiration_count()
<< "/"
<< t2_shutdown_->options().max_restarts.value_or(-1);
if (!t2_shutdown_->is_running()) {
packet_sender_.Send(tcb_->PacketBuilder().Add(
AbortChunk(true, Parameters::Builder()
.Add(UserInitiatedAbortCause(
"Too many retransmissions of SHUTDOWN"))
.Build())));
InternalClose(ErrorKind::kTooManyRetries, "No SHUTDOWN_ACK received");
RTC_DCHECK(IsConsistent());
return absl::nullopt;
}
SendShutdown();
RTC_DCHECK(IsConsistent());
return tcb_->current_rto();
}
void DcSctpSocket::OnSentPacket(rtc::ArrayView<const uint8_t> packet,
SendPacketStatus status) {
if (packet_observer_ != nullptr) {
packet_observer_->OnSentPacket(callbacks_.TimeMillis(), packet);
}
if (status == SendPacketStatus::kSuccess) {
if (RTC_DLOG_IS_ON) {
DebugPrintOutgoing(packet);
}
if (tcb_ != nullptr) {
tcb_->heartbeat_handler().RestartTimer();
}
++metrics_.tx_packets_count;
}
}
bool DcSctpSocket::ValidateHasTCB() {
if (tcb_ != nullptr) {
return true;
}
callbacks_.OnError(
ErrorKind::kNotConnected,
"Received unexpected commands on socket that is not connected");
return false;
}
void DcSctpSocket::ReportFailedToParseChunk(int chunk_type) {
rtc::StringBuilder sb;
sb << "Failed to parse chunk of type: " << chunk_type;
callbacks_.OnError(ErrorKind::kParseFailed, sb.str());
}
void DcSctpSocket::HandleData(const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<DataChunk> chunk = DataChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk) && ValidateHasTCB()) {
HandleDataCommon(*chunk);
}
}
void DcSctpSocket::HandleIData(const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<IDataChunk> chunk = IDataChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk) && ValidateHasTCB()) {
HandleDataCommon(*chunk);
}
}
void DcSctpSocket::HandleDataCommon(AnyDataChunk& chunk) {
TSN tsn = chunk.tsn();
AnyDataChunk::ImmediateAckFlag immediate_ack = chunk.options().immediate_ack;
Data data = std::move(chunk).extract();
if (data.payload.empty()) {
packet_sender_.Send(tcb_->PacketBuilder().Add(
ErrorChunk(Parameters::Builder().Add(NoUserDataCause(tsn)).Build())));
callbacks_.OnError(ErrorKind::kProtocolViolation,
"Received DATA chunk with no user data");
return;
}
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Handle DATA, queue_size="
<< tcb_->reassembly_queue().queued_bytes()
<< ", water_mark="
<< tcb_->reassembly_queue().watermark_bytes()
<< ", full=" << tcb_->reassembly_queue().is_full()
<< ", above="
<< tcb_->reassembly_queue().is_above_watermark();
if (tcb_->reassembly_queue().is_full()) {
packet_sender_.Send(tcb_->PacketBuilder().Add(AbortChunk(
true, Parameters::Builder().Add(OutOfResourceErrorCause()).Build())));
InternalClose(ErrorKind::kResourceExhaustion,
"Reassembly Queue is exhausted");
return;
}
if (tcb_->reassembly_queue().is_above_watermark()) {
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Is above high watermark";
if (!tcb_->data_tracker().will_increase_cum_ack_tsn(tsn)) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Rejected data because of exceeding watermark";
tcb_->data_tracker().ForceImmediateSack();
return;
}
}
if (!tcb_->data_tracker().IsTSNValid(tsn)) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Rejected data because of failing TSN validity";
return;
}
if (tcb_->data_tracker().Observe(tsn, immediate_ack)) {
tcb_->reassembly_queue().Add(tsn, std::move(data));
MaybeDeliverMessages();
}
}
void DcSctpSocket::HandleInit(const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<InitChunk> chunk = InitChunk::Parse(descriptor.data);
if (!ValidateParseSuccess(chunk)) {
return;
}
if (chunk->initiate_tag() == VerificationTag(0) ||
chunk->nbr_outbound_streams() == 0 || chunk->nbr_inbound_streams() == 0) {
packet_sender_.Send(
SctpPacket::Builder(VerificationTag(0), options_)
.Add(AbortChunk(
false,
Parameters::Builder()
.Add(ProtocolViolationCause("INIT malformed"))
.Build())));
InternalClose(ErrorKind::kProtocolViolation, "Received invalid INIT");
return;
}
if (state_ == State::kShutdownAckSent) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Received Init indicating lost ShutdownComplete";
SendShutdownAck();
return;
}
TieTag tie_tag(0);
if (state_ == State::kClosed) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Received Init in closed state (normal)";
MakeConnectionParameters();
} else if (state_ == State::kCookieWait || state_ == State::kCookieEchoed) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Received Init indicating simultaneous connections";
} else {
RTC_DCHECK(tcb_ != nullptr);
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Received Init indicating restarted connection";
for (int tries = 0; tries < 10; ++tries) {
connect_params_.verification_tag = VerificationTag(
callbacks_.GetRandomInt(kMinVerificationTag, kMaxVerificationTag));
if (connect_params_.verification_tag != tcb_->my_verification_tag()) {
break;
}
}
connect_params_.initial_tsn =
TSN(*tcb_->retransmission_queue().next_tsn() + 1000000);
tie_tag = tcb_->tie_tag();
}
RTC_DLOG(LS_VERBOSE)
<< log_prefix()
<< rtc::StringFormat(
"Proceeding with connection. my_verification_tag=%08x, "
"my_initial_tsn=%u, peer_verification_tag=%08x, "
"peer_initial_tsn=%u",
*connect_params_.verification_tag, *connect_params_.initial_tsn,
*chunk->initiate_tag(), *chunk->initial_tsn());
Capabilities capabilities =
ComputeCapabilities(options_, chunk->nbr_outbound_streams(),
chunk->nbr_inbound_streams(), chunk->parameters());
SctpPacket::Builder b(chunk->initiate_tag(), options_);
Parameters::Builder params_builder =
Parameters::Builder().Add(StateCookieParameter(
StateCookie(chunk->initiate_tag(), chunk->initial_tsn(),
chunk->a_rwnd(), tie_tag, capabilities)
.Serialize()));
AddCapabilityParameters(options_, params_builder);
InitAckChunk init_ack(connect_params_.verification_tag,
options_.max_receiver_window_buffer_size,
options_.announced_maximum_outgoing_streams,
options_.announced_maximum_incoming_streams,
connect_params_.initial_tsn, params_builder.Build());
b.Add(init_ack);
packet_sender_.Send(b, !capabilities.zero_checksum);
}
void DcSctpSocket::HandleInitAck(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<InitAckChunk> chunk = InitAckChunk::Parse(descriptor.data);
if (!ValidateParseSuccess(chunk)) {
return;
}
if (state_ != State::kCookieWait) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Received INIT_ACK in unexpected state";
return;
}
auto cookie = chunk->parameters().get<StateCookieParameter>();
if (!cookie.has_value()) {
packet_sender_.Send(
SctpPacket::Builder(connect_params_.verification_tag, options_)
.Add(AbortChunk(
false,
Parameters::Builder()
.Add(ProtocolViolationCause("INIT-ACK malformed"))
.Build())));
InternalClose(ErrorKind::kProtocolViolation,
"InitAck chunk doesn't contain a cookie");
return;
}
Capabilities capabilities =
ComputeCapabilities(options_, chunk->nbr_outbound_streams(),
chunk->nbr_inbound_streams(), chunk->parameters());
t1_init_->Stop();
metrics_.peer_implementation = DeterminePeerImplementation(cookie->data());
send_queue_.Reset();
CreateTransmissionControlBlock(capabilities, connect_params_.verification_tag,
connect_params_.initial_tsn,
chunk->initiate_tag(), chunk->initial_tsn(),
chunk->a_rwnd(), MakeTieTag(callbacks_));
SetState(State::kCookieEchoed, "INIT_ACK received");
tcb_->SetCookieEchoChunk(CookieEchoChunk(cookie->data()));
tcb_->SendBufferedPackets(callbacks_.TimeMillis());
t1_cookie_->Start();
}
void DcSctpSocket::HandleCookieEcho(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<CookieEchoChunk> chunk =
CookieEchoChunk::Parse(descriptor.data);
if (!ValidateParseSuccess(chunk)) {
return;
}
absl::optional<StateCookie> cookie =
StateCookie::Deserialize(chunk->cookie());
if (!cookie.has_value()) {
callbacks_.OnError(ErrorKind::kParseFailed, "Failed to parse state cookie");
return;
}
if (tcb_ != nullptr) {
if (!HandleCookieEchoWithTCB(header, *cookie)) {
return;
}
} else {
if (header.verification_tag != connect_params_.verification_tag) {
callbacks_.OnError(
ErrorKind::kParseFailed,
rtc::StringFormat(
"Received CookieEcho with invalid verification tag: %08x, "
"expected %08x",
*header.verification_tag, *connect_params_.verification_tag));
return;
}
}
t1_init_->Stop();
t1_cookie_->Stop();
if (state_ != State::kEstablished) {
if (tcb_ != nullptr) {
tcb_->ClearCookieEchoChunk();
}
SetState(State::kEstablished, "COOKIE_ECHO received");
callbacks_.OnConnected();
}
if (tcb_ == nullptr) {
send_queue_.Reset();
CreateTransmissionControlBlock(
cookie->capabilities(), connect_params_.verification_tag,
connect_params_.initial_tsn, cookie->initiate_tag(),
cookie->initial_tsn(), cookie->a_rwnd(), MakeTieTag(callbacks_));
}
SctpPacket::Builder b = tcb_->PacketBuilder();
b.Add(CookieAckChunk());
tcb_->SendBufferedPackets(b, callbacks_.TimeMillis());
}
bool DcSctpSocket::HandleCookieEchoWithTCB(const CommonHeader& header,
const StateCookie& cookie) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Handling CookieEchoChunk with TCB. local_tag="
<< *tcb_->my_verification_tag()
<< ", peer_tag=" << *header.verification_tag
<< ", tcb_tag=" << *tcb_->peer_verification_tag()
<< ", cookie_tag=" << *cookie.initiate_tag()
<< ", local_tie_tag=" << *tcb_->tie_tag()
<< ", peer_tie_tag=" << *cookie.tie_tag();
if (header.verification_tag != tcb_->my_verification_tag() &&
tcb_->peer_verification_tag() != cookie.initiate_tag() &&
cookie.tie_tag() == tcb_->tie_tag()) {
if (state_ == State::kShutdownAckSent) {
SctpPacket::Builder b(cookie.initiate_tag(), options_);
b.Add(ShutdownAckChunk());
b.Add(ErrorChunk(Parameters::Builder()
.Add(CookieReceivedWhileShuttingDownCause())
.Build()));
packet_sender_.Send(b);
callbacks_.OnError(ErrorKind::kWrongSequence,
"Received COOKIE-ECHO while shutting down");
return false;
}
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Received COOKIE-ECHO indicating a restarted peer";
tcb_ = nullptr;
callbacks_.OnConnectionRestarted();
} else if (header.verification_tag == tcb_->my_verification_tag() &&
tcb_->peer_verification_tag() != cookie.initiate_tag()) {
RTC_DLOG(LS_VERBOSE)
<< log_prefix()
<< "Received COOKIE-ECHO indicating simultaneous connections";
tcb_ = nullptr;
} else if (header.verification_tag != tcb_->my_verification_tag() &&
tcb_->peer_verification_tag() == cookie.initiate_tag() &&
cookie.tie_tag() == TieTag(0)) {
RTC_DLOG(LS_VERBOSE)
<< log_prefix()
<< "Received COOKIE-ECHO indicating a late COOKIE-ECHO. Discarding";
return false;
} else if (header.verification_tag == tcb_->my_verification_tag() &&
tcb_->peer_verification_tag() == cookie.initiate_tag()) {
RTC_DLOG(LS_VERBOSE)
<< log_prefix()
<< "Received duplicate COOKIE-ECHO, probably because of peer not "
"receiving COOKIE-ACK and retransmitting COOKIE-ECHO. Continuing.";
}
return true;
}
void DcSctpSocket::HandleCookieAck(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<CookieAckChunk> chunk = CookieAckChunk::Parse(descriptor.data);
if (!ValidateParseSuccess(chunk)) {
return;
}
if (state_ != State::kCookieEchoed) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Received COOKIE_ACK not in COOKIE_ECHOED state";
return;
}
t1_cookie_->Stop();
tcb_->ClearCookieEchoChunk();
SetState(State::kEstablished, "COOKIE_ACK received");
tcb_->SendBufferedPackets(callbacks_.TimeMillis());
callbacks_.OnConnected();
}
void DcSctpSocket::MaybeDeliverMessages() {
for (auto& message : tcb_->reassembly_queue().FlushMessages()) {
++metrics_.rx_messages_count;
callbacks_.OnMessageReceived(std::move(message));
}
}
void DcSctpSocket::HandleSack(const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<SackChunk> chunk = SackChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk) && ValidateHasTCB()) {
TimeMs now = callbacks_.TimeMillis();
SackChunk sack = ChunkValidators::Clean(*std::move(chunk));
if (tcb_->retransmission_queue().HandleSack(now, sack)) {
MaybeSendShutdownOrAck();
tcb_->MaybeSendFastRetransmit();
tcb_->SendBufferedPackets(now);
} else {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Dropping out-of-order SACK with TSN "
<< *sack.cumulative_tsn_ack();
}
}
}
void DcSctpSocket::HandleHeartbeatRequest(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<HeartbeatRequestChunk> chunk =
HeartbeatRequestChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk) && ValidateHasTCB()) {
tcb_->heartbeat_handler().HandleHeartbeatRequest(*std::move(chunk));
}
}
void DcSctpSocket::HandleHeartbeatAck(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<HeartbeatAckChunk> chunk =
HeartbeatAckChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk) && ValidateHasTCB()) {
tcb_->heartbeat_handler().HandleHeartbeatAck(*std::move(chunk));
}
}
void DcSctpSocket::HandleAbort(const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<AbortChunk> chunk = AbortChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk)) {
std::string error_string = ErrorCausesToString(chunk->error_causes());
if (tcb_ == nullptr) {
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Received ABORT (" << error_string
<< ") on a connection with no TCB. Ignoring";
return;
}
RTC_DLOG(LS_WARNING) << log_prefix() << "Received ABORT (" << error_string
<< ") - closing connection.";
InternalClose(ErrorKind::kPeerReported, error_string);
}
}
void DcSctpSocket::HandleError(const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<ErrorChunk> chunk = ErrorChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk)) {
std::string error_string = ErrorCausesToString(chunk->error_causes());
if (tcb_ == nullptr) {
RTC_DLOG(LS_VERBOSE) << log_prefix() << "Received ERROR (" << error_string
<< ") on a connection with no TCB. Ignoring";
return;
}
RTC_DLOG(LS_WARNING) << log_prefix() << "Received ERROR: " << error_string;
callbacks_.OnError(ErrorKind::kPeerReported,
"Peer reported error: " + error_string);
}
}
void DcSctpSocket::HandleReconfig(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
TimeMs now = callbacks_.TimeMillis();
absl::optional<ReConfigChunk> chunk = ReConfigChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk) && ValidateHasTCB()) {
tcb_->stream_reset_handler().HandleReConfig(*std::move(chunk));
MaybeSendResetStreamsRequest();
tcb_->SendBufferedPackets(now);
MaybeDeliverMessages();
}
}
void DcSctpSocket::HandleShutdown(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
if (!ValidateParseSuccess(ShutdownChunk::Parse(descriptor.data))) {
return;
}
if (state_ == State::kClosed) {
return;
} else if (state_ == State::kCookieWait || state_ == State::kCookieEchoed) {
} else if (state_ == State::kShutdownSent) {
SendShutdownAck();
SetState(State::kShutdownAckSent, "SHUTDOWN received");
} else if (state_ == State::kShutdownAckSent) {
return;
} else if (state_ != State::kShutdownReceived) {
RTC_DLOG(LS_VERBOSE) << log_prefix()
<< "Received SHUTDOWN - shutting down the socket";
SetState(State::kShutdownReceived, "SHUTDOWN received");
MaybeSendShutdownOrAck();
}
}
void DcSctpSocket::HandleShutdownAck(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
if (!ValidateParseSuccess(ShutdownAckChunk::Parse(descriptor.data))) {
return;
}
if (state_ == State::kShutdownSent || state_ == State::kShutdownAckSent) {
SctpPacket::Builder b = tcb_->PacketBuilder();
b.Add(ShutdownCompleteChunk(false));
packet_sender_.Send(b);
InternalClose(ErrorKind::kNoError, "");
} else {
SctpPacket::Builder b(header.verification_tag, options_);
b.Add(ShutdownCompleteChunk(true));
packet_sender_.Send(b);
}
}
void DcSctpSocket::HandleShutdownComplete(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
if (!ValidateParseSuccess(ShutdownCompleteChunk::Parse(descriptor.data))) {
return;
}
if (state_ == State::kShutdownAckSent) {
InternalClose(ErrorKind::kNoError, "");
}
}
void DcSctpSocket::HandleForwardTsn(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<ForwardTsnChunk> chunk =
ForwardTsnChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk) && ValidateHasTCB()) {
HandleForwardTsnCommon(*chunk);
}
}
void DcSctpSocket::HandleIForwardTsn(
const CommonHeader& header,
const SctpPacket::ChunkDescriptor& descriptor) {
absl::optional<IForwardTsnChunk> chunk =
IForwardTsnChunk::Parse(descriptor.data);
if (ValidateParseSuccess(chunk) && ValidateHasTCB()) {
HandleForwardTsnCommon(*chunk);
}
}
void DcSctpSocket::HandleForwardTsnCommon(const AnyForwardTsnChunk& chunk) {
if (!tcb_->capabilities().partial_reliability) {
SctpPacket::Builder b = tcb_->PacketBuilder();
b.Add(AbortChunk(true,
Parameters::Builder()
.Add(ProtocolViolationCause(
"I-FORWARD-TSN received, but not indicated "
"during connection establishment"))
.Build()));
packet_sender_.Send(b);
callbacks_.OnError(ErrorKind::kProtocolViolation,
"Received a FORWARD_TSN without announced peer support");
return;
}
if (tcb_->data_tracker().HandleForwardTsn(chunk.new_cumulative_tsn())) {
tcb_->reassembly_queue().HandleForwardTsn(chunk.new_cumulative_tsn(),
chunk.skipped_streams());
}
MaybeDeliverMessages();
}
void DcSctpSocket::MaybeSendShutdownOrAck() {
if (tcb_->retransmission_queue().outstanding_bytes() != 0) {
return;
}
if (state_ == State::kShutdownPending) {
SendShutdown();
t2_shutdown_->set_duration(tcb_->current_rto());
t2_shutdown_->Start();
SetState(State::kShutdownSent, "No more outstanding data");
} else if (state_ == State::kShutdownReceived) {
SendShutdownAck();
SetState(State::kShutdownAckSent, "No more outstanding data");
}
}
void DcSctpSocket::SendShutdown() {
SctpPacket::Builder b = tcb_->PacketBuilder();
b.Add(ShutdownChunk(tcb_->data_tracker().last_cumulative_acked_tsn()));
packet_sender_.Send(b);
}
void DcSctpSocket::SendShutdownAck() {
packet_sender_.Send(tcb_->PacketBuilder().Add(ShutdownAckChunk()));
t2_shutdown_->set_duration(tcb_->current_rto());
t2_shutdown_->Start();
}
HandoverReadinessStatus DcSctpSocket::GetHandoverReadiness() const {
RTC_DCHECK_RUN_ON(&thread_checker_);
HandoverReadinessStatus status;
if (state_ != State::kClosed && state_ != State::kEstablished) {
status.Add(HandoverUnreadinessReason::kWrongConnectionState);
}
status.Add(send_queue_.GetHandoverReadiness());
if (tcb_) {
status.Add(tcb_->GetHandoverReadiness());
}
return status;
}
absl::optional<DcSctpSocketHandoverState>
DcSctpSocket::GetHandoverStateAndClose() {
RTC_DCHECK_RUN_ON(&thread_checker_);
CallbackDeferrer::ScopedDeferrer deferrer(callbacks_);
if (!GetHandoverReadiness().IsReady()) {
return absl::nullopt;
}
DcSctpSocketHandoverState state;
if (state_ == State::kClosed) {
state.socket_state = DcSctpSocketHandoverState::SocketState::kClosed;
} else if (state_ == State::kEstablished) {
state.socket_state = DcSctpSocketHandoverState::SocketState::kConnected;
tcb_->AddHandoverState(state);
send_queue_.AddHandoverState(state);
InternalClose(ErrorKind::kNoError, "handover");
}
return std::move(state);
}
}