Harden the java serialization fallback bridge with a JEP-290 serial filter

docs/SECURITY.md

@@ -663,6 +663,20 @@ Storm uses Kryo for serializing tuple data between spouts and bolts. By default,
 
 **Do not set `topology.fall.back.on.java.serialization` to `true` in production.** While topology submitters already run arbitrary code via their spouts and bolts, enabling the Java serialization fallback broadens the attack surface and may allow malicious data from external sources (e.g. message queues) to trigger unintended code execution during deserialization.
 
+As defense in depth, the fallback bridge can be constrained with `topology.fall.back.on.java.serialization.filter`, a [JEP-290](https://openjdk.org/jeps/290) serial-filter pattern applied whenever the bridge deserializes. The filter covers the fallback bridge of the default kryo factory (`DefaultKryoFactory`) only; a custom `topology.kryo.factory` implementation is outside its reach and must handle filtering itself, and the pre-kryo worker state channel (`DefaultStateSerializer`) is not covered either. An empty or unset value leaves the bridge unfiltered, as before.
+
+When the bridge is used, the configured filter is merged with any JVM-wide `jdk.serialFilter` (e.g. set in `worker.childopts`) rather than replacing it, so enabling this setting can only tighten stream filtering, never loosen an operator-set allow-list.
+
+The pattern below is a starting point for operators who enable the fallback anyway and know which classes their payloads contain. It is not a security boundary, and the deny-list is not exhaustive. Following [JEP-290](https://openjdk.org/jeps/290) guidance, an allow-list of the exact classes a topology actually needs is the preferred design; deny-lists only block known gadget classes.
+
+```yaml
+topology.fall.back.on.java.serialization.filter: "!org.apache.commons.collections.functors.*;!org.apache.commons.collections.comparators.*;!org.apache.commons.collections4.functors.*;!org.apache.commons.collections4.comparators.*;!org.apache.commons.beanutils.*;!org.apache.xalan.xsltc.trax.*;!com.sun.org.apache.xalan.internal.**;!com.sun.rowset.*;!com.sun.org.apache.rowset.internal.*;!com.mchange.v2.c3p0.**;!org.codehaus.groovy.runtime.ConvertedClosure;!org.codehaus.groovy.runtime.MethodClosure;!javax.management.BadAttributeValueExpException;!sun.reflect.annotation.AnnotationInvocationHandler;!com.sun.jndi.**;!java.rmi.**;!clojure.**;!org.apache.commons.fileupload.**;!bsh.**;!org.python.**;!org.jboss.**;maxdepth=64;maxrefs=2097152;maxarray=1048576;maxbytes=10485760"
+```
+
+The size limits need a caveat. `maxbytes` counts stream bytes per object, not per tuple, and it is not exact: the filter sees a primitive array when the array is created, before its contents are read, so one big array can slip past the byte cap. `maxarray`, set well below `maxbytes`, is what bounds that case; `maxdepth` and `maxrefs` bound the depth and reference count of the graph. Workloads that legitimately use large arrays should raise these limits. On buffered input the bridge also refuses a declared value length larger than the bytes left in the frame before allocating it; streaming programmatic use of the deserializer is outside this guard.
+
+An invalid pattern is rejected when the topology is submitted, and otherwise fails worker startup.
+
 For tuple encryption, use TLS-based transport encryption (`storm.messaging.netty.tls.enable`) instead of the deprecated `BlowfishTupleSerializer`, which uses a 64-bit block cipher vulnerable to birthday attacks.
 
 ### Log Cleanup

docs/Serialization.md

@@ -61,6 +61,12 @@ Beware that Java serialization is extremely expensive, both in terms of CPU cost
 
 You can turn on/off the behavior to fall back on Java serialization by setting the `Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION` config to true/false. The default value is false for security reasons.
 
+When the fallback is enabled, the bridge can be constrained with `Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER`, a [JEP-290](https://openjdk.org/jeps/290) serial-filter pattern (e.g. `!org.apache.commons.collections4.functors.*;maxbytes=10485760`) applied to every `ObjectInputStream` the bridge uses for deserialization. An invalid pattern is refused when the topology is submitted, and if one reaches a worker anyway its setup fails with the config key in the error. Unlike a JVM-wide `-Djdk.serialFilter`, this filter is topology-scoped and also applies when the deserializer is constructed programmatically, e.g. in local mode. When both are present, the two filters are merged, so both apply.
+
+This only filters the fallback path of the default kryo factory. A custom `topology.kryo.factory` gets no filter from Storm and must set its own; `DefaultStateSerializer`, used for worker state before kryo is set up, is unfiltered as well.
+
+There is no default value; see `docs/SECURITY.md` for a sample pattern. `maxbytes` is a per-object budget (one `ObjectInputStream` per value, not per tuple) and can be overshot by a single large array, since the array is measured before its contents are read. On buffered input the bridge rejects a declared length that exceeds the bytes remaining in the frame before allocating; streaming programmatic use is not covered. An empty or unset value leaves the bridge unfiltered, as before.
+
 ### Tuple compression
 
 For inter-worker (remote) traffic, Storm can optionally compress serialized tuples with [Zstandard](https://facebook.github.io/zstd/) before they are sent over the network. This is intended for one specific scenario: components that emit **large** payloads to a remote worker, where the bytes saved on the wire outweigh the CPU cost of compression. A good example is a spout that emits entire lines of text to a downstream bolt running on a different worker.

storm-client/src/jvm/org/apache/storm/Config.java

@@ -660,6 +660,24 @@ public class Config extends HashMap<String, Object> {
      */
     @IsBoolean
     public static final String TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION = "topology.fall.back.on.java.serialization";
+    /**
+     * Optional <a href="https://openjdk.org/jeps/290">JEP-290</a> serial-filter pattern applied to the
+     * Java-serialization fallback bridge that {@link #TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION} enables for
+     * unregistered classes. When set to a non-empty pattern, it is parsed once at kryo construction and installed
+     * on every {@code ObjectInputStream} used to deserialize fallback values, so stream classes rejected by the
+     * filter are neither instantiated nor have their {@code readObject} logic invoked. Unset by default;
+     * {@code docs/SECURITY.md} has a sample deny-list pattern to start from. An empty or unset value leaves the
+     * bridge unfiltered, as before. An invalid pattern fails topology submission and worker startup with a clear error.
+     * Note: unlike a JVM-wide {@code jdk.serialFilter}, this is topology-scoped and also applies when the
+     * deserializer is built programmatically, e.g. local mode; when both are present the two filters are merged,
+     * so each takes effect. Filters only {@code DefaultKryoFactory}'s fallback path; a custom
+     * {@code topology.kryo.factory} or the pre-kryo state serializer must arrange its own filtering.
+     * {@code maxbytes} is per value, not per tuple, and approximate: a large primitive array is measured at
+     * creation and may exceed it. Example deny-list: {@code !org.apache.commons.collections4.functors.*}.
+     */
+    @IsString(validatorClass = ConfigValidation.SerialFilterPatternValidator.class)
+    public static final String TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER =
+            "topology.fall.back.on.java.serialization.filter";
     /**
      * Topology-specific options for the worker child process. This is used in addition to WORKER_CHILDOPTS.
      */

storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java

@@ -15,6 +15,7 @@
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.util.Util;
+import java.io.ObjectInputFilter;
 import java.util.Map;
 import org.apache.storm.Config;
 import org.slf4j.Logger;
@@ -26,12 +27,29 @@ public class DefaultKryoFactory implements IKryoFactory {
 
     @Override
     public Kryo getKryo(Map<String, Object> conf) {
-        KryoSerializableDefault k = new KryoSerializableDefault();
+        KryoSerializableDefault k = new KryoSerializableDefault(getJavaSerializationFilter(conf));
         k.setRegistrationRequired(!((Boolean) conf.get(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION)));
         k.setReferences(false);
         return k;
     }
 
+    /**
+     * Parses the pattern once at kryo construction so an invalid pattern fails worker setup with a clear error
+     * instead of failing per-tuple on the read path. Returns null when the key is unset or empty (no filter).
+     */
+    private static ObjectInputFilter getJavaSerializationFilter(Map<String, Object> conf) {
+        String filterSpec = (String) conf.get(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER);
+        if (filterSpec == null || filterSpec.isEmpty()) {
+            return null;
+        }
+        try {
+            return ObjectInputFilter.Config.createFilter(filterSpec);
+        } catch (IllegalArgumentException e) {
+            throw new IllegalArgumentException("Invalid " + Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER
+                    + " pattern: \"" + filterSpec + "\"", e);
+        }
+    }
+
     @Override
     public void preRegister(Kryo k, Map<String, Object> conf) {
     }
@@ -47,6 +65,11 @@ public void postDecorate(Kryo k, Map<String, Object> conf) {
 
     public static class KryoSerializableDefault extends Kryo {
         boolean override = false;
+        private final ObjectInputFilter javaSerializationFilter;
+
+        KryoSerializableDefault(ObjectInputFilter javaSerializationFilter) {
+            this.javaSerializationFilter = javaSerializationFilter;
+        }
 
         public void overrideDefault(boolean value) {
             override = value;
@@ -61,7 +84,7 @@ public Serializer getDefaultSerializer(Class type) {
                         Util.className(type), Util.className(type)
                 );
 
-                return new SerializableSerializer();
+                return new SerializableSerializer(javaSerializationFilter);
             } else {
                 return super.getDefaultSerializer(type);
             }

storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java

@@ -13,18 +13,36 @@
 package org.apache.storm.serialization;
 
 import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.KryoException;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.ObjectInputFilter;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 
 
 public class SerializableSerializer extends Serializer<Object> {
 
+    /**
+     * Optional JEP-290 filter applied to each ObjectInputStream used for deserialization (null means unfiltered,
+     * as before). The filter itself is created once from
+     * {@link org.apache.storm.Config#TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER} by {@link DefaultKryoFactory};
+     * instances returned by {@link ObjectInputFilter.Config#createFilter} are immutable and safe to share across streams.
+     */
+    private final ObjectInputFilter serialFilter;
+
+    public SerializableSerializer() {
+        this(null);
+    }
+
+    public SerializableSerializer(ObjectInputFilter serialFilter) {
+        this.serialFilter = serialFilter;
+    }
+
     @Override
     public void write(Kryo kryo, Output output, Object object) {
         ByteArrayOutputStream bos = new ByteArrayOutputStream();
@@ -43,14 +61,39 @@ public void write(Kryo kryo, Output output, Object object) {
     @Override
     public Object read(Kryo kryo, Input input, Class c) {
         int len = input.readInt();
+        if (len < 0) {
+            throw new KryoException("Invalid java-serialized value length: " + len);
+        }
+        // For a buffer-backed Input the remaining bytes are known (position/limit), so a declared length larger than the
+        // bytes actually left is refused before the new byte[len] allocation; a stream-backed Input may still deliver the
+        // declared bytes later, so the upper bound is not checked there.
+        if (input.getInputStream() == null) {
+            int remaining = input.limit() - input.position();
+            if (len > remaining) {
+                throw new KryoException("Declared java-serialized value length exceeds the input's remaining bytes "
+                        + "(declared: " + len + ", remaining: " + remaining + ")");
+            }
+        }
         byte[] ser = new byte[len];
         input.readBytes(ser);
         ByteArrayInputStream bis = new ByteArrayInputStream(ser);
         try {
             ObjectInputStream ois = new ObjectInputStream(bis);
+            if (serialFilter != null) {
+                ois.setObjectInputFilter(mergeWithExisting(serialFilter, ois.getObjectInputFilter()));
+            }
             return ois.readObject();
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
     }
+
+    /**
+     * Combines the configured filter with the stream's existing filter (a JVM-wide {@code jdk.serialFilter}, if any), so both
+     * the configured pattern and any process-wide filter apply to the stream: per JEP-290,
+     * {@link ObjectInputStream#setObjectInputFilter} overrides the process-wide filter for that stream unless the two are merged.
+     */
+    static ObjectInputFilter mergeWithExisting(ObjectInputFilter configured, ObjectInputFilter existing) {
+        return existing != null ? ObjectInputFilter.merge(configured, existing) : configured;
+    }
 }

storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java

@@ -15,6 +15,7 @@
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
+import java.io.ObjectInputFilter;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Field;
 import java.lang.reflect.InvocationTargetException;
@@ -590,6 +591,30 @@ public void validateField(String name, Object o) {
         }
     }
 
+    /**
+     * Validates that a String is a well-formed JEP-290 serial-filter pattern.
+     */
+    public static class SerialFilterPatternValidator extends Validator {
+
+        @Override
+        public void validateField(String name, Object o) {
+            if (o == null) {
+                return;
+            }
+            SimpleTypeValidator.validateField(name, String.class, o);
+            String pattern = (String) o;
+            if (pattern.isEmpty()) {
+                return; // empty means no filter
+            }
+            try {
+                ObjectInputFilter.Config.createFilter(pattern);
+            } catch (IllegalArgumentException e) {
+                throw new IllegalArgumentException(
+                    "Field " + name + " is not a valid JEP-290 serial-filter pattern: '" + pattern + "'", e);
+            }
+        }
+    }
+
     /**
      * Validates each entry in a list.
      */

storm-client/test/jvm/org/apache/storm/TestConfigValidate.java

@@ -92,6 +92,30 @@ public void invalidPacemakerAuthTest() {
         assertThrows(IllegalArgumentException.class, () -> ConfigValidation.validateFields(conf));
     }
 
+    @Test
+    public void fallbackJavaSerializationFilterPatternTest() {
+        // A malformed JEP-290 pattern must be rejected at conf validation time, with the key named in the error.
+        for (String invalid : Arrays.asList("!", "maxbytes=not-a-number")) {
+            Map<String, Object> conf = new HashMap<>();
+            conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER, invalid);
+            IllegalArgumentException ex =
+                assertThrows(IllegalArgumentException.class, () -> ConfigValidation.validateFields(conf));
+            // The validator reports the config field name (the convention for all validators here).
+            assertTrue(ex.getMessage().contains("TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER"),
+                "message should name the offending field: " + ex.getMessage());
+        }
+
+        // Valid, empty (no filter), and absent values all pass.
+        Map<String, Object> conf = new HashMap<>();
+        conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER,
+                 "!org.apache.commons.collections4.functors.*;maxarray=1048576;maxbytes=10485760");
+        ConfigValidation.validateFields(conf);
+        conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER, "");
+        ConfigValidation.validateFields(conf);
+        conf.remove(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER);
+        ConfigValidation.validateFields(conf);
+    }
+
     @Test
     public void upstreamFeedbackRequiresEwmaTest() {
         // Cross-field rule: enabling feedback without EWMA stats is a no-op, so it is rejected.