ENH: encryption extension (#5581)

ADD: encryption extension

Signed-off-by: George Chen <qchea@amazon.com>

Author: chenqi0805

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/encryption/EncryptionEngine.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.encryption;
+
+public interface EncryptionEngine {
+    /**
+     * Encrypts raw data into {@link EncryptionEnvelope}.
+     *
+     * @param data the raw data in bytes
+     * @return returns the encryption envelope
+     */
+    EncryptionEnvelope encrypt(byte[] data);
+
+    /**
+     * Decrypts the encryption envelope into raw data.
+     *
+     * @param encryptionEnvelope the encryption envelope
+     * @return returns the raw data in bytes
+     */
+    byte[] decrypt(EncryptionEnvelope encryptionEnvelope);
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/encryption/EncryptionEnvelope.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.encryption;
+
+public interface EncryptionEnvelope {
+    /**
+     * The encrypted data.
+     */
+    byte[] getEncryptedData();
+
+    /**
+     * The encrypted data key.
+     */
+    String getEncryptedDataKey();
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/encryption/KeyProvider.java

@@ -0,0 +1,11 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.encryption;
+
+@FunctionalInterface
+public interface KeyProvider {
+    byte[] decryptKey(byte[] encryptedKey);
+}

data-prepper-core/build.gradle

@@ -23,6 +23,7 @@ dependencies {
     implementation project(':data-prepper-event')
     implementation project(':data-prepper-plugins:blocking-buffer')
     implementation project(':data-prepper-plugins:common')
+    implementation project(':data-prepper-plugins:encryption-plugin')
     implementation project(':data-prepper-logstash-configuration')
     implementation project(':data-prepper-pipeline-parser')
     implementation project(':data-prepper-plugin-framework')

data-prepper-core/src/main/java/org/opensearch/dataprepper/core/pipeline/server/DataPrepperServer.java

@@ -10,6 +10,7 @@
 import com.sun.net.httpserver.HttpHandler;
 import com.sun.net.httpserver.HttpServer;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
+import org.opensearch.dataprepper.plugins.encryption.EncryptionHttpHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -32,6 +33,7 @@ public class DataPrepperServer {
     private final ListPipelinesHandler listPipelinesHandler;
     private final GetPipelinesHandler getPipelinesHandler;
     private final ShutdownHandler shutdownHandler;
+    private final EncryptionHttpHandler encryptionHttpHandler;
     private final PrometheusMeterRegistry prometheusMeterRegistry;
     private final Authenticator authenticator;
     private final ExecutorService executorService;
@@ -43,13 +45,15 @@ public DataPrepperServer(
             final ListPipelinesHandler listPipelinesHandler,
             final ShutdownHandler shutdownHandler,
             final GetPipelinesHandler getPipelinesHandler,
+            @Autowired(required = false) @Nullable final EncryptionHttpHandler encryptionHttpHandler,
             @Autowired(required = false) @Nullable final PrometheusMeterRegistry prometheusMeterRegistry,
             @Autowired(required = false) @Nullable final Authenticator authenticator
     ) {
         this.serverProvider = serverProvider;
         this.listPipelinesHandler = listPipelinesHandler;
         this.shutdownHandler = shutdownHandler;
         this.getPipelinesHandler = getPipelinesHandler;
+        this.encryptionHttpHandler = encryptionHttpHandler;
         this.prometheusMeterRegistry = prometheusMeterRegistry;
         this.authenticator = authenticator;
         executorService = Executors.newFixedThreadPool(3);
@@ -72,6 +76,10 @@ private HttpServer createServer() {
         createContext(server, shutdownHandler, authenticator, "/shutdown");
         createContext(server, getPipelinesHandler, authenticator, "/pipelines");
 
+        if (encryptionHttpHandler != null) {
+            createContext(server, encryptionHttpHandler, authenticator, "/encryption/rotate");
+        }
+
         if (prometheusMeterRegistry != null) {
             final PrometheusMetricsHandler prometheusMetricsHandler = new PrometheusMetricsHandler(prometheusMeterRegistry);
             createContext(server, prometheusMetricsHandler, authenticator, "/metrics/prometheus", "/metrics/sys");

data-prepper-core/src/test/java/org/opensearch/dataprepper/core/pipeline/server/DataPrepperServerTest.java

@@ -16,6 +16,7 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.opensearch.dataprepper.plugins.encryption.EncryptionHttpHandler;
 
 import java.net.InetSocketAddress;
 import java.util.concurrent.ExecutorService;
@@ -51,6 +52,9 @@ public class DataPrepperServerTest {
     @Mock
     private GetPipelinesHandler getPipelinesHandler;
 
+    @Mock
+    private EncryptionHttpHandler encryptionHttpHandler;
+
     @Mock
     private PrometheusMeterRegistry prometheusMeterRegistry;
 
@@ -71,15 +75,15 @@ public void tearDown() {
 
     @Test
     public void testDataPrepperServerConstructor() {
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator, null);
         assertThat(dataPrepperServer, is(notNullValue()));
     }
 
     @Test
     public void testGivenValidServerWhenStartThenShouldCallServerStart() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator, null);
         dataPrepperServer.start();
 
         verifyServerStart();
@@ -88,11 +92,25 @@ public void testGivenValidServerWhenStartThenShouldCallServerStart() {
         verify(context, times(5)).setAuthenticator(eq(authenticator));
     }
 
+    @Test
+    public void testServerStartWithEncryptionHttpHandler() {
+        mockServerStart();
+
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator, encryptionHttpHandler);
+        dataPrepperServer.start();
+
+        verifyServerStart();
+        verify(server).createContext(eq("/metrics/prometheus"), any(PrometheusMetricsHandler.class));
+        verify(server).createContext(eq("/metrics/sys"), any(PrometheusMetricsHandler.class));
+        verify(server).createContext(eq("/encryption/rotate"), any(EncryptionHttpHandler.class));
+        verify(context, times(6)).setAuthenticator(eq(authenticator));
+    }
+
     @Test
     public void testGivenValidServerWhenStartThenShouldCallServerStart_NullPrometheus() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, authenticator);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, authenticator, null);
         dataPrepperServer.start();
 
         verifyServerStart();
@@ -103,7 +121,7 @@ public void testGivenValidServerWhenStartThenShouldCallServerStart_NullPrometheu
     public void testGivenValidServerWhenStartThenShouldCallServerStart_NullAuthenticator() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, null);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, null, null);
         dataPrepperServer.start();
 
         verifyServerStart();
@@ -115,7 +133,7 @@ public void testGivenValidServerWhenStartThenShouldCallServerStart_NullAuthentic
     public void testGivenValidServerWhenStartThenShouldCallServerStart_NullPrometheusAndAuthenticator() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, null);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, null, null);
         dataPrepperServer.start();
 
         verifyServerStart();
@@ -126,7 +144,7 @@ public void testGivenValidServerWhenStartThenShouldCallServerStart_NullPrometheu
     public void testGivenValidServerWhenStopThenShouldCallServerStopWithNoDelay() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, null);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, null, null);
         dataPrepperServer.start();
         dataPrepperServer.stop();
 
