GH-3168: Restrict trusted packages in the parquet-avro module

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

@@ -21,6 +21,8 @@
 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;
@@ -34,6 +36,15 @@
 
 public class AvroConverters {
 
+  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(",");
+  }
+
   public abstract static class AvroGroupConverter extends GroupConverter {
     protected final ParentValueContainer parent;
 
@@ -261,6 +272,7 @@ 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);
@@ -277,6 +289,33 @@ 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> {