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> {

parquet-avro/src/test/java/org/apache/parquet/UntrustedStringableClass.java

@@ -0,0 +1,32 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.parquet;
+
+public class UntrustedStringableClass {
+  private final String value;
+
+  public UntrustedStringableClass(String value) {
+    this.value = value;
+  }
+
+  @Override
+  public String toString() {
+    return this.value;
+  }
+}

parquet-avro/src/test/java/org/apache/parquet/avro/TestReflectReadWrite.java

@@ -37,6 +37,7 @@
 import org.apache.avro.util.Utf8;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
+import org.apache.parquet.UntrustedStringableClass;
 import org.apache.parquet.hadoop.ParquetReader;
 import org.apache.parquet.hadoop.ParquetWriter;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
@@ -76,6 +77,40 @@ public void testWriteReflectReadGeneric() throws IOException {
     }
   }
 
+  @Test(expected = SecurityException.class)
+  public void testUntrustedStringableClass() {
+    new AvroConverters.FieldStringableConverter(
+        new ParentValueContainer() {
+          @Override
+          public void add(Object value) {}
+
+          @Override
+          public void addBoolean(boolean value) {}
+
+          @Override
+          public void addInt(int value) {}
+
+          @Override
+          public void addLong(long value) {}
+
+          @Override
+          public void addFloat(float value) {}
+
+          @Override
+          public void addDouble(double value) {}
+
+          @Override
+          public void addChar(char value) {}
+
+          @Override
+          public void addByte(byte value) {}
+
+          @Override
+          public void addShort(short value) {}
+        },
+        UntrustedStringableClass.class);
+  }
+
   private GenericRecord getGenericPojoUtf8() {
     Schema schema = ReflectData.get().getSchema(Pojo.class);
     GenericData.Record record = new GenericData.Record(schema);