ApiKey authentication

src/main/java/io/confluent/connect/elasticsearch/ConfigCallbackHandler.java

@@ -19,9 +19,11 @@
 import java.security.AccessController;
 import java.security.PrivilegedActionException;
 import java.security.PrivilegedExceptionAction;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import javax.net.ssl.HostnameVerifier;
 import javax.net.ssl.SSLContext;
@@ -30,6 +32,7 @@
 import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.auth.login.Configuration;
 import javax.security.auth.login.LoginContext;
+import org.apache.http.Header;
 import org.apache.http.HttpHost;
 import org.apache.http.auth.AuthSchemeProvider;
 import org.apache.http.auth.AuthScope;
@@ -49,6 +52,7 @@
 import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
 import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
 import org.apache.http.impl.nio.reactor.IOReactorConfig;
+import org.apache.http.message.BasicHeader;
 import org.apache.http.nio.conn.NoopIOSessionStrategy;
 import org.apache.http.nio.conn.SchemeIOSessionStrategy;
 import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
@@ -132,6 +136,13 @@ private void configureAuthentication(HttpAsyncClientBuilder builder) {
           )
       );
       builder.setDefaultCredentialsProvider(credentialsProvider);
+    } else if (config.isApikeyConfigured()) {
+      String apiKey = config.apikey();
+      List<Header> defaultHeaders = Arrays.asList(
+          new BasicHeader("Authorization",
+              "ApiKey " + apiKey));
+
+      builder.setDefaultHeaders(defaultHeaders);
     }
 
     if (config.isBasicProxyConfigured()) {

src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfig.java

@@ -71,6 +71,12 @@ public class ElasticsearchSinkConnectorConfig extends AbstractConfig {
   private static final String CONNECTION_PASSWORD_DISPLAY = "Connection Password";
   private static final String CONNECTION_PASSWORD_DEFAULT = null;
 
+  public static final String CONNECTION_APIKEY_CONFIG = "connection.apikey";
+  private static final String CONNECTION_APIKEY_DOC =
+      "The API key id used to authenticate with Elasticsearch.";
+  private static final String CONNECTION_APIKEY_DISPLAY = "Connection API key ID";
+  private static final String CONNECTION_APIKEY_DEFAULT = null;
+
   public static final String BATCH_SIZE_CONFIG = "batch.size";
   private static final String BATCH_SIZE_DOC =
       "The number of records to process as a batch when writing to Elasticsearch.";
@@ -480,6 +486,16 @@ private static void addConnectorConfigs(ConfigDef configDef) {
             ++order,
             Width.SHORT,
             CONNECTION_PASSWORD_DISPLAY
+        ).define(
+            CONNECTION_APIKEY_CONFIG,
+            Type.STRING,
+            CONNECTION_APIKEY_DEFAULT,
+            Importance.MEDIUM,
+            CONNECTION_APIKEY_DOC,
+            CONNECTOR_GROUP,
+            ++order,
+            Width.LONG,
+            CONNECTION_APIKEY_DISPLAY
         ).define(
             BATCH_SIZE_CONFIG,
             Type.INT,
@@ -899,6 +915,10 @@ public boolean isAuthenticatedConnection() {
     return username() != null && password() != null;
   }
 
+  public boolean isApikeyConfigured() {
+    return apikey() != null;
+  }
+
   public boolean isBasicProxyConfigured() {
     return !getString(PROXY_HOST_CONFIG).isEmpty();
   }
@@ -1050,6 +1070,10 @@ public Password password() {
     return getPassword(CONNECTION_PASSWORD_CONFIG);
   }
 
+  public String apikey() {
+    return getString(CONNECTION_APIKEY_CONFIG);
+  }
+
   public String proxyHost() {
     return getString(PROXY_HOST_CONFIG);
   }

src/main/java/io/confluent/connect/elasticsearch/Validator.java

@@ -39,6 +39,7 @@
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.BATCH_SIZE_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.BEHAVIOR_ON_NULL_VALUES_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.BehaviorOnNullValues;
+import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_APIKEY_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_PASSWORD_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_USERNAME_CONFIG;
@@ -101,6 +102,7 @@ public Config validate() {
     }
 
     validateCredentials();
+    validateApiKey();
     validateDataStreamConfigs();
     validateIgnoreConfigs();
     validateKerberos();
@@ -137,6 +139,22 @@ private void validateCredentials() {
     }
   }
 
+  private void validateApiKey() {
+    if (config.isApikeyConfigured()) {
+      if (config.isAuthenticatedConnection()) {
+        String errorMessage = String.format(
+            "Either only API Key (%s) or connection credentials (%s, %s) must be set.",
+            CONNECTION_APIKEY_CONFIG,
+            CONNECTION_USERNAME_CONFIG,
+            CONNECTION_PASSWORD_CONFIG
+        );
+        addErrorMessage(CONNECTION_APIKEY_CONFIG, errorMessage);
+        addErrorMessage(CONNECTION_USERNAME_CONFIG, errorMessage);
+        addErrorMessage(CONNECTION_PASSWORD_CONFIG, errorMessage);
+      }
+    }
+  }
+
   private void validateDataStreamConfigs() {
     if (config.dataStreamType().toUpperCase().equals(DataStreamType.NONE.name())
             ^ config.dataStreamDataset().isEmpty()) {

src/test/java/io/confluent/connect/elasticsearch/helper/ElasticsearchContainer.java

@@ -15,23 +15,18 @@
 
 package io.confluent.connect.elasticsearch.helper;
 
-import io.confluent.connect.elasticsearch.ElasticsearchClient;
-import io.confluent.connect.elasticsearch.RetryUtil;
+import co.elastic.clients.elasticsearch.security.GrantApiKeyResponse;
 import org.apache.kafka.common.config.SslConfigs;
-import org.apache.kafka.test.TestUtils;
 import org.elasticsearch.client.security.user.User;
 import org.elasticsearch.client.security.user.privileges.Role;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.ContainerLaunchException;
 import org.testcontainers.containers.output.OutputFrame;
 import org.testcontainers.containers.wait.strategy.Wait;
-import org.testcontainers.containers.wait.strategy.WaitStrategyTarget;
-import org.testcontainers.images.RemoteDockerImage;
 import org.testcontainers.images.builder.ImageFromDockerfile;
 import org.testcontainers.images.builder.dockerfile.DockerfileBuilder;
 import org.testcontainers.shaded.org.apache.commons.io.IOUtils;
-import org.testcontainers.utility.DockerImageName;
 
 import java.io.File;
 import java.io.FileOutputStream;
@@ -40,8 +35,10 @@
 import java.net.InetAddress;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -147,6 +144,7 @@ public static ElasticsearchContainer withESVersion(String ESVersion) {
   private Map<User, String> usersToCreate;
   private String localKeystorePath;
   private String localTruststorePath;
+  private Map<String, String> apiKeysToCreate;
 
   /**
    * Create an Elasticsearch container with the given image name with version qualifier.
@@ -179,6 +177,9 @@ public void start() {
       ElasticsearchHelperClient helperClient = getHelperClient(props);
       helperClient.waitForConnection(30000);
       createUsersAndRoles(helperClient);
+      if (isApikeyEnabled()) {
+        createApikeys(helperClient);
+      }
     }
   }
 
@@ -204,6 +205,21 @@ private void createUsersAndRoles(ElasticsearchHelperClient helperClient ) {
     }
   }
 
+  private void createApikeys(ElasticsearchHelperClient helperClient ) {
+    try {
+      for (Map.Entry<User,String> userToPassword: this.usersToCreate.entrySet()) {
+        GrantApiKeyResponse grantApiKeyResponse = helperClient.grantApiKey(userToPassword, this.rolesToCreate);
+        String apiKey =
+            Base64.getEncoder().encodeToString(
+                (grantApiKeyResponse.id() + ":" + grantApiKeyResponse.apiKey())
+                    .getBytes(StandardCharsets.UTF_8));
+        apiKeysToCreate.put(userToPassword.getKey().getUsername(), apiKey);
+      }
+    } catch (IOException e) {
+      throw new ContainerLaunchException("Container startup failed", e);
+    }
+  }
+
   public ElasticsearchContainer withSslEnabled(boolean enable) {
     enableSsl(enable);
     return this;
@@ -219,6 +235,12 @@ public ElasticsearchContainer withBasicAuth(Map<User, String> users, List<Role>
     return this;
   }
 
+  public ElasticsearchContainer withApikey(Map<User, String> users, List<Role> roles) {
+    enableBasicAuth(users, roles);
+    enableApikeys(users);
+    return this;
+  }
+
   /**
    * Set whether the Elasticsearch instance should use SSL.
    *
@@ -289,10 +311,24 @@ private void enableBasicAuth(Map<User, String> users, List<Role> roles) {
     this.rolesToCreate = roles;
   }
 
+  private void enableApikeys(Map<User, String> users) {
+    if (isKerberosEnabled()) {
+      throw new IllegalStateException(
+          "Api Keys and Kerberos are mutually exclusive."
+      );
+    }
+    this.apiKeysToCreate = new HashMap<>();
+    users.keySet().stream().forEach(user -> this.apiKeysToCreate.put(user.getUsername(), null));
+  }
+
   public boolean isBasicAuthEnabled() {
     return usersToCreate != null && !this.usersToCreate.isEmpty();
   }
 
+  public boolean isApikeyEnabled() {
+    return apiKeysToCreate != null && !this.apiKeysToCreate.isEmpty();
+  }
+
   private String getFullResourcePath(String resourceName) {
     if (isSslEnabled() && isKerberosEnabled()) {
       return "/both/" + resourceName;
@@ -601,4 +637,8 @@ public boolean shouldStartClientInCompatibilityMode() {
   public int esMajorVersion() {
     return getImageVersion().get(0);
   }
+
+  public Map<String, String> getAPIkeys() {
+    return apiKeysToCreate;
+  }
 }

src/test/java/io/confluent/connect/elasticsearch/helper/ElasticsearchHelperClient.java

@@ -17,8 +17,13 @@
 
 import co.elastic.clients.elasticsearch.ElasticsearchClient;
 import co.elastic.clients.elasticsearch.indices.GetDataStreamRequest.Builder;
+import co.elastic.clients.elasticsearch.security.GrantApiKeyRequest;
+import co.elastic.clients.elasticsearch.security.GrantApiKeyResponse;
+import co.elastic.clients.elasticsearch.security.grant_api_key.ApiKeyGrantType;
+import co.elastic.clients.elasticsearch.security.grant_api_key.GrantApiKey;
 import co.elastic.clients.json.jackson.JacksonJsonpMapper;
 import co.elastic.clients.transport.rest_client.RestClientTransport;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.http.HttpHost;
 import org.apache.kafka.test.TestUtils;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
@@ -62,6 +67,7 @@ public class ElasticsearchHelperClient {
   private final String url;
   private final ElasticsearchSinkConnectorConfig config;
   private RestHighLevelClient client;
+  private ElasticsearchClient elasticsearchClient;
 
   public ElasticsearchHelperClient(String url, ElasticsearchSinkConnectorConfig config,
       boolean compatibilityMode) {
@@ -75,6 +81,11 @@ public ElasticsearchHelperClient(String url, ElasticsearchSinkConnectorConfig co
         .build()
         // compatibility mode should be true for 7.17 high level rest clients while talking to ES 8.
     ).setApiCompatibilityMode(compatibilityMode).build();
+
+    elasticsearchClient = new ElasticsearchClient(new RestClientTransport(
+        RestClient.builder(HttpHost.create(url)).setHttpClientConfigCallback(configCallbackHandler).build(),
+        new JacksonJsonpMapper()
+    ));
   }
 
   public ElasticsearchHelperClient(String url, ElasticsearchSinkConnectorConfig config) {
@@ -167,6 +178,23 @@ public void createUser(Entry<User, String> userToPassword) throws IOException {
       throw new RuntimeException(String.format("Failed to create a user %s", userToPassword.getKey().getUsername()));
     }
   }
+  public GrantApiKeyResponse grantApiKey(Entry<User, String> userToPassword, List<Role> roles) throws IOException {
+    GrantApiKey grantApiKey =
+        new GrantApiKey.Builder().name("apikey_".concat(userToPassword.getKey().getUsername())).build();
+    GrantApiKeyRequest grantApiKeyRequest = new GrantApiKeyRequest.Builder()
+        .grantType(ApiKeyGrantType.Password)
+        .apiKey(grantApiKey)
+        .username(userToPassword.getKey().getUsername())
+        .password(userToPassword.getValue())
+        .build();
+
+    GrantApiKeyResponse grantApiKeyResponse = elasticsearchClient.security().grantApiKey(grantApiKeyRequest);
+    if (StringUtils.isEmpty(grantApiKeyResponse.apiKey())) {
+      throw new RuntimeException(String.format("Failed to create API key for user %s", userToPassword.getKey().getUsername()));
+    }
+        return grantApiKeyResponse;
+  }
+
 
   public void waitForConnection(long timeMs) {
     try {

src/test/java/io/confluent/connect/elasticsearch/integration/ElasticsearchConnectorApikeyIT.java

@@ -0,0 +1,43 @@
+package io.confluent.connect.elasticsearch.integration;
+
+import io.confluent.common.utils.IntegrationTest;
+import io.confluent.connect.elasticsearch.helper.ElasticsearchContainer;
+import org.elasticsearch.client.security.user.User;
+import org.elasticsearch.client.security.user.privileges.Role;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import java.util.List;
+import java.util.Map;
+import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_APIKEY_CONFIG;
+
+@Category(IntegrationTest.class)
+public class ElasticsearchConnectorApikeyIT extends ElasticsearchConnectorBaseIT {
+
+  @BeforeClass
+  public static void setupBeforeAll() throws Exception {
+    Map<User, String> users = getUsers();
+    List<Role> roles = getRoles();
+    container = ElasticsearchContainer.fromSystemProperties().withApikey(users, roles);
+    container.start();
+  }
+
+  @AfterClass
+  public static void cleanupAfterAll() {
+    container.close();
+  }
+
+  @Test
+  public void testApikey() throws Exception {
+    addApikeyConfigConfigs(props);
+    helperClient = container.getHelperClient(props);
+    helperClient.waitForConnection(60000);
+    runSimpleTest(props);
+  }
+
+  protected static void addApikeyConfigConfigs(Map<String, String> props) {
+    String apikey = container.getAPIkeys().get(ELASTIC_MINIMAL_PRIVILEGES_NAME);
+    props.put(CONNECTION_APIKEY_CONFIG, apikey);
+  }
+}