@@ -161,7 +179,10 @@ private void verifyServerStart() {
         verify(socketAddress).getPort();
     }
 
-    private DataPrepperServer createObjectUnderTest(final PrometheusMeterRegistry prometheusMeterRegistry, final Authenticator authenticator) {
-        return new DataPrepperServer(httpServerProvider, listPipelinesHandler, shutdownHandler, getPipelinesHandler, prometheusMeterRegistry, authenticator);
+    private DataPrepperServer createObjectUnderTest(final PrometheusMeterRegistry prometheusMeterRegistry,
+                                                    final Authenticator authenticator,
+                                                    final EncryptionHttpHandler encryptionHttpHandler) {
+        return new DataPrepperServer(
+                httpServerProvider, listPipelinesHandler, shutdownHandler, getPipelinesHandler, encryptionHttpHandler, prometheusMeterRegistry, authenticator);
     }
 }

data-prepper-plugins/aws-plugin-api/build.gradle

@@ -2,6 +2,7 @@
 dependencies {
     implementation 'software.amazon.awssdk:auth'
     implementation 'software.amazon.awssdk:apache-client'
+    implementation 'software.amazon.awssdk:sts'
     implementation 'org.apache.httpcomponents.client5:httpclient5:5.3.1'
     implementation 'com.fasterxml.jackson.core:jackson-annotations' 
     testImplementation libs.commons.lang3

data-prepper-plugins/aws-plugin-api/src/main/java/org/opensearch/dataprepper/aws/api/AwsContext.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.aws.api;
+
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+
+/**
+ * An interface available to plugins which provides the default AWS credentials and region.
+ */
+public interface AwsContext {
+    AwsCredentialsProvider getOrDefault();
+
+    Region getRegionOrDefault();
+}

data-prepper-plugins/aws-plugin-api/src/main/java/org/opensearch/dataprepper/aws/api/AwsContextImpl.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.aws.api;
+
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
+
+import java.util.UUID;
+
+// TODO: This class should be provided as Bean in aws-plugin once the issue is resolved:
+//  https://github.com/opensearch-project/data-prepper/issues/2825
+public class AwsContextImpl implements AwsContext {
+    private final AwsCredentialsConfig awsCredentialsConfig;
+
+    public AwsContextImpl(final AwsCredentialsConfig awsCredentialsConfig) {
+        this.awsCredentialsConfig = awsCredentialsConfig;
+    }
+
+    @Override
+    public AwsCredentialsProvider getOrDefault() {
+        if (awsCredentialsConfig == null || awsCredentialsConfig.getStsRoleArn() == null) {
+            return getDefault();
+        }
+
+        return getFromOptions(awsCredentialsConfig.toCredentialsOptions());
+    }
+
+    @Override
+    public Region getRegionOrDefault() {
+        if (awsCredentialsConfig != null && awsCredentialsConfig.getRegion() != null) {
+            return Region.of(awsCredentialsConfig.getRegion());
+        }
+        return null;
+    }
+
+    private AwsCredentialsProvider getDefault() {
+        final AwsCredentialsOptions credentialsOptions;
+        if (awsCredentialsConfig != null) {
+            credentialsOptions = awsCredentialsConfig.toCredentialsOptions();
+        } else {
+            credentialsOptions = AwsCredentialsOptions.defaultOptions();
+        }
+
+        return getFromOptions(credentialsOptions);
+    }
+
+    private AwsCredentialsProvider getFromOptions(AwsCredentialsOptions awsCredentialsOptions) {
+        final AwsCredentialsProvider awsCredentialsProvider;
+        final String awsStsRoleArn = awsCredentialsOptions.getStsRoleArn();
+        if (awsStsRoleArn != null && !awsStsRoleArn.isEmpty()) {
+
+            final StsClient stsClient = StsClient.builder()
+                    .region(awsCredentialsOptions.getRegion())
+                    .build();
+
+            AssumeRoleRequest.Builder assumeRoleRequestBuilder = AssumeRoleRequest.builder()
+                    .roleSessionName("aws-iam-" + UUID.randomUUID())
+                    .roleArn(awsStsRoleArn);
+
+            awsCredentialsProvider = StsAssumeRoleCredentialsProvider.builder()
+                    .stsClient(stsClient)
+                    .refreshRequest(assumeRoleRequestBuilder.build())
+                    .build();
+
+        } else {
+            // use default credential provider
+            awsCredentialsProvider = DefaultCredentialsProvider.create();
+        }
+
+        return awsCredentialsProvider;
+    }
+}

data-prepper-plugins/aws-plugin-api/src/main/java/org/opensearch/dataprepper/aws/api/AwsCredentialsConfig.java

@@ -0,0 +1,15 @@
+/*
+ *
+ *  * Copyright OpenSearch Contributors
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.opensearch.dataprepper.aws.api;
+
+public interface AwsCredentialsConfig {
+    String getRegion();
+    String getStsRoleArn();
+
+    AwsCredentialsOptions toCredentialsOptions();
+}