This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new b2b62cb5716b [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 (#28050)
b2b62cb5716b is described below
commit b2b62cb5716bee929418445603dffdb17aced73a
Author: Andrea Cosentino <[email protected]>
AuthorDate: Sun Oct 18 21:06:41 2026 +0200
[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 (#28050)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../client/ElasticsearchRestClientProducer.java | 38 +++++--
.../rest/client/NodeScopedCredentialsProvider.java | 81 +++++++++++++++
...ticsearchRestClientBasicAuthenticationTest.java | 95 +++++++++++++++++
...ticsearchRestClientProducerCredentialsTest.java | 81 +++++++++++++++
.../client/NodeScopedCredentialsProviderTest.java | 103 +++++++++++++++++++
.../camel/component/es/ElasticsearchProducer.java | 35 +++++--
.../es/NodeScopedCredentialsProvider.java | 81 +++++++++++++++
.../es/ElasticsearchProducerCredentialsTest.java | 83 +++++++++++++++
.../es/NodeScopedCredentialsProviderTest.java | 103 +++++++++++++++++++
.../opensearch/NodeScopedCredentialsProvider.java | 68 +++++++++++++
.../component/opensearch/OpensearchProducer.java | 37 +++++--
.../NodeScopedCredentialsProviderTest.java | 94 +++++++++++++++++
.../OpensearchBasicAuthenticationTest.java | 112 +++++++++++++++++++++
.../OpensearchProducerCredentialsTest.java | 82 +++++++++++++++
14 files changed, 1073 insertions(+), 20 deletions(-)
diff --git
a/components/camel-elasticsearch-rest-client/src/main/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientProducer.java
b/components/camel-elasticsearch-rest-client/src/main/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientProducer.java
index 68aeea3cd4cb..401e78593f90 100644
---
a/components/camel-elasticsearch-rest-client/src/main/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientProducer.java
+++
b/components/camel-elasticsearch-rest-client/src/main/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientProducer.java
@@ -22,8 +22,11 @@ import java.security.KeyStore;
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.ObjectHelper;
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 @@ public class ElasticsearchRestClientProducer extends
DefaultAsyncProducer {
* 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 @@ public class ElasticsearchRestClientProducer extends
DefaultAsyncProducer {
return httpClientBuilder;
});
final RestClient restClient = builder.build();
+ restClientRef.set(restClient);
// initiate Sniffer
if (this.endpoint.isEnableSniffer()) {
@@ -413,6 +418,25 @@ public class ElasticsearchRestClientProducer extends
DefaultAsyncProducer {
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(
diff --git
a/components/camel-elasticsearch-rest-client/src/main/java/org/apache/camel/component/elasticsearch/rest/client/NodeScopedCredentialsProvider.java
b/components/camel-elasticsearch-rest-client/src/main/java/org/apache/camel/component/elasticsearch/rest/client/NodeScopedCredentialsProvider.java
new file mode 100644
index 000000000000..6ef1fc26690a
--- /dev/null
+++
b/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;
+ }
+}
diff --git
a/components/camel-elasticsearch-rest-client/src/test/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientBasicAuthenticationTest.java
b/components/camel-elasticsearch-rest-client/src/test/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientBasicAuthenticationTest.java
new file mode 100644
index 000000000000..58b056d476f4
--- /dev/null
+++
b/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();
+ }
+ }
+}
diff --git
a/components/camel-elasticsearch-rest-client/src/test/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientProducerCredentialsTest.java
b/components/camel-elasticsearch-rest-client/src/test/java/org/apache/camel/component/elasticsearch/rest/client/ElasticsearchRestClientProducerCredentialsTest.java
new file mode 100644
index 000000000000..c45e1fb9c12f
--- /dev/null
+++
b/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");
+ }
+}
diff --git
a/components/camel-elasticsearch-rest-client/src/test/java/org/apache/camel/component/elasticsearch/rest/client/NodeScopedCredentialsProviderTest.java
b/components/camel-elasticsearch-rest-client/src/test/java/org/apache/camel/component/elasticsearch/rest/client/NodeScopedCredentialsProviderTest.java
new file mode 100644
index 000000000000..32a6447b38d2
--- /dev/null
+++
b/components/camel-elasticsearch-rest-client/src/test/java/org/apache/camel/component/elasticsearch/rest/client/NodeScopedCredentialsProviderTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.ArrayList;
+import java.util.List;
+
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.Credentials;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+class NodeScopedCredentialsProviderTest {
+
+ private static final Credentials CREDENTIALS = new
UsernamePasswordCredentials("elastic", "secret");
+
+ private final List<HttpHost> nodes = new ArrayList<>(
+ List.of(new HttpHost("es1.example.com", 9200, "http"),
+ new HttpHost("es2.example.com", 9243, "https"),
+ new HttpHost("es3.example.com", -1, "https")));
+
+ private final NodeScopedCredentialsProvider provider = new
NodeScopedCredentialsProvider(CREDENTIALS, () -> nodes);
+
+ @Test
+ void offersTheCredentialsToEveryNode() {
+ assertSame(CREDENTIALS, provider.getCredentials(scope("http",
"es1.example.com", 9200)));
+ assertSame(CREDENTIALS, provider.getCredentials(scope("https",
"es2.example.com", 9243)));
+ }
+
+ @Test
+ void usesTheDefaultPortOfTheSchemeWhenTheNodeHasNone() {
+ assertSame(CREDENTIALS, provider.getCredentials(scope("https",
"es3.example.com", 443)));
+ assertNull(provider.getCredentials(scope("https", "es3.example.com",
9200)));
+ }
+
+ @Test
+ void matchesTheHostNameRegardlessOfCase() {
+ assertSame(CREDENTIALS, provider.getCredentials(scope("HTTP",
"ES1.Example.COM", 9200)));
+ }
+
+ @Test
+ void offersNoCredentialsToAHostThatIsNotANode() {
+ assertNull(provider.getCredentials(scope("http", "other.example.com",
9200)));
+ }
+
+ @Test
+ void offersNoCredentialsToAnotherPortOrSchemeOfANode() {
+ assertNull(provider.getCredentials(scope("http", "es1.example.com",
9300)));
+ assertNull(provider.getCredentials(scope("https", "es1.example.com",
9200)));
+ assertNull(provider.getCredentials(scope("http", "es2.example.com",
9243)));
+ }
+
+ @Test
+ void followsTheCurrentNodes() {
+ HttpHost sniffed = new HttpHost("es4.example.com", 9200, "http");
+ assertNull(provider.getCredentials(scope("http", "es4.example.com",
9200)));
+
+ nodes.add(sniffed);
+ assertSame(CREDENTIALS, provider.getCredentials(scope("http",
"es4.example.com", 9200)));
+
+ nodes.remove(sniffed);
+ assertNull(provider.getCredentials(scope("http", "es4.example.com",
9200)));
+ }
+
+ @Test
+ void offersNoCredentialsToAScopeWithoutAnOrigin() {
+ assertNull(provider.getCredentials(AuthScope.ANY));
+ assertNull(provider.getCredentials(new AuthScope("es1.example.com",
9200)));
+ assertNull(provider.getCredentials(null));
+ }
+
+ @Test
+ void ignoresCredentialsSetAfterwards() {
+ provider.setCredentials(scope("http", "other.example.com", 9200), new
UsernamePasswordCredentials("user", "pass"));
+ assertNull(provider.getCredentials(scope("http", "other.example.com",
9200)));
+
+ provider.clear();
+ assertSame(CREDENTIALS, provider.getCredentials(scope("http",
"es1.example.com", 9200)));
+ }
+
+ private static AuthScope scope(String scheme, String host, int port) {
+ // the shape of the scope HttpClient looks the credentials up with,
preemptively or after a challenge
+ return new AuthScope(new HttpHost(host, port, scheme), "realm",
"basic");
+ }
+}
diff --git
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
index 20027487a64a..03a6dae53561 100644
---
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
+++
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
@@ -20,7 +20,11 @@ import java.io.InputStream;
import java.security.KeyStore;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
+import java.util.Collection;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;
@@ -59,10 +63,9 @@ import org.apache.camel.support.DefaultAsyncProducer;
import org.apache.camel.support.ResourceHelper;
import org.apache.camel.util.IOHelper;
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.impl.client.BasicCredentialsProvider;
+import org.elasticsearch.client.Node;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
@@ -509,14 +512,14 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
private RestClient createClient() {
final RestClientBuilder builder =
RestClient.builder(configuration.getHostAddressesList().toArray(new
HttpHost[0]));
+ // 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(configuration.getConnectionTimeout()).setSocketTimeout(configuration.getSocketTimeout()));
builder.setHttpClientConfigCallback(httpClientBuilder -> {
- if (configuration.getUser() != null && configuration.getPassword()
!= null) {
- final CredentialsProvider credentialsProvider = new
BasicCredentialsProvider();
- credentialsProvider.setCredentials(AuthScope.ANY,
- new
UsernamePasswordCredentials(configuration.getUser(),
configuration.getPassword()));
+ CredentialsProvider credentialsProvider =
createCredentialsProvider(() -> nodeHosts(restClientRef.get()));
+ if (credentialsProvider != null) {
httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
}
if (configuration.getSslContextParameters() != null) {
@@ -534,6 +537,7 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
return httpClientBuilder;
});
final RestClient restClient = builder.build();
+ restClientRef.set(restClient);
if (configuration.isEnableSniffer()) {
SnifferBuilder snifferBuilder = Sniffer.builder(restClient);
snifferBuilder.setSniffIntervalMillis(configuration.getSnifferInterval());
@@ -543,6 +547,25 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
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 (configuration.getUser() == null || configuration.getPassword() ==
null) {
+ return null;
+ }
+ return new NodeScopedCredentialsProvider(
+ new UsernamePasswordCredentials(configuration.getUser(),
configuration.getPassword()), nodes);
+ }
+
+ private List<HttpHost> nodeHosts(RestClient restClient) {
+ if (restClient == null) {
+ return configuration.getHostAddressesList();
+ }
+ return restClient.getNodes().stream().map(Node::getHost).toList();
+ }
+
@Override
protected void doStop() throws Exception {
if (client != null) {
diff --git
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/NodeScopedCredentialsProvider.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/NodeScopedCredentialsProvider.java
new file mode 100644
index 000000000000..d4a2fac52a04
--- /dev/null
+++
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/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.es;
+
+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;
+ }
+}
diff --git
a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/ElasticsearchProducerCredentialsTest.java
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/ElasticsearchProducerCredentialsTest.java
new file mode 100644
index 000000000000..3f619e96bd0b
--- /dev/null
+++
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/ElasticsearchProducerCredentialsTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.es;
+
+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 ElasticsearchProducerCredentialsTest {
+
+ @Test
+ void offersTheConfiguredCredentialsOnlyToTheConfiguredHosts() throws
Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ ElasticsearchConfiguration configuration
+ = configuration(context,
"es1.example.com:9200,es2.example.com:9201", "elastic", "secret");
+
+ CredentialsProvider provider = new
ElasticsearchProducer(endpoint(context), configuration)
+
.createCredentialsProvider(configuration::getHostAddressesList);
+
+ 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("http",
"es2.example.com", 9201)));
+
+ assertNull(provider.getCredentials(scope("http",
"other.example.com", 9200)));
+ assertNull(provider.getCredentials(scope("http",
"es1.example.com", 9201)));
+ assertNull(provider.getCredentials(scope("https",
"es1.example.com", 9200)));
+ }
+ }
+
+ @Test
+ void createsNoProviderWithoutCredentials() throws Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ ElasticsearchConfiguration configuration = configuration(context,
"es1.example.com:9200", null, null);
+
+ assertNull(new ElasticsearchProducer(endpoint(context),
configuration)
+
.createCredentialsProvider(configuration::getHostAddressesList));
+ }
+ }
+
+ private static ElasticsearchConfiguration configuration(
+ CamelContext context, String hostAddresses, String user, String
password) {
+ ElasticsearchComponent component = new ElasticsearchComponent(context);
+ component.setHostAddresses(hostAddresses);
+ component.setUser(user);
+ component.setPassword(password);
+ context.addComponent("elasticsearch", component);
+ return endpoint(context).getConfiguration();
+ }
+
+ private static ElasticsearchEndpoint endpoint(CamelContext context) {
+ return context.getEndpoint("elasticsearch:test-cluster",
ElasticsearchEndpoint.class);
+ }
+
+ private static AuthScope scope(String scheme, String host, int port) {
+ return new AuthScope(new HttpHost(host, port, scheme), "realm",
"basic");
+ }
+}
diff --git
a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/NodeScopedCredentialsProviderTest.java
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/NodeScopedCredentialsProviderTest.java
new file mode 100644
index 000000000000..b23e2e10ef79
--- /dev/null
+++
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/NodeScopedCredentialsProviderTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.es;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.Credentials;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+class NodeScopedCredentialsProviderTest {
+
+ private static final Credentials CREDENTIALS = new
UsernamePasswordCredentials("elastic", "secret");
+
+ private final List<HttpHost> nodes = new ArrayList<>(
+ List.of(new HttpHost("es1.example.com", 9200, "http"),
+ new HttpHost("es2.example.com", 9243, "https"),
+ new HttpHost("es3.example.com", -1, "https")));
+
+ private final NodeScopedCredentialsProvider provider = new
NodeScopedCredentialsProvider(CREDENTIALS, () -> nodes);
+
+ @Test
+ void offersTheCredentialsToEveryNode() {
+ assertSame(CREDENTIALS, provider.getCredentials(scope("http",
"es1.example.com", 9200)));
+ assertSame(CREDENTIALS, provider.getCredentials(scope("https",
"es2.example.com", 9243)));
+ }
+
+ @Test
+ void usesTheDefaultPortOfTheSchemeWhenTheNodeHasNone() {
+ assertSame(CREDENTIALS, provider.getCredentials(scope("https",
"es3.example.com", 443)));
+ assertNull(provider.getCredentials(scope("https", "es3.example.com",
9200)));
+ }
+
+ @Test
+ void matchesTheHostNameRegardlessOfCase() {
+ assertSame(CREDENTIALS, provider.getCredentials(scope("HTTP",
"ES1.Example.COM", 9200)));
+ }
+
+ @Test
+ void offersNoCredentialsToAHostThatIsNotANode() {
+ assertNull(provider.getCredentials(scope("http", "other.example.com",
9200)));
+ }
+
+ @Test
+ void offersNoCredentialsToAnotherPortOrSchemeOfANode() {
+ assertNull(provider.getCredentials(scope("http", "es1.example.com",
9300)));
+ assertNull(provider.getCredentials(scope("https", "es1.example.com",
9200)));
+ assertNull(provider.getCredentials(scope("http", "es2.example.com",
9243)));
+ }
+
+ @Test
+ void followsTheCurrentNodes() {
+ HttpHost sniffed = new HttpHost("es4.example.com", 9200, "http");
+ assertNull(provider.getCredentials(scope("http", "es4.example.com",
9200)));
+
+ nodes.add(sniffed);
+ assertSame(CREDENTIALS, provider.getCredentials(scope("http",
"es4.example.com", 9200)));
+
+ nodes.remove(sniffed);
+ assertNull(provider.getCredentials(scope("http", "es4.example.com",
9200)));
+ }
+
+ @Test
+ void offersNoCredentialsToAScopeWithoutAnOrigin() {
+ assertNull(provider.getCredentials(AuthScope.ANY));
+ assertNull(provider.getCredentials(new AuthScope("es1.example.com",
9200)));
+ assertNull(provider.getCredentials(null));
+ }
+
+ @Test
+ void ignoresCredentialsSetAfterwards() {
+ provider.setCredentials(scope("http", "other.example.com", 9200), new
UsernamePasswordCredentials("user", "pass"));
+ assertNull(provider.getCredentials(scope("http", "other.example.com",
9200)));
+
+ provider.clear();
+ assertSame(CREDENTIALS, provider.getCredentials(scope("http",
"es1.example.com", 9200)));
+ }
+
+ private static AuthScope scope(String scheme, String host, int port) {
+ // the shape of the scope HttpClient looks the credentials up with,
preemptively or after a challenge
+ return new AuthScope(new HttpHost(host, port, scheme), "realm",
"basic");
+ }
+}
diff --git
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/NodeScopedCredentialsProvider.java
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/NodeScopedCredentialsProvider.java
new file mode 100644
index 000000000000..c03030907bfc
--- /dev/null
+++
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/NodeScopedCredentialsProvider.java
@@ -0,0 +1,68 @@
+/*
+ * 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.opensearch;
+
+import java.util.Collection;
+import java.util.function.Supplier;
+
+import org.apache.hc.client5.http.auth.AuthScope;
+import org.apache.hc.client5.http.auth.Credentials;
+import org.apache.hc.client5.http.auth.CredentialsProvider;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.protocol.HttpContext;
+
+/**
+ * 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, HttpContext
context) {
+ if (authScope != null && authScope.getProtocol() != null &&
authScope.getHost() != null) {
+ for (HttpHost node : nodes.get()) {
+ if (sameAuthority(node, authScope)) {
+ return credentials;
+ }
+ }
+ }
+ return null;
+ }
+
+ private static boolean sameAuthority(HttpHost node, AuthScope authScope) {
+ return node.getSchemeName().equalsIgnoreCase(authScope.getProtocol())
+ && node.getHostName().equalsIgnoreCase(authScope.getHost())
+ && effectivePort(node.getSchemeName(), node.getPort())
+ == effectivePort(authScope.getProtocol(),
authScope.getPort());
+ }
+
+ private static int effectivePort(String scheme, int port) {
+ if (port >= 0) {
+ return port;
+ }
+ return "https".equalsIgnoreCase(scheme) ? 443 : 80;
+ }
+}
diff --git
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java
index 2d96335235bf..b490935cd064 100644
---
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java
+++
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java
@@ -22,7 +22,11 @@ import java.security.KeyStore;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;
@@ -37,16 +41,16 @@ import org.apache.camel.support.DefaultAsyncProducer;
import org.apache.camel.support.ResourceHelper;
import org.apache.camel.util.IOHelper;
import org.apache.camel.util.ObjectHelper;
-import org.apache.hc.client5.http.auth.AuthScope;
+import org.apache.hc.client5.http.auth.CredentialsProvider;
import org.apache.hc.client5.http.auth.UsernamePasswordCredentials;
import org.apache.hc.client5.http.config.ConnectionConfig;
-import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
import
org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
import org.apache.hc.client5.http.ssl.ClientTlsStrategyBuilder;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
import org.apache.hc.core5.util.Timeout;
+import org.opensearch.client.Node;
import org.opensearch.client.RestClient;
import org.opensearch.client.RestClientBuilder;
import org.opensearch.client.json.jackson.JacksonJsonpMapper;
@@ -495,17 +499,16 @@ class OpensearchProducer extends DefaultAsyncProducer {
private RestClient createClient() {
final RestClientBuilder builder =
RestClient.builder(configuration.getHostAddressesList().toArray(new
HttpHost[0]));
+ // 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(Timeout.of(Duration.ofMillis(configuration.getConnectionTimeout())))
// apply the socket/read timeout for both plain-HTTP and SSL
connections, not only when SSL is enabled
.setResponseTimeout(Timeout.of(Duration.ofMillis(configuration.getSocketTimeout()))));
builder.setHttpClientConfigCallback(httpClientBuilder -> {
- if (ObjectHelper.isNotEmpty(configuration.getUser()) &&
ObjectHelper.isNotEmpty(configuration.getPassword())) {
- final BasicCredentialsProvider credentialsProvider = new
BasicCredentialsProvider();
- // match-all AuthScope so basic auth is sent to every node,
not only the first configured host
- credentialsProvider.setCredentials(new AuthScope(null, null,
-1, null, null),
- new
UsernamePasswordCredentials(configuration.getUser(),
configuration.getPassword().toCharArray()));
+ CredentialsProvider credentialsProvider =
createCredentialsProvider(() -> nodeHosts(restClientRef.get()));
+ if (credentialsProvider != null) {
httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
}
@@ -558,6 +561,7 @@ class OpensearchProducer extends DefaultAsyncProducer {
});
final RestClient restClient = builder.build();
+ restClientRef.set(restClient);
if (configuration.isEnableSniffer()) {
SnifferBuilder snifferBuilder = Sniffer.builder(restClient);
snifferBuilder.setSniffIntervalMillis(configuration.getSnifferInterval());
@@ -567,6 +571,25 @@ class OpensearchProducer extends DefaultAsyncProducer {
return restClient;
}
+ /**
+ * Creates the provider of the configured user and password, which offers
them only to the given nodes - every node
+ * of the cluster, not only the first configured host - or returns null
when no credentials are configured.
+ */
+ CredentialsProvider createCredentialsProvider(Supplier<? extends
Collection<HttpHost>> nodes) {
+ if (ObjectHelper.isEmpty(configuration.getUser()) ||
ObjectHelper.isEmpty(configuration.getPassword())) {
+ return null;
+ }
+ return new NodeScopedCredentialsProvider(
+ new UsernamePasswordCredentials(configuration.getUser(),
configuration.getPassword().toCharArray()), nodes);
+ }
+
+ private List<HttpHost> nodeHosts(RestClient restClient) {
+ if (restClient == null) {
+ return configuration.getHostAddressesList();
+ }
+ return restClient.getNodes().stream().map(Node::getHost).toList();
+ }
+
@Override
protected void doStop() throws Exception {
if (client != null && openSearchClient == null) {
diff --git
a/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/NodeScopedCredentialsProviderTest.java
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/NodeScopedCredentialsProviderTest.java
new file mode 100644
index 000000000000..2f2df73ed3e3
--- /dev/null
+++
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/NodeScopedCredentialsProviderTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.opensearch;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hc.client5.http.auth.AuthScope;
+import org.apache.hc.client5.http.auth.Credentials;
+import org.apache.hc.client5.http.auth.UsernamePasswordCredentials;
+import org.apache.hc.core5.http.HttpHost;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+class NodeScopedCredentialsProviderTest {
+
+ private static final Credentials CREDENTIALS = new
UsernamePasswordCredentials("admin", "secret".toCharArray());
+
+ private final List<HttpHost> nodes = new ArrayList<>(
+ List.of(new HttpHost("http", "os1.example.com", 9200),
+ new HttpHost("https", "os2.example.com", 9243),
+ new HttpHost("https", "os3.example.com", -1)));
+
+ private final NodeScopedCredentialsProvider provider = new
NodeScopedCredentialsProvider(CREDENTIALS, () -> nodes);
+
+ @Test
+ void offersTheCredentialsToEveryNode() {
+ assertSame(CREDENTIALS, credentialsFor("http", "os1.example.com",
9200));
+ assertSame(CREDENTIALS, credentialsFor("https", "os2.example.com",
9243));
+ }
+
+ @Test
+ void usesTheDefaultPortOfTheSchemeWhenTheNodeHasNone() {
+ assertSame(CREDENTIALS, credentialsFor("https", "os3.example.com",
443));
+ assertNull(credentialsFor("https", "os3.example.com", 9200));
+ }
+
+ @Test
+ void matchesTheHostNameRegardlessOfCase() {
+ assertSame(CREDENTIALS, credentialsFor("HTTP", "OS1.Example.COM",
9200));
+ }
+
+ @Test
+ void offersNoCredentialsToAHostThatIsNotANode() {
+ assertNull(credentialsFor("http", "other.example.com", 9200));
+ }
+
+ @Test
+ void offersNoCredentialsToAnotherPortOrSchemeOfANode() {
+ assertNull(credentialsFor("http", "os1.example.com", 9300));
+ assertNull(credentialsFor("https", "os1.example.com", 9200));
+ assertNull(credentialsFor("http", "os2.example.com", 9243));
+ }
+
+ @Test
+ void followsTheCurrentNodes() {
+ HttpHost sniffed = new HttpHost("http", "os4.example.com", 9200);
+ assertNull(credentialsFor("http", "os4.example.com", 9200));
+
+ nodes.add(sniffed);
+ assertSame(CREDENTIALS, credentialsFor("http", "os4.example.com",
9200));
+
+ nodes.remove(sniffed);
+ assertNull(credentialsFor("http", "os4.example.com", 9200));
+ }
+
+ @Test
+ void offersNoCredentialsToAScopeWithoutProtocolOrHost() {
+ assertNull(provider.getCredentials(new AuthScope(null, null, -1, null,
null), null));
+ assertNull(provider.getCredentials(new AuthScope("os1.example.com",
9200), null));
+ assertNull(provider.getCredentials(null, null));
+ }
+
+ private Credentials credentialsFor(String scheme, String host, int port) {
+ // the shape of the scope HttpClient looks the credentials up with,
preemptively or after a challenge
+ return provider.getCredentials(new AuthScope(new HttpHost(scheme,
host, port), "realm", "Basic"), null);
+ }
+}
diff --git
a/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/OpensearchBasicAuthenticationTest.java
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/OpensearchBasicAuthenticationTest.java
new file mode 100644
index 000000000000..2b99695c56ec
--- /dev/null
+++
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/OpensearchBasicAuthenticationTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.opensearch;
+
+import java.io.IOException;
+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.HttpServer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+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 every node it is configured with, through the
client it creates itself.
+ */
+class OpensearchBasicAuthenticationTest {
+
+ private static final String AUTHORIZATION
+ = "Basic " +
Base64.getEncoder().encodeToString("admin:secret".getBytes(StandardCharsets.UTF_8));
+
+ private final ClusterNode node1 = new ClusterNode();
+ private final ClusterNode node2 = new ClusterNode();
+
+ @BeforeEach
+ void startNodes() throws IOException {
+ node1.start();
+ node2.start();
+ }
+
+ @AfterEach
+ void stopNodes() {
+ node1.stop();
+ node2.stop();
+ }
+
+ @Test
+ void sendsTheConfiguredCredentialsToEveryNode() throws Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+ OpensearchComponent component = new OpensearchComponent(context);
+ component.setHostAddresses(node1.address() + "," +
node2.address());
+ component.setUser("admin");
+ component.setPassword("secret");
+ context.addComponent("opensearch", component);
+ context.start();
+
+ // the client sends the requests to the nodes in turn
+ ProducerTemplate template = context.createProducerTemplate();
+ assertEquals(Boolean.TRUE,
template.requestBody("opensearch:test-cluster?operation=Ping", (Object) null));
+ assertEquals(Boolean.TRUE,
template.requestBody("opensearch:test-cluster?operation=Ping", (Object) null));
+
+ assertEquals(List.of(AUTHORIZATION), node1.authorizations);
+ assertEquals(List.of(AUTHORIZATION), node2.authorizations);
+ }
+ }
+
+ /**
+ * A node that answers a ping with the configured credentials, and asks
for them otherwise.
+ */
+ private static final class ClusterNode {
+
+ private final List<String> authorizations = new
CopyOnWriteArrayList<>();
+ private HttpServer server;
+
+ void start() throws IOException {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0),
0);
+ server.createContext("/", exchange -> {
+ exchange.getRequestBody().readAllBytes();
+ String authorization =
exchange.getRequestHeaders().getFirst("Authorization");
+ authorizations.add(String.valueOf(authorization));
+ if (AUTHORIZATION.equals(authorization)) {
+ exchange.sendResponseHeaders(200, -1);
+ } else {
+ exchange.getResponseHeaders().add("WWW-Authenticate",
"Basic realm=\"camel\"");
+ exchange.sendResponseHeaders(401, -1);
+ }
+ exchange.close();
+ });
+ server.start();
+ }
+
+ String address() {
+ return "127.0.0.1:" + server.getAddress().getPort();
+ }
+
+ void stop() {
+ server.stop(0);
+ }
+ }
+}
diff --git
a/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/OpensearchProducerCredentialsTest.java
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/OpensearchProducerCredentialsTest.java
new file mode 100644
index 000000000000..1f89865ca829
--- /dev/null
+++
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/OpensearchProducerCredentialsTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.opensearch;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.hc.client5.http.auth.AuthScope;
+import org.apache.hc.client5.http.auth.Credentials;
+import org.apache.hc.client5.http.auth.CredentialsProvider;
+import org.apache.hc.core5.http.HttpHost;
+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 OpensearchProducerCredentialsTest {
+
+ @Test
+ void offersTheConfiguredCredentialsToEveryConfiguredHostOnly() throws
Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ OpensearchConfiguration configuration
+ = configuration(context,
"os1.example.com:9200,os2.example.com:9201", "admin", "secret");
+
+ CredentialsProvider provider = new
OpensearchProducer(endpoint(context), configuration)
+
.createCredentialsProvider(configuration::getHostAddressesList);
+
+ Credentials credentials = provider.getCredentials(scope("http",
"os1.example.com", 9200), null);
+ assertNotNull(credentials);
+ assertEquals("admin", credentials.getUserPrincipal().getName());
+ assertNotNull(provider.getCredentials(scope("http",
"os2.example.com", 9201), null));
+
+ assertNull(provider.getCredentials(scope("http",
"other.example.com", 9200), null));
+ assertNull(provider.getCredentials(scope("http",
"os1.example.com", 9201), null));
+ assertNull(provider.getCredentials(scope("https",
"os1.example.com", 9200), null));
+ }
+ }
+
+ @Test
+ void createsNoProviderWithoutCredentials() throws Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ OpensearchConfiguration configuration = configuration(context,
"os1.example.com:9200", null, null);
+
+ assertNull(new OpensearchProducer(endpoint(context), configuration)
+
.createCredentialsProvider(configuration::getHostAddressesList));
+ }
+ }
+
+ private static OpensearchConfiguration configuration(
+ CamelContext context, String hostAddresses, String user, String
password) {
+ OpensearchComponent component = new OpensearchComponent(context);
+ component.setHostAddresses(hostAddresses);
+ component.setUser(user);
+ component.setPassword(password);
+ context.addComponent("opensearch", component);
+ return endpoint(context).getConfiguration();
+ }
+
+ private static OpensearchEndpoint endpoint(CamelContext context) {
+ return context.getEndpoint("opensearch:test-cluster",
OpensearchEndpoint.class);
+ }
+
+ private static AuthScope scope(String scheme, String host, int port) {
+ return new AuthScope(new HttpHost(scheme, host, port), "realm",
"Basic");
+ }
+}
