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 68aeea3cd4cbf..401e78593f90c 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.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 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> 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 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 0000000000000..6ef1fc26690a0 --- /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> nodes; + + NodeScopedCredentialsProvider(Credentials credentials, Supplier> 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 0000000000000..58b056d476f43 --- /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 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 0000000000000..c45e1fb9c12fe --- /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 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 0000000000000..32a6447b38d24 --- /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 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 20027487a64ae..03a6dae53561c 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.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.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 @@ private void startClient() { 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 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 @@ private RestClient createClient() { 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 @@ private RestClient createClient() { 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> nodes) { + if (configuration.getUser() == null || configuration.getPassword() == null) { + return null; + } + return new NodeScopedCredentialsProvider( + new UsernamePasswordCredentials(configuration.getUser(), configuration.getPassword()), nodes); + } + + private List 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 0000000000000..d4a2fac52a04a --- /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> nodes; + + NodeScopedCredentialsProvider(Credentials credentials, Supplier> 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 0000000000000..3f619e96bd0b8 --- /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 0000000000000..b23e2e10ef792 --- /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 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 0000000000000..c03030907bfc1 --- /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> nodes; + + NodeScopedCredentialsProvider(Credentials credentials, Supplier> 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 2d96335235bf1..b490935cd0646 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.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.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 @@ private void startClient() { 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 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 @@ private RestClient createClient() { }); final RestClient restClient = builder.build(); + restClientRef.set(restClient); if (configuration.isEnableSniffer()) { SnifferBuilder snifferBuilder = Sniffer.builder(restClient); snifferBuilder.setSniffIntervalMillis(configuration.getSnifferInterval()); @@ -567,6 +571,25 @@ private RestClient createClient() { 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> 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 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 0000000000000..2f2df73ed3e34 --- /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 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 0000000000000..2b99695c56ec5 --- /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 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 0000000000000..1f89865ca829f --- /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"); + } +}