diff --git a/clickhouse/client.cpp b/clickhouse/client.cpp index 05f03bcd..1bba7a9f 100644 --- a/clickhouse/client.cpp +++ b/clickhouse/client.cpp @@ -9,6 +9,7 @@ #include "columns/factory.h" #include +#include #include #include #include @@ -157,7 +158,6 @@ struct ProfileEvents { struct EndOfStream { }; using DecodedPacket = std::variant< - std::monostate, Block, ServerException, Profile, @@ -247,7 +247,17 @@ class Client::Impl { bool ReceiveData(Block & block); /// Reads exception packet form input stream. - bool ReceiveException(bool rethrow = false, ServerError * error = nullptr); + bool ReceiveException(ServerError & error); + + bool ReceiveProfileInfo(Profile & profile); + + bool ReceiveProgress(Progress & progress); + + bool ReceiveLog(Block & block); + + bool ReceiveTableColumns(); + + bool ReceiveProfileEvents(Block & block); void WriteBlock(const Block& block, OutputStream& output); @@ -386,7 +396,9 @@ std::optional Client::Impl::NextBlock() { return {std::move(block)}; } case VariantIndex(): - case VariantIndex(): + // Normally `ReceivePacket()` throws on server exceptions, but it can be + // suppressed by `ClientOptions::SetRethrowException`. In that case the + // error marks the end of the response. case VariantIndex(): ResetState(); return std::nullopt; @@ -674,7 +686,7 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { uint64_t packet_type = 0; if (!WireFormat::ReadVarint64(*input_, &packet_type)) { - return {}; + throw ProtocolError{"can't read packet type from input stream"}; } if (server_packet) { *server_packet = packet_type; @@ -684,76 +696,37 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { case ServerCodes::Data: { Block ret{}; if (!ReceiveData(ret)) { - throw ProtocolError("can't read data packet from input stream"); + throw ProtocolError{"can't read data packet from input stream"}; } return ret; } case ServerCodes::Exception: { ServerError ret{std::make_shared()}; - if (!ReceiveException(false, &ret)) { - throw ProtocolError("can't read exception packet from input stream"); + if (!ReceiveException(ret)) { + throw ProtocolError{"server reported an error, but the exception packet could not be decoded (error details lost)"}; + } + + if (options_.rethrow_exceptions) { + throw ret; } + return ret; } case ServerCodes::ProfileInfo: { Profile ret{}; - - if (!WireFormat::ReadUInt64(*input_, &ret.rows)) { - return {}; - } - if (!WireFormat::ReadUInt64(*input_, &ret.blocks)) { - return {}; - } - if (!WireFormat::ReadUInt64(*input_, &ret.bytes)) { - return {}; - } - if (!WireFormat::ReadFixed(*input_, &ret.applied_limit)) { - return {}; + if (!ReceiveProfileInfo(ret)) { + throw ProtocolError{"can't read profile info packet from input stream"}; } - if (!WireFormat::ReadUInt64(*input_, &ret.rows_before_limit)) { - return {}; - } - if (!WireFormat::ReadFixed(*input_, &ret.calculated_rows_before_limit)) { - return {}; - } - - if (events_) { - events_->OnProfile(ret); - } - return ret; } case ServerCodes::Progress: { Progress ret{}; - - if (!WireFormat::ReadUInt64(*input_, &ret.rows)) { - return {}; - } - if (!WireFormat::ReadUInt64(*input_, &ret.bytes)) { - return {}; - } - if constexpr(DMBS_PROTOCOL_REVISION >= DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS) { - if (!WireFormat::ReadUInt64(*input_, &ret.total_rows)) { - return {}; - } - } - if (server_info_.revision >= DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO) - { - if (!WireFormat::ReadUInt64(*input_, &ret.written_rows)) { - return {}; - } - if (!WireFormat::ReadUInt64(*input_, &ret.written_bytes)) { - return {}; - } - } - - if (events_) { - events_->OnProgress(ret); + if (!ReceiveProgress(ret)) { + throw ProtocolError{"can't read progress packet from input stream"}; } - return ret; } @@ -773,55 +746,30 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { } case ServerCodes::Log: { - // log tag - if (!WireFormat::SkipString(*input_)) { - return {}; - } Log ret; - - // Use uncompressed stream since log blocks usually contain only one row - if (!ReadBlock(*input_, &ret.block)) { - return {}; - } - - if (events_) { - events_->OnServerLog(ret.block); + if (!ReceiveLog(ret.block)) { + throw ProtocolError{"can't read log packet from input stream"}; } return ret; } case ServerCodes::TableColumns: { - // external table name - if (!WireFormat::SkipString(*input_)) { - return {}; - } - - // columns metadata - if (!WireFormat::SkipString(*input_)) { - return {}; + if (!ReceiveTableColumns()) { + throw ProtocolError{"can't read table columns packet from input stream"}; } return TableColumns{}; } case ServerCodes::ProfileEvents: { - if (!WireFormat::SkipString(*input_)) { - return {}; - } - ProfileEvents ret; - if (!ReadBlock(*input_, &ret.block)) { - return {}; - } - - if (events_) { - events_->OnProfileEvents(ret.block); + if (!ReceiveProfileEvents(ret.block)) { + throw ProtocolError{"can't read profile events packet from input stream"}; } return ret; } default: throw UnimplementedError("unimplemented " + std::to_string((int)packet_type)); - break; } } @@ -829,7 +777,9 @@ bool Client::Impl::ProcessPacket(uint64_t* server_packet) { auto packet = ReceivePacket(server_packet); switch (packet.index()) { case VariantIndex(): - case VariantIndex(): + // Normally `ReceivePacket()` throws on server exceptions, but it can be + // suppressed by `ClientOptions::SetRethrowException`. In that case the + // error marks the end of the response. case VariantIndex(): return false; default: @@ -946,29 +896,97 @@ bool Client::Impl::ReceiveData(Block & block) { return true; } -bool Client::Impl::ReceiveException(bool rethrow, ServerError * error) { +bool Client::Impl::ReceiveException(ServerError & error) { std::shared_ptr e(new Exception); bool has_nested = false; // obsolete: https://github.com/ClickHouse/ClickHouse/blob/ef11941cf5a/src/IO/ReadHelpers.cpp#L2017 - bool exception_received = - WireFormat::ReadFixed(*input_, &e->code) - && WireFormat::ReadString(*input_, &e->name) - && WireFormat::ReadString(*input_, &e->display_text) - && WireFormat::ReadString(*input_, &e->stack_trace) - && WireFormat::ReadFixed(*input_, &has_nested); + if (!WireFormat::ReadFixed(*input_, &e->code) || + !WireFormat::ReadString(*input_, &e->name) || + !WireFormat::ReadString(*input_, &e->display_text) || + !WireFormat::ReadString(*input_, &e->stack_trace) || + !WireFormat::ReadFixed(*input_, &has_nested)) { + return false; + } if (events_) { events_->OnServerException(*e); } - if (rethrow || options_.rethrow_exceptions) { - throw ServerError(e); + error = ServerError(e); + + return true; +} + +bool Client::Impl::ReceiveProfileInfo(Profile & profile) { + if (!WireFormat::ReadUInt64(*input_, &profile.rows) || + !WireFormat::ReadUInt64(*input_, &profile.blocks) || + !WireFormat::ReadUInt64(*input_, &profile.bytes) || + !WireFormat::ReadFixed(*input_, &profile.applied_limit) || + !WireFormat::ReadUInt64(*input_, &profile.rows_before_limit) || + !WireFormat::ReadFixed(*input_, &profile.calculated_rows_before_limit)) { + return false; } - if (exception_received && error != nullptr) { - *error = ServerError(e); + if (events_) { + events_->OnProfile(profile); } - return exception_received; + + return true; +} + +bool Client::Impl::ReceiveProgress(Progress & progress) { + if (!WireFormat::ReadUInt64(*input_, &progress.rows) || + !WireFormat::ReadUInt64(*input_, &progress.bytes)) { + return false; + } + + if constexpr(DMBS_PROTOCOL_REVISION >= DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS) { + if (!WireFormat::ReadUInt64(*input_, &progress.total_rows)) { + return false; + } + } + if (server_info_.revision >= DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO) + { + if (!WireFormat::ReadUInt64(*input_, &progress.written_rows) || + !WireFormat::ReadUInt64(*input_, &progress.written_bytes)) { + return false; + } + } + + if (events_) { + events_->OnProgress(progress); + } + + return true; +} + +bool Client::Impl::ReceiveLog(Block & block) { + if (!WireFormat::SkipString(*input_) || // log tag + !ReadBlock(*input_, &block)) { // Use uncompressed stream since log blocks usually contain only one row + return false; + } + if (events_) { + events_->OnServerLog(block); + } + return true; +} + +bool Client::Impl::ReceiveTableColumns() { + if (!WireFormat::SkipString(*input_) || // external table name + !WireFormat::SkipString(*input_)) { // columns metadata + return false; + } + return true; +} + +bool Client::Impl::ReceiveProfileEvents(Block & block) { + if (!WireFormat::SkipString(*input_) || !ReadBlock(*input_, &block)) { + return false; + } + if (events_) { + events_->OnProfileEvents(block); + } + return true; } void Client::Impl::SendCancel() { @@ -1213,8 +1231,11 @@ bool Client::Impl::ReceiveHello() { return true; } else if (packet_type == ServerCodes::Exception) { - ReceiveException(true); - return false; + ServerError ret{std::make_shared()}; + if (!ReceiveException(ret)) { + throw ProtocolError{"server rejected the connection, but its exception packet could not be decoded"}; + } + throw ret; } return false; diff --git a/ut/CMakeLists.txt b/ut/CMakeLists.txt index c827e5e3..b830ae17 100644 --- a/ut/CMakeLists.txt +++ b/ut/CMakeLists.txt @@ -5,6 +5,7 @@ SET ( clickhouse-cpp-ut-src bignum_ut.cpp bignum_round_trip.cpp block_ut.cpp + client_protocol_ut.cpp client_ut.cpp columns_ut.cpp column_as_ut.cpp diff --git a/ut/client_protocol_ut.cpp b/ut/client_protocol_ut.cpp new file mode 100644 index 00000000..72cfdd5d --- /dev/null +++ b/ut/client_protocol_ut.cpp @@ -0,0 +1,240 @@ +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include + +namespace { + +using namespace clickhouse; + +/// A socket that serves a pre-recorded byte script as the server response +/// and discards (but keeps) everything written by the client. +class ScriptedSocket : public SocketBase { +public: + explicit ScriptedSocket(std::vector script) + : script_(std::move(script)) + {} + + std::unique_ptr makeInputStream() const override { + return std::make_unique(script_.data(), script_.size()); + } + + std::unique_ptr makeOutputStream() const override { + return std::make_unique(&written_); + } + +private: + const std::vector script_; + mutable std::vector written_; +}; + +class ScriptedSocketFactory : public SocketFactory { +public: + explicit ScriptedSocketFactory(std::vector script) + : script_(std::move(script)) + {} + + std::unique_ptr connect(const ClientOptions&, const Endpoint&) override { + return std::make_unique(script_); + } + +private: + std::vector script_; +}; + +ClientOptions ScriptedClientOptions() { + return ClientOptions() + .SetHost("scripted.test") + .SetPingBeforeQuery(false) + .SetSendRetries(0); +} + +/// A minimal well-formed ServerHello. Revision 50000 is below +/// DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE (54058), so the server sends only +/// name, version_major, version_minor and revision. +const std::vector kServerHello = { + 0x00, // packet type: ServerCodes::Hello (varint) + 0x02, 0x43, 0x48, // server name: "CH" (string, len 2) + 0x01, // version_major = 1 (varint) + 0x01, // version_minor = 1 (varint) + 0xD0, 0x86, 0x03, // revision = 50000 (varint) +}; + +/// An exception packet with a valid code but a corrupt `name` string. +const std::vector kCorruptExceptionPacket = { + 0x02, // packet type: ServerCodes::Exception (varint) + 0x3C, 0x00, 0x00, 0x00, // code = 60 (int32, little-endian) + 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, // name string length = 0xFFFFFFFF (varint), + // exceeds the 0x00FFFFFF limit in WireFormat::ReadString +}; + +/// A complete well-formed exception packet. +const std::vector kWellFormedExceptionPacket = { + 0x02, // packet type: ServerCodes::Exception (varint) + 0x3C, 0x00, 0x00, 0x00, // code = 60 (int32, little-endian) + 0x02, 0x44, 0x42, // name: "DB" (string, len 2) + 0x04, 0x6F, 0x6F, 0x70, 0x73, // display_text: "oops" (string, len 4) + 0x00, // stack_trace: "" (string, len 0) + 0x00, // has_nested = false (fixed, 1 byte) +}; + +/// Performs the handshake against kServerHello, then runs a Select that +/// receives `response` and expects a ProtocolError containing +/// `expected_message`. +void ExpectSelectThrowsProtocolError(const std::vector& response, + const std::string& expected_message) { + std::vector script = kServerHello; + script.insert(script.end(), response.begin(), response.end()); + // Trailing garbage: decoding must fail on validation, not on end-of-stream. + script.insert(script.end(), 64, 0xAA); + + // The handshake must succeed; only Select is expected to throw. + Client client(ScriptedClientOptions(), + std::make_unique(script)); + + try { + client.Select("SELECT 1", [](const Block&) {}); + FAIL() << "expected ProtocolError"; + } catch (const ProtocolError& e) { + EXPECT_TRUE(std::string(e.what()).find(expected_message) != std::string::npos) + << "unexpected message: " << e.what(); + } +} + +} + +TEST(ClientProtocol, MalformedServerHelloThrowsProtocolError) { + std::vector script = { + 0x00, // packet type: ServerCodes::Hello (varint) + 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, // server-name string length = 0xFFFFFFFF (varint), + // exceeds the 0x00FFFFFF limit in WireFormat::ReadString + }; + + // Trailing garbage: decoding must fail on validation, not on end-of-stream. + script.insert(script.end(), 64, 0xAA); + + try { + Client client(ScriptedClientOptions(), + std::make_unique(script)); + FAIL() << "expected ProtocolError"; + } catch (const ProtocolError& e) { + EXPECT_TRUE(std::string(e.what()).find("fail to connect") != std::string::npos) + << "unexpected message: " << e.what(); + } +} + +TEST(ClientProtocol, MalformedSelectResponseThrowsProtocolError) { + // Malformed packet type: a varint whose continuation bit never clears, + // so ReadVarint64 gives up after MAX_VARINT_BYTES. + const std::vector response(10, 0x80); + ExpectSelectThrowsProtocolError(response, "can't read packet type"); +} + +TEST(ClientProtocol, CorruptProgressPacketThrowsProtocolError) { + std::vector response = { + 0x03, // packet type: ServerCodes::Progress (varint) + }; + // Malformed `rows` varint: the continuation bit never clears. + response.insert(response.end(), 10, 0x80); + ExpectSelectThrowsProtocolError(response, "can't read progress packet"); +} + +TEST(ClientProtocol, CorruptProfileInfoPacketThrowsProtocolError) { + std::vector response = { + 0x06, // packet type: ServerCodes::ProfileInfo (varint) + }; + // Malformed `rows` varint: the continuation bit never clears. + response.insert(response.end(), 10, 0x80); + ExpectSelectThrowsProtocolError(response, "can't read profile info packet"); +} + +TEST(ClientProtocol, CorruptLogPacketThrowsProtocolError) { + const std::vector response = { + 0x0A, // packet type: ServerCodes::Log (varint) + 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, // log-tag string length = 0xFFFFFFFF (varint), + // exceeds the 0x00FFFFFF limit in WireFormat::SkipString + }; + ExpectSelectThrowsProtocolError(response, "can't read log packet"); +} + +TEST(ClientProtocol, CorruptExceptionPacketInSelectThrowsProtocolError) { + ExpectSelectThrowsProtocolError(kCorruptExceptionPacket, "error details lost"); +} + +TEST(ClientProtocol, CorruptExceptionPacketInHandshakeThrowsProtocolError) { + // The exception packet arrives instead of ServerHello. + std::vector script = kCorruptExceptionPacket; + // Trailing garbage: decoding must fail on validation, not on end-of-stream. + script.insert(script.end(), 64, 0xAA); + + try { + Client client(ScriptedClientOptions(), + std::make_unique(script)); + FAIL() << "expected ProtocolError"; + } catch (const ProtocolError& e) { + EXPECT_TRUE(std::string(e.what()).find("exception packet could not be decoded") != std::string::npos) + << "unexpected message: " << e.what(); + } +} + +TEST(ClientProtocol, ExceptionPacketInSelectThrowsServerException) { + std::vector script = kServerHello; + script.insert(script.end(), + kWellFormedExceptionPacket.begin(), kWellFormedExceptionPacket.end()); + + // The handshake must succeed; only Select is expected to throw. + Client client(ScriptedClientOptions(), + std::make_unique(script)); + + try { + client.Select("SELECT 1", [](const Block&) {}); + FAIL() << "expected ServerException"; + } catch (const ServerException& e) { + EXPECT_EQ(e.GetCode(), 60); + EXPECT_EQ(e.GetException().name, "DB"); + EXPECT_STREQ(e.what(), "oops"); + } +} + +TEST(ClientProtocol, ExceptionPacketInHandshakeThrowsServerException) { + // The exception packet arrives instead of ServerHello. + try { + Client client(ScriptedClientOptions(), + std::make_unique(kWellFormedExceptionPacket)); + FAIL() << "expected ServerException"; + } catch (const ServerException& e) { + EXPECT_EQ(e.GetCode(), 60); + EXPECT_EQ(e.GetException().name, "DB"); + EXPECT_STREQ(e.what(), "oops"); + } +} + +TEST(ClientProtocol, WellFormedSelectResponseSucceeds) { + std::vector script = kServerHello; + const std::vector response = { + 0x01, // packet type: ServerCodes::Data (varint) + // (revision 50000 has neither temp-table name nor BlockInfo) + 0x00, // num_columns = 0 (varint) + 0x00, // num_rows = 0 (varint) + 0x05, // packet type: ServerCodes::EndOfStream (varint) + }; + script.insert(script.end(), response.begin(), response.end()); + + Client client(ScriptedClientOptions(), + std::make_unique(script)); + + size_t blocks = 0; + EXPECT_NO_THROW(client.Select("SELECT 1", [&blocks](const Block& block) { + ++blocks; + EXPECT_EQ(block.GetRowCount(), 0u); + })); + EXPECT_EQ(blocks, 1u); +}