diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index 19df72888..8c12ddf59 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -441,9 +441,11 @@ See the link:https://github.com/apache/stormcrawler/tree/main/external/opensearc |=== | key | default value | description -| opensearch.addresses | - | OpenSearch server address(es). -| opensearch.user | - | Username for authentication (optional). +| opensearch.addresses | - | OpenSearch server address(es). An address without a scheme is contacted over plain http, on port 9200 if none is given; use `https://` when credentials are configured. +| opensearch.user | - | Username for Basic authentication (optional). The credentials are only sent to the addresses listed, matched on scheme, host and port, not to nodes discovered by sniffing under another address; list those nodes in the addresses or disable `opensearch..sniff`, which is on by default. A warning is logged when sniffing is enabled together with credentials, and if an address other than a loopback one uses plain http. | opensearch.password | - | Password for authentication (optional). +| opensearch..sniff | true | Discover the other nodes of the cluster and send requests to them as well. The nodes found are contacted over https if one of the addresses uses https, over plain http otherwise. +| opensearch.disable.tls.validation | false | Do not check the certificates and host names of the OpenSearch nodes. A warning is logged when this is enabled. | opensearch.concurrentRequests | 2 | Number of concurrent bulk requests. | opensearch.indexer.index.name | content | Index name for crawled documents. | opensearch.indexer.create | false | Auto-create index if it does not exist. diff --git a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java index 4c31a74a2..5a1ef1fa7 100644 --- a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java +++ b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java @@ -24,10 +24,14 @@ import java.net.URI; import java.util.ArrayList; import java.util.Arrays; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Locale; import java.util.Map; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Pattern; import javax.net.ssl.SSLContext; import org.apache.commons.lang3.StringUtils; import org.apache.hc.client5.http.auth.AuthScope; @@ -217,6 +221,55 @@ public static String getBulkOperationId(BulkOperation op) { return null; } + /** + * Returns the scopes the Basic credentials are registered for: the scheme, host and port of + * each configured address. A request to any other address does not receive them. + */ + static List credentialScopes(List hosts) { + final Set scopes = new LinkedHashSet<>(); + for (HttpHost host : hosts) { + scopes.add(new AuthScope(host)); + } + return new ArrayList<>(scopes); + } + + /** + * Returns the addresses which use plain http and are not a loopback address, i.e. those the + * Basic credentials would be sent to in the clear over the network. + */ + static List plainHttpHosts(List hosts) { + final List plain = new ArrayList<>(); + for (HttpHost host : hosts) { + if ("http".equalsIgnoreCase(host.getSchemeName()) && !isLoopback(host.getHostName())) { + plain.add(host); + } + } + return plain; + } + + private static final Pattern IPV4_LOOPBACK = Pattern.compile("127(\\.\\d{1,3}){3}"); + + /** Whether the host is localhost or a loopback IP literal. No name is resolved. */ + static boolean isLoopback(String hostname) { + String host = StringUtils.strip(hostname.toLowerCase(Locale.ROOT), "[]"); + return host.equals("localhost") + || IPV4_LOOPBACK.matcher(host).matches() + || host.equals("::1") + || host.equals("0:0:0:0:0:0:0:1"); + } + + private static void warnAboutPlainHttp(String boltType, List hosts) { + final List plain = plainHttpHosts(hosts); + if (!plain.isEmpty()) { + LOG.warn( + "OpenSearch credentials are configured for {} but the addresses {} use plain " + + "http, the credentials are sent unencrypted. Give the addresses an " + + "https:// scheme.", + boltType, + plain); + } + } + // internal helpers private record ClientResources(OpenSearchClient client, OpenSearchTransport transport) {} @@ -349,6 +402,16 @@ private static ClientResources buildClientResources( .setConnectTimeout(Timeout.ofMilliseconds(connectTimeout)) .setSocketTimeout(Timeout.ofMilliseconds(socketTimeout))); + if (needsUser) { + warnAboutPlainHttp(boltType, hosts); + } + if (disableTlsValidation) { + LOG.warn( + "opensearch.disable.tls.validation is set: the certificates and host names of " + + "the OpenSearch nodes used for {} are not checked", + boltType); + } + // Auth, proxy, and/or trust-all SSL via HttpClient customisation if (needsUser || needsProxy || disableTlsValidation) { builder.setHttpClientConfigCallback( @@ -357,9 +420,11 @@ private static ClientResources buildClientResources( if (needsUser) { final BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider(); - credentialsProvider.setCredentials( - new AuthScope(null, -1), - new UsernamePasswordCredentials(user, password.toCharArray())); + final UsernamePasswordCredentials credentials = + new UsernamePasswordCredentials(user, password.toCharArray()); + for (AuthScope scope : credentialScopes(hosts)) { + credentialsProvider.setCredentials(scope, credentials); + } httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider); } // hc.client5 proxy: HttpHost(scheme, host, port) diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/OpenSearchConnectionCredentialsTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/OpenSearchConnectionCredentialsTest.java new file mode 100644 index 000000000..bb484e843 --- /dev/null +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/OpenSearchConnectionCredentialsTest.java @@ -0,0 +1,128 @@ +/* + * 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.stormcrawler.opensearch; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import org.apache.hc.client5.http.auth.AuthScope; +import org.apache.hc.client5.http.auth.UsernamePasswordCredentials; +import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider; +import org.apache.hc.core5.http.HttpHost; +import org.junit.jupiter.api.Test; + +class OpenSearchConnectionCredentialsTest { + + private static BasicCredentialsProvider providerFor(List hosts) { + final BasicCredentialsProvider provider = new BasicCredentialsProvider(); + final UsernamePasswordCredentials credentials = + new UsernamePasswordCredentials("crawler", "s3cret".toCharArray()); + for (AuthScope scope : OpenSearchConnection.credentialScopes(hosts)) { + provider.setCredentials(scope, credentials); + } + return provider; + } + + @Test + void scopesCoverTheConfiguredHostsAndPorts() { + final List scopes = + OpenSearchConnection.credentialScopes( + List.of( + new HttpHost("https", "opensearch1.example.org", 9200), + new HttpHost("https", "opensearch2.example.org", 9201))); + assertEquals( + List.of( + new AuthScope("https", "opensearch1.example.org", 9200, null, null), + new AuthScope("https", "opensearch2.example.org", 9201, null, null)), + scopes); + } + + @Test + void duplicateAddressesGiveOneScope() { + final List scopes = + OpenSearchConnection.credentialScopes( + List.of( + new HttpHost("https", "opensearch1.example.org", 9200), + new HttpHost("https", "OpenSearch1.example.org", 9200))); + assertEquals(1, scopes.size()); + } + + @Test + void noAddressGivesNoScope() { + assertTrue(OpenSearchConnection.credentialScopes(List.of()).isEmpty()); + } + + @Test + void credentialsAreOnlyGivenToTheConfiguredHosts() { + final BasicCredentialsProvider provider = + providerFor(List.of(new HttpHost("https", "opensearch1.example.org", 9200))); + + assertNotNull( + provider.getCredentials(request("https", "opensearch1.example.org", 9200), null)); + // another node of the cluster which is not listed in the addresses + assertNull(provider.getCredentials(request("https", "10.0.0.12", 9200), null)); + // same host, another port + assertNull( + provider.getCredentials(request("https", "opensearch1.example.org", 9300), null)); + assertNull(provider.getCredentials(request("https", "other.example.org", 9200), null)); + } + + @Test + void credentialsAreMatchedOnTheScheme() { + final BasicCredentialsProvider provider = + providerFor(List.of(new HttpHost("https", "opensearch1.example.org", 9200))); + assertNull(provider.getCredentials(request("http", "opensearch1.example.org", 9200), null)); + } + + /** The scope HttpClient looks the credentials up with for a request to the given node. */ + private static AuthScope request(String scheme, String host, int port) { + return new AuthScope(new HttpHost(scheme, host, port), null, "Basic"); + } + + @Test + void plainHttpToARemoteHostIsReported() { + final HttpHost remote = new HttpHost("http", "opensearch1.example.org", 9200); + final List plain = + OpenSearchConnection.plainHttpHosts( + List.of( + remote, + new HttpHost("https", "opensearch2.example.org", 9200), + new HttpHost("http", "localhost", 9200), + new HttpHost("http", "127.0.0.1", 9200), + new HttpHost("http", "[::1]", 9200))); + assertEquals(List.of(remote), plain); + } + + @Test + void loopbackAddresses() { + assertTrue(OpenSearchConnection.isLoopback("localhost")); + assertTrue(OpenSearchConnection.isLoopback("LOCALHOST")); + assertTrue(OpenSearchConnection.isLoopback("127.0.0.1")); + assertTrue(OpenSearchConnection.isLoopback("[::1]")); + assertTrue(OpenSearchConnection.isLoopback("0:0:0:0:0:0:0:1")); + assertFalse(OpenSearchConnection.isLoopback("opensearch1.example.org")); + assertFalse(OpenSearchConnection.isLoopback("10.0.0.12")); + assertFalse(OpenSearchConnection.isLoopback("localhost.example.org")); + assertFalse(OpenSearchConnection.isLoopback("127.example.org")); + assertTrue(OpenSearchConnection.isLoopback("127.1.2.3")); + } +} diff --git a/external/opensearch/archetype/src/main/resources/archetype-resources/opensearch-conf.yaml b/external/opensearch/archetype/src/main/resources/archetype-resources/opensearch-conf.yaml index 597ff4f85..d1b8d4500 100644 --- a/external/opensearch/archetype/src/main/resources/archetype-resources/opensearch-conf.yaml +++ b/external/opensearch/archetype/src/main/resources/archetype-resources/opensearch-conf.yaml @@ -22,11 +22,14 @@ config: # also accepts a list or multiple values in a single line # separated by a semi-colon e.g. "opensearch1:9200; opensearch2:9200" opensearch.addresses: "http://localhost:9200" + # credentials are only sent to the hosts and ports listed in the addresses, + # use https:// addresses for any host other than localhost #opensearch.user: "USERNAME" #opensearch.password: "PASSWORD" opensearch.concurrentRequests: 2 - # Disable TLS validation for connection to OpenSearch + # Disable TLS validation for connection to OpenSearch, a warning is logged + # when this is enabled # opensearch.disable.tls.validation: false # Indexer bolt diff --git a/external/opensearch/opensearch-conf.yaml b/external/opensearch/opensearch-conf.yaml index dbbbc8d8f..595e170c9 100644 --- a/external/opensearch/opensearch-conf.yaml +++ b/external/opensearch/opensearch-conf.yaml @@ -22,6 +22,8 @@ config: # also accepts a list or multiple values in a single line # separated by a semi-colon e.g. "opensearch1:9200; opensearch2:9200" opensearch.addresses: "http://localhost:9200" + # credentials are only sent to the hosts and ports listed in the addresses, + # use https:// addresses for any host other than localhost #opensearch.user: "USERNAME" #opensearch.password: "PASSWORD" opensearch.concurrentRequests: 2 @@ -29,7 +31,8 @@ config: # Sets the response buffer to the specified value in MB. # opensearch.responseBufferSize: 100 - # Disable TLS validation for connection to OpenSearch + # Disable TLS validation for connection to OpenSearch, a warning is logged + # when this is enabled # opensearch.disable.tls.validation: false # Indexer bolt diff --git a/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java b/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java index 58d38df24..f9a5dea1a 100644 --- a/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java +++ b/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java @@ -24,17 +24,21 @@ import java.net.URI; import java.util.ArrayList; import java.util.Arrays; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Locale; import java.util.Map; +import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; import org.apache.commons.lang3.StringUtils; 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.apache.http.client.CredentialsProvider; import org.apache.http.conn.ssl.NoopHostnameVerifier; import org.apache.http.conn.ssl.TrustAllStrategy; -import org.apache.http.impl.client.BasicCredentialsProvider; import org.apache.http.ssl.SSLContextBuilder; import org.apache.stormcrawler.util.ConfUtils; import org.jetbrains.annotations.NotNull; @@ -49,6 +53,7 @@ import org.opensearch.client.RestClient; import org.opensearch.client.RestClientBuilder; import org.opensearch.client.RestHighLevelClient; +import org.opensearch.client.sniff.OpenSearchNodesSniffer; import org.opensearch.client.sniff.Sniffer; import org.opensearch.common.unit.TimeValue; import org.slf4j.Logger; @@ -139,18 +144,27 @@ public static RestHighLevelClient getClient(Map stormConf, Strin ConfUtils.getBoolean( stormConf, Constants.PARAMPREFIX, "", "disable.tls.validation", false); - final boolean needsUser = StringUtils.isNotBlank(user) && StringUtils.isNotBlank(password); + final boolean needsUser = hasCredentials(user, password); final boolean needsProxy = StringUtils.isNotBlank(proxyhost) && proxyport != -1; + if (needsUser) { + warnAboutPlainHttp(boltType, hosts); + } + if (disableTlsValidation) { + LOG.warn( + "opensearch.disable.tls.validation is set: the certificates and host names of " + + "the OpenSearch nodes used for {} are not checked", + boltType); + } + if (needsUser || needsProxy || disableTlsValidation) { builder.setHttpClientConfigCallback( httpClientBuilder -> { if (needsUser) { - final CredentialsProvider credentialsProvider = - new BasicCredentialsProvider(); - credentialsProvider.setCredentials( - AuthScope.ANY, new UsernamePasswordCredentials(user, password)); - httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider); + httpClientBuilder.setDefaultCredentialsProvider( + new OriginCredentialsProvider( + hosts, + new UsernamePasswordCredentials(user, password))); } if (needsProxy) { httpClientBuilder.setProxy( @@ -230,6 +244,106 @@ public static RestHighLevelClient getClient(Map stormConf, Strin return new RestHighLevelClient(builder); } + /** Basic authentication is only set up when both the user and the password are given. */ + static boolean hasCredentials(String user, String password) { + return StringUtils.isNotBlank(user) && StringUtils.isNotBlank(password); + } + + /** + * Gives the Basic credentials only to requests for one of the configured addresses, matched on + * scheme, host and port. A node reached under any other address, such as one found by the + * sniffer under the address it publishes, does not receive them. + */ + static final class OriginCredentialsProvider implements CredentialsProvider { + + private final Set origins = new LinkedHashSet<>(); + private final Credentials credentials; + + OriginCredentialsProvider(List hosts, Credentials credentials) { + for (HttpHost host : hosts) { + origins.add(origin(host)); + } + this.credentials = credentials; + } + + private static String origin(HttpHost host) { + final String scheme = host.getSchemeName().toLowerCase(Locale.ROOT); + int port = host.getPort(); + if (port < 0) { + port = "https".equals(scheme) ? 443 : 80; + } + return scheme + "://" + host.getHostName().toLowerCase(Locale.ROOT) + ":" + port; + } + + @Override + public Credentials getCredentials(AuthScope scope) { + final HttpHost origin = scope.getOrigin(); + if (origin != null && origins.contains(origin(origin))) { + return credentials; + } + return null; + } + + @Override + public void setCredentials(AuthScope scope, Credentials credentials) { + throw new UnsupportedOperationException(); + } + + @Override + public void clear() {} + } + + /** + * Returns the addresses which use plain http and are not a loopback address, i.e. those the + * Basic credentials would be sent to in the clear over the network. + */ + static List plainHttpHosts(List hosts) { + final List plain = new ArrayList<>(); + for (HttpHost host : hosts) { + if ("http".equalsIgnoreCase(host.getSchemeName()) && !isLoopback(host.getHostName())) { + plain.add(host); + } + } + return plain; + } + + /** + * Returns the scheme under which the sniffer registers the nodes it finds. The nodes publish no + * scheme, so https is used as soon as one configured address uses it; otherwise sniffing would + * move the requests to plain http after the first round. + */ + static OpenSearchNodesSniffer.Scheme sniffScheme(List hosts) { + for (HttpHost host : hosts) { + if ("https".equalsIgnoreCase(host.getSchemeName())) { + return OpenSearchNodesSniffer.Scheme.HTTPS; + } + } + return OpenSearchNodesSniffer.Scheme.HTTP; + } + + private static final Pattern IPV4_LOOPBACK = Pattern.compile("127(\\.\\d{1,3}){3}"); + + /** Whether the host is localhost or a loopback IP literal. No name is resolved. */ + static boolean isLoopback(String hostname) { + String host = StringUtils.strip(hostname.toLowerCase(Locale.ROOT), "[]"); + return host.equals("localhost") + || IPV4_LOOPBACK.matcher(host).matches() + || host.equals("::1") + || host.equals("0:0:0:0:0:0:0:1"); + } + + private static void warnAboutPlainHttp(String boltType, List hosts) { + final List plain = plainHttpHosts(hosts); + if (!plain.isEmpty()) { + LOG.warn( + "OpenSearch credentials are configured for {} but the addresses {} use plain " + + "http, the credentials are sent unencrypted. Give the addresses an " + + "https:// scheme.", + boltType, + plain); + } + } + public void addToProcessor(final DocWriteRequest request) { processor.add(request); } @@ -306,7 +420,35 @@ public static OpenSearchConnection getConnection( ConfUtils.getBoolean( stormConf, Constants.PARAMPREFIX, dottedType, "sniff", true); if (sniff) { - sniffer = Sniffer.builder(client.getLowLevelClient()).build(); + if (hasCredentials( + ConfUtils.getString(stormConf, Constants.PARAMPREFIX, dottedType, "user"), + ConfUtils.getString( + stormConf, Constants.PARAMPREFIX, dottedType, "password"))) { + LOG.warn( + "Sniffing is enabled for {} and OpenSearch credentials are configured. " + + "The credentials are only sent to the configured addresses: " + + "requests to a node the sniffer finds under another host or " + + "port are sent without them and fail if the cluster requires " + + "authentication. List every node in opensearch.{}.addresses " + + "or set opensearch.{}.sniff: false.", + boltType, + boltType, + boltType); + } + final RestClient lowLevelClient = client.getLowLevelClient(); + final List configured = new ArrayList<>(); + for (Node node : lowLevelClient.getNodes()) { + configured.add(node.getHost()); + } + sniffer = + Sniffer.builder(lowLevelClient) + .setNodesSniffer( + new OpenSearchNodesSniffer( + lowLevelClient, + OpenSearchNodesSniffer + .DEFAULT_SNIFF_REQUEST_TIMEOUT, + sniffScheme(configured))) + .build(); } return new OpenSearchConnection(client, bulkProcessor, sniffer); diff --git a/external/opensearch/src/test/java/org/apache/stormcrawler/opensearch/OpenSearchConnectionCredentialsTest.java b/external/opensearch/src/test/java/org/apache/stormcrawler/opensearch/OpenSearchConnectionCredentialsTest.java new file mode 100644 index 000000000..ae90bebde --- /dev/null +++ b/external/opensearch/src/test/java/org/apache/stormcrawler/opensearch/OpenSearchConnectionCredentialsTest.java @@ -0,0 +1,219 @@ +/* + * 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.stormcrawler.opensearch; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.UncheckedIOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +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.apache.http.client.CredentialsProvider; +import org.junit.jupiter.api.Test; +import org.opensearch.client.Request; +import org.opensearch.client.RestClient; +import org.opensearch.client.sniff.OpenSearchNodesSniffer; + +class OpenSearchConnectionCredentialsTest { + + private static final Credentials CREDENTIALS = + new UsernamePasswordCredentials("crawler", "s3cret"); + + private static CredentialsProvider providerFor(HttpHost... hosts) { + return new OpenSearchConnection.OriginCredentialsProvider(List.of(hosts), CREDENTIALS); + } + + /** The scope HttpClient looks the credentials up with for a request to the given node. */ + private static AuthScope request(String scheme, String host, int port) { + return new AuthScope(new HttpHost(host, port, scheme), AuthScope.ANY_REALM, "Basic"); + } + + @Test + void credentialsAreGivenToTheConfiguredAddresses() { + final CredentialsProvider provider = + providerFor( + new HttpHost("opensearch1.example.org", 9200, "https"), + new HttpHost("OpenSearch2.example.org", 9201, "https")); + assertEquals( + CREDENTIALS, + provider.getCredentials(request("https", "opensearch1.example.org", 9200))); + assertEquals( + CREDENTIALS, + provider.getCredentials(request("HTTPS", "opensearch2.example.org", 9201))); + } + + @Test + void credentialsAreOnlyGivenToTheConfiguredHosts() { + final CredentialsProvider provider = + providerFor(new HttpHost("opensearch1.example.org", 9200, "https")); + + // a node found by the sniffer under the address it publishes + assertNull(provider.getCredentials(request("https", "10.0.0.12", 9200))); + // same host, another port + assertNull(provider.getCredentials(request("https", "opensearch1.example.org", 9300))); + assertNull(provider.getCredentials(request("https", "other.example.org", 9200))); + } + + @Test + void credentialsAreMatchedOnTheScheme() { + final CredentialsProvider provider = + providerFor(new HttpHost("opensearch1.example.org", 9200, "https")); + assertNull(provider.getCredentials(request("http", "opensearch1.example.org", 9200))); + } + + @Test + void credentialsNeedAnOrigin() { + final CredentialsProvider provider = + providerFor(new HttpHost("opensearch1.example.org", 9200, "https")); + assertNull(provider.getCredentials(new AuthScope("opensearch1.example.org", 9200))); + assertNull(provider.getCredentials(AuthScope.ANY)); + } + + @Test + void noAddressGivesTheCredentialsToNoOne() { + assertNull(providerFor().getCredentials(request("https", "opensearch1.example.org", 9200))); + } + + /** + * Sends a request with the client to a local server and returns the Authorization header it + * received, if any. + */ + private static String authorizationSentTo(HttpHost configured) throws Exception { + try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + final int port = server.getLocalPort(); + final CompletableFuture authorization = + CompletableFuture.supplyAsync( + () -> { + try (Socket socket = server.accept()) { + final BufferedReader in = + new BufferedReader( + new InputStreamReader( + socket.getInputStream(), + StandardCharsets.US_ASCII)); + String header = null; + for (String line = in.readLine(); + line != null && !line.isEmpty(); + line = in.readLine()) { + if (line.regionMatches(true, 0, "Authorization:", 0, 14)) { + header = line.substring(14).trim(); + } + } + socket.getOutputStream() + .write( + "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + .getBytes(StandardCharsets.US_ASCII)); + return header; + } catch (IOException e) { + throw new UncheckedIOException(e); + } + }); + final HttpHost configuredHere = + new HttpHost(configured.getHostName(), port, configured.getSchemeName()); + try (RestClient client = + RestClient.builder(new HttpHost("127.0.0.1", port, "http")) + .setHttpClientConfigCallback( + b -> + b.setDefaultCredentialsProvider( + providerFor(configuredHere))) + .build()) { + client.performRequest(new Request("GET", "/")); + } + return authorization.get(10, TimeUnit.SECONDS); + } + } + + @Test + void clientSendsTheCredentialsToAConfiguredAddress() throws Exception { + assertNotNull(authorizationSentTo(new HttpHost("127.0.0.1", 0, "http"))); + } + + @Test + void clientSendsNoCredentialsToAnotherAddress() throws Exception { + assertNull(authorizationSentTo(new HttpHost("127.0.0.1", 0, "https"))); + } + + @Test + void plainHttpToARemoteHostIsReported() { + final HttpHost remote = new HttpHost("opensearch1.example.org", 9200, "http"); + final List plain = + OpenSearchConnection.plainHttpHosts( + List.of( + remote, + new HttpHost("opensearch2.example.org", 9200, "https"), + new HttpHost("localhost", 9200, "http"), + new HttpHost("127.0.0.1", 9200, "http"), + new HttpHost("[::1]", 9200, "http"))); + assertEquals(List.of(remote), plain); + } + + @Test + void sniffedNodesKeepTheSchemeOfTheAddresses() { + assertEquals( + OpenSearchNodesSniffer.Scheme.HTTPS, + OpenSearchConnection.sniffScheme( + List.of(new HttpHost("opensearch1.example.org", 9200, "https")))); + assertEquals( + OpenSearchNodesSniffer.Scheme.HTTPS, + OpenSearchConnection.sniffScheme( + List.of( + new HttpHost("opensearch1.example.org", 9200, "http"), + new HttpHost("opensearch2.example.org", 9200, "https")))); + assertEquals( + OpenSearchNodesSniffer.Scheme.HTTP, + OpenSearchConnection.sniffScheme( + List.of(new HttpHost("opensearch1.example.org", 9200, "http")))); + } + + @Test + void credentialsNeedUserAndPassword() { + assertTrue(OpenSearchConnection.hasCredentials("crawler", "s3cret")); + assertFalse(OpenSearchConnection.hasCredentials("crawler", null)); + assertFalse(OpenSearchConnection.hasCredentials("crawler", " ")); + assertFalse(OpenSearchConnection.hasCredentials(null, "s3cret")); + assertFalse(OpenSearchConnection.hasCredentials("", "")); + } + + @Test + void loopbackAddresses() { + assertTrue(OpenSearchConnection.isLoopback("localhost")); + assertTrue(OpenSearchConnection.isLoopback("LOCALHOST")); + assertTrue(OpenSearchConnection.isLoopback("127.0.0.1")); + assertTrue(OpenSearchConnection.isLoopback("[::1]")); + assertTrue(OpenSearchConnection.isLoopback("0:0:0:0:0:0:0:1")); + assertFalse(OpenSearchConnection.isLoopback("opensearch1.example.org")); + assertFalse(OpenSearchConnection.isLoopback("10.0.0.12")); + assertFalse(OpenSearchConnection.isLoopback("localhost.example.org")); + assertFalse(OpenSearchConnection.isLoopback("127.example.org")); + assertTrue(OpenSearchConnection.isLoopback("127.1.2.3")); + } +}