diff --git a/docs/integration-api-v1.md b/docs/integration-api-v1.md index e5e1ec0a9..1e9d857de 100644 --- a/docs/integration-api-v1.md +++ b/docs/integration-api-v1.md @@ -181,6 +181,31 @@ refuses the persistent backend with `ConfigurationError` rather than generating with nothing stored. The catalog takes one publisher per `(database, table_prefix)`, so a second engine on the same catalog is refused at `create_record_runtime`. +To reach a secured catalog, set `clickhouse_scheme="https"` (and the server's +TLS HTTP port, usually 8443) on `NativeCaptureStorageConfig`. The client always +verifies the server's certificate and name, against libcurl's built-in CA +bundle/directory, or a private CA given as `clickhouse_ca_file` (a PEM bundle) +or `clickhouse_ca_path` (a hashed directory). Each replaces libcurl's built-in +default for that option rather than adding to it: a libcurl built with both a +bundle and a directory (Debian, Ubuntu) keeps trusting the system roots +through the other, a bundle-only build (RHEL, Fedora) does not, so to trust +both there, pass a bundle holding the system roots and the private CA. +`clickhouse_user`/`clickhouse_password` travel as +`X-ClickHouse-User`/`X-ClickHouse-Key` headers, never in a URL, and the +password is left out of the config's repr. A password over plain http is +refused unless `clickhouse_allow_insecure_http=True`, and `clickhouse_host` +must be a bare host (no scheme, port or `user:password@`). +`clickhouse_reader_user`/`clickhouse_reader_password` give +`NativeCaptureReader` a separate account, for example one limited with +`GRANT SELECT` on the catalog tables (or a `readonly=2` settings profile). Do +not use a `readonly=1` profile: the reader sends its query limits +(`max_rows_to_read`, `max_execution_time`, ...) as settings, which +`readonly=1` refuses (Code 164, READONLY). The storage service always uses +`clickhouse_user`. Every catalog request is bounded by +`clickhouse_request_timeout_s`, which must be at least twice +`publish_timeout_s` (10 s at the default 5 s publish timeout). A refused connection is retried for any +statement; a reset, or a 5xx that is not a permanent ClickHouse error (such +as a row limit or a denied grant), for reads only; and a timeout never. The schedule's default factory creates a distinct `CaptureSchedule` for each config instance. `MonitoringEngine` stores the config, while concrete adaptors decide whether diff --git a/native/csrc/catalog/bindings_store.cpp b/native/csrc/catalog/bindings_store.cpp index 90be3dc99..0d552487f 100644 --- a/native/csrc/catalog/bindings_store.cpp +++ b/native/csrc/catalog/bindings_store.cpp @@ -52,11 +52,27 @@ dmi_store::S3Config s3_config(const py::dict& d) { return c; } -dc::ClickHouseTimeouts clickhouse_timeouts(const py::dict& d) { - dc::ClickHouseTimeouts t; - t.connect_s = get(d, "clickhouse_connect_timeout_s", t.connect_s); - t.request_s = get(d, "clickhouse_request_timeout_s", t.request_s); - return t; +// The catalog connection, for the service and the reader alike: both read +// the same keys, so a reader is pointed at a secured catalog exactly as the +// service is (NativeCaptureReader passes its own account, when configured, +// under the same key names). The client validates it on construction. +dc::ClickHouseConnection clickhouse_connection(const py::dict& d) { + dc::ClickHouseConnection c; + c.scheme = get(d, "clickhouse_scheme", c.scheme); + c.host = get(d, "clickhouse_host", c.host); + c.port = get(d, "clickhouse_port", c.port); + c.user = get(d, "clickhouse_user", c.user); + c.password = get(d, "clickhouse_password", c.password); + c.ca_file = get(d, "clickhouse_ca_file", c.ca_file); + c.ca_path = get(d, "clickhouse_ca_path", c.ca_path); + c.allow_insecure_http = + get(d, "clickhouse_allow_insecure_http", c.allow_insecure_http); + c.timeouts.connect_s = + get(d, "clickhouse_connect_timeout_s", c.timeouts.connect_s); + c.timeouts.request_s = + get(d, "clickhouse_request_timeout_s", c.timeouts.request_s); + c.max_attempts = get(d, "clickhouse_max_attempts", c.max_attempts); + return c; } dc::StorageServiceConfig service_config(const py::dict& d) { @@ -69,9 +85,7 @@ dc::StorageServiceConfig service_config(const py::dict& d) { c.uploader.max_in_flight_bytes = get(d, "uploader_max_in_flight_bytes", c.uploader.max_in_flight_bytes); c.uploader.max_attempts = get(d, "uploader_max_attempts", c.uploader.max_attempts); - c.clickhouse_host = get(d, "clickhouse_host", c.clickhouse_host); - c.clickhouse_port = get(d, "clickhouse_port", c.clickhouse_port); - c.clickhouse_timeouts = clickhouse_timeouts(d); + c.clickhouse = clickhouse_connection(d); c.max_index_attempts = get(d, "max_index_attempts", c.max_index_attempts); c.writer.database = get(d, "database", "default"); c.writer.table_prefix = get(d, "table_prefix", "dmi"); @@ -184,9 +198,7 @@ class CaptureReader { explicit CaptureReader(const py::dict& d) : s3_(s3_config(d)), client_(std::make_shared( - get(d, "clickhouse_host", "127.0.0.1"), - get(d, "clickhouse_port", 8123), - clickhouse_timeouts(d))), + clickhouse_connection(d))), config_{get(d, "database", "default"), get(d, "table_prefix", "dmi")}, catalog_(client_, config_), diff --git a/native/csrc/catalog/clickhouse_client.cpp b/native/csrc/catalog/clickhouse_client.cpp index 8b816a8b5..230233455 100644 --- a/native/csrc/catalog/clickhouse_client.cpp +++ b/native/csrc/catalog/clickhouse_client.cpp @@ -1,6 +1,13 @@ #include "clickhouse_client.h" +#include +#include +#include +#include +#include #include +#include +#include #include "../common/curl_init.h" #include "sql_escape.h" @@ -14,6 +21,93 @@ size_t write_body(char* ptr, size_t size, size_t nmemb, void* userp) { return size * nmemb; } +// Leading decimal digits of `text` from `at`, as long as they fit an int; +// -1 when there are none. +int leading_int(const std::string& text, size_t at) { + int value = -1; + while (at < text.size() && text[at] >= '0' && text[at] <= '9') { + const int digit = text[at] - '0'; + if (value > (INT_MAX - digit) / 10) return -1; + value = (value < 0 ? 0 : value * 10) + digit; + ++at; + } + return value; +} + +// Keeps the X-ClickHouse-Exception-Code of the LAST response: libcurl hands +// the header callback every response's headers, a 100 Continue's included, +// and each status line starts a new response. +size_t read_header(char* ptr, size_t size, size_t nmemb, void* userp) { + const size_t length = size * nmemb; + const std::string line(ptr, length); + int* code = static_cast(userp); + static const std::string name = "x-clickhouse-exception-code:"; + if (line.compare(0, 5, "HTTP/") == 0) { + *code = -1; + } else if (line.size() > name.size() && + std::equal(name.begin(), name.end(), line.begin(), + [](char a, char b) { + return a == std::tolower(static_cast(b)); + })) { + size_t at = name.size(); + while (at < line.size() && (line[at] == ' ' || line[at] == '\t')) ++at; + *code = leading_int(line, at); + } + return length; +} + +// A ClickHouse error body starts "Code: . DB::Exception: ...": the +// fallback when no X-ClickHouse-Exception-Code header came with it. +int body_exception_code(const std::string& body) { + static const std::string prefix = "Code: "; + if (body.compare(0, prefix.size(), prefix) != 0) return -1; + const int code = leading_int(body, prefix.size()); + size_t end = prefix.size(); + while (end < body.size() && body[end] >= '0' && body[end] <= '9') ++end; + return end < body.size() && body[end] == '.' ? code : -1; +} + +// The ClickHouse errors (src/Common/ErrorCodes.cpp; the names checked with +// errorCodeToName on 25.12) that a repeat of a READ can plausibly cure: the +// server, or something it depends on, was briefly unable to answer. Every +// other code arriving with a 5xx names a failure of the statement itself -- +// TOO_MANY_ROWS (158), ACCESS_DENIED (497), READONLY (164), +// FUNCTION_THROW_IF_VALUE_IS_NON_ZERO (395), ILLEGAL_TYPE_OF_ARGUMENT (43) +// all come back as 500 -- and a repeat would only run it again. +// +// Deliberately absent: TIMEOUT_EXCEEDED (159), since timeouts are never +// retried (see execute()); MEMORY_LIMIT_EXCEEDED (241), which is as often the +// query's own max_memory_usage as a busy server, and repeating it adds to the +// pressure; QUERY_WAS_CANCELLED (394), which is usually a KILL QUERY; and the +// write-side codes (TABLE_IS_READ_ONLY 242, TOO_MANY_PARTS 252), since a +// write is never retried once it reached the server anyway. +bool transient_clickhouse_error(int code) { + switch (code) { + case 3: // UNEXPECTED_END_OF_FILE + case 202: // TOO_MANY_SIMULTANEOUS_QUERIES + case 209: // SOCKET_TIMEOUT (the server's own socket, not this request) + case 210: // NETWORK_ERROR + case 236: // ABORTED + case 279: // ALL_CONNECTION_TRIES_FAILED + case 425: // SYSTEM_ERROR + case 999: // KEEPER_EXCEPTION (select_sequential_consistency asks Keeper) + case 1000: // POCO_EXCEPTION + return true; + default: + return false; + } +} + +// libcurl takes whole milliseconds, and 0 means "its default" -- no bound at +// all for the whole request. A positive timeout therefore rounds UP, so one +// below a millisecond still bounds the request (at 1 ms); validate() has +// already refused anything that is not positive. +long timeout_ms(double seconds) { + const double ms = std::ceil(seconds * 1000.0); + if (!(ms < static_cast(LONG_MAX))) return LONG_MAX; + return std::max(1L, static_cast(ms)); +} + std::string url_encode(const std::string& value) { char* escaped = curl_easy_escape(nullptr, value.c_str(), static_cast(value.size())); @@ -27,8 +121,181 @@ std::string url_encode(const std::string& value) { return out; } +bool has_header_breaking_byte(const std::string& value) { + return value.find_first_of(std::string("\r\n\0", 3)) != std::string::npos; +} + +// The host goes into the URL verbatim, so anything that is not a host +// would change what the URL means: `user:pw@host` sends credentials in the +// URL (and into every log that records it), `host:port` or `host/path` +// reaches a different endpoint than the port field says. Only a bare name +// or address is admitted, and the message never repeats the value -- the +// likeliest mistake is a URL with a password in it. +void validate_host(const std::string& host) { + const auto refuse = [](const std::string& why) { + throw ClickHouseError( + "clickhouse host " + why + ": pass a bare host name or address; " + "the scheme, port and credentials each have their own option"); + }; + if (host.empty()) refuse("is empty"); + if (host.find('@') != std::string::npos) { + refuse("must not carry userinfo (user:password@)"); + } + if (host.front() == '[') { + const bool closed = host.size() > 2 && host.back() == ']'; + const bool inner_ok = + closed && std::all_of(host.begin() + 1, host.end() - 1, [](char c) { + return std::isxdigit(static_cast(c)) || c == ':' || + c == '.'; + }); + if (!inner_ok) refuse("is not a bracketed IPv6 address"); + return; + } + const bool name_ok = std::all_of(host.begin(), host.end(), [](char c) { + return std::isalnum(static_cast(c)) || c == '-' || + c == '.' || c == '_'; + }); + if (!name_ok) refuse("is not a bare host name or address"); +} + +// Failures where the request never left this process: repeating the +// statement cannot run it twice, whatever it is. +bool never_connected(CURLcode code) { + return code == CURLE_COULDNT_CONNECT || code == CURLE_COULDNT_RESOLVE_HOST || + code == CURLE_COULDNT_RESOLVE_PROXY; +} + +// Failures after connecting that a repeat can plausibly cure: the +// connection broke, or the answer was cut short. Timeouts are deliberately +// absent (see execute()), as are TLS verification failures, which a repeat +// cannot cure. +bool transient_transport(CURLcode code) { + return code == CURLE_SEND_ERROR || code == CURLE_RECV_ERROR || + code == CURLE_GOT_NOTHING || code == CURLE_PARTIAL_FILE; +} + +struct Attempt { + CURLcode code = CURLE_OK; + long status = 0; + int exception_code = -1; // X-ClickHouse-Exception-Code; -1: none sent + std::string body; + std::string detail; // libcurl's error buffer: says WHICH certificate check +}; + +struct SlistFree { + void operator()(curl_slist* list) const { curl_slist_free_all(list); } +}; +using Headers = std::unique_ptr; + +void append_header(Headers* headers, const std::string& line) { + curl_slist* grown = curl_slist_append(headers->get(), line.c_str()); + if (grown == nullptr) throw ClickHouseError("libcurl failed to add a header"); + (void)headers->release(); + headers->reset(grown); +} + +std::vector parse_tsv(const std::string& body) { + std::vector rows; + size_t start = 0; + while (start < body.size()) { + size_t end = body.find('\n', start); + if (end == std::string::npos) end = body.size(); + const std::string line = body.substr(start, end - start); + start = end + 1; + if (line.empty()) continue; + Row row; + size_t field_start = 0; + while (field_start <= line.size()) { + size_t field_end = line.find('\t', field_start); + if (field_end == std::string::npos) field_end = line.size(); + row.push_back(line.substr(field_start, field_end - field_start)); + field_start = field_end + 1; + if (field_end == line.size()) break; + } + rows.push_back(std::move(row)); + } + return rows; +} + } // namespace +void validate(const ClickHouseConnection& c) { + if (c.scheme != "http" && c.scheme != "https") { + throw ClickHouseError("clickhouse scheme must be \"http\" or \"https\""); + } + validate_host(c.host); + if (c.port == 0) throw ClickHouseError("clickhouse port must be in 1..65535"); + if (has_header_breaking_byte(c.user)) { + throw ClickHouseError("clickhouse user must not contain CR, LF or NUL"); + } + if (has_header_breaking_byte(c.password)) { + throw ClickHouseError("clickhouse password must not contain CR, LF or NUL"); + } + if (!c.password.empty() && c.user.empty()) { + throw ClickHouseError( + "clickhouse password is set without a clickhouse user; name the " + "account it belongs to"); + } + if (c.scheme == "http") { + if (!c.ca_file.empty() || !c.ca_path.empty()) { + throw ClickHouseError( + "clickhouse ca_file/ca_path need the https scheme; over http they " + "would verify nothing"); + } + if (!c.password.empty() && !c.allow_insecure_http) { + throw ClickHouseError( + "clickhouse password over plain http is refused: use https, or set " + "allow_insecure_http to send it in the clear"); + } + } else if (c.allow_insecure_http) { + throw ClickHouseError( + "clickhouse allow_insecure_http admits plain http and never " + "downgrades TLS; leave it unset for https"); + } + if (!(c.timeouts.connect_s > 0) || !(c.timeouts.request_s > 0)) { + throw ClickHouseError("clickhouse timeouts must be positive"); + } + if (c.max_attempts < 1) { + throw ClickHouseError("clickhouse max_attempts must be at least 1"); + } +} + +bool is_read_statement(const std::string& statement) { + size_t at = 0; + while (at < statement.size() && + std::isspace(static_cast(statement[at]))) { + ++at; + } + std::string keyword; + while (at < statement.size() && + std::isalpha(static_cast(statement[at]))) { + keyword.push_back(static_cast( + std::toupper(static_cast(statement[at])))); + ++at; + } + if (keyword == "WITH") { + // ClickHouse parses `WITH 1 AS x INSERT INTO t SELECT x` as an INSERT + // whose WITH clause comes first, so a WITH statement is a read only if + // INSERT appears nowhere in it as a word. Conservative on purpose: an + // INSERT inside a string literal or a quoted name also makes it a + // write, and a read misjudged as a write only loses its retries. + std::string word; + for (size_t i = at; i <= statement.size(); ++i) { + const unsigned char c = + i < statement.size() ? static_cast(statement[i]) : 0; + if (std::isalnum(c) || c == '_') { + word.push_back(static_cast(std::toupper(c))); + continue; + } + if (word == "INSERT") return false; + word.clear(); + } + return true; + } + return keyword == "SELECT" || keyword == "SHOW" || keyword == "DESCRIBE" || + keyword == "DESC" || keyword == "EXISTS" || keyword == "CHECK"; +} + std::string substitute(const std::string& query, const Params& params) { // ONE left-to-right pass, appending to an output buffer, because // clickhouse-driver's substitution is `query % escaped` @@ -108,12 +375,9 @@ std::map deciding_read() { return {{"select_sequential_consistency", "1"}}; } -ClickHouseClient::ClickHouseClient(std::string host, uint16_t port, - ClickHouseTimeouts timeouts) - : host_(std::move(host)), port_(port), timeouts_(timeouts) { - if (!(timeouts_.connect_s > 0) || !(timeouts_.request_s > 0)) { - throw ClickHouseError("clickhouse timeouts must be positive"); - } +ClickHouseClient::ClickHouseClient(ClickHouseConnection connection) + : connection_(std::move(connection)) { + validate(connection_); // Process-lifetime, not per-object: the matching curl_global_cleanup used // to run in the destructor below, which tore libcurl down for the WHOLE // process while the uploader's worker threads were inside @@ -121,79 +385,139 @@ ClickHouseClient::ClickHouseClient(std::string host, uint16_t port, dmi_common::EnsureCurlGlobalInit(); } +ClickHouseClient::ClickHouseClient(std::string host, uint16_t port, + ClickHouseTimeouts timeouts) + : ClickHouseClient([&] { + ClickHouseConnection c; + c.host = std::move(host); + c.port = port; + c.timeouts = timeouts; + return c; + }()) {} + ClickHouseClient::~ClickHouseClient() = default; std::vector ClickHouseClient::execute( const std::string& query, const Params& params, - const std::map& settings) const { + const std::map& settings, int* attempts) const { const std::string statement = substitute(query, params); + const bool read = is_read_statement(statement); // Settings ride as URL parameters; the statement is the POST body - // (GET-with-query is evaluated as readonly — writes are refused). - std::string url = "http://" + host_ + ":" + std::to_string(port_) + "/?"; - for (const auto& [key, value] : settings) { + // (GET-with-query is evaluated as readonly — writes are refused). A + // caller's own wait_end_of_query wins over the default added here. + std::map url_settings = settings; + if (read) url_settings.emplace("wait_end_of_query", "1"); + std::string url = connection_.scheme + "://" + connection_.host + ":" + + std::to_string(connection_.port) + "/?"; + for (const auto& [key, value] : url_settings) { url += url_encode(key) + "=" + url_encode(value) + "&"; } url.pop_back(); - CURL* curl = curl_easy_init(); - if (curl == nullptr) throw ClickHouseError("libcurl init failed"); - std::string body; - curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); - // The body is a LENGTH, not a C string. Without an explicit size - // libcurl measures the POST body with strlen, so a NUL anywhere in the - // statement silently drops everything after it and the server answers - // the prefix as if that were the whole query. The escaper never emits a - // raw NUL, but statement text assembled outside it still can, so the - // length is its own guard rather than a consequence of the escaping. - // - // Order is free for CURLOPT_POSTFIELDS, which only borrows the buffer; - // anyone switching to CURLOPT_COPYPOSTFIELDS must set the size FIRST, - // because that option copies using the size known at the time. - curl_easy_setopt(curl, CURLOPT_POSTFIELDS, statement.c_str()); - curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE_LARGE, - static_cast(statement.size())); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_body); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, &body); - // NOSIGNAL: timeouts must not use SIGALRM in a multi-threaded process. - curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L); - curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, - static_cast(timeouts_.connect_s * 1000)); - curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, - static_cast(timeouts_.request_s * 1000)); - const CURLcode code = curl_easy_perform(curl); - long status = 0; - curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &status); - curl_easy_cleanup(curl); - if (code != CURLE_OK) { - throw ClickHouseError(std::string("curl: ") + - curl_easy_strerror(code)); - } - if (status != 200) { - throw ClickHouseError("clickhouse " + std::to_string(status) + ": " + - body.substr(0, 4096)); + // Credentials as headers: ClickHouse reads X-ClickHouse-User/-Key, and + // unlike URL parameters or userinfo they do not end up in access logs or + // the error text below. libcurl follows no redirect (no + // CURLOPT_FOLLOWLOCATION), so no other server is sent them -- but libcurl + // does honor the http_proxy/https_proxy/all_proxy/no_proxy environment. + // Over https a proxy only sees the CONNECT, which carries none of these + // headers; over plain http (allow_insecure_http) a configured proxy + // receives the whole request, credentials included. + Headers headers; + if (!connection_.user.empty()) { + append_header(&headers, "X-ClickHouse-User: " + connection_.user); + if (!connection_.password.empty()) { + append_header(&headers, "X-ClickHouse-Key: " + connection_.password); + } } - std::vector rows; - size_t start = 0; - while (start < body.size()) { - size_t end = body.find('\n', start); - if (end == std::string::npos) end = body.size(); - const std::string line = body.substr(start, end - start); - start = end + 1; - if (line.empty()) continue; - Row row; - size_t field_start = 0; - while (field_start <= line.size()) { - size_t field_end = line.find('\t', field_start); - if (field_end == std::string::npos) field_end = line.size(); - row.push_back(line.substr(field_start, field_end - field_start)); - field_start = field_end + 1; - if (field_end == line.size()) break; + const auto perform = [&]() { + Attempt attempt; + CURL* curl = curl_easy_init(); + if (curl == nullptr) throw ClickHouseError("libcurl init failed"); + char error_buffer[CURL_ERROR_SIZE] = {0}; + curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); + // The body is a LENGTH, not a C string. Without an explicit size + // libcurl measures the POST body with strlen, so a NUL anywhere in the + // statement silently drops everything after it and the server answers + // the prefix as if that were the whole query. The escaper never emits a + // raw NUL, but statement text assembled outside it still can, so the + // length is its own guard rather than a consequence of the escaping. + // + // Order is free for CURLOPT_POSTFIELDS, which only borrows the buffer; + // anyone switching to CURLOPT_COPYPOSTFIELDS must set the size FIRST, + // because that option copies using the size known at the time. + curl_easy_setopt(curl, CURLOPT_POSTFIELDS, statement.c_str()); + curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE_LARGE, + static_cast(statement.size())); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_body); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, &attempt.body); + curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, read_header); + curl_easy_setopt(curl, CURLOPT_HEADERDATA, &attempt.exception_code); + curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, error_buffer); + if (headers) curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers.get()); + if (connection_.scheme == "https") { + // libcurl's defaults, stated: https without verification is not an + // option this client offers. + curl_easy_setopt(curl, CURLOPT_SSL_VERIFYPEER, 1L); + curl_easy_setopt(curl, CURLOPT_SSL_VERIFYHOST, 2L); + if (!connection_.ca_file.empty()) { + curl_easy_setopt(curl, CURLOPT_CAINFO, connection_.ca_file.c_str()); + } + if (!connection_.ca_path.empty()) { + curl_easy_setopt(curl, CURLOPT_CAPATH, connection_.ca_path.c_str()); + } } - rows.push_back(std::move(row)); + // NOSIGNAL: timeouts must not use SIGALRM in a multi-threaded process. + curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L); + curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, + timeout_ms(connection_.timeouts.connect_s)); + curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, + timeout_ms(connection_.timeouts.request_s)); + attempt.code = curl_easy_perform(curl); + curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &attempt.status); + curl_easy_cleanup(curl); + attempt.detail = error_buffer; + return attempt; + }; + + for (int number = 1;; ++number) { + const Attempt attempt = perform(); + if (attempt.code == CURLE_OK && attempt.status == 200) { + if (attempts != nullptr) *attempts = number; + return parse_tsv(attempt.body); + } + std::string error; + bool retry = false; + if (attempt.code != CURLE_OK) { + error = std::string("curl: ") + curl_easy_strerror(attempt.code); + if (!attempt.detail.empty()) error += ": " + attempt.detail; + retry = never_connected(attempt.code) || + (read && transient_transport(attempt.code)); + } else { + error = "clickhouse " + std::to_string(attempt.status) + ": " + + attempt.body.substr(0, 4096); + // A 5xx that names a ClickHouse error is retried only when that error + // is transient; one that names none came from something in front of + // the server (a proxy's 502/503/504) or from a server too broken to + // say, and is retried as before. Reads only, either way. + int code = attempt.exception_code; + if (code < 0) code = body_exception_code(attempt.body); + retry = read && attempt.status >= 500 && attempt.status < 600 && + (code < 0 || transient_clickhouse_error(code)); + } + if (!retry || number >= connection_.max_attempts) { + if (number > 1) { + error += " (after " + std::to_string(number) + " attempts)"; + } + throw ClickHouseError(error); + } + // 100 ms, doubling, capped at 1 s: enough for a restarting server or a + // flapping connection, short beside the request timeout it adds to. + const int shift = std::min(number - 1, 4); + std::this_thread::sleep_for(std::chrono::milliseconds( + std::min(100 << shift, 1000))); } - return rows; } } // namespace dmi_catalog diff --git a/native/csrc/catalog/clickhouse_client.h b/native/csrc/catalog/clickhouse_client.h index 51f291c54..185051731 100644 --- a/native/csrc/catalog/clickhouse_client.h +++ b/native/csrc/catalog/clickhouse_client.h @@ -56,8 +56,57 @@ struct ClickHouseTimeouts { double request_s = 60.0; // the whole request, connect included }; +// Where the catalog lives and how to reach it: the HTTP interface, plain or +// TLS, with optional credentials. Validated when a client is built from it +// (validate() below), so an inconsistent connection is refused before any +// request rather than at the first statement. +struct ClickHouseConnection { + std::string scheme = "http"; // "http" or "https", lower case + // A bare host name, IPv4 address or bracketed IPv6 address -- never a + // URL: no scheme, port, path or userinfo ("user:pw@host"). + std::string host = "127.0.0.1"; + uint16_t port = 8123; + // Sent as X-ClickHouse-User / X-ClickHouse-Key headers, never in the URL + // (a URL lands in proxy, server and error logs). Empty user: no auth + // headers, which the server reads as its `default` user. + std::string user; + std::string password; + // https always verifies the peer and its name, against libcurl's built-in + // CA bundle and/or directory unless these name a private CA + // (CURLOPT_CAINFO / CURLOPT_CAPATH). Each REPLACES that option's built-in + // default rather than adding to it, so whether the system roots are still + // trusted depends on the libcurl build: one configured with both a bundle + // and a directory (Debian, Ubuntu) keeps the other; a bundle-only build + // (RHEL, Fedora) trusts only ca_file once it is set. To trust both, pass a + // bundle holding the system roots and the private CA. Refused with http, + // where they would silently do nothing. + std::string ca_file; + std::string ca_path; + // A password over plain http must be opted into. Refused with https: + // the flag admits plain http, it never downgrades TLS. + bool allow_insecure_http = false; + ClickHouseTimeouts timeouts; + // Attempts per statement, first included, for the failures it is safe to + // repeat (see execute()). 1 disables retries. + int max_attempts = 3; +}; + +// Throws ClickHouseError naming the first inconsistent field. Messages never +// contain the password. +void validate(const ClickHouseConnection& connection); + +// Whether a statement only reads, judged by its first keyword: SELECT, +// SHOW, DESCRIBE/DESC, EXISTS, CHECK, or WITH when the word INSERT appears +// nowhere in the statement (ClickHouse reads `WITH ... INSERT INTO ...` as +// an INSERT). Anything else -- including a statement that opens with a +// parenthesis or a comment -- counts as a write, the safe default, since a +// write is never repeated once it may have reached the server. +bool is_read_statement(const std::string& statement); + class ClickHouseClient { public: + explicit ClickHouseClient(ClickHouseConnection connection); + // Plain http, no credentials: the conformance drivers and local servers. ClickHouseClient(std::string host, uint16_t port, ClickHouseTimeouts timeouts = {}); ~ClickHouseClient(); @@ -68,14 +117,39 @@ class ClickHouseClient { // Runs one statement with `%(name)s` parameters substituted client-side // and `settings` appended as URL parameters. Returns the parsed // FORMAT TSV rows (empty for writes). + // + // Retries, up to connection.max_attempts, with a short backoff: + // * a connection that was never made (refused, or the name did not + // resolve) -- for ANY statement, since nothing reached the server; + // * a transport error after connecting (reset, empty reply, short read) + // or a 5xx -- for reads only. A write that may have reached the + // server has an unknown outcome, and repeating it is the caller's + // decision (the fenced publish quarantines instead). ClickHouse + // answers 500 for permanent errors too (a row limit, a denied grant), + // so a 5xx that names a ClickHouse error (X-ClickHouse-Exception-Code, + // or a "Code: N." body) is retried only for the few transient codes + // listed in the .cpp; a 5xx naming none (a proxy's 502/503/504) is. + // A timeout is never retried, so each attempt's bound is the whole + // call's: at most max_attempts request timeouts plus the backoff, and a + // single request timeout for anything that timed out. TLS failures (an + // untrusted or misnamed certificate) and 4xx answers are not retried. + // Timeouts go to libcurl in whole milliseconds, rounded up, so a positive + // timeout below 1 ms bounds the request at 1 ms rather than not at all. + // + // Reads also carry wait_end_of_query=1, so the server buffers the result + // and an exception part-way through it arrives as an error status rather + // than as a 200 whose truncated body would parse as rows. A URL setting: + // the statement bytes are unchanged. + // + // `attempts`, when given, receives the number of requests the statement + // took (1 without a retry); it is set only when execute() returns. std::vector execute( const std::string& query, const Params& params = {}, - const std::map& settings = {}) const; + const std::map& settings = {}, + int* attempts = nullptr) const; private: - std::string host_; - uint16_t port_; - ClickHouseTimeouts timeouts_; + ClickHouseConnection connection_; }; } // namespace dmi_catalog diff --git a/native/csrc/catalog/conformance_catalog.cpp b/native/csrc/catalog/conformance_catalog.cpp index 02a7223c3..d82ee5a5a 100644 --- a/native/csrc/catalog/conformance_catalog.cpp +++ b/native/csrc/catalog/conformance_catalog.cpp @@ -340,9 +340,45 @@ Session make_session(const std::string& line) { // HTTP interface: the driver speaks HTTP (libcurl), not the native TCP // protocol clickhouse-driver uses, so the port differs from the // Python-side suites' DMI_CLICKHOUSE_PORT. 8123 is ClickHouse's default. - session.client = std::make_shared( - host != nullptr ? host : "127.0.0.1", - static_cast(port != nullptr ? std::atoi(port) : 8123)); + dmi_catalog::ClickHouseConnection connection; + connection.host = host != nullptr ? host : "127.0.0.1"; + connection.port = + static_cast(port != nullptr ? std::atoi(port) : 8123); + // Optional per-session overrides of the connection, for the suites that + // pin what the client sends (tests/test_native_catalog_connection.py). + // Absent keys keep the environment's plain-http defaults above. + if (jc::HasKey(line, "clickhouse_scheme")) { + connection.scheme = jc::FindString(line, "clickhouse_scheme"); + } + if (jc::HasKey(line, "clickhouse_host")) { + connection.host = jc::FindString(line, "clickhouse_host"); + } + if (jc::HasKey(line, "clickhouse_port")) { + connection.port = static_cast(field_int(line, "clickhouse_port")); + } + connection.user = jc::FindString(line, "clickhouse_user"); + connection.password = jc::FindString(line, "clickhouse_password"); + connection.ca_file = jc::FindString(line, "clickhouse_ca_file"); + connection.ca_path = jc::FindString(line, "clickhouse_ca_path"); + connection.allow_insecure_http = + jc::FindBool(line, "clickhouse_allow_insecure_http"); + if (jc::HasKey(line, "clickhouse_request_timeout_ms")) { + connection.timeouts.request_s = + static_cast(field_int(line, "clickhouse_request_timeout_ms")) / + 1000.0; + } + // Microseconds, for the sub-millisecond bound libcurl cannot express. + if (jc::HasKey(line, "clickhouse_request_timeout_us")) { + connection.timeouts.request_s = + static_cast(field_int(line, "clickhouse_request_timeout_us")) / + 1e6; + } + if (jc::HasKey(line, "clickhouse_max_attempts")) { + connection.max_attempts = + static_cast(field_int(line, "clickhouse_max_attempts")); + } + session.client = + std::make_shared(connection); session.writer = std::make_unique(session.client, config); session.database = config.database; session.table_prefix = config.table_prefix; @@ -1039,8 +1075,10 @@ std::string respond(const std::string& line, Session* session) { static_cast(field_int(line, "clock_skew_ns"))); out = std::string(",\"admits\":") + (admits ? "1" : "0"); } else if (op == "execute") { + int attempts = 0; out = rows_to_json(session->client->execute( - jc::FindString(line, "query"))); + jc::FindString(line, "query"), {}, {}, &attempts)); + out += ",\"attempts\":" + std::to_string(attempts); } else { return prefix + "false,\"what\":\"unknown op\"}"; } diff --git a/native/csrc/catalog/storage_service.cpp b/native/csrc/catalog/storage_service.cpp index 63be51074..941a3a038 100644 --- a/native/csrc/catalog/storage_service.cpp +++ b/native/csrc/catalog/storage_service.cpp @@ -71,9 +71,7 @@ uint64_t lease_tick_ns(uint64_t ttl_ns) { CaptureStorageService::CaptureStorageService(StorageServiceConfig config) : config_(std::move(config)), s3_(config_.s3), - clickhouse_(std::make_shared( - config_.clickhouse_host, config_.clickhouse_port, - config_.clickhouse_timeouts)), + clickhouse_(std::make_shared(config_.clickhouse)), writer_(clickhouse_, config_.writer), indexer_(&s3_, &writer_, config_.indexer) { if (config_.spool_root.empty()) { diff --git a/native/csrc/catalog/storage_service.h b/native/csrc/catalog/storage_service.h index 965387c91..13b77dd87 100644 --- a/native/csrc/catalog/storage_service.h +++ b/native/csrc/catalog/storage_service.h @@ -70,9 +70,9 @@ struct StorageServiceConfig { dmi_store::S3Config s3; dmi_store::UploaderConfig uploader; // uploader.store_id names the store - std::string clickhouse_host = "127.0.0.1"; - uint16_t clickhouse_port = 8123; - ClickHouseTimeouts clickhouse_timeouts; + // The catalog's HTTP interface: scheme, host, port, credentials, TLS + // trust, timeouts and retry attempts. + ClickHouseConnection clickhouse; WriterConfig writer; // database, table_prefix, lease TTLs IndexerConfig indexer; std::string holder; // the publisher lease holder id diff --git a/src/dmi/storage/native_capture.py b/src/dmi/storage/native_capture.py index bfd1e2e8c..209ac2f56 100644 --- a/src/dmi/storage/native_capture.py +++ b/src/dmi/storage/native_capture.py @@ -28,6 +28,7 @@ import json import math import os +import re import socket import uuid from dataclasses import dataclass, field @@ -56,6 +57,18 @@ def _load_native_store_extension() -> Any: ) from exc +# A bare host name or IPv4 address, or a bracketed IPv6 address. Anything +# else would change what the URL the native client builds means. +_BARE_HOST = re.compile(r"[A-Za-z0-9._-]+|\[[0-9A-Fa-f:.]+\]") + +def _text(name: str, value: Any) -> None: + if type(value) is not str: + raise TypeError(f"{name} must be str") + # Credentials travel as HTTP headers; a line break would end the header. + if any(c in value for c in "\r\n\x00"): + raise ValueError(f"{name} must not contain CR, LF or NUL") + + def _finite(name: str, value: Any) -> None: if type(value) not in (int, float): raise TypeError(f"{name} must be float") @@ -106,8 +119,29 @@ class NativeCaptureStorageConfig: # The name packs are indexed under; readers resolve it to this store. store_id: str = "s3" - clickhouse_host: str = "127.0.0.1" - clickhouse_port: int = 8123 # the HTTP interface + clickhouse_host: str = "127.0.0.1" # a bare host: no scheme, port or user + clickhouse_port: int = 8123 # the HTTP interface (8443 for its TLS port) + # "https" always verifies the server's certificate and name, against + # libcurl's built-in CA bundle/directory, or a private CA given as + # clickhouse_ca_file (a PEM bundle) or clickhouse_ca_path (an OpenSSL + # hashed directory). Each REPLACES libcurl's default for that option: + # whether the system roots still count depends on the libcurl build + # (kept on Debian/Ubuntu, dropped by bundle-only builds such as RHEL). + clickhouse_scheme: str = "http" + # Sent as X-ClickHouse-User / X-ClickHouse-Key headers, never in a URL. + # Empty: no credentials, which ClickHouse reads as its `default` user. + clickhouse_user: str = "" + clickhouse_password: str = field(default="", repr=False) + clickhouse_ca_file: str = "" + clickhouse_ca_path: str = "" + # A password over plain http must be opted into, as for s3. + clickhouse_allow_insecure_http: bool = False + # An optional separate account for NativeCaptureReader, typically one + # limited with GRANT SELECT (or a readonly=2 profile) -- not readonly=1, + # which refuses the query-limit settings every read sends (Code 164). + # Empty: the reader uses clickhouse_user. + clickhouse_reader_user: str = "" + clickhouse_reader_password: str = field(default="", repr=False) # Every catalog request is bounded, so a server that stops answering # cannot hold a flush, a publish or the lease renewal indefinitely. clickhouse_connect_timeout_s: float = 10.0 @@ -192,6 +226,7 @@ def __post_init__(self) -> None: self.clickhouse_connect_timeout_s, float) _positive("clickhouse_request_timeout_s", self.clickhouse_request_timeout_s, float) + self._validate_clickhouse_connection() _positive("close_flush_timeout_s", self.close_flush_timeout_s, float) if type(self.reconcile_interval_s) not in (int, float): raise TypeError("reconcile_interval_s must be float") @@ -200,6 +235,16 @@ def __post_init__(self) -> None: if self.reconcile_interval_s < 0: raise ValueError("reconcile_interval_s must be non-negative") self._validate_lease() + # A publish runs server-side for up to publish_timeout_s. A client + # that gives up first reports an outcome it does not know, and the + # writer quarantines itself over a statement that may have committed. + minimum = 2 * float(self.publish_timeout_s) + if self.clickhouse_request_timeout_s < minimum: + raise ValueError( + "clickhouse_request_timeout_s must be at least " + f"{minimum:g} s: twice publish_timeout_s " + f"({float(self.publish_timeout_s):g} s), so a publish is " + "never abandoned while the server may still commit it") def _validate_lease(self) -> None: for name in ("lease_ttl_s", "publish_timeout_s", "clock_skew_s"): @@ -239,6 +284,50 @@ def _lease_native(self) -> dict[str, int]: "start_lease_wait_ns": _ns(wait), } + def _validate_clickhouse_connection(self) -> None: + """What the native client refuses, refused here with field names.""" + for name in ("clickhouse_scheme", "clickhouse_host", "clickhouse_user", + "clickhouse_password", "clickhouse_ca_file", + "clickhouse_ca_path", "clickhouse_reader_user", + "clickhouse_reader_password"): + _text(name, getattr(self, name)) + if type(self.clickhouse_allow_insecure_http) is not bool: + raise TypeError("clickhouse_allow_insecure_http must be bool") + if self.clickhouse_scheme not in ("http", "https"): + raise ValueError('clickhouse_scheme must be "http" or "https"') + # The value is never repeated: the likeliest mistake here is a URL + # with a password in it. + if "@" in self.clickhouse_host: + raise ValueError( + "clickhouse_host must not carry userinfo (user:password@); " + "set clickhouse_user and clickhouse_password instead") + if _BARE_HOST.fullmatch(self.clickhouse_host) is None: + raise ValueError( + "clickhouse_host must be a bare host name or address; the " + "scheme and port have their own fields") + for user, password in ( + ("clickhouse_user", "clickhouse_password"), + ("clickhouse_reader_user", "clickhouse_reader_password")): + if getattr(self, password) and not getattr(self, user): + raise ValueError(f"{password} needs {user}: name the account " + "it belongs to") + if self.clickhouse_scheme == "http": + for name in ("clickhouse_ca_file", "clickhouse_ca_path"): + if getattr(self, name): + raise ValueError( + f"{name} needs clickhouse_scheme='https'; over http " + "it would verify nothing") + if ((self.clickhouse_password or self.clickhouse_reader_password) + and not self.clickhouse_allow_insecure_http): + raise ValueError( + "a ClickHouse password over plain http is refused: set " + "clickhouse_scheme='https', or clickhouse_allow_insecure_http" + "=True to send it in the clear") + elif self.clickhouse_allow_insecure_http: + raise ValueError( + "clickhouse_allow_insecure_http admits plain http and never " + "downgrades TLS; leave it False for https") + def _native_dict(self) -> dict[str, Any]: return { "s3_endpoint": self.s3_endpoint, @@ -251,14 +340,28 @@ def _native_dict(self) -> dict[str, Any]: "s3_ca_file": self.s3_ca_file, "s3_ca_path": self.s3_ca_path, "store_id": self.store_id, + "clickhouse_scheme": self.clickhouse_scheme, "clickhouse_host": self.clickhouse_host, "clickhouse_port": self.clickhouse_port, + "clickhouse_user": self.clickhouse_user, + "clickhouse_password": self.clickhouse_password, + "clickhouse_ca_file": self.clickhouse_ca_file, + "clickhouse_ca_path": self.clickhouse_ca_path, + "clickhouse_allow_insecure_http": self.clickhouse_allow_insecure_http, "clickhouse_connect_timeout_s": float(self.clickhouse_connect_timeout_s), "clickhouse_request_timeout_s": float(self.clickhouse_request_timeout_s), "database": self.database, "table_prefix": self.table_prefix, } + def _native_reader_dict(self) -> dict[str, Any]: + """The reader's native config: the reader account, when one is set.""" + native = self._native_dict() + if self.clickhouse_reader_user: + native["clickhouse_user"] = self.clickhouse_reader_user + native["clickhouse_password"] = self.clickhouse_reader_password + return native + class NativeCaptureStorage: """The in-process storage service: spool -> object store -> catalog.""" @@ -379,7 +482,7 @@ def __init__( if not isinstance(config, NativeCaptureStorageConfig): raise TypeError("config must be a NativeCaptureStorageConfig") module = _load_native_store_extension() - native = config._native_dict() + native = config._native_reader_dict() native["max_coalesce_gap_bytes"] = max_coalesce_gap_bytes self._columns: tuple[str, ...] = tuple(module.SEARCH_ITEM_COLUMNS) self._reader = module.CaptureReader(native) diff --git a/tests/_private_ca.py b/tests/_private_ca.py new file mode 100644 index 000000000..a099f09d0 --- /dev/null +++ b/tests/_private_ca.py @@ -0,0 +1,274 @@ +"""A throwaway private CA and TLS endpoints for the catalog client's tests. + +The certificates are generated per test session with the ``openssl`` CLI -- +nothing secret is committed -- and live for two days. The server +certificate names ``IP:127.0.0.1`` and ``DNS:localhost`` only, so a client +that verifies the peer accepts it for those names and nothing else. + +Two endpoints are built on it: + +* :class:`FakeClickHouse` -- a tiny HTTP(S) server that records every request + (method, path, headers, body) and answers through a caller's function, so + a test can see exactly what reached the wire; +* :class:`TlsTerminator` -- TLS in front of a plain TCP port (the local + ClickHouse HTTP interface), for the live round trip over verified TLS. +""" + +from __future__ import annotations + +import shutil +import socket +import ssl +import subprocess +import threading +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Callable, Optional, Union + +_CA_CONFIG = """\ +[req] +distinguished_name = dn +prompt = no +x509_extensions = v3_ca +[dn] +CN = DMI catalog test CA +[v3_ca] +basicConstraints = critical,CA:TRUE +keyUsage = critical,keyCertSign,cRLSign +subjectKeyIdentifier = hash +""" + +_LEAF_CONFIG = """\ +[req] +distinguished_name = dn +prompt = no +[dn] +CN = {cn} +""" + +_LEAF_EXTENSIONS = """\ +basicConstraints = CA:FALSE +keyUsage = critical,digitalSignature,keyEncipherment +extendedKeyUsage = serverAuth +subjectKeyIdentifier = hash +authorityKeyIdentifier = keyid,issuer +subjectAltName = {san} +""" + + +@dataclass(frozen=True) +class PrivateCa: + ca_file: Path # the CA certificate, PEM + ca_path: Path # a hashed directory holding only that CA (CURLOPT_CAPATH) + cert: Path # server certificate for 127.0.0.1 / localhost + key: Path + # Signed by the same CA for a name that is NOT this host: a client that + # checks the name refuses it even though it trusts the issuer. + wrong_name_cert: Path + wrong_name_key: Path + + def server_context(self, *, wrong_name: bool = False) -> ssl.SSLContext: + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + if wrong_name: + context.load_cert_chain(str(self.wrong_name_cert), + str(self.wrong_name_key)) + else: + context.load_cert_chain(str(self.cert), str(self.key)) + return context + + +def _openssl(*args: str, cwd: Path) -> str: + return subprocess.run(["openssl", *args], cwd=cwd, check=True, + capture_output=True, text=True).stdout + + +def make_private_ca(directory: Path) -> PrivateCa: + """Generate a CA and a server certificate it signs, under `directory`.""" + if shutil.which("openssl") is None: + raise RuntimeError("the openssl CLI is required to generate the test CA") + directory.mkdir(parents=True, exist_ok=True) + (directory / "ca.cnf").write_text(_CA_CONFIG) + _openssl("req", "-x509", "-config", "ca.cnf", "-newkey", "rsa:2048", + "-nodes", "-keyout", "ca.key", "-out", "ca.pem", "-days", "2", + cwd=directory) + for stem, cn, san in ( + ("server", "127.0.0.1", "IP:127.0.0.1,DNS:localhost"), + ("wrong-name", "not-this-host.invalid", + "DNS:not-this-host.invalid")): + (directory / f"{stem}.cnf").write_text(_LEAF_CONFIG.format(cn=cn)) + (directory / f"{stem}.ext").write_text( + _LEAF_EXTENSIONS.format(san=san)) + _openssl("req", "-new", "-config", f"{stem}.cnf", "-newkey", + "rsa:2048", "-nodes", "-keyout", f"{stem}.key", "-out", + f"{stem}.csr", cwd=directory) + _openssl("x509", "-req", "-in", f"{stem}.csr", "-CA", "ca.pem", + "-CAkey", "ca.key", "-CAcreateserial", "-out", f"{stem}.pem", + "-days", "2", "-extfile", f"{stem}.ext", cwd=directory) + # CURLOPT_CAPATH reads OpenSSL's hashed layout: .0. + subject_hash = _openssl("x509", "-hash", "-noout", "-in", "ca.pem", + cwd=directory).strip() + hashed = directory / "hashed" + hashed.mkdir(exist_ok=True) + shutil.copy(directory / "ca.pem", hashed / f"{subject_hash}.0") + return PrivateCa(ca_file=directory / "ca.pem", ca_path=hashed, + cert=directory / "server.pem", + key=directory / "server.key", + wrong_name_cert=directory / "wrong-name.pem", + wrong_name_key=directory / "wrong-name.key") + + +@dataclass +class Request: + method: str + path: str + headers: dict[str, str] # lower-cased names + body: bytes + + +# A responder returns (status, body) or (status, body, headers) -- headers a +# dict of extra response headers, such as the X-ClickHouse-Exception-Code a +# real server sends with an error -- or None to close the connection without +# answering (the client sees an empty reply: a transport error). +Responder = Callable[ + [Request], + Optional[Union[tuple[int, bytes], tuple[int, bytes, dict[str, str]]]]] + + +class FakeClickHouse: + """Records every request and answers through `respond`. + + With `tls`, the listening socket speaks TLS with the private CA's server + certificate; `handshakes` counts accepted connections and + `failed_handshakes` those whose handshake failed (a client that refused + the certificate). + """ + + def __init__(self, respond: Optional[Responder] = None, *, + tls: Optional[PrivateCa] = None, wrong_name: bool = False): + self.requests: list[Request] = [] + self.handshakes = 0 + self.failed_handshakes = 0 + self._respond = respond or (lambda request: (200, b"")) + fake = self + + class _Handler(BaseHTTPRequestHandler): + def do_POST(self): # noqa: N802 - http.server's naming + length = int(self.headers.get("Content-Length", "0")) + request = Request( + method="POST", path=self.path, + headers={k.lower(): v for k, v in self.headers.items()}, + body=self.rfile.read(length)) + fake.requests.append(request) + answer = fake._respond(request) + if answer is None: + self.close_connection = True + try: + self.connection.shutdown(socket.SHUT_RDWR) + except OSError: + pass + return + status, body, *extra = answer + self.send_response(status) + for name, value in (extra[0] if extra else {}).items(): + self.send_header(name, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + do_GET = do_POST # noqa: N815 + + def log_message(self, *args): + pass + + class _Server(ThreadingHTTPServer): + daemon_threads = True + + def get_request(self): + sock, address = self.socket.accept() + if tls is None: + return sock, address + fake.handshakes += 1 + try: + return context.wrap_socket(sock, server_side=True), address + except (ssl.SSLError, OSError): + fake.failed_handshakes += 1 + sock.close() + raise + + def handle_error(self, request, client_address): + pass + + context = (tls.server_context(wrong_name=wrong_name) + if tls is not None else None) + self._server = _Server(("127.0.0.1", 0), _Handler) + self.port = self._server.server_address[1] + self._thread = threading.Thread(target=self._server.serve_forever, + daemon=True) + self._thread.start() + + def close(self): + self._server.shutdown() + self._server.server_close() + + def __enter__(self): + return self + + def __exit__(self, *exc): + self.close() + + +class TlsTerminator: + """TLS on a local port, forwarded as plain TCP to `target`.""" + + def __init__(self, ca: PrivateCa, target: tuple[str, int]): + self._context = ca.server_context() + self._target = target + self._listener = socket.create_server(("127.0.0.1", 0)) + self.port = self._listener.getsockname()[1] + self.handshakes = 0 + self.failed_handshakes = 0 + threading.Thread(target=self._accept, daemon=True).start() + + def _accept(self): + while True: + try: + client, _ = self._listener.accept() + except OSError: + return + threading.Thread(target=self._serve, args=(client,), + daemon=True).start() + + def _serve(self, client): + self.handshakes += 1 + try: + tls = self._context.wrap_socket(client, server_side=True) + except (ssl.SSLError, OSError): + self.failed_handshakes += 1 + client.close() + return + try: + upstream = socket.create_connection(self._target) + except OSError: + tls.close() + return + threading.Thread(target=self._pump, args=(upstream, tls), + daemon=True).start() + self._pump(tls, upstream) + + @staticmethod + def _pump(source, sink): + try: + while data := source.recv(65536): + sink.sendall(data) + except OSError: + pass + finally: + for end in (source, sink): + try: + end.shutdown(socket.SHUT_RDWR) + except OSError: + pass + + def close(self): + self._listener.close() diff --git a/tests/test_native_capture_storage_live.py b/tests/test_native_capture_storage_live.py index 6e6f051bc..683d27150 100644 --- a/tests/test_native_capture_storage_live.py +++ b/tests/test_native_capture_storage_live.py @@ -773,6 +773,63 @@ def _native(spool, holder): assert snapshot["lease_renewals"] >= 3, snapshot +def test_capture_round_trips_through_a_verified_tls_catalog(fake_s3, tmp_path): + """The whole path -- schema, lease, index, publish, search, resolve -- + over https to the catalog, verified against a private CA. + + A TLS terminator with a freshly generated CA stands in front of the local + ClickHouse HTTP port, so the server itself is untouched. The same catalog + refuses the service when the client is not given that CA: https never + falls back to trusting whatever answers.""" + from tests._private_ca import TlsTerminator, make_private_ca + + ca = make_private_ca(tmp_path / "ca") + terminator = TlsTerminator(ca, (CLICKHOUSE_HOST, CLICKHOUSE_HTTP_PORT)) + spool_root = tmp_path / "spool" + tensors = _stage(spool_root, range(4)) + try: + with _catalog() as (_client, catalog): + tls = dict(clickhouse_scheme="https", clickhouse_host="127.0.0.1", + clickhouse_port=terminator.port, + # The local server's passwordless `default`, named so + # the credential headers are on every request. + clickhouse_user="default") + + untrusted = _storage_config(fake_s3, catalog.table_prefix, **tls) + with pytest.raises(RuntimeError, match="(?i)certificate"): + _service(untrusted, spool_root).start() + # Counted on the terminator's thread, which may learn of the + # refused handshake a moment after the client has given up. + deadline = time.monotonic() + 5.0 + while (terminator.failed_handshakes < 1 + and time.monotonic() < deadline): + time.sleep(0.01) + assert terminator.failed_handshakes >= 1 + + config = _storage_config(fake_s3, catalog.table_prefix, + clickhouse_ca_file=str(ca.ca_file), **tls) + service = _service(config, spool_root) + service.start() + try: + service.flush(30.0) + snapshot = service.snapshot() + finally: + service.stop() + assert snapshot["indexed_packs"] == 2, snapshot + assert snapshot["indexed_rows"] == 4, snapshot + + captures = _read_all(config) + assert sorted(captures) == sorted(tensors) + for capture_id, tensor in tensors.items(): + assert captures[capture_id].payload == tensor.numpy().tobytes() + # The CA as a hashed directory reaches the same catalog. + by_path = _storage_config(fake_s3, catalog.table_prefix, + clickhouse_ca_path=str(ca.ca_path), **tls) + assert sorted(_read_all(by_path)) == sorted(tensors) + finally: + terminator.close() + + # --- the publisher lease through ClickHouse errors and restarts --------------- diff --git a/tests/test_native_capture_storage_wiring.py b/tests/test_native_capture_storage_wiring.py index 842040a07..dbe9c8a20 100644 --- a/tests/test_native_capture_storage_wiring.py +++ b/tests/test_native_capture_storage_wiring.py @@ -113,6 +113,158 @@ def test_https_with_the_insecure_flag_is_refused(): s3_allow_insecure_http=True) +# --- the catalog connection: scheme, credentials, TLS -------------------------- + + +def test_catalog_passwords_stay_out_of_the_repr(): + text = repr(_storage_config( + clickhouse_scheme="https", clickhouse_user="writer", + clickhouse_password="writer-pw-test", + clickhouse_reader_user="reader", + clickhouse_reader_password="reader-pw-test")) + assert "writer-pw-test" not in text + assert "reader-pw-test" not in text + # The account names are not secrets, and help say which one failed. + assert "clickhouse_user='writer'" in text + + +def test_the_catalog_defaults_are_the_local_plain_http_server(): + config = _storage_config() + assert (config.clickhouse_scheme, config.clickhouse_host, + config.clickhouse_port) == ("http", "127.0.0.1", 8123) + native = config._native_dict() + assert native["clickhouse_scheme"] == "http" + assert native["clickhouse_user"] == "" + assert native["clickhouse_password"] == "" + assert native["clickhouse_allow_insecure_http"] is False + + +def test_the_catalog_connection_reaches_the_native_dict(): + config = _storage_config( + clickhouse_scheme="https", clickhouse_host="ch.example.test", + clickhouse_port=8443, clickhouse_user="writer", + clickhouse_password="writer-pw", clickhouse_ca_file="/etc/ca.pem", + clickhouse_ca_path="/etc/ca.d") + native = config._native_dict() + assert {k: native[k] for k in ( + "clickhouse_scheme", "clickhouse_host", "clickhouse_port", + "clickhouse_user", "clickhouse_password", "clickhouse_ca_file", + "clickhouse_ca_path", "clickhouse_allow_insecure_http")} == { + "clickhouse_scheme": "https", "clickhouse_host": "ch.example.test", + "clickhouse_port": 8443, "clickhouse_user": "writer", + "clickhouse_password": "writer-pw", "clickhouse_ca_file": "/etc/ca.pem", + "clickhouse_ca_path": "/etc/ca.d", + "clickhouse_allow_insecure_http": False} + # The reader account is the reader's alone; the service never sees it. + assert not any(k.startswith("clickhouse_reader") for k in native) + + +def test_the_reader_account_replaces_the_writer_account_for_the_reader(): + config = _storage_config( + clickhouse_scheme="https", clickhouse_user="writer", + clickhouse_password="writer-pw", clickhouse_reader_user="reader", + clickhouse_reader_password="reader-pw") + native = config._native_reader_dict() + assert (native["clickhouse_user"], native["clickhouse_password"]) == ( + "reader", "reader-pw") + without = _storage_config(clickhouse_scheme="https", + clickhouse_user="writer", + clickhouse_password="writer-pw") + native = without._native_reader_dict() + assert (native["clickhouse_user"], native["clickhouse_password"]) == ( + "writer", "writer-pw") + + +def test_a_catalog_password_over_plain_http_must_be_opted_into(): + with pytest.raises(ValueError, match="clickhouse_allow_insecure_http"): + _storage_config(clickhouse_user="writer", clickhouse_password="pw") + with pytest.raises(ValueError, match="clickhouse_allow_insecure_http"): + _storage_config(clickhouse_reader_user="reader", + clickhouse_reader_password="pw") + config = _storage_config(clickhouse_user="writer", + clickhouse_password="pw", + clickhouse_allow_insecure_http=True) + assert config.clickhouse_allow_insecure_http is True + # A user name alone is not a secret: the local passwordless default. + assert _storage_config(clickhouse_user="default").clickhouse_user == "default" + + +def test_the_catalog_insecure_flag_never_downgrades_https(): + with pytest.raises(ValueError, match="clickhouse_allow_insecure_http"): + _storage_config(clickhouse_scheme="https", + clickhouse_allow_insecure_http=True) + + +@pytest.mark.parametrize("scheme", ["", "ftp", "HTTP", "https://", None]) +def test_the_catalog_scheme_is_http_or_https(scheme): + with pytest.raises((ValueError, TypeError), match="clickhouse_scheme"): + _storage_config(clickhouse_scheme=scheme) + + +@pytest.mark.parametrize("host", [ + "writer:pw@ch.example.test", "@ch.example.test", "https://ch.example.test", + "ch.example.test:8443", "ch.example.test/db", "ch.example.test?x=1", + "ch.example.test#x", "ch example", "", "[::1", +]) +def test_the_catalog_host_is_only_a_host(host): + with pytest.raises(ValueError, match="clickhouse_host"): + _storage_config(clickhouse_host=host) + + +def test_a_bracketed_ipv6_catalog_host_is_accepted(): + assert _storage_config(clickhouse_host="[::1]").clickhouse_host == "[::1]" + + +@pytest.mark.parametrize("name", ["clickhouse_ca_file", "clickhouse_ca_path"]) +def test_a_catalog_ca_needs_https(name): + with pytest.raises(ValueError, match=name): + _storage_config(**{name: "/etc/ca.pem"}) + assert getattr(_storage_config(clickhouse_scheme="https", + **{name: "/etc/ca.pem"}), name) + + +@pytest.mark.parametrize("fields, match", [ + (dict(clickhouse_password="pw"), "clickhouse_user"), + (dict(clickhouse_reader_password="pw"), "clickhouse_reader_user"), + (dict(clickhouse_user="a\nb"), "clickhouse_user"), + (dict(clickhouse_password="a\r\nb", clickhouse_user="u"), + "clickhouse_password"), + (dict(clickhouse_reader_user="a\x00b"), "clickhouse_reader_user"), +]) +def test_catalog_credentials_are_checked(fields, match): + with pytest.raises(ValueError, match=match): + _storage_config(clickhouse_scheme="https", **fields) + + +@pytest.mark.parametrize("name", [ + "clickhouse_user", "clickhouse_password", "clickhouse_ca_file", + "clickhouse_reader_user", "clickhouse_reader_password"]) +def test_catalog_text_options_must_be_text(name): + with pytest.raises(TypeError, match=name): + _storage_config(clickhouse_scheme="https", **{name: 7}) + + +def test_the_request_timeout_rule_follows_the_configured_publish_timeout(): + # Twice publish_timeout_s, not twice a fixed 5 s: a 7 s publish cap needs + # a 14 s request timeout, and 12 s would abandon a publish mid-flight. + with pytest.raises(ValueError, match="twice publish_timeout_s"): + _storage_config(publish_timeout_s=7.0, + clickhouse_request_timeout_s=12.0) + assert _storage_config( + publish_timeout_s=7.0, + clickhouse_request_timeout_s=14.0).clickhouse_request_timeout_s == 14.0 + + +def test_the_catalog_request_timeout_outlasts_a_publish(): + """A publish runs server-side for up to its 5 s publish timeout; a client + that gives up sooner reports an outcome it does not know, and the writer + quarantines itself over a statement that may well have committed.""" + with pytest.raises(ValueError, match="clickhouse_request_timeout_s"): + _storage_config(clickhouse_request_timeout_s=9.9) + assert _storage_config( + clickhouse_request_timeout_s=10.0).clickhouse_request_timeout_s == 10.0 + + # --- a private CA for an https object store ---------------------------------- diff --git a/tests/test_native_catalog_connection.py b/tests/test_native_catalog_connection.py new file mode 100644 index 000000000..4fbd99b97 --- /dev/null +++ b/tests/test_native_catalog_connection.py @@ -0,0 +1,680 @@ +"""The native catalog client's connection: auth, TLS, retries, bounds. + +The catalog's ClickHouse client used to speak plain ``http://host:port`` with +no credentials, so capture storage could not reach a secured ClickHouse at +all. These tests pin what it now puts on the wire, against a local fake +ClickHouse that records every request (no server needed): + +* credentials travel as ``X-ClickHouse-User`` / ``X-ClickHouse-Key`` headers + and never in the URL, and a password is refused over plain HTTP unless the + caller opts in; +* ``https`` always verifies the peer and its name, against libcurl's + built-in roots or a private CA given as a file or a hashed directory; +* a statement that never reached the server (connection refused) is retried + whatever it is; a read is retried on a transport error or a 5xx, unless the + 5xx names a ClickHouse error a repeat cannot cure (ClickHouse answers 500 + for a row limit or a denied grant too); a write is + never retried once it may have reached the server, since its outcome is + unknown; and a timeout is not retried at all, so one request timeout stays + the bound it claims to be; +* reads ask the server to finish the query before answering + (``wait_end_of_query=1``), so an error part-way through a result arrives as + an error status instead of a truncated body that parses as rows. + +Most of it drives the client through the ``conformance_catalog`` driver's +``execute`` op; the last section drives the ``_dmi_native_store`` bindings +and ``NativeCaptureStorageConfig`` to show the options reach both the +storage service and the reader. +""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess +import threading +import time +from pathlib import Path +from urllib.parse import parse_qs, urlsplit + +import pytest + +from tests._private_ca import FakeClickHouse, make_private_ca + +REPO = Path(__file__).resolve().parents[1] +BUILD = REPO / "native" / "build" +DRIVER = BUILD / "conformance_catalog" +STORE_BUILT = bool(sorted(BUILD.glob("_dmi_native_store*.so"))) + +pytestmark = [ + pytest.mark.cpu, + pytest.mark.skipif( + not DRIVER.exists(), + reason="native/build/conformance_catalog is not built; run " + "`make -C native build/conformance_catalog`"), +] + +OPEN = { + "op": "open", + "database": "default", + "table_prefix": "connection_test", + "lease_ttl_ns": 30_000_000_000, + "publish_timeout_ns": 5_000_000_000, + "clock_skew_ns": 0, + "allocation_attempts": 3, +} + +PASSWORD = "pw-5e1f0c7a-do-not-log" +READ = "SELECT 1" +WRITE = "INSERT INTO `default`.`connection_test_t` VALUES (1)" + + +@pytest.fixture(scope="module") +def ca(tmp_path_factory): + return make_private_ca(tmp_path_factory.mktemp("private-ca")) + + +class Driver: + """One conformance_catalog process; the env points it at a dead port, + so a test reaches a server only through the connection it opens.""" + + def __init__(self): + env = dict(os.environ) + env["DMI_CLICKHOUSE_HOST"] = "127.0.0.1" + env["DMI_CLICKHOUSE_HTTP_PORT"] = "1" + self.proc = subprocess.Popen( + [str(DRIVER)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, + text=True, bufsize=1, env=env) + + def call(self, **fields) -> dict: + self.proc.stdin.write(json.dumps(fields) + "\n") + self.proc.stdin.flush() + return json.loads(self.proc.stdout.readline()) + + def open(self, **connection) -> dict: + return self.call(**OPEN, **connection) + + def execute(self, query: str) -> dict: + return self.call(op="execute", query=query) + + def close(self): + try: + self.proc.stdin.close() + except BrokenPipeError: + pass + self.proc.wait(timeout=60) + + +@pytest.fixture +def driver(): + d = Driver() + yield d + d.close() + + +def _opened(driver, **connection): + response = driver.open(**connection) + assert response["ok"], response + return driver + + +def _query(request) -> dict[str, list[str]]: + return parse_qs(urlsplit(request.path).query) + + +# --- credentials --------------------------------------------------------------- + + +def test_credentials_travel_as_headers_and_never_in_the_url(driver): + with FakeClickHouse(lambda r: (200, b"1\n")) as fake: + _opened(driver, clickhouse_port=fake.port, + clickhouse_user="catalog_writer", clickhouse_password=PASSWORD, + clickhouse_allow_insecure_http=True) + response = driver.execute(READ) + assert response["ok"], response + assert response["rows"] == [["1"]] + [request] = fake.requests + assert request.headers["x-clickhouse-user"] == "catalog_writer" + assert request.headers["x-clickhouse-key"] == PASSWORD + assert PASSWORD not in request.path + assert "catalog_writer" not in request.path + assert "authorization" not in request.headers + assert request.body == READ.encode() + + +def test_no_credentials_send_no_auth_headers(driver): + with FakeClickHouse() as fake: + _opened(driver, clickhouse_port=fake.port) + assert driver.execute(READ)["ok"] + [request] = fake.requests + assert "x-clickhouse-user" not in request.headers + assert "x-clickhouse-key" not in request.headers + + +def test_a_user_without_a_password_sends_only_the_user(driver): + with FakeClickHouse() as fake: + _opened(driver, clickhouse_port=fake.port, clickhouse_user="reader") + assert driver.execute(READ)["ok"] + [request] = fake.requests + assert request.headers["x-clickhouse-user"] == "reader" + assert "x-clickhouse-key" not in request.headers + + +def test_a_password_over_plain_http_is_refused_unless_opted_into(driver): + with FakeClickHouse() as fake: + response = driver.open(clickhouse_port=fake.port, + clickhouse_user="catalog_writer", + clickhouse_password=PASSWORD) + assert not response["ok"] + assert "allow_insecure_http" in response["message"], response + assert PASSWORD not in response["message"] + assert fake.requests == [] + + +@pytest.mark.parametrize("host", [ + "catalog_writer:secret@127.0.0.1", # userinfo + "@127.0.0.1", + "http://127.0.0.1", # a scheme belongs in the scheme + "127.0.0.1:8123", # a port belongs in the port + "127.0.0.1/db", + "127.0.0.1?user=x", + "127.0.0.1#x", + "127.0.0.1 ", + "", + "[::1", +]) +def test_a_host_that_is_not_just_a_host_is_refused(driver, host): + response = driver.open(clickhouse_host=host) + assert not response["ok"], response + assert "clickhouse host" in response["message"], response + assert "secret" not in response["message"] + + +def test_a_bracketed_ipv6_host_is_accepted(driver): + assert driver.open(clickhouse_host="[::1]")["ok"] + + +@pytest.mark.parametrize("connection, needle", [ + (dict(clickhouse_scheme="ftp"), "scheme"), + (dict(clickhouse_scheme="HTTPS"), "scheme"), + (dict(clickhouse_scheme="https", clickhouse_allow_insecure_http=True), + "allow_insecure_http"), + (dict(clickhouse_ca_file="/etc/ssl/certs/ca-certificates.crt"), + "https"), + (dict(clickhouse_ca_path="/etc/ssl/certs"), "https"), + (dict(clickhouse_scheme="https", clickhouse_password=PASSWORD), + "user"), + (dict(clickhouse_user="a\r\nX-Injected: 1"), "user"), + (dict(clickhouse_scheme="https", clickhouse_user="u", + clickhouse_password="a\nb"), "password"), + (dict(clickhouse_max_attempts=0), "max_attempts"), +]) +def test_inconsistent_connection_options_are_refused(driver, connection, needle): + response = driver.open(**connection) + assert not response["ok"], response + assert needle in response["message"], response + assert PASSWORD not in response["message"] + + +# --- what a read asks of the server -------------------------------------------- + + +def test_a_read_waits_for_the_end_of_the_query_and_a_write_does_not(driver): + with FakeClickHouse() as fake: + _opened(driver, clickhouse_port=fake.port) + assert driver.execute(READ)["ok"] + assert driver.execute(WRITE)["ok"] + read, write = fake.requests + assert _query(read).get("wait_end_of_query") == ["1"] + assert "wait_end_of_query" not in _query(write) + # A URL setting: the statement bytes are exactly what the caller sent. + assert read.body == READ.encode() + assert write.body == WRITE.encode() + + +@pytest.mark.parametrize("statement", [ + "SELECT 1", " select 1", "\nWITH 1 AS x SELECT x", "SHOW TABLES", + "DESCRIBE TABLE t", "EXISTS TABLE t", "CHECK GRANT SHOW TABLES ON t", + # INSERT is matched as a whole word: a name that merely contains it + # leaves a WITH statement a read. + "WITH 1 AS inserted SELECT inserted", +]) +def test_statements_that_only_read_are_classified_as_reads(driver, statement): + with FakeClickHouse() as fake: + _opened(driver, clickhouse_port=fake.port) + assert driver.execute(statement)["ok"] + assert _query(fake.requests[0]).get("wait_end_of_query") == ["1"] + + +@pytest.mark.parametrize("statement", [ + "INSERT INTO t SELECT 1", "CREATE TABLE t (x UInt8) ENGINE = Memory", + "ALTER TABLE t DELETE WHERE 1", "DROP TABLE t", "TRUNCATE TABLE t", + "SELECTED", "(SELECT 1)", "SYSTEM FLUSH LOGS", + # ClickHouse parses a WITH clause in front of an INSERT as the INSERT's: + # the first keyword alone would call this a read, and retry it. + "WITH 1 AS x INSERT INTO t SELECT x", + "with 1 as x\ninsert into t select x", +]) +def test_everything_else_is_treated_as_a_write(driver, statement): + with FakeClickHouse() as fake: + _opened(driver, clickhouse_port=fake.port) + assert driver.execute(statement)["ok"] + assert "wait_end_of_query" not in _query(fake.requests[0]) + + +# --- retries ------------------------------------------------------------------- + + +def _failing(times: int, answer): + """Answer `answer` for the first `times` requests, then 200 "1".""" + seen = [] + + def respond(request): + seen.append(request) + return answer if len(seen) <= times else (200, b"1\n") + return respond + + +def test_a_read_is_retried_after_a_server_error(driver): + with FakeClickHouse(_failing(2, (503, b"overloaded"))) as fake: + _opened(driver, clickhouse_port=fake.port) + response = driver.execute(READ) + assert response["ok"], response + assert response["rows"] == [["1"]] + assert len(fake.requests) == 3 + + +def test_a_read_gives_up_after_its_attempts(driver): + with FakeClickHouse(lambda r: (500, b"still broken")) as fake: + _opened(driver, clickhouse_port=fake.port, clickhouse_max_attempts=4) + response = driver.execute(READ) + assert not response["ok"] + assert "500" in response["message"] and "still broken" in response["message"] + assert len(fake.requests) == 4 + + +TOO_MANY_ROWS = (b"Code: 158. DB::Exception: Limit for rows (controlled by " + b"'max_rows_to_read' setting) exceeded, max rows: 10.00. " + b"(TOO_MANY_ROWS) (version 25.12.2.54 (official build))\n") + + +def test_a_read_that_failed_for_good_is_not_retried(driver): + """ClickHouse answers 500 for errors a repeat cannot cure -- a row limit, + a denied grant, throwIf -- and names the error in a response header. A + repeat would only run the query again (a row limit rescans up to it).""" + with FakeClickHouse(lambda r: ( + 500, TOO_MANY_ROWS, {"X-ClickHouse-Exception-Code": "158"})) as fake: + _opened(driver, clickhouse_port=fake.port, + clickhouse_user="catalog_reader", clickhouse_password=PASSWORD, + clickhouse_allow_insecure_http=True) + response = driver.execute(READ) + assert not response["ok"] + assert len(fake.requests) == 1 + assert "500" in response["message"] + assert "TOO_MANY_ROWS" in response["message"], response + assert "attempts" not in response["message"], response + assert PASSWORD not in response["message"] + + +def test_the_error_code_is_read_from_the_body_without_the_header(driver): + with FakeClickHouse(lambda r: (500, TOO_MANY_ROWS)) as fake: + _opened(driver, clickhouse_port=fake.port) + response = driver.execute(READ) + assert not response["ok"] + assert len(fake.requests) == 1 + assert "TOO_MANY_ROWS" in response["message"], response + + +@pytest.mark.parametrize("code, name", [ + ("210", "NETWORK_ERROR"), ("202", "TOO_MANY_SIMULTANEOUS_QUERIES"), + ("999", "KEEPER_EXCEPTION"), +]) +def test_a_read_that_failed_transiently_is_retried(driver, code, name): + answer = (500, f"Code: {code}. DB::Exception: x. ({name})".encode(), + {"X-ClickHouse-Exception-Code": code}) + with FakeClickHouse(_failing(2, answer)) as fake: + _opened(driver, clickhouse_port=fake.port) + response = driver.execute(READ) + assert response["ok"], response + assert response["rows"] == [["1"]] + assert len(fake.requests) == 3 + + +def test_a_server_error_without_a_clickhouse_code_is_retried_for_a_read(driver): + """A 502/503/504 from a proxy or load balancer in front of ClickHouse + names no ClickHouse error, and is what a restart looks like from here.""" + with FakeClickHouse(_failing(1, (503, b"Service Unavailable")) + ) as fake: + _opened(driver, clickhouse_port=fake.port) + response = driver.execute(READ) + assert response["ok"], response + assert len(fake.requests) == 2 + + +def test_a_write_is_not_retried_after_a_server_error(driver): + with FakeClickHouse(_failing(1, (503, b"overloaded"))) as fake: + _opened(driver, clickhouse_port=fake.port) + response = driver.execute(WRITE) + assert not response["ok"] + assert "503" in response["message"] + assert len(fake.requests) == 1 + # Nor after a transient ClickHouse error: the write may have been applied. + with FakeClickHouse(_failing(1, (500, b"Code: 210. (NETWORK_ERROR)", + {"X-ClickHouse-Exception-Code": "210"})) + ) as fake: + _opened(driver, clickhouse_port=fake.port) + response = driver.execute(WRITE) + assert not response["ok"] + assert len(fake.requests) == 1 + + +def test_a_read_is_retried_when_the_connection_drops_and_a_write_is_not(driver): + with FakeClickHouse(_failing(1, None)) as fake: + _opened(driver, clickhouse_port=fake.port) + response = driver.execute(READ) + assert response["ok"], response + assert len(fake.requests) == 2 + with FakeClickHouse(_failing(1, None)) as fake: + _opened(driver, clickhouse_port=fake.port) + response = driver.execute(WRITE) + assert not response["ok"] + assert len(fake.requests) == 1 + + +def test_a_client_error_is_not_retried_even_for_a_read(driver): + with FakeClickHouse(lambda r: (404, b"Code: 60. UNKNOWN_TABLE")) as fake: + _opened(driver, clickhouse_port=fake.port) + response = driver.execute(READ) + assert not response["ok"] + assert len(fake.requests) == 1 + + +def test_a_timed_out_read_is_not_retried(driver): + """Retrying a timeout would multiply the bound the timeout promises.""" + release = threading.Event() + + def stall(request): + release.wait(10) + return (200, b"1\n") + + with FakeClickHouse(stall) as fake: + _opened(driver, clickhouse_port=fake.port, + clickhouse_request_timeout_ms=300) + started = time.monotonic() + response = driver.execute(READ) + elapsed = time.monotonic() - started + time.sleep(0.3) # a retry would have landed by now + release.set() + assert not response["ok"] + assert "timeout" in response["message"].lower() or \ + "timed out" in response["message"].lower(), response + assert len(fake.requests) == 1 + assert elapsed < 2.0, elapsed + + +def test_a_sub_millisecond_timeout_still_bounds_the_request(driver): + """libcurl takes whole milliseconds, and 0 means its default -- no bound + at all for the whole request. A positive timeout below 1 ms rounds up.""" + release = threading.Event() + + def stall(request): + release.wait(5) + return (200, b"1\n") + + with FakeClickHouse(stall) as fake: + _opened(driver, clickhouse_port=fake.port, + clickhouse_request_timeout_us=400) + started = time.monotonic() + response = driver.execute(READ) + elapsed = time.monotonic() - started + release.set() + assert not response["ok"], response + assert "timeout" in response["message"].lower() or \ + "timed out" in response["message"].lower(), response + assert elapsed < 2.0, elapsed + + +def test_a_refused_connection_is_retried_even_for_a_write(driver): + """Nothing reached the server, so even a write is safe to send again. + + The port refuses the first attempt and starts listening while the client + backs off: the write lands exactly once, on a later attempt. The driver + reports how many attempts the statement took, so a first attempt that + happened to land after listen() fails the test instead of passing it + without a retry. + """ + port_holder = socket.socket() + port_holder.bind(("127.0.0.1", 0)) # bound, not listening: refused + port = port_holder.getsockname()[1] + _opened(driver, clickhouse_port=port, clickhouse_max_attempts=5) + + received = [] + + def serve(): + # Past the first attempt, well inside the 100+200+400+800 ms backoff. + time.sleep(0.15) + port_holder.listen(1) + conn, _ = port_holder.accept() + with conn: + data = b"" + while b"\r\n\r\n" not in data: + data += conn.recv(65536) + head, _, body = data.partition(b"\r\n\r\n") + length = next(int(line.split(b":")[1]) for line in + head.split(b"\r\n") + if line.lower().startswith(b"content-length")) + while len(body) < length: + body += conn.recv(65536) + received.append(body) + conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n" + b"Connection: close\r\n\r\n") + + server = threading.Thread(target=serve, daemon=True) + server.start() + response = driver.execute(WRITE) + server.join(timeout=10) + port_holder.close() + assert response["ok"], response + assert response["attempts"] >= 2, response + assert received == [WRITE.encode()] + + +def test_a_refused_connection_gives_up_after_its_attempts(driver): + port_holder = socket.socket() + port_holder.bind(("127.0.0.1", 0)) + port = port_holder.getsockname()[1] + try: + _opened(driver, clickhouse_port=port, clickhouse_max_attempts=3) + started = time.monotonic() + response = driver.execute(WRITE) + elapsed = time.monotonic() - started + finally: + port_holder.close() + assert not response["ok"] + assert "3 attempts" in response["message"], response + assert elapsed < 5.0, elapsed + + +# --- TLS ----------------------------------------------------------------------- + + +def test_https_verifies_the_server_against_a_private_ca_file(driver, ca): + with FakeClickHouse(lambda r: (200, b"1\n"), tls=ca) as fake: + _opened(driver, clickhouse_scheme="https", clickhouse_port=fake.port, + clickhouse_ca_file=str(ca.ca_file), + clickhouse_user="catalog_writer", clickhouse_password=PASSWORD) + response = driver.execute(READ) + assert response["ok"], response + assert response["rows"] == [["1"]] + [request] = fake.requests + assert request.headers["x-clickhouse-key"] == PASSWORD + assert PASSWORD not in request.path + assert (fake.handshakes, fake.failed_handshakes) == (1, 0) + + +def test_https_verifies_the_server_against_a_private_ca_directory(driver, ca): + with FakeClickHouse(lambda r: (200, b"1\n"), tls=ca) as fake: + _opened(driver, clickhouse_scheme="https", clickhouse_port=fake.port, + clickhouse_ca_path=str(ca.ca_path)) + response = driver.execute(READ) + assert response["ok"], response + + +def test_https_refuses_a_server_its_roots_do_not_vouch_for(driver, ca): + """Without the private CA the handshake fails: no request is sent, the + password never leaves, and a refused certificate is not retried.""" + with FakeClickHouse(tls=ca) as fake: + _opened(driver, clickhouse_scheme="https", clickhouse_port=fake.port, + clickhouse_user="catalog_writer", clickhouse_password=PASSWORD) + response = driver.execute(READ) + # The fake counts on its own thread, which may learn of the refused + # handshake a moment after the client has given up. + deadline = time.monotonic() + 5.0 + while fake.failed_handshakes < 1 and time.monotonic() < deadline: + time.sleep(0.01) + assert not response["ok"] + assert "certificate" in response["message"].lower(), response + assert fake.requests == [] + assert (fake.handshakes, fake.failed_handshakes) == (1, 1) + + +def test_https_refuses_a_certificate_for_another_name(driver, ca): + with FakeClickHouse(tls=ca, wrong_name=True) as fake: + _opened(driver, clickhouse_scheme="https", clickhouse_port=fake.port, + clickhouse_ca_file=str(ca.ca_file)) + response = driver.execute(READ) + assert not response["ok"] + assert fake.requests == [] + + +def test_https_does_not_fall_back_to_plain_http(driver, ca): + with FakeClickHouse() as fake: # a plain HTTP server + _opened(driver, clickhouse_scheme="https", clickhouse_port=fake.port, + clickhouse_ca_file=str(ca.ca_file), + clickhouse_user="catalog_writer", clickhouse_password=PASSWORD) + response = driver.execute(READ) + assert not response["ok"] + assert fake.requests == [] + + +# --- the bindings and the config ------------------------------------------------- + +needs_store = pytest.mark.skipif( + not STORE_BUILT, + reason="the native store module is not built; run " + "`make -C native build/_dmi_native_store PYTHON=/bin/python`") + + +def _store_dict(ca, port, **overrides): + native = { + "s3_endpoint": "http://127.0.0.1:1", "s3_bucket": "bucket", + "s3_access_key": "AKIA-test", "s3_secret_key": "secret-test", + "s3_allow_insecure_http": True, + "clickhouse_scheme": "https", "clickhouse_host": "127.0.0.1", + "clickhouse_port": port, "clickhouse_ca_file": str(ca.ca_file), + "clickhouse_user": "catalog_writer", "clickhouse_password": PASSWORD, + "database": "default", "table_prefix": "connection_test", + } + native.update(overrides) + return native + + +@needs_store +def test_the_reader_binding_reaches_a_verified_tls_catalog(ca): + from dmi.storage.native_capture import _load_native_store_extension + + module = _load_native_store_extension() + with FakeClickHouse(lambda r: (400, b"Code: 62. refused"), tls=ca) as fake: + reader = module.CaptureReader(_store_dict(ca, fake.port)) + with pytest.raises(RuntimeError, match="400"): + reader.search({"tenant_id": "t"}) + assert fake.requests, "the reader never reached the catalog" + request = fake.requests[0] + assert request.headers["x-clickhouse-user"] == "catalog_writer" + assert request.headers["x-clickhouse-key"] == PASSWORD + assert PASSWORD not in request.path + assert _query(request).get("wait_end_of_query") == ["1"] + + +@needs_store +def test_the_service_binding_reaches_a_verified_tls_catalog(ca, tmp_path): + from dmi.storage.native_capture import _load_native_store_extension + + module = _load_native_store_extension() + with FakeClickHouse(lambda r: (400, b"Code: 62. refused"), tls=ca) as fake: + service = module.StorageService(_store_dict( + ca, fake.port, spool_root=str(tmp_path / "spool"), + holder="connection-test", reconcile_on_start=False)) + with pytest.raises(Exception, match="400"): + service.start() + service.stop() + assert fake.requests, "the service never reached the catalog" + request = fake.requests[0] + assert request.headers["x-clickhouse-user"] == "catalog_writer" + assert request.headers["x-clickhouse-key"] == PASSWORD + assert PASSWORD not in request.path + + +@needs_store +def test_the_bindings_refuse_a_password_over_plain_http(ca): + from dmi.storage.native_capture import _load_native_store_extension + + module = _load_native_store_extension() + native = _store_dict(ca, 1, clickhouse_scheme="http") + native.pop("clickhouse_ca_file") + with pytest.raises(RuntimeError, match="allow_insecure_http"): + module.CaptureReader(native) + + +def _storage_config(ca, port, **overrides): + from dmi.storage.native_capture import NativeCaptureStorageConfig + + fields = dict( + s3_endpoint="http://127.0.0.1:1", s3_bucket="bucket", + s3_access_key="AKIA-test", s3_secret_key="secret-test", + s3_allow_insecure_http=True, clickhouse_scheme="https", + clickhouse_port=port, clickhouse_ca_file=str(ca.ca_file), + clickhouse_user="catalog_writer", clickhouse_password=PASSWORD, + table_prefix="connection_test") + fields.update(overrides) + return NativeCaptureStorageConfig(**fields) + + +@needs_store +def test_the_reader_uses_its_own_account_and_the_service_the_writer(ca, tmp_path): + from dmi.storage.native_capture import ( + NativeCaptureReader, NativeCaptureStorage, + ) + + with FakeClickHouse(lambda r: (400, b"Code: 62. refused"), tls=ca) as fake: + config = _storage_config(ca, fake.port, + clickhouse_reader_user="catalog_reader", + clickhouse_reader_password="reader-pw", + reconcile_on_start=False) + with pytest.raises(RuntimeError, match="400"): + NativeCaptureReader(config).search(tenant_id="t") + reader_request = fake.requests[-1] + service = NativeCaptureStorage(config, spool_root=str(tmp_path / "s"), + spool_max_bytes=1 << 30, + sweep_spool=False) + with pytest.raises(Exception, match="400"): + service.start() + service.stop() + writer_request = fake.requests[-1] + assert reader_request.headers["x-clickhouse-user"] == "catalog_reader" + assert reader_request.headers["x-clickhouse-key"] == "reader-pw" + assert writer_request.headers["x-clickhouse-user"] == "catalog_writer" + assert writer_request.headers["x-clickhouse-key"] == PASSWORD + + +@needs_store +def test_without_a_reader_account_the_reader_uses_the_writer_account(ca): + from dmi.storage.native_capture import NativeCaptureReader + + with FakeClickHouse(lambda r: (400, b"Code: 62. refused"), tls=ca) as fake: + with pytest.raises(RuntimeError, match="400"): + NativeCaptureReader(_storage_config(ca, fake.port)).search( + tenant_id="t") + assert fake.requests[-1].headers["x-clickhouse-user"] == "catalog_writer" + assert fake.requests[-1].headers["x-clickhouse-key"] == PASSWORD