diff --git a/docs/SECURITY.md b/docs/SECURITY.md index 03c71993c81..36154271f13 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 covered either. An empty or unset value leaves the bridge unfiltered, as before. + +When the bridge is used, the configured filter is merged with any JVM-wide `jdk.serialFilter` (e.g. set in `worker.childopts`) rather than replacing it, so enabling this setting can only tighten stream filtering, never loosen an operator-set allow-list. + +The pattern below is a starting point for operators who enable the fallback anyway and know which classes their payloads contain. It is not a security boundary, and the deny-list is not exhaustive. Following [JEP-290](https://openjdk.org/jeps/290) guidance, an allow-list of the exact classes a topology actually needs is the preferred design; deny-lists only block known gadget classes. + +```yaml +topology.fall.back.on.java.serialization.filter: "!org.apache.commons.collections.functors.*;!org.apache.commons.collections.comparators.*;!org.apache.commons.collections4.functors.*;!org.apache.commons.collections4.comparators.*;!org.apache.commons.beanutils.*;!org.apache.xalan.xsltc.trax.*;!com.sun.org.apache.xalan.internal.**;!com.sun.rowset.*;!com.sun.org.apache.rowset.internal.*;!com.mchange.v2.c3p0.**;!org.codehaus.groovy.runtime.ConvertedClosure;!org.codehaus.groovy.runtime.MethodClosure;!javax.management.BadAttributeValueExpException;!sun.reflect.annotation.AnnotationInvocationHandler;!com.sun.jndi.**;!java.rmi.**;!clojure.**;!org.apache.commons.fileupload.**;!bsh.**;!org.python.**;!org.jboss.**;maxdepth=64;maxrefs=2097152;maxarray=1048576;maxbytes=10485760" +``` + +The size limits need a caveat. `maxbytes` counts stream bytes per object, not per tuple, and it is not exact: the filter sees a primitive array when the array is created, before its contents are read, so one big array can slip past the byte cap. `maxarray`, set well below `maxbytes`, is what bounds that case; `maxdepth` and `maxrefs` bound the depth and reference count of the graph. Workloads that legitimately use large arrays should raise these limits. On buffered input the bridge also refuses a declared value length larger than the bytes left in the frame before allocating it; streaming programmatic use of the deserializer is outside this guard. + +An invalid pattern is rejected when the topology is submitted, and otherwise fails worker startup. + For tuple encryption, use TLS-based transport encryption (`storm.messaging.netty.tls.enable`) instead of the deprecated `BlowfishTupleSerializer`, which uses a 64-bit block cipher vulnerable to birthday attacks. ### Log Cleanup diff --git a/docs/Serialization.md b/docs/Serialization.md index 8f87ba6b043..b8701e8aacc 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-wide `-Djdk.serialFilter`, this filter is topology-scoped and also applies when the deserializer is constructed programmatically, e.g. in local mode. When both are present, the two filters are merged, so both apply. + +This only filters the fallback path of the default kryo factory. A custom `topology.kryo.factory` gets no filter from Storm and must set its own; `DefaultStateSerializer`, used for worker state before kryo is set up, is unfiltered as well. + +There is no default value; see `docs/SECURITY.md` for a sample pattern. `maxbytes` is a per-object budget (one `ObjectInputStream` per value, not per tuple) and can be overshot by a single large array, since the array is measured before its contents are read. On buffered input the bridge rejects a declared length that exceeds the bytes remaining in the frame before allocating; streaming programmatic use is not covered. An empty or unset value leaves the bridge unfiltered, as before. + ### Tuple compression For inter-worker (remote) traffic, Storm can optionally compress serialized tuples with [Zstandard](https://facebook.github.io/zstd/) before they are sent over the network. This is intended for one specific scenario: components that emit **large** payloads to a remote worker, where the bytes saved on the wire outweigh the CPU cost of compression. A good example is a spout that emits entire lines of text to a downstream bolt running on a different worker. diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index 4e3dbe061bf..53f8eefdf27 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 { */ @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. 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 ef00ea47f75..6ed3be28273 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; @@ -26,12 +27,29 @@ 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); 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 IllegalArgumentException("Invalid " + Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER + + " pattern: \"" + filterSpec + "\"", e); + } + } + @Override public void preRegister(Kryo k, Map conf) { } @@ -47,6 +65,11 @@ public void postDecorate(Kryo k, Map conf) { public static class KryoSerializableDefault extends Kryo { boolean override = false; + private final ObjectInputFilter javaSerializationFilter; + + KryoSerializableDefault(ObjectInputFilter javaSerializationFilter) { + this.javaSerializationFilter = javaSerializationFilter; + } public void overrideDefault(boolean value) { override = value; @@ -61,7 +84,7 @@ public Serializer getDefaultSerializer(Class type) { Util.className(type), Util.className(type) ); - return new SerializableSerializer(); + return new SerializableSerializer(javaSerializationFilter); } else { return super.getDefaultSerializer(type); } 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..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,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 { + /** + * Optional JEP-290 filter applied to each ObjectInputStream used for deserialization (null means unfiltered, + * as before). The filter itself is created once from + * {@link org.apache.storm.Config#TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_FILTER} by {@link DefaultKryoFactory}; + * instances returned by {@link ObjectInputFilter.Config#createFilter} are immutable and safe to share across streams. + */ + private final ObjectInputFilter serialFilter; + + public SerializableSerializer() { + this(null); + } + + public SerializableSerializer(ObjectInputFilter serialFilter) { + this.serialFilter = serialFilter; + } + @Override public void write(Kryo kryo, Output output, Object object) { ByteArrayOutputStream bos = new ByteArrayOutputStream(); @@ -43,14 +61,39 @@ public void write(Kryo kryo, Output output, Object object) { @Override public Object read(Kryo kryo, Input input, Class c) { int len = input.readInt(); + if (len < 0) { + throw new KryoException("Invalid java-serialized value length: " + len); + } + // For a buffer-backed Input the remaining bytes are known (position/limit), so a declared length larger than the + // bytes actually left is refused before the new byte[len] allocation; a stream-backed Input may still deliver the + // declared bytes later, so the upper bound is not checked there. + if (input.getInputStream() == null) { + int remaining = input.limit() - input.position(); + if (len > remaining) { + throw new KryoException("Declared java-serialized value length exceeds the input's remaining bytes " + + "(declared: " + len + ", remaining: " + remaining + ")"); + } + } byte[] ser = new byte[len]; input.readBytes(ser); ByteArrayInputStream bis = new ByteArrayInputStream(ser); try { ObjectInputStream ois = new ObjectInputStream(bis); + if (serialFilter != null) { + ois.setObjectInputFilter(mergeWithExisting(serialFilter, ois.getObjectInputFilter())); + } return ois.readObject(); } catch (Exception e) { throw new RuntimeException(e); } } + + /** + * Combines the configured filter with the stream's existing filter (a JVM-wide {@code jdk.serialFilter}, if any), so both + * the configured pattern and any process-wide filter apply to the stream: per JEP-290, + * {@link ObjectInputStream#setObjectInputFilter} overrides the process-wide filter for that stream unless the two are merged. + */ + static ObjectInputFilter mergeWithExisting(ObjectInputFilter configured, ObjectInputFilter existing) { + return existing != null ? ObjectInputFilter.merge(configured, existing) : configured; + } } 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/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 new file mode 100644 index 00000000000..4c4732ea65a --- /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.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} + * 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; + } + + 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(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 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")); + 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)); + 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 + 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); + 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 conf = bridgeConf(SAMPLE_PATTERN); + 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 > 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 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))); + } +}