[fix][client] Apply connect and request timeouts to the OAuth2 HTTP requests

include/pulsar/Authentication.h

@@ -531,6 +531,10 @@ typedef std::shared_ptr<CachedToken> 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
      */

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<long>(options.timeoutInSeconds));
+    if (options.connectTimeoutInSeconds > 0) {
+        curl_easy_setopt(handle_, CURLOPT_CONNECTTIMEOUT, static_cast<long>(options.connectTimeoutInSeconds));
+    }
     if (!options.userAgent.empty()) {
         curl_easy_setopt(handle_, CURLOPT_USERAGENT, options.userAgent.c_str());
     }

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,16 +294,19 @@ static std::unique_ptr<CurlWrapper::TlsContext> 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()) {
         LOG_ERROR("Failed to initialize curl");
         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

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_;
 };

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;