diff --git a/include/pulsar/Authentication.h b/include/pulsar/Authentication.h index a6a02b8f..85f122f9 100644 --- a/include/pulsar/Authentication.h +++ b/include/pulsar/Authentication.h @@ -531,6 +531,10 @@ typedef std::shared_ptr CachedTokenPtr; * "tls_cert_file": "/path/to/cert.pem", * "tls_key_file": "/path/to/key.pem" * ``` + * + * Both forms accept the optional "connect_timeout_seconds" and "request_timeout_seconds" keys, see + * AuthOauth2::create(ParamMap&). + * * If passed in as std::string, it should be in Json format. */ class PULSAR_PUBLIC AuthOauth2 : public Authentication { @@ -550,6 +554,11 @@ class PULSAR_PUBLIC AuthOauth2 : public Authentication { * Optional keys: `client_id`, `audience`, `scope`. If `client_id` is omitted, the client * uses `pulsar-client`. * + * Both methods accept the optional keys `connect_timeout_seconds` (default: 10) and + * `request_timeout_seconds` (default: 30), which bound the HTTP requests sent to the issuer so + * that an unresponsive issuer cannot block the client forever. 0 falls back to the underlying + * libcurl defaults: 300 seconds to connect and no limit for the whole request. + * * @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..82ca36ed 100644 --- a/lib/CurlWrapper.h +++ b/lib/CurlWrapper.h @@ -52,7 +52,10 @@ class CurlWrapper { std::string method; std::string postFields; std::string userAgent; + // The maximum time the whole request is allowed to take. 0 means no limit. int timeoutInSeconds{0}; + // The maximum time the connection phase is allowed to take. 0 means libcurl's default (300s). + int connectTimeoutInSeconds{0}; int maxLookupRedirects{-1}; bool authAllowRedirect{false}; }; @@ -120,7 +123,10 @@ 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_TIMEOUT, static_cast(options.timeoutInSeconds)); + if (options.connectTimeoutInSeconds > 0) { + curl_easy_setopt(handle_, CURLOPT_CONNECTTIMEOUT, static_cast(options.connectTimeoutInSeconds)); + } 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..11c5f379 100644 --- a/lib/auth/AuthOauth2.cc +++ b/lib/auth/AuthOauth2.cc @@ -51,6 +51,24 @@ OAuth2TokenEndpointAuthMethod parseTokenEndpointAuthMethod(const std::string& au return OAuth2TokenEndpointAuthMethod::Unknown; } +int parseTimeoutSeconds(const ParamMap& params, const std::string& key, int defaultValue) { + const auto it = params.find(key); + if (it == params.cend() || it->second.empty()) { + return defaultValue; + } + try { + size_t numParsed; + const int value = std::stoi(it->second, &numParsed); + if (numParsed == it->second.size() && value >= 0) { + return value; + } + } catch (const std::exception&) { + // Fall through to the warning below + } + LOG_WARN("Ignore invalid " << key << " \"" << it->second << "\", use the default value " << defaultValue); + return defaultValue; +} + std::string toFlowName(OAuth2TokenEndpointAuthMethod authMethod) { switch (authMethod) { case OAuth2TokenEndpointAuthMethod::TlsClientAuth: @@ -62,6 +80,17 @@ std::string toFlowName(OAuth2TokenEndpointAuthMethod authMethod) { } } // namespace +// Oauth2TimeoutSettings + +Oauth2TimeoutSettings Oauth2TimeoutSettings::fromParamMap(const ParamMap& params) { + Oauth2TimeoutSettings settings; + settings.connectTimeoutSeconds = + parseTimeoutSeconds(params, "connect_timeout_seconds", DEFAULT_CONNECT_TIMEOUT_SECONDS); + settings.requestTimeoutSeconds = + parseTimeoutSeconds(params, "request_timeout_seconds", DEFAULT_REQUEST_TIMEOUT_SECONDS); + return settings; +} + // AuthDataOauth2 AuthDataOauth2::AuthDataOauth2(const std::string& accessToken) { accessToken_ = accessToken; } @@ -265,8 +294,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 Oauth2TimeoutSettings& timeouts) { const auto wellKnownUrl = getWellKnownUrl(issuerUrl); CurlWrapper curl; if (!curl.init()) { @@ -274,7 +303,10 @@ static std::string fetchTokenEndpoint(const std::string& issuerUrl, return ""; } - auto result = curl.get(wellKnownUrl, "Accept: application/json", {}, tlsContext); + CurlWrapper::Options options; + options.timeoutInSeconds = timeouts.requestTimeoutSeconds; + options.connectTimeoutInSeconds = timeouts.connectTimeoutSeconds; + 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 ""; @@ -315,7 +347,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 Oauth2TimeoutSettings& timeouts) { Oauth2TokenResultPtr resultPtr = Oauth2TokenResultPtr(new Oauth2TokenResult()); if (tokenEndpoint.empty()) { return resultPtr; @@ -335,6 +368,8 @@ static Oauth2TokenResultPtr fetchOauth2Token(const std::string& tokenEndpoint, c CurlWrapper::Options options; options.postFields = std::move(postData); + options.timeoutInSeconds = timeouts.requestTimeoutSeconds; + options.connectTimeoutInSeconds = timeouts.connectTimeoutSeconds; auto result = curl.get(tokenEndpoint, "Content-Type: application/x-www-form-urlencoded", options, tlsContext); if (!result.error.empty()) { @@ -394,7 +429,8 @@ ClientCredentialFlow::ClientCredentialFlow(ParamMap& params) audience_(params["audience"]), scope_(params["scope"]), tlsCertFilePath_(params["tls_cert_file"]), - tlsKeyFilePath_(params["tls_key_file"]) {} + tlsKeyFilePath_(params["tls_key_file"]), + timeouts_(Oauth2TimeoutSettings::fromParamMap(params)) {} std::string ClientCredentialFlow::getTokenEndPoint() const { return tokenEndPoint_; } @@ -408,7 +444,7 @@ void ClientCredentialFlow::initialize() { } const auto tlsContext = createTlsContext(tlsTrustCertsFilePath_, tlsCertFilePath_, tlsKeyFilePath_); - this->tokenEndPoint_ = fetchTokenEndpoint(issuerUrl_, tlsContext.get()); + this->tokenEndPoint_ = fetchTokenEndpoint(issuerUrl_, tlsContext.get(), timeouts_); if (!this->tokenEndPoint_.empty()) { LOG_DEBUG("Get token endpoint: " << this->tokenEndPoint_); } @@ -466,7 +502,7 @@ Oauth2TokenResultPtr ClientCredentialFlow::authenticate() { const auto params = generateParamMap(); const auto tlsContext = createTlsContext(tlsTrustCertsFilePath_, tlsCertFilePath_, tlsKeyFilePath_); return fetchOauth2Token(tokenEndPoint_, params, tlsContext.get(), - OAuth2TokenEndpointAuthMethod::ClientSecretPost); + OAuth2TokenEndpointAuthMethod::ClientSecretPost, timeouts_); } TlsClientAuthFlow::TlsClientAuthFlow(ParamMap& params) @@ -475,7 +511,8 @@ TlsClientAuthFlow::TlsClientAuthFlow(ParamMap& params) audience_(params["audience"]), scope_(params["scope"]), tlsCertFilePath_(params["tls_cert_file"]), - tlsKeyFilePath_(params["tls_key_file"]) {} + tlsKeyFilePath_(params["tls_key_file"]), + timeouts_(Oauth2TimeoutSettings::fromParamMap(params)) {} std::string TlsClientAuthFlow::getTokenEndPoint() const { return tokenEndPoint_; } @@ -494,7 +531,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(), timeouts_); if (!this->tokenEndPoint_.empty()) { LOG_DEBUG("Get token endpoint: " << this->tokenEndPoint_); } @@ -523,7 +560,7 @@ Oauth2TokenResultPtr TlsClientAuthFlow::authenticate() { return resultPtr; } return fetchOauth2Token(tokenEndPoint_, params, tlsContext.get(), - OAuth2TokenEndpointAuthMethod::TlsClientAuth); + OAuth2TokenEndpointAuthMethod::TlsClientAuth, timeouts_); } // AuthOauth2 diff --git a/lib/auth/AuthOauth2.h b/lib/auth/AuthOauth2.h index b402f37e..906028c5 100644 --- a/lib/auth/AuthOauth2.h +++ b/lib/auth/AuthOauth2.h @@ -51,6 +51,31 @@ class KeyFile { static KeyFile fromBase64(const std::string& encoded); }; +/** + * The timeouts applied to the HTTP requests sent to the OAuth 2.0 issuer. + * + * Without them, an issuer that accepts the connection but never replies would block the caller of + * Authentication::getAuthData() forever. + */ +struct Oauth2TimeoutSettings { + // The maximum time to wait for the connection to the issuer to be established. + static constexpr int DEFAULT_CONNECT_TIMEOUT_SECONDS = 10; + // The maximum time to wait for a whole request to the issuer to complete, including the connection. + static constexpr int DEFAULT_REQUEST_TIMEOUT_SECONDS = 30; + + int connectTimeoutSeconds{DEFAULT_CONNECT_TIMEOUT_SECONDS}; + int requestTimeoutSeconds{DEFAULT_REQUEST_TIMEOUT_SECONDS}; + + /** + * Read the "connect_timeout_seconds" and "request_timeout_seconds" keys from `params`. + * + * A key that is missing, or whose value is not a non-negative integer, falls back to the default + * above. 0 falls back to the underlying libcurl default, i.e. 300 seconds to connect and no + * limit for the whole request. + */ + static Oauth2TimeoutSettings fromParamMap(const ParamMap& params); +}; + class ClientCredentialFlow : public Oauth2Flow { public: ClientCredentialFlow(ParamMap& params); @@ -73,6 +98,7 @@ class ClientCredentialFlow : public Oauth2Flow { const std::string scope_; const std::string tlsCertFilePath_; const std::string tlsKeyFilePath_; + const Oauth2TimeoutSettings timeouts_; std::string tlsTrustCertsFilePath_; std::once_flag initializeOnce_; }; @@ -101,6 +127,7 @@ class TlsClientAuthFlow : public Oauth2Flow { const std::string scope_; const std::string tlsCertFilePath_; const std::string tlsKeyFilePath_; + const Oauth2TimeoutSettings timeouts_; std::string tlsTrustCertsFilePath_; std::once_flag initializeOnce_; }; diff --git a/tests/AuthPluginTest.cc b/tests/AuthPluginTest.cc index 7ab151cb..65097dfa 100644 --- a/tests/AuthPluginTest.cc +++ b/tests/AuthPluginTest.cc @@ -1036,6 +1036,72 @@ TEST(AuthPluginTest, testOauth2UnknownTokenEndpointAuthMethod) { ASSERT_THROW(AuthOauth2::create(params), std::invalid_argument); } +namespace testOauth2Timeout { + +// An issuer that accepts the TCP connection but never sends a response back. Without any timeout +// configured on the OAuth2 HTTP requests, talking to such an issuer blocks the caller forever. +// +// The connection is never accepted by the application: the kernel completes the handshake from the +// listen backlog, so the client connects and then waits for a response that never comes. +class UnresponsiveServer { + public: + UnresponsiveServer() : acceptor_(io_, ASIO::ip::tcp::endpoint(ASIO::ip::tcp::v4(), 0)) {} + + uint16_t port() const { return acceptor_.local_endpoint().port(); } + + private: + ASIO::io_context io_; + ASIO::ip::tcp::acceptor acceptor_; +}; + +} // namespace testOauth2Timeout + +TEST(AuthPluginTest, testOauth2TimeoutSettings) { + ParamMap params; + auto settings = Oauth2TimeoutSettings::fromParamMap(params); + ASSERT_EQ(settings.connectTimeoutSeconds, Oauth2TimeoutSettings::DEFAULT_CONNECT_TIMEOUT_SECONDS); + ASSERT_EQ(settings.requestTimeoutSeconds, Oauth2TimeoutSettings::DEFAULT_REQUEST_TIMEOUT_SECONDS); + + // 0 is accepted and falls back to libcurl's own defaults + params["connect_timeout_seconds"] = "5"; + params["request_timeout_seconds"] = "0"; + settings = Oauth2TimeoutSettings::fromParamMap(params); + ASSERT_EQ(settings.connectTimeoutSeconds, 5); + ASSERT_EQ(settings.requestTimeoutSeconds, 0); + + // Invalid values fall back to the defaults + params["connect_timeout_seconds"] = "-1"; + params["request_timeout_seconds"] = "not-a-number"; + settings = Oauth2TimeoutSettings::fromParamMap(params); + ASSERT_EQ(settings.connectTimeoutSeconds, Oauth2TimeoutSettings::DEFAULT_CONNECT_TIMEOUT_SECONDS); + ASSERT_EQ(settings.requestTimeoutSeconds, Oauth2TimeoutSettings::DEFAULT_REQUEST_TIMEOUT_SECONDS); +} + +TEST(AuthPluginTest, testOauth2UnresponsiveIssuer) { + testOauth2Timeout::UnresponsiveServer server; + const std::string issuerUrl = "http://127.0.0.1:" + std::to_string(server.port()); + + const int requestTimeoutSeconds = 2; + ParamMap params; + params["issuer_url"] = issuerUrl; + params["client_id"] = "client-id"; + params["client_secret"] = "client-secret"; + params["audience"] = "audience"; + params["request_timeout_seconds"] = std::to_string(requestTimeoutSeconds); + + AuthenticationPtr auth = AuthOauth2::create(params); + AuthenticationDataPtr data; + + const auto start = std::chrono::steady_clock::now(); + ASSERT_EQ(auth->getAuthData(data), ResultAuthenticationError); + const auto elapsed = std::chrono::steady_clock::now() - start; + + // The request must be cut off by the timeout, not by a connection failure + ASSERT_GE(elapsed, std::chrono::seconds(requestTimeoutSeconds - 1)); + // Before the timeout was configured, getAuthData() never returned + ASSERT_LT(elapsed, std::chrono::seconds(15)); +} + TEST(AuthPluginTest, testInvalidPlugin) { Client client("pulsar://localhost:6650", ClientConfiguration{}.setAuth(AuthFactory::create("invalid"))); Producer producer;