(kafka) branch trunk updated: KAFKA-20450 Allowlist based SafeObjectInputStream (#22056)

[chia7712]

This is an automated email from the ASF dual-hosted git repository.

chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new eef6cab6481 KAFKA-20450 Allowlist based SafeObjectInputStream (#22056)
eef6cab6481 is described below

commit eef6cab6481e14d6567d66c9705985394d1ba8ea
Author: subbudvk <115633743+subbu...@users.noreply.github.com>
AuthorDate: Tue Apr 28 11:33:00 2026 +0530

    KAFKA-20450 Allowlist based SafeObjectInputStream (#22056)
    
    The current **SafeObjectInputStream** uses a denylist based approach -
    having a fixed denylist to be validated against for deserialization.
    This is a bad security practise and has also been advised so in the
    original PR.
    
    Making this as a allowlist instead and allowing safe BASE_TYPES which
    are required by current caller
    (_org.apache.kafka.connect.storage.FileOffsetBackingStore_)
    
    Also providing a `SafeObjectInputStream(InputStream in, Set<String>
    allowedClasses)` so if any consumer require any specific allowedClasses
    they can pass in here.
    
    Reviewers: Mickael Maison <mickael.mai...@gmail.com>, Chia-Ping Tsai
     <chia7...@gmail.com>
---
 .../kafka/connect/util/SafeObjectInputStream.java  | 54 ++++++++++++----------
 1 file changed, 30 insertions(+), 24 deletions(-)

diff --git 
a/connect/runtime/src/main/java/org/apache/kafka/connect/util/SafeObjectInputStream.java
 
b/connect/runtime/src/main/java/org/apache/kafka/connect/util/SafeObjectInputStream.java
index bd0279fb3d9..c020de04f58 100644
--- 
a/connect/runtime/src/main/java/org/apache/kafka/connect/util/SafeObjectInputStream.java
+++ 
b/connect/runtime/src/main/java/org/apache/kafka/connect/util/SafeObjectInputStream.java
@@ -18,8 +18,10 @@ package org.apache.kafka.connect.util;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InvalidClassException;
 import java.io.ObjectInputStream;
 import java.io.ObjectStreamClass;
+import java.util.Objects;
 import java.util.Set;
 
 
@@ -30,40 +32,44 @@ import java.util.Set;
  */
 public class SafeObjectInputStream extends ObjectInputStream {
 
-    protected static final Set<String> DEFAULT_NO_DESERIALIZE_CLASS_NAMES = 
Set.of(
-            "org.apache.commons.collections.functors.InvokerTransformer",
-            "org.apache.commons.collections.functors.InstantiateTransformer",
-            "org.apache.commons.collections4.functors.InvokerTransformer",
-            "org.apache.commons.collections4.functors.InstantiateTransformer",
-            "org.codehaus.groovy.runtime.ConvertedClosure",
-            "org.codehaus.groovy.runtime.MethodClosure",
-            "org.springframework.beans.factory.ObjectFactory",
-            "com.sun.org.apache.xalan.internal.xsltc.trax.TemplatesImpl",
-            "org.apache.xalan.xsltc.trax.TemplatesImpl"
+    /**
+     * The exact class descriptors produced when deserializing {@code 
HashMap<byte[], byte[]>},
+     * the format written by {@link 
org.apache.kafka.connect.storage.FileOffsetBackingStore}.
+     * Verified by inspection: only {@code java.util.HashMap} and {@code [B} 
(byte[]) appear.
+     * Allowing any additional type would widen the attack surface without 
justification.
+     */
+    public static final Set<String> BASE_TYPES = Set.of(
+            "java.util.HashMap",
+            "[B"   // JVM descriptor for byte[]
     );
 
+    private final Set<String> allowedClasses;
+
+    /**
+     * Uses {@link #BASE_TYPES} as the allowlist. Suitable for {@code 
FileOffsetBackingStore}.
+     */
     public SafeObjectInputStream(InputStream in) throws IOException {
+        this(in, BASE_TYPES);
+    }
+
+    /**
+     * Permits only the classes in {@code allowedClasses}. Use when the stream 
contains
+     * types beyond {@link #BASE_TYPES}; the caller must enumerate every 
expected type.
+     */
+    public SafeObjectInputStream(InputStream in, Set<String> allowedClasses) 
throws IOException {
         super(in);
+        this.allowedClasses = Objects.requireNonNull(allowedClasses, 
"allowedClasses");
     }
 
     @Override
     protected Class<?> resolveClass(ObjectStreamClass desc) throws 
IOException, ClassNotFoundException {
         String name = desc.getName();
-
-        if (isBlocked(name)) {
-            throw new SecurityException("Illegal type to deserialize: 
prevented for security reasons");
+        if (!allowedClasses.contains(name)) {
+            throw new InvalidClassException(name,
+                    "Rejected by deserialization allowlist. If this class is 
legitimately " +
+                    "required, pass an explicit allowedClasses set to " +
+                    "SafeObjectInputStream(InputStream, Set).");
         }
-
         return super.resolveClass(desc);
     }
-
-    private boolean isBlocked(String name) {
-        for (String list : DEFAULT_NO_DESERIALIZE_CLASS_NAMES) {
-            if (name.endsWith(list)) {
-                return true;
-            }
-        }
-
-        return false;
-    }
 }