diff --git a/docs/en/antalya/protocol.md b/docs/en/antalya/protocol.md new file mode 100644 index 000000000000..78065bb47afc --- /dev/null +++ b/docs/en/antalya/protocol.md @@ -0,0 +1,91 @@ +--- +description: 'How the Antalya fork versions its own wire-protocol changes independently of upstream ClickHouse' +sidebar_label: 'Antalya Protocol Version' +sidebar_position: 40 +slug: /antalya/protocol +title: 'Antalya Protocol Version' +doc_type: 'reference' +--- + +# Antalya protocol version {#antalya-protocol-version} + +Antalya versions its own wire-protocol changes with `DBMS_ANTALYA_PROTOCOL_VERSION`, a counter that +upstream ClickHouse cannot reach, defined in `src/Core/AntalyaProtocol.h`. A server advertises it in +the `ServerHello` name string, on every connection: + +```text +server -> client "ClickHouse (antalya:1)" +``` + +The client strips the suffix, caps the value with `min(own, server)` and keeps the result. `0` means +the peer is not an Antalya build. Negotiation is per hop and not transitive: initiator to worker and +worker to worker negotiate independently. + +Version 1 is the advertisement itself. Nothing is gated on it yet. + +## Adding an Antalya-only wire change {#adding-a-wire-change} + +- Bump `DBMS_ANTALYA_PROTOCOL_VERSION` by one and gate the change on the negotiated value. +- Never bump `DBMS_TCP_PROTOCOL_VERSION`, and never take a slot in + `DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION` for a feature upstream does not have. +- Keep the counter cumulative. A backport takes the whole contiguous range up to the value it needs, + or does not bump at all - the `min(own, server)` cap is only sound for a cumulative feature set. +- Gate only what the *client* decides to do. The server never learns the client's version, because + only the server advertises. +- Document it here. Also document it in `docs/en/interfaces/specs/NativeProtocol.md` if it changes + the layout of a packet that file describes field by field: a third-party client (`ch-go`, + `clickhouse-go`, `clickhouse-driver`) that does not know an added field cannot parse the stream + past it. Everything else stays here, so that the spec keeps tracking upstream through rebases. + +## Why a counter of our own {#why-a-counter-of-our-own} + +An Antalya-only change that takes a slot in an upstream counter has to be renumbered whenever +upstream claims that slot for something else. The same number then means two different things in two +shipped builds, and two nodes that negotiate it disagree about the bytes on the wire. + +The constant lives in its own header rather than in `src/Core/ProtocolDefines.h`, whose tail is +where upstream adds its own constants, and therefore where every rebase conflicts. + +## Why the marker rides in `ServerHello` {#why-the-marker-rides-in-serverhello} + +**The client `Hello` is never marked.** The client writes that packet before it has read a byte from +the peer, so it cannot gate a marker on what the peer is. Every field in it - `client_name`, +`default_database`, `user`, `password` - is one an upstream server acts on. `client_name` is the +worst: a server persists it, and `validate_tcp_client_information` compares it against the Query +packet's `ClientInfo`, so a marker there fails a `remote()` query with `CLIENT_INFO_DOES_NOT_MATCH` +on any peer that does not strip it. + +`ServerHello` has no such field. `server_name` is client-side display text and reaches no system +table, which is why the server is the side that speaks. + +**The marker is a suffix, not extra bytes.** Neither Hello nor the Addendum has a length prefix or a +terminator: a reader stops after the last field its negotiated revision knows about, so a peer that +does not expect extra bytes reads them as the next packet. + +## What you see {#what-you-see} + +An Antalya client appends the negotiated version to its existing connection log line +(`Connected to ... server version ...`). The server logs nothing, and degradation to `0` is silent. + +Nothing a server stores changes: `system.query_log` and `system.processes` keep the `client_name` +the peer sent, so filters written as `client_name = 'ClickHouse server'` keep working. + +The suffix is visible in one place - an **upstream** client's `server_name`, which that client does +not strip, so its banner reads `Connected to ClickHouse (antalya:1) server version ...`. + +## No opt-out {#no-opt-out} + +No setting suppresses the marker: a node either speaks this protocol or is not an Antalya build. + +`server_name` is already a build-time value - upstream sets it from the CMake project name, and any +fork may set it to anything - so a native client cannot treat it as a fixed string. What a switch +would protect is the displayed name on a peer that does not strip the suffix, which is cosmetic, and +it would cost a state in which two Antalya nodes that both support a feature fail to negotiate it +because one was configured not to say so. If a peer is ever found that breaks on the suffix, the fix +is to stop appending it. + +## Implementation {#implementation} + +`src/Core/AntalyaProtocol.h` holds the version constant, the marker grammar and the `appendMarker` / +`parseMarker` / `stripMarker` / `negotiate` helpers. `TCPHandler::sendHello` appends the marker; +`Connection::receiveHello` strips it and stores the capped value. diff --git a/docs/en/interfaces/specs/NativeProtocol.md b/docs/en/interfaces/specs/NativeProtocol.md index 34b9809b1328..02eb626f2598 100644 --- a/docs/en/interfaces/specs/NativeProtocol.md +++ b/docs/en/interfaces/specs/NativeProtocol.md @@ -440,7 +440,7 @@ Server → Client. The reply to ClientHello on successful authentication. | # | Field | Type | Role | Condition | Description | |---|------------------|---------|-----------|------------------------|-------------| -| 1 | server_name | String | universal | always | Server identifier | +| 1 | server_name | String | universal | always | Server identifier. An Altinity Antalya build appends `" (antalya:N)"`, where `N` is its Antalya protocol version; a client may ignore or strip the suffix. See [Antalya protocol version](/antalya/protocol). | | 2 | version_major | VarUInt | universal | always | Server major version | | 3 | version_minor | VarUInt | universal | always | Server minor version | | 4 | protocol_version | VarUInt | universal | always | Server's protocol version | diff --git a/src/Client/Connection.cpp b/src/Client/Connection.cpp index 746c1bc0b8cc..e732e9071ced 100644 --- a/src/Client/Connection.cpp +++ b/src/Client/Connection.cpp @@ -26,7 +26,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -369,8 +371,11 @@ void Connection::connect(const ConnectionTimeouts & timeouts) if (proto_recv_chunked == "chunked") in->enableChunked(); - LOG_TRACE(log_wrapper.get(), "Connected to {} server version {}.{}.{}.", - server_name, server_version_major, server_version_minor, server_version_patch); + LOG_TRACE(log_wrapper.get(), "Connected to {} server version {}.{}.{}{}.", + server_name, server_version_major, server_version_minor, server_version_patch, + (server_antalya_protocol_version > 0 + ? ", Antalya protocol: " + std::to_string(server_antalya_protocol_version) + : "")); /// Now that the handshake is complete, use the regular timeouts socket->setReceiveTimeout(timeouts.receive_timeout); @@ -622,6 +627,7 @@ void Connection::receiveHello() { readStringBinary(server_name, *in, DBMS_MAX_HELLO_STRING_SIZE); sanitizeUntrustedServerString(server_name); + server_antalya_protocol_version = AntalyaProtocol::negotiate(AntalyaProtocol::stripMarker(server_name)); readVarUInt(server_version_major, *in); readVarUInt(server_version_minor, *in); readVarUInt(server_revision, *in); diff --git a/src/Client/Connection.h b/src/Client/Connection.h index c9d64c55c3be..d2a4a5504a35 100644 --- a/src/Client/Connection.h +++ b/src/Client/Connection.h @@ -238,6 +238,7 @@ class Connection : public IServerConnection UInt64 server_parallel_replicas_protocol_version = 0; UInt64 worker_cluster_function_protocol_version = 0; UInt64 server_query_plan_serialization_version = 0; + UInt64 server_antalya_protocol_version = 0; String server_timezone; String server_display_name; SettingsChanges settings_from_server; diff --git a/src/Client/tests/gtest_sanitize_untrusted_server_string.cpp b/src/Client/tests/gtest_sanitize_untrusted_server_string.cpp index 195202691eb5..cad0530b94f2 100644 --- a/src/Client/tests/gtest_sanitize_untrusted_server_string.cpp +++ b/src/Client/tests/gtest_sanitize_untrusted_server_string.cpp @@ -1,6 +1,7 @@ #include #include +#include #include using namespace DB; @@ -155,3 +156,21 @@ TEST(SanitizeUntrustedServerString, MaxPasswordComplexityRulesCapIsTight) EXPECT_LE(DBMS_MAX_PASSWORD_COMPLEXITY_RULES, 4096u); EXPECT_GE(DBMS_MAX_PASSWORD_COMPLEXITY_RULES, 16u); } + +TEST(SanitizeUntrustedServerString, PreservesAnAntalyaProtocolMarker) +{ + /// The client parses the marker only after sanitizing, so it must pass through byte for byte. + String name = AntalyaProtocol::appendMarker("ClickHouse"); + const String before = name; + sanitizeUntrustedServerString(name); + EXPECT_EQ(name, before); + EXPECT_EQ(AntalyaProtocol::parseMarker(name), static_cast(DBMS_ANTALYA_PROTOCOL_VERSION)); +} + +TEST(SanitizeUntrustedServerString, CannotForgeAnAntalyaProtocolMarker) +{ + /// Control bytes are replaced, never deleted, so a hostile name cannot collapse into a marker. + String hostile = "ClickHouse\x01 (antalya:\x02" "1)"; + sanitizeUntrustedServerString(hostile); + EXPECT_EQ(AntalyaProtocol::parseMarker(hostile), 0u); +} diff --git a/src/Core/AntalyaProtocol.cpp b/src/Core/AntalyaProtocol.cpp new file mode 100644 index 000000000000..f20948e07d98 --- /dev/null +++ b/src/Core/AntalyaProtocol.cpp @@ -0,0 +1,81 @@ +#include + +#include +#include + +#include + + +namespace DB +{ + +namespace AntalyaProtocol +{ + +constexpr UInt64 MAX_VERSION = intExp10(static_cast(MAX_MARKER_DIGITS)) - 1; + +static_assert( + DBMS_ANTALYA_PROTOCOL_VERSION >= 1 && DBMS_ANTALYA_PROTOCOL_VERSION <= MAX_VERSION, + "DBMS_ANTALYA_PROTOCOL_VERSION does not fit the marker grammar"); + +String appendMarker(std::string_view name) +{ + String result; + result.reserve(name.size() + MAX_MARKER_SIZE); + result.append(name); + result.append(MARKER_PREFIX); + result.append(std::to_string(DBMS_ANTALYA_PROTOCOL_VERSION)); + result.push_back(MARKER_TERMINATOR); + return result; +} + +UInt64 parseMarker(std::string_view name) noexcept +{ + if (name.empty() || name.back() != MARKER_TERMINATOR) + return 0; + + const size_t close = name.size() - 1; + + size_t first_digit = close; + while (first_digit > 0 && isNumericASCII(name[first_digit - 1])) + { + --first_digit; + if (close - first_digit > MAX_MARKER_DIGITS) + return 0; + } + + const size_t digits = close - first_digit; + /// Exactly one spelling per version: no empty digit run, no leading zero. + if (digits == 0 || name[first_digit] == '0') + return 0; + + if (first_digit < MARKER_PREFIX.size()) + return 0; + if (name.substr(first_digit - MARKER_PREFIX.size(), MARKER_PREFIX.size()) != MARKER_PREFIX) + return 0; + + UInt64 version = 0; + for (size_t i = first_digit; i < close; ++i) + version = version * 10 + static_cast(name[i] - '0'); + return version; +} + +UInt64 stripMarker(String & name) +{ + const UInt64 version = parseMarker(name); + if (version != 0) + { + /// Only the canonical spelling parses, so the digits on the wire are the ones `to_string` gives. + name.resize(name.size() - MARKER_PREFIX.size() - std::to_string(version).size() - 1); + } + return version; +} + +UInt64 negotiate(UInt64 peer_version) noexcept +{ + return std::min(peer_version, DBMS_ANTALYA_PROTOCOL_VERSION); +} + +} + +} diff --git a/src/Core/AntalyaProtocol.h b/src/Core/AntalyaProtocol.h new file mode 100644 index 000000000000..7f1c6a5044c8 --- /dev/null +++ b/src/Core/AntalyaProtocol.h @@ -0,0 +1,46 @@ +#pragma once + +#include + +#include + + +namespace DB +{ + +/// Counter for Antalya-only wire features, independent of every upstream counter in +/// `Core/ProtocolDefines.h`, so that a rebase can never renumber an Antalya feature. Bump it by one +/// per feature. The client caps the server's value with `min(own, server)`, so the feature set must +/// stay cumulative: a backport takes the whole contiguous range up to the value it needs, or does +/// not bump at all. See `docs/en/antalya/protocol.md`. +static constexpr auto DBMS_ANTALYA_PROTOCOL_VERSION = 1; + +namespace AntalyaProtocol +{ + +/// The version is advertised in the `ServerHello` name string - "ClickHouse (antalya:M)" - and +/// nowhere else: the client `Hello` is never marked, because `client_name` reaches the peer's +/// `system.query_log` and `validate_tcp_client_information`. See `docs/en/antalya/protocol.md`. + +/// " (antalya:" ")", where is [1-9][0-9]{0,8}. +inline constexpr std::string_view MARKER_PREFIX = " (antalya:"; +inline constexpr char MARKER_TERMINATOR = ')'; +inline constexpr size_t MAX_MARKER_DIGITS = 9; +inline constexpr size_t MAX_MARKER_SIZE = MARKER_PREFIX.size() + MAX_MARKER_DIGITS + 1; + +String appendMarker(std::string_view name); + +/// Returns 0 unless `name` ends with exactly one canonical marker. Scans at most `MAX_MARKER_SIZE` +/// bytes back from the end, because the client runs it on a `server_name` it has not authenticated. +UInt64 parseMarker(std::string_view name) noexcept; + +/// Removes a trailing canonical marker from `name` and returns the version it spelled; returns 0 and +/// leaves `name` alone when there is none. +UInt64 stripMarker(String & name); + +/// The version to speak with a server that advertised `peer_version`; 0 means it is not Antalya. +UInt64 negotiate(UInt64 peer_version) noexcept; + +} + +} diff --git a/src/Core/Protocol.h b/src/Core/Protocol.h index 3f58e654d9ea..2f8dde808e6e 100644 --- a/src/Core/Protocol.h +++ b/src/Core/Protocol.h @@ -68,6 +68,9 @@ const char JWT_AUTHENTICAION_MARKER[] = " JWT AUTHENTICATION "; }; +/// `client_name` for server-to-server connections, the counterpart of `DEFAULT_CLIENT_NAME`. +static constexpr std::string_view SERVER_CLIENT_NAME = "server"; + namespace Protocol { /// Packet types that server transmits. diff --git a/src/Core/tests/gtest_antalya_protocol.cpp b/src/Core/tests/gtest_antalya_protocol.cpp new file mode 100644 index 000000000000..a9bfefdc0fe4 --- /dev/null +++ b/src/Core/tests/gtest_antalya_protocol.cpp @@ -0,0 +1,76 @@ +#include + +#include + +using namespace DB; +using namespace DB::AntalyaProtocol; + + +TEST(AntalyaProtocol, AppendMarkerSpellsTheWireForm) +{ + EXPECT_EQ( + appendMarker("ClickHouse server"), + "ClickHouse server (antalya:" + std::to_string(DBMS_ANTALYA_PROTOCOL_VERSION) + ")"); +} + +TEST(AntalyaProtocol, AppendThenParseRoundTrips) +{ + EXPECT_EQ(parseMarker(appendMarker("ClickHouse server")), static_cast(DBMS_ANTALYA_PROTOCOL_VERSION)); +} + +TEST(AntalyaProtocol, ParseRejectsNonCanonicalMarkers) +{ + const std::string_view rejected[] = { + "", + "ClickHouse server", + "ClickHouse client", + "(antalya:1)", + "ClickHouse server(antalya:1)", + "ClickHouse server (antalya:1", + "ClickHouse server (antalya:)", + "ClickHouse server (antalya:0)", + "ClickHouse server (antalya:01)", + "ClickHouse server (antalya:1234567890)", + "ClickHouse server (antalya:1) v2", + "ClickHouse server (ANTALYA:1)", + "ClickHouse server (antalya:1x)", + "ClickHouse server (antalya:1 )", + "antalya:1)", + ")", + }; + + for (const auto & name : rejected) + EXPECT_EQ(parseMarker(name), 0u) << "should not have parsed: " << name; +} + +TEST(AntalyaProtocol, ParseTakesTheTrailingMarkerWhenRepeated) +{ + EXPECT_EQ(parseMarker("ClickHouse server (antalya:99) (antalya:1)"), 1u); +} + +TEST(AntalyaProtocol, StripMarkerRemovesOnlyTheMarker) +{ + String marked = "ClickHouse server (antalya:42)"; + EXPECT_EQ(stripMarker(marked), 42u); + EXPECT_EQ(marked, "ClickHouse server"); + + String widest = "ClickHouse (antalya:999999999)"; + EXPECT_EQ(stripMarker(widest), 999999999u); + EXPECT_EQ(widest, "ClickHouse"); + + String plain = "ClickHouse server"; + EXPECT_EQ(stripMarker(plain), 0u); + EXPECT_EQ(plain, "ClickHouse server"); + + String malformed = "ClickHouse server (antalya:01)"; + EXPECT_EQ(stripMarker(malformed), 0u); + EXPECT_EQ(malformed, "ClickHouse server (antalya:01)"); +} + +TEST(AntalyaProtocol, NegotiateCapsToOurVersion) +{ + EXPECT_EQ(negotiate(0), 0u); + EXPECT_EQ(negotiate(DBMS_ANTALYA_PROTOCOL_VERSION), static_cast(DBMS_ANTALYA_PROTOCOL_VERSION)); + EXPECT_EQ(negotiate(DBMS_ANTALYA_PROTOCOL_VERSION + 1), static_cast(DBMS_ANTALYA_PROTOCOL_VERSION)); + EXPECT_EQ(negotiate(999999999), static_cast(DBMS_ANTALYA_PROTOCOL_VERSION)); +} diff --git a/src/Interpreters/Cluster.cpp b/src/Interpreters/Cluster.cpp index f4fc30f9e8c7..613cf81f9b87 100644 --- a/src/Interpreters/Cluster.cpp +++ b/src/Interpreters/Cluster.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -516,7 +517,7 @@ Cluster::Cluster(const Poco::Util::AbstractConfiguration & config, address.quota_key, address.cluster, address.cluster_secret, - "server", + String(SERVER_CLIENT_NAME), address.compression, address.secure, address.bind_host, @@ -676,7 +677,7 @@ void Cluster::addShard( replica.quota_key, replica.cluster, replica.cluster_secret, - "server", + String(SERVER_CLIENT_NAME), replica.compression, replica.secure, replica.bind_host, @@ -845,7 +846,7 @@ Cluster::Cluster(Cluster::ReplicasAsShardsTag, const Cluster & from, const Setti address.quota_key, address.cluster, address.cluster_secret, - "server", + String(SERVER_CLIENT_NAME), address.compression, address.secure, address.bind_host, diff --git a/src/Server/TCPHandler.cpp b/src/Server/TCPHandler.cpp index 24e4eefff641..c85dc0847448 100644 --- a/src/Server/TCPHandler.cpp +++ b/src/Server/TCPHandler.cpp @@ -10,6 +10,7 @@ #include #include #include +#include #include #include #include @@ -2136,7 +2137,7 @@ void TCPHandler::processUnexpectedHello() void TCPHandler::sendHello() { writeVarUInt(Protocol::Server::Hello, *out); - writeStringBinary(VERSION_NAME, *out); + writeStringBinary(AntalyaProtocol::appendMarker(VERSION_NAME), *out); writeVarUInt(VERSION_MAJOR, *out); writeVarUInt(VERSION_MINOR, *out); writeVarUInt(DBMS_TCP_PROTOCOL_VERSION, *out); diff --git a/tests/integration/test_antalya_protocol/__init__.py b/tests/integration/test_antalya_protocol/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/integration/test_antalya_protocol/test.py b/tests/integration/test_antalya_protocol/test.py new file mode 100644 index 000000000000..8f43bdb89667 --- /dev/null +++ b/tests/integration/test_antalya_protocol/test.py @@ -0,0 +1,74 @@ +"""Antalya protocol version negotiation over the native TCP handshake. + +The log assertions count occurrences before and after the query because the cluster fixture is +module-scoped, so a plain substring check would pass on an earlier test's output. +""" + +import pytest + +from helpers.cluster import CLICKHOUSE_CI_MIN_TESTED_VERSION, ClickHouseCluster + +cluster = ClickHouseCluster(__file__) + +node1 = cluster.add_instance("node1") +node2 = cluster.add_instance("node2") +# A build from before the marker existed: it sends an unmarked `ServerHello` and does not know to +# strip one. +node_old = cluster.add_instance( + "node_old", + image="altinity/clickhouse-server", + tag=CLICKHOUSE_CI_MIN_TESTED_VERSION, + with_installed_binary=True, +) + +NEGOTIATED = "Antalya protocol: " + + +@pytest.fixture(scope="module") +def started_cluster(): + try: + cluster.start() + yield cluster + finally: + cluster.shutdown() + + +def count_in_log(node, substring): + return int(node.count_in_log(substring)) + + +def test_remote_function_negotiates(started_cluster): + """Only the initiator logs the version: the worker learns nothing about the peer.""" + initiator_before = count_in_log(node1, NEGOTIATED) + worker_before = count_in_log(node2, NEGOTIATED) + + assert node1.query("SELECT count() FROM remote('node2', system.one)") == "1\n" + + assert count_in_log(node1, NEGOTIATED) > initiator_before + assert count_in_log(node2, NEGOTIATED) == worker_before + + +def test_new_initiator_against_an_unmarked_worker(started_cluster): + """A worker that predates the marker neither strips nor expects one.""" + assert ( + node1.query( + "SELECT count() FROM remote('node_old', numbers(10))" + " SETTINGS log_comment = 'antalya_marker_old_worker'" + ) + == "10\n" + ) + + node_old.query("SYSTEM FLUSH LOGS") + assert ( + node_old.query( + "SELECT DISTINCT client_name FROM system.query_log" + " WHERE log_comment = 'antalya_marker_old_worker' AND type = 'QueryFinish'" + " AND is_initial_query" + ) + == "ClickHouse server\n" + ) + + +def test_unmarked_initiator_against_a_marked_server(started_cluster): + """An old client gets a marked `ServerHello` it does not know to strip, and only displays it.""" + assert node_old.query("SELECT count() FROM remote('node1', numbers(10))") == "10\n" diff --git a/tests/queries/0_stateless/05053_antalya_protocol_marker.reference b/tests/queries/0_stateless/05053_antalya_protocol_marker.reference new file mode 100644 index 000000000000..231e92f406e6 --- /dev/null +++ b/tests/queries/0_stateless/05053_antalya_protocol_marker.reference @@ -0,0 +1,3 @@ +10 +ClickHouse client +ClickHouse server diff --git a/tests/queries/0_stateless/05053_antalya_protocol_marker.sql b/tests/queries/0_stateless/05053_antalya_protocol_marker.sql new file mode 100644 index 000000000000..df3a18612973 --- /dev/null +++ b/tests/queries/0_stateless/05053_antalya_protocol_marker.sql @@ -0,0 +1,26 @@ +-- No Antalya protocol string may reach `system.query_log`: only the `ServerHello` carries the +-- marker, and the client strips it. See `Core/AntalyaProtocol.h`. + +SELECT count() FROM remote('127.0.0.2', numbers(10)) +SETTINGS log_queries = 1, log_comment = '05053_antalya_protocol_marker'; + +SYSTEM FLUSH LOGS query_log; + +-- The worker runs in `default` rather than in the test database, so its row is reached through the +-- initiating query's id. `IN` rather than a scalar subquery: the worker's row is also +-- `is_initial_query`, so a second match must narrow the result rather than throw. +SELECT DISTINCT client_name +FROM system.query_log +WHERE event_date >= yesterday() + AND type = 'QueryFinish' + AND initial_query_id IN + ( + SELECT query_id + FROM system.query_log + WHERE event_date >= yesterday() + AND current_database = currentDatabase() + AND log_comment = '05053_antalya_protocol_marker' + AND type = 'QueryFinish' + AND is_initial_query + ) +ORDER BY 1;