Add OAuth2 HTTP request timeouts

Motivation:
OAuth2 issuer discovery and token requests previously used libcurl without application-level connection or total request deadlines. An unavailable or stalled issuer could therefore delay client startup and recovery for several minutes.

Modification:
Add a distinct CurlWrapper connection timeout and configurable OAuth2 connect_timeout_seconds and request_timeout_seconds parameters, defaulting to 10 and 30 seconds. Validate both parameters as positive integers, apply them to discovery, token acquisition, and refresh for both OAuth2 flows, document the public configuration, and add black-box regression coverage.

Testing:
Built the modified production and AuthPluginTest objects with -Werror. Ran three new timeout and validation tests plus four existing OAuth TLS tests; all 7 passed. The complete pulsar-tests target remains blocked by the pre-existing DagWatchSession incompatibility with the installed Boost.Asio API.

Usage:
Set connect_timeout_seconds and request_timeout_seconds as positive integer strings in the AuthOauth2 parameter map or JSON passed to AuthOauth2::create. If omitted, the client uses 10-second connection and 30-second total-request timeouts.

Author: RobertIndie

include/pulsar/Authentication.h

@@ -519,7 +519,9 @@ typedef std::shared_ptr<CachedToken> 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

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

lib/auth/AuthOauth2.cc

@@ -20,9 +20,11 @@
 
 #include <boost/property_tree/json_parser.hpp>
 #include <boost/property_tree/ptree.hpp>
+#include <charconv>
 #include <cstdint>
 #include <sstream>
 #include <stdexcept>
+#include <system_error>
 
 #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,16 +301,17 @@ 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 Oauth2HttpTimeouts& 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);
+    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

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