diff --git a/CMakeLists.txt b/CMakeLists.txt index 32a9bc6df8c..28056125435 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -285,6 +285,7 @@ include(CheckOpenSSLIsQuictls) include(CheckOpenSSLIsAwsLc) include(CheckOpenSSLHasQuicTlsCbs) include(CheckOpenSSLHasNativeQuic) +include(CheckQuicheHasQmux) find_package(OpenSSL REQUIRED) check_openssl_is_boringssl(SSLLIB_IS_BORINGSSL BORINGSSL_VERSION "${OPENSSL_INCLUDE_DIR}") check_openssl_is_awslc(SSLLIB_IS_AWSLC AWSLC_VERSION "${OPENSSL_INCLUDE_DIR}") @@ -329,41 +330,19 @@ endif() check_openssl_has_native_quic(SSLLIB_HAS_NATIVE_QUIC "${OPENSSL_INCLUDE_DIR}") -if(DEFINED ENABLE_OPENSSL_QUIC - AND NOT ENABLE_OPENSSL_QUIC STREQUAL "AUTO" - AND ENABLE_OPENSSL_QUIC -) - if(ENABLE_QUICHE) - message(FATAL_ERROR "ENABLE_OPENSSL_QUIC and ENABLE_QUICHE are mutually exclusive QUIC backends") - endif() - if(NOT SSLLIB_HAS_NATIVE_QUIC) - message(FATAL_ERROR "OpenSSL native QUIC support requires OpenSSL 3.5 or newer with OSSL_QUIC_server_method") - endif() - if(SSLLIB_IS_BORINGSSL - OR SSLLIB_IS_AWSLC - OR SSLLIB_IS_QUICTLS - ) - message(FATAL_ERROR "OpenSSL native QUIC support requires upstream OpenSSL 3.5 or newer") - endif() -endif() - -set(OPENSSL_QUIC_AVAILABLE ${SSLLIB_HAS_NATIVE_QUIC}) -if(SSLLIB_IS_BORINGSSL - OR SSLLIB_IS_AWSLC - OR SSLLIB_IS_QUICTLS - OR ENABLE_QUICHE +# OpenSSL's native QUIC is a capability of the mandatory OpenSSL dependency, not an optional +# component to opt into -- same footing as SSLLIB_IS_BORINGSSL or SSLLIB_HAS_QUIC_TLS_CBS above. +# It's mutually exclusive with quiche by construction: quiche requires BoringSSL or the TLS +# callback compat shim, neither of which implements this upstream-OpenSSL-3.5+ API. +set(TS_HAS_OPENSSL_QUIC FALSE) +if(SSLLIB_HAS_NATIVE_QUIC + AND NOT SSLLIB_IS_BORINGSSL + AND NOT SSLLIB_IS_AWSLC + AND NOT SSLLIB_IS_QUICTLS + AND NOT ENABLE_QUICHE ) - set(OPENSSL_QUIC_AVAILABLE FALSE) + set(TS_HAS_OPENSSL_QUIC TRUE) endif() -auto_option( - OPENSSL_QUIC - FEATURE_VAR - TS_HAS_OPENSSL_QUIC - DESCRIPTION - "Use OpenSSL native QUIC" - VAR_DEPENDS - OPENSSL_QUIC_AVAILABLE -) if(ENABLE_PROFILER) find_package(profiler REQUIRED) @@ -380,11 +359,6 @@ elseif(TS_HAS_MIMALLOC) link_libraries(mimalloc) endif() -if(TS_HAS_OPENSSL_QUIC) - set(TS_USE_QUIC TRUE) - message(STATUS "Using OpenSSL native QUIC") -endif() - if(ENABLE_QUICHE) if(TS_OPENSSL_QUIC_TLS_CBS_COMPAT) set(quiche_USE_STATIC TRUE) @@ -392,7 +366,6 @@ if(ENABLE_QUICHE) find_package(quiche REQUIRED) set(TS_HAS_QUICHE ${quiche_FOUND}) - set(TS_USE_QUIC ${TS_HAS_QUICHE}) if(NOT SSLLIB_IS_BORINGSSL AND NOT SSLLIB_IS_QUICTLS AND NOT TS_OPENSSL_QUIC_TLS_CBS_COMPAT @@ -411,6 +384,53 @@ if(ENABLE_QUICHE) elseif(TS_OPENSSL_QUIC_TLS_CBS_COMPAT) message(STATUS "Using OpenSSL QUIC TLS callbacks compatibility for quiche") endif() + + check_quiche_has_qmux(TS_QUICHE_HAS_QMUX) +endif() + +# A QUIC backend supplies the protocol implementation; the transport options +# below decide how QUIC streams are actually carried on the wire. At least one +# transport must be enabled for QUIC to be reachable. +if(TS_HAS_OPENSSL_QUIC OR TS_HAS_QUICHE) + set(TS_HAS_QUIC_BACKEND TRUE) +else() + set(TS_HAS_QUIC_BACKEND FALSE) +endif() + +auto_option( + QUIC + DESCRIPTION + "Carry QUIC over UDP (default AUTO: on when a QUIC backend is available)" + FEATURE_VAR + TS_USE_QUIC + VAR_DEPENDS + TS_HAS_QUIC_BACKEND +) + +if(TS_USE_QUIC) + if(TS_HAS_OPENSSL_QUIC) + message(STATUS "Using OpenSSL native QUIC") + elseif(TS_HAS_QUICHE) + message(STATUS "Using quiche for QUIC transport") + endif() +endif() + +auto_option( + QMUX + DESCRIPTION + "Carry QUIC over TLS/TCP as QMux (default AUTO: on when quiche has qmux support)" + FEATURE_VAR + TS_USE_QMUX + VAR_DEPENDS + TS_HAS_QUICHE + TS_QUICHE_HAS_QMUX +) + +if(TS_HAS_QUIC_BACKEND + AND NOT TS_USE_QUIC + AND NOT TS_USE_QMUX +) + message(WARNING "A QUIC backend is enabled but neither UDP QUIC nor QMux is, so QUIC will not be served.") endif() find_package(maxminddb) # Header_rewrite experimental/maxmind_acl diff --git a/cmake/CheckQuicheHasQmux.cmake b/cmake/CheckQuicheHasQmux.cmake new file mode 100644 index 00000000000..cef8744c3cf --- /dev/null +++ b/cmake/CheckQuicheHasQmux.cmake @@ -0,0 +1,41 @@ +####################### +# +# Licensed to the Apache Software Foundation (ASF) under one or more contributor license +# agreements. See the NOTICE file distributed with this work for additional information regarding +# copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software distributed under the License +# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing permissions and limitations under +# the License. +# +####################### + +# quiche.h always declares quiche_config_enable_qmux(), regardless of whether quiche was built +# with its `qmux` Rust feature -- only the compiled library conditionally exports the symbol. So +# this must be a full compile-and-link check against the actual quiche library, not a header-only +# check, or it would report qmux support as available even when the linked quiche lacks it. +function(CHECK_QUICHE_HAS_QMUX OUT_VAR) + set(CHECK_PROGRAM + " + #include + + int main() { + quiche_config *config = quiche_config_new(QUICHE_PROTOCOL_VERSION); + quiche_config_enable_qmux(config, true); + return 0; + } + " + ) + set(CMAKE_REQUIRED_LIBRARIES quiche::quiche) + include(CheckCXXSourceCompiles) + check_cxx_source_compiles("${CHECK_PROGRAM}" ${OUT_VAR}) + set(${OUT_VAR} + ${${OUT_VAR}} + PARENT_SCOPE + ) +endfunction() diff --git a/include/iocore/net/qmux/QMuxConnection.h b/include/iocore/net/qmux/QMuxConnection.h new file mode 100644 index 00000000000..d1940253729 --- /dev/null +++ b/include/iocore/net/qmux/QMuxConnection.h @@ -0,0 +1,132 @@ +/** @file + + QMux connection wrapping quiche_conn (draft-opik-quic-qmux-01) + + @section license License + + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + */ + +#pragma once + +#include "iocore/net/quic/QUICConnection.h" +#include "iocore/net/quic/QUICStream.h" +#include "iocore/eventsystem/Continuation.h" +#include "tscore/ink_inet.h" + +#include +#include +#include +#include + +struct quiche_conn; +struct quiche_config; + +class Event; +class NetVConnection; +class VIO; +class MIOBuffer; +class IOBufferReader; +class QUICContext; +class QUICApplicationMap; +class QUICStreamManager; + +/** + * QMux connection implementing QUICConnection interface. + * Wraps a quiche_conn* created with QMux-enabled config. + * Also acts as the I/O event handler (Continuation) that bridges + * the SSLNetVConnection byte stream to quiche framing, and as the + * QUICStreamIO backend that QUICStream uses to move stream data. + */ +class QMuxConnection : public QUICConnection, public Continuation, public QUICStreamIO +{ +public: + explicit QMuxConnection(NetVConnection *netvc); + ~QMuxConnection() override; + + // QUICConnectionInfoProvider + QUICConnectionId peer_connection_id() const override; + QUICConnectionId original_connection_id() const override; + QUICConnectionId first_connection_id() const override; + QUICConnectionId retry_source_connection_id() const override; + QUICConnectionId initial_source_connection_id() const override; + QUICConnectionId connection_id() const override; + std::string_view cids() const override; + const QUICFiveTuple five_tuple() const override; + uint32_t pmtu() const override; + NetVConnectionContext_t direction() const override; + bool is_closed() const override; + bool is_at_anti_amplification_limit() const override; + bool is_address_validation_completed() const override; + bool is_handshake_completed() const override; + QUICVersion negotiated_version() const override; + std::string_view negotiated_application_name() const override; + void on_stream_updated() override; + + // QUICStreamIO + int64_t read_stream(QUICStreamId stream_id, uint8_t *buf, size_t len, bool &fin, QUICStreamIO::ErrorCode &error_code) override; + bool stream_read_finished(QUICStreamId stream_id) override; + int64_t stream_write_capacity(QUICStreamId stream_id) override; + int64_t write_stream(QUICStreamId stream_id, uint8_t const *buf, size_t len, bool fin, + QUICStreamIO::ErrorCode &error_code) override; + + // QUICConnection + QUICStreamManager *stream_manager() override; + void close_quic_connection(QUICConnectionErrorUPtr error) override; + void reset_quic_connection() override; + void handle_received_packet(UDPPacket *packet) override; + void ping() override; + + void start(NetVConnection *netvc); + void signal_write_ready(); + +private: + int main_event(int event, void *data); + void _handle_read(); + void _handle_write(); + void _flush_quiche_output(); + void _handle_read_streams(); + void _handle_write_streams(); + void _schedule_quiche_timeout(); + void _unschedule_quiche_timeout(); + + static quiche_config *_shared_config; + static void _init_shared_config(); + + quiche_conn *_quiche_con = nullptr; + + sockaddr_storage _local_addr = {}; + socklen_t _local_addr_len = 0; + sockaddr_storage _peer_addr = {}; + socklen_t _peer_addr_len = 0; + + std::unique_ptr _app_map; + std::unique_ptr _context; + std::unique_ptr _stream_manager; + + QUICConnectionId _synthetic_cid; + std::string _cids_str; + + bool _closed = false; + bool _in_write = false; + + MIOBuffer *_read_buf = nullptr; + IOBufferReader *_read_reader = nullptr; + MIOBuffer *_write_buf = nullptr; + VIO *_write_vio = nullptr; + Event *_quiche_timeout = nullptr; +}; diff --git a/include/iocore/net/quic/QUICStream.h b/include/iocore/net/quic/QUICStream.h index e57c49b6dc5..e287daa5c90 100644 --- a/include/iocore/net/quic/QUICStream.h +++ b/include/iocore/net/quic/QUICStream.h @@ -61,7 +61,7 @@ class QUICStream QUICStream() {} QUICStream(QUICConnectionInfoProvider *cinfo, QUICStreamId sid); - ~QUICStream(); + virtual ~QUICStream(); QUICStreamId id() const; const QUICConnectionInfoProvider *connection_info(); diff --git a/include/ts/apidefs.h.in b/include/ts/apidefs.h.in index fc2403892bc..f34ffdb1a3f 100644 --- a/include/ts/apidefs.h.in +++ b/include/ts/apidefs.h.in @@ -1512,6 +1512,7 @@ extern const char *const TS_ALPN_PROTOCOL_HTTP_3; extern const char *const TS_ALPN_PROTOCOL_HTTP_3_D29; extern const char *const TS_ALPN_PROTOCOL_HTTP_QUIC; extern const char *const TS_ALPN_PROTOCOL_HTTP_QUIC_D29; +extern const char *const TS_ALPN_PROTOCOL_H3QX; extern int TS_ALPN_PROTOCOL_INDEX_HTTP_0_9; extern int TS_ALPN_PROTOCOL_INDEX_HTTP_1_0; @@ -1519,6 +1520,7 @@ extern int TS_ALPN_PROTOCOL_INDEX_HTTP_1_1; extern int TS_ALPN_PROTOCOL_INDEX_HTTP_2_0; extern int TS_ALPN_PROTOCOL_INDEX_HTTP_3; extern int TS_ALPN_PROTOCOL_INDEX_HTTP_QUIC; +extern int TS_ALPN_PROTOCOL_INDEX_H3QX; extern const char *const TS_ALPN_PROTOCOL_GROUP_HTTP; extern const char *const TS_ALPN_PROTOCOL_GROUP_HTTP2; @@ -1528,6 +1530,7 @@ extern const char *const TS_PROTO_TAG_HTTP_1_1; extern const char *const TS_PROTO_TAG_HTTP_2_0; extern const char *const TS_PROTO_TAG_HTTP_3; extern const char *const TS_PROTO_TAG_HTTP_QUIC; +extern const char *const TS_PROTO_TAG_H3QX; extern const char *const TS_PROTO_TAG_TLS_1_3; extern const char *const TS_PROTO_TAG_TLS_1_2; extern const char *const TS_PROTO_TAG_TLS_1_1; diff --git a/include/tscore/ink_config.h.cmake.in b/include/tscore/ink_config.h.cmake.in index 3898e3e7dc1..c914eae89d3 100644 --- a/include/tscore/ink_config.h.cmake.in +++ b/include/tscore/ink_config.h.cmake.in @@ -163,6 +163,7 @@ const int DEFAULT_STACKSIZE = @DEFAULT_STACK_SIZE@; #cmakedefine01 TS_USE_ALLOCATOR_METRICS #cmakedefine01 TS_USE_POSIX_CAP #cmakedefine01 TS_USE_QUIC +#cmakedefine01 TS_USE_QMUX #cmakedefine01 TS_USE_REMOTE_UNWINDING #cmakedefine01 TS_USE_TLS13 #cmakedefine01 TS_USE_TLS_ASYNC diff --git a/include/tscore/ink_inet.h b/include/tscore/ink_inet.h index d0cc2433644..efdad4b5662 100644 --- a/include/tscore/ink_inet.h +++ b/include/tscore/ink_inet.h @@ -78,6 +78,7 @@ extern const std::string_view IP_PROTO_TAG_HTTP_QUIC; extern const std::string_view IP_PROTO_TAG_HTTP_3; extern const std::string_view IP_PROTO_TAG_HTTP_QUIC_D29; extern const std::string_view IP_PROTO_TAG_HTTP_3_D29; +extern const std::string_view IP_PROTO_TAG_H3QX; struct IpAddr; // forward declare. struct UnAddr; // forward declare. diff --git a/src/iocore/net/CMakeLists.txt b/src/iocore/net/CMakeLists.txt index c76322b77c9..bd1abce1b6c 100644 --- a/src/iocore/net/CMakeLists.txt +++ b/src/iocore/net/CMakeLists.txt @@ -77,12 +77,15 @@ add_library( ) add_library(ts::inknet ALIAS inknet) -if(TS_USE_QUIC) +if(TS_USE_QUIC OR TS_USE_QMUX) add_subdirectory(quic) - target_sources( - inknet PRIVATE QUICClosedConCollector.cc QUICMultiCertConfigLoader.cc QUICNextProtocolAccept.cc QUICSupport.cc - ) + target_sources(inknet PRIVATE QUICSupport.cc) + target_link_libraries(inknet PUBLIC ts::quic) +endif() + +if(TS_USE_QUIC) + target_sources(inknet PRIVATE QUICClosedConCollector.cc QUICMultiCertConfigLoader.cc QUICNextProtocolAccept.cc) if(TS_HAS_OPENSSL_QUIC) target_sources(inknet PRIVATE OpenSSLQUICNetProcessor.cc OpenSSLQUICNetVConnection.cc OpenSSLQUICPacketHandler.cc) @@ -90,8 +93,11 @@ if(TS_USE_QUIC) target_sources(inknet PRIVATE QUICNet.cc QUICNetProcessor.cc QUICNetVConnection.cc QUICPacketHandler.cc) target_link_libraries(inknet PUBLIC quiche::quiche) endif() +endif() - target_link_libraries(inknet PUBLIC ts::quic) +if(TS_USE_QMUX) + add_subdirectory(qmux) + target_link_libraries(inknet PUBLIC quiche::quiche ts::qmux) endif() if(BUILD_REGRESSION_TESTING OR BUILD_TESTING) @@ -165,12 +171,15 @@ if(BUILD_TESTING) ts::http ts::http_remap ) - if(TS_USE_QUIC) + if(TS_USE_QUIC OR TS_USE_QMUX) list(APPEND LINK_GROUP_LIBS quic http3) if(TS_HAS_QUICHE) list(APPEND LINK_GROUP_LIBS quiche::quiche) endif() endif() + if(TS_USE_QMUX) + list(APPEND LINK_GROUP_LIBS qmux) + endif() if(CMAKE_LINK_GROUP_USING_RESCAN_SUPPORTED OR CMAKE_CXX_LINK_GROUP_USING_RESCAN_SUPPORTED) string(JOIN "," LINK_GROUP_LIBS_CSV ${LINK_GROUP_LIBS}) target_link_libraries( diff --git a/src/iocore/net/P_SSLNetVConnection.h b/src/iocore/net/P_SSLNetVConnection.h index 161748796d1..41fe9a5cdaf 100644 --- a/src/iocore/net/P_SSLNetVConnection.h +++ b/src/iocore/net/P_SSLNetVConnection.h @@ -45,6 +45,13 @@ #include "P_SSLUtils.h" #include "P_SSLConfig.h" +#include "tscore/ink_config.h" + +#if TS_USE_QMUX +#include "iocore/net/QUICSupport.h" +#include "iocore/net/qmux/QMuxConnection.h" +#endif + #include #include #include @@ -103,6 +110,10 @@ class SSLNetVConnection : public UnixNetVConnection, public TLSCertSwitchSupport, public TLSEventSupport, public TLSBasicSupport +#if TS_USE_QMUX + , + public QUICSupport +#endif { using super = UnixNetVConnection; ///< Parent type. @@ -417,6 +428,17 @@ class SSLNetVConnection : public UnixNetVConnection, IOBufferReader *_early_data_reader = nullptr; #endif +#if TS_USE_QMUX + // QUICSupport + QUICConnection * + get_quic_connection() override + { + return _qmux_connection.get(); + } + + std::unique_ptr _qmux_connection; +#endif + private: void _make_ssl_connection(SSL_CTX *ctx); void _bindSSLObject(); diff --git a/src/iocore/net/SSLNetVConnection.cc b/src/iocore/net/SSLNetVConnection.cc index a1b5ea43bdc..bb7ccb5ddec 100644 --- a/src/iocore/net/SSLNetVConnection.cc +++ b/src/iocore/net/SSLNetVConnection.cc @@ -1030,6 +1030,15 @@ SSLNetVConnection::clear() ssl = nullptr; } +#if TS_USE_QMUX + // The destructor never runs (ClassAllocator), so the + // QMux connection has to be released here or it leaks on every VC recycle. + // Clear the QUICSupport service slot too, or a recycled non-QMux VC would + // still report a (now null) QUIC connection to get_service(). + _qmux_connection.reset(); + this->_set_service(static_cast(nullptr)); +#endif + ALPNSupport::clear(); TLSBasicSupport::clear(); TLSEventSupport::clear(); @@ -1482,6 +1491,20 @@ SSLNetVConnection::sslServerHandShakeEvent(int &err) this->set_negotiated_protocol_id({reinterpret_cast(proto), static_cast(len)}); Dbg(dbg_ctl_ssl, "Origin selected next protocol '%.*s'", len, proto); + +#if TS_USE_QMUX + if (this->get_negotiated_protocol_id() == TS_ALPN_PROTOCOL_INDEX_H3QX) { + Dbg(dbg_ctl_ssl, "ALPN h3qx-01: creating QMuxConnection"); + _qmux_connection = std::make_unique(this); + if (_qmux_connection->is_closed()) { + // Config or quiche_accept() failed. Don't advertise a QUIC connection that can + // never make progress -- fail the connection instead of completing the handshake. + _qmux_connection.reset(); + return EVENT_ERROR; + } + this->_set_service(static_cast(this)); + } +#endif } else { Dbg(dbg_ctl_ssl, "Origin did not select a next protocol"); } diff --git a/src/iocore/net/qmux/CMakeLists.txt b/src/iocore/net/qmux/CMakeLists.txt new file mode 100644 index 00000000000..4ec8c648120 --- /dev/null +++ b/src/iocore/net/qmux/CMakeLists.txt @@ -0,0 +1,23 @@ +####################### +# +# Licensed to the Apache Software Foundation (ASF) under one or more contributor license +# agreements. See the NOTICE file distributed with this work for additional information regarding +# copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software distributed under the License +# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +# or implied. See the License for the specific language governing permissions and limitations under +# the License. +# +####################### + +add_library(qmux STATIC QMuxConnection.cc) +add_library(ts::qmux ALIAS qmux) + +target_link_libraries(qmux PUBLIC quiche::quiche ts::quic ts::inkevent ts::tscore) + +clang_tidy_check(qmux) diff --git a/src/iocore/net/qmux/QMuxConnection.cc b/src/iocore/net/qmux/QMuxConnection.cc new file mode 100644 index 00000000000..3b73b16c5bc --- /dev/null +++ b/src/iocore/net/qmux/QMuxConnection.cc @@ -0,0 +1,543 @@ +/** @file + + QMux connection implementation wrapping quiche_conn (draft-opik-quic-qmux-01) + + @section license License + + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + */ + +#include "iocore/net/qmux/QMuxConnection.h" +#include "iocore/net/NetVConnection.h" +#include "iocore/net/quic/QUICContext.h" +#include "iocore/net/quic/QUICApplicationMap.h" +#include "iocore/net/quic/QUICStreamManager.h" +#include "iocore/eventsystem/EThread.h" +#include "iocore/eventsystem/IOBuffer.h" +#include "iocore/eventsystem/VIO.h" +#include "tscore/Diags.h" +#include "tscore/ink_hrtime.h" +#include "tsutil/DbgCtl.h" + +#include +#include +#include +#include +#include +#include + +namespace +{ +DbgCtl dbg_ctl_qmux{"qmux"}; +constexpr int QMUX_IO_BUFFER_SIZE_INDEX = BUFFER_SIZE_INDEX_32K; + +// Largest Frames field we advertise via qmux_max_record_size. quiche rejects +// anything smaller than this, and enforces the limit on records the peer sends. +constexpr uint64_t QMUX_MAX_RECORD_SIZE = 16382; + +// A record is Size (varint, at most 8 bytes) followed by Frames, so this bounds +// the bytes that must be contiguous for quiche to parse one record. +constexpr int64_t QMUX_MAX_RECORD_BYTES = QMUX_MAX_RECORD_SIZE + 8; + +// Staging size for records handed to the transport on each send. +constexpr int64_t QMUX_SEND_BUFFER_SIZE = 65535; + +constexpr QUICVersion QMUX_QUIC_VERSION = 0x00000001; + +std::once_flag qmux_shared_config_once; +} // end anonymous namespace + +quiche_config *QMuxConnection::_shared_config = nullptr; + +void +QMuxConnection::_init_shared_config() +{ + std::call_once(qmux_shared_config_once, []() { + quiche_config *config = quiche_config_new(QUICHE_PROTOCOL_VERSION); + if (config == nullptr) { + Error("failed to create a QMux config"); + return; + } + + std::string alpn("\x07h3qx-01"); + quiche_config_set_application_protos(config, reinterpret_cast(alpn.c_str()), alpn.size()); + + quiche_config_set_max_idle_timeout(config, 30000); + quiche_config_set_initial_max_data(config, 10000000); + quiche_config_set_initial_max_stream_data_bidi_local(config, 1000000); + quiche_config_set_initial_max_stream_data_bidi_remote(config, 1000000); + quiche_config_set_initial_max_stream_data_uni(config, 1000000); + quiche_config_set_initial_max_streams_bidi(config, 100); + quiche_config_set_initial_max_streams_uni(config, 100); + quiche_config_set_disable_active_migration(config, true); + + quiche_config_enable_qmux(config, true); + quiche_config_set_qmux_max_record_size(config, QMUX_MAX_RECORD_SIZE); + + _shared_config = config; + }); +} + +QMuxConnection::QMuxConnection(NetVConnection *netvc) : Continuation(netvc->mutex) +{ + _init_shared_config(); + SET_HANDLER(&QMuxConnection::main_event); + + _synthetic_cid.randomize(); + _cids_str = _synthetic_cid.hex(); + + auto *local_ep = netvc->get_local_addr(); + auto *peer_ep = netvc->get_remote_addr(); + + _local_addr_len = ats_ip_size(local_ep); + _peer_addr_len = ats_ip_size(peer_ep); + memcpy(&_local_addr, local_ep, _local_addr_len); + memcpy(&_peer_addr, peer_ep, _peer_addr_len); + + if (_shared_config != nullptr) { + _quiche_con = + quiche_accept(_synthetic_cid, _synthetic_cid.length(), nullptr, 0, reinterpret_cast(&_local_addr), + _local_addr_len, reinterpret_cast(&_peer_addr), _peer_addr_len, _shared_config); + } + if (_quiche_con == nullptr) { + Error("failed to create a QMux connection"); + _closed = true; + } + + _context = std::make_unique(this); + _app_map = std::make_unique(); + _stream_manager = std::make_unique(_context.get(), _app_map.get()); +} + +QMuxConnection::~QMuxConnection() +{ + _unschedule_quiche_timeout(); + if (_read_reader) { + _read_reader->dealloc(); + } + if (_read_buf) { + free_MIOBuffer(_read_buf); + } + if (_write_buf) { + free_MIOBuffer(_write_buf); + } + if (_quiche_con != nullptr) { + quiche_conn_free(_quiche_con); + _quiche_con = nullptr; + } +} + +void +QMuxConnection::start(NetVConnection *netvc) +{ + _read_buf = new_MIOBuffer(QMUX_IO_BUFFER_SIZE_INDEX); + _read_buf->water_mark = QMUX_MAX_RECORD_BYTES; + _read_reader = _read_buf->alloc_reader(); + _write_buf = new_MIOBuffer(QMUX_IO_BUFFER_SIZE_INDEX); + + netvc->do_io_read(this, INT64_MAX, _read_buf); + _write_vio = netvc->do_io_write(this, INT64_MAX, _write_buf->alloc_reader()); + + _schedule_quiche_timeout(); +} + +void +QMuxConnection::_schedule_quiche_timeout() +{ + if (!_quiche_timeout && _quiche_con != nullptr) { + _quiche_timeout = this_ethread()->schedule_in(this, HRTIME_MSECONDS(quiche_conn_timeout_as_millis(_quiche_con))); + } +} + +void +QMuxConnection::_unschedule_quiche_timeout() +{ + if (_quiche_timeout) { + _quiche_timeout->cancel(); + _quiche_timeout = nullptr; + } +} + +void +QMuxConnection::signal_write_ready() +{ + if (_in_write) { + return; + } + if (_write_vio) { + SCOPED_MUTEX_LOCK(lock, this->mutex, this_ethread()); + _write_vio->reenable(); + } +} + +int +QMuxConnection::main_event(int event, void *data) +{ + if (_quiche_con == nullptr) { + return EVENT_DONE; + } + + switch (event) { + case VC_EVENT_READ_READY: + case VC_EVENT_READ_COMPLETE: + _handle_read(); + break; + case VC_EVENT_WRITE_READY: + case VC_EVENT_WRITE_COMPLETE: + _handle_write(); + break; + case EVENT_INTERVAL: + ink_assert(_quiche_timeout == data); + _quiche_timeout = nullptr; + quiche_conn_on_timeout(_quiche_con); + _flush_quiche_output(); + if (quiche_conn_is_closed(_quiche_con)) { + close_quic_connection(nullptr); + } else { + _schedule_quiche_timeout(); + } + break; + case VC_EVENT_EOS: + case VC_EVENT_ERROR: + case VC_EVENT_INACTIVITY_TIMEOUT: + case VC_EVENT_ACTIVE_TIMEOUT: + Dbg(dbg_ctl_qmux, "connection event %d, closing", event); + close_quic_connection(nullptr); + break; + default: + break; + } + + return EVENT_CONT; +} + +void +QMuxConnection::_handle_read() +{ + if (_read_reader->read_avail() <= 0) { + return; + } + + // quiche parses at most one record per call and needs it in contiguous + // memory. A record can straddle IOBufferBlock boundaries, so the spanning + // case is staged through this buffer; the common case reads in place. + uint8_t staging[QMUX_MAX_RECORD_BYTES]; + + while (_read_reader->read_avail() > 0) { + int64_t avail = _read_reader->read_avail(); + int64_t blk_len = _read_reader->block_read_avail(); + + if (blk_len <= 0) { + _read_reader->skip_empty_blocks(); + continue; + } + + uint8_t *buf = nullptr; + int64_t len = 0; + + if (blk_len == avail) { + buf = reinterpret_cast(_read_reader->start()); + len = blk_len; + } else { + len = std::min(avail, QMUX_MAX_RECORD_BYTES); + _read_reader->memcpy(staging, len, 0); + buf = staging; + } + + quiche_recv_info recv_info = {}; + recv_info.from = const_cast(reinterpret_cast(&_peer_addr)); + recv_info.from_len = _peer_addr_len; + recv_info.to = const_cast(reinterpret_cast(&_local_addr)); + recv_info.to_len = _local_addr_len; + + ssize_t done = quiche_conn_recv(_quiche_con, buf, len, &recv_info); + if (done < 0) { + if (done == QUICHE_ERR_DONE) { + // No complete record in what's buffered yet. Leave the bytes for the next read event. + } else { + // quiche has already classified this as an unrecoverable per-connection error and + // started its own internal close/drain sequence -- these bytes will never parse + // successfully, so close now instead of leaving them to linger until the next + // scheduled quiche_conn_on_timeout() notices the connection is closed. + Dbg(dbg_ctl_qmux, "quiche_conn_recv error: %zd", done); + close_quic_connection(nullptr); + } + break; + } + _read_reader->consume(done); + } + + _handle_read_streams(); + _handle_write(); +} + +void +QMuxConnection::_handle_read_streams() +{ + quiche_stream_iter *readable = quiche_conn_readable(_quiche_con); + uint64_t stream_id; + + while (quiche_stream_iter_next(readable, &stream_id)) { + QUICStream *stream = _stream_manager->find_stream(stream_id); + if (stream == nullptr) { + QUICConnectionError err; + stream = _stream_manager->create_stream(stream_id, err); + if (stream == nullptr) { + Dbg(dbg_ctl_qmux, "failed to create stream %" PRIu64, stream_id); + continue; + } + } + stream->receive_data(*this); + } + quiche_stream_iter_free(readable); +} + +void +QMuxConnection::_handle_write() +{ + _in_write = true; + if (!quiche_conn_is_established(_quiche_con)) { + // Our own QX_TRANSPORT_PARAMETERS haven't been sent yet. Flush now so that, + // if the peer's have already been received, the connection is established + // before the _handle_write_streams() check below -- otherwise a stream + // queued before this call (e.g. the HTTP/3 control stream) misses this + // cycle, and nothing else is guaranteed to trigger another one. + _flush_quiche_output(); + } + _handle_write_streams(); + _flush_quiche_output(); + _in_write = false; +} + +void +QMuxConnection::_flush_quiche_output() +{ + bool wrote = false; + uint8_t out[QMUX_SEND_BUFFER_SIZE]; + quiche_send_info send_info; + + for (;;) { + ssize_t written = quiche_conn_send(_quiche_con, out, sizeof(out), &send_info); + if (written == QUICHE_ERR_DONE) { + break; + } + if (written < 0) { + Dbg(dbg_ctl_qmux, "quiche_conn_send error: %zd", written); + break; + } + _write_buf->write(out, written); + wrote = true; + } + + if (wrote && _write_vio) { + _write_vio->reenable(); + } +} + +void +QMuxConnection::_handle_write_streams() +{ + if (!quiche_conn_is_established(_quiche_con)) { + return; + } + + quiche_stream_iter *writable = quiche_conn_writable(_quiche_con); + uint64_t stream_id; + + while (quiche_stream_iter_next(writable, &stream_id)) { + QUICStream *stream = _stream_manager->find_stream(stream_id); + if (stream != nullptr) { + stream->send_data(*this); + } + } + quiche_stream_iter_free(writable); +} + +// --- QUICConnectionInfoProvider --- + +QUICConnectionId +QMuxConnection::peer_connection_id() const +{ + return QUICConnectionId::ZERO(); +} + +QUICConnectionId +QMuxConnection::original_connection_id() const +{ + return QUICConnectionId::ZERO(); +} + +QUICConnectionId +QMuxConnection::first_connection_id() const +{ + return _synthetic_cid; +} + +QUICConnectionId +QMuxConnection::retry_source_connection_id() const +{ + return QUICConnectionId::ZERO(); +} + +QUICConnectionId +QMuxConnection::initial_source_connection_id() const +{ + return _synthetic_cid; +} + +QUICConnectionId +QMuxConnection::connection_id() const +{ + return _synthetic_cid; +} + +std::string_view +QMuxConnection::cids() const +{ + return _cids_str; +} + +const QUICFiveTuple +QMuxConnection::five_tuple() const +{ + return QUICFiveTuple(); +} + +uint32_t +QMuxConnection::pmtu() const +{ + // Not meaningful over TCP. + return QMUX_SEND_BUFFER_SIZE; +} + +NetVConnectionContext_t +QMuxConnection::direction() const +{ + return NET_VCONNECTION_IN; +} + +bool +QMuxConnection::is_closed() const +{ + return _closed; +} + +bool +QMuxConnection::is_at_anti_amplification_limit() const +{ + return false; +} + +bool +QMuxConnection::is_address_validation_completed() const +{ + return true; +} + +bool +QMuxConnection::is_handshake_completed() const +{ + return true; +} + +QUICVersion +QMuxConnection::negotiated_version() const +{ + return QMUX_QUIC_VERSION; +} + +std::string_view +QMuxConnection::negotiated_application_name() const +{ + return "h3qx-01"; +} + +void +QMuxConnection::on_stream_updated() +{ + this->signal_write_ready(); +} + +// --- QUICStreamIO --- + +int64_t +QMuxConnection::read_stream(QUICStreamId stream_id, uint8_t *buf, size_t len, bool &fin, QUICStreamIO::ErrorCode &error_code) +{ + return quiche_conn_stream_recv(_quiche_con, stream_id, buf, len, &fin, &error_code); +} + +bool +QMuxConnection::stream_read_finished(QUICStreamId stream_id) +{ + return quiche_conn_stream_finished(_quiche_con, stream_id); +} + +int64_t +QMuxConnection::stream_write_capacity(QUICStreamId stream_id) +{ + return quiche_conn_stream_capacity(_quiche_con, stream_id); +} + +int64_t +QMuxConnection::write_stream(QUICStreamId stream_id, uint8_t const *buf, size_t len, bool fin, QUICStreamIO::ErrorCode &error_code) +{ + return quiche_conn_stream_send(_quiche_con, stream_id, const_cast(buf), len, fin, &error_code); +} + +// --- QUICConnection --- + +QUICStreamManager * +QMuxConnection::stream_manager() +{ + return _stream_manager.get(); +} + +void +QMuxConnection::close_quic_connection(QUICConnectionErrorUPtr error) +{ + if (_closed) { + return; + } + _closed = true; + + const bool is_app_error = error != nullptr && error->cls == QUICErrorClass::APPLICATION; + const uint64_t err_code = error == nullptr ? static_cast(QUICTransErrorCode::NO_ERROR) : error->code; + + if (int rv = quiche_conn_close(_quiche_con, is_app_error, err_code, nullptr, 0); rv < 0) { + Dbg(dbg_ctl_qmux, "[%s] quiche_conn_close error: %d", _cids_str.c_str(), rv); + } + // quiche_conn_close() only queues the CLOSE frame; it has to be flushed like any other + // outgoing data or the peer never sees it. + _flush_quiche_output(); + Dbg(dbg_ctl_qmux, "[%s] connection closed with error %" PRIu64, _cids_str.c_str(), err_code); +} + +void +QMuxConnection::reset_quic_connection() +{ + _closed = true; +} + +void +QMuxConnection::handle_received_packet(UDPPacket * /* packet ATS_UNUSED */) +{ +} + +void +QMuxConnection::ping() +{ +} diff --git a/src/proxy/CMakeLists.txt b/src/proxy/CMakeLists.txt index c0814f6009f..f4c74c8c8de 100644 --- a/src/proxy/CMakeLists.txt +++ b/src/proxy/CMakeLists.txt @@ -52,7 +52,7 @@ add_subdirectory(http) add_subdirectory(http2) add_subdirectory(logging) -if(TS_USE_QUIC) +if(TS_USE_QUIC OR TS_USE_QMUX) add_subdirectory(http3) endif() diff --git a/src/proxy/http/CMakeLists.txt b/src/proxy/http/CMakeLists.txt index 55cf7ebd36e..bf36cd4a783 100644 --- a/src/proxy/http/CMakeLists.txt +++ b/src/proxy/http/CMakeLists.txt @@ -50,7 +50,7 @@ target_link_libraries( PRIVATE ts::http2 ts::http_remap ts::inkcache ts::inkutils ts::logging ) -if(TS_USE_QUIC) +if(TS_USE_QUIC OR TS_USE_QMUX) target_link_libraries(http PRIVATE ts::http3) endif() diff --git a/src/proxy/http/HttpProxyServerMain.cc b/src/proxy/http/HttpProxyServerMain.cc index 133f70116ee..467fcad36a0 100644 --- a/src/proxy/http/HttpProxyServerMain.cc +++ b/src/proxy/http/HttpProxyServerMain.cc @@ -40,6 +40,8 @@ #include "../../iocore/net/P_QUICNetProcessor.h" #include "../../iocore/net/P_QUICNextProtocolAccept.h" #include "proxy/http3/Http3SessionAccept.h" +#elif TS_USE_QMUX == 1 +#include "proxy/http3/Http3SessionAccept.h" #endif #include @@ -223,6 +225,9 @@ MakeHttpProxyAcceptor(HttpProxyAcceptor &acceptor, HttpProxyPort &port, unsigned ssl->registerEndpoint(TS_ALPN_PROTOCOL_HTTP_1_0, http); ssl->registerEndpoint(TS_ALPN_PROTOCOL_HTTP_1_1, http); ssl->registerEndpoint(TS_ALPN_PROTOCOL_HTTP_2_0, new Http2SessionAccept(accept_opt)); +#if TS_USE_QMUX + ssl->registerEndpoint(TS_ALPN_PROTOCOL_H3QX, new Http3SessionAccept(accept_opt)); +#endif SCOPED_MUTEX_LOCK(lock, ssl_plugin_mutex, this_ethread()); ssl_plugin_acceptors.push(ssl); diff --git a/src/proxy/http3/CMakeLists.txt b/src/proxy/http3/CMakeLists.txt index c4c71dba53f..42beb522023 100644 --- a/src/proxy/http3/CMakeLists.txt +++ b/src/proxy/http3/CMakeLists.txt @@ -47,6 +47,10 @@ target_link_libraries( PRIVATE ts::proxy ) +if(TS_USE_QMUX) + target_link_libraries(http3 PUBLIC ts::qmux) +endif() + if(BUILD_TESTING) add_executable( test_http3 diff --git a/src/proxy/http3/Http3SessionAccept.cc b/src/proxy/http3/Http3SessionAccept.cc index d82546b6060..8a7e9da0360 100644 --- a/src/proxy/http3/Http3SessionAccept.cc +++ b/src/proxy/http3/Http3SessionAccept.cc @@ -32,6 +32,10 @@ #include "proxy/http3/Http09App.h" #include "proxy/http3/Http3App.h" +#if TS_USE_QMUX +#include "iocore/net/qmux/QMuxConnection.h" +#endif + namespace { DbgCtl dbg_ctl_http3{"http3"}; @@ -77,10 +81,26 @@ Http3SessionAccept::accept(NetVConnection *netvc, MIOBuffer * /* iobuf ATS_UNUSE if (IP_PROTO_TAG_HTTP_QUIC.compare(alpn) == 0 || IP_PROTO_TAG_HTTP_QUIC_D29.compare(alpn) == 0) { Dbg(dbg_ctl_http3, "[%s] start HTTP/0.9 app (ALPN=%.*s)", qc->cids().data(), static_cast(alpn.length()), alpn.data()); new Http09App(netvc, qc, std::move(session_acl), this->options); - } else if (IP_PROTO_TAG_HTTP_3.compare(alpn) == 0 || IP_PROTO_TAG_HTTP_3_D29.compare(alpn) == 0) { + } else if (IP_PROTO_TAG_HTTP_3.compare(alpn) == 0 || IP_PROTO_TAG_HTTP_3_D29.compare(alpn) == 0 || + IP_PROTO_TAG_H3QX.compare(alpn) == 0) { Dbg(dbg_ctl_http3, "[%s] start HTTP/3 app (ALPN=%.*s)", qc->cids().data(), static_cast(alpn.length()), alpn.data()); Http3App *app = new Http3App(netvc, qc, std::move(session_acl), this->options); + +#if TS_USE_QMUX + if (IP_PROTO_TAG_H3QX.compare(alpn) == 0) { + // Http3App's constructor runs the generic ProxySession start-up (HQSession::start()), + // which claims the netvc's read/write VIOs for itself. Reclaim them for QMuxConnection + // here, after that clobber and before app->start() can generate any stream I/O that + // would need them. + auto *qmux_con = dynamic_cast(qc); + if (!qmux_con) { + ink_abort("negotiated h3qx-01 but QUICConnection is not a QMuxConnection"); + } + qmux_con->start(netvc); + } +#endif + app->start(); } else { ink_abort("Negotiated App Name is unknown"); diff --git a/src/records/RecHttp.cc b/src/records/RecHttp.cc index 7f1dbd68f95..5b689a3d0b1 100644 --- a/src/records/RecHttp.cc +++ b/src/records/RecHttp.cc @@ -51,6 +51,7 @@ const char *const TS_ALPN_PROTOCOL_HTTP_3 = IP_PROTO_TAG_HTTP_3.data(); const char *const TS_ALPN_PROTOCOL_HTTP_QUIC = IP_PROTO_TAG_HTTP_QUIC.data(); const char *const TS_ALPN_PROTOCOL_HTTP_3_D29 = IP_PROTO_TAG_HTTP_3_D29.data(); const char *const TS_ALPN_PROTOCOL_HTTP_QUIC_D29 = IP_PROTO_TAG_HTTP_QUIC_D29.data(); +const char *const TS_ALPN_PROTOCOL_H3QX = IP_PROTO_TAG_H3QX.data(); const char *const TS_ALPN_PROTOCOL_GROUP_HTTP = "http"; const char *const TS_ALPN_PROTOCOL_GROUP_HTTP2 = "http2"; @@ -62,6 +63,7 @@ const char *const TS_PROTO_TAG_HTTP_3 = TS_ALPN_PROTOCOL_HTTP_3; const char *const TS_PROTO_TAG_HTTP_QUIC = TS_ALPN_PROTOCOL_HTTP_QUIC; const char *const TS_PROTO_TAG_HTTP_3_D29 = TS_ALPN_PROTOCOL_HTTP_3_D29; const char *const TS_PROTO_TAG_HTTP_QUIC_D29 = TS_ALPN_PROTOCOL_HTTP_QUIC_D29; +const char *const TS_PROTO_TAG_H3QX = TS_ALPN_PROTOCOL_H3QX; const char *const TS_PROTO_TAG_TLS_1_3 = IP_PROTO_TAG_TLS_1_3.data(); const char *const TS_PROTO_TAG_TLS_1_2 = IP_PROTO_TAG_TLS_1_2.data(); const char *const TS_PROTO_TAG_TLS_1_1 = IP_PROTO_TAG_TLS_1_1.data(); @@ -82,6 +84,7 @@ int TS_ALPN_PROTOCOL_INDEX_HTTP_3 = SessionProtocolNameRegistry::INVALID; int TS_ALPN_PROTOCOL_INDEX_HTTP_QUIC = SessionProtocolNameRegistry::INVALID; int TS_ALPN_PROTOCOL_INDEX_HTTP_3_D29 = SessionProtocolNameRegistry::INVALID; int TS_ALPN_PROTOCOL_INDEX_HTTP_QUIC_D29 = SessionProtocolNameRegistry::INVALID; +int TS_ALPN_PROTOCOL_INDEX_H3QX = SessionProtocolNameRegistry::INVALID; // Predefined protocol sets for ease of use. SessionProtocolSet HTTP_PROTOCOL_SET; @@ -221,6 +224,7 @@ constexpr std::string_view TS_ALPN_PROTO_ID_OPENSSL_HTTP_1_0("\x8http/1.0"); constexpr std::string_view TS_ALPN_PROTO_ID_OPENSSL_HTTP_1_1("\x8http/1.1"); constexpr std::string_view TS_ALPN_PROTO_ID_OPENSSL_HTTP_2("\x2h2"); constexpr std::string_view TS_ALPN_PROTO_ID_OPENSSL_HTTP_3("\x2h3"); +constexpr std::string_view TS_ALPN_PROTO_ID_OPENSSL_H3QX("\x7h3qx-01"); bool parse_octal_mode(const char *s, mode_t &out) @@ -836,6 +840,7 @@ ts_session_protocol_well_known_name_indices_init() TS_ALPN_PROTOCOL_INDEX_HTTP_QUIC = globalSessionProtocolNameRegistry.toIndexConst(std::string_view{TS_ALPN_PROTOCOL_HTTP_QUIC}); TS_ALPN_PROTOCOL_INDEX_HTTP_QUIC_D29 = globalSessionProtocolNameRegistry.toIndexConst(std::string_view{TS_ALPN_PROTOCOL_HTTP_QUIC_D29}); + TS_ALPN_PROTOCOL_INDEX_H3QX = globalSessionProtocolNameRegistry.toIndexConst(std::string_view{TS_ALPN_PROTOCOL_H3QX}); // Now do the predefined protocol sets. HTTP_PROTOCOL_SET.markIn(TS_ALPN_PROTOCOL_INDEX_HTTP_0_9); @@ -846,6 +851,7 @@ ts_session_protocol_well_known_name_indices_init() DEFAULT_TLS_SESSION_PROTOCOL_SET.markAllIn(); DEFAULT_TLS_SESSION_PROTOCOL_SET.markOut(TS_ALPN_PROTOCOL_INDEX_HTTP_3); DEFAULT_TLS_SESSION_PROTOCOL_SET.markOut(TS_ALPN_PROTOCOL_INDEX_HTTP_QUIC); + DEFAULT_TLS_SESSION_PROTOCOL_SET.markOut(TS_ALPN_PROTOCOL_INDEX_H3QX); DEFAULT_QUIC_SESSION_PROTOCOL_SET.markIn(TS_ALPN_PROTOCOL_INDEX_HTTP_3); DEFAULT_QUIC_SESSION_PROTOCOL_SET.markIn(TS_ALPN_PROTOCOL_INDEX_HTTP_QUIC); @@ -861,6 +867,7 @@ ts_session_protocol_well_known_name_indices_init() TSProtoTags.insert(TS_PROTO_TAG_HTTP_QUIC); TSProtoTags.insert(TS_PROTO_TAG_HTTP_3_D29); TSProtoTags.insert(TS_PROTO_TAG_HTTP_QUIC_D29); + TSProtoTags.insert(TS_PROTO_TAG_H3QX); TSProtoTags.insert(TS_PROTO_TAG_TLS_1_3); TSProtoTags.insert(TS_PROTO_TAG_TLS_1_2); TSProtoTags.insert(TS_PROTO_TAG_TLS_1_1); @@ -898,6 +905,8 @@ SessionProtocolNameRegistry::convert_openssl_alpn_wire_format(int index) return TS_ALPN_PROTO_ID_OPENSSL_HTTP_2; } else if (index == TS_ALPN_PROTOCOL_INDEX_HTTP_3) { return TS_ALPN_PROTO_ID_OPENSSL_HTTP_3; + } else if (index == TS_ALPN_PROTOCOL_INDEX_H3QX) { + return TS_ALPN_PROTO_ID_OPENSSL_H3QX; } return {}; diff --git a/src/traffic_layout/info.cc b/src/traffic_layout/info.cc index 9b1e6c331c1..746eeca499b 100644 --- a/src/traffic_layout/info.cc +++ b/src/traffic_layout/info.cc @@ -160,6 +160,7 @@ produce_features(bool json) print_feature("TS_USE_HWLOC", TS_USE_HWLOC, json); print_feature("TS_USE_TLS13", TS_USE_TLS13, json); print_feature("TS_USE_QUIC", TS_USE_QUIC, json); + print_feature("TS_USE_QMUX", TS_USE_QMUX, json); print_feature("TS_HAS_OPENSSL_QUIC", TS_HAS_OPENSSL_QUIC, json); print_feature("TS_HAS_QUICHE", TS_HAS_QUICHE, json); print_feature("TS_HAS_SO_PEERCRED", TS_HAS_SO_PEERCRED, json); diff --git a/src/traffic_server/CMakeLists.txt b/src/traffic_server/CMakeLists.txt index f50ff2ef824..9f1e1fc20f9 100644 --- a/src/traffic_server/CMakeLists.txt +++ b/src/traffic_server/CMakeLists.txt @@ -50,10 +50,14 @@ if(NOT APPLE) target_link_options(traffic_server PRIVATE -Wl,--no-undefined,--no-allow-shlib-undefined) endif() -if(TS_USE_QUIC) +if(TS_USE_QUIC OR TS_USE_QMUX) target_link_libraries(traffic_server PRIVATE ts::http3 ts::quic) endif() +if(TS_USE_QMUX) + target_link_libraries(traffic_server PRIVATE ts::qmux) +endif() + if(TS_HAS_PROFILER) target_link_libraries(traffic_server PRIVATE gperftools::profiler) endif() diff --git a/src/traffic_server/traffic_server.cc b/src/traffic_server/traffic_server.cc index 57158fd66b1..f253c0cdead 100644 --- a/src/traffic_server/traffic_server.cc +++ b/src/traffic_server/traffic_server.cc @@ -127,7 +127,7 @@ extern "C" int plock(int); #include "mgmt/config/FileManager.h" -#if TS_USE_QUIC == 1 +#if TS_USE_QUIC == 1 || TS_USE_QMUX == 1 #include "proxy/http3/Http3.h" #include "proxy/http3/Http3Config.h" #endif @@ -2159,7 +2159,7 @@ main(int /* argc ATS_UNUSED */, const char **argv) // We want to initialize Machine as early as possible because it // has other dependencies. Hopefully not in prep_HttpProxyServer(). HttpConfig::startup(); -#if TS_USE_QUIC == 1 +#if TS_USE_QUIC == 1 || TS_USE_QMUX == 1 ts::Http3Config::startup(); #endif @@ -2353,7 +2353,7 @@ main(int /* argc ATS_UNUSED */, const char **argv) // Initialize HTTP/2 Http2::init(); -#if TS_USE_QUIC == 1 +#if TS_USE_QUIC == 1 || TS_USE_QMUX == 1 // Initialize HTTP/QUIC Http3::init(); #endif diff --git a/src/tscore/ink_inet.cc b/src/tscore/ink_inet.cc index 0194dc9d942..6d8f9eda5e3 100644 --- a/src/tscore/ink_inet.cc +++ b/src/tscore/ink_inet.cc @@ -56,6 +56,7 @@ const std::string_view IP_PROTO_TAG_HTTP_QUIC("hq"sv); // HTTP/0.9 over Q const std::string_view IP_PROTO_TAG_HTTP_3("h3"sv); // HTTP/3 over QUIC const std::string_view IP_PROTO_TAG_HTTP_QUIC_D29("hq-29"sv); // HTTP/0.9 over QUIC (draft-29) const std::string_view IP_PROTO_TAG_HTTP_3_D29("h3-29"sv); // HTTP/3 over QUIC (draft-29) +const std::string_view IP_PROTO_TAG_H3QX("h3qx-01"sv); // HTTP/3 over QMux (TLS/TCP) const std::string_view UNIX_PROTO_TAG{"unix"sv}; diff --git a/tests/gold_tests/qmux/go_qmux_client/go.mod b/tests/gold_tests/qmux/go_qmux_client/go.mod new file mode 100644 index 00000000000..531eb56c069 --- /dev/null +++ b/tests/gold_tests/qmux/go_qmux_client/go.mod @@ -0,0 +1,15 @@ +module qmux_client + +go 1.26.1 + +require ( + github.com/okdaichi/qmux-go v0.2.0 + github.com/quic-go/qpack v0.6.0 + github.com/quic-go/quic-go v0.59.0 +) + +require ( + golang.org/x/crypto v0.50.0 // indirect + golang.org/x/net v0.53.0 // indirect + golang.org/x/sys v0.43.0 // indirect +) diff --git a/tests/gold_tests/qmux/go_qmux_client/go.sum b/tests/gold_tests/qmux/go_qmux_client/go.sum new file mode 100644 index 00000000000..b10735cd87f --- /dev/null +++ b/tests/gold_tests/qmux/go_qmux_client/go.sum @@ -0,0 +1,26 @@ +github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g= +github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/okdaichi/qmux-go v0.2.0 h1:FiAJN99zhe9CcEHbJCBonOhjVhNRNLS1GPCQbE+Etx4= +github.com/okdaichi/qmux-go v0.2.0/go.mod h1:M3k3+VbBl98QagraePQqavelrSM15FQSHJgnS+RjKOU= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= +github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= +github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= +github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko= +go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o= +golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= +golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= +golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA= +golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs= +golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/tests/gold_tests/qmux/go_qmux_client/main.go b/tests/gold_tests/qmux/go_qmux_client/main.go new file mode 100644 index 00000000000..2fccebe4769 --- /dev/null +++ b/tests/gold_tests/qmux/go_qmux_client/main.go @@ -0,0 +1,340 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "flag" + "fmt" + "io" + "net" + "os" + "strconv" + "time" + + "github.com/okdaichi/qmux-go/qmux" + "github.com/quic-go/qpack" + "github.com/quic-go/quic-go/quicvarint" +) + +const ( + qmuxALPN = "h3qx-01" + bodyChunkSize = 8 * 1024 + largeBodySize = 300000 + + h3FrameData = 0x00 + h3FrameHeaders = 0x01 + h3FrameSettings = 0x04 + + h3ControlStream = 0x00 + h3QPACKEncoderStream = 0x02 + h3QPACKDecoderStream = 0x03 +) + +type requestCase struct { + name string + method string + path string + requestSize int + responseSize int +} + +func generatedBody(size int) []byte { + var body bytes.Buffer + for i := 0; body.Len() < size; i++ { + fmt.Fprintf(&body, "%07x ", i) + } + return body.Bytes()[:size] +} + +func writeVarInt(w io.Writer, value uint64) error { + encoded := quicvarint.Append(nil, value) + _, err := w.Write(encoded) + return err +} + +func writeFrame(w io.Writer, frameType uint64, payload []byte) error { + header := quicvarint.Append(nil, frameType) + header = quicvarint.Append(header, uint64(len(payload))) + if _, err := w.Write(header); err != nil { + return err + } + _, err := w.Write(payload) + return err +} + +func writeRequestBody(w io.Writer, body []byte) error { + for len(body) > 0 { + chunkSize := min(len(body), bodyChunkSize) + if err := writeFrame(w, h3FrameData, body[:chunkSize]); err != nil { + return err + } + body = body[chunkSize:] + } + return nil +} + +func openUniStream(ctx context.Context, conn *qmux.Conn, streamType uint64) error { + stream, err := conn.OpenUniStreamSync(ctx) + if err != nil { + return err + } + return writeVarInt(stream, streamType) +} + +func initializeHTTP3(ctx context.Context, conn *qmux.Conn) error { + control, err := conn.OpenUniStreamSync(ctx) + if err != nil { + return fmt.Errorf("open control stream: %w", err) + } + if err := writeVarInt(control, h3ControlStream); err != nil { + return fmt.Errorf("write control stream type: %w", err) + } + if err := writeFrame(control, h3FrameSettings, nil); err != nil { + return fmt.Errorf("write SETTINGS frame: %w", err) + } + + if err := openUniStream(ctx, conn, h3QPACKEncoderStream); err != nil { + return fmt.Errorf("open QPACK encoder stream: %w", err) + } + if err := openUniStream(ctx, conn, h3QPACKDecoderStream); err != nil { + return fmt.Errorf("open QPACK decoder stream: %w", err) + } + return nil +} + +func encodeRequestHeaders(authority string, tc requestCase) ([]byte, error) { + var block bytes.Buffer + + encoder := qpack.NewEncoder(&block) + fields := []qpack.HeaderField{ + {Name: ":method", Value: tc.method}, + {Name: ":scheme", Value: "https"}, + {Name: ":authority", Value: authority}, + {Name: ":path", Value: tc.path}, + {Name: "user-agent", Value: "ats-qmux-go-autest"}, + {Name: "x-qmux-client", Value: "qmux-go"}, + {Name: "x-qmux-test-case", Value: tc.name}, + {Name: "uuid", Value: tc.name}, + } + if tc.requestSize > 0 { + fields = append( + fields, + qpack.HeaderField{Name: "content-type", Value: "application/octet-stream"}, + qpack.HeaderField{Name: "content-length", Value: strconv.Itoa(tc.requestSize)}, + ) + } + for _, field := range fields { + if err := encoder.WriteField(field); err != nil { + return nil, err + } + } + return block.Bytes(), nil +} + +func decodeResponseHeaders(block []byte) (string, string, string, error) { + var status string + var marker string + var contentLength string + + decode := qpack.NewDecoder().Decode(block) + for { + field, err := decode() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return "", "", "", err + } + switch field.Name { + case ":status": + status = field.Value + case "x-qmux-response": + marker = field.Value + case "content-length": + contentLength = field.Value + } + } + return status, marker, contentLength, nil +} + +func readResponse(stream *qmux.Stream) (string, string, string, []byte, error) { + reader := quicvarint.NewReader(stream) + var status string + var marker string + var contentLength string + var body bytes.Buffer + + for { + frameType, err := quicvarint.Read(reader) + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return "", "", "", nil, err + } + length, err := quicvarint.Read(reader) + if err != nil { + return "", "", "", nil, err + } + payload := make([]byte, length) + if _, err := io.ReadFull(reader, payload); err != nil { + return "", "", "", nil, err + } + + switch frameType { + case h3FrameHeaders: + decodedStatus, decodedMarker, decodedContentLength, err := decodeResponseHeaders(payload) + if err != nil { + return "", "", "", nil, fmt.Errorf("decode response headers: %w", err) + } + if decodedStatus != "" { + status = decodedStatus + } + if decodedMarker != "" { + marker = decodedMarker + } + if decodedContentLength != "" { + contentLength = decodedContentLength + } + case h3FrameData: + body.Write(payload) + } + } + return status, marker, contentLength, body.Bytes(), nil +} + +func request(ctx context.Context, conn *qmux.Conn, authority string, tc requestCase) error { + stream, err := conn.OpenStreamSync(ctx) + if err != nil { + return fmt.Errorf("%s: open request stream: %w", tc.name, err) + } + stream.SetDeadline(time.Now().Add(20 * time.Second)) + + headerBlock, err := encodeRequestHeaders(authority, tc) + if err != nil { + return fmt.Errorf("%s: encode request headers: %w", tc.name, err) + } + if err := writeFrame(stream, h3FrameHeaders, headerBlock); err != nil { + return fmt.Errorf("%s: write request headers: %w", tc.name, err) + } + if tc.requestSize > 0 { + if err := writeRequestBody(stream, generatedBody(tc.requestSize)); err != nil { + return fmt.Errorf("%s: write request body: %w", tc.name, err) + } + } + if err := stream.Close(); err != nil { + return fmt.Errorf("%s: finish request stream: %w", tc.name, err) + } + + status, marker, contentLength, body, err := readResponse(stream) + if err != nil { + return fmt.Errorf("%s: read response: %w", tc.name, err) + } + if status != "200" { + return fmt.Errorf("%s: expected status 200, got %q", tc.name, status) + } + if marker != "success" { + return fmt.Errorf("%s: expected X-QMux-Response success, got %q", tc.name, marker) + } + if contentLength != strconv.Itoa(tc.responseSize) { + return fmt.Errorf("%s: expected Content-Length %d, got %q", tc.name, tc.responseSize, contentLength) + } + expectedBody := generatedBody(tc.responseSize) + if !bytes.Equal(body, expectedBody) { + return fmt.Errorf("%s: response body mismatch: got %d bytes, expected %d", tc.name, len(body), len(expectedBody)) + } + + fmt.Printf("ok %s request=%d response=%d\n", tc.name, tc.requestSize, tc.responseSize) + return nil +} + +func run(addr string, authority string, serverName string) error { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + tcpConn, err := (&net.Dialer{}).DialContext(ctx, "tcp", addr) + if err != nil { + return fmt.Errorf("dial TCP: %w", err) + } + tlsConn := tls.Client(tcpConn, &tls.Config{ + InsecureSkipVerify: true, + MinVersion: tls.VersionTLS13, + NextProtos: []string{qmuxALPN}, + ServerName: serverName, + }) + if err := tlsConn.HandshakeContext(ctx); err != nil { + return fmt.Errorf("TLS handshake: %w", err) + } + if negotiated := tlsConn.ConnectionState().NegotiatedProtocol; negotiated != qmuxALPN { + return fmt.Errorf("expected ALPN %q, got %q", qmuxALPN, negotiated) + } + + config := qmux.DefaultConfig() + // qmux-go v0.2.0 uses a nonstandard code point for this optional parameter. + // Omitting it selects the interoperable protocol default of 16,382 bytes. + config.MaxRecordSize = 0 + config.InitialConnectionReceiveWindow = 10000000 + config.InitialStreamReceiveWindow = 1000000 + conn, err := qmux.Dial(newQMuxCompatConn(tlsConn), config) + if err != nil { + return fmt.Errorf("start QMux: %w", err) + } + defer conn.Close() + + if err := initializeHTTP3(ctx, conn); err != nil { + return err + } + cases := []requestCase{ + {name: "qmux-get-empty", method: "GET", path: "/qmux-get-empty"}, + {name: "qmux-post-small", method: "POST", path: "/qmux-post-small", requestSize: 100, responseSize: 100}, + { + name: "qmux-post-large", + method: "POST", + path: "/qmux-post-large", + requestSize: largeBodySize, + responseSize: largeBodySize, + }, + } + for _, tc := range cases { + if err := request(ctx, conn, authority, tc); err != nil { + return err + } + } + + fmt.Printf("completed %d QMux HTTP/3 requests: alpn=%s\n", len(cases), qmuxALPN) + return nil +} + +func main() { + addr := flag.String("addr", "", "ATS QMux address in host:port form") + authority := flag.String("authority", "", "HTTP/3 request authority") + serverName := flag.String("server-name", "", "TLS SNI server name") + flag.Parse() + + if *addr == "" || *authority == "" || *serverName == "" { + flag.Usage() + os.Exit(2) + } + if err := run(*addr, *authority, *serverName); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} diff --git a/tests/gold_tests/qmux/go_qmux_client/qmux_compat.go b/tests/gold_tests/qmux/go_qmux_client/qmux_compat.go new file mode 100644 index 00000000000..e7f164389b3 --- /dev/null +++ b/tests/gold_tests/qmux/go_qmux_client/qmux_compat.go @@ -0,0 +1,223 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "bytes" + "fmt" + "io" + "net" + "sync" + + "github.com/quic-go/quic-go/quicvarint" +) + +const qmuxTransportParametersFrameType = 0x3f5153300d0a0d0a + +const ( + qmuxStreamFrameType = 0x08 + qmuxStreamFrameTypeMask = 0xf8 + qmuxStreamFrameOffsetBit = 0x04 + qmuxStreamFrameLengthBit = 0x02 +) + +// qmuxCompatConn adapts qmux-go v0.2.0's initial transport-parameter frame to +// draft-ietf-quic-qmux-01. The release omits the transport-parameter frame's +// payload length and cannot parse STREAM frames without a LEN field, so the +// adapter normalizes both differences before qmux-go sees them. +type qmuxCompatConn struct { + net.Conn + readMutex sync.Mutex + readBuffer bytes.Buffer + readReady bool + writeMutex sync.Mutex + writeDone bool +} + +func newQMuxCompatConn(conn net.Conn) net.Conn { + return &qmuxCompatConn{Conn: conn} +} + +func (conn *qmuxCompatConn) Read(data []byte) (int, error) { + conn.readMutex.Lock() + defer conn.readMutex.Unlock() + + if conn.readBuffer.Len() == 0 { + var adapted []byte + var err error + if conn.readReady { + adapted, err = conn.readRecord() + } else { + adapted, err = conn.readInitialRecord() + conn.readReady = true + } + if err != nil { + return 0, err + } + conn.readBuffer.Write(adapted) + } + return conn.readBuffer.Read(data) +} + +func (conn *qmuxCompatConn) readRecord() ([]byte, error) { + reader := quicvarint.NewReader(conn.Conn) + recordLength, err := quicvarint.Read(reader) + if err != nil { + return nil, err + } + payload := make([]byte, recordLength) + if _, err := io.ReadFull(reader, payload); err != nil { + return nil, err + } + return adaptStreamFrameRecord(payload) +} + +func adaptStreamFrameRecord(payload []byte) ([]byte, error) { + frameType, frameTypeBytes, err := quicvarint.Parse(payload) + if err != nil { + return nil, err + } + if frameType&qmuxStreamFrameTypeMask != qmuxStreamFrameType || frameType&qmuxStreamFrameLengthBit != 0 { + return appendRecord(nil, payload), nil + } + + headerEnd := frameTypeBytes + _, streamIDBytes, err := quicvarint.Parse(payload[headerEnd:]) + if err != nil { + return nil, err + } + headerEnd += streamIDBytes + if frameType&qmuxStreamFrameOffsetBit != 0 { + _, offsetBytes, err := quicvarint.Parse(payload[headerEnd:]) + if err != nil { + return nil, err + } + headerEnd += offsetBytes + } + + adaptedPayload := quicvarint.Append(nil, frameType|qmuxStreamFrameLengthBit) + adaptedPayload = append(adaptedPayload, payload[frameTypeBytes:headerEnd]...) + adaptedPayload = quicvarint.Append(adaptedPayload, uint64(len(payload)-headerEnd)) + adaptedPayload = append(adaptedPayload, payload[headerEnd:]...) + return appendRecord(nil, adaptedPayload), nil +} + +func (conn *qmuxCompatConn) readInitialRecord() ([]byte, error) { + reader := quicvarint.NewReader(conn.Conn) + recordLength, err := quicvarint.Read(reader) + if err != nil { + return nil, err + } + payload := make([]byte, recordLength) + if _, err := io.ReadFull(reader, payload); err != nil { + return nil, err + } + + payloadReader := bytes.NewReader(payload) + frameType, err := quicvarint.Read(quicvarint.NewReader(payloadReader)) + if err != nil { + return nil, err + } + if frameType != qmuxTransportParametersFrameType { + return nil, fmt.Errorf("expected initial QX_TRANSPORT_PARAMETERS frame, got %#x", frameType) + } + parameterLength, err := quicvarint.Read(quicvarint.NewReader(payloadReader)) + if err != nil { + return nil, err + } + if parameterLength > uint64(payloadReader.Len()) { + return nil, fmt.Errorf("QMux transport parameters length %d exceeds record payload", parameterLength) + } + + parameterBytes := make([]byte, parameterLength) + if _, err := io.ReadFull(payloadReader, parameterBytes); err != nil { + return nil, err + } + transportParameters := quicvarint.Append(nil, frameType) + transportParameters = append(transportParameters, parameterBytes...) + adapted := appendRecord(nil, transportParameters) + if payloadReader.Len() > 0 { + remainingFrames := make([]byte, payloadReader.Len()) + if _, err := io.ReadFull(payloadReader, remainingFrames); err != nil { + return nil, err + } + adapted = appendRecord(adapted, remainingFrames) + } + return adapted, nil +} + +func (conn *qmuxCompatConn) Write(data []byte) (int, error) { + conn.writeMutex.Lock() + defer conn.writeMutex.Unlock() + + if conn.writeDone { + return conn.Conn.Write(data) + } + adapted, err := adaptInitialWrite(data) + if err != nil { + return 0, err + } + if err := writeAll(conn.Conn, adapted); err != nil { + return 0, err + } + conn.writeDone = true + return len(data), nil +} + +func adaptInitialWrite(data []byte) ([]byte, error) { + recordLength, recordLengthBytes, err := quicvarint.Parse(data) + if err != nil { + return nil, err + } + if recordLength > uint64(len(data)-recordLengthBytes) { + return nil, fmt.Errorf("incomplete initial QMux record") + } + payload := data[recordLengthBytes : recordLengthBytes+int(recordLength)] + frameType, frameTypeBytes, err := quicvarint.Parse(payload) + if err != nil { + return nil, err + } + if frameType != qmuxTransportParametersFrameType { + return nil, fmt.Errorf("expected initial QX_TRANSPORT_PARAMETERS frame, got %#x", frameType) + } + + parameters := payload[frameTypeBytes:] + adaptedPayload := quicvarint.Append(nil, frameType) + adaptedPayload = quicvarint.Append(adaptedPayload, uint64(len(parameters))) + adaptedPayload = append(adaptedPayload, parameters...) + adapted := appendRecord(nil, adaptedPayload) + return append(adapted, data[recordLengthBytes+int(recordLength):]...), nil +} + +func appendRecord(destination []byte, payload []byte) []byte { + destination = quicvarint.Append(destination, uint64(len(payload))) + return append(destination, payload...) +} + +func writeAll(writer io.Writer, data []byte) error { + for len(data) > 0 { + written, err := writer.Write(data) + if err != nil { + return err + } + if written == 0 { + return io.ErrShortWrite + } + data = data[written:] + } + return nil +} diff --git a/tests/gold_tests/qmux/qmux.replay.yaml b/tests/gold_tests/qmux/qmux.replay.yaml new file mode 100644 index 00000000000..d93ca9badb8 --- /dev/null +++ b/tests/gold_tests/qmux/qmux.replay.yaml @@ -0,0 +1,126 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# This is a server-only replay file. The Go client generates the downstream +# HTTP/3 requests, while Proxy Verifier validates ATS's origin requests and +# generates the origin responses. + +meta: + version: '1.0' + + blocks: + - request_base: &request_base + version: '1.1' + - empty_response: &empty_response + status: 200 + reason: OK + headers: + fields: + - [ Content-Length, '0' ] + - [ X-QMux-Response, success ] + content: + size: 0 + - generated_100_response: &generated_100_response + status: 200 + reason: OK + headers: + fields: + - [ Content-Type, application/octet-stream ] + - [ Content-Length, '100' ] + - [ X-QMux-Response, success ] + content: + size: 100 + - generated_300k_response: &generated_300k_response + status: 200 + reason: OK + headers: + fields: + - [ Content-Type, application/octet-stream ] + - [ Content-Length, '300000' ] + - [ X-QMux-Response, success ] + content: + size: 300000 + +sessions: +- transactions: + - client-request: + <<: *request_base + method: GET + url: /qmux-get-empty + headers: + fields: + - [ X-QMux-Client, qmux-go ] + - [ X-QMux-Test-Case, qmux-get-empty ] + - [ uuid, qmux-get-empty ] + + proxy-request: + headers: + fields: + - [ X-QMux-Client, { value: qmux-go, as: equal } ] + - [ X-QMux-Test-Case, { value: qmux-get-empty, as: equal } ] + + server-response: + <<: *empty_response + + - client-request: + <<: *request_base + method: POST + url: /qmux-post-small + headers: + fields: + - [ X-QMux-Client, qmux-go ] + - [ X-QMux-Test-Case, qmux-post-small ] + - [ Content-Type, application/octet-stream ] + - [ Content-Length, '100' ] + - [ uuid, qmux-post-small ] + content: + size: 100 + verify: { as: equal } + + proxy-request: + headers: + fields: + - [ X-QMux-Client, { value: qmux-go, as: equal } ] + - [ X-QMux-Test-Case, { value: qmux-post-small, as: equal } ] + - [ Content-Length, { value: '100', as: equal } ] + + server-response: + <<: *generated_100_response + + - client-request: + <<: *request_base + method: POST + url: /qmux-post-large + headers: + fields: + - [ X-QMux-Client, qmux-go ] + - [ X-QMux-Test-Case, qmux-post-large ] + - [ Content-Type, application/octet-stream ] + - [ Content-Length, '300000' ] + - [ uuid, qmux-post-large ] + content: + size: 300000 + verify: { as: equal } + + proxy-request: + headers: + fields: + - [ X-QMux-Client, { value: qmux-go, as: equal } ] + - [ X-QMux-Test-Case, { value: qmux-post-large, as: equal } ] + - [ Content-Length, { value: '300000', as: equal } ] + + server-response: + <<: *generated_300k_response diff --git a/tests/gold_tests/qmux/qmux_go_client.test.py b/tests/gold_tests/qmux/qmux_go_client.test.py new file mode 100644 index 00000000000..8a16a8619e6 --- /dev/null +++ b/tests/gold_tests/qmux/qmux_go_client.test.py @@ -0,0 +1,128 @@ +''' +Verify HTTP/3 over QMux interoperability with a Go client. +''' +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import os + +Test.Summary = '''Verify that a Go QMux client can complete HTTP/3 transactions through ATS.''' + +Test.SkipUnless( + Condition.HasATSFeature('TS_USE_QMUX'), + Condition.HasGoVersion('1.26'), +) + + +class TestQMuxGoClient: + '''Configure a Go client interoperability test for HTTP/3 over QMux.''' + + replay_file: str = 'qmux.replay.yaml' + + def __init__(self) -> None: + '''Configure the test run.''' + tr = Test.AddTestRun('Go HTTP/3 over QMux client request') + self._configure_server(tr) + self._configure_traffic_server(tr) + self._configure_client(tr) + + def _configure_server(self, tr: 'TestRun') -> 'Process': + '''Configure the Proxy Verifier origin server. + + :param tr: The TestRun to add the server process to. + :return: The server process. + ''' + server = tr.AddVerifierServerProcess('server', self.replay_file, verbose=False) + self._server = server + return server + + def _configure_traffic_server(self, tr: 'TestRun') -> 'Process': + '''Configure Traffic Server. + + :param tr: The TestRun to add the Traffic Server process to. + :return: The Traffic Server process. + ''' + ts = tr.MakeATSProcess('ts', enable_tls=True, enable_cache=False) + self._ts = ts + + ts.StartupTimeout = 60 + ts.addDefaultSSLFiles() + ts.Disk.ssl_multicert_yaml.AddLines( + ''' +ssl_multicert: + - dest_ip: "*" + ssl_cert_name: server.pem + ssl_key_name: server.key +'''.split('\n')) + ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'qmux|http3', + 'proxy.config.http.server_ports': (f'{ts.Variables.port} {ts.Variables.ssl_port}:ssl:proto=h3qx-01'), + 'proxy.config.ssl.server.cert.path': ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': ts.Variables.SSLDir, + }) + ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._server.Variables.http_port}') + ts.Disk.logging_yaml.AddLines( + ''' +logging: + formats: + - name: qmux_access + format: 'c_alpn=% client_version=% c_method=% c_url=%' + + logs: + - filename: qmux_access + format: qmux_access +'''.split('\n')) + + access_log = Test.Disk.File(os.path.join(ts.Variables.LOGDIR, 'qmux_access.log'), exists=True) + access_log.Content = Testers.ContainsExpression( + r'c_alpn=h3qx-01 client_version=http/3 c_method=GET ' + r'c_url=https://qmux\.example\.com:[0-9]+/qmux-get-empty', + 'ATS should log the empty QMux request as HTTP/3 over the h3qx-01 ALPN.') + access_log.Content += Testers.ContainsExpression( + r'c_alpn=h3qx-01 client_version=http/3 c_method=POST ' + r'c_url=https://qmux\.example\.com:[0-9]+/qmux-post-large', + 'ATS should log the large QMux request as HTTP/3 over the h3qx-01 ALPN.') + return ts + + def _configure_client(self, tr: 'TestRun') -> 'Process': + '''Configure the Go QMux client. + + :param tr: The TestRun to add the client process to. + :return: The client process. + ''' + tr.Setup.Copy('go_qmux_client') + client = tr.Processes.Default + client.Env['GOFLAGS'] = '-mod=readonly -modcacherw' + client.Env['GOCACHE'] = os.path.join(tr.RunDirectory, 'gocache') + client.Env['GOMODCACHE'] = os.path.join(tr.RunDirectory, 'gomodcache') + client.Env['GOTOOLCHAIN'] = 'local' + client.Command = ( + f'cd "{os.path.join(tr.RunDirectory, "go_qmux_client")}" && ' + f'go run . --addr 127.0.0.1:{self._ts.Variables.ssl_port} ' + f'--authority qmux.example.com:{self._ts.Variables.ssl_port} ' + '--server-name qmux.example.com') + client.ReturnCode = 0 + client.Streams.stdout = Testers.ContainsExpression( + 'completed 3 QMux HTTP/3 requests: alpn=h3qx-01', + 'The Go client should complete all HTTP/3 requests over one QMux session.') + client.StartBefore(self._server) + client.StartBefore(self._ts) + return client + + +TestQMuxGoClient()