crypto plugin integration changes

server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java

@@ -45,6 +45,7 @@
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.InternalClusterInfoService;
 import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.cluster.metadata.RepositoryMetadata;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.routing.RecoverySource;
 import org.opensearch.cluster.routing.ShardRouting;
@@ -63,6 +64,7 @@
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.xcontent.XContentType;
 import org.opensearch.common.util.io.IOUtils;
+import org.opensearch.crypto.CryptoClient;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
 import org.opensearch.env.ShardLock;
@@ -81,6 +83,7 @@
 import org.opensearch.index.translog.InternalTranslogFactory;
 import org.opensearch.index.translog.TestTranslog;
 import org.opensearch.index.translog.Translog;
+import org.opensearch.index.translog.TranslogFactory;
 import org.opensearch.index.translog.TranslogStats;
 import org.opensearch.indices.IndicesService;
 import org.opensearch.indices.breaker.CircuitBreakerService;
@@ -699,7 +702,17 @@ public static final IndexShard newIndexShard(
             () -> {},
             RetentionLeaseSyncer.EMPTY,
             cbs,
-            (indexSettings, shardRouting) -> new InternalTranslogFactory(),
+            new IndicesService.TranslogFactorySupplier() {
+                @Override
+                public TranslogFactory createTranslogFactory(IndexSettings indexSettings, ShardRouting shardRouting) {
+                    return new InternalTranslogFactory();
+                }
+
+                @Override
+                public CryptoClient createCryptoClient(RepositoryMetadata repositoryMetadata) {
+                    return null;
+                }
+            },
             SegmentReplicationCheckpointPublisher.EMPTY,
             null
         );

server/src/main/java/org/opensearch/cluster/metadata/RepositoryMetadata.java

