diff --git a/include/pulsar/Authentication.h b/include/pulsar/Authentication.h index a6a02b8f..957ed1ab 100644 --- a/include/pulsar/Authentication.h +++ b/include/pulsar/Authentication.h @@ -519,7 +519,9 @@ typedef std::shared_ptr CachedTokenPtr; * "issuer_url": "https://accounts.google.com", * "client_id": "d9ZyX97q1ef8Cr81WHVC4hFQ64vSlDK3", * "client_secret": "on1uJ...k6F6R", - * "audience": "https://broker.example.com" + * "audience": "https://broker.example.com", + * "connect_timeout_seconds": "10", + * "request_timeout_seconds": "30" * ``` * * For `tokenEndpointAuthMethod = "tls_client_auth"`: @@ -543,12 +545,17 @@ class PULSAR_PUBLIC AuthOauth2 : public Authentication { * * For `tokenEndpointAuthMethod = "client_secret_post"` (default), the required parameter * keys are “issuer_url”, “private_key”, and “audience”. - * Optional keys: `scope`, `tls_cert_file`, `tls_key_file`. + * Optional keys: `scope`, `tls_cert_file`, `tls_key_file`, `connect_timeout_seconds`, + * and `request_timeout_seconds`. * * For `tokenEndpointAuthMethod = "tls_client_auth"`, the required parameter keys are * `issuer_url`, `tls_cert_file`, and `tls_key_file`. - * Optional keys: `client_id`, `audience`, `scope`. If `client_id` is omitted, the client - * uses `pulsar-client`. + * Optional keys: `client_id`, `audience`, `scope`, `connect_timeout_seconds`, and + * `request_timeout_seconds`. If `client_id` is omitted, the client uses `pulsar-client`. + * + * `connect_timeout_seconds` controls the OAuth HTTP connection timeout and defaults to 10 seconds. + * `request_timeout_seconds` controls the total OAuth HTTP request timeout and defaults to 30 seconds. + * Both values must be positive integers and apply to issuer discovery and token requests. * * @param parameters the key-value to create OAuth 2.0 client credentials * @see http://pulsar.apache.org/docs/en/security-oauth2/#client-credentials diff --git a/lib/CurlWrapper.h b/lib/CurlWrapper.h index cf68b635..192f7e50 100644 --- a/lib/CurlWrapper.h +++ b/lib/CurlWrapper.h @@ -52,6 +52,7 @@ class CurlWrapper { std::string method; std::string postFields; std::string userAgent; + int connectTimeoutInSeconds{0}; int timeoutInSeconds{0}; int maxLookupRedirects{-1}; bool authAllowRedirect{false}; @@ -120,7 +121,8 @@ inline CurlWrapper::Result CurlWrapper::get(const std::string& url, const std::s // Without this config, Curl_resolv_timeout might crash in multi-threads environment curl_easy_setopt(handle_, CURLOPT_NOSIGNAL, 1L); - curl_easy_setopt(handle_, CURLOPT_TIMEOUT, options.timeoutInSeconds); + curl_easy_setopt(handle_, CURLOPT_CONNECTTIMEOUT, static_cast(options.connectTimeoutInSeconds)); + curl_easy_setopt(handle_, CURLOPT_TIMEOUT, static_cast(options.timeoutInSeconds)); if (!options.userAgent.empty()) { curl_easy_setopt(handle_, CURLOPT_USERAGENT, options.userAgent.c_str()); } diff --git a/lib/auth/AuthOauth2.cc b/lib/auth/AuthOauth2.cc index 94d9fc67..31732e93 100644 --- a/lib/auth/AuthOauth2.cc +++ b/lib/auth/AuthOauth2.cc @@ -20,9 +20,11 @@ #include #include +#include #include #include #include +#include #include "InitialAuthData.h" #include "lib/Base64Utils.h" @@ -34,6 +36,11 @@ namespace pulsar { const std::string TlsClientAuthFlow::DEFAULT_CLIENT_ID = "pulsar-client"; namespace { +constexpr int DEFAULT_OAUTH2_CONNECT_TIMEOUT_SECONDS = 10; +constexpr int DEFAULT_OAUTH2_REQUEST_TIMEOUT_SECONDS = 30; +constexpr char CONNECT_TIMEOUT_PARAM[] = "connect_timeout_seconds"; +constexpr char REQUEST_TIMEOUT_PARAM[] = "request_timeout_seconds"; + enum class OAuth2TokenEndpointAuthMethod : std::uint8_t { ClientSecretPost, @@ -60,8 +67,37 @@ std::string toFlowName(OAuth2TokenEndpointAuthMethod authMethod) { return "ClientCredentialFlow"; } } + +int parsePositiveTimeout(const ParamMap& params, const char* name, int defaultValue) { + const auto it = params.find(name); + if (it == params.end()) { + return defaultValue; + } + + const auto& rawValue = it->second; + int value = 0; + const auto result = std::from_chars(rawValue.data(), rawValue.data() + rawValue.size(), value); + if (rawValue.empty() || result.ec != std::errc() || result.ptr != rawValue.data() + rawValue.size() || + value <= 0) { + throw std::invalid_argument(std::string("OAuth2 parameter ") + name + " must be a positive integer"); + } + return value; +} + +CurlWrapper::Options createHttpOptions(const Oauth2HttpTimeouts& timeouts) { + CurlWrapper::Options options; + options.connectTimeoutInSeconds = timeouts.connectTimeoutInSeconds; + options.timeoutInSeconds = timeouts.requestTimeoutInSeconds; + return options; +} } // namespace +Oauth2HttpTimeouts::Oauth2HttpTimeouts(const ParamMap& params) + : connectTimeoutInSeconds( + parsePositiveTimeout(params, CONNECT_TIMEOUT_PARAM, DEFAULT_OAUTH2_CONNECT_TIMEOUT_SECONDS)), + requestTimeoutInSeconds( + parsePositiveTimeout(params, REQUEST_TIMEOUT_PARAM, DEFAULT_OAUTH2_REQUEST_TIMEOUT_SECONDS)) {} + // AuthDataOauth2 AuthDataOauth2::AuthDataOauth2(const std::string& accessToken) { accessToken_ = accessToken; } @@ -265,8 +301,8 @@ static std::unique_ptr createTlsContext(const std::stri return tlsContext; } -static std::string fetchTokenEndpoint(const std::string& issuerUrl, - const CurlWrapper::TlsContext* tlsContext) { +static std::string fetchTokenEndpoint(const std::string& issuerUrl, const CurlWrapper::TlsContext* tlsContext, + const Oauth2HttpTimeouts& timeouts) { const auto wellKnownUrl = getWellKnownUrl(issuerUrl); CurlWrapper curl; if (!curl.init()) { @@ -274,7 +310,8 @@ static std::string fetchTokenEndpoint(const std::string& issuerUrl, return ""; } - auto result = curl.get(wellKnownUrl, "Accept: application/json", {}, tlsContext); + const auto options = createHttpOptions(timeouts); + auto result = curl.get(wellKnownUrl, "Accept: application/json", options, tlsContext); if (!result.error.empty()) { LOG_ERROR("Failed to get the well-known configuration " << issuerUrl << ": " << result.error); return ""; @@ -305,6 +342,11 @@ static std::string fetchTokenEndpoint(const std::string& issuerUrl, << issuerUrl << ". response Code " << responseCode); } break; + case CURLE_OPERATION_TIMEDOUT: + LOG_ERROR("Timed out retrieving OAuth2 issuer metadata from " + << issuerUrl << " (connect timeout: " << timeouts.connectTimeoutInSeconds + << " seconds, request timeout: " << timeouts.requestTimeoutInSeconds << " seconds)"); + break; default: LOG_ERROR("Response failed for getting the well-known configuration " << issuerUrl << ". Error Code " << res << ": " << errorBuffer); @@ -315,7 +357,8 @@ static std::string fetchTokenEndpoint(const std::string& issuerUrl, static Oauth2TokenResultPtr fetchOauth2Token(const std::string& tokenEndpoint, const ParamMap& params, const CurlWrapper::TlsContext* tlsContext, - OAuth2TokenEndpointAuthMethod authMethod) { + OAuth2TokenEndpointAuthMethod authMethod, + const Oauth2HttpTimeouts& timeouts) { Oauth2TokenResultPtr resultPtr = Oauth2TokenResultPtr(new Oauth2TokenResult()); if (tokenEndpoint.empty()) { return resultPtr; @@ -333,7 +376,7 @@ static Oauth2TokenResultPtr fetchOauth2Token(const std::string& tokenEndpoint, c } LOG_DEBUG("Generate URL encoded body for " << toFlowName(authMethod) << ": " << postData); - CurlWrapper::Options options; + auto options = createHttpOptions(timeouts); options.postFields = std::move(postData); auto result = curl.get(tokenEndpoint, "Content-Type: application/x-www-form-urlencoded", options, tlsContext); @@ -379,6 +422,11 @@ static Oauth2TokenResultPtr fetchOauth2Token(const std::string& tokenEndpoint, c << responseCode); } break; + case CURLE_OPERATION_TIMEDOUT: + LOG_ERROR("Timed out fetching OAuth2 token from " + << tokenEndpoint << " (connect timeout: " << timeouts.connectTimeoutInSeconds + << " seconds, request timeout: " << timeouts.requestTimeoutInSeconds << " seconds)"); + break; default: LOG_ERROR("Response failed for token endpoint " << tokenEndpoint << ". ErrorCode " << res << ": " << errorBuffer); @@ -389,7 +437,8 @@ static Oauth2TokenResultPtr fetchOauth2Token(const std::string& tokenEndpoint, c } ClientCredentialFlow::ClientCredentialFlow(ParamMap& params) - : issuerUrl_(params["issuer_url"]), + : httpTimeouts_(params), + issuerUrl_(params["issuer_url"]), keyFile_(KeyFile::fromParamMap(params)), audience_(params["audience"]), scope_(params["scope"]), @@ -408,7 +457,7 @@ void ClientCredentialFlow::initialize() { } const auto tlsContext = createTlsContext(tlsTrustCertsFilePath_, tlsCertFilePath_, tlsKeyFilePath_); - this->tokenEndPoint_ = fetchTokenEndpoint(issuerUrl_, tlsContext.get()); + this->tokenEndPoint_ = fetchTokenEndpoint(issuerUrl_, tlsContext.get(), httpTimeouts_); if (!this->tokenEndPoint_.empty()) { LOG_DEBUG("Get token endpoint: " << this->tokenEndPoint_); } @@ -466,11 +515,12 @@ Oauth2TokenResultPtr ClientCredentialFlow::authenticate() { const auto params = generateParamMap(); const auto tlsContext = createTlsContext(tlsTrustCertsFilePath_, tlsCertFilePath_, tlsKeyFilePath_); return fetchOauth2Token(tokenEndPoint_, params, tlsContext.get(), - OAuth2TokenEndpointAuthMethod::ClientSecretPost); + OAuth2TokenEndpointAuthMethod::ClientSecretPost, httpTimeouts_); } TlsClientAuthFlow::TlsClientAuthFlow(ParamMap& params) - : issuerUrl_(params["issuer_url"]), + : httpTimeouts_(params), + issuerUrl_(params["issuer_url"]), clientId_(params["client_id"].empty() ? DEFAULT_CLIENT_ID : params["client_id"]), audience_(params["audience"]), scope_(params["scope"]), @@ -494,7 +544,7 @@ void TlsClientAuthFlow::initialize() { LOG_ERROR("Failed to initialize TlsClientAuthFlow: tls_cert_file or tls_key_file is not set"); return; } - this->tokenEndPoint_ = fetchTokenEndpoint(issuerUrl_, tlsContext.get()); + this->tokenEndPoint_ = fetchTokenEndpoint(issuerUrl_, tlsContext.get(), httpTimeouts_); if (!this->tokenEndPoint_.empty()) { LOG_DEBUG("Get token endpoint: " << this->tokenEndPoint_); } @@ -523,7 +573,7 @@ Oauth2TokenResultPtr TlsClientAuthFlow::authenticate() { return resultPtr; } return fetchOauth2Token(tokenEndPoint_, params, tlsContext.get(), - OAuth2TokenEndpointAuthMethod::TlsClientAuth); + OAuth2TokenEndpointAuthMethod::TlsClientAuth, httpTimeouts_); } // AuthOauth2 diff --git a/lib/auth/AuthOauth2.h b/lib/auth/AuthOauth2.h index b402f37e..8ac4b5ba 100644 --- a/lib/auth/AuthOauth2.h +++ b/lib/auth/AuthOauth2.h @@ -51,6 +51,13 @@ class KeyFile { static KeyFile fromBase64(const std::string& encoded); }; +struct Oauth2HttpTimeouts { + explicit Oauth2HttpTimeouts(const ParamMap& params); + + int connectTimeoutInSeconds; + int requestTimeoutInSeconds; +}; + class ClientCredentialFlow : public Oauth2Flow { public: ClientCredentialFlow(ParamMap& params); @@ -66,6 +73,7 @@ class ClientCredentialFlow : public Oauth2Flow { } private: + const Oauth2HttpTimeouts httpTimeouts_; std::string tokenEndPoint_; const std::string issuerUrl_; const KeyFile keyFile_; @@ -94,6 +102,7 @@ class TlsClientAuthFlow : public Oauth2Flow { } private: + const Oauth2HttpTimeouts httpTimeouts_; std::string tokenEndPoint_; const std::string issuerUrl_; const std::string clientId_; diff --git a/pkg/apk/APKBUILD b/pkg/apk/APKBUILD index 464fecc6..19c5f6c6 100644 --- a/pkg/apk/APKBUILD +++ b/pkg/apk/APKBUILD @@ -38,7 +38,14 @@ build() { if [ "$CBUILD" != "$CHOST" ]; then CMAKE_CROSSOPTS="-DCMAKE_SYSTEM_NAME=Linux -DCMAKE_HOST_SYSTEM_NAME=Linux" fi + VCPKG_BASELINE=$(sed -n 's/.*"builtin-baseline"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' \ + "$ROOT_DIR/vcpkg.json") + if [ -z "$VCPKG_BASELINE" ]; then + echo "Failed to read builtin-baseline from vcpkg.json" >&2 + return 1 + fi git clone https://github.com/microsoft/vcpkg.git + git -C vcpkg checkout "$VCPKG_BASELINE" mv vcpkg $ROOT_DIR/ export VCPKG_FORCE_SYSTEM_BINARIES=1 # On aarch64 musl, vcpkg has no prebuilt binary and builds from source. diff --git a/tests/AuthPluginTest.cc b/tests/AuthPluginTest.cc index 7ab151cb..8d6bd896 100644 --- a/tests/AuthPluginTest.cc +++ b/tests/AuthPluginTest.cc @@ -23,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -548,11 +549,13 @@ static const auto mockServerTimeout = std::chrono::seconds(10); class MockOauth2Server { public: MockOauth2Server(const std::string& responseBody, const std::string& responseContentType, int listenPort, - bool requireClientCert = true) + bool requireClientCert = true, + std::chrono::milliseconds responseDelay = std::chrono::milliseconds::zero()) : responseBody_(responseBody), responseContentType_(responseContentType), acceptor_(io_, ASIO::ip::tcp::endpoint(ASIO::ip::tcp::v4(), static_cast(listenPort))), - sslCtx_(ASIO::ssl::context::sslv23) { + sslCtx_(ASIO::ssl::context::sslv23), + responseDelay_(responseDelay) { sslCtx_.set_options(ASIO::ssl::context::default_workarounds | ASIO::ssl::context::no_sslv2 | ASIO::ssl::context::no_sslv3); sslCtx_.use_certificate_chain_file(brokerPublicKeyPath); @@ -565,6 +568,8 @@ class MockOauth2Server { const std::string& request() const { return request_; } + int port() const { return acceptor_.local_endpoint().port(); } + bool mockServe() { ASIO_ERROR error; auto socket = std::make_shared(io_); @@ -573,11 +578,39 @@ class MockOauth2Server { activeSocket_ = socket; } - acceptor_.accept(*socket, error); + // Closing a synchronous accept from another thread does not reliably unblock it on all platforms. + // Poll in non-blocking mode so stop() can terminate a server that never receives a connection. + acceptor_.non_blocking(true, error); if (error) { clearActiveSocket(); return false; } + while (true) { + acceptor_.accept(*socket, error); + if (!error) { + break; + } + if (error != ASIO::error::would_block && error != ASIO::error::try_again) { + clearActiveSocket(); + return false; + } + + std::unique_lock lock(mutex_); + if (stopCondition_.wait_for(lock, std::chrono::milliseconds(10), [this]() { return stopped_; })) { + activeSocket_.reset(); + return false; + } + } + + { + std::lock_guard lock(mutex_); + connectionAccepted_ = true; + if (stopped_) { + activeSocket_.reset(); + connectionAccepted_ = false; + return false; + } + } ASIO::ssl::stream sslStream(*socket, sslCtx_); sslStream.handshake(ASIO::ssl::stream_base::server, error); @@ -586,6 +619,14 @@ class MockOauth2Server { return false; } + { + std::unique_lock lock(mutex_); + if (stopCondition_.wait_for(lock, responseDelay_, [this]() { return stopped_; })) { + activeSocket_.reset(); + connectionAccepted_ = false; + return false; + } + } const std::string response = "HTTP/1.1 200 OK\r\nContent-Type: " + responseContentType_ + "\r\nContent-Length: " + std::to_string(responseBody_.size()) + "\r\nConnection: close\r\n\r\n" + responseBody_; @@ -599,15 +640,14 @@ class MockOauth2Server { ASIO_ERROR error; { std::lock_guard lock(mutex_); - if (acceptor_.is_open()) { - acceptor_.close(error); - } - if (activeSocket_ && activeSocket_->is_open()) { + stopped_ = true; + if (connectionAccepted_ && activeSocket_ && activeSocket_->is_open()) { activeSocket_->cancel(error); activeSocket_->shutdown(ASIO::ip::tcp::socket::shutdown_both, error); activeSocket_->close(error); } } + stopCondition_.notify_all(); io_.stop(); } @@ -615,6 +655,7 @@ class MockOauth2Server { void clearActiveSocket() { std::lock_guard lock(mutex_); activeSocket_.reset(); + connectionAccepted_ = false; } bool readRequest(ASIO::ssl::stream& sslStream) { @@ -652,8 +693,12 @@ class MockOauth2Server { ASIO::io_context io_; ASIO::ip::tcp::acceptor acceptor_; ASIO::ssl::context sslCtx_; + const std::chrono::milliseconds responseDelay_; std::shared_ptr activeSocket_; std::mutex mutex_; + std::condition_variable stopCondition_; + bool stopped_{false}; + bool connectionAccepted_{false}; }; static bool awaitMockServeResult(std::future& future, MockOauth2Server& server, std::thread& thread, @@ -678,6 +723,109 @@ static bool awaitMockServeResult(std::future& future, MockOauth2Server& se } // namespace testOauth2Tls +TEST(AuthPluginTest, testOauth2IssuerDiscoveryRequestTimeout) { + using testOauth2Tls::MockOauth2Server; + + const std::string tokenBody = R"({"access_token":"mockToken","expires_in":3600,"token_type":"Bearer"})"; + MockOauth2Server tokenServer(tokenBody, "application/json", 0, false); + + std::ostringstream wellKnownBody; + wellKnownBody << R"({"token_endpoint":"https://localhost:)" << tokenServer.port() << R"(/oauth/token"})"; + MockOauth2Server wellKnownServer(wellKnownBody.str(), "application/json", 0, false, + std::chrono::seconds(3)); + + std::thread tokenThread([&tokenServer]() { tokenServer.mockServe(); }); + std::thread wellKnownThread([&wellKnownServer]() { wellKnownServer.mockServe(); }); + + ParamMap params; + params["issuer_url"] = "https://localhost:" + std::to_string(wellKnownServer.port()); + params["client_id"] = "test-client"; + params["client_secret"] = "test-secret"; + params["connect_timeout_seconds"] = "1"; + params["request_timeout_seconds"] = "1"; + + AuthenticationDataPtr data = + std::static_pointer_cast(std::make_shared(caPath)); + AuthenticationPtr auth = AuthOauth2::create(params); + + const auto start = std::chrono::steady_clock::now(); + const Result result = auth->getAuthData(data); + const auto elapsedMs = + std::chrono::duration_cast(std::chrono::steady_clock::now() - start) + .count(); + + wellKnownServer.stop(); + tokenServer.stop(); + wellKnownThread.join(); + tokenThread.join(); + + EXPECT_EQ(result, ResultAuthenticationError); + EXPECT_LT(elapsedMs, 2500); + EXPECT_NE(wellKnownServer.request().find("GET /.well-known/openid-configuration "), std::string::npos); +} + +TEST(AuthPluginTest, testOauth2TokenRequestTimeout) { + using testOauth2Tls::MockOauth2Server; + + const std::string tokenBody = R"({"access_token":"mockToken","expires_in":3600,"token_type":"Bearer"})"; + MockOauth2Server tokenServer(tokenBody, "application/json", 0, false, std::chrono::seconds(3)); + + std::ostringstream wellKnownBody; + wellKnownBody << R"({"token_endpoint":"https://localhost:)" << tokenServer.port() << R"(/oauth/token"})"; + MockOauth2Server wellKnownServer(wellKnownBody.str(), "application/json", 0, false); + + std::thread tokenThread([&tokenServer]() { tokenServer.mockServe(); }); + std::thread wellKnownThread([&wellKnownServer]() { wellKnownServer.mockServe(); }); + + ParamMap params; + params["issuer_url"] = "https://localhost:" + std::to_string(wellKnownServer.port()); + params["client_id"] = "test-client"; + params["client_secret"] = "test-secret"; + params["connect_timeout_seconds"] = "1"; + params["request_timeout_seconds"] = "1"; + + AuthenticationDataPtr data = + std::static_pointer_cast(std::make_shared(caPath)); + AuthenticationPtr auth = AuthOauth2::create(params); + + const auto start = std::chrono::steady_clock::now(); + const Result result = auth->getAuthData(data); + const auto elapsedMs = + std::chrono::duration_cast(std::chrono::steady_clock::now() - start) + .count(); + + wellKnownServer.stop(); + tokenServer.stop(); + wellKnownThread.join(); + tokenThread.join(); + + EXPECT_EQ(result, ResultAuthenticationError); + EXPECT_LT(elapsedMs, 2500); + EXPECT_NE(wellKnownServer.request().find("GET /.well-known/openid-configuration "), std::string::npos); + EXPECT_NE(tokenServer.request().find("POST /oauth/token "), std::string::npos); + EXPECT_NE(tokenServer.request().find("grant_type=client_credentials"), std::string::npos); +} + +TEST(AuthPluginTest, testOauth2HttpTimeoutValidation) { + ParamMap validParams; + validParams["issuer_url"] = "https://localhost"; + validParams["client_id"] = "test-client"; + validParams["client_secret"] = "test-secret"; + + for (const std::string key : {"connect_timeout_seconds", "request_timeout_seconds"}) { + for (const std::string value : {"not-a-number", "0", "-1", "999999999999999999999999"}) { + SCOPED_TRACE(key + "=" + value); + auto params = validParams; + params[key] = value; + EXPECT_THROW(AuthOauth2::create(params), std::invalid_argument); + } + } + + validParams["connect_timeout_seconds"] = "1"; + validParams["request_timeout_seconds"] = "2"; + EXPECT_NO_THROW(AuthOauth2::create(validParams)); +} + TEST(AuthPluginTest, testOauth2) { // test success get token from oauth2 server. pulsar::AuthenticationDataPtr data;