GH-3198: Allow specifying trusted classes by class name (#3199)

Author: gszadovszky

parquet-avro/README.md

@@ -32,7 +32,8 @@ Apache Avro integration
 | `parquet.avro.read.schema`              | `String`  | The Avro schema to be used for reading. It shall be compatible with the file schema. The file schema will be used directly if not set. |
 | `parquet.avro.projection`               | `String`  | The Avro schema to be used for projection.                           |
 | `parquet.avro.compatible`               | `boolean` | Flag for compatibility mode. `true` for materializing Avro `IndexedRecord` objects, `false` for materializing the related objects for either generic, specific, or reflect records.<br/>The default value is `true`. |
-| `parquet.avro.readInt96AsFixed`        | `boolean` | Flag for handling the `INT96` Parquet types. `true` for converting it to the `fixed` Avro type, `false` for not handling `INT96` types (throwing exception).<br/>The default value is `false`.<br/>**NOTE: The `INT96` Parquet type is deprecated. This option is only to support old data.** |
+| `parquet.avro.readInt96AsFixed`         | `boolean` | Flag for handling the `INT96` Parquet types. `true` for converting it to the `fixed` Avro type, `false` for not handling `INT96` types (throwing exception).<br/>The default value is `false`.<br/>**NOTE: The `INT96` Parquet type is deprecated. This option is only to support old data.** |
+| `parquet.avro.serializable.classes`     | `String`  | List of the fully qualified class names separated by ',' that may be referenced from the Avro schema by "java-class" or "java-key-class" and are allowed to be loaded. |
 
 ### Configuration for writing
 

parquet-avro/src/main/java/org/apache/parquet/avro/AvroConverters.java

@@ -21,8 +21,6 @@
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.List;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.util.Utf8;
@@ -36,13 +34,19 @@
 
 public class AvroConverters {
 
+  /**
+   * Contains the packages which classes are allowed to be loaded that may be referenced from the Avro schema by
+   * "java-class" or "java-key-class". It contains the packages parsed from system variable
+   * "org.apache.parquet.avro.SERIALIZABLE_PACKAGES".
+   *
+   * @deprecated will be removed in 2.0.0
+   */
+  @Deprecated
   public static final String[] SERIALIZABLE_PACKAGES;
 
   static {
-    SERIALIZABLE_PACKAGES = System.getProperty(
-            "org.apache.parquet.avro.SERIALIZABLE_PACKAGES",
-            "java.lang,java.math,java.io,java.net,org.apache.parquet.avro")
-        .split(",");
+    String prop = System.getProperty("org.apache.parquet.avro.SERIALIZABLE_PACKAGES");
+    SERIALIZABLE_PACKAGES = prop == null ? new String[0] : prop.split(",");
   }
 
   public abstract static class AvroGroupConverter extends GroupConverter {
@@ -272,7 +276,6 @@ static final class FieldStringableConverter extends BinaryConverter<Object> {
 
     public FieldStringableConverter(ParentValueContainer parent, Class<?> stringableClass) {
       super(parent);
-      checkSecurity(stringableClass);
       stringableName = stringableClass.getName();
       try {
         this.ctor = stringableClass.getConstructor(String.class);
@@ -289,33 +292,6 @@ public Object convert(Binary binary) {
         throw new ParquetDecodingException("Cannot convert binary to " + stringableName, e);
       }
     }
-
-    private void checkSecurity(Class<?> clazz) throws SecurityException {
-      List<String> trustedPackages = Arrays.asList(SERIALIZABLE_PACKAGES);
-
-      boolean trustAllPackages = trustedPackages.size() == 1 && "*".equals(trustedPackages.get(0));
-      if (trustAllPackages || clazz.isPrimitive()) {
-        return;
-      }
-
-      boolean found = false;
-      Package thePackage = clazz.getPackage();
-      if (thePackage != null) {
-        for (String trustedPackage : trustedPackages) {
-          if (thePackage.getName().equals(trustedPackage)
-              || thePackage.getName().startsWith(trustedPackage + ".")) {
-            found = true;
-            break;
-          }
-        }
-        if (!found) {
-          throw new SecurityException("Forbidden " + clazz
-              + "! This class is not trusted to be included in Avro schema using java-class."
-              + " Please set org.apache.parquet.avro.SERIALIZABLE_PACKAGES system property"
-              + " with the packages you trust.");
-        }
-      }
-    }
   }
 
   static final class FieldEnumConverter extends BinaryConverter<Object> {

parquet-avro/src/main/java/org/apache/parquet/avro/AvroParquetReader.java

@@ -19,6 +19,8 @@
 package org.apache.parquet.avro;
 
 import java.io.IOException;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.specific.SpecificData;
@@ -179,6 +181,26 @@ public Builder<T> withDataModel(GenericData model) {
       return this;
     }
 
+    public Builder<T> withSerializableClasses(String... classNames) {
+      if (classNames.length == 0) {
+        configuration.set(AvroReadSupport.SERIALIZABLE_CLASSES, null);
+      } else {
+        configuration.set(AvroReadSupport.SERIALIZABLE_CLASSES, String.join(",", classNames));
+      }
+      return this;
+    }
+
+    public Builder<T> withSerializableClasses(Class<?>... classes) {
+      if (classes.length == 0) {
+        configuration.set(AvroReadSupport.SERIALIZABLE_CLASSES, null);
+      } else {
+        configuration.set(
+            AvroReadSupport.SERIALIZABLE_CLASSES,
+            Stream.of(classes).map(Class::getName).collect(Collectors.joining(",")));
+      }
+      return this;
+    }
+
     public Builder<T> disableCompatibility() {
       this.enableCompatibility = false;
       return this;

parquet-avro/src/main/java/org/apache/parquet/avro/AvroReadSupport.java

@@ -20,6 +20,8 @@
 
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
 import org.apache.hadoop.conf.Configuration;
@@ -61,6 +63,12 @@ public class AvroReadSupport<T> extends ReadSupport<T> {
   public static final String READ_INT96_AS_FIXED = "parquet.avro.readInt96AsFixed";
   public static final boolean READ_INT96_AS_FIXED_DEFAULT = false;
 
+  /**
+   * List of the fully qualified class names separated by ',' that may be referenced from the Avro schema by
+   * "java-class" or "java-key-class" and are allowed to be loaded.
+   */
+  public static final String SERIALIZABLE_CLASSES = "parquet.avro.serializable.classes";
+
   /**
    * @param configuration       a configuration
    * @param requestedProjection the requested projection schema
@@ -83,6 +91,24 @@ public static void setAvroDataSupplier(Configuration configuration, Class<? exte
     configuration.set(AVRO_DATA_SUPPLIER, clazz.getName());
   }
 
+  public static void setSerializableClasses(Configuration configuration, String... classNames) {
+    if (classNames.length == 0) {
+      configuration.set(AvroReadSupport.SERIALIZABLE_CLASSES, null);
+    } else {
+      configuration.set(AvroReadSupport.SERIALIZABLE_CLASSES, String.join(",", classNames));
+    }
+  }
+
+  public static void setSerializableClasses(Configuration configuration, Class<?>... classes) {
+    if (classes.length == 0) {
+      configuration.set(AvroReadSupport.SERIALIZABLE_CLASSES, null);
+    } else {
+      configuration.set(
+          AvroReadSupport.SERIALIZABLE_CLASSES,
+          Stream.of(classes).map(Class::getName).collect(Collectors.joining(",")));
+    }
+  }
+
   private GenericData model = null;
 
   public AvroReadSupport() {}
@@ -158,7 +184,15 @@ public RecordMaterializer<T> prepareForRead(
     if (Boolean.parseBoolean(compatEnabled)) {
       return newCompatMaterializer(parquetSchema, avroSchema, model);
     }
-    return new AvroRecordMaterializer<T>(parquetSchema, avroSchema, model);
+    String[] serializableClasses = configuration.getStrings(SERIALIZABLE_CLASSES, null);
+
+    return new AvroRecordMaterializer<T>(
+        parquetSchema,
+        avroSchema,
+        model,
+        serializableClasses == null
+            ? new ReflectClassValidator.PackageValidator()
+            : new ReflectClassValidator.ClassValidator(serializableClasses));
   }
 
   @SuppressWarnings("unchecked")