#2103 OpenSearchConnection: send the credentials only to the configured addresses

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.<type>.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.<type>.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.

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<AuthScope> credentialScopes(List<HttpHost> hosts) {
+        final Set<AuthScope> 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<HttpHost> plainHttpHosts(List<HttpHost> hosts) {
+        final List<HttpHost> 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<HttpHost> hosts) {
+        final List<HttpHost> 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)

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<HttpHost> 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<AuthScope> 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<AuthScope> 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<HttpHost> 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"));
+    }
+}

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

external/opensearch/opensearch-conf.yaml

@@ -22,14 +22,17 @@ 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
 
   # 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