Add client-side encryption

The Java Cryptography Extension (JCE) has to be installed to use this feature.

core/src/main/java/org/elasticsearch/http/HttpServerTransport.java

@@ -19,6 +19,11 @@
 
 package org.elasticsearch.http;
 
+<<<<<<< HEAD:core/src/main/java/org/elasticsearch/http/HttpServerTransport.java
+=======
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.EncryptionMaterials;
+>>>>>>> 98d508f... Add client-side encryption:src/main/java/org/elasticsearch/cloud/aws/AwsS3Service.java
 import org.elasticsearch.common.component.LifecycleComponent;
 import org.elasticsearch.common.transport.BoundTransportAddress;
 
@@ -29,9 +34,15 @@ public interface HttpServerTransport extends LifecycleComponent<HttpServerTransp
 
     BoundTransportAddress boundAddress();
 
+<<<<<<< HEAD:core/src/main/java/org/elasticsearch/http/HttpServerTransport.java
     HttpInfo info();
 
     HttpStats stats();
 
     void httpServerAdapter(HttpServerAdapter httpServerAdapter);
+=======
+    AmazonS3 client(String endpoint, String protocol, String region, String account, String key, Integer maxRetries);
+
+    AmazonS3 client(String endpoint, String protocol, String region, String account, String key, Integer maxRetries, EncryptionMaterials clientSideEncryptionMaterials);
+>>>>>>> 98d508f... Add client-side encryption:src/main/java/org/elasticsearch/cloud/aws/AwsS3Service.java
 }

plugins/repository-s3/src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java

@@ -31,6 +31,15 @@ import com.amazonaws.http.IdleConnectionReaper;
 import com.amazonaws.internal.StaticCredentialsProvider;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.AmazonS3Client;
+<<<<<<< HEAD:plugins/repository-s3/src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java
+=======
+
+import com.amazonaws.services.s3.AmazonS3EncryptionClient;
+import com.amazonaws.services.s3.model.CryptoConfiguration;
+import com.amazonaws.services.s3.model.EncryptionMaterials;
+import com.amazonaws.services.s3.model.EncryptionMaterialsProvider;
+import com.amazonaws.services.s3.model.StaticEncryptionMaterialsProvider;
+>>>>>>> 98d508f... Add client-side encryption:src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.Tuple;
@@ -47,9 +56,15 @@ import java.util.Map;
 public class InternalAwsS3Service extends AbstractLifecycleComponent<AwsS3Service> implements AwsS3Service {
 
     /**
+<<<<<<< HEAD:plugins/repository-s3/src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java
      * (acceskey, endpoint) -&gt; client
      */
     private Map<Tuple<String, String>, AmazonS3Client> clients = new HashMap<>();
+=======
+     * (acceskey, (endpoint, clientSideEncryptionKey)) -> client
+     */
+    private Map<Tuple<String, Tuple<String, EncryptionMaterials>>, AmazonS3Client> clients = new HashMap<Tuple<String,Tuple<String, EncryptionMaterials>>, AmazonS3Client>();
+>>>>>>> 98d508f... Add client-side encryption:src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java
 
     @Inject
     public InternalAwsS3Service(Settings settings) {
@@ -57,6 +72,7 @@ public class InternalAwsS3Service extends AbstractLifecycleComponent<AwsS3Servic
     }
 
     @Override
+<<<<<<< HEAD:plugins/repository-s3/src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java
     public synchronized AmazonS3 client(String endpoint, Protocol protocol, String region, String account, String key, Integer maxRetries) {
         if (Strings.isNullOrEmpty(endpoint)) {
             // We need to set the endpoint based on the region
@@ -67,13 +83,50 @@ public class InternalAwsS3Service extends AbstractLifecycleComponent<AwsS3Servic
                 // No region has been set so we will use the default endpoint
                 endpoint = getDefaultEndpoint();
             }
+=======
+    public synchronized AmazonS3 client() {
+        String endpoint = getDefaultEndpoint();
+        String account = settings.get("cloud.aws.access_key", settings.get("cloud.account"));
+        String key = settings.get("cloud.aws.secret_key", settings.get("cloud.key"));
+
+        return getClient(endpoint, null, account, key, null, null);
     }
 
-        return getClient(endpoint, protocol, account, key, maxRetries);
+    @Override
+    public AmazonS3 client(String endpoint, String protocol, String region, String account, String key) {
+        return client(endpoint, protocol, region, account, key, null);
     }
 
+    @Override
+    public synchronized AmazonS3 client(String endpoint, String protocol, String region, String account, String key, Integer maxRetries) {
+        return client(endpoint, protocol, region, account, key, maxRetries, null);
+    }
+
+    @Override
+    public synchronized AmazonS3 client(String endpoint, String protocol, String region, String account, String key, Integer maxRetries, EncryptionMaterials clientSideEncryptionMaterials) {
+        if (region != null && endpoint == null) {
+            endpoint = getEndpoint(region);
+            logger.debug("using s3 region [{}], with endpoint [{}]", region, endpoint);
+        } else if (endpoint == null) {
+            endpoint = getDefaultEndpoint();
+        }
+        if (account == null || key == null) {
+            account = settings.get("cloud.aws.access_key", settings.get("cloud.account"));
+            key = settings.get("cloud.aws.secret_key", settings.get("cloud.key"));
+>>>>>>> 98d508f... Add client-side encryption:src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java
+        }
+
+        return getClient(endpoint, protocol, account, key, maxRetries, clientSideEncryptionMaterials);
+    }
+
+<<<<<<< HEAD:plugins/repository-s3/src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java
     private synchronized AmazonS3 getClient(String endpoint, Protocol protocol, String account, String key, Integer maxRetries) {
         Tuple<String, String> clientDescriptor = new Tuple<>(endpoint, account);
+=======
+
+    private synchronized AmazonS3 getClient(String endpoint, String protocol, String account, String key, Integer maxRetries, EncryptionMaterials clientSideEncryptionMaterials) {
+        Tuple<String, Tuple<String, EncryptionMaterials>> clientDescriptor = new Tuple<String, Tuple<String, EncryptionMaterials>>(endpoint, new Tuple(account, clientSideEncryptionMaterials));
+>>>>>>> 98d508f... Add client-side encryption:src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java
         AmazonS3Client client = clients.get(clientDescriptor);
         if (client != null) {
             return client;
@@ -123,7 +176,19 @@ public class InternalAwsS3Service extends AbstractLifecycleComponent<AwsS3Servic
                     new StaticCredentialsProvider(new BasicAWSCredentials(account, key))
             );
         }
+
+        if(clientSideEncryptionMaterials != null) {
+            EncryptionMaterialsProvider encryptionMaterialsProvider = new StaticEncryptionMaterialsProvider(clientSideEncryptionMaterials);
+            CryptoConfiguration cryptoConfiguration = new CryptoConfiguration();
+            client = new AmazonS3EncryptionClient(
+                    credentials,
+                    encryptionMaterialsProvider,
+                    clientConfiguration,
+                    cryptoConfiguration
+            );
+        } else {
             client = new AmazonS3Client(credentials, clientConfiguration);
+        }
 
         if (endpoint != null) {
             client.setEndpoint(endpoint);

plugins/repository-s3/src/main/java/org/elasticsearch/cloud/aws/blobstore/S3BlobStore.java

@@ -22,8 +22,12 @@ package org.elasticsearch.cloud.aws.blobstore;
 import com.amazonaws.AmazonClientException;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.model.AmazonS3Exception;
+<<<<<<< HEAD:plugins/repository-s3/src/main/java/org/elasticsearch/cloud/aws/blobstore/S3BlobStore.java
 import com.amazonaws.services.s3.model.CannedAccessControlList;
 import com.amazonaws.services.s3.model.CreateBucketRequest;
+=======
+import com.amazonaws.services.s3.AmazonS3EncryptionClient;
+>>>>>>> 98d508f... Add client-side encryption:src/main/java/org/elasticsearch/cloud/aws/blobstore/S3BlobStore.java
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion;
 import com.amazonaws.services.s3.model.ObjectListing;
@@ -70,6 +74,12 @@ public class S3BlobStore extends AbstractComponent implements BlobStore {
         this.region = region;
         this.serverSideEncryption = serverSideEncryption;
         this.bufferSize = bufferSize;
+
+        if (client instanceof AmazonS3EncryptionClient && this.bufferSize.getBytes() % 16 > 0) {
+            throw new BlobStoreException("Detected client-side encryption " +
+                "and a buffer_size for the S3 storage not a multiple of the cipher block size (16)");
+        }
+        
         this.cannedACL = initCannedACL(cannedACL);
         this.numberOfRetries = maxRetries;
         this.storageClass = initStorageClass(storageClass);

plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/AbstractS3SnapshotRestoreTest.java

@@ -23,7 +23,15 @@ import com.amazonaws.Protocol;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
 import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3Object;
 import com.amazonaws.services.s3.model.S3ObjectSummary;
+<<<<<<< HEAD:plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/AbstractS3SnapshotRestoreTest.java
+=======
+
+import com.amazonaws.util.Base64;
+import com.carrotsearch.ant.tasks.junit4.dependencies.com.google.gson.stream.JsonReader;
+import com.carrotsearch.ant.tasks.junit4.dependencies.com.google.gson.stream.MalformedJsonException;
+>>>>>>> 98d508f... Add client-side encryption:src/test/java/org/elasticsearch/repositories/s3/AbstractS3SnapshotRestoreTest.java
 import org.elasticsearch.action.admin.cluster.repositories.put.PutRepositoryResponse;
 import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
@@ -33,6 +41,11 @@ import org.elasticsearch.cloud.aws.AbstractAwsTestCase;
 import org.elasticsearch.cloud.aws.AwsS3Service;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.settings.Settings;
+<<<<<<< HEAD:plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/AbstractS3SnapshotRestoreTest.java
+=======
+import org.elasticsearch.plugins.PluginsService;
+import org.elasticsearch.repositories.RepositoryException;
+>>>>>>> 98d508f... Add client-side encryption:src/test/java/org/elasticsearch/repositories/s3/AbstractS3SnapshotRestoreTest.java
 import org.elasticsearch.repositories.RepositoryMissingException;
 import org.elasticsearch.repositories.RepositoryVerificationException;
 import org.elasticsearch.snapshots.SnapshotMissingException;
@@ -42,7 +55,11 @@ import org.elasticsearch.test.ESIntegTestCase.Scope;
 import org.junit.After;
 import org.junit.Before;
 
+import javax.crypto.KeyGenerator;
+import java.io.InputStreamReader;
+import java.security.*;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.hamcrest.Matchers.equalTo;
@@ -118,6 +135,7 @@ abstract public class AbstractS3SnapshotRestoreTest extends AbstractAwsTestCase
         assertThat(createSnapshotResponse.getSnapshotInfo().successfulShards(), equalTo(createSnapshotResponse.getSnapshotInfo().totalShards()));
 
         assertThat(client.admin().cluster().prepareGetSnapshots("test-repo").setSnapshots("test-snap").get().getSnapshots().get(0).state(), equalTo(SnapshotState.SUCCESS));
+        assertMetadataFileIsNotEncrypted("test-snap");
 
         logger.info("--> delete some data");
         for (int i = 0; i < 50; i++) {
@@ -249,6 +267,77 @@ abstract public class AbstractS3SnapshotRestoreTest extends AbstractAwsTestCase
         assertThat(clusterState.getMetaData().hasIndex("test-idx-2"), equalTo(false));
     }
 
+    @Test @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch-cloud-aws/issues/211")
+    public void testClientSideEncryption() throws NoSuchAlgorithmException {
+
+        KeyGenerator keyGenerator1 = KeyGenerator.getInstance("AES");
+        keyGenerator1.init(128);
+        String symmetricEncryptionKeyBase64 = Base64.encodeAsString(keyGenerator1.generateKey().getEncoded());
+
+        KeyPairGenerator keyGenerator2 = KeyPairGenerator.getInstance("RSA");
+        keyGenerator2.initialize(512, new SecureRandom());
+        KeyPair keyPair = keyGenerator2.generateKeyPair();
+        String publicEncryptionKeyBase64 = Base64.encodeAsString(keyPair.getPublic().getEncoded());
+        String privateEncryptionKeyBase64 = Base64.encodeAsString(keyPair.getPrivate().getEncoded());
+
+        Client client = client();
+        try {
+            PutRepositoryResponse putRepositoryResponse = client.admin().cluster().preparePutRepository("test-repo")
+                    .setType("s3").setSettings(ImmutableSettings.settingsBuilder()
+                                    .put("base_path", basePath)
+                                    .put("client_side_encryption_key.symmetric", symmetricEncryptionKeyBase64)
+                                    .put("client_side_encryption_key.public", publicEncryptionKeyBase64)
+                                    .put("client_side_encryption_key.private", privateEncryptionKeyBase64)
+                                    .put("chunk_size", randomIntBetween(1000, 10000))
+                    ).get();
+            fail("Symmetric and public/private key pairs are exclusive options. An exception should be thrown.");
+        } catch(RepositoryException e) {
+        }
+
+        List<ImmutableSettings.Builder> allSettings = Arrays.asList(
+            ImmutableSettings.settingsBuilder()
+                    .put("base_path", basePath)
+                    .put("client_side_encryption_key.symmetric", symmetricEncryptionKeyBase64)
+                    .put("chunk_size", randomIntBetween(1000, 10000)),
+            ImmutableSettings.settingsBuilder()
+                    .put("base_path", basePath)
+                    .put("client_side_encryption_key.public", publicEncryptionKeyBase64)
+                    .put("client_side_encryption_key.private", privateEncryptionKeyBase64)
+                    .put("chunk_size", randomIntBetween(1000, 10000))
+        );
+        for(ImmutableSettings.Builder settings: allSettings) {
+            PutRepositoryResponse putRepositoryResponse = client.admin().cluster().preparePutRepository("test-repo")
+                    .setType("s3").setSettings(settings).get();
+
+            // Create the index and index some data
+            createIndex("test-idx-1");
+            for (int i = 0; i < 100; i++) {
+                index("test-idx-1", "doc", Integer.toString(i), "foo", "bar" + i);
+            }
+            refresh();
+
+            // Take the snapshot
+            CreateSnapshotResponse createSnapshotResponse = client.admin().cluster().prepareCreateSnapshot("test-repo", "test-snap").setWaitForCompletion(true).setIndices("test-idx-1").get();
+            assertThat(createSnapshotResponse.getSnapshotInfo().successfulShards(), greaterThan(0));
+            assertThat(createSnapshotResponse.getSnapshotInfo().successfulShards(), equalTo(createSnapshotResponse.getSnapshotInfo().totalShards()));
+
+            assertMetadataFileIsEncrypted("test-snap");
+
+            // Restore
+            cluster().wipeIndices("test-idx-1");
+            RestoreSnapshotResponse restoreSnapshotResponse = client.admin().cluster().prepareRestoreSnapshot("test-repo", "test-snap").setWaitForCompletion(true).setIndices("test-idx-1").execute().actionGet();
+            ensureGreen();
+            assertThat(client.prepareCount("test-idx-1").get().getCount(), equalTo(100L));
+            ClusterState clusterState = client.admin().cluster().prepareState().get().getState();
+            assertThat(clusterState.getMetaData().hasIndex("test-idx-1"), equalTo(true));
+
+            // Clean, the test will bbe run with different settings
+            cluster().wipeIndices("test-idx-1");
+            wipeRepositories();
+            cleanRepositoryFiles(basePath);
+        }
+    }
+
     /**
      * This test verifies that the test configuration is set up in a manner that
      * does not make the test {@link #testRepositoryWithCustomCredentials()} pointless.
@@ -435,6 +524,51 @@ abstract public class AbstractS3SnapshotRestoreTest extends AbstractAwsTestCase
         assertThat(client.prepareSearch("test-idx-1").setSize(0).get().getHits().totalHits(), equalTo(100L));
     }
 
+    private void assertMetadataFileIsEncrypted(String snapshotName) {
+
+        Settings settings = internalCluster().getInstance(Settings.class);
+        AmazonS3 s3Client = internalCluster().getInstance(AwsS3Service.class).client(
+                settings.get("repositories.s3.endpoint"),
+                settings.get("repositories.s3.protocol"),
+                settings.get("repositories.s3.region"),
+                settings.get("cloud.aws.access_key"),
+                settings.get("cloud.aws.secret_key"));
+        String bucket = settings.get("repositories.s3.bucket");
+        String objectKey = basePath + "/metadata-" + snapshotName;
+        S3Object object = s3Client.getObject(bucket, objectKey);
+
+        try {
+            JsonReader jsonReader = new JsonReader(new InputStreamReader(object.getObjectContent()));
+            jsonReader.beginObject();
+            assertThat("The file hasn't been encrypted properly, its content is still readable!", jsonReader.nextName(), not(equalTo("meta-data")));
+        } catch(Exception e) {
+            // The json is not valid, the file is encrypted
+
+            // MalformedJsonException can't be catched directly so the following
+            //   assertion is necessary to avoid silent failures.
+            assertThat(e, instanceOf(MalformedJsonException.class));
+        }
+    }
+
+    private void assertMetadataFileIsNotEncrypted(String snapshotName) {
+
+        Settings settings = internalCluster().getInstance(Settings.class);
+        AmazonS3 s3Client = internalCluster().getInstance(AwsS3Service.class).client(
+                settings.get("repositories.s3.endpoint"),
+                settings.get("repositories.s3.protocol"),
+                settings.get("repositories.s3.region"),
+                settings.get("cloud.aws.access_key"),
+                settings.get("cloud.aws.secret_key"));
+        String bucket = settings.get("repositories.s3.bucket");
+        String objectKey = basePath + "/metadata-" + snapshotName;
+        S3Object object = s3Client.getObject(bucket, objectKey);
+
+        JsonReader jsonReader = new JsonReader(new InputStreamReader(object.getObjectContent()));
+        jsonReader.beginObject();
+        assertThat("The file wasn't decrypted properly", jsonReader.nextName(), equalTo("meta-data"));
+
+        // The beginning of the file looks like json. If it was encrypted, it wouldn't.
+    }
 
     /**
      * Deletes repositories, supports wildcard notation.