[feat] Add httpLookupAuthAllowRedirect option to forward auth credentials on HTTP lookup redirects

include/pulsar/ClientConfiguration.h

@@ -186,6 +186,33 @@ class PULSAR_PUBLIC ClientConfiguration {
      */
     int getMaxBackoffIntervalMs() const;
 
+    /**
+     * Configure whether to send authentication credentials when following HTTP redirects
+     * to a different host during lookup requests.
+     *
+     * When enabled, the Authorization header will be forwarded on cross-origin redirects.
+     *
+     * HTTP lookup redirects typically occur when the broker receiving the lookup request
+     * is not the owner of the requested topic. In this case, the broker responds with
+     * an HTTP redirect (3xx) pointing to the correct owner broker. If authentication is
+     * enabled, the redirected request needs to carry the auth credentials to be accepted
+     * by the target broker.
+     *
+     * If this option is not enabled and the cluster has authentication enabled, the
+     * redirected request will not carry credentials, which may result in a 401
+     * Unauthorized error from the target broker.
+     *
+     * The default value is false.
+     *
+     * @param allow whether to allow sending auth credentials on redirects
+     */
+    ClientConfiguration& setHttpLookupAuthAllowRedirect(bool allow);
+
+    /**
+     * @return whether auth credentials are sent on HTTP redirects
+     */
+    bool isHttpLookupAuthAllowRedirect() const;
+
     /**
      * Configure a custom logger backend to route of Pulsar client library
      * to a different logger implementation.

include/pulsar/c/client_configuration.h

@@ -203,6 +203,12 @@ PULSAR_PUBLIC void pulsar_client_configuration_set_keep_alive_interval_in_second
 PULSAR_PUBLIC unsigned int pulsar_client_configuration_get_keep_alive_interval_in_seconds(
     pulsar_client_configuration_t *conf);
 
+PULSAR_PUBLIC void pulsar_client_configuration_set_http_lookup_auth_allow_redirect(
+    pulsar_client_configuration_t *conf, int httpLookupAuthAllowRedirect);
+
+PULSAR_PUBLIC int pulsar_client_configuration_is_http_lookup_auth_allow_redirect(
+    pulsar_client_configuration_t *conf);
+
 #ifdef __cplusplus
 }
 #endif

lib/ClientConfiguration.cc

@@ -231,4 +231,11 @@ ClientConfiguration& ClientConfiguration::setDescription(const std::string& desc
 
 const std::string& ClientConfiguration::getDescription() const noexcept { return impl_->description; }
 
+ClientConfiguration& ClientConfiguration::setHttpLookupAuthAllowRedirect(bool allow) {
+    impl_->httpLookupAuthAllowRedirect = allow;
+    return *this;
+}
+
+bool ClientConfiguration::isHttpLookupAuthAllowRedirect() const { return impl_->httpLookupAuthAllowRedirect; }
+
 }  // namespace pulsar

lib/ClientConfigurationImpl.h

@@ -53,6 +53,7 @@ struct ClientConfigurationImpl {
     std::string description;
     std::string proxyServiceUrl;
     ClientConfiguration::ProxyProtocol proxyProtocol;
+    bool httpLookupAuthAllowRedirect{false};
 
     std::unique_ptr<LoggerFactory> takeLogger() { return std::move(loggerFactory); }
 

lib/CurlWrapper.h

@@ -54,6 +54,7 @@ class CurlWrapper {
         std::string userAgent;
         int timeoutInSeconds{0};
         int maxLookupRedirects{-1};
+        bool authAllowRedirect{false};
     };
 
     struct TlsContext {
@@ -128,6 +129,9 @@ inline CurlWrapper::Result CurlWrapper::get(const std::string& url, const std::s
     // Redirects
     curl_easy_setopt(handle_, CURLOPT_FOLLOWLOCATION, 1L);
     curl_easy_setopt(handle_, CURLOPT_MAXREDIRS, options.maxLookupRedirects);
+    if (options.authAllowRedirect) {
+        curl_easy_setopt(handle_, CURLOPT_UNRESTRICTED_AUTH, 1L);
+    }
 
     char errorBuffer[CURL_ERROR_SIZE] = "";
     curl_easy_setopt(handle_, CURLOPT_ERRORBUFFER, errorBuffer);

lib/HTTPLookupService.cc

@@ -61,7 +61,8 @@ HTTPLookupService::HTTPLookupService(const ServiceInfo &serviceInfo,
       tlsTrustCertsFilePath_(serviceInfo.tlsTrustCertsFilePath().value_or("")),
       isUseTls_(serviceInfo.useTls()),
       tlsAllowInsecure_(clientConfiguration.isTlsAllowInsecureConnection()),
-      tlsValidateHostname_(clientConfiguration.isValidateHostName()) {}
+      tlsValidateHostname_(clientConfiguration.isValidateHostName()),
+      httpLookupAuthAllowRedirect_(clientConfiguration.isHttpLookupAuthAllowRedirect()) {}
 
 auto HTTPLookupService::getBroker(const TopicName &topicName) -> LookupResultFuture {
     LookupResultPromise promise;
@@ -228,6 +229,7 @@ Error HTTPLookupService::sendHTTPRequest(const std::string &completeUrl, std::st
     options.timeoutInSeconds = lookupTimeoutInSeconds_;
     options.userAgent = std::string("Pulsar-CPP-v") + PULSAR_VERSION_STR;
     options.maxLookupRedirects = maxLookupRedirects_;
+    options.authAllowRedirect = httpLookupAuthAllowRedirect_;
     auto result = curl.get(completeUrl, authDataContent->getHttpHeaders(), options, tlsContext.get());
 
     responseData = result.responseData;

lib/HTTPLookupService.h

@@ -54,6 +54,7 @@ class HTTPLookupService : public LookupService, public std::enable_shared_from_t
     bool isUseTls_;
     bool tlsAllowInsecure_;
     bool tlsValidateHostname_;
+    bool httpLookupAuthAllowRedirect_;
 
     static LookupDataResultPtr parsePartitionData(const std::string&);
     static LookupDataResultPtr parseLookupData(const std::string&);

lib/c/c_ClientConfiguration.cc

@@ -204,3 +204,12 @@ unsigned int pulsar_client_configuration_get_keep_alive_interval_in_seconds(
     pulsar_client_configuration_t *conf) {
     return conf->conf.getKeepAliveIntervalInSeconds();
 }
+
+void pulsar_client_configuration_set_http_lookup_auth_allow_redirect(pulsar_client_configuration_t *conf,
+                                                                     int httpLookupAuthAllowRedirect) {
+    conf->conf.setHttpLookupAuthAllowRedirect(httpLookupAuthAllowRedirect);
+}
+
+int pulsar_client_configuration_is_http_lookup_auth_allow_redirect(pulsar_client_configuration_t *conf) {
+    return conf->conf.isHttpLookupAuthAllowRedirect();
+}

tests/c/c_ClientConfigurationTest.cc

@@ -37,4 +37,12 @@ TEST(C_ClientConfigurationTest, testCApiConfig) {
 
     pulsar_client_configuration_set_keep_alive_interval_in_seconds(conf, 60);
     ASSERT_EQ(pulsar_client_configuration_get_keep_alive_interval_in_seconds(conf), 60);
+
+    ASSERT_EQ(pulsar_client_configuration_is_http_lookup_auth_allow_redirect(conf), 0);
+
+    pulsar_client_configuration_set_http_lookup_auth_allow_redirect(conf, 1);
+    ASSERT_EQ(pulsar_client_configuration_is_http_lookup_auth_allow_redirect(conf), 1);
+
+    pulsar_client_configuration_set_http_lookup_auth_allow_redirect(conf, 0);
+    ASSERT_EQ(pulsar_client_configuration_is_http_lookup_auth_allow_redirect(conf), 0);
 }