[backport camel-4.22.x] CAMEL-25706, CAMEL-25707, CAMEL-25708: camel-elasticsearch, camel-elasticsearch-rest-client, camel-opensearch - bind the basic authentication credentials to the cluster nodes

components/camel-elasticsearch-rest-client/src/main/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientProducer.java

@@ -22,8 +22,11 @@
 import java.security.cert.Certificate;
 import java.security.cert.CertificateFactory;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import javax.net.ssl.SSLContext;
@@ -44,13 +47,12 @@
 import org.apache.camel.util.json.JsonArray;
 import org.apache.camel.util.json.JsonObject;
 import org.apache.http.HttpHost;
-import org.apache.http.auth.AuthScope;
 import org.apache.http.auth.UsernamePasswordCredentials;
 import org.apache.http.client.CredentialsProvider;
 import org.apache.http.entity.ContentType;
-import org.apache.http.impl.client.BasicCredentialsProvider;
 import org.apache.http.nio.entity.NStringEntity;
 import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Node;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseListener;
@@ -375,16 +377,18 @@ String extractSearch(JsonObject doc) {
      * Creates Rest Client from information in the Endpoint
      */
     private RestClient createClient() throws Exception {
-        final RestClientBuilder builder = RestClient.builder(getHttpHosts());
+        final HttpHost[] httpHosts = getHttpHosts();
+        final RestClientBuilder builder = RestClient.builder(httpHosts);
+        // the credentials provider reads the nodes from the client, as the sniffer can change them
+        final AtomicReference<RestClient> restClientRef = new AtomicReference<>();
 
         builder.setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder
                 .setConnectTimeout(this.endpoint.getConnectionTimeout()).setSocketTimeout(this.endpoint.getSocketTimeout()));
 
         builder.setHttpClientConfigCallback(httpClientBuilder -> {
-            if (this.endpoint.getUser() != null && this.endpoint.getPassword() != null) {
-                final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
-                credentialsProvider.setCredentials(AuthScope.ANY,
-                        new UsernamePasswordCredentials(this.endpoint.getUser(), this.endpoint.getPassword()));
+            CredentialsProvider credentialsProvider
+                    = createCredentialsProvider(() -> nodeHosts(restClientRef.get(), httpHosts));
+            if (credentialsProvider != null) {
                 httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
             }
             if (this.endpoint.getSslContextParameters() != null) {
@@ -402,6 +406,7 @@ private RestClient createClient() throws Exception {
             return httpClientBuilder;
         });
         final RestClient restClient = builder.build();
+        restClientRef.set(restClient);
 
         // initiate Sniffer
         if (this.endpoint.isEnableSniffer()) {
@@ -413,6 +418,25 @@ private RestClient createClient() throws Exception {
         return restClient;
     }
 
+    /**
+     * Creates the provider of the configured user and password, which offers them only to the given nodes, or returns
+     * null when no credentials are configured.
+     */
+    CredentialsProvider createCredentialsProvider(Supplier<? extends Collection<HttpHost>> nodes) {
+        if (this.endpoint.getUser() == null || this.endpoint.getPassword() == null) {
+            return null;
+        }
+        return new NodeScopedCredentialsProvider(
+                new UsernamePasswordCredentials(this.endpoint.getUser(), this.endpoint.getPassword()), nodes);
+    }
+
+    private static List<HttpHost> nodeHosts(RestClient restClient, HttpHost[] configuredHosts) {
+        if (restClient == null) {
+            return List.of(configuredHosts);
+        }
+        return restClient.getNodes().stream().map(Node::getHost).toList();
+    }
+
     private HttpHost[] getHttpHosts() {
         if (ObjectHelper.isEmpty(this.endpoint.getHostAddressesList())) {
             throw new IllegalArgumentException(

components/camel-elasticsearch-rest-client/src/main/java/org/apache/camel/component/elasticsearch/rest/client/NodeScopedCredentialsProvider.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.elasticsearch.rest.client;
+
+import java.util.Collection;
+import java.util.function.Supplier;
+
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.Credentials;
+import org.apache.http.client.CredentialsProvider;
+
+/**
+ * Offers the configured credentials only to the nodes of the cluster the client talks to: the configured hosts, and the
+ * nodes found by the sniffer when it is enabled. A node matches when the scheme, the host name and the port are the
+ * same, the default port of the scheme standing in for a missing one.
+ */
+final class NodeScopedCredentialsProvider implements CredentialsProvider {
+
+    private final Credentials credentials;
+    private final Supplier<? extends Collection<HttpHost>> nodes;
+
+    NodeScopedCredentialsProvider(Credentials credentials, Supplier<? extends Collection<HttpHost>> nodes) {
+        this.credentials = credentials;
+        this.nodes = nodes;
+    }
+
+    @Override
+    public Credentials getCredentials(AuthScope authScope) {
+        HttpHost origin = authScope != null ? authScope.getOrigin() : null;
+        if (origin != null) {
+            for (HttpHost node : nodes.get()) {
+                if (sameAuthority(node, origin)) {
+                    return credentials;
+                }
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Ignored: the credentials come from the endpoint configuration only. HttpClient calls this for the user info of a
+     * request URI.
+     */
+    @Override
+    public void setCredentials(AuthScope authScope, Credentials credentials) {
+        // read-only
+    }
+
+    @Override
+    public void clear() {
+        // read-only
+    }
+
+    private static boolean sameAuthority(HttpHost node, HttpHost origin) {
+        return node.getSchemeName().equalsIgnoreCase(origin.getSchemeName())
+                && node.getHostName().equalsIgnoreCase(origin.getHostName())
+                && effectivePort(node) == effectivePort(origin);
+    }
+
+    private static int effectivePort(HttpHost host) {
+        if (host.getPort() >= 0) {
+            return host.getPort();
+        }
+        return "https".equalsIgnoreCase(host.getSchemeName()) ? 443 : 80;
+    }
+}

components/camel-elasticsearch-rest-client/src/test/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientBasicAuthenticationTest.java

@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.elasticsearch.rest.client;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The producer authenticates to the node it is configured with, through the client it creates itself.
+ */
+class ElasticsearchRestClientBasicAuthenticationTest {
+
+    private static final String AUTHORIZATION
+            = "Basic " + Base64.getEncoder().encodeToString("elastic:secret".getBytes(StandardCharsets.UTF_8));
+
+    private final List<String> authorizations = new CopyOnWriteArrayList<>();
+    private HttpServer node;
+
+    @BeforeEach
+    void startNode() throws IOException {
+        node = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        node.createContext("/", this::handle);
+        node.start();
+    }
+
+    @AfterEach
+    void stopNode() {
+        node.stop(0);
+    }
+
+    @Test
+    void sendsTheConfiguredCredentialsToTheNode() throws Exception {
+        try (CamelContext context = new DefaultCamelContext()) {
+            ElasticsearchRestClientComponent component = new ElasticsearchRestClientComponent();
+            component.setHostAddressesList("127.0.0.1:" + node.getAddress().getPort());
+            component.setUser("elastic");
+            component.setPassword("secret");
+            context.addComponent("elasticsearch-rest-client", component);
+            context.start();
+
+            Object acknowledged = context.createProducerTemplate()
+                    .requestBody("elasticsearch-rest-client:camel?operation=CREATE_INDEX&indexName=camel", (Object) null);
+
+            assertEquals(Boolean.TRUE, acknowledged);
+            assertEquals(List.of(AUTHORIZATION), authorizations);
+        }
+    }
+
+    private void handle(HttpExchange exchange) throws IOException {
+        exchange.getRequestBody().readAllBytes();
+        String authorization = exchange.getRequestHeaders().getFirst("Authorization");
+        authorizations.add(String.valueOf(authorization));
+        if (AUTHORIZATION.equals(authorization)) {
+            byte[] body = "{\"acknowledged\":true}".getBytes(StandardCharsets.UTF_8);
+            exchange.getResponseHeaders().add("Content-Type", "application/json");
+            exchange.sendResponseHeaders(200, body.length);
+            try (OutputStream out = exchange.getResponseBody()) {
+                out.write(body);
+            }
+        } else {
+            exchange.getResponseHeaders().add("WWW-Authenticate", "Basic realm=\"camel\"");
+            exchange.sendResponseHeaders(401, -1);
+            exchange.close();
+        }
+    }
+}

components/camel-elasticsearch-rest-client/src/test/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientProducerCredentialsTest.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.elasticsearch.rest.client;
+
+import java.util.List;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.Credentials;
+import org.apache.http.client.CredentialsProvider;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class ElasticsearchRestClientProducerCredentialsTest {
+
+    private static final List<HttpHost> NODES
+            = List.of(HttpHost.create("es1.example.com:9200"), HttpHost.create("https://es2.example.com:9243"));
+
+    @Test
+    void offersTheConfiguredCredentialsOnlyToTheNodes() throws Exception {
+        try (CamelContext context = new DefaultCamelContext()) {
+            context.start();
+            ElasticsearchRestClientEndpoint endpoint = endpoint(context, "elastic", "secret");
+
+            CredentialsProvider provider = new ElasticsearchRestClientProducer(endpoint).createCredentialsProvider(() -> NODES);
+
+            Credentials credentials = provider.getCredentials(scope("http", "es1.example.com", 9200));
+            assertNotNull(credentials);
+            assertEquals("elastic", credentials.getUserPrincipal().getName());
+            assertEquals("secret", credentials.getPassword());
+            assertNotNull(provider.getCredentials(scope("https", "es2.example.com", 9243)));
+
+            assertNull(provider.getCredentials(scope("http", "other.example.com", 9200)));
+            assertNull(provider.getCredentials(scope("http", "es1.example.com", 9201)));
+            assertNull(provider.getCredentials(scope("http", "es2.example.com", 9243)));
+        }
+    }
+
+    @Test
+    void createsNoProviderWithoutCredentials() throws Exception {
+        try (CamelContext context = new DefaultCamelContext()) {
+            context.start();
+            ElasticsearchRestClientEndpoint endpoint = endpoint(context, null, null);
+
+            assertNull(new ElasticsearchRestClientProducer(endpoint).createCredentialsProvider(() -> NODES));
+        }
+    }
+
+    private static ElasticsearchRestClientEndpoint endpoint(CamelContext context, String user, String password) {
+        ElasticsearchRestClientComponent component = new ElasticsearchRestClientComponent();
+        component.setHostAddressesList("es1.example.com:9200,https://es2.example.com:9243");
+        component.setUser(user);
+        component.setPassword(password);
+        context.addComponent("elasticsearch-rest-client", component);
+        return context.getEndpoint("elasticsearch-rest-client:camel?operation=CREATE_INDEX",
+                ElasticsearchRestClientEndpoint.class);
+    }
+
+    private static AuthScope scope(String scheme, String host, int port) {
+        return new AuthScope(new HttpHost(host, port, scheme), "realm", "basic");
+    }
+}