Manifest list encryption

api/src/main/java/org/apache/iceberg/ManifestListFile.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import java.nio.ByteBuffer;
+import org.apache.iceberg.encryption.EncryptionManager;
+
+public interface ManifestListFile {
+
+  /** Location of manifest list file. */
+  String location();
+
+  /** Snapshot ID of the manifest list. */
+  long snapshotId();
+
+  /**
+   * The manifest list key metadata is encrypted with a "key encryption key" (KEK). Returns the KEK
+   * ID for this manifest file.
+   */
+  String keyMetadataKeyId();
+
+  /** Returns the manifest list key metadata, encrypted with its KEK. */
+  ByteBuffer encryptedKeyMetadata();
+
+  /** Decrypt and return the encrypted key metadata */
+  ByteBuffer decryptKeyMetadata(EncryptionManager em);
+}

api/src/main/java/org/apache/iceberg/Snapshot.java

@@ -162,6 +162,16 @@ default Iterable<DeleteFile> removedDeleteFiles(FileIO io) {
    */
   String manifestListLocation();
 
+  /**
+   * This snapshot's manifest list file info: size, encryption key metadata and location
+   *
+   * @return manifest list file info
+   */
+  default ManifestListFile manifestListFile() {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " doesn't implement manifestListFile method");
+  }
+
   /**
    * Return the id of the schema used when this snapshot was created, or null if this information is
    * not available.

api/src/main/java/org/apache/iceberg/encryption/EncryptingFileIO.java

@@ -28,6 +28,7 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestListFile;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
@@ -109,13 +110,21 @@ public InputFile newInputFile(ManifestFile manifest) {
     }
   }
 
+  @Override
+  public InputFile newInputFile(ManifestListFile manifestList) {
+    if (manifestList.encryptedKeyMetadata() != null) {
+      ByteBuffer keyMetadata = manifestList.decryptKeyMetadata(em);
+      return newDecryptingInputFile(manifestList.location(), keyMetadata);
+    } else {
+      return newInputFile(manifestList.location());
+    }
+  }
+
   public InputFile newDecryptingInputFile(String path, ByteBuffer buffer) {
     return em.decrypt(wrap(io.newInputFile(path), buffer));
   }
 
   public InputFile newDecryptingInputFile(String path, long length, ByteBuffer buffer) {
-    // TODO: is the length correct for the encrypted file? It may be the length of the plaintext
-    // stream
     return em.decrypt(wrap(io.newInputFile(path, length), buffer));
   }
 

api/src/main/java/org/apache/iceberg/io/FileIO.java

@@ -24,6 +24,7 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestListFile;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
 /**
@@ -70,6 +71,15 @@ default InputFile newInputFile(ManifestFile manifest) {
     return newInputFile(manifest.path(), manifest.length());
   }
 
+  default InputFile newInputFile(ManifestListFile manifestList) {
+    Preconditions.checkArgument(
+        manifestList.encryptedKeyMetadata() == null,
+        "Cannot decrypt manifest list: %s (use EncryptingFileIO)",
+        manifestList.location());
+    // cannot pass length because it is not tracked outside of key metadata
+    return newInputFile(manifestList.location());
+  }
+
   /** Get a {@link OutputFile} instance to write bytes to the file at the given path. */
   OutputFile newOutputFile(String path);
 

core/src/main/java/org/apache/iceberg/BaseManifestListFile.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.encryption.EncryptionUtil;
+import org.apache.iceberg.util.ByteBuffers;
+
+class BaseManifestListFile implements ManifestListFile, Serializable {
+  private final String location;
+  private final long snapshotId;
+  private final String keyMetadataKeyID;
+  // stored as a byte array to be Serializable
+  private final byte[] encryptedKeyMetadata;
+
+  BaseManifestListFile(
+      String location,
+      long snapshotId,
+      String keyMetadataKeyID,
+      ByteBuffer encryptedKeyMetadata) {
+    this.location = location;
+    this.snapshotId = snapshotId;
+    this.encryptedKeyMetadata = ByteBuffers.toByteArray(encryptedKeyMetadata);
+    this.keyMetadataKeyID = keyMetadataKeyID;
+  }
+
+  @Override
+  public String location() {
+    return location;
+  }
+
+  @Override
+  public long snapshotId() {
+    return snapshotId;
+  }
+
+  @Override
+  public String keyMetadataKeyId() {
+    return keyMetadataKeyID;
+  }
+
+  @Override
+  public ByteBuffer encryptedKeyMetadata() {
+    return ByteBuffer.wrap(encryptedKeyMetadata);
+  }
+
+  @Override
+  public ByteBuffer decryptKeyMetadata(EncryptionManager em) {
+    return EncryptionUtil.decryptSnapshotKeyMetadata(this, em);
+  }
+}

core/src/main/java/org/apache/iceberg/BaseSnapshot.java

@@ -26,6 +26,7 @@
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Objects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -38,11 +39,11 @@ class BaseSnapshot implements Snapshot {
   private final Long parentId;
   private final long sequenceNumber;
   private final long timestampMillis;
-  private final String manifestListLocation;
   private final String operation;
   private final Map<String, String> summary;
   private final Integer schemaId;
   private final String[] v1ManifestLocations;
+  private final ManifestListFile manifestListFile;
 
   // lazily initialized
   private transient List<ManifestFile> allManifests = null;
@@ -53,6 +54,7 @@ class BaseSnapshot implements Snapshot {
   private transient List<DeleteFile> addedDeleteFiles = null;
   private transient List<DeleteFile> removedDeleteFiles = null;
 
+  @VisibleForTesting
   BaseSnapshot(
       long sequenceNumber,
       long snapshotId,
@@ -62,14 +64,34 @@ class BaseSnapshot implements Snapshot {
       Map<String, String> summary,
       Integer schemaId,
       String manifestList) {
+    this(
+        sequenceNumber,
+        snapshotId,
+        parentId,
+        timestampMillis,
+        operation,
+        summary,
+        schemaId,
+        new BaseManifestListFile(manifestList, snapshotId, null, null));
+  }
+
+  BaseSnapshot(
+      long sequenceNumber,
+      long snapshotId,
+      Long parentId,
+      long timestampMillis,
+      String operation,
+      Map<String, String> summary,
+      Integer schemaId,
+      ManifestListFile manifestListFile) {
     this.sequenceNumber = sequenceNumber;
     this.snapshotId = snapshotId;
     this.parentId = parentId;
     this.timestampMillis = timestampMillis;
     this.operation = operation;
     this.summary = summary;
     this.schemaId = schemaId;
-    this.manifestListLocation = manifestList;
+    this.manifestListFile = manifestListFile;
     this.v1ManifestLocations = null;
   }
 
@@ -89,7 +111,7 @@ class BaseSnapshot implements Snapshot {
     this.operation = operation;
     this.summary = summary;
     this.schemaId = schemaId;
-    this.manifestListLocation = null;
+    this.manifestListFile = new BaseManifestListFile(null, snapshotId, null, null);
     this.v1ManifestLocations = v1ManifestLocations;
   }
 
@@ -128,6 +150,11 @@ public Integer schemaId() {
     return schemaId;
   }
 
+  @Override
+  public ManifestListFile manifestListFile() {
+    return manifestListFile;
+  }
+
   private void cacheManifests(FileIO fileIO) {
     if (fileIO == null) {
       throw new IllegalArgumentException("Cannot cache changes: FileIO is null");
@@ -143,7 +170,7 @@ private void cacheManifests(FileIO fileIO) {
 
     if (allManifests == null) {
       // if manifests isn't set, then the snapshotFile is set and should be read to get the list
-      this.allManifests = ManifestLists.read(fileIO.newInputFile(manifestListLocation));
+      this.allManifests = ManifestLists.read(fileIO.newInputFile(manifestListFile));
     }
 
     if (dataManifests == null || deleteManifests == null) {
@@ -216,7 +243,7 @@ public Iterable<DeleteFile> removedDeleteFiles(FileIO fileIO) {
 
   @Override
   public String manifestListLocation() {
-    return manifestListLocation;
+    return manifestListFile.location();
   }
 
   private void cacheDeleteFileChanges(FileIO fileIO) {
@@ -317,7 +344,7 @@ public String toString() {
         .add("timestamp_ms", timestampMillis)
         .add("operation", operation)
         .add("summary", summary)
-        .add("manifest-list", manifestListLocation)
+        .add("manifest-list", manifestListFile.location())
         .add("schema-id", schemaId)
         .toString();
   }

core/src/main/java/org/apache/iceberg/CatalogProperties.java

@@ -159,4 +159,10 @@ private CatalogProperties() {}
 
   public static final String ENCRYPTION_KMS_TYPE = "encryption.kms-type";
   public static final String ENCRYPTION_KMS_IMPL = "encryption.kms-impl";
+  public static final String WRITER_KEK_TIMEOUT_SEC = "encryption.kek-timeout-sec";
+
+  /**
+   * Default time-out of key encryption keys. Per NIST SP 800-57 P1 R5 section 5.3.6, set to 1 week.
+   */
+  public static final long WRITER_KEK_TIMEOUT_SEC_DEFAULT = TimeUnit.DAYS.toSeconds(7);
 }