@@ -61,6 +61,11 @@ public class RepositoryMetadata implements Writeable {
      */
     private final long pendingGeneration;
 
+    /**
+     * Whether repository is encrypted
+     */
+    private final Boolean encrypted;
+
     /**
      * Constructs new repository metadata
      *
@@ -69,14 +74,22 @@ public class RepositoryMetadata implements Writeable {
      * @param settings repository settings
      */
     public RepositoryMetadata(String name, String type, Settings settings) {
-        this(name, type, settings, RepositoryData.UNKNOWN_REPO_GEN, RepositoryData.EMPTY_REPO_GEN);
+        this(name, type, settings, RepositoryData.UNKNOWN_REPO_GEN, RepositoryData.EMPTY_REPO_GEN, null);
+    }
+
+    public RepositoryMetadata(String name, String type, Settings settings, Boolean encrypted) {
+        this(name, type, settings, RepositoryData.UNKNOWN_REPO_GEN, RepositoryData.EMPTY_REPO_GEN, encrypted);
     }
 
     public RepositoryMetadata(RepositoryMetadata metadata, long generation, long pendingGeneration) {
-        this(metadata.name, metadata.type, metadata.settings, generation, pendingGeneration);
+        this(metadata.name, metadata.type, metadata.settings, generation, pendingGeneration, null);
     }
 
     public RepositoryMetadata(String name, String type, Settings settings, long generation, long pendingGeneration) {
+        this(name, type, settings, generation, pendingGeneration, null);
+    }
+
+    public RepositoryMetadata(String name, String type, Settings settings, long generation, long pendingGeneration, Boolean encrypted) {
         this.name = name;
         this.type = type;
         this.settings = settings;
@@ -87,6 +100,7 @@ public RepositoryMetadata(String name, String type, Settings settings, long gene
             + "] must be greater or equal to generation ["
             + generation
             + "]";
+        this.encrypted = encrypted;
     }
 
     /**
@@ -116,6 +130,15 @@ public Settings settings() {
         return this.settings;
     }
 
+    /**
+     * Returns whether repository is encrypted
+     *
+     * @return whether repository is encrypted
+     */
+    public Boolean encrypted() {
+        return null;
+    }
+
     /**
      * Returns the safe repository generation. {@link RepositoryData} for this generation is assumed to exist in the repository.
      * All operations on the repository must be based on the {@link RepositoryData} at this generation.
@@ -146,6 +169,7 @@ public RepositoryMetadata(StreamInput in) throws IOException {
         settings = Settings.readSettingsFromStream(in);
         generation = in.readLong();
         pendingGeneration = in.readLong();
+        encrypted = null;
     }
 
     /**

server/src/main/java/org/opensearch/common/Stream.java

@@ -0,0 +1,55 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.common;
+
+import java.io.InputStream;
+
+/**
+ * Model composed of an input stream, the total content length and offset
+ */
+public class Stream {
+
+    private final InputStream inputStream;
+    private final long contentLength;
+    private final long offset;
+
+    /**
+     * Construct a new stream object
+     *
+     * @param inputStream The input stream that is to be encapsulated
+     * @param contentLength The total content length that is to be read from the stream
+     * @param offset The offset pointer that this stream reads from in the file
+     */
+    public Stream(InputStream inputStream, long contentLength, long offset) {
+        this.inputStream = inputStream;
+        this.contentLength = contentLength;
+        this.offset = offset;
+    }
+
+    /**
+     * @return The input stream this object is reading from
+     */
+    public InputStream getInputStream() {
+        return inputStream;
+    }
+
+    /**
+     * @return The total length of the content that has to be read from this stream
+     */
+    public long getContentLength() {
+        return contentLength;
+    }
+
+    /**
+     * @return The offset pointer in the file that this stream is reading from
+     */
+    public long getOffset() {
+        return offset;
+    }
+}

server/src/main/java/org/opensearch/crypto/CryptoClient.java

@@ -0,0 +1,92 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.crypto;
+
+import org.opensearch.common.Stream;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.concurrent.RefCounted;
+
+import java.io.InputStream;
+
+/**
+ * Crypto plugin interface used for encryption and decryption.
+ */
+public interface CryptoClient extends RefCounted {
+
+    /**
+     * A factory interface for constructing crypto client.
+     *
+     */
+    interface Factory {
+
+        /**
+         * Constructs a crypto client used for encryption and decryption
+         *
+         * @param cryptoSettings Settings needed for creating crypto client.
+         * @param keyProviderName Name of the key provider.
+         * @return instance of CryptoClient
+         */
+        CryptoClient create(Settings cryptoSettings, String keyProviderName);
+    }
+
+    /**
+     * @return key provider type
+     */
+    String type();
+
+    /**
+     * @return key provider name
+     */
+    String name();
+
+    /**
+     * To Initialise a crypto context used in encryption. This might be needed to set the context before beginning
+     * encryption.
+     *
+     * @return crypto context instance
+     */
+    Object initCryptoContext();
+
+    /**
+     * In scenarios where content is divided into multiple parts and streams are emitted against each part,
+     * it is sometimes required to adjust the size of a part.
+     *
+     * @param cryptoContextObj crypto context instance
+     * @param streamSize Size of the raw stream
+     * @return Adjusted size of the stream.
+     */
+    long adjustStreamSize(Object cryptoContextObj, long streamSize);
+
+    /**
+     * Wraps a raw InputStream with encrypting stream
+     *
+     * @param cryptoContext created earlier to set the crypto context.
+     * @param stream Raw InputStream to encrypt
+     * @return encrypting stream wrapped around raw InputStream.
+     */
+    Stream createEncryptingStream(Object cryptoContext, Stream stream);
+
+    /**
+     * Provides encrypted stream for a raw stream emitted for a part of content.
+     *
+     * @param cryptoContextObj crypto context instance.
+     * @param stream raw stream for which encrypted stream has to be created.
+     * @param totalStreams Number of streams being used for the entire content.
+     * @param streamIdx Index of the current stream.
+     * @return Encrypted stream for the provided raw stream.
+     */
+    Stream createEncryptingStreamOfPart(Object cryptoContextObj, Stream stream, int totalStreams, int streamIdx);
+
+    /**
+     * This method accepts an encrypted stream and provides a decrypting wrapper.
+     * @param encryptingStream to be decrypted.
+     * @return Decrypting wrapper stream
+     */
+    InputStream createDecryptingStream(InputStream encryptingStream);
+}

server/src/main/java/org/opensearch/crypto/package-info.java

@@ -0,0 +1,12 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+/**
+ * Package for crypto client abstractions and exceptions.
+ */
+package org.opensearch.crypto;

server/src/main/java/org/opensearch/index/IndexModule.java

@@ -43,7 +43,6 @@
 import org.opensearch.Version;
 import org.opensearch.client.Client;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
-import org.opensearch.cluster.routing.ShardRouting;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.CheckedFunction;
 import org.opensearch.common.SetOnce;
@@ -73,8 +72,8 @@
 import org.opensearch.index.store.FsDirectoryFactory;
 import org.opensearch.index.store.remote.directory.RemoteSnapshotDirectoryFactory;
 import org.opensearch.index.store.remote.filecache.FileCache;
-import org.opensearch.index.translog.TranslogFactory;
 import org.opensearch.indices.IndicesQueryCache;
+import org.opensearch.indices.IndicesService;
 import org.opensearch.indices.breaker.CircuitBreakerService;
 import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache;
 import org.opensearch.indices.mapper.MapperRegistry;
@@ -504,7 +503,7 @@ public IndexService newIndexService(
         BooleanSupplier idFieldDataEnabled,
         ValuesSourceRegistry valuesSourceRegistry,
         IndexStorePlugin.DirectoryFactory remoteDirectoryFactory,
-        BiFunction<IndexSettings, ShardRouting, TranslogFactory> translogFactorySupplier
+        IndicesService.TranslogFactorySupplier translogFactorySupplier
     ) throws IOException {
         final IndexEventListener eventListener = freeze();
         Function<IndexService, CheckedFunction<DirectoryReader, DirectoryReader, IOException>> readerWrapperFactory = indexReaderWrapper

server/src/main/java/org/opensearch/index/IndexService.java

@@ -89,7 +89,7 @@
 import org.opensearch.index.similarity.SimilarityService;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.translog.Translog;
-import org.opensearch.index.translog.TranslogFactory;
+import org.opensearch.indices.IndicesService;
 import org.opensearch.indices.breaker.CircuitBreakerService;
 import org.opensearch.indices.cluster.IndicesClusterStateService;
 import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache;
@@ -114,7 +114,6 @@
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.BiFunction;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -174,7 +173,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final IndexNameExpressionResolver expressionResolver;
     private final Supplier<Sort> indexSortSupplier;
     private final ValuesSourceRegistry valuesSourceRegistry;
-    private final BiFunction<IndexSettings, ShardRouting, TranslogFactory> translogFactorySupplier;
+    private final IndicesService.TranslogFactorySupplier translogFactorySupplier;
 
     public IndexService(
         IndexSettings indexSettings,
@@ -207,7 +206,7 @@ public IndexService(
         IndexNameExpressionResolver expressionResolver,
         ValuesSourceRegistry valuesSourceRegistry,
         IndexStorePlugin.RecoveryStateFactory recoveryStateFactory,
-        BiFunction<IndexSettings, ShardRouting, TranslogFactory> translogFactorySupplier
+        IndicesService.TranslogFactorySupplier translogFactorySupplier
     ) {
         super(indexSettings);
         this.allowExpensiveQueries = allowExpensiveQueries;