Add Server Side Encryption support for S3 repository

For legal reasons, we need to store our ES backups in S3 using server side encryption. Closes elastic#78. (cherry picked from commit b3f9e12)

Author: wilkes

README.md

@@ -99,8 +99,9 @@ The following settings are supported:
 * `secret_key`: The secret key to use for authentication. Defaults to value of `cloud.aws.secret_key`.
 * `chunk_size`: Big files can be broken down into chunks during snapshotting if needed. The chunk size can be specified in bytes or by using size value notation, i.e. `1g`, `10m`, `5k`. Defaults to `100m`.
 * `compress`: When set to `true` metadata files are stored in compressed format. This setting doesn't affect index files that are already compressed by default. Defaults to `false`.
+* `server_side_encryption`: When set to `true` files are encrypted on server side using AES256 algorithm. Defaults to `false`.
 
-The S3 repositories are using the same credentials as the rest of the AWSS3 Repo documentation #83 services provided by this plugin (`discovery` and `gateway`). They can be configured the following way:
+The S3 repositories are using the same credentials as the rest of the AWS services provided by this plugin (`discovery` and `gateway`). They can be configured the following way:
 
  cloud:
   aws:

src/main/java/org/elasticsearch/cloud/aws/blobstore/S3BlobStore.java

@@ -52,12 +52,15 @@ public class S3BlobStore extends AbstractComponent implements BlobStore {
 
  private final int bufferSizeInBytes;
 
- public S3BlobStore(Settings settings, AmazonS3 client, String bucket, @Nullable String region, ThreadPool threadPool) {
+ private final boolean serverSideEncryption;
+
+ public S3BlobStore(Settings settings, AmazonS3 client, String bucket, @Nullable String region, ThreadPool threadPool, boolean serverSideEncryption) {
  super(settings);
  this.client = client;
  this.bucket = bucket;
  this.region = region;
  this.threadPool = threadPool;
+ this.serverSideEncryption = serverSideEncryption;
 
  this.bufferSizeInBytes = (int) settings.getAsBytesSize("buffer_size", new ByteSizeValue(100, ByteSizeUnit.KB)).bytes();
 
@@ -87,6 +90,8 @@ public Executor executor() {
  return threadPool.executor(ThreadPool.Names.SNAPSHOT_DATA);
  }
 
+ public boolean serverSideEncryption() { return serverSideEncryption; }
+
  public int bufferSizeInBytes() {
  return bufferSizeInBytes;
  }

src/main/java/org/elasticsearch/cloud/aws/blobstore/S3ImmutableBlobContainer.java

@@ -44,6 +44,9 @@ public void writeBlob(final String blobName, final InputStream is, final long si
  public void run() {
  try {
  ObjectMetadata md = new ObjectMetadata();
+ if (blobStore.serverSideEncryption()) {
+ md.setServerSideEncryption(ObjectMetadata.AES_256_SERVER_SIDE_ENCRYPTION);
+ }
  md.setContentLength(sizeInBytes);
  PutObjectResult objectResult = blobStore.client().putObject(blobStore.bucket(), buildKey(blobName), is, md);
  listener.onCompleted();

src/main/java/org/elasticsearch/repositories/s3/S3Repository.java

@@ -36,8 +36,6 @@
 
 import java.io.IOException;
 import java.util.Locale;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
 
 /**
  * Shared file system implementation of the BlobStoreRepository
@@ -121,8 +119,9 @@ public S3Repository(RepositoryName name, RepositorySettings repositorySettings,
  }
  }
 
- logger.debug("using bucket [{}], region [{}], chunk_size [{}]", bucket, region, chunkSize);
- blobStore = new S3BlobStore(settings, s3Service.client(region, repositorySettings.settings().get("access_key"), repositorySettings.settings().get("secret_key")), bucket, region, threadPool);
+ boolean serverSideEncryption = repositorySettings.settings().getAsBoolean("server_side_encryption", componentSettings.getAsBoolean("server_side_encryption", false));
+ logger.debug("using bucket [{}], region [{}], chunk_size [{}], server_side_encryption [{}]", bucket, region, chunkSize, serverSideEncryption);
+ blobStore = new S3BlobStore(settings, s3Service.client(region, repositorySettings.settings().get("access_key"), repositorySettings.settings().get("secret_key")), bucket, region, threadPool, serverSideEncryption);
  this.chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size", componentSettings.getAsBytesSize("chunk_size", new ByteSizeValue(100, ByteSizeUnit.MB)));
  this.compress = repositorySettings.settings().getAsBoolean("compress", componentSettings.getAsBoolean("compress", false));
  String basePath = repositorySettings.settings().get("base_path", null);

src/test/java/org/elasticsearch/repositories/s3/S3SnapshotRestoreTest.java

@@ -44,6 +44,7 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.List;
 import java.util.ArrayList;
 
 import static org.hamcrest.Matchers.equalTo;
@@ -152,6 +153,93 @@ public void testSimpleWorkflow() {
  assertThat(clusterState.getMetaData().hasIndex("test-idx-1"), equalTo(true));
  assertThat(clusterState.getMetaData().hasIndex("test-idx-2"), equalTo(false));
  }
+ 
+ @Test
+ public void testEncryption() {
+Client client = client();
+logger.info("--> creating s3 repository with bucket[{}] and path [{}]", cluster().getInstance(Settings.class).get("repositories.s3.bucket"), basePath);
+PutRepositoryResponse putRepositoryResponse = client.admin().cluster().preparePutRepository("test-repo")
+.setType("s3").setSettings(ImmutableSettings.settingsBuilder()
+.put("base_path", basePath)
+.put("chunk_size", randomIntBetween(1000, 10000))
+.put("server_side_encryption", true)
+).get();
+assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
+
+createIndex("test-idx-1", "test-idx-2", "test-idx-3");
+ensureGreen();
+
+logger.info("--> indexing some data");
+for (int i = 0; i < 100; i++) {
+ index("test-idx-1", "doc", Integer.toString(i), "foo", "bar" + i);
+ index("test-idx-2", "doc", Integer.toString(i), "foo", "baz" + i);
+ index("test-idx-3", "doc", Integer.toString(i), "foo", "baz" + i);
+}
+refresh();
+assertThat(client.prepareCount("test-idx-1").get().getCount(), equalTo(100L));
+assertThat(client.prepareCount("test-idx-2").get().getCount(), equalTo(100L));
+assertThat(client.prepareCount("test-idx-3").get().getCount(), equalTo(100L));
+
+logger.info("--> snapshot");
+CreateSnapshotResponse createSnapshotResponse = client.admin().cluster().prepareCreateSnapshot("test-repo", "test-snap").setWaitForCompletion(true).setIndices("test-idx-*", "-test-idx-3").get();
+assertThat(createSnapshotResponse.getSnapshotInfo().successfulShards(), greaterThan(0));
+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));
+
+Settings settings = cluster().getInstance(Settings.class);
+Settings bucket = settings.getByPrefix("repositories.s3.");
+AmazonS3 s3Client = cluster().getInstance(AwsS3Service.class).client(
+bucket.get("region", settings.get("repositories.s3.region")),
+bucket.get("access_key", settings.get("cloud.aws.access_key")),
+bucket.get("secret_key", settings.get("cloud.aws.secret_key")));
+
+ String bucketName = bucket.get("bucket");
+logger.info("--> verify encryption for bucket [{}], prefix [{}]", bucketName, basePath);
+List<S3ObjectSummary> summaries = s3Client.listObjects(bucketName, basePath).getObjectSummaries();
+for (S3ObjectSummary summary : summaries) {
+ assertThat(s3Client.getObjectMetadata(bucketName, summary.getKey()).getServerSideEncryption(), equalTo("AES256"));
+}
+
+logger.info("--> delete some data");
+for (int i = 0; i < 50; i++) {
+ client.prepareDelete("test-idx-1", "doc", Integer.toString(i)).get();
+}
+for (int i = 50; i < 100; i++) {
+ client.prepareDelete("test-idx-2", "doc", Integer.toString(i)).get();
+}
+for (int i = 0; i < 100; i += 2) {
+ client.prepareDelete("test-idx-3", "doc", Integer.toString(i)).get();
+}
+refresh();
+assertThat(client.prepareCount("test-idx-1").get().getCount(), equalTo(50L));
+assertThat(client.prepareCount("test-idx-2").get().getCount(), equalTo(50L));
+assertThat(client.prepareCount("test-idx-3").get().getCount(), equalTo(50L));
+
+logger.info("--> close indices");
+client.admin().indices().prepareClose("test-idx-1", "test-idx-2").get();
+
+logger.info("--> restore all indices from the snapshot");
+RestoreSnapshotResponse restoreSnapshotResponse = client.admin().cluster().prepareRestoreSnapshot("test-repo", "test-snap").setWaitForCompletion(true).execute().actionGet();
+assertThat(restoreSnapshotResponse.getRestoreInfo().totalShards(), greaterThan(0));
+
+ensureGreen();
+assertThat(client.prepareCount("test-idx-1").get().getCount(), equalTo(100L));
+assertThat(client.prepareCount("test-idx-2").get().getCount(), equalTo(100L));
+assertThat(client.prepareCount("test-idx-3").get().getCount(), equalTo(50L));
+
+// Test restore after index deletion
+logger.info("--> delete indices");
+cluster().wipeIndices("test-idx-1", "test-idx-2");
+logger.info("--> restore one index after deletion");
+restoreSnapshotResponse = client.admin().cluster().prepareRestoreSnapshot("test-repo", "test-snap").setWaitForCompletion(true).setIndices("test-idx-*", "-test-idx-2").execute().actionGet();
+assertThat(restoreSnapshotResponse.getRestoreInfo().totalShards(), greaterThan(0));
+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));
+assertThat(clusterState.getMetaData().hasIndex("test-idx-2"), equalTo(false));
+ }
 
  /**
  * This test verifies that the test configuration is set up in a manner that