DataNode: using JWT for authentication/authorisation (#16334)

* jwt implementation, adding preflight + versionprobe

* adding JWT auth to anomaly detection client, some minor refactorings

* added missing license

* added provider, refactorings

* fixing small bugs

* set smaller default expiration

* using the correct OpenSearch distro for the erchitecture

* added config options

* small refactorings, fixing tests, adding dependencies for running tests from inside IntelliJ

* fixing import to correct Duration class

* removing unnecessary debug output

* fixing tests

* fixing tests

* adding JWT to Cluster test

* fixing path

* fixing path problems

* fix architecture logic for DatanodeContainerizedBackend

* fixing path problems

* fixing JWT settings in tests

* fixing JWT settings in tests

* fixing cluster test

* fixing cluster test

* fixing cluster test

* adding missing jwt token

* small refactoring to force thread-safety on creation

* the datanode uses mongodb - so the closing sequence needs to be adjusted

* moving config changes into config file, keep only change that has to be done by code

* adding breaking change info for JWT and clock synchronization.

* renaming config settings

* renaming config settings

* removed debug settings that are no longer in use

* renamed provider/annotation to be more precise

* lowering log level

* smaller refactorings/renamings

* keep config change in code until PR is merged, remove code in a follow-up PR

* moved changelog file to correct folder

* clock synchronsation

Co-authored-by: Dennis Oelkers <dennis@graylog.com>

* added jwt auth into changes after merge

* upped the timeout

* log curl connection information in DatanodeOpensearchWait

* replacing with jwt auth

* changing wait strategy

* fixing imports

* remove uname/pwd from opensearch rest base url

* propagate jwt token provider to http rest client callback, obtain fresh token every time

* inject IndexerJwtAuthTokenProvider in DatanodeConfigurationProvider

* remove rest uname/pwd from datanode configuration

---------

Co-authored-by: Tomas Dvorak <tomas.dvorak@graylog.com>
Co-authored-by: Dennis Oelkers <dennis@graylog.com>

data-node/src/test/java/org/graylog/datanode/DatanodeOpensearchWait.java

@@ -38,6 +38,8 @@ import org.slf4j.LoggerFactory;
 import javax.net.ssl.SSLHandshakeException;
 import java.net.SocketException;
 import java.security.KeyStore;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Locale;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
@@ -46,13 +48,12 @@ import java.util.function.Predicate;
 
 public class DatanodeOpensearchWait {
 
-    private static final Logger LOG = LoggerFactory.getLogger(DatanodeSecuritySetupIT.class);
+    private static final Logger LOG = LoggerFactory.getLogger(DatanodeOpensearchWait.class);
     private static final int ATTEMPTS_COUNT = 160;
 
     private final int opensearchPort;
     private KeyStore truststore;
-    private String username;
-    private String password;
+    private String jwtToken;
     private String lastRecordedResponse;
 
     private DatanodeOpensearchWait(int opensearchPort) {
@@ -68,9 +69,8 @@ public class DatanodeOpensearchWait {
         return this;
     }
 
-    public DatanodeOpensearchWait withBasicAuth(String username, String password) {
-        this.username = username;
-        this.password = password;
+    public DatanodeOpensearchWait withJwtAuth(String jwtToken) {
+        this.jwtToken = jwtToken;
         return this;
     }
 
@@ -90,6 +90,11 @@ public class DatanodeOpensearchWait {
 
     @SafeVarargs
     private ValidatableResponse waitForOpensearch(Predicate<ValidatableResponse>... predicates) throws ExecutionException, RetryException {
+
+        if(LOG.isInfoEnabled()) {
+            LOG.info("cURL request: " + formatCurlConnectionInfo());
+        }
+
         //noinspection UnstableApiUsage
         final RetryerBuilder<ValidatableResponse> builder = RetryerBuilder.<ValidatableResponse>newBuilder()
                 .withWaitStrategy(WaitStrategies.fixedWait(1, TimeUnit.SECONDS))
@@ -122,8 +127,8 @@ public class DatanodeOpensearchWait {
 
                 Optional.ofNullable(truststore).ifPresent(ts -> req.trustStore(truststore));
 
-                if (username != null && password != null) {
-                    req.auth().basic(username, password);
+                if (jwtToken != null) {
+                    req.header( "Authorization", "Bearer " + jwtToken);
                 }
 
                 return req.get(formatClusterHeathUrl())
@@ -138,6 +143,16 @@ public class DatanodeOpensearchWait {
         }
     }
 
+    private String formatCurlConnectionInfo() {
+        List<String> builder = new LinkedList<>();
+        builder.add("curl -k"); // trust self-signed certs
+        if(jwtToken != null) {
+            builder.add("-H \"Authorization: Bearer " + jwtToken + "\"");
+        }
+        builder.add(formatClusterHeathUrl());
+        return String.join(" ", builder);
+    }
+
     private void printShardsInfo() {
         try {
             final RequestSpecification req = RestAssured.given()
@@ -145,8 +160,8 @@ public class DatanodeOpensearchWait {
 
             Optional.ofNullable(truststore).ifPresent(ts -> req.trustStore(truststore));
 
-            if (username != null && password != null) {
-                req.auth().basic(username, password);
+            if (jwtToken != null) {
+                req.header( "Authorization", "Bearer " + jwtToken);
             }
 
             final String shardsResponse = req.get("https://localhost:" + opensearchPort + "/_cat/shards?v=true&h=index,shard,prirep,state,node,unassigned.reason&s=state")

data-node/src/test/java/org/graylog/datanode/integration/DatanodeClusterIT.java

@@ -16,11 +16,14 @@
  */
 package org.graylog.datanode.integration;
 
+import com.github.joschi.jadconfig.util.Duration;
 import com.github.rholder.retry.RetryException;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.graylog.datanode.DatanodeOpensearchWait;
 import org.graylog.datanode.configuration.variants.KeystoreInformation;
 import org.graylog.datanode.testinfra.DatanodeContainerizedBackend;
+import org.graylog2.security.IndexerJwtAuthTokenProvider;
+import org.hamcrest.Matchers;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -30,8 +33,11 @@ import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
 
+import javax.net.ssl.SSLHandshakeException;
 import javax.validation.constraints.NotNull;
 import java.io.IOException;
+import java.net.SocketException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.security.GeneralSecurityException;
 import java.security.KeyStore;
@@ -144,6 +150,7 @@ public class DatanodeClusterIT {
                 mongodb,
                 hostname,
                 datanodeContainer -> {
+                    datanodeContainer.withEnv("GRAYLOG_DATANODE_PASSWORD_SECRET", DatanodeContainerizedBackend.SIGNING_SECRET);
                     datanodeContainer.withEnv("GRAYLOG_DATANODE_CLUSTER_INITIAL_MANAGER_NODES", hostnameNodeA);
                     datanodeContainer.withEnv("GRAYLOG_DATANODE_OPENSEARCH_DISCOVERY_SEED_HOSTS", hostnameNodeA + ":9300");
 
@@ -174,11 +181,13 @@ public class DatanodeClusterIT {
                 });
     }
 
-    private void waitForNodesCount(int countOfNodes) throws ExecutionException, RetryException {
+    private void waitForNodesCount(final int countOfNodes) throws ExecutionException, RetryException {
+        final String jwtToken = IndexerJwtAuthTokenProvider.createToken(DatanodeContainerizedBackend.SIGNING_SECRET.getBytes(StandardCharsets.UTF_8), Duration.seconds(120));
+
         try {
             DatanodeOpensearchWait.onPort(nodeA.getOpensearchRestPort())
                     .withTruststore(trustStore)
-                    .withBasicAuth(initialAdminUsername, initialAdminPassword)
+                    .withJwtAuth(jwtToken)
                     .waitForNodesCount(countOfNodes);
         } catch (Exception retryException) {
             LOG.error("DataNode Container logs from nodeA follow:\n" + nodeA.getLogs());

data-node/src/test/java/org/graylog/datanode/integration/DatanodeSecuritySetupIT.java

@@ -16,6 +16,7 @@
  */
 package org.graylog.datanode.integration;
 
+import com.github.joschi.jadconfig.util.Duration;
 import com.github.rholder.retry.RetryException;
 import com.github.rholder.retry.Retryer;
 import com.github.rholder.retry.RetryerBuilder;
@@ -28,6 +29,7 @@ import org.apache.http.NoHttpResponseException;
 import org.graylog.datanode.configuration.variants.KeystoreInformation;
 import org.graylog.datanode.testinfra.DatanodeContainerizedBackend;
 import org.graylog2.plugin.Tools;
+import org.graylog2.security.IndexerJwtAuthTokenProvider;
 import org.graylog2.shared.utilities.StringUtils;
 import org.hamcrest.Matchers;
 import org.junit.jupiter.api.AfterEach;
@@ -40,6 +42,7 @@ import org.slf4j.LoggerFactory;
 import javax.net.ssl.SSLHandshakeException;
 import java.io.IOException;
 import java.net.SocketException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.security.GeneralSecurityException;
 import java.security.KeyStore;
@@ -86,6 +89,7 @@ public class DatanodeSecuritySetupIT {
             datanodeContainer.withEnv("GRAYLOG_DATANODE_TRANSPORT_CERTIFICATE", "datanode-transport-certificates.p12");
             datanodeContainer.withEnv("GRAYLOG_DATANODE_TRANSPORT_CERTIFICATE_PASSWORD", transportCert.passwordAsString());
             datanodeContainer.withEnv("GRAYLOG_DATANODE_INSECURE_STARTUP", "false");
+            datanodeContainer.withEnv("GRAYLOG_DATANODE_PASSWORD_SECRET", DatanodeContainerizedBackend.SIGNING_SECRET);
 
             // configure http security
             datanodeContainer.withEnv("GRAYLOG_DATANODE_HTTP_CERTIFICATE", "datanode-https-certificates.p12");
@@ -115,24 +119,24 @@ public class DatanodeSecuritySetupIT {
 
     @Test
     void testSecuredSetup() throws ExecutionException, RetryException {
+        final String jwtAuthToken = IndexerJwtAuthTokenProvider.createToken(DatanodeContainerizedBackend.SIGNING_SECRET.getBytes(StandardCharsets.UTF_8), Duration.seconds(120));
 
-        waitForOpensearchAvailableStatus(backend.getDatanodeRestPort(), trustStore);
+        waitForOpensearchAvailableStatus(backend.getDatanodeRestPort(), backend.getOpensearchRestPort(), trustStore);
 
         try {
-            given()
-                .auth().basic(httpUsername, httpPassword)
-                .trustStore(trustStore)
-                .get("https://localhost:" + backend.getOpensearchRestPort())
-                .then().assertThat()
-                .body("name", Matchers.equalTo(containerHostname))
-                .body("cluster_name", Matchers.equalTo("datanode-cluster"));
+            given().header( "Authorization", "Bearer " + jwtAuthToken)
+                    .trustStore(trustStore)
+                    .get("https://localhost:" + backend.getOpensearchRestPort())
+                    .then().assertThat()
+                    .body("name", Matchers.equalTo(containerHostname))
+                    .body("cluster_name", Matchers.equalTo("datanode-cluster"));
         } catch (Exception ex) {
             LOG.error("Error connecting to OpenSearch in the DataNode, showing logs:\n{}", backend.getLogs());
             throw ex;
         }
     }
 
-    private void waitForOpensearchAvailableStatus(Integer datanodeRestPort, KeyStore trustStore) throws ExecutionException, RetryException {
+    private void waitForOpensearchAvailableStatus(final Integer datanodeRestPort, final Integer opensearchPort, final KeyStore trustStore) throws ExecutionException, RetryException {
         final Retryer<ValidatableResponse> retryer = RetryerBuilder.<ValidatableResponse>newBuilder()
                 .withWaitStrategy(WaitStrategies.fixedWait(1, TimeUnit.SECONDS))
                 .withStopStrategy(StopStrategies.stopAfterAttempt(120))
@@ -145,11 +149,13 @@ public class DatanodeSecuritySetupIT {
 
         try {
             var hostname = Tools.getLocalCanonicalHostname();
-            var url = StringUtils.f("https://%s:%d", hostname, datanodeRestPort);
-            LOG.info("Trying to connect to: {}", url);
+            var rest = StringUtils.f("https://%s:%d", hostname, datanodeRestPort);
+            var os = StringUtils.f("https://%s:%d", hostname, opensearchPort);
+            LOG.info("Trying to connect to REST API at: {}, OpenSearch is at: {}", rest, os);
             retryer.call(() -> RestAssured.given()
+                    .auth().basic(httpUsername, httpPassword)
                     .trustStore(trustStore)
-                    .get(url)
+                    .get(rest)
                     .then());
         } catch (Exception ex) {
             LOG.error("Error starting the DataNode, showing logs:\n" + backend.getLogs());

data-node/src/test/java/org/graylog/datanode/testinfra/DatanodeContainerizedBackend.java

@@ -38,6 +38,7 @@ import java.util.Properties;
 import java.util.function.Supplier;
 
 public class DatanodeContainerizedBackend {
+    static public final String SIGNING_SECRET = "0123456789012345678901234567890123456789012345678901234567890987654321098765432109876543211";
 
     public static final int DATANODE_REST_PORT = 8999;
     public static final int DATANODE_OPENSEARCH_PORT = 9200;
@@ -106,7 +107,7 @@ public class DatanodeContainerizedBackend {
 
                 .withEnv("GRAYLOG_DATANODE_REST_API_USERNAME", "admin")
                 .withEnv("GRAYLOG_DATANODE_REST_API_PASSWORD", "admin")
-                .withEnv("GRAYLOG_DATANODE_PASSWORD_SECRET", "this_is_not_used_but_required")
+                .withEnv("GRAYLOG_DATANODE_PASSWORD_SECRET", SIGNING_SECRET)
 
                 .withEnv("GRAYLOG_DATANODE_NODE_ID_FILE", "./node-id")
                 .withEnv("GRAYLOG_DATANODE_HTTP_BIND_ADDRESS", "0.0.0.0:" + DATANODE_REST_PORT)