This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new 415212892 Harden the java serialization fallback bridge with a JEP-290
serial filter (#9075)
415212892 is described below
commit 415212892e73b1b63d3909b48027932668f4694a
Author: L1nq <[email protected]>
AuthorDate: Tue Sep 8 21:48:21 2026 +0800
Harden the java serialization fallback bridge with a JEP-290 serial filter
(#9075)
* 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.
* 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.
---
docs/SECURITY.md | 14 +
docs/Serialization.md | 6 +
storm-client/src/jvm/org/apache/storm/Config.java | 18 ++
.../storm/serialization/DefaultKryoFactory.java | 27 +-
.../serialization/SerializableSerializer.java | 43 +++
.../apache/storm/validation/ConfigValidation.java | 25 ++
.../jvm/org/apache/storm/TestConfigValidate.java | 24 ++
.../SerializableSerializerFilterTest.java | 311 +++++++++++++++++++++
8 files changed, 466 insertions(+), 2 deletions(-)
diff --git a/docs/SECURITY.md b/docs/SECURITY.md
index 03c71993c..36154271f 100644
--- a/docs/SECURITY.md
+++ b/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 [...]
+
+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.Method
[...]
+```
+
+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 [...]
+
+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
diff --git a/docs/Serialization.md b/docs/Serialization.md
index 8f87ba6b0..b8701e8aa 100644
--- a/docs/Serialization.md
+++ b/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- [...]
+
+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.
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java
b/storm-client/src/jvm/org/apache/storm/Config.java
index 4e3dbe061..53f8eefdf 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/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.
*/
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 ef00ea47f..6ed3be282 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 @@ package org.apache.storm.serialization;
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 class DefaultKryoFactory implements IKryoFactory {
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 class DefaultKryoFactory implements IKryoFactory {
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 f17689cb6..042dceade 100644
---
a/storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java
+++
b/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 class SerializableSerializer extends
Serializer<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;
+ }
}
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 7d59b4a2e..4ebef6e43 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 @@ package org.apache.storm.validation;
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 class ConfigValidation {
}
}
+ /**
+ * 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/org/apache/storm/TestConfigValidate.java
b/storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
index 32bb14338..6105bc987 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 class TestConfigValidate {
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.
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 000000000..4c4732ea6
--- /dev/null
+++
b/storm-client/test/jvm/org/apache/storm/serialization/SerializableSerializerFilterTest.java
@@ -0,0 +1,311 @@
+/**
+ * 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 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 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.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. 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 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.Method
[...]
+
+ /**
+ * 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<String, Object> bridgeConf(String filterSpec) {
+ Map<String, Object> 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;
+ }
+
+ private Object roundTrip(Map<String, Object> 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<String, Object> 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(Utils.exceptionCauseIsInstanceOf(InvalidClassException.class, ex),
+ "expected the JEP-290 filter rejection in the cause chain,
got: " + ex);
+ }
+
+ /** 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);
+ }
+
+ /** 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
+ public void testFilterRejectsDeniedClassOnDeserialization() {
+ Map<String, Object> conf = bridgeConf("!java.util.PriorityQueue");
+
+ PriorityQueue<Integer> original = new PriorityQueue<>(Arrays.asList(3,
1, 2));
+ assertRejectedOnRead(conf, original);
+ }
+
+ @Test
+ public void testFilterAllowsNonDeniedClassesRoundTrip() {
+ Map<String, Object> conf = bridgeConf("!java.util.PriorityQueue");
+
+ // HashMap has a dedicated kryo serializer: ordinary payloads must
keep round-tripping.
+ HashMap<String, Integer> 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<String> deque = new ArrayDeque<>(Arrays.asList("a", "b",
"c"));
+ assertIterableEquals(deque, (Iterable<String>) 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<Integer> original = new PriorityQueue<>(Arrays.asList(5,
4, 6));
+ assertIterableEquals(original, (Iterable<Integer>)
roundTrip(bridgeConf(null), original));
+ }
+
+ @Test
+ public void testWildcardDepthCoversDirectMembersAndSubpackages() {
+ // '.*' denies direct package members only: PriorityQueue (member of
java.util) is rejected...
+ Map<String, Object> 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<String, Object> recursive = bridgeConf("!java.util.**");
+ assertRejectedOnRead(recursive, new PriorityQueue<>(Arrays.asList(3,
1, 2)));
+ assertRejectedOnRead(recursive,
Pattern.compile("bridge-wildcard-probe"));
+ }
+
+ @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<String, Object> conf = bridgeConf(invalid);
+ 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 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 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 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<String, Object> conf = bridgeConf(SAMPLE_PATTERN);
+ ArrayDeque<byte[]> 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 > SAMPLE_MAX_BYTES, "payload must exceed the
maxbytes limit, was " + bytes.length);
+
+ RuntimeException ex = assertThrows(RuntimeException.class, () ->
deserializer.deserialize(bytes));
+
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<String, Object> conf = bridgeConf("maxarray=1024");
+ ArrayDeque<byte[]> 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)));
+ }
+}
