diff --git a/lib/http/HttpClient_Curl.cpp b/lib/http/HttpClient_Curl.cpp index 6195bdedd..9005b74f4 100644 --- a/lib/http/HttpClient_Curl.cpp +++ b/lib/http/HttpClient_Curl.cpp @@ -283,7 +283,12 @@ namespace MAT_NS_BEGIN { m_state(std::make_shared()) { TRACE("Initializing HttpClient_Curl...\n"); - EnsureCurlGlobalInit(); + const CURLcode initResult = EnsureCurlGlobalInit(); + if (initResult != CURLE_OK) + { + LOG_ERROR("libcurl global initialization failed: %d", static_cast(initResult)); + return; + } TRACE("libcurl version = %s\n", curl_version_info(CURLVERSION_NOW)->version); } diff --git a/lib/http/HttpClient_Curl.hpp b/lib/http/HttpClient_Curl.hpp index ff9bef44b..edd3497a9 100644 --- a/lib/http/HttpClient_Curl.hpp +++ b/lib/http/HttpClient_Curl.hpp @@ -75,10 +75,10 @@ namespace MAT_NS_BEGIN { * for the life of the process is the only correct choice for an embedded * library; the host may still call curl_global_cleanup() itself at exit. */ -inline void EnsureCurlGlobalInit() noexcept +inline CURLcode EnsureCurlGlobalInit() noexcept { static const CURLcode initResult = curl_global_init(CURL_GLOBAL_ALL); - (void)initResult; + return initResult; } // Private per-client shared state. Defined in HttpClient_Curl.cpp: it owns the @@ -170,7 +170,10 @@ class CurlHttpOperation { if (m_callback != nullptr) { HookScope callbackScope(m_callbackHooks); - m_callback->OnHttpStateEvent(type, static_cast(curl), 0); + // OnSending runs inside curl_easy_perform with the prereq callback. + // Do not expose a handle observers could mutate during that transfer. + void* handle = (m_usePrereqCallback && type == OnSending) ? nullptr : static_cast(curl); + m_callback->OnHttpStateEvent(type, handle, 0); } } @@ -272,7 +275,15 @@ class CurlHttpOperation { // A directly constructed operation may be the process's first libcurl // user, so it shares the client's init-once rather than assuming an // HttpClient_Curl was built first. - EnsureCurlGlobalInit(); + const CURLcode initResult = EnsureCurlGlobalInit(); + if (initResult != CURLE_OK) + { + LOG_ERROR("libcurl global initialization failed: %d", static_cast(initResult)); + m_transportError = initResult; + m_setupError = initResult; + EmitCreationEvent(OnCreateFailed); + return; + } /* get a curl handle */ curl = curl_easy_init(); @@ -285,6 +296,19 @@ class CurlHttpOperation { return; } +#if LIBCURL_VERSION_NUM >= 0x075000 + const curl_version_info_data* versionInfo = curl_version_info(CURLVERSION_NOW); + if (versionInfo == nullptr) + { + LOG_ERROR("libcurl version query failed"); + m_transportError = CURLE_FAILED_INIT; + m_setupError = CURLE_FAILED_INIT; + EmitCreationEvent(OnCreateFailed); + return; + } + m_usePrereqCallback = versionInfo->version_num >= 0x075000; +#endif + if (!SetOption(CURLOPT_VERBOSE, 0L) || !SetOption(CURLOPT_URL, m_url.c_str()) || !SetOption(CURLOPT_SSL_VERIFYPEER, 1L) || @@ -302,6 +326,10 @@ class CurlHttpOperation { // inside libcurl, and aborts the transfer in an orderly way. !SetOption(CURLOPT_NOPROGRESS, 0L) || !SetAbortProgressOption() || +#if LIBCURL_VERSION_NUM >= 0x075000 // libcurl 7.80.0 + (m_usePrereqCallback && (!SetOption(CURLOPT_PREREQFUNCTION, &OnConnectionReady) || + !SetOption(CURLOPT_PREREQDATA, static_cast(this)))) || +#endif // HTTP/2 when the linked libcurl supports it, otherwise HTTP/1.1 !SetOption(CURLOPT_HTTP_VERSION, GetPreferredHttpVersion())) { @@ -379,6 +407,7 @@ class CurlHttpOperation { { TRACE("method=%s\n", this->m_method.c_str()); + m_connectionReady = false; ReleaseResponse(); // Request buffer const void *request = m_requestBody.empty() ? nullptr : m_requestBody.data(); @@ -408,68 +437,69 @@ class CurlHttpOperation { // TODO: should we control what local source port we use? // curl_easy_setopt(curl, CURLOPT_LOCALPORT, dcf_port); - // Perform initial connect, handling the timeout if needed - if (!SetOption(CURLOPT_CONNECT_ONLY, 1L)) + if (!m_usePrereqCallback) { - DispatchEvent(OnConnectFailed); - goto cleanup; - } - DispatchEvent(OnConnecting); - m_transportError = curl_easy_perform(curl); - if(CURLE_OK != m_transportError) - { - DispatchEvent(OnConnectFailed); // couldn't connect - stage 1 - TRACE("Error #1: %s\n", curl_easy_strerror(m_transportError)); - goto cleanup; - } + if (!SetOption(CURLOPT_CONNECT_ONLY, 1L)) + { + DispatchEvent(OnConnectFailed); + goto cleanup; + } + DispatchEvent(OnConnecting); + m_transportError = curl_easy_perform(curl); + if (CURLE_OK != m_transportError) + { + DispatchEvent(OnConnectFailed); // couldn't connect - stage 1 + TRACE("Error #1: %s\n", curl_easy_strerror(m_transportError)); + goto cleanup; + } - /* Extract the socket from the curl handle - we'll need it for waiting. - * Note that this API takes a pointer to a 'long' while we use - * curl_socket_t for sockets otherwise. - */ + /* Extract the socket from the curl handle - we'll need it for waiting. + * Note that this API takes a pointer to a 'long' while we use + * curl_socket_t for sockets otherwise. + */ #if LIBCURL_VERSION_NUM >= 0x072D00 // Version 7.45.00 - m_transportError = curl_easy_getinfo(curl, CURLINFO_ACTIVESOCKET, &sockextr); + m_transportError = curl_easy_getinfo(curl, CURLINFO_ACTIVESOCKET, &sockextr); #else - { - long lastSocket = -1; - m_transportError = curl_easy_getinfo(curl, CURLINFO_LASTSOCKET, &lastSocket); - if (m_transportError == CURLE_OK) { - sockextr = static_cast(lastSocket); + long lastSocket = -1; + m_transportError = curl_easy_getinfo(curl, CURLINFO_LASTSOCKET, &lastSocket); + if (m_transportError == CURLE_OK) + { + sockextr = static_cast(lastSocket); + } } - } #endif + if (CURLE_OK != m_transportError) + { + DispatchEvent(OnConnectFailed); // couldn't connect - stage 2 + TRACE("Error #2: %s\n", curl_easy_strerror(m_transportError)); + goto cleanup; + } + if (sockextr == CURL_SOCKET_BAD) + { + m_transportError = CURLE_FAILED_INIT; + DispatchEvent(OnConnectFailed); // couldn't connect - no socket + TRACE("Error #2: curl returned an invalid socket\n"); + goto cleanup; + } - if(CURLE_OK != m_transportError) - { - DispatchEvent(OnConnectFailed); // couldn't connect - stage 2 - TRACE("Error #2: %s\n", curl_easy_strerror(m_transportError)); - goto cleanup; - } - if (sockextr == CURL_SOCKET_BAD) - { - m_transportError = CURLE_FAILED_INIT; - DispatchEvent(OnConnectFailed); // couldn't connect - no socket - TRACE("Error #2: curl returned an invalid socket\n"); - goto cleanup; - } - - /* wait for the socket to become ready for sending */ - sockfd = sockextr; - if (WaitOnSocket(sockfd, 0, static_cast(httpConnTimeout) * 1000L) <= 0 || isAborted) - { - TRACE("Error #3: timeout, aborted=%u\n", isAborted.load() ); - m_transportError = CURLE_OPERATION_TIMEDOUT; - DispatchEvent(OnConnectFailed); // couldn't connect - stage 3 - goto cleanup; - } + /* wait for the socket to become ready for sending */ + sockfd = sockextr; + if (WaitOnSocket(sockfd, 0, static_cast(httpConnTimeout) * 1000L) <= 0 || isAborted) + { + TRACE("Error #3: timeout, aborted=%u\n", isAborted.load()); + m_transportError = CURLE_OPERATION_TIMEDOUT; + DispatchEvent(OnConnectFailed); // couldn't connect - stage 3 + goto cleanup; + } - // once connection is there - switch back to easy perform for HTTP post - if (!SetOption(CURLOPT_CONNECT_ONLY, 0L)) - { - DispatchEvent(OnSendFailed); - goto cleanup; + // once connection is there - switch back to easy perform for HTTP post + if (!SetOption(CURLOPT_CONNECT_ONLY, 0L)) + { + DispatchEvent(OnSendFailed); + goto cleanup; + } } // send all data to our callback function @@ -521,11 +551,11 @@ class CurlHttpOperation { DispatchEvent(OnSendFailed); goto cleanup; } - DispatchEvent(OnSending); + DispatchEvent(m_usePrereqCallback ? OnConnecting : OnSending); m_transportError = curl_easy_perform(curl); if(CURLE_OK != m_transportError) { - DispatchEvent(OnSendFailed); + DispatchEvent(!m_usePrereqCallback || m_connectionReady ? OnSendFailed : OnConnectFailed); TRACE("Error: %s\n", curl_easy_strerror(m_transportError)); goto cleanup; } @@ -746,7 +776,7 @@ class CurlHttpOperation { const bool rawResponse; // Do not split response headers from response body const long httpConnTimeout; // Timeout for connect. Default: 5s - CURL *curl; // Local curl instance + CURL *curl = nullptr; // Local curl instance CURLcode m_transportError = CURLE_OK; CURLcode m_setupError = CURLE_OK; long m_httpStatusCode = 0; @@ -780,6 +810,9 @@ class CurlHttpOperation { curl_socket_t sockextr = CURL_SOCKET_BAD; + bool m_usePrereqCallback{false}; // Chosen once from the loaded libcurl version. + bool m_connectionReady{false}; // Only accessed by the transfer's worker thread. + curl_off_t nread = 0; size_t sendlen = 0; // # bytes sent by client size_t acklen = 0; // # bytes ack by server @@ -953,6 +986,41 @@ class CurlHttpOperation { #endif } +#if LIBCURL_VERSION_NUM >= 0x075000 + static int OnConnectionReady(void* clientp, char*, char*, int, int) noexcept + { + auto* operation = static_cast(clientp); + if (operation->isAborted.load(std::memory_order_acquire)) + { + return CURL_PREREQFUNC_ABORT; + } + + operation->m_connectionReady = true; +#if HAVE_EXCEPTIONS + try + { +#endif + operation->DispatchEvent(OnSending); +#if HAVE_EXCEPTIONS + } + catch (const std::exception& ex) + { + LOG_ERROR("HTTP sending state callback failed: %s", ex.what()); + operation->m_setupError = CURLE_FAILED_INIT; + return CURL_PREREQFUNC_ABORT; + } + catch (...) + { + LOG_ERROR("HTTP sending state callback failed with a non-standard exception"); + operation->m_setupError = CURLE_FAILED_INIT; + return CURL_PREREQFUNC_ABORT; + } +#endif + return operation->isAborted.load(std::memory_order_acquire) + ? CURL_PREREQFUNC_ABORT : CURL_PREREQFUNC_OK; + } +#endif + #if LIBCURL_VERSION_NUM >= 0x072000 // Version 7.32.0 static int XferInfoAbortCallback(void* clientp, curl_off_t, curl_off_t, curl_off_t, curl_off_t) noexcept { @@ -1037,7 +1105,7 @@ class CurlHttpOperation { * @param data * @return */ - static size_t WriteVectorCallback(char* ptr, size_t size, size_t nmemb, void* userp) + static size_t WriteVectorCallback(char* ptr, size_t size, size_t nmemb, void* userp) noexcept { // Guard the size * nmemb product against size_t overflow before using it. if (nmemb != 0 && size > static_cast(-1) / nmemb) { @@ -1055,7 +1123,19 @@ class CurlHttpOperation { } const auto* begin = reinterpret_cast(ptr); const auto* end = begin + realsize; - data->insert( data->end(), begin, end); +#if HAVE_EXCEPTIONS + try + { +#endif + data->insert(data->end(), begin, end); +#if HAVE_EXCEPTIONS + } + catch (...) + { + // A short write reports CURLE_WRITE_ERROR without unwinding through libcurl. + return 0; + } +#endif } return realsize; } diff --git a/lib/include/public/IHttpClient.hpp b/lib/include/public/IHttpClient.hpp index 29b57edc0..a96cdcebd 100644 --- a/lib/include/public/IHttpClient.hpp +++ b/lib/include/public/IHttpClient.hpp @@ -500,7 +500,9 @@ namespace MAT_NS_BEGIN /// /// /// HttpStateEvent - see diagram - /// HTTP client implementation-specific data structure (optional) + /// HTTP client implementation-specific data structure (optional). + /// On curl 7.80+ the OnSending event runs within a libcurl transfer and does not + /// expose its active handle; configure curl options from OnConnecting instead. /// HTTP client implementation-specific data structure size (optional) virtual void OnHttpStateEvent(HttpStateEvent state, void* data = nullptr, size_t size = 0) { diff --git a/tests/unittests/HttpClientCurlTests.cpp b/tests/unittests/HttpClientCurlTests.cpp index 23598d772..792229917 100644 --- a/tests/unittests/HttpClientCurlTests.cpp +++ b/tests/unittests/HttpClientCurlTests.cpp @@ -51,6 +51,7 @@ class HttpClientCurlTests : public ::testing::Test TEST_F(HttpClientCurlTests, SslVerification_DefaultsToTrue) { + ASSERT_EQ(EnsureCurlGlobalInit(), CURLE_OK); CurlHttpOperation op("GET", "https://example.com", nullptr, m_headers, m_body); ASSERT_NE(op.GetHandle(), nullptr); } @@ -99,12 +100,61 @@ TEST(HttpClientCurlOperationTests, ClampsConnectionTimeoutBeforeMillisecondsConv std::numeric_limits::max() / 1000L); } +class CountingHttpServer : public HttpServer +{ +public: + size_t acceptedConnections() const + { + return m_acceptedConnections.load(); + } + + bool waitForConnections(size_t count) + { + std::unique_lock lock(m_acceptedMutex); + return m_acceptedCv.wait_for(lock, std::chrono::seconds(2), + [this, count] { return acceptedConnections() >= count; }); + } + +protected: + void onSocketAcceptable(Socket socket) override + { + const size_t previousCount = m_connections.size(); + HttpServer::onSocketAcceptable(socket); + if (m_connections.size() > previousCount) + { + { + std::lock_guard lock(m_acceptedMutex); + ++m_acceptedConnections; + } + m_acceptedCv.notify_all(); + } + } + +private: + std::atomic m_acceptedConnections {0}; + std::mutex m_acceptedMutex; + std::condition_variable m_acceptedCv; +}; + +static bool UsesPrereqCallback() +{ +#if LIBCURL_VERSION_NUM >= 0x075000 + const curl_version_info_data* versionInfo = curl_version_info(CURLVERSION_NOW); + return versionInfo != nullptr && versionInfo->version_num >= 0x075000; +#else + return false; +#endif +} + class HttpClientCurlHeaderTests : public ::testing::Test, public HttpServer::Callback { protected: - HttpServer m_server; + CountingHttpServer m_server; std::string m_url; + std::mutex m_requestMutex; + std::string m_requestMethod; + std::string m_requestContent; void SetUp() override { @@ -122,8 +172,11 @@ class HttpClientCurlHeaderTests : public ::testing::Test, m_server.stop(); } - int onHttpRequest(HttpServer::Request const&, HttpServer::Response& response) override + int onHttpRequest(HttpServer::Request const& request, HttpServer::Response& response) override { + std::lock_guard lock(m_requestMutex); + m_requestMethod = request.method; + m_requestContent = request.content; response.headers["X-MAT-Test"] = "header-value"; response.content = "body-value"; return 200; @@ -132,11 +185,24 @@ class HttpClientCurlHeaderTests : public ::testing::Test, TEST_F(HttpClientCurlHeaderTests, CapturesResponseHeadersAndBody) { + struct StateCallback : public IHttpResponseCallback + { + std::vector states; + std::vector handles; + + void OnHttpResponse(IHttpResponse* response) override { delete response; } + void OnHttpStateEvent(HttpStateEvent state, void* handle, size_t) override + { + states.push_back(state); + handles.push_back(handle); + } + } callback; + const std::map requestHeaders; const std::vector requestBody; const HttpClient_Curl client; (void)client; // Initialize curl globally before constructing the operation. - CurlHttpOperation operation("GET", m_url, nullptr, requestHeaders, requestBody); + CurlHttpOperation operation("GET", m_url, &callback, requestHeaders, requestBody); operation.Send(); ASSERT_EQ(operation.GetTransportError(), CURLE_OK); @@ -147,8 +213,64 @@ TEST_F(HttpClientCurlHeaderTests, CapturesResponseHeadersAndBody) ASSERT_EQ(responseHeaders.count("X-MAT-Test"), 1u); EXPECT_EQ(responseHeaders.at("X-MAT-Test"), "header-value"); EXPECT_EQ(std::string(responseBody.begin(), responseBody.end()), "body-value"); + EXPECT_EQ(callback.states, (std::vector{OnCreated, OnConnecting, OnSending, OnResponse})); + ASSERT_EQ(callback.handles.size(), 4u); + EXPECT_EQ(callback.handles[1], operation.GetHandle()); + EXPECT_EQ(callback.handles[2], UsesPrereqCallback() ? nullptr : operation.GetHandle()); + EXPECT_EQ(m_server.acceptedConnections(), UsesPrereqCallback() ? 1u : 2u); +} + +TEST_F(HttpClientCurlHeaderTests, SendsBinaryPostWithoutRedundantConnection) +{ + const std::map requestHeaders; + const std::vector requestBody {'a', '\0', 'b'}; + CurlHttpOperation operation("POST", m_url, nullptr, requestHeaders, requestBody); + + operation.Send(); + + ASSERT_EQ(operation.GetTransportError(), CURLE_OK); + ASSERT_EQ(operation.GetHttpStatusCode(), 200L); + { + std::lock_guard lock(m_requestMutex); + EXPECT_EQ(m_requestMethod, "POST"); + EXPECT_EQ(m_requestContent, std::string(requestBody.begin(), requestBody.end())); + } + EXPECT_EQ(m_server.acceptedConnections(), UsesPrereqCallback() ? 1u : 2u); } +#if LIBCURL_VERSION_NUM >= 0x075000 +TEST_F(HttpClientCurlHeaderTests, AbortsWhenSendingStateCallbackThrows) +{ + if (!UsesPrereqCallback()) + { + GTEST_SKIP() << "OnSending runs inside libcurl only on libcurl 7.80+"; + } + + struct ThrowingCallback : public IHttpResponseCallback + { + void OnHttpResponse(IHttpResponse* response) override { delete response; } + void OnHttpStateEvent(HttpStateEvent state, void*, size_t) override + { + if (state == OnSending) + { + throw std::runtime_error("state callback failed"); + } + } + } callback; + + const std::map requestHeaders; + const std::vector requestBody; + CurlHttpOperation operation("GET", m_url, &callback, requestHeaders, requestBody); + + operation.Send(); + + EXPECT_EQ(operation.GetTransportError(), CURLE_ABORTED_BY_CALLBACK); + EXPECT_EQ(operation.GetSetupError(), CURLE_FAILED_INIT); + ASSERT_TRUE(m_server.waitForConnections(1)); + EXPECT_EQ(m_server.acceptedConnections(), 1u); +} +#endif + // --- ILogConfiguration integration --- TEST(HttpClientCurlConfigTests, LogConfiguration_SslVerify_DefaultIsTrue) @@ -659,6 +781,28 @@ TEST_F(HttpClientCurlLifetimeTests, InternalRegistryDoesNotDereferenceDeletedReq EXPECT_EQ(callback.responses(), 1u); } +TEST_F(HttpClientCurlLifetimeTests, CancelFromOnSendingAbortsTheTransfer) +{ + RecordingCallback callback; + std::unique_ptr request(m_client.CreateRequest()); + request->SetUrl(m_endpoint.url()); + const std::string id = request->GetId(); + callback.setStateHook([this, id](HttpStateEvent state) { + if (state == OnSending) + { + m_client.CancelRequestAsync(id); + } + }); + + m_client.SendRequestAsync(request.get(), &callback); + + ASSERT_TRUE(callback.waitForResponses(1, kTerminalTimeout)); + m_client.CancelAllRequests(); + EXPECT_EQ(callback.responses(), 1u); + EXPECT_EQ(callback.responsesWithResult(HttpResult_Aborted), 1u); + EXPECT_EQ(callback.stateCount(OnSending), 1u); +} + // A full drain returns only when every operation has completed and been // destroyed, for all of them, not just the first. TEST_F(HttpClientCurlLifetimeTests, CancelAllRequestsFullyDrainsEveryOperation)