[Java] Add class checker API (#890)

* add class check API

* add white/black list based class checker

* refine class checker

* add missing header

* refine documentation

* refine documentation

* add deserialize tests

* refine security warning message

README.md

@@ -355,6 +355,9 @@ For example, the deserialization may invoke `init` constructor or `equals`/`hash
 Fury provides a class registration option and enabled by default for such protocols, which allows only deserializing trusted registered types or built-in types.
  **Do not disable class registration unless you can ensure your environment is indeed secure**.
 
+If this option is disabled, you are responsible for serialization security. You can configure `io.fury.resolver.ClassChecker` by
+`ClassResolver#setClassChecker` to control which classes are allowed for serialization.
+
 ## RoadMap
 - Meta compression, auto meta sharing and cross-language schema compatibility.
 - AOT Framework for c++/golang/rust to generate code statically.

docs/guide/java_object_graph_guide.md

@@ -160,10 +160,31 @@ should have same registration order.
 
 ```java
 Fury fury=xxx;
-  fury.register(SomeClass.class);
-  fury.register(SomeClass1.class,200);
+fury.register(SomeClass.class);
+fury.register(SomeClass1.class, 200);
 ```
 
+If you invoke `FuryBuilder#requireClassRegistration(false)` to disable class registration check,
+you can set `io.fury.resolver.ClassChecker` by `ClassResolver#setClassChecker` to control which classes are allowed 
+for serialization. For example,you can allow classes started with `org.example.*` by:
+```java
+Fury fury = xxx;
+fury.getClassResolver().setClassChecker((classResolver, className) -> className.startsWith("org.example."));
+```
+```java
+AllowListChecker checker = new AllowListChecker(AllowListChecker.CheckLevel.STRICT);
+ThreadSafeFury fury = new ThreadLocalFury(classLoader -> {
+  Fury f = Fury.builder().requireClassRegistration(false).withClassLoader(classLoader).build();
+  f.getClassResolver().setClassChecker(checker);
+  checker.addListener(f.getClassResolver());
+  return f;
+});
+checker.allowClass("org.example.*");
+```
+
+Fury also provided a `io.fury.resolver.AllowListChecker` which is white/blacklist based checker to simplify 
+the customization of class check mechanism. You can use this checker or implement more sophisticated checker by yourself.
+
 ### Serializer Registration
 
 You can also register a custom serializer for a class by `Fury#registerSerializer` API.

java/fury-core/src/main/java/io/fury/Fury.java

@@ -1405,10 +1405,12 @@ public FuryBuilder registerGuavaTypes(boolean register) {
 
     /**
      * Whether to require registering classes for serialization, enabled by default. If disabled,
-     * unknown insecure classes can be deserialized, which can be insecure and cause remote code
-     * execution attack if the classes `constructor`/`equals`/`hashCode` method contain malicious
-     * code. Do not disable class registration if you can't ensure your environment are *indeed
-     * secure*. We are not responsible for security risks if you disable this option.
+     * unknown classes can be deserialized, which may be insecure and cause remote code execution
+     * attack if the classes `constructor`/`equals`/`hashCode` method contain malicious code. Do not
+     * disable class registration if you can't ensure your environment are *indeed secure*. We are
+     * not responsible for security risks if you disable this option. If you disable this option,
+     * you can configure {@link io.fury.resolver.ClassChecker} by {@link
+     * ClassResolver#setClassChecker} to control which classes are allowed being serialized.
      */
     public FuryBuilder requireClassRegistration(boolean requireClassRegistration) {
       this.requireClassRegistration = requireClassRegistration;
@@ -1478,9 +1480,10 @@ private void finish() {
       }
       if (!requireClassRegistration) {
         LOG.warn(
-            "Class registration isn't forced, unknown insecure classes can be deserialized. "
+            "Class registration isn't forced, unknown classes can be deserialized. "
                 + "If the environment isn't 100% secure, please enable class registration by "
-                + "`FuryBuilder#requireClassRegistration(true)`.");
+                + "`FuryBuilder#requireClassRegistration(true)` or configure ClassChecker by "
+                + "`ClassResolver.setClassChecker`");
       }
     }
 

java/fury-core/src/main/java/io/fury/resolver/AllowListChecker.java

@@ -0,0 +1,210 @@
+/*
+ * Copyright 2023 The Fury Authors
+ *
+ * Licensed 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 io.fury.resolver;
+
+import io.fury.Fury;
+import io.fury.exception.InsecureException;
+import io.fury.memory.MemoryBuffer;
+import io.fury.serializer.Serializer;
+import io.fury.util.LoggerFactory;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.WeakHashMap;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import javax.annotation.concurrent.ThreadSafe;
+import org.slf4j.Logger;
+
+/**
+ * White/black list based class checker.
+ *
+ * @author chaokunyang
+ */
+@ThreadSafe
+public class AllowListChecker implements ClassChecker {
+  private static final Logger LOG = LoggerFactory.getLogger(AllowListChecker.class);
+
+  public enum CheckLevel {
+    /** Disable serialize check for all classes. */
+    DISABLE,
+
+    /** Only deny danger classes, warn if other classes are not in allow list. */
+    WARN,
+
+    /** Only allow classes in allow list, deny if other classes are not in allow list. */
+    STRICT
+  }
+
+  private final CheckLevel checkLevel;
+  private final Set<String> allowList;
+  private final Set<String> allowListPrefix;
+  private final Set<String> disallowList;
+  private final Set<String> disallowListPrefix;
+  private final transient WeakHashMap<ClassResolver, Boolean> listeners;
+  private final transient ReadWriteLock lock;
+
+  public AllowListChecker() {
+    this(CheckLevel.WARN);
+  }
+
+  public AllowListChecker(CheckLevel checkLevel) {
+    this.checkLevel = checkLevel;
+    allowList = new HashSet<>();
+    allowListPrefix = new HashSet<>();
+    disallowList = new HashSet<>();
+    disallowListPrefix = new HashSet<>();
+    lock = new ReentrantReadWriteLock();
+    listeners = new WeakHashMap<>();
+  }
+
+  @Override
+  public boolean checkClass(ClassResolver classResolver, String className) {
+    switch (checkLevel) {
+      case DISABLE:
+        return true;
+      case WARN:
+        if (containsPrefix(disallowList, disallowListPrefix, className)) {
+          throw new InsecureException(
+              String.format("Class %s is forbidden for serialization.", className));
+        }
+        if (!containsPrefix(allowList, allowListPrefix, className)) {
+          LOG.warn(
+              "Class {} not in allow list, please check whether objects of this class "
+                  + "are allowed for serialization.",
+              className);
+        }
+        return true;
+      case STRICT:
+        if (containsPrefix(disallowList, disallowListPrefix, className)) {
+          throw new InsecureException(
+              String.format("Class %s is forbidden for serialization.", className));
+        }
+        if (!containsPrefix(allowList, allowListPrefix, className)) {
+          throw new InsecureException(
+              String.format(
+                  "Class %s isn't in allow list for serialization. If this class is allowed for "
+                      + "serialization, please add it to allow list by AllowListChecker#addAllowClass",
+                  className));
+        }
+        return true;
+      default:
+        throw new UnsupportedOperationException("Unsupported check level " + checkLevel);
+    }
+  }
+
+  boolean containsPrefix(Set<String> set, Set<String> prefixSet, String className) {
+    try {
+      lock.readLock().lock();
+      if (set.contains(className)) {
+        return true;
+      }
+      for (String prefix : prefixSet) {
+        if (className.startsWith(prefix)) {
+          return true;
+        }
+      }
+      return false;
+    } finally {
+      lock.readLock().unlock();
+    }
+  }
+
+  /**
+   * Add class to allow list.
+   *
+   * @param classNameOrPrefix class name or class name prefix ends with *.
+   */
+  public void allowClass(String classNameOrPrefix) {
+    try {
+      lock.writeLock().lock();
+      if (classNameOrPrefix.endsWith("*")) {
+        allowListPrefix.add(classNameOrPrefix.substring(0, classNameOrPrefix.length() - 1));
+      } else {
+        allowList.add(classNameOrPrefix);
+      }
+    } finally {
+      lock.writeLock().unlock();
+    }
+  }
+
+  /**
+   * Add class to disallow list.
+   *
+   * @param classNameOrPrefix class name or class name prefix ends with *.
+   */
+  public void disallowClass(String classNameOrPrefix) {
+    try {
+      lock.writeLock().lock();
+      if (classNameOrPrefix.endsWith("*")) {
+        String prefix = classNameOrPrefix.substring(0, classNameOrPrefix.length() - 1);
+        disallowListPrefix.add(prefix);
+        for (ClassResolver classResolver : listeners.keySet()) {
+          try {
+            classResolver.getFury().getJITContext().lock();
+            // clear serializer may throw NullPointerException for field serialization.
+            classResolver.setSerializers(prefix, DisallowSerializer.class);
+          } finally {
+            classResolver.getFury().getJITContext().unlock();
+          }
+        }
+      } else {
+        disallowList.add(classNameOrPrefix);
+        for (ClassResolver classResolver : listeners.keySet()) {
+          try {
+            classResolver.getFury().getJITContext().lock();
+            // clear serializer may throw NullPointerException for field serialization.
+            classResolver.setSerializer(classNameOrPrefix, DisallowSerializer.class);
+          } finally {
+            classResolver.getFury().getJITContext().unlock();
+          }
+        }
+      }
+    } finally {
+      lock.writeLock().unlock();
+    }
+  }
+
+  /**
+   * Add listener to in response to disallow list. So if object of a class is serialized before,
+   * future serialization will be refused.
+   */
+  public void addListener(ClassResolver classResolver) {
+    try {
+      lock.writeLock().lock();
+    } finally {
+      listeners.put(classResolver, true);
+    }
+  }
+
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  private static class DisallowSerializer extends Serializer {
+
+    public DisallowSerializer(Fury fury, Class type) {
+      super(fury, type);
+    }
+
+    @Override
+    public void write(MemoryBuffer buffer, Object value) {
+      throw new InsecureException(String.format("Class %s not allowed for serialization.", type));
+    }
+
+    @Override
+    public Object read(MemoryBuffer buffer) {
+      throw new InsecureException(String.format("Class %s not allowed for serialization.", type));
+    }
+  }
+}

java/fury-core/src/main/java/io/fury/resolver/ClassChecker.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2023 The Fury Authors
+ *
+ * Licensed 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 io.fury.resolver;
+
+/**
+ * Check whether class or objects of class should be serialized. If class checker will be invoked by
+ * multiple {@link ClassResolver}, class checker should be thread safe.
+ *
+ * @author chaokunyang
+ */
+public interface ClassChecker {
+  /**
+   * Check whether class should be allowed for serialization.
+   *
+   * @param classResolver class resolver
+   * @param className full name of class
+   * @return true if class is allowed for serialization.
+   */
+  boolean checkClass(ClassResolver classResolver, String className);
+}