blob: 1a9fc9bb0ff3096380bcf6271b8b7d2ffa1ed029 [file] [log] [blame] [edit]
// Copyright 2019 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "quic/masque/masque_client_session.h"
#include "absl/algorithm/container.h"
#include "quic/core/http/spdy_utils.h"
#include "quic/core/quic_data_reader.h"
#include "quic/core/quic_utils.h"
#include "common/platform/api/quiche_text_utils.h"
namespace quic {
MasqueClientSession::MasqueClientSession(
MasqueMode masque_mode,
const QuicConfig& config,
const ParsedQuicVersionVector& supported_versions,
QuicConnection* connection,
const QuicServerId& server_id,
QuicCryptoClientConfig* crypto_config,
QuicClientPushPromiseIndex* push_promise_index,
Owner* owner)
: QuicSpdyClientSession(config,
supported_versions,
connection,
server_id,
crypto_config,
push_promise_index),
masque_mode_(masque_mode),
owner_(owner),
compression_engine_(this) {}
void MasqueClientSession::OnMessageReceived(absl::string_view message) {
if (masque_mode_ == MasqueMode::kLegacy) {
QUIC_DVLOG(1) << "Received DATAGRAM frame of length " << message.length();
QuicConnectionId client_connection_id, server_connection_id;
QuicSocketAddress target_server_address;
std::vector<char> packet;
bool version_present;
if (!compression_engine_.DecompressDatagram(
message, &client_connection_id, &server_connection_id,
&target_server_address, &packet, &version_present)) {
return;
}
auto connection_id_registration =
client_connection_id_registrations_.find(client_connection_id);
if (connection_id_registration ==
client_connection_id_registrations_.end()) {
QUIC_DLOG(ERROR) << "MasqueClientSession failed to dispatch "
<< client_connection_id;
return;
}
EncapsulatedClientSession* encapsulated_client_session =
connection_id_registration->second;
encapsulated_client_session->ProcessPacket(
absl::string_view(packet.data(), packet.size()), target_server_address);
QUIC_DVLOG(1) << "Sent " << packet.size() << " bytes to connection for "
<< client_connection_id;
return;
}
QUICHE_DCHECK_EQ(masque_mode_, MasqueMode::kOpen);
QuicSpdySession::OnMessageReceived(message);
}
void MasqueClientSession::OnMessageAcked(QuicMessageId message_id,
QuicTime /*receive_timestamp*/) {
QUIC_DVLOG(1) << "Received ack for DATAGRAM frame " << message_id;
}
void MasqueClientSession::OnMessageLost(QuicMessageId message_id) {
QUIC_DVLOG(1) << "We believe DATAGRAM frame " << message_id << " was lost";
}
const MasqueClientSession::ConnectUdpClientState*
MasqueClientSession::GetOrCreateConnectUdpClientState(
const QuicSocketAddress& target_server_address,
EncapsulatedClientSession* encapsulated_client_session) {
for (const ConnectUdpClientState& client_state : connect_udp_client_states_) {
if (client_state.target_server_address() == target_server_address &&
client_state.encapsulated_client_session() ==
encapsulated_client_session) {
// Found existing CONNECT-UDP request.
return &client_state;
}
}
// No CONNECT-UDP request found, create a new one.
QuicSpdyClientStream* stream = CreateOutgoingBidirectionalStream();
if (stream == nullptr) {
// Stream flow control limits prevented us from opening a new stream.
QUIC_DLOG(ERROR) << "Failed to open CONNECT-UDP stream";
return nullptr;
}
QuicDatagramFlowId flow_id = GetNextDatagramFlowId();
// Send the request.
spdy::Http2HeaderBlock headers;
headers[":method"] = "CONNECT-UDP";
headers[":scheme"] = "masque";
headers[":path"] = "/";
headers[":authority"] = target_server_address.ToString();
SpdyUtils::AddDatagramFlowIdHeader(&headers, flow_id);
size_t bytes_sent =
stream->SendRequest(std::move(headers), /*body=*/"", /*fin=*/false);
if (bytes_sent == 0) {
QUIC_DLOG(ERROR) << "Failed to send CONNECT-UDP request";
return nullptr;
}
connect_udp_client_states_.push_back(
ConnectUdpClientState(stream, encapsulated_client_session, this, flow_id,
target_server_address));
return &connect_udp_client_states_.back();
}
void MasqueClientSession::SendPacket(
QuicConnectionId client_connection_id,
QuicConnectionId server_connection_id,
absl::string_view packet,
const QuicSocketAddress& target_server_address,
EncapsulatedClientSession* encapsulated_client_session) {
if (masque_mode_ == MasqueMode::kLegacy) {
compression_engine_.CompressAndSendPacket(packet, client_connection_id,
server_connection_id,
target_server_address);
return;
}
const ConnectUdpClientState* connect_udp = GetOrCreateConnectUdpClientState(
target_server_address, encapsulated_client_session);
if (connect_udp == nullptr) {
QUIC_DLOG(ERROR) << "Failed to create CONNECT-UDP request";
return;
}
QuicDatagramFlowId flow_id = connect_udp->flow_id();
MessageStatus message_status =
SendHttp3Datagram(connect_udp->flow_id(), packet);
QUIC_DVLOG(1) << "Sent packet to " << target_server_address
<< " compressed with flow ID " << flow_id
<< " and got message status " << message_status;
}
void MasqueClientSession::RegisterConnectionId(
QuicConnectionId client_connection_id,
EncapsulatedClientSession* encapsulated_client_session) {
QUIC_DLOG(INFO) << "Registering " << client_connection_id
<< " to encapsulated client";
QUICHE_DCHECK(
client_connection_id_registrations_.find(client_connection_id) ==
client_connection_id_registrations_.end() ||
client_connection_id_registrations_[client_connection_id] ==
encapsulated_client_session);
client_connection_id_registrations_[client_connection_id] =
encapsulated_client_session;
}
void MasqueClientSession::UnregisterConnectionId(
QuicConnectionId client_connection_id,
EncapsulatedClientSession* encapsulated_client_session) {
QUIC_DLOG(INFO) << "Unregistering " << client_connection_id;
if (masque_mode_ == MasqueMode::kLegacy) {
if (client_connection_id_registrations_.find(client_connection_id) !=
client_connection_id_registrations_.end()) {
client_connection_id_registrations_.erase(client_connection_id);
owner_->UnregisterClientConnectionId(client_connection_id);
compression_engine_.UnregisterClientConnectionId(client_connection_id);
}
return;
}
for (auto it = connect_udp_client_states_.begin();
it != connect_udp_client_states_.end();) {
if (it->encapsulated_client_session() == encapsulated_client_session) {
QUIC_DLOG(INFO) << "Removing state for flow_id " << it->flow_id();
auto* stream = it->stream();
it = connect_udp_client_states_.erase(it);
if (!stream->write_side_closed()) {
stream->Reset(QUIC_STREAM_CANCELLED);
}
} else {
++it;
}
}
}
void MasqueClientSession::OnConnectionClosed(
const QuicConnectionCloseFrame& frame,
ConnectionCloseSource source) {
QuicSpdyClientSession::OnConnectionClosed(frame, source);
// Close all encapsulated sessions.
for (const auto& client_state : connect_udp_client_states_) {
client_state.encapsulated_client_session()->CloseConnection(
QUIC_CONNECTION_CANCELLED, "Underlying MASQUE connection was closed",
ConnectionCloseBehavior::SILENT_CLOSE);
}
}
void MasqueClientSession::OnStreamClosed(QuicStreamId stream_id) {
if (QuicUtils::IsBidirectionalStreamId(stream_id, version()) &&
QuicUtils::IsClientInitiatedStreamId(transport_version(), stream_id)) {
QuicSpdyClientStream* stream =
reinterpret_cast<QuicSpdyClientStream*>(GetActiveStream(stream_id));
if (stream != nullptr) {
QUIC_DLOG(INFO) << "Stream " << stream_id
<< " closed, got response headers:"
<< stream->response_headers().DebugString();
}
}
for (auto it = connect_udp_client_states_.begin();
it != connect_udp_client_states_.end();) {
if (it->stream()->id() == stream_id) {
QUIC_DLOG(INFO) << "Stream " << stream_id
<< " was closed, removing state for flow_id "
<< it->flow_id();
auto* encapsulated_client_session = it->encapsulated_client_session();
it = connect_udp_client_states_.erase(it);
encapsulated_client_session->CloseConnection(
QUIC_CONNECTION_CANCELLED,
"Underlying MASQUE CONNECT-UDP stream was closed",
ConnectionCloseBehavior::SILENT_CLOSE);
} else {
++it;
}
}
QuicSpdyClientSession::OnStreamClosed(stream_id);
}
MasqueClientSession::ConnectUdpClientState::ConnectUdpClientState(
QuicSpdyClientStream* stream,
EncapsulatedClientSession* encapsulated_client_session,
MasqueClientSession* masque_session,
QuicDatagramFlowId flow_id,
const QuicSocketAddress& target_server_address)
: stream_(stream),
encapsulated_client_session_(encapsulated_client_session),
masque_session_(masque_session),
flow_id_(flow_id),
target_server_address_(target_server_address) {
QUICHE_DCHECK_NE(masque_session_, nullptr);
masque_session_->RegisterHttp3FlowId(this->flow_id(), this);
}
MasqueClientSession::ConnectUdpClientState::~ConnectUdpClientState() {
if (flow_id_.has_value()) {
masque_session_->UnregisterHttp3FlowId(flow_id());
}
}
MasqueClientSession::ConnectUdpClientState::ConnectUdpClientState(
MasqueClientSession::ConnectUdpClientState&& other) {
*this = std::move(other);
}
MasqueClientSession::ConnectUdpClientState&
MasqueClientSession::ConnectUdpClientState::operator=(
MasqueClientSession::ConnectUdpClientState&& other) {
stream_ = other.stream_;
encapsulated_client_session_ = other.encapsulated_client_session_;
masque_session_ = other.masque_session_;
flow_id_ = other.flow_id_;
target_server_address_ = other.target_server_address_;
other.flow_id_.reset();
if (flow_id_.has_value()) {
masque_session_->UnregisterHttp3FlowId(flow_id());
masque_session_->RegisterHttp3FlowId(flow_id(), this);
}
return *this;
}
void MasqueClientSession::ConnectUdpClientState::OnHttp3Datagram(
QuicDatagramFlowId flow_id,
absl::string_view payload) {
QUICHE_DCHECK_EQ(flow_id, this->flow_id());
encapsulated_client_session_->ProcessPacket(payload, target_server_address_);
QUIC_DVLOG(1) << "Sent " << payload.size()
<< " bytes to connection for flow_id " << flow_id;
}
} // namespace quic