Implement MASQUE CONNECT-UDP Previously, the MASQUE code was supporting the legacy MASQUE protocol. This CL adds support for CONNECT-UDP, the new MASQUE wire format. This is test code currently meant for IETF interop events. We're mainly landing it to prevent bitrot. This code isn't used in production. PiperOrigin-RevId: 359575658 Change-Id: Ia883662fb6ffe7827c15b1460f2b316494b7ffbd
diff --git a/quic/masque/masque_client_bin.cc b/quic/masque/masque_client_bin.cc index 8e09a12..afd991a 100644 --- a/quic/masque/masque_client_bin.cc +++ b/quic/masque/masque_client_bin.cc
@@ -29,6 +29,12 @@ false, "If true, don't verify the server certificate."); +DEFINE_QUIC_COMMAND_LINE_FLAG(std::string, + masque_mode, + "", + "Allows setting MASQUE mode, valid values are " + "open and legacy. Defaults to open."); + namespace quic { namespace { @@ -54,7 +60,8 @@ masque_url = QuicUrl(absl::StrCat("https://", urls[0]), "https"); } if (masque_url.host().empty()) { - QUIC_LOG(ERROR) << "Failed to parse MASQUE server address " << urls[0]; + std::cerr << "Failed to parse MASQUE server address \"" << urls[0] << "\"" + << std::endl; return 1; } std::unique_ptr<ProofVerifier> proof_verifier; @@ -63,15 +70,23 @@ } else { proof_verifier = CreateDefaultProofVerifier(masque_url.host()); } - std::unique_ptr<MasqueEpollClient> masque_client = - MasqueEpollClient::Create(masque_url.host(), masque_url.port(), - &epoll_server, std::move(proof_verifier)); + MasqueMode masque_mode = MasqueMode::kOpen; + std::string mode_string = GetQuicFlag(FLAGS_masque_mode); + if (mode_string == "legacy") { + masque_mode = MasqueMode::kLegacy; + } else if (!mode_string.empty() && mode_string != "open") { + std::cerr << "Invalid masque_mode \"" << mode_string << "\"" << std::endl; + return 1; + } + std::unique_ptr<MasqueEpollClient> masque_client = MasqueEpollClient::Create( + masque_url.host(), masque_url.port(), masque_mode, &epoll_server, + std::move(proof_verifier)); if (masque_client == nullptr) { return 1; } std::cerr << "MASQUE is connected " << masque_client->connection_id() - << std::endl; + << " in " << masque_mode << " mode" << std::endl; for (size_t i = 1; i < urls.size(); ++i) { if (!tools::SendEncapsulatedMasqueRequest(
diff --git a/quic/masque/masque_client_session.cc b/quic/masque/masque_client_session.cc index eeedccf..5663c4a 100644 --- a/quic/masque/masque_client_session.cc +++ b/quic/masque/masque_client_session.cc
@@ -3,10 +3,14 @@ // found in the LICENSE file. #include "quic/masque/masque_client_session.h" +#include "absl/algorithm/container.h" +#include "quic/core/quic_data_reader.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, @@ -20,36 +24,65 @@ server_id, crypto_config, push_promise_index), + masque_mode_(masque_mode), owner_(owner), compression_engine_(this) {} void MasqueClientSession::OnMessageReceived(absl::string_view message) { QUIC_DVLOG(1) << "Received DATAGRAM frame of length " << message.length(); + if (masque_mode_ == MasqueMode::kLegacy) { + 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; + } - QuicConnectionId client_connection_id, server_connection_id; - QuicSocketAddress server_address; - std::vector<char> packet; - bool version_present; - if (!compression_engine_.DecompressDatagram( - message, &client_connection_id, &server_connection_id, - &server_address, &packet, &version_present)) { + 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; } - - 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; + QuicDataReader reader(message); + QuicDatagramFlowId flow_id; + if (!reader.ReadVarInt62(&flow_id)) { + QUIC_DLOG(ERROR) << "Failed to parse flow_id"; + return; + } + auto it = + absl::c_find_if(connect_udp_client_states_, + [flow_id](const ConnectUdpClientState& connect_udp) { + return connect_udp.flow_id() == flow_id; + }); + if (it == connect_udp_client_states_.end()) { + QUIC_DLOG(ERROR) << "Received unknown flow_id " << flow_id; return; } EncapsulatedClientSession* encapsulated_client_session = - connection_id_registration->second; - encapsulated_client_session->ProcessPacket( - absl::string_view(packet.data(), packet.size()), server_address); + it->encapsulated_client_session(); + QuicSocketAddress target_server_address = it->target_server_address(); + QUICHE_DCHECK_NE(encapsulated_client_session, nullptr); + QUICHE_DCHECK(target_server_address.IsInitialized()); + absl::string_view packet = reader.ReadRemainingPayload(); + encapsulated_client_session->ProcessPacket(packet, target_server_address); - QUIC_DVLOG(1) << "Sent " << packet.size() << " bytes to connection for " - << client_connection_id; + QUIC_DVLOG(1) << "Sent " << packet.size() + << " bytes to connection for flow_id " << flow_id; } void MasqueClientSession::OnMessageAcked(QuicMessageId message_id, @@ -61,12 +94,87 @@ QUIC_DVLOG(1) << "We believe DATAGRAM frame " << message_id << " was lost"; } -void MasqueClientSession::SendPacket(QuicConnectionId client_connection_id, - QuicConnectionId server_connection_id, - absl::string_view packet, - const QuicSocketAddress& server_address) { - compression_engine_.CompressAndSendPacket( - packet, client_connection_id, server_connection_id, server_address); +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 = compression_engine_.GetNextFlowId(); + + // Send the request. + spdy::Http2HeaderBlock headers; + headers[":method"] = "CONNECT-UDP"; + headers[":scheme"] = "masque"; + headers[":path"] = "/"; + headers[":authority"] = target_server_address.ToString(); + headers["datagram-flow-id"] = absl::StrCat(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, 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(); + size_t slice_length = + QuicDataWriter::GetVarInt62Len(flow_id) + packet.length(); + QuicUniqueBufferPtr buffer = MakeUniqueBuffer( + connection()->helper()->GetStreamSendBufferAllocator(), slice_length); + QuicDataWriter writer(slice_length, buffer.get()); + if (!writer.WriteVarInt62(flow_id)) { + QUIC_BUG << "Failed to write flow_id"; + return; + } + if (!writer.WriteBytes(packet.data(), packet.length())) { + QUIC_BUG << "Failed to write packet"; + return; + } + + QuicMemSlice slice(std::move(buffer), slice_length); + MessageResult message_result = SendMessage(QuicMemSliceSpan(&slice)); + + QUIC_DVLOG(1) << "Sent packet to " << target_server_address + << " compressed with flow ID " << flow_id + << " and got message result " << message_result; } void MasqueClientSession::RegisterConnectionId( @@ -84,14 +192,65 @@ } void MasqueClientSession::UnregisterConnectionId( - QuicConnectionId client_connection_id) { + QuicConnectionId client_connection_id, + EncapsulatedClientSession* encapsulated_client_session) { QUIC_DLOG(INFO) << "Unregistering " << client_connection_id; - 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); + 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 (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) { + 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); } } // namespace quic
diff --git a/quic/masque/masque_client_session.h b/quic/masque/masque_client_session.h index 0eab5b2..f849c61 100644 --- a/quic/masque/masque_client_session.h +++ b/quic/masque/masque_client_session.h
@@ -9,6 +9,7 @@ #include "absl/strings/string_view.h" #include "quic/core/http/quic_spdy_client_session.h" #include "quic/masque/masque_compression_engine.h" +#include "quic/masque/masque_utils.h" #include "quic/platform/api/quic_export.h" #include "quic/platform/api/quic_socket_address.h" @@ -39,14 +40,21 @@ // Process packet that was just decapsulated. virtual void ProcessPacket(absl::string_view packet, - QuicSocketAddress server_address) = 0; + QuicSocketAddress target_server_address) = 0; + + // Close the encapsulated connection. + virtual void CloseConnection( + QuicErrorCode error, + const std::string& details, + ConnectionCloseBehavior connection_close_behavior) = 0; }; // Takes ownership of |connection|, but not of |crypto_config| or // |push_promise_index| or |owner|. All pointers must be non-null. Caller // must ensure that |push_promise_index| and |owner| stay valid for the // lifetime of the newly created MasqueClientSession. - MasqueClientSession(const QuicConfig& config, + MasqueClientSession(MasqueMode masque_mode, + const QuicConfig& config, const ParsedQuicVersionVector& supported_versions, QuicConnection* connection, const QuicServerId& server_id, @@ -60,17 +68,19 @@ // From QuicSession. void OnMessageReceived(absl::string_view message) override; - void OnMessageAcked(QuicMessageId message_id, QuicTime receive_timestamp) override; - void OnMessageLost(QuicMessageId message_id) override; + void OnConnectionClosed(const QuicConnectionCloseFrame& frame, + ConnectionCloseSource source) override; + void OnStreamClosed(QuicStreamId stream_id) override; // Send encapsulated packet. void SendPacket(QuicConnectionId client_connection_id, QuicConnectionId server_connection_id, absl::string_view packet, - const QuicSocketAddress& server_address); + const QuicSocketAddress& target_server_address, + EncapsulatedClientSession* encapsulated_client_session); // Register encapsulated client. This allows clients that are encapsulated // within this MASQUE session to indicate they own a given client connection @@ -84,9 +94,48 @@ // Unregister encapsulated client. |client_connection_id| must match a // value previously passed to RegisterConnectionId. - void UnregisterConnectionId(QuicConnectionId client_connection_id); + void UnregisterConnectionId( + QuicConnectionId client_connection_id, + EncapsulatedClientSession* encapsulated_client_session); private: + // State that the MasqueClientSession keeps for each CONNECT-UDP request. + class QUIC_NO_EXPORT ConnectUdpClientState { + public: + // |stream| and |encapsulated_client_session| must be valid for the lifetime + // of the ConnectUdpClientState. + explicit ConnectUdpClientState( + QuicSpdyClientStream* stream, + EncapsulatedClientSession* encapsulated_client_session, + QuicDatagramFlowId flow_id, + const QuicSocketAddress& target_server_address) + : stream_(stream), + encapsulated_client_session_(encapsulated_client_session), + flow_id_(flow_id), + target_server_address_(target_server_address) {} + + QuicSpdyClientStream* stream() const { return stream_; } + EncapsulatedClientSession* encapsulated_client_session() const { + return encapsulated_client_session_; + } + QuicDatagramFlowId flow_id() const { return flow_id_; } + const QuicSocketAddress& target_server_address() const { + return target_server_address_; + } + + private: + QuicSpdyClientStream* stream_; // Unowned. + EncapsulatedClientSession* encapsulated_client_session_; // Unowned. + QuicDatagramFlowId flow_id_; + QuicSocketAddress target_server_address_; + }; + + const ConnectUdpClientState* GetOrCreateConnectUdpClientState( + const QuicSocketAddress& target_server_address, + EncapsulatedClientSession* encapsulated_client_session); + + MasqueMode masque_mode_; + std::list<ConnectUdpClientState> connect_udp_client_states_; absl::flat_hash_map<QuicConnectionId, EncapsulatedClientSession*, QuicConnectionIdHash>
diff --git a/quic/masque/masque_compression_engine.h b/quic/masque/masque_compression_engine.h index 96c5948..16b4549 100644 --- a/quic/masque/masque_compression_engine.h +++ b/quic/masque/masque_compression_engine.h
@@ -70,6 +70,9 @@ // compression table. void UnregisterClientConnectionId(QuicConnectionId client_connection_id); + // Generates a new datagram flow ID. + QuicDatagramFlowId GetNextFlowId(); + private: struct QUIC_NO_EXPORT MasqueCompressionContext { QuicConnectionId client_connection_id; @@ -78,9 +81,6 @@ bool validated = false; }; - // Generates a new datagram flow ID. - QuicDatagramFlowId GetNextFlowId(); - // Finds or creates a new compression context to use during compression. // |client_connection_id_present| and |server_connection_id_present| indicate // whether the corresponding connection ID is present in the current packet.
diff --git a/quic/masque/masque_dispatcher.cc b/quic/masque/masque_dispatcher.cc index 6edca62..03eb8f8 100644 --- a/quic/masque/masque_dispatcher.cc +++ b/quic/masque/masque_dispatcher.cc
@@ -8,9 +8,11 @@ namespace quic { MasqueDispatcher::MasqueDispatcher( + MasqueMode masque_mode, const QuicConfig* config, const QuicCryptoServerConfig* crypto_config, QuicVersionManager* version_manager, + QuicEpollServer* epoll_server, std::unique_ptr<QuicConnectionHelperInterface> helper, std::unique_ptr<QuicCryptoServerStreamBase::Helper> session_helper, std::unique_ptr<QuicAlarmFactory> alarm_factory, @@ -24,6 +26,8 @@ std::move(alarm_factory), masque_server_backend, expected_server_connection_id_length), + masque_mode_(masque_mode), + epoll_server_(epoll_server), masque_server_backend_(masque_server_backend) {} std::unique_ptr<QuicSession> MasqueDispatcher::CreateQuicSession( @@ -40,9 +44,9 @@ ParsedQuicVersionVector{version}); auto session = std::make_unique<MasqueServerSession>( - config(), GetSupportedVersions(), connection, this, this, - session_helper(), crypto_config(), compressed_certs_cache(), - masque_server_backend_); + masque_mode_, config(), GetSupportedVersions(), connection, this, this, + epoll_server_, session_helper(), crypto_config(), + compressed_certs_cache(), masque_server_backend_); session->Initialize(); return session; }
diff --git a/quic/masque/masque_dispatcher.h b/quic/masque/masque_dispatcher.h index 2e0186c..ce371b5 100644 --- a/quic/masque/masque_dispatcher.h +++ b/quic/masque/masque_dispatcher.h
@@ -8,6 +8,8 @@ #include "absl/container/flat_hash_map.h" #include "quic/masque/masque_server_backend.h" #include "quic/masque/masque_server_session.h" +#include "quic/masque/masque_utils.h" +#include "quic/platform/api/quic_epoll.h" #include "quic/platform/api/quic_export.h" #include "quic/tools/quic_simple_dispatcher.h" @@ -19,9 +21,11 @@ public MasqueServerSession::Visitor { public: explicit MasqueDispatcher( + MasqueMode masque_mode, const QuicConfig* config, const QuicCryptoServerConfig* crypto_config, QuicVersionManager* version_manager, + QuicEpollServer* epoll_server, std::unique_ptr<QuicConnectionHelperInterface> helper, std::unique_ptr<QuicCryptoServerStreamBase::Helper> session_helper, std::unique_ptr<QuicAlarmFactory> alarm_factory, @@ -51,6 +55,8 @@ QuicConnectionId client_connection_id) override; private: + MasqueMode masque_mode_; + QuicEpollServer* epoll_server_; // Unowned. MasqueServerBackend* masque_server_backend_; // Unowned. // Mapping from client connection IDs to server sessions, allows routing // incoming packets to the right MASQUE connection.
diff --git a/quic/masque/masque_encapsulated_client_session.cc b/quic/masque/masque_encapsulated_client_session.cc index 983a094..3d557c3 100644 --- a/quic/masque/masque_encapsulated_client_session.cc +++ b/quic/masque/masque_encapsulated_client_session.cc
@@ -31,11 +31,19 @@ received_packet); } +void MasqueEncapsulatedClientSession::CloseConnection( + QuicErrorCode error, + const std::string& details, + ConnectionCloseBehavior connection_close_behavior) { + connection()->CloseConnection(error, details, connection_close_behavior); +} + void MasqueEncapsulatedClientSession::OnConnectionClosed( - const QuicConnectionCloseFrame& /*frame*/, - ConnectionCloseSource /*source*/) { + const QuicConnectionCloseFrame& frame, + ConnectionCloseSource source) { + QuicSpdyClientSession::OnConnectionClosed(frame, source); masque_client_session_->UnregisterConnectionId( - connection()->client_connection_id()); + connection()->client_connection_id(), this); } } // namespace quic
diff --git a/quic/masque/masque_encapsulated_client_session.h b/quic/masque/masque_encapsulated_client_session.h index 570ccdb..0952159 100644 --- a/quic/masque/masque_encapsulated_client_session.h +++ b/quic/masque/masque_encapsulated_client_session.h
@@ -44,6 +44,10 @@ // From MasqueClientSession::EncapsulatedClientSession. void ProcessPacket(absl::string_view packet, QuicSocketAddress server_address) override; + void CloseConnection( + QuicErrorCode error, + const std::string& details, + ConnectionCloseBehavior connection_close_behavior) override; // From QuicSession. void OnConnectionClosed(const QuicConnectionCloseFrame& frame,
diff --git a/quic/masque/masque_encapsulated_epoll_client.cc b/quic/masque/masque_encapsulated_epoll_client.cc index 8fb938d..8891e4a 100644 --- a/quic/masque/masque_encapsulated_epoll_client.cc +++ b/quic/masque/masque_encapsulated_epoll_client.cc
@@ -30,8 +30,8 @@ absl::string_view packet(buffer, buf_len); client_->masque_client()->masque_client_session()->SendPacket( client_->session()->connection()->client_connection_id(), - client_->session()->connection()->connection_id(), packet, - peer_address); + client_->session()->connection()->connection_id(), packet, peer_address, + client_->masque_encapsulated_client_session()); return WriteResult(WRITE_STATUS_OK, buf_len); } @@ -94,7 +94,7 @@ MasqueEncapsulatedEpollClient::~MasqueEncapsulatedEpollClient() { masque_client_->masque_client_session()->UnregisterConnectionId( - client_connection_id_); + client_connection_id_, masque_encapsulated_client_session()); } std::unique_ptr<QuicSession>
diff --git a/quic/masque/masque_epoll_client.cc b/quic/masque/masque_epoll_client.cc index 1d77e55..bf09da4 100644 --- a/quic/masque/masque_epoll_client.cc +++ b/quic/masque/masque_epoll_client.cc
@@ -12,6 +12,7 @@ MasqueEpollClient::MasqueEpollClient( QuicSocketAddress server_address, const QuicServerId& server_id, + MasqueMode masque_mode, QuicEpollServer* epoll_server, std::unique_ptr<ProofVerifier> proof_verifier, const std::string& authority) @@ -20,6 +21,7 @@ MasqueSupportedVersions(), epoll_server, std::move(proof_verifier)), + masque_mode_(masque_mode), authority_(authority) {} std::unique_ptr<QuicSession> MasqueEpollClient::CreateQuicClientSession( @@ -28,8 +30,8 @@ QUIC_DLOG(INFO) << "Creating MASQUE session for " << connection->connection_id(); return std::make_unique<MasqueClientSession>( - *config(), supported_versions, connection, server_id(), crypto_config(), - push_promise_index(), this); + masque_mode_, *config(), supported_versions, connection, server_id(), + crypto_config(), push_promise_index(), this); } MasqueClientSession* MasqueEpollClient::masque_client_session() { @@ -44,6 +46,7 @@ std::unique_ptr<MasqueEpollClient> MasqueEpollClient::Create( const std::string& host, int port, + MasqueMode masque_mode, QuicEpollServer* epoll_server, std::unique_ptr<ProofVerifier> proof_verifier) { // Build the masque_client, and try to connect. @@ -57,7 +60,7 @@ // std::make_unique<MasqueEpollClient>(...) because the constructor for // MasqueEpollClient is private and therefore not accessible from make_unique. auto masque_client = QuicWrapUnique(new MasqueEpollClient( - addr, server_id, epoll_server, std::move(proof_verifier), + addr, server_id, masque_mode, epoll_server, std::move(proof_verifier), absl::StrCat(host, ":", port))); if (masque_client == nullptr) { @@ -78,33 +81,35 @@ return nullptr; } - std::string body = "foo"; + if (masque_client->masque_mode() == MasqueMode::kLegacy) { + // Construct the legacy mode init request. + spdy::Http2HeaderBlock header_block; + header_block[":method"] = "POST"; + header_block[":scheme"] = "https"; + header_block[":authority"] = masque_client->authority_; + header_block[":path"] = "/.well-known/masque/init"; + std::string body = "foo"; - // Construct a GET or POST request for supplied URL. - spdy::Http2HeaderBlock header_block; - header_block[":method"] = "POST"; - header_block[":scheme"] = "https"; - header_block[":authority"] = masque_client->authority_; - header_block[":path"] = "/.well-known/masque/init"; + // Make sure to store the response, for later output. + masque_client->set_store_response(true); - // Make sure to store the response, for later output. - masque_client->set_store_response(true); + // Send the MASQUE init command. + masque_client->SendRequestAndWaitForResponse(header_block, body, + /*fin=*/true); - // Send the MASQUE init command. - masque_client->SendRequestAndWaitForResponse(header_block, body, - /*fin=*/true); + if (!masque_client->connected()) { + QUIC_LOG(ERROR) + << "MASQUE init request caused connection failure. Error: " + << QuicErrorCodeToString(masque_client->session()->error()); + return nullptr; + } - if (!masque_client->connected()) { - QUIC_LOG(ERROR) << "MASQUE init request caused connection failure. Error: " - << QuicErrorCodeToString(masque_client->session()->error()); - return nullptr; - } - - const int response_code = masque_client->latest_response_code(); - if (response_code != 200) { - QUIC_LOG(ERROR) << "MASQUE init request failed with HTTP response code " - << response_code; - return nullptr; + const int response_code = masque_client->latest_response_code(); + if (response_code != 200) { + QUIC_LOG(ERROR) << "MASQUE init request failed with HTTP response code " + << response_code; + return nullptr; + } } return masque_client; }
diff --git a/quic/masque/masque_epoll_client.h b/quic/masque/masque_epoll_client.h index 345c18c..1db161f 100644 --- a/quic/masque/masque_epoll_client.h +++ b/quic/masque/masque_epoll_client.h
@@ -6,6 +6,7 @@ #define QUICHE_QUIC_MASQUE_MASQUE_EPOLL_CLIENT_H_ #include "quic/masque/masque_client_session.h" +#include "quic/masque/masque_utils.h" #include "quic/platform/api/quic_export.h" #include "quic/tools/quic_client.h" @@ -19,6 +20,7 @@ static std::unique_ptr<MasqueEpollClient> Create( const std::string& host, int port, + MasqueMode masque_mode, QuicEpollServer* epoll_server, std::unique_ptr<ProofVerifier> proof_verifier); @@ -37,10 +39,13 @@ void UnregisterClientConnectionId( QuicConnectionId client_connection_id) override; + MasqueMode masque_mode() const { return masque_mode_; } + private: // Constructor is private, use Create() instead. MasqueEpollClient(QuicSocketAddress server_address, const QuicServerId& server_id, + MasqueMode masque_mode, QuicEpollServer* epoll_server, std::unique_ptr<ProofVerifier> proof_verifier, const std::string& authority); @@ -49,6 +54,7 @@ MasqueEpollClient(const MasqueEpollClient&) = delete; MasqueEpollClient& operator=(const MasqueEpollClient&) = delete; + MasqueMode masque_mode_; std::string authority_; };
diff --git a/quic/masque/masque_epoll_server.cc b/quic/masque/masque_epoll_server.cc index 34769a3..e8a0850 100644 --- a/quic/masque/masque_epoll_server.cc +++ b/quic/masque/masque_epoll_server.cc
@@ -11,15 +11,18 @@ namespace quic { -MasqueEpollServer::MasqueEpollServer(MasqueServerBackend* masque_server_backend) +MasqueEpollServer::MasqueEpollServer(MasqueMode masque_mode, + MasqueServerBackend* masque_server_backend) : QuicServer(CreateDefaultProofSource(), masque_server_backend, MasqueSupportedVersions()), + masque_mode_(masque_mode), masque_server_backend_(masque_server_backend) {} QuicDispatcher* MasqueEpollServer::CreateQuicDispatcher() { return new MasqueDispatcher( - &config(), &crypto_config(), version_manager(), + masque_mode_, &config(), &crypto_config(), version_manager(), + epoll_server(), std::make_unique<QuicEpollConnectionHelper>(epoll_server(), QuicAllocator::BUFFER_POOL), std::make_unique<QuicSimpleCryptoServerStreamHelper>(),
diff --git a/quic/masque/masque_epoll_server.h b/quic/masque/masque_epoll_server.h index 57eab6a..88161ee 100644 --- a/quic/masque/masque_epoll_server.h +++ b/quic/masque/masque_epoll_server.h
@@ -6,6 +6,7 @@ #define QUICHE_QUIC_MASQUE_MASQUE_EPOLL_SERVER_H_ #include "quic/masque/masque_server_backend.h" +#include "quic/masque/masque_utils.h" #include "quic/platform/api/quic_export.h" #include "quic/tools/quic_server.h" @@ -14,7 +15,8 @@ // QUIC server that implements MASQUE. class QUIC_NO_EXPORT MasqueEpollServer : public QuicServer { public: - explicit MasqueEpollServer(MasqueServerBackend* masque_server_backend); + explicit MasqueEpollServer(MasqueMode masque_mode, + MasqueServerBackend* masque_server_backend); // Disallow copy and assign. MasqueEpollServer(const MasqueEpollServer&) = delete; @@ -24,6 +26,7 @@ QuicDispatcher* CreateQuicDispatcher() override; private: + MasqueMode masque_mode_; MasqueServerBackend* masque_server_backend_; // Unowned. };
diff --git a/quic/masque/masque_server_backend.cc b/quic/masque/masque_server_backend.cc index 1994819..230f7e7 100644 --- a/quic/masque/masque_server_backend.cc +++ b/quic/masque/masque_server_backend.cc
@@ -19,9 +19,10 @@ } // namespace -MasqueServerBackend::MasqueServerBackend(const std::string& server_authority, +MasqueServerBackend::MasqueServerBackend(MasqueMode masque_mode, + const std::string& server_authority, const std::string& cache_directory) - : server_authority_(server_authority) { + : masque_mode_(masque_mode), server_authority_(server_authority) { if (!cache_directory.empty()) { QuicMemoryCacheBackend::InitializeBackend(cache_directory); } @@ -43,16 +44,24 @@ absl::string_view path = path_pair->second; absl::string_view scheme = scheme_pair->second; absl::string_view method = method_pair->second; - if (scheme != "https" || method != "POST" || request_body.empty()) { - // MASQUE requests MUST be a non-empty https POST. - return false; - } + std::string masque_path = ""; + if (masque_mode_ == MasqueMode::kLegacy) { + if (scheme != "https" || method != "POST" || request_body.empty()) { + // MASQUE requests MUST be a non-empty https POST. + return false; + } - if (path.rfind("/.well-known/masque/", 0) != 0) { - // This request is not a MASQUE path. - return false; + if (path.rfind("/.well-known/masque/", 0) != 0) { + // This request is not a MASQUE path. + return false; + } + masque_path = path.substr(sizeof("/.well-known/masque/") - 1); + } else { + if (method != "CONNECT-UDP") { + // Unexpected method. + return false; + } } - std::string masque_path(path.substr(sizeof("/.well-known/masque/") - 1)); if (!server_authority_.empty()) { auto authority_pair = request_headers.find(":authority");
diff --git a/quic/masque/masque_server_backend.h b/quic/masque/masque_server_backend.h index 1809359..86b1e00 100644 --- a/quic/masque/masque_server_backend.h +++ b/quic/masque/masque_server_backend.h
@@ -6,6 +6,7 @@ #define QUICHE_QUIC_MASQUE_MASQUE_SERVER_BACKEND_H_ #include "absl/container/flat_hash_map.h" +#include "quic/masque/masque_utils.h" #include "quic/platform/api/quic_export.h" #include "quic/tools/quic_memory_cache_backend.h" @@ -27,7 +28,8 @@ virtual ~BackendClient() = default; }; - explicit MasqueServerBackend(const std::string& server_authority, + explicit MasqueServerBackend(MasqueMode masque_mode, + const std::string& server_authority, const std::string& cache_directory); // Disallow copy and assign. @@ -57,6 +59,7 @@ const std::string& request_body, QuicSimpleServerBackend::RequestHandler* request_handler); + MasqueMode masque_mode_; std::string server_authority_; absl::flat_hash_map<std::string, std::unique_ptr<QuicBackendResponse>> active_response_map_;
diff --git a/quic/masque/masque_server_bin.cc b/quic/masque/masque_server_bin.cc index b25bf83..a6b6225 100644 --- a/quic/masque/masque_server_bin.cc +++ b/quic/masque/masque_server_bin.cc
@@ -35,6 +35,12 @@ "Specifies the authority over which the server will accept MASQUE " "requests. Defaults to empty which allows all authorities."); +DEFINE_QUIC_COMMAND_LINE_FLAG(std::string, + masque_mode, + "", + "Allows setting MASQUE mode, valid values are " + "open and legacy. Defaults to open."); + int main(int argc, char* argv[]) { const char* usage = "Usage: masque_server [options]"; std::vector<std::string> non_option_args = @@ -44,17 +50,28 @@ return 0; } - auto backend = std::make_unique<quic::MasqueServerBackend>( - GetQuicFlag(FLAGS_server_authority), GetQuicFlag(FLAGS_cache_dir)); + quic::MasqueMode masque_mode = quic::MasqueMode::kOpen; + std::string mode_string = GetQuicFlag(FLAGS_masque_mode); + if (mode_string == "legacy") { + masque_mode = quic::MasqueMode::kLegacy; + } else if (!mode_string.empty() && mode_string != "open") { + std::cerr << "Invalid masque_mode \"" << mode_string << "\"" << std::endl; + return 1; + } - auto server = std::make_unique<quic::MasqueEpollServer>(backend.get()); + auto backend = std::make_unique<quic::MasqueServerBackend>( + masque_mode, GetQuicFlag(FLAGS_server_authority), + GetQuicFlag(FLAGS_cache_dir)); + + auto server = + std::make_unique<quic::MasqueEpollServer>(masque_mode, backend.get()); if (!server->CreateUDPSocketAndListen(quic::QuicSocketAddress( quic::QuicIpAddress::Any6(), GetQuicFlag(FLAGS_port)))) { return 1; } - std::cerr << "Started MASQUE server" << std::endl; + std::cerr << "Started " << masque_mode << " MASQUE server" << std::endl; server->HandleEventsForever(); return 0; }
diff --git a/quic/masque/masque_server_session.cc b/quic/masque/masque_server_session.cc index 228444a..de231ba 100644 --- a/quic/masque/masque_server_session.cc +++ b/quic/masque/masque_server_session.cc
@@ -4,14 +4,79 @@ #include "quic/masque/masque_server_session.h" +#include <netdb.h> + +#include "absl/strings/str_cat.h" +#include "quic/core/quic_data_reader.h" +#include "quic/core/quic_udp_socket.h" +#include "quic/tools/quic_url.h" +#include "common/platform/api/quiche_text_utils.h" + namespace quic { +namespace { +// RAII wrapper for QuicUdpSocketFd. +class FdWrapper { + public: + // Takes ownership of |fd| and closes the file descriptor on destruction. + explicit FdWrapper(int address_family) { + QuicUdpSocketApi socket_api; + fd_ = + socket_api.Create(address_family, + /*receive_buffer_size =*/kDefaultSocketReceiveBuffer, + /*send_buffer_size =*/kDefaultSocketReceiveBuffer); + } + + ~FdWrapper() { + if (fd_ == kQuicInvalidSocketFd) { + return; + } + QuicUdpSocketApi socket_api; + socket_api.Destroy(fd_); + } + + // Hands ownership of the file descriptor to the caller. + QuicUdpSocketFd extract_fd() { + QuicUdpSocketFd fd = fd_; + fd_ = kQuicInvalidSocketFd; + return fd; + } + + // Keeps ownership of the file descriptor. + QuicUdpSocketFd fd() { return fd_; } + + // Disallow copy and move. + FdWrapper(const FdWrapper&) = delete; + FdWrapper(FdWrapper&&) = delete; + FdWrapper& operator=(const FdWrapper&) = delete; + FdWrapper& operator=(FdWrapper&&) = delete; + + private: + QuicUdpSocketFd fd_; +}; + +std::unique_ptr<QuicBackendResponse> CreateBackendErrorResponse( + absl::string_view status, + absl::string_view body) { + spdy::Http2HeaderBlock response_headers; + response_headers[":status"] = status; + auto response = std::make_unique<QuicBackendResponse>(); + response->set_response_type(QuicBackendResponse::REGULAR_RESPONSE); + response->set_headers(std::move(response_headers)); + response->set_body(body); + return response; +} + +} // namespace + MasqueServerSession::MasqueServerSession( + MasqueMode masque_mode, const QuicConfig& config, const ParsedQuicVersionVector& supported_versions, QuicConnection* connection, QuicSession::Visitor* visitor, Visitor* owner, + QuicEpollServer* epoll_server, QuicCryptoServerStreamBase::Helper* helper, const QuicCryptoServerConfig* crypto_config, QuicCompressedCertsCache* compressed_certs_cache, @@ -26,41 +91,74 @@ masque_server_backend), masque_server_backend_(masque_server_backend), owner_(owner), - compression_engine_(this) { + epoll_server_(epoll_server), + compression_engine_(this), + masque_mode_(masque_mode) { masque_server_backend_->RegisterBackendClient(connection_id(), this); } void MasqueServerSession::OnMessageReceived(absl::string_view message) { QUIC_DVLOG(1) << "Received DATAGRAM frame of length " << message.length(); + if (masque_mode_ == MasqueMode::kLegacy) { + 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; + } - QuicConnectionId client_connection_id, server_connection_id; - QuicSocketAddress server_address; - std::vector<char> packet; - bool version_present; - if (!compression_engine_.DecompressDatagram( - message, &client_connection_id, &server_connection_id, - &server_address, &packet, &version_present)) { + QUIC_DVLOG(1) << "Received packet of length " << packet.size() << " for " + << target_server_address << " client " + << client_connection_id; + + if (version_present) { + if (client_connection_id.length() != kQuicDefaultConnectionIdLength) { + QUIC_DLOG(ERROR) + << "Dropping long header with invalid client_connection_id " + << client_connection_id; + return; + } + owner_->RegisterClientConnectionId(client_connection_id, this); + } + + WriteResult write_result = connection()->writer()->WritePacket( + packet.data(), packet.size(), connection()->self_address().host(), + target_server_address, nullptr); + QUIC_DVLOG(1) << "Got " << write_result << " for " << packet.size() + << " bytes to " << target_server_address; + return; + } + QUICHE_DCHECK_EQ(masque_mode_, MasqueMode::kOpen); + QuicDataReader reader(message); + QuicDatagramFlowId flow_id; + if (!reader.ReadVarInt62(&flow_id)) { + QUIC_DLOG(ERROR) << "Failed to read flow_id"; return; } - QUIC_DVLOG(1) << "Received packet of length " << packet.size() << " for " - << server_address << " client " << client_connection_id; - - if (version_present) { - if (client_connection_id.length() != kQuicDefaultConnectionIdLength) { - QUIC_DLOG(ERROR) - << "Dropping long header with invalid client_connection_id " - << client_connection_id; - return; - } - owner_->RegisterClientConnectionId(client_connection_id, this); + auto it = + absl::c_find_if(connect_udp_server_states_, + [flow_id](const ConnectUdpServerState& connect_udp) { + return connect_udp.flow_id() == flow_id; + }); + if (it == connect_udp_server_states_.end()) { + QUIC_DLOG(ERROR) << "Received unknown flow_id " << flow_id; + return; } - - WriteResult write_result = connection()->writer()->WritePacket( - packet.data(), packet.size(), connection()->self_address().host(), - server_address, nullptr); - QUIC_DVLOG(1) << "Got " << write_result << " for " << packet.size() - << " bytes to " << server_address; + QuicSocketAddress target_server_address = it->target_server_address(); + QUICHE_DCHECK(target_server_address.IsInitialized()); + QuicUdpSocketFd fd = it->fd(); + QUICHE_DCHECK_NE(fd, kQuicInvalidSocketFd); + absl::string_view packet = reader.ReadRemainingPayload(); + QuicUdpSocketApi socket_api; + QuicUdpPacketInfo packet_info; + packet_info.SetPeerAddress(target_server_address); + WriteResult write_result = + socket_api.WritePacket(fd, packet.data(), packet.length(), packet_info); + QUIC_DVLOG(1) << "Wrote packet to server with result " << write_result; } void MasqueServerSession::OnMessageAcked(QuicMessageId message_id, @@ -73,17 +171,123 @@ } void MasqueServerSession::OnConnectionClosed( - const QuicConnectionCloseFrame& /*frame*/, - ConnectionCloseSource /*source*/) { + const QuicConnectionCloseFrame& frame, + ConnectionCloseSource source) { + QuicSimpleServerSession::OnConnectionClosed(frame, source); QUIC_DLOG(INFO) << "Closing connection for " << connection_id(); masque_server_backend_->RemoveBackendClient(connection_id()); + // Clearing this state will close all sockets. + connect_udp_server_states_.clear(); +} + +void MasqueServerSession::OnStreamClosed(QuicStreamId stream_id) { + connect_udp_server_states_.remove_if( + [stream_id](const ConnectUdpServerState& connect_udp) { + return connect_udp.stream_id() == stream_id; + }); + + QuicSimpleServerSession::OnStreamClosed(stream_id); } std::unique_ptr<QuicBackendResponse> MasqueServerSession::HandleMasqueRequest( const std::string& masque_path, - const spdy::Http2HeaderBlock& /*request_headers*/, + const spdy::Http2HeaderBlock& request_headers, const std::string& request_body, - QuicSimpleServerBackend::RequestHandler* /*request_handler*/) { + QuicSimpleServerBackend::RequestHandler* request_handler) { + if (masque_mode_ != MasqueMode::kLegacy) { + auto path_pair = request_headers.find(":path"); + auto scheme_pair = request_headers.find(":scheme"); + auto method_pair = request_headers.find(":method"); + auto flow_id_pair = request_headers.find("datagram-flow-id"); + auto authority_pair = request_headers.find(":authority"); + if (path_pair == request_headers.end() || + scheme_pair == request_headers.end() || + method_pair == request_headers.end() || + flow_id_pair == request_headers.end() || + authority_pair == request_headers.end()) { + QUIC_DLOG(ERROR) << "MASQUE request is missing required headers"; + return CreateBackendErrorResponse("400", "Missing required headers"); + } + absl::string_view path = path_pair->second; + absl::string_view scheme = scheme_pair->second; + absl::string_view method = method_pair->second; + absl::string_view flow_id_str = flow_id_pair->second; + absl::string_view authority = authority_pair->second; + if (path.empty()) { + QUIC_DLOG(ERROR) << "MASQUE request with empty path"; + return CreateBackendErrorResponse("400", "Empty path"); + } + if (scheme.empty()) { + return CreateBackendErrorResponse("400", "Empty scheme"); + return nullptr; + } + if (method != "CONNECT-UDP") { + QUIC_DLOG(ERROR) << "MASQUE request with bad method \"" << method << "\""; + return CreateBackendErrorResponse("400", "Bad method"); + } + QuicDatagramFlowId flow_id; + if (!absl::SimpleAtoi(flow_id_str, &flow_id)) { + QUIC_DLOG(ERROR) << "MASQUE request with bad flow_id \"" << flow_id_str + << "\""; + return CreateBackendErrorResponse("400", "Bad flow ID"); + } + QuicUrl url(absl::StrCat("https://", authority)); + if (!url.IsValid() || url.PathParamsQuery() != "/") { + QUIC_DLOG(ERROR) << "MASQUE request with bad authority \"" << authority + << "\""; + return CreateBackendErrorResponse("400", "Bad authority"); + } + + std::string port = absl::StrCat(url.port()); + addrinfo hint = {}; + hint.ai_protocol = IPPROTO_UDP; + + addrinfo* info_list = nullptr; + int result = + getaddrinfo(url.host().c_str(), port.c_str(), &hint, &info_list); + if (result != 0) { + QUIC_DLOG(ERROR) << "Failed to resolve " << authority << ": " + << gai_strerror(result); + return CreateBackendErrorResponse("500", "DNS resolution failed"); + } + + QUICHE_CHECK_NE(info_list, nullptr); + std::unique_ptr<addrinfo, void (*)(addrinfo*)> info_list_owned( + info_list, freeaddrinfo); + QuicSocketAddress target_server_address(info_list->ai_addr, + info_list->ai_addrlen); + QUIC_DLOG(INFO) << "Got CONNECT_UDP request flow_id=" << flow_id + << " target_server_address=\"" << target_server_address + << "\""; + + FdWrapper fd_wrapper(target_server_address.host().AddressFamilyToInt()); + if (fd_wrapper.fd() == kQuicInvalidSocketFd) { + QUIC_DLOG(ERROR) << "Socket creation failed"; + return CreateBackendErrorResponse("500", "Socket creation failed"); + } + QuicSocketAddress any_v6_address(QuicIpAddress::Any6(), 0); + QuicUdpSocketApi socket_api; + if (!socket_api.Bind(fd_wrapper.fd(), any_v6_address)) { + QUIC_DLOG(ERROR) << "Socket bind failed"; + return CreateBackendErrorResponse("500", "Socket bind failed"); + } + epoll_server_->RegisterFDForRead(fd_wrapper.fd(), this); + + connect_udp_server_states_.emplace_back(ConnectUdpServerState( + flow_id, request_handler->stream_id(), target_server_address, + fd_wrapper.extract_fd(), epoll_server_)); + + spdy::Http2HeaderBlock response_headers; + response_headers[":status"] = "200"; + response_headers["datagram-flow-id"] = absl::StrCat(flow_id); + auto response = std::make_unique<QuicBackendResponse>(); + response->set_response_type(QuicBackendResponse::INCOMPLETE_RESPONSE); + response->set_headers(std::move(response_headers)); + response->set_body(""); + + return response; + } + QUIC_DLOG(INFO) << "MasqueServerSession handling MASQUE request"; if (masque_path == "init") { @@ -120,9 +324,166 @@ void MasqueServerSession::HandlePacketFromServer( const ReceivedPacketInfo& packet_info) { QUIC_DVLOG(1) << "MasqueServerSession received " << packet_info; - compression_engine_.CompressAndSendPacket( - packet_info.packet.AsStringPiece(), packet_info.destination_connection_id, - packet_info.source_connection_id, packet_info.peer_address); + if (masque_mode_ == MasqueMode::kLegacy) { + compression_engine_.CompressAndSendPacket( + packet_info.packet.AsStringPiece(), + packet_info.destination_connection_id, packet_info.source_connection_id, + packet_info.peer_address); + return; + } + QUIC_LOG(ERROR) << "Ignoring packet from server in " << masque_mode_ + << " mode"; +} + +void MasqueServerSession::OnRegistration(QuicEpollServer* /*eps*/, + QuicUdpSocketFd fd, + int event_mask) { + QUIC_DVLOG(1) << "OnRegistration " << fd << " event_mask " << event_mask; +} + +void MasqueServerSession::OnModification(QuicUdpSocketFd fd, int event_mask) { + QUIC_DVLOG(1) << "OnModification " << fd << " event_mask " << event_mask; +} + +void MasqueServerSession::OnEvent(QuicUdpSocketFd fd, QuicEpollEvent* event) { + if ((event->in_events & EPOLLIN) == 0) { + QUIC_DVLOG(1) << "Ignoring OnEvent fd " << fd << " event mask " + << event->in_events; + return; + } + auto it = absl::c_find_if(connect_udp_server_states_, + [fd](const ConnectUdpServerState& connect_udp) { + return connect_udp.fd() == fd; + }); + if (it == connect_udp_server_states_.end()) { + QUIC_BUG << "Got unexpected event mask " << event->in_events + << " on unknown fd " << fd; + return; + } + QuicDatagramFlowId flow_id = it->flow_id(); + QuicSocketAddress expected_target_server_address = + it->target_server_address(); + QUICHE_DCHECK(expected_target_server_address.IsInitialized()); + QUIC_DVLOG(1) << "Received readable event on fd " << fd << " (mask " + << event->in_events << ") flow_id " << flow_id << " server " + << expected_target_server_address; + QuicUdpSocketApi socket_api; + BitMask64 packet_info_interested(QuicUdpPacketInfoBit::PEER_ADDRESS); + char packet_buffer[kMaxIncomingPacketSize]; + char control_buffer[kDefaultUdpPacketControlBufferSize]; + while (true) { + QuicUdpSocketApi::ReadPacketResult read_result; + read_result.packet_buffer = {packet_buffer, sizeof(packet_buffer)}; + read_result.control_buffer = {control_buffer, sizeof(control_buffer)}; + socket_api.ReadPacket(fd, packet_info_interested, &read_result); + if (!read_result.ok) { + // Most likely there is nothing left to read, break out of read loop. + break; + } + if (!read_result.packet_info.HasValue(QuicUdpPacketInfoBit::PEER_ADDRESS)) { + QUIC_BUG << "Missing peer address when reading from fd " << fd; + continue; + } + if (read_result.packet_info.peer_address() != + expected_target_server_address) { + QUIC_DLOG(ERROR) << "Ignoring UDP packet on fd " << fd + << " from unexpected server address " + << read_result.packet_info.peer_address() + << " (expected " << expected_target_server_address + << ")"; + continue; + } + if (!connection()->connected()) { + QUIC_BUG << "Unexpected incoming UDP packet on fd " << fd << " from " + << expected_target_server_address + << " because MASQUE connection is closed"; + return; + } + // The packet is valid, send it to the client in a DATAGRAM frame. + size_t slice_length = QuicDataWriter::GetVarInt62Len(flow_id) + + read_result.packet_buffer.buffer_len; + QuicUniqueBufferPtr buffer = MakeUniqueBuffer( + connection()->helper()->GetStreamSendBufferAllocator(), slice_length); + QuicDataWriter writer(slice_length, buffer.get()); + if (!writer.WriteVarInt62(flow_id)) { + QUIC_BUG << "Failed to write flow_id"; + continue; + } + if (!writer.WriteBytes(read_result.packet_buffer.buffer, + read_result.packet_buffer.buffer_len)) { + QUIC_BUG << "Failed to write packet"; + continue; + } + QUICHE_DCHECK_EQ(writer.remaining(), 0u); + QuicMemSlice slice(std::move(buffer), slice_length); + MessageResult message_result = SendMessage(QuicMemSliceSpan(&slice)); + QUIC_DVLOG(1) << "Sent UDP packet from target server of length " + << read_result.packet_buffer.buffer_len << " with flow ID " + << flow_id << " and got message result " << message_result; + } +} + +void MasqueServerSession::OnUnregistration(QuicUdpSocketFd fd, bool replaced) { + QUIC_DVLOG(1) << "OnUnregistration " << fd << " " << (replaced ? "" : "!") + << " replaced"; +} + +void MasqueServerSession::OnShutdown(QuicEpollServer* /*eps*/, + QuicUdpSocketFd fd) { + QUIC_DVLOG(1) << "OnShutdown " << fd; +} + +std::string MasqueServerSession::Name() const { + return std::string("MasqueServerSession-") + connection_id().ToString(); +} + +MasqueServerSession::ConnectUdpServerState::ConnectUdpServerState( + QuicDatagramFlowId flow_id, + QuicStreamId stream_id, + const QuicSocketAddress& target_server_address, + QuicUdpSocketFd fd, + QuicEpollServer* epoll_server) + : flow_id_(flow_id), + stream_id_(stream_id), + target_server_address_(target_server_address), + fd_(fd), + epoll_server_(epoll_server) { + QUICHE_DCHECK_NE(fd_, kQuicInvalidSocketFd); + QUICHE_DCHECK_NE(epoll_server_, nullptr); +} + +MasqueServerSession::ConnectUdpServerState::~ConnectUdpServerState() { + if (fd_ == kQuicInvalidSocketFd) { + return; + } + QuicUdpSocketApi socket_api; + QUIC_DLOG(INFO) << "Closing fd " << fd_; + epoll_server_->UnregisterFD(fd_); + socket_api.Destroy(fd_); +} + +MasqueServerSession::ConnectUdpServerState::ConnectUdpServerState( + MasqueServerSession::ConnectUdpServerState&& other) { + fd_ = kQuicInvalidSocketFd; + *this = std::move(other); +} + +MasqueServerSession::ConnectUdpServerState& +MasqueServerSession::ConnectUdpServerState::operator=( + MasqueServerSession::ConnectUdpServerState&& other) { + if (fd_ != kQuicInvalidSocketFd) { + QuicUdpSocketApi socket_api; + QUIC_DLOG(INFO) << "Closing fd " << fd_; + epoll_server_->UnregisterFD(fd_); + socket_api.Destroy(fd_); + } + flow_id_ = other.flow_id_; + stream_id_ = other.stream_id_; + target_server_address_ = other.target_server_address_; + fd_ = other.fd_; + epoll_server_ = other.epoll_server_; + other.fd_ = kQuicInvalidSocketFd; + return *this; } } // namespace quic
diff --git a/quic/masque/masque_server_session.h b/quic/masque/masque_server_session.h index 9e43b63..a73e641 100644 --- a/quic/masque/masque_server_session.h +++ b/quic/masque/masque_server_session.h
@@ -5,8 +5,12 @@ #ifndef QUICHE_QUIC_MASQUE_MASQUE_SERVER_SESSION_H_ #define QUICHE_QUIC_MASQUE_MASQUE_SERVER_SESSION_H_ +#include "quic/core/quic_types.h" +#include "quic/core/quic_udp_socket.h" #include "quic/masque/masque_compression_engine.h" #include "quic/masque/masque_server_backend.h" +#include "quic/masque/masque_utils.h" +#include "quic/platform/api/quic_epoll.h" #include "quic/platform/api/quic_export.h" #include "quic/tools/quic_simple_server_session.h" @@ -15,7 +19,8 @@ // QUIC server session for connection to MASQUE proxy. class QUIC_NO_EXPORT MasqueServerSession : public QuicSimpleServerSession, - public MasqueServerBackend::BackendClient { + public MasqueServerBackend::BackendClient, + public QuicEpollCallbackInterface { public: // Interface meant to be implemented by owner of this MasqueServerSession // instance. @@ -33,11 +38,13 @@ }; explicit MasqueServerSession( + MasqueMode masque_mode, const QuicConfig& config, const ParsedQuicVersionVector& supported_versions, QuicConnection* connection, QuicSession::Visitor* visitor, Visitor* owner, + QuicEpollServer* epoll_server, QuicCryptoServerStreamBase::Helper* helper, const QuicCryptoServerConfig* crypto_config, QuicCompressedCertsCache* compressed_certs_cache, @@ -54,6 +61,7 @@ void OnMessageLost(QuicMessageId message_id) override; void OnConnectionClosed(const QuicConnectionCloseFrame& frame, ConnectionCloseSource source) override; + void OnStreamClosed(QuicStreamId stream_id) override; // From MasqueServerBackend::BackendClient. std::unique_ptr<QuicBackendResponse> HandleMasqueRequest( @@ -62,13 +70,61 @@ const std::string& request_body, QuicSimpleServerBackend::RequestHandler* request_handler) override; + // From QuicEpollCallbackInterface. + void OnRegistration(QuicEpollServer* eps, + QuicUdpSocketFd fd, + int event_mask) override; + void OnModification(QuicUdpSocketFd fd, int event_mask) override; + void OnEvent(QuicUdpSocketFd fd, QuicEpollEvent* event) override; + void OnUnregistration(QuicUdpSocketFd fd, bool replaced) override; + void OnShutdown(QuicEpollServer* eps, QuicUdpSocketFd fd) override; + std::string Name() const override; + // Handle packet for client, meant to be called by MasqueDispatcher. void HandlePacketFromServer(const ReceivedPacketInfo& packet_info); private: + // State that the MasqueServerSession keeps for each CONNECT-UDP request. + class QUIC_NO_EXPORT ConnectUdpServerState { + public: + // ConnectUdpServerState takes ownership of |fd|. It will unregister it + // from |epoll_server| and close the file descriptor when destructed. + explicit ConnectUdpServerState( + QuicDatagramFlowId flow_id, + QuicStreamId stream_id, + const QuicSocketAddress& target_server_address, + QuicUdpSocketFd fd, + QuicEpollServer* epoll_server); + + ~ConnectUdpServerState(); + + // Disallow copy but allow move. + ConnectUdpServerState(const ConnectUdpServerState&) = delete; + ConnectUdpServerState(ConnectUdpServerState&&); + ConnectUdpServerState& operator=(const ConnectUdpServerState&) = delete; + ConnectUdpServerState& operator=(ConnectUdpServerState&&); + + QuicDatagramFlowId flow_id() const { return flow_id_; } + QuicStreamId stream_id() const { return stream_id_; } + const QuicSocketAddress& target_server_address() const { + return target_server_address_; + } + QuicUdpSocketFd fd() const { return fd_; } + + private: + QuicDatagramFlowId flow_id_; + QuicStreamId stream_id_; + QuicSocketAddress target_server_address_; + QuicUdpSocketFd fd_; // Owned. + QuicEpollServer* epoll_server_; // Unowned. + }; + MasqueServerBackend* masque_server_backend_; // Unowned. Visitor* owner_; // Unowned. + QuicEpollServer* epoll_server_; // Unowned. MasqueCompressionEngine compression_engine_; + MasqueMode masque_mode_; + std::list<ConnectUdpServerState> connect_udp_server_states_; bool masque_initialized_ = false; };
diff --git a/quic/masque/masque_utils.cc b/quic/masque/masque_utils.cc index f4f77f5..17dec0d 100644 --- a/quic/masque/masque_utils.cc +++ b/quic/masque/masque_utils.cc
@@ -27,4 +27,21 @@ return config; } +std::string MasqueModeToString(MasqueMode masque_mode) { + switch (masque_mode) { + case MasqueMode::kInvalid: + return "Invalid"; + case MasqueMode::kLegacy: + return "Legacy"; + case MasqueMode::kOpen: + return "Open"; + } + return absl::StrCat("Unknown(", static_cast<int>(masque_mode), ")"); +} + +std::ostream& operator<<(std::ostream& os, const MasqueMode& masque_mode) { + os << MasqueModeToString(masque_mode); + return os; +} + } // namespace quic
diff --git a/quic/masque/masque_utils.h b/quic/masque/masque_utils.h index f151f55..8113047 100644 --- a/quic/masque/masque_utils.h +++ b/quic/masque/masque_utils.h
@@ -18,7 +18,24 @@ QUIC_NO_EXPORT QuicConfig MasqueEncapsulatedConfig(); // Maximum packet size for encapsulated connections. -const QuicByteCount kMasqueMaxEncapsulatedPacketSize = 1300; +enum : QuicByteCount { kMasqueMaxEncapsulatedPacketSize = 1300 }; + +// Mode that MASQUE is operating in. +enum class MasqueMode : uint8_t { + kInvalid = 0, // Should never be used. + kLegacy = 1, // Legacy mode uses the legacy MASQUE protocol as documented in + // <https://tools.ietf.org/html/draft-schinazi-masque-protocol>. That version + // of MASQUE uses a custom application-protocol over HTTP/3, and also allows + // unauthenticated clients. + kOpen = 2, // Open mode uses the MASQUE HTTP CONNECT-UDP method as documented + // in <https://tools.ietf.org/html/draft-ietf-masque-connect-udp>. This mode + // allows unauthenticated clients (a more restricted mode will be added to + // this enum at a later date). +}; + +QUIC_NO_EXPORT std::string MasqueModeToString(MasqueMode masque_mode); +QUIC_NO_EXPORT std::ostream& operator<<(std::ostream& os, + const MasqueMode& masque_mode); } // namespace quic