Encryption for REST catalog

.palantir/revapi.yml

@@ -1177,6 +1177,10 @@ acceptedBreaks:
       old: "class org.apache.iceberg.Metrics"
       new: "class org.apache.iceberg.Metrics"
       justification: "Java serialization across versions is not guaranteed"
+    - code: "java.class.defaultSerializationChanged"
+      old: "class org.apache.iceberg.encryption.EncryptingFileIO"
+      new: "class org.apache.iceberg.encryption.EncryptingFileIO"
+      justification: "New method for Manifest List reading"
     org.apache.iceberg:iceberg-core:
     - code: "java.method.removed"
       old: "method java.lang.String[] org.apache.iceberg.hadoop.Util::blockLocations(org.apache.iceberg.CombinedScanTask,\

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

@@ -0,0 +1,34 @@
+/*
+ * 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();
+
+  /** The manifest list key metadata can be encrypted. Returns ID of encryption key */
+  String encryptionKeyID();
+
+  /** Decrypt and return the manifest list key metadata */
+  ByteBuffer decryptKeyMetadata(EncryptionManager em);
+}

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;
@@ -108,13 +109,21 @@ public InputFile newInputFile(ManifestFile manifest) {
     }
   }
 
+  @Override
+  public InputFile newInputFile(ManifestListFile manifestList) {
+    if (manifestList.encryptionKeyID() != 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.encryptionKeyID() == 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,49 @@
+/*
+ * 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;
+
+class BaseManifestListFile implements ManifestListFile, Serializable {
+  private final String location;
+  private final String encryptionKeyID;
+
+  BaseManifestListFile(String location, String encryptionKeyID) {
+    this.location = location;
+    this.encryptionKeyID = encryptionKeyID;
+  }
+
+  @Override
+  public String location() {
+    return location;
+  }
+
+  @Override
+  public String encryptionKeyID() {
+    return encryptionKeyID;
+  }
+
+  @Override
+  public ByteBuffer decryptKeyMetadata(EncryptionManager em) {
+    return EncryptionUtil.decryptManifestListKeyMetadata(this, em);
+  }
+}

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

@@ -182,7 +182,9 @@ 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(new BaseManifestListFile(manifestListLocation, keyId)));
     }
 
     if (dataManifests == null || deleteManifests == null) {

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

@@ -21,6 +21,10 @@
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.Map;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.encryption.NativeEncryptionKeyMetadata;
+import org.apache.iceberg.encryption.NativeEncryptionOutputFile;
+import org.apache.iceberg.encryption.StandardEncryptionManager;
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.io.OutputFile;
@@ -29,9 +33,25 @@
 
 abstract class ManifestListWriter implements FileAppender<ManifestFile> {
   private final FileAppender<ManifestFile> writer;
+  private final StandardEncryptionManager standardEncryptionManager;
+  private final NativeEncryptionKeyMetadata manifestListKeyMetadata;
+  private final OutputFile outputFile;
+
+  private ManifestListWriter(
+      OutputFile file, EncryptionManager encryptionManager, Map<String, String> meta) {
+    if (encryptionManager instanceof StandardEncryptionManager) {
+      // ability to encrypt the manifest list key is introduced for standard encryption.
+      this.standardEncryptionManager = (StandardEncryptionManager) encryptionManager;
+      NativeEncryptionOutputFile encryptedFile = this.standardEncryptionManager.encrypt(file);
+      this.outputFile = encryptedFile.encryptingOutputFile();
+      this.manifestListKeyMetadata = encryptedFile.keyMetadata();
+    } else {
+      this.standardEncryptionManager = null;
+      this.outputFile = file;
+      this.manifestListKeyMetadata = null;
+    }
 
-  private ManifestListWriter(OutputFile file, Map<String, String> meta) {
-    this.writer = newAppender(file, meta);
+    this.writer = newAppender(outputFile, meta);
   }
 
   protected abstract ManifestFile prepare(ManifestFile manifest);
@@ -73,18 +93,31 @@ public Long nextRowId() {
     return null;
   }
 
+  public ManifestListFile toManifestListFile() {
+    if (manifestListKeyMetadata != null && manifestListKeyMetadata.encryptionKey() != null) {
+      manifestListKeyMetadata.copyWithLength(writer.length());
+      String manifestListKeyID =
+          standardEncryptionManager.addManifestListKeyMetadata(manifestListKeyMetadata);
+      return new BaseManifestListFile(outputFile.location(), manifestListKeyID);
+    } else {
+      return new BaseManifestListFile(outputFile.location(), null);
+    }
+  }
+
   static class V3Writer extends ManifestListWriter {
     private final V3Metadata.ManifestFileWrapper wrapper;
     private Long nextRowId;
 
     V3Writer(
         OutputFile snapshotFile,
+        EncryptionManager encryptionManager,
         long snapshotId,
         Long parentSnapshotId,
         long sequenceNumber,
         long firstRowId) {
       super(
           snapshotFile,
+          encryptionManager,
           ImmutableMap.of(
               "snapshot-id", String.valueOf(snapshotId),
               "parent-snapshot-id", String.valueOf(parentSnapshotId),
@@ -134,15 +167,22 @@ public Long nextRowId() {
   static class V2Writer extends ManifestListWriter {
     private final V2Metadata.ManifestFileWrapper wrapper;
 
-    V2Writer(OutputFile snapshotFile, long snapshotId, Long parentSnapshotId, long sequenceNumber) {
+    V2Writer(
+        OutputFile snapshotFile,
+        EncryptionManager encryptionManager,
+        long snapshotId,
+        Long parentSnapshotId,
+        long sequenceNumber) {
       super(
           snapshotFile,
+          encryptionManager,
           ImmutableMap.of(
               "snapshot-id", String.valueOf(snapshotId),
               "parent-snapshot-id", String.valueOf(parentSnapshotId),
               "sequence-number", String.valueOf(sequenceNumber),
               "format-version", "2"));
       this.wrapper = new V2Metadata.ManifestFileWrapper(snapshotId, sequenceNumber);
+      // todo encryption only in v3? throw exception if e.manager is not plaintext?
     }
 
     @Override
@@ -170,13 +210,19 @@ protected FileAppender<ManifestFile> newAppender(OutputFile file, Map<String, St
   static class V1Writer extends ManifestListWriter {
     private final V1Metadata.ManifestFileWrapper wrapper = new V1Metadata.ManifestFileWrapper();
 
-    V1Writer(OutputFile snapshotFile, long snapshotId, Long parentSnapshotId) {
+    V1Writer(
+        OutputFile snapshotFile,
+        EncryptionManager encryptionManager,
+        long snapshotId,
+        Long parentSnapshotId) {
       super(
           snapshotFile,
+          encryptionManager,
           ImmutableMap.of(
               "snapshot-id", String.valueOf(snapshotId),
               "parent-snapshot-id", String.valueOf(parentSnapshotId),
               "format-version", "1"));
+      // todo encryption only in v3? throw exception if e.manager is not plaintext?
     }
 
     @Override

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

@@ -20,10 +20,13 @@
 
 import java.io.IOException;
 import java.util.List;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.encryption.PlaintextEncryptionManager;
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
@@ -47,26 +50,52 @@ static List<ManifestFile> read(InputFile manifestList) {
     }
   }
 
+  // or should we modify all related tests (to pass PlaintextEncryptionManager)?
+  @VisibleForTesting
   static ManifestListWriter write(
       int formatVersion,
       OutputFile manifestListFile,
       long snapshotId,
       Long parentSnapshotId,
       long sequenceNumber,
       Long firstRowId) {
+    return write(
+        formatVersion,
+        manifestListFile,
+        PlaintextEncryptionManager.instance(),
+        snapshotId,
+        parentSnapshotId,
+        sequenceNumber,
+        firstRowId);
+  }
+
+  static ManifestListWriter write(
+      int formatVersion,
+      OutputFile manifestListFile,
+      EncryptionManager encryptionManager,
+      long snapshotId,
+      Long parentSnapshotId,
+      long sequenceNumber,
+      Long firstRowId) {
     switch (formatVersion) {
       case 1:
         Preconditions.checkArgument(
             sequenceNumber == TableMetadata.INITIAL_SEQUENCE_NUMBER,
             "Invalid sequence number for v1 manifest list: %s",
             sequenceNumber);
-        return new ManifestListWriter.V1Writer(manifestListFile, snapshotId, parentSnapshotId);
+        return new ManifestListWriter.V1Writer(
+            manifestListFile, encryptionManager, snapshotId, parentSnapshotId);
       case 2:
         return new ManifestListWriter.V2Writer(
-            manifestListFile, snapshotId, parentSnapshotId, sequenceNumber);
+            manifestListFile, encryptionManager, snapshotId, parentSnapshotId, sequenceNumber);
       case 3:
         return new ManifestListWriter.V3Writer(
-            manifestListFile, snapshotId, parentSnapshotId, sequenceNumber, firstRowId);
+            manifestListFile,
+            encryptionManager,
+            snapshotId,
+            parentSnapshotId,
+            sequenceNumber,
+            firstRowId);
     }
     throw new UnsupportedOperationException(
         "Cannot write manifest list for table version: " + formatVersion);

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

@@ -21,6 +21,8 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.encryption.NativeEncryptionKeyMetadata;
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.io.OutputFile;
@@ -38,7 +40,7 @@ public abstract class ManifestWriter<F extends ContentFile<F>> implements FileAp
   static final long UNASSIGNED_SEQ = -1L;
 
   private final OutputFile file;
-  private final ByteBuffer keyMetadataBuffer;
+  private final EncryptionKeyMetadata keyMetadata;
   private final int specId;
   private final FileAppender<ManifestEntry<F>> writer;
   private final Long snapshotId;
@@ -65,7 +67,7 @@ private ManifestWriter(
         new GenericManifestEntry<>(V1Metadata.entrySchema(spec.partitionType()).asStruct());
     this.stats = new PartitionSummary(spec);
     this.firstRowId = firstRowId;
-    this.keyMetadataBuffer = (file.keyMetadata() == null) ? null : file.keyMetadata().buffer();
+    this.keyMetadata = file.keyMetadata();
   }
 
   protected abstract ManifestEntry<F> prepare(ManifestEntry<F> entry);
@@ -192,6 +194,18 @@ public long length() {
 
   public ManifestFile toManifestFile() {
     Preconditions.checkState(closed, "Cannot build ManifestFile, writer is not closed");
+
+    // if key metadata can store the length, add it
+    ByteBuffer keyMetadataBuffer;
+    if (keyMetadata instanceof NativeEncryptionKeyMetadata) {
+      keyMetadataBuffer =
+          ((NativeEncryptionKeyMetadata) keyMetadata).copyWithLength(length()).buffer();
+    } else if (keyMetadata != null) {
+      keyMetadataBuffer = keyMetadata.buffer();
+    } else {
+      keyMetadataBuffer = null;
+    }
+
     // if the minSequenceNumber is null, then no manifests with a sequence number have been written,
     // so the min data sequence number is the one that will be assigned when this is committed.
     // pass UNASSIGNED_SEQ to inherit it.