GGraziadei commented on code in PR #9075:
URL: https://github.com/apache/storm/pull/9075#discussion_r3943810695
##########
storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java:
##########
@@ -47,6 +66,11 @@ public void postDecorate(Kryo k, Map<String, Object> conf) {
public static class KryoSerializableDefault extends Kryo {
boolean override = false;
+ private ObjectInputFilter javaSerializationFilter;
+
+ public void setJavaSerializationFilter(ObjectInputFilter filter) {
Review Comment:
Minor: this mutable setter (plus the non-final field) isn't needed —
`KryoSerializableDefault` is constructed at exactly one site (`getKryo`) and
the filter never changes afterwards. As written, any code that gets hold of the
`Kryo` instance (e.g. an `IKryoDecorator` registered via
`topology.kryo.decorators` receives it in `decorate`) can silently swap or null
out the filter after construction.
Since this is a security control, a constructor argument + `final` field
would make post-construction mutation impossible and avoid adding a public
mutator to the API.
##########
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"
Review Comment:
Two size-cap gaps worth closing or at least documenting, since
SECURITY.md/Serialization.md present `maxbytes=10485760` as the bridge's size
cap:
1. `maxbytes` does not bound a single large primitive array. Verified on JDK
25: a holder object wrapping one 30MB `byte[]` (3x the limit) **passes**
`createFilter("maxbytes=10485760")` and fully deserializes — the filter check
for the array fires at array creation, before the body bytes are counted into
`streamBytes`. Adding `maxarray=` to the pattern rejects it. (The test file's
own comment concedes this: "one huge primitive payload would not re-invoke the
filter".)
2. Independently of the filter, `SerializableSerializer.read()` does `int
len = input.readInt(); byte[] ser = new byte[len]` *before* the
`ObjectInputStream` is constructed, so the attacker-sized buffer is fully
allocated regardless of any filter.
Suggestion: append a `maxarray=` entry to this default pattern, and soften
the docs claim to describe `maxbytes` as best-effort rather than a hard cap.
##########
storm-client/src/jvm/org/apache/storm/serialization/SerializableSerializer.java:
##########
@@ -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);
Review Comment:
The stream-level filter set here **replaces** any operator-set JVM-wide
`-Djdk.serialFilter` on these streams instead of combining with it. Per JEP-290
semantics, `ObjectInputStream.setObjectInputFilter` overrides the process-wide
filter for that stream (verified empirically on JDK 25: with
`-Djdk.serialFilter='!java.util.PriorityQueue;*'` a `PriorityQueue` is rejected
on a plain `ObjectInputStream`, but deserializes once
`setObjectInputFilter(...)` is called with a filter that allows it).
So a cluster that is already hardened with a strict JVM-wide allowlist
would, after picking up this default, end up with a *weaker* filter on exactly
the bridge streams this PR is trying to protect — anything outside the
deny-list here that the operator's allowlist used to block now gets through.
Suggested fix: merge instead of replace, so the stream filter can only
tighten:
```java
ObjectInputFilter existing = ois.getObjectInputFilter();
ois.setObjectInputFilter(existing != null
? ObjectInputFilter.merge(serialFilter, existing)
: serialFilter);
```
##########
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<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;
+ }
+
+ /** conf assembled exactly like a worker's would be: defaults.yaml +
topology-level overrides. */
+ private Map<String, Object> defaultsBridgeConf() {
+ Map<String, Object> conf = new Config();
+ conf.putAll(Utils.readDefaultConfig());
+ conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, true);
+ 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(hasCause(ex, InvalidClassException.class),
+ "expected the JEP-290 filter rejection in the cause chain,
got: " + ex);
+ }
+
+ private static boolean hasCause(Throwable throwable, Class<? extends
Throwable> type) {
Review Comment:
Nit: `hasCause()` re-implements `Utils.exceptionCauseIsInstanceOf(Class,
Throwable)` (Utils.java), which this test can use directly:
```java
assertTrue(Utils.exceptionCauseIsInstanceOf(InvalidClassException.class,
ex), ...);
```
Deletes the private cause-walking loop with no behavior change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
