From 740be7761a75b002d08d0050727f8efe6f68ceaa Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Fri, 4 Sep 2026 11:40:58 +0800 Subject: [PATCH 1/2] Harden the java serialization fallback bridge with a JEP-290 serial filter Add topology.fall.back.on.java.serialization.filter, a JEP-290 filter pattern for the java serialization fallback bridge. DefaultKryoFactory parses the pattern once at kryo construction, so an invalid pattern fails worker setup with the config key in the error, and SerializableSerializer installs it via setObjectInputFilter whenever it deserializes. The filter is topology-scoped and also covers programmatic construction such as local mode, which a JVM-wide jdk.serialFilter in worker.childopts does not reach. conf/defaults.yaml sets a default pattern: a deny-list of well-known gadget namespaces (commons-collections 3/4 functors and comparators, beanutils, xalan external and JDK-internal, rowset, c3p0, groovy closures) plus maxbytes=10485760. An empty or unset value leaves the bridge unfiltered, as before. The pattern uses JEP-290 wildcards: pkg.* covers direct package members and pkg.** also covers subpackages; tests exercise both depths against loadable classes in denied packages, end to end through KryoValuesSerializer and KryoValuesDeserializer. --- conf/defaults.yaml | 1 + docs/SECURITY.md | 2 + docs/Serialization.md | 2 + .../src/jvm/org/apache/storm/Config.java | 14 ++ .../serialization/DefaultKryoFactory.java | 26 ++- .../serialization/SerializableSerializer.java | 20 ++ .../mchange/v2/c3p0/impl/SimulatedGadget.java | 23 +++ .../comparators/SimulatedGadget.java | 23 +++ .../collections/functors/SimulatedGadget.java | 24 +++ .../SerializableSerializerFilterTest.java | 187 ++++++++++++++++++ 10 files changed, 321 insertions(+), 1 deletion(-) create mode 100644 storm-client/test/jvm/com/mchange/v2/c3p0/impl/SimulatedGadget.java create mode 100644 storm-client/test/jvm/org/apache/commons/collections/comparators/SimulatedGadget.java create mode 100644 storm-client/test/jvm/org/apache/commons/collections/functors/SimulatedGadget.java create mode 100644 storm-client/test/jvm/org/apache/storm/serialization/SerializableSerializerFilterTest.java diff --git a/conf/defaults.yaml b/conf/defaults.yaml index 6fd7a04b9d4..d8e0422849e 100644 --- a/conf/defaults.yaml +++ b/conf/defaults.yaml @@ -309,6 +309,7 @@ topology.upstream.feedback.freq.secs: 10 topology.upstream.feedback.enable: false topology.builtin.metrics.bucket.size.secs: 60 topology.fall.back.on.java.serialization: false +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;maxbytes=10485760" topology.worker.childopts: null topology.worker.logwriter.childopts: "-Xmx64m" topology.tick.tuple.freq.secs: null diff --git a/docs/SECURITY.md b/docs/SECURITY.md index 03c71993c81..028e870c360 100644 --- a/docs/SECURITY.md +++ b/docs/SECURITY.md @@ -663,6 +663,8 @@ 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 is constrained by `topology.fall.back.on.java.serialization.filter`, a [JEP-290](https://openjdk.org/jeps/290) serial-filter pattern applied whenever the bridge deserializes. `conf/defaults.yaml` sets a default pattern: a deny-list of well-known gadget namespaces plus `maxbytes=10485760`. An empty or unset value leaves the bridge unfiltered, as before. This reduces the impact of a misconfigured cluster. + 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 diff --git a/docs/Serialization.md b/docs/Serialization.md index 8f87ba6b043..81f9def5548 100644 --- a/docs/Serialization.md +++ b/docs/Serialization.md @@ -61,6 +61,8 @@ 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. The pattern is parsed when the serialization stack is created, so an invalid pattern fails worker setup with the config key in the error. `conf/defaults.yaml` carries a default deny-list of well-known gadget namespaces with a `maxbytes=10485760` limit; an empty or unset value leaves the bridge unfiltered, as before. 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. + ### 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. diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index 4e3dbe061bf..304210bcda1 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -660,6 +660,20 @@ public class Config extends HashMap { */ @IsBoolean public static final String TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION = "topology.fall.back.on.java.serialization"; + /** + * Optional JEP-290 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. {@code conf/defaults.yaml} + * sets a default gadget deny-list with a {@code maxbytes=10485760} limit; an empty or unset value leaves the + * bridge unfiltered, as before. An invalid pattern fails 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. Example deny-list: {@code !org.apache.commons.collections4.functors.*}. + */ + @IsString + 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. */ diff --git a/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java b/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java index ef00ea47f75..aba08fe2001 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java +++ b/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; @@ -29,9 +30,27 @@ public Kryo getKryo(Map conf) { KryoSerializableDefault k = new KryoSerializableDefault(); k.setRegistrationRequired(!((Boolean) conf.get(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION))); k.setReferences(false); + k.setJavaSerializationFilter(getJavaSerializationFilter(conf)); 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 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 RuntimeException("Invalid " + Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER + + " pattern: \"" + filterSpec + "\"", e); + } + } + @Override public void preRegister(Kryo k, Map conf) { } @@ -47,6 +66,11 @@ public void postDecorate(Kryo k, Map conf) { public static class KryoSerializableDefault extends Kryo { boolean override = false; + private ObjectInputFilter javaSerializationFilter; + + public void setJavaSerializationFilter(ObjectInputFilter filter) { + this.javaSerializationFilter = filter; + } public void overrideDefault(boolean value) { override = value; @@ -61,7 +85,7 @@ public Serializer getDefaultSerializer(Class type) { Util.className(type), Util.className(type) ); - return new SerializableSerializer(); + return new SerializableSerializer(javaSerializationFilter); } else { return super.getDefaultSerializer(type); } diff --git a/storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java b/storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java index f17689cb6de..b905419a041 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java @@ -19,12 +19,29 @@ 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 { + /** + * 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(); @@ -48,6 +65,9 @@ public Object read(Kryo kryo, Input input, Class c) { ByteArrayInputStream bis = new ByteArrayInputStream(ser); try { ObjectInputStream ois = new ObjectInputStream(bis); + if (serialFilter != null) { + ois.setObjectInputFilter(serialFilter); + } return ois.readObject(); } catch (Exception e) { throw new RuntimeException(e); diff --git a/storm-client/test/jvm/com/mchange/v2/c3p0/impl/SimulatedGadget.java b/storm-client/test/jvm/com/mchange/v2/c3p0/impl/SimulatedGadget.java new file mode 100644 index 00000000000..2d0d161714c --- /dev/null +++ b/storm-client/test/jvm/com/mchange/v2/c3p0/impl/SimulatedGadget.java @@ -0,0 +1,23 @@ +/** + * 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 + * + * 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 com.mchange.v2.c3p0.impl; + +import java.io.Serializable; + +/** + * Placeholder under com.mchange.v2.c3p0.impl; sits in a subpackage so the tests cover the difference between pkg.* and pkg.** entries. + */ +public class SimulatedGadget implements Serializable { + + private static final long serialVersionUID = 1L; +} diff --git a/storm-client/test/jvm/org/apache/commons/collections/comparators/SimulatedGadget.java b/storm-client/test/jvm/org/apache/commons/collections/comparators/SimulatedGadget.java new file mode 100644 index 00000000000..82648a0da18 --- /dev/null +++ b/storm-client/test/jvm/org/apache/commons/collections/comparators/SimulatedGadget.java @@ -0,0 +1,23 @@ +/** + * 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 + * + * 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 org.apache.commons.collections.comparators; + +import java.io.Serializable; + +/** + * Placeholder in the commons-collections comparators package; TransformingComparator gadgets live here, not in functors. + */ +public class SimulatedGadget implements Serializable { + + private static final long serialVersionUID = 1L; +} diff --git a/storm-client/test/jvm/org/apache/commons/collections/functors/SimulatedGadget.java b/storm-client/test/jvm/org/apache/commons/collections/functors/SimulatedGadget.java new file mode 100644 index 00000000000..f82278d9e6f --- /dev/null +++ b/storm-client/test/jvm/org/apache/commons/collections/functors/SimulatedGadget.java @@ -0,0 +1,24 @@ +/** + * 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 + * + * 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 org.apache.commons.collections.functors; + +import java.io.Serializable; + +/** + * Placeholder class in the commons-collections functors package so the filter tests deny a class that actually loads (the gadget + * here is InvokerTransformer). + */ +public class SimulatedGadget implements Serializable { + + private static final long serialVersionUID = 1L; +} diff --git a/storm-client/test/jvm/org/apache/storm/serialization/SerializableSerializerFilterTest.java b/storm-client/test/jvm/org/apache/storm/serialization/SerializableSerializerFilterTest.java new file mode 100644 index 00000000000..5983cbe5d7f --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/serialization/SerializableSerializerFilterTest.java @@ -0,0 +1,187 @@ +/** + * 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 + * + * 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 org.apache.storm.serialization; + +import java.io.InvalidClassException; +import java.io.ObjectInputFilter; +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.PriorityQueue; +import org.apache.commons.collections.functors.SimulatedGadget; +import org.apache.storm.Config; +import org.apache.storm.serialization.types.ListDelegateSerializer; +import org.apache.storm.utils.Utils; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertIterableEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests for the JEP-290 serial filter ({@link Config#TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER}) protecting the + * java-serialization fallback bridge. Every allow and deny case exercises an actual round-trip through the bridge, not the + * filter API in isolation. All round-trips go through KryoValuesSerializer and KryoValuesDeserializer end to end. + */ +public class SerializableSerializerFilterTest { + + /** The maxbytes limit set in conf/defaults.yaml. */ + private static final long DEFAULT_MAX_BYTES = 10485760L; + + /** + * Minimal conf that routes unregistered classes through the java-serialization fallback bridge. {@code filterSpec == null} + * means the filter key is absent from the conf entirely (the pre-existing behavior). + */ + private Map bridgeConf(String filterSpec) { + Map conf = new Config(); + conf.put(Config.TOPOLOGY_KRYO_FACTORY, DefaultKryoFactory.class.getName()); + conf.put(Config.TOPOLOGY_TUPLE_SERIALIZER, ListDelegateSerializer.class.getName()); + conf.put(Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS, false); + conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, true); + if (filterSpec != null) { + conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER, filterSpec); + } + return conf; + } + + /** conf assembled exactly like a worker's would be: defaults.yaml + topology-level overrides. */ + private Map defaultsBridgeConf() { + Map conf = new Config(); + conf.putAll(Utils.readDefaultConfig()); + conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, true); + return conf; + } + + private Object roundTrip(Map conf, Object value) { + KryoValuesSerializer serializer = new KryoValuesSerializer(conf); + KryoValuesDeserializer deserializer = new KryoValuesDeserializer(conf); + return deserializer.deserialize(serializer.serialize(Collections.singletonList(value))).get(0); + } + + /** Serializes {@code value} and asserts that reading it back fails with a JEP-290 rejection in the cause chain. */ + private void assertRejectedOnRead(Map conf, Object value) { + KryoValuesSerializer serializer = new KryoValuesSerializer(conf); + KryoValuesDeserializer deserializer = new KryoValuesDeserializer(conf); + // Writing is plain java serialization (filters apply to deserialization only), so this must succeed. + byte[] bytes = serializer.serialize(Collections.singletonList(value)); + RuntimeException ex = assertThrows(RuntimeException.class, () -> deserializer.deserialize(bytes)); + assertTrue(hasCause(ex, InvalidClassException.class), + "expected the JEP-290 filter rejection in the cause chain, got: " + ex); + } + + private static boolean hasCause(Throwable throwable, Class type) { + for (Throwable t = throwable; t != null; t = t.getCause()) { + if (type.isInstance(t)) { + return true; + } + } + return false; + } + + /** ArrayDeque and PriorityQueue do not override equals, so round-trips are compared by iteration content. */ + private static void assertSameContent(Iterable expected, Iterable actual) { + assertIterableEquals(expected, actual); + } + + @Test + public void testFilterRejectsDeniedClassOnDeserialization() { + Map conf = bridgeConf("!java.util.PriorityQueue"); + + PriorityQueue original = new PriorityQueue<>(Arrays.asList(3, 1, 2)); + assertRejectedOnRead(conf, original); + } + + @Test + public void testFilterAllowsNonDeniedClassesRoundTrip() { + Map conf = bridgeConf("!java.util.PriorityQueue"); + + // HashMap has a dedicated kryo serializer: ordinary payloads must keep round-tripping. + HashMap hashMap = new HashMap<>(Collections.singletonMap("one", 1)); + assertEquals(hashMap, roundTrip(conf, hashMap)); + + // ArrayDeque is unregistered and Serializable, so it travels through the java-serialization bridge itself. + ArrayDeque deque = new ArrayDeque<>(Arrays.asList("a", "b", "c")); + assertSameContent(deque, (ArrayDeque) roundTrip(conf, deque)); + } + + @Test + public void testUnsetFilterKeyKeepsUnfilteredBehavior() { + // No filter key in the conf at all: PriorityQueue must round-trip like it did before the filter existed. + PriorityQueue original = new PriorityQueue<>(Arrays.asList(5, 4, 6)); + assertSameContent(original, (PriorityQueue) roundTrip(bridgeConf(null), original)); + } + + @Test + public void testInvalidPatternFailsFastAtKryoConstruction() { + // The parser only rejects a few inputs: '!' (no pattern) and a non-numeric maxbytes; malformed class patterns + // are ignored, not rejected. + for (String invalid : Arrays.asList("!", "maxbytes=not-a-number")) { + Map conf = bridgeConf(invalid); + RuntimeException ex = assertThrows(RuntimeException.class, () -> new KryoValuesSerializer(conf)); + assertTrue(ex.getMessage().contains(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER), + "error must name the offending config key: " + ex.getMessage()); + } + } + + @Test + public void testDefaultsYamlFilterParsesAndDeniesGadgetPackage() { + Map defaults = Utils.readDefaultConfig(); + Object filterSpec = defaults.get(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER); + assertNotNull(filterSpec, "conf/defaults.yaml must define a default " + + Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER); + // The default pattern must parse (bad patterns throw IllegalArgumentException here). + ObjectInputFilter.Config.createFilter((String) filterSpec); + + // Through an actual round-trip: a class in a deny-listed package is rejected on read... + assertRejectedOnRead(defaultsBridgeConf(), new SimulatedGadget()); + // ...while ordinary JDK collections keep round-tripping under the same default. + PriorityQueue queue = new PriorityQueue<>(Arrays.asList(5, 4, 6)); + assertSameContent(queue, (PriorityQueue) roundTrip(defaultsBridgeConf(), queue)); + } + + @Test + public void testDefaultFilterRejectsCommonsCollections3Comparators() { + // CC3's TransformingComparator gadget chain lives in the comparators package (the functors pair alone is not enough). + assertRejectedOnRead(defaultsBridgeConf(), new org.apache.commons.collections.comparators.SimulatedGadget()); + } + + @Test + public void testDefaultFilterRejectsSubpackagesOfRecursiveWildcardEntries() { + // Wildcard depth matters: 'pkg.**' denies subpackages too; 'pkg.*' does not (c3p0.impl sits under the + // shipped '!com.mchange.v2.c3p0.**' entry). + assertRejectedOnRead(defaultsBridgeConf(), new com.mchange.v2.c3p0.impl.SimulatedGadget()); + } + + @Test + public void testDefaultFilterEnforcesMaxBytesLimit() { + Map conf = defaultsBridgeConf(); + // Every array read re-invokes the filter, so a stream built from many small arrays makes the cumulative + // streamBytes() limit bite mid-deserialization (one huge primitive payload would not re-invoke the filter). + ArrayDeque big = new ArrayDeque<>(); + for (int i = 0; i < 11000; i++) { + big.add(new byte[1024]); + } + KryoValuesSerializer serializer = new KryoValuesSerializer(conf); + KryoValuesDeserializer deserializer = new KryoValuesDeserializer(conf); + byte[] bytes = serializer.serialize(Collections.singletonList(big)); + assertTrue(bytes.length > DEFAULT_MAX_BYTES, "payload must exceed the default limit, was " + bytes.length); + + RuntimeException ex = assertThrows(RuntimeException.class, () -> deserializer.deserialize(bytes)); + assertTrue(hasCause(ex, InvalidClassException.class), + "expected the maxbytes rejection in the cause chain, got: " + ex); + } +} From 54e4926f23852160ec35ccd03d891961383dec7a Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Mon, 7 Sep 2026 11:00:03 +0800 Subject: [PATCH 2/2] Rework the fallback serial filter: opt-in sample, merge, validation Remove the default value of topology.fall.back.on.java.serialization.filter from conf/defaults.yaml. A shipped deny-list is not a security boundary, and defaulting it on silently replaced the operator's JVM-wide jdk.serialFilter on the bridge streams. docs/SECURITY.md now carries a sample pattern instead: the previous gadget deny-list plus well-known additions (JMX reflection gadget, JNDI, RMI, Clojure, FileUpload, BeanShell, Jython, JBoss). maxbytes alone would not bound a single large primitive array, so the pattern also carries maxdepth/maxrefs/maxarray. SerializableSerializer now reads the stream's existing filter (a JVM-wide jdk.serialFilter, if set) and merges the two via ObjectInputFilter.merge, so a JVM-wide allow-list still applies on this path. The bridge also validates the declared length before allocating: a negative length is always refused, and on buffered input (the tuple paths) a declared length larger than the bytes remaining in the frame is refused before the byte[] allocation. Stream-backed programmatic use keeps the previous behavior, since the stream may still deliver the declared bytes. KryoSerializableDefault takes the filter as a constructor argument and holds it in a final field; the mutable setter is removed so the filter cannot be swapped after construction. ConfigValidation gains SerialFilterPatternValidator, wired to the TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER field, so a malformed pattern fails topology submission with the key named; the parse error in DefaultKryoFactory is now an IllegalArgumentException. The filter applies to the default factory's fallback bridge; a custom topology.kryo.factory or the pre-kryo state serializer is not covered. Tests now use JDK classes only and cover wildcard depth, the sample pattern, the size limits, the length guard, and the merge helper. --- conf/defaults.yaml | 1 - docs/SECURITY.md | 14 +- docs/Serialization.md | 6 +- .../src/jvm/org/apache/storm/Config.java | 18 +- .../serialization/DefaultKryoFactory.java | 11 +- .../serialization/SerializableSerializer.java | 25 +- .../storm/validation/ConfigValidation.java | 25 ++ .../mchange/v2/c3p0/impl/SimulatedGadget.java | 23 -- .../comparators/SimulatedGadget.java | 23 -- .../collections/functors/SimulatedGadget.java | 24 -- .../org/apache/storm/TestConfigValidate.java | 24 ++ .../SerializableSerializerFilterTest.java | 240 +++++++++++++----- 12 files changed, 289 insertions(+), 145 deletions(-) delete mode 100644 storm-client/test/jvm/com/mchange/v2/c3p0/impl/SimulatedGadget.java delete mode 100644 storm-client/test/jvm/org/apache/commons/collections/comparators/SimulatedGadget.java delete mode 100644 storm-client/test/jvm/org/apache/commons/collections/functors/SimulatedGadget.java diff --git a/conf/defaults.yaml b/conf/defaults.yaml index d8e0422849e..6fd7a04b9d4 100644 --- a/conf/defaults.yaml +++ b/conf/defaults.yaml @@ -309,7 +309,6 @@ topology.upstream.feedback.freq.secs: 10 topology.upstream.feedback.enable: false topology.builtin.metrics.bucket.size.secs: 60 topology.fall.back.on.java.serialization: false -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;maxbytes=10485760" topology.worker.childopts: null topology.worker.logwriter.childopts: "-Xmx64m" topology.tick.tuple.freq.secs: null diff --git a/docs/SECURITY.md b/docs/SECURITY.md index 028e870c360..36154271f13 100644 --- a/docs/SECURITY.md +++ b/docs/SECURITY.md @@ -663,7 +663,19 @@ 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 is constrained by `topology.fall.back.on.java.serialization.filter`, a [JEP-290](https://openjdk.org/jeps/290) serial-filter pattern applied whenever the bridge deserializes. `conf/defaults.yaml` sets a default pattern: a deny-list of well-known gadget namespaces plus `maxbytes=10485760`. An empty or unset value leaves the bridge unfiltered, as before. This reduces the impact of a misconfigured cluster. +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. diff --git a/docs/Serialization.md b/docs/Serialization.md index 81f9def5548..b8701e8aacc 100644 --- a/docs/Serialization.md +++ b/docs/Serialization.md @@ -61,7 +61,11 @@ 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. The pattern is parsed when the serialization stack is created, so an invalid pattern fails worker setup with the config key in the error. `conf/defaults.yaml` carries a default deny-list of well-known gadget namespaces with a `maxbytes=10485760` limit; an empty or unset value leaves the bridge unfiltered, as before. 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 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 diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index 304210bcda1..53f8eefdf27 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -665,13 +665,17 @@ public class Config extends HashMap { * 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. {@code conf/defaults.yaml} - * sets a default gadget deny-list with a {@code maxbytes=10485760} limit; an empty or unset value leaves the - * bridge unfiltered, as before. An invalid pattern fails 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. Example deny-list: {@code !org.apache.commons.collections4.functors.*}. - */ - @IsString + * 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"; /** diff --git a/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java b/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java index aba08fe2001..6ed3be28273 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java @@ -27,10 +27,9 @@ public class DefaultKryoFactory implements IKryoFactory { @Override public Kryo getKryo(Map 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); - k.setJavaSerializationFilter(getJavaSerializationFilter(conf)); return k; } @@ -46,7 +45,7 @@ private static ObjectInputFilter getJavaSerializationFilter(Map try { return ObjectInputFilter.Config.createFilter(filterSpec); } catch (IllegalArgumentException e) { - throw new RuntimeException("Invalid " + Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER + throw new IllegalArgumentException("Invalid " + Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER + " pattern: \"" + filterSpec + "\"", e); } } @@ -66,10 +65,10 @@ public void postDecorate(Kryo k, Map conf) { public static class KryoSerializableDefault extends Kryo { boolean override = false; - private ObjectInputFilter javaSerializationFilter; + private final ObjectInputFilter javaSerializationFilter; - public void setJavaSerializationFilter(ObjectInputFilter filter) { - this.javaSerializationFilter = filter; + KryoSerializableDefault(ObjectInputFilter javaSerializationFilter) { + this.javaSerializationFilter = javaSerializationFilter; } public void overrideDefault(boolean value) { diff --git a/storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java b/storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java index b905419a041..042dceade10 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java @@ -13,6 +13,7 @@ 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; @@ -60,17 +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(serialFilter); + 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; + } } diff --git a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java index 7d59b4a2ebd..4ebef6e4325 100644 --- a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java +++ b/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. */ diff --git a/storm-client/test/jvm/com/mchange/v2/c3p0/impl/SimulatedGadget.java b/storm-client/test/jvm/com/mchange/v2/c3p0/impl/SimulatedGadget.java deleted file mode 100644 index 2d0d161714c..00000000000 --- a/storm-client/test/jvm/com/mchange/v2/c3p0/impl/SimulatedGadget.java +++ /dev/null @@ -1,23 +0,0 @@ -/** - * 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 - * - * 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 com.mchange.v2.c3p0.impl; - -import java.io.Serializable; - -/** - * Placeholder under com.mchange.v2.c3p0.impl; sits in a subpackage so the tests cover the difference between pkg.* and pkg.** entries. - */ -public class SimulatedGadget implements Serializable { - - private static final long serialVersionUID = 1L; -} diff --git a/storm-client/test/jvm/org/apache/commons/collections/comparators/SimulatedGadget.java b/storm-client/test/jvm/org/apache/commons/collections/comparators/SimulatedGadget.java deleted file mode 100644 index 82648a0da18..00000000000 --- a/storm-client/test/jvm/org/apache/commons/collections/comparators/SimulatedGadget.java +++ /dev/null @@ -1,23 +0,0 @@ -/** - * 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 - * - * 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 org.apache.commons.collections.comparators; - -import java.io.Serializable; - -/** - * Placeholder in the commons-collections comparators package; TransformingComparator gadgets live here, not in functors. - */ -public class SimulatedGadget implements Serializable { - - private static final long serialVersionUID = 1L; -} diff --git a/storm-client/test/jvm/org/apache/commons/collections/functors/SimulatedGadget.java b/storm-client/test/jvm/org/apache/commons/collections/functors/SimulatedGadget.java deleted file mode 100644 index f82278d9e6f..00000000000 --- a/storm-client/test/jvm/org/apache/commons/collections/functors/SimulatedGadget.java +++ /dev/null @@ -1,24 +0,0 @@ -/** - * 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 - * - * 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 org.apache.commons.collections.functors; - -import java.io.Serializable; - -/** - * Placeholder class in the commons-collections functors package so the filter tests deny a class that actually loads (the gadget - * here is InvokerTransformer). - */ -public class SimulatedGadget implements Serializable { - - private static final long serialVersionUID = 1L; -} diff --git a/storm-client/test/jvm/org/apache/storm/TestConfigValidate.java b/storm-client/test/jvm/org/apache/storm/TestConfigValidate.java index 32bb1433858..6105bc98737 100644 --- a/storm-client/test/jvm/org/apache/storm/TestConfigValidate.java +++ b/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 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 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. diff --git a/storm-client/test/jvm/org/apache/storm/serialization/SerializableSerializerFilterTest.java b/storm-client/test/jvm/org/apache/storm/serialization/SerializableSerializerFilterTest.java index 5983cbe5d7f..4c4732ea65a 100644 --- a/storm-client/test/jvm/org/apache/storm/serialization/SerializableSerializerFilterTest.java +++ b/storm-client/test/jvm/org/apache/storm/serialization/SerializableSerializerFilterTest.java @@ -12,35 +12,57 @@ package org.apache.storm.serialization; +import com.esotericsoftware.kryo.KryoException; +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.InvalidClassException; import java.io.ObjectInputFilter; +import java.io.ObjectOutputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; import java.util.ArrayDeque; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.PriorityQueue; -import org.apache.commons.collections.functors.SimulatedGadget; +import java.util.logging.Level; +import java.util.regex.Pattern; +import javax.management.BadAttributeValueExpException; import org.apache.storm.Config; import org.apache.storm.serialization.types.ListDelegateSerializer; import org.apache.storm.utils.Utils; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertIterableEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; /** * Tests for the JEP-290 serial filter ({@link Config#TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER}) protecting the - * java-serialization fallback bridge. Every allow and deny case exercises an actual round-trip through the bridge, not the - * filter API in isolation. All round-trips go through KryoValuesSerializer and KryoValuesDeserializer end to end. + * java-serialization fallback bridge. Round-trip cases exercise an actual pass through the bridge via KryoValuesSerializer and + * KryoValuesDeserializer end to end, using JDK classes only (no fixtures under third-party package names). Filter semantics + * that a round-trip cannot express (merging, limit tightening) are asserted through checkInput with synthetic FilterInfos + * (rejections only). */ public class SerializableSerializerFilterTest { - /** The maxbytes limit set in conf/defaults.yaml. */ - private static final long DEFAULT_MAX_BYTES = 10485760L; + /** The maxbytes limit carried by {@link #SAMPLE_PATTERN}. */ + private static final long SAMPLE_MAX_BYTES = 10485760L; + + /** + * The sample filter pattern documented in docs/SECURITY.md: a deny-list of well-known gadget namespaces plus + * depth/reference/array/byte limits. Only entries whose classes exist on a plain JDK classpath are asserted against real + * Class objects; the rest are covered by the parse (createFilter) and by the doc-sync test below. + */ + private static final String SAMPLE_PATTERN = "!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"; /** * Minimal conf that routes unregistered classes through the java-serialization fallback bridge. {@code filterSpec == null} @@ -58,14 +80,6 @@ private Map bridgeConf(String filterSpec) { return conf; } - /** conf assembled exactly like a worker's would be: defaults.yaml + topology-level overrides. */ - private Map defaultsBridgeConf() { - Map conf = new Config(); - conf.putAll(Utils.readDefaultConfig()); - conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, true); - return conf; - } - private Object roundTrip(Map conf, Object value) { KryoValuesSerializer serializer = new KryoValuesSerializer(conf); KryoValuesDeserializer deserializer = new KryoValuesDeserializer(conf); @@ -79,22 +93,43 @@ private void assertRejectedOnRead(Map conf, Object value) { // Writing is plain java serialization (filters apply to deserialization only), so this must succeed. byte[] bytes = serializer.serialize(Collections.singletonList(value)); RuntimeException ex = assertThrows(RuntimeException.class, () -> deserializer.deserialize(bytes)); - assertTrue(hasCause(ex, InvalidClassException.class), + assertTrue(Utils.exceptionCauseIsInstanceOf(InvalidClassException.class, ex), "expected the JEP-290 filter rejection in the cause chain, got: " + ex); } - private static boolean hasCause(Throwable throwable, Class type) { - for (Throwable t = throwable; t != null; t = t.getCause()) { - if (type.isInstance(t)) { - return true; - } - } - return false; + /** A FilterInfo describing only the candidate class; depth/references/streamBytes stay in-range so they cannot reject on their own. */ + private static ObjectInputFilter.FilterInfo info(Class serialClass) { + return info(serialClass, -1); } - /** ArrayDeque and PriorityQueue do not override equals, so round-trips are compared by iteration content. */ - private static void assertSameContent(Iterable expected, Iterable actual) { - assertIterableEquals(expected, actual); + /** A FilterInfo describing an array class of the given length. */ + private static ObjectInputFilter.FilterInfo info(Class serialClass, long arrayLength) { + return new ObjectInputFilter.FilterInfo() { + @Override + public Class serialClass() { + return serialClass; + } + + @Override + public long arrayLength() { + return arrayLength; + } + + @Override + public long depth() { + return 1; + } + + @Override + public long references() { + return 1; + } + + @Override + public long streamBytes() { + return 0; + } + }; } @Test @@ -115,14 +150,33 @@ public void testFilterAllowsNonDeniedClassesRoundTrip() { // ArrayDeque is unregistered and Serializable, so it travels through the java-serialization bridge itself. ArrayDeque deque = new ArrayDeque<>(Arrays.asList("a", "b", "c")); - assertSameContent(deque, (ArrayDeque) roundTrip(conf, deque)); + assertIterableEquals(deque, (Iterable) roundTrip(conf, deque)); } @Test public void testUnsetFilterKeyKeepsUnfilteredBehavior() { // No filter key in the conf at all: PriorityQueue must round-trip like it did before the filter existed. PriorityQueue original = new PriorityQueue<>(Arrays.asList(5, 4, 6)); - assertSameContent(original, (PriorityQueue) roundTrip(bridgeConf(null), original)); + assertIterableEquals(original, (Iterable) roundTrip(bridgeConf(null), original)); + } + + @Test + public void testWildcardDepthCoversDirectMembersAndSubpackages() { + // '.*' denies direct package members only: PriorityQueue (member of java.util) is rejected... + Map shallow = bridgeConf("!java.util.*"); + assertRejectedOnRead(shallow, new PriorityQueue<>(Arrays.asList(3, 1, 2))); + + // ...while classes in subpackages keep round-tripping through the bridge: java.util.regex.Pattern and + // java.util.logging.Level are unregistered, non-trivial, Serializable, and their java-serialized graphs stay + // inside java.lang for fields, so the pass/fail outcome is decided by their own package. + Pattern compiled = (Pattern) roundTrip(shallow, Pattern.compile("bridge-wildcard-probe")); + assertEquals("bridge-wildcard-probe", compiled.pattern()); + assertEquals(Level.WARNING, roundTrip(shallow, Level.WARNING)); + + // '**' also covers subpackages, so both depths are rejected. + Map recursive = bridgeConf("!java.util.**"); + assertRejectedOnRead(recursive, new PriorityQueue<>(Arrays.asList(3, 1, 2))); + assertRejectedOnRead(recursive, Pattern.compile("bridge-wildcard-probe")); } @Test @@ -131,46 +185,35 @@ public void testInvalidPatternFailsFastAtKryoConstruction() { // are ignored, not rejected. for (String invalid : Arrays.asList("!", "maxbytes=not-a-number")) { Map conf = bridgeConf(invalid); - RuntimeException ex = assertThrows(RuntimeException.class, () -> new KryoValuesSerializer(conf)); + IllegalArgumentException ex = assertThrows(IllegalArgumentException.class, () -> new KryoValuesSerializer(conf)); assertTrue(ex.getMessage().contains(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER), "error must name the offending config key: " + ex.getMessage()); } } @Test - public void testDefaultsYamlFilterParsesAndDeniesGadgetPackage() { - Map defaults = Utils.readDefaultConfig(); - Object filterSpec = defaults.get(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER); - assertNotNull(filterSpec, "conf/defaults.yaml must define a default " - + Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER); - // The default pattern must parse (bad patterns throw IllegalArgumentException here). - ObjectInputFilter.Config.createFilter((String) filterSpec); - - // Through an actual round-trip: a class in a deny-listed package is rejected on read... - assertRejectedOnRead(defaultsBridgeConf(), new SimulatedGadget()); - // ...while ordinary JDK collections keep round-tripping under the same default. - PriorityQueue queue = new PriorityQueue<>(Arrays.asList(5, 4, 6)); - assertSameContent(queue, (PriorityQueue) roundTrip(defaultsBridgeConf(), queue)); + public void testSamplePatternRejectsDenyListedJdkClasses() { + // Parsing the full sample also syntax-checks the third-party gadget entries, whose classes are not on the + // classpath and thus cannot be asserted as Class objects. + ObjectInputFilter sample = ObjectInputFilter.Config.createFilter(SAMPLE_PATTERN); + assertEquals(ObjectInputFilter.Status.REJECTED, sample.checkInput(info(BadAttributeValueExpException.class))); + assertEquals(ObjectInputFilter.Status.REJECTED, sample.checkInput(info(java.rmi.MarshalledObject.class))); } @Test - public void testDefaultFilterRejectsCommonsCollections3Comparators() { - // CC3's TransformingComparator gadget chain lives in the comparators package (the functors pair alone is not enough). - assertRejectedOnRead(defaultsBridgeConf(), new org.apache.commons.collections.comparators.SimulatedGadget()); + public void testSecurityDocCarriesTheSamplePattern() throws IOException { + // Surefire runs with the module directory as working directory, so the repo-root docs are one level up. + Path securityDoc = Paths.get("..", "docs", "SECURITY.md").toAbsolutePath().normalize(); + assertTrue(Files.exists(securityDoc), "docs/SECURITY.md not found at " + securityDoc); + String doc = Files.readString(securityDoc, StandardCharsets.UTF_8); + assertTrue(doc.contains(SAMPLE_PATTERN), "docs/SECURITY.md must carry this test's sample pattern verbatim"); } @Test - public void testDefaultFilterRejectsSubpackagesOfRecursiveWildcardEntries() { - // Wildcard depth matters: 'pkg.**' denies subpackages too; 'pkg.*' does not (c3p0.impl sits under the - // shipped '!com.mchange.v2.c3p0.**' entry). - assertRejectedOnRead(defaultsBridgeConf(), new com.mchange.v2.c3p0.impl.SimulatedGadget()); - } - - @Test - public void testDefaultFilterEnforcesMaxBytesLimit() { - Map conf = defaultsBridgeConf(); - // Every array read re-invokes the filter, so a stream built from many small arrays makes the cumulative - // streamBytes() limit bite mid-deserialization (one huge primitive payload would not re-invoke the filter). + public void testSamplePatternEnforcesMaxBytesLimit() { + // ~11MB of heap churn per run: the payload must exceed the pattern's maxbytes for the cumulative limit to bite + // mid-deserialization (every array read re-invokes the filter, so many small arrays make streamBytes add up). + Map conf = bridgeConf(SAMPLE_PATTERN); ArrayDeque big = new ArrayDeque<>(); for (int i = 0; i < 11000; i++) { big.add(new byte[1024]); @@ -178,10 +221,91 @@ public void testDefaultFilterEnforcesMaxBytesLimit() { KryoValuesSerializer serializer = new KryoValuesSerializer(conf); KryoValuesDeserializer deserializer = new KryoValuesDeserializer(conf); byte[] bytes = serializer.serialize(Collections.singletonList(big)); - assertTrue(bytes.length > DEFAULT_MAX_BYTES, "payload must exceed the default limit, was " + bytes.length); + assertTrue(bytes.length > SAMPLE_MAX_BYTES, "payload must exceed the maxbytes limit, was " + bytes.length); RuntimeException ex = assertThrows(RuntimeException.class, () -> deserializer.deserialize(bytes)); - assertTrue(hasCause(ex, InvalidClassException.class), + assertTrue(Utils.exceptionCauseIsInstanceOf(InvalidClassException.class, ex), "expected the maxbytes rejection in the cause chain, got: " + ex); } + + @Test + public void testMaxArrayLimitRejectsOversizedArray() { + // One big array can ride past the byte cap, because an array passes the filter before its contents are read; + // maxarray is what bounds the allocation itself. + Map conf = bridgeConf("maxarray=1024"); + ArrayDeque payload = new ArrayDeque<>(); + payload.add(new byte[4096]); + assertRejectedOnRead(conf, payload); + } + + @Test + public void testOversizedDeclaredLengthRejectedOnBufferedInput() { + // A fixed-width prefix (Output.writeInt and Input.readInt are symmetric 4-byte reads) declares a thousand + // bytes where only one follows; the mismatch fails up front, before the new byte[len] allocation can balloon. + Output out = new Output(16); + out.writeInt(1000); + out.writeByte(0); + SerializableSerializer serializer = new SerializableSerializer(); + KryoException ex = assertThrows(KryoException.class, + () -> serializer.read(null, new Input(out.toBytes()), Object.class)); + assertTrue(ex.getMessage().contains("declared: 1000"), + "error should name the declared length, got: " + ex.getMessage()); + } + + @Test + public void testStreamBackedInputIsExemptFromUpperBoundGuard() throws IOException { + // Stream-backed input is not length-checked: at prefix-read time the stream may have delivered only part of + // the value, with the rest still arriving, so an upper bound there would reject well-formed input and only + // the negative-length check applies. + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + try (ObjectOutputStream oos = new ObjectOutputStream(bos)) { + oos.writeObject(new byte[100000]); + } + byte[] payload = bos.toByteArray(); + + Output out = new Output(4096, Integer.MAX_VALUE); + out.writeInt(payload.length); + out.writeBytes(payload); + // Small buffer on purpose: after the 4-byte prefix the buffer holds fewer bytes than declared. + Input streamBacked = new Input(new ByteArrayInputStream(out.toBytes()), 1024); + byte[] result = (byte[]) new SerializableSerializer().read(null, streamBacked, Object.class); + // The declared length counted the java-serialization framing; the object that comes back is the original array. + assertEquals(100000, result.length); + } + + @Test + public void testNegativeDeclaredLengthRejected() { + // An all-ones 4-byte length prefix decodes as -1: no legitimate writer produces a negative length, buffered or streamed. + SerializableSerializer serializer = new SerializableSerializer(); + KryoException ex = assertThrows(KryoException.class, + () -> serializer.read(null, new Input(new byte[]{(byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF}), + Object.class)); + assertTrue(ex.getMessage().contains("-1"), "error should name the negative length, got: " + ex.getMessage()); + } + + @Test + public void testMergeWithExistingReturnsConfiguredWhenNoExistingFilter() { + ObjectInputFilter configured = ObjectInputFilter.Config.createFilter("!java.util.PriorityQueue"); + assertSame(configured, SerializableSerializer.mergeWithExisting(configured, null)); + } + + @Test + public void testMergeWithExistingRejectsClassDeniedByEitherFilter() { + ObjectInputFilter configured = ObjectInputFilter.Config.createFilter("!java.util.PriorityQueue"); + ObjectInputFilter existing = ObjectInputFilter.Config.createFilter("!java.util.ArrayDeque"); + ObjectInputFilter merged = SerializableSerializer.mergeWithExisting(configured, existing); + // The configured filter's denial survives the merge... + assertEquals(ObjectInputFilter.Status.REJECTED, merged.checkInput(info(PriorityQueue.class))); + // ...and the existing (e.g. JVM-wide) filter's denial is not replaced by it. + assertEquals(ObjectInputFilter.Status.REJECTED, merged.checkInput(info(ArrayDeque.class))); + } + + @Test + public void testMergeWithExistingEnforcesTighterLimit() { + ObjectInputFilter configured = ObjectInputFilter.Config.createFilter("maxarray=1000"); + ObjectInputFilter existing = ObjectInputFilter.Config.createFilter("maxarray=10"); + ObjectInputFilter merged = SerializableSerializer.mergeWithExisting(configured, existing); + // 500 fits the configured limit but exceeds the existing one; the merge keeps the tighter bound. + assertEquals(ObjectInputFilter.Status.REJECTED, merged.checkInput(info(byte[].class, 500))); + } }