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

[L1nq0]

Closes#9073

Upgrade note: no default pattern ships with this change. The key is unset in conf/defaults.yaml and the fallback bridge behaves exactly as on master unless a topology sets it. The recommended pattern lives in docs/SECURITY.md as a block the operator opts into. For operators who also run a JVM-wide -Djdk.serialFilter: when the topology key is set, the per-stream filter is merged with the process-wide one (ObjectInputFilter.merge), so a reject from either side still rejects and the operator's own filter keeps applying on this path.

What this adds

A new config key, topology.fall.back.on.java.serialization.filter, holding a JEP-290 filter pattern for the java serialization fallback bridge. The pattern is parsed once at kryo construction, and an invalid pattern is rejected at topology submission: a ConfigValidation validator runs it through ObjectInputFilter.Config.createFilter, so a typo fails on nimbus with an IllegalArgumentException naming the key instead of producing a supervisor restart loop. SerializableSerializer installs the filter on every ObjectInputStream it opens, merged with any JVM-wide serial filter as above. The filter is topology-scoped and also covers programmatic construction such as local mode, which a JVM-wide filter in worker.childopts cannot reach.

The filter covers the default factory's fallback bridge. A custom topology.kryo.factory is on its own, since SerializationFactory only ever gets back a finished Kryo with no way to reach a third-party factory's fallback path; DefaultStateSerializer, the pre-kryo state escape hatch, is not covered either.

The recommended pattern

docs/SECURITY.md carries the block: the gadget-namespace deny entries (commons-collections 3/4 functors and comparators, beanutils, xalan external and JDK-internal, rowset, c3p0, groovy closures, plus BadAttributeValueExpException, AnnotationInvocationHandler, com.sun.jndi, java.rmi, clojure, commons-fileupload, bsh, org.python, org.jboss) followed by maxdepth=64;maxrefs=2097152;maxarray=1048576;maxbytes=10485760. It carries an explicit not-exhaustive note and is worded per JEP-290's allow-list-first guidance: a starting point for an operator who has turned the fallback on and knows which classes cross it, not a security boundary. Payloads that legitimately need bigger arrays or sizes should raise maxarray/maxbytes accordingly.

On the size limits: maxbytes counts per object, not per tuple, because a new ObjectInputStream is opened for each value. It can also be overshot by a single large array, since the filter sees the array before reading its contents; that is the reason the sample also carries maxarray.

Hardening in the bridge itself

SerializableSerializer.read() no longer allocates from an unvalidated length field. A negative declared length is rejected on any input, and on buffered input (which is what the tuple path uses) a declared length that exceeds the bytes remaining in the input is rejected before the allocation. Stream-backed programmatic use of the deserializer is exempt from the upper bound, because a stream may still deliver the declared bytes after the buffer runs out (examples/storm-loadgen feeds it an Input over a request InputStream). The filter itself is passed to KryoSerializableDefault at construction and held in a private final field; there is no post-construction mutator.

Tests

15 cases in SerializableSerializerFilterTest, most running end to end through KryoValuesSerializer/KryoValuesDeserializer. They cover reject and allow round-trips on both wildcard depths (pkg.* matches direct members, pkg.** also matches subpackages) on JDK classes that really go through the bridge, the unset-key no-op, invalid patterns failing fast, the sample pattern checked against deny entries whose classes exist on the classpath, the maxbytes limit across many small arrays, and the maxarray limit. The declared-length guard gets three cases: oversized and negative lengths on buffered input, and the stream-backed exemption. A separate test asserts the sample pattern appears verbatim in docs/SECURITY.md, and the merge helper gets unit tests for the null case, a denial from either side, and the tighter limit winning. TestConfigValidate additionally covers submit-time rejection of invalid patterns with the key named.

Merge order: this is best merged after #9076, so what lands on master is the composed behaviour: a payload the filter rejects is dropped and counted as a deserializationFailure rather than killing the receiving worker. I will rebase onto #9076 once it merges.

[rzo1]

Thanks for this — and for the thorough writeup and the issue thread beforehand. I built the branch and ran the new tests locally on JDK 25: 8/8 pass, and CI is green.

The mechanism is right, and I want to say that up front: topology-scoped rather than JVM-wide, parsed once at kryo construction so a bad pattern fails fast, installed per ObjectInputStream, null/empty preserving the old behaviour, and the no-arg SerializableSerializer constructor kept for compatibility. That is the shape I would have asked for.

My objections are about the default, not the code.

1. Please don't ship the deny-list as the conf/defaults.yaml default

This is a behavioural break in a patch release for every cluster running topology.fall.back.on.java.serialization: true — new class rejections and a 10 MB cap on a path that has no cap today. You flagged this yourself in the description; I'd like to take you up on the offer and keep option (2) pure.

Ship the mechanism with an empty default, and put the recommended pattern in docs/SECURITY.md as a copy-paste block operators can adopt deliberately. Point 2 is the real argument for this.

2. As a security default, the deny-list is materially incomplete

Missing entry points that are on the JDK or the Storm worker classpath, so reachable on a stock cluster:

• javax.management.BadAttributeValueExpException — the trigger for a large share of published chains
• sun.reflect.annotation.AnnotationInvocationHandler
• com.sun.jndi.**, java.rmi.**
• clojure.** — Storm ships Clojure in the worker classpath
• org.apache.commons.fileupload.**, bsh.**, org.python.**, org.jboss.**

A twelve-entry list that stops the names we happened to think of, shipped as a default, buys false confidence: an operator reads "hardened" and stops looking. JEP-290's own guidance is allow-list first. That is a fine thing to document as a starting point for someone who has consciously enabled the fallback and knows their payload classes; it is not a good thing to enable silently on their behalf. If a deny-list does ship as the default it needs a much wider set plus an explicit "not exhaustive" note.

3. The filter only exists in DefaultKryoFactory, but defaults.yaml and the docs present it as unconditional

SerializationFactory.getKryo() (SerializationFactory.java:56) instantiates whatever topology.kryo.factory names. A custom IKryoFactory — a documented extension point — silently gets no filter even with the key set.

Either install it in SerializationFactory.getKryo after the kryoFactory.getKryo(conf) call, or state the DefaultKryoFactory-only scope in the Config javadoc and both doc pages.

4. Validate at submit time, not at worker start

@IsString only checks the type, so a bad pattern currently fails every worker at kryo construction and turns into a supervisor restart loop, with nothing surfaced at submission. A ConfigValidation validator that calls ObjectInputFilter.Config.createFilter would let nimbus reject it on submit — much better than discovering it per-worker.

Also RuntimeException → IllegalArgumentException in DefaultKryoFactory.getJavaSerializationFilter; the message itself is good.

5. Doc scoping is too broad

topology.fall.back.on.java.serialization: true also loosens DefaultStateSerializer — see docs/State-checkpointing.md:230, where it is documented as the escape hatch for state that predates kryo registration. That path builds its own new Kryo(...) rather than going through DefaultKryoFactory, so the filter never reaches it. (Lower risk, since unregistered classes there go through FieldSerializer and no readObject runs — but the new prose reads as if the whole config switch is now filtered.) Please scope the wording in Serialization.md and SECURITY.md to the tuple fallback bridge.

Smaller things

• Test fixtures squat third-party package names (org.apache.commons.collections.functors, ...comparators, com.mchange.v2.c3p0.impl) under storm-client/test/jvm. I checked and none of those artifacts is on storm-client's test classpath today, so there's no split package yet — but it's a trap for whoever adds one. Suggestion: keep one end-to-end round-trip on a Storm-owned class with a topology-scoped pattern (which the !java.util.PriorityQueue tests already do nicely), and cover the shipped defaults by implementing ObjectInputFilter.FilterInfo in the test and calling checkInput directly. Same coverage, no fake packages.
• KryoSerializableDefault.setJavaSerializationFilter is a public setter for a security control. A constructor parameter would be harder to get wrong.
• testDefaultFilterEnforcesMaxBytesLimit allocates ~11 MB per run. It's fine — the whole class runs in 0.26s — just noting it in case it ever moves somewhere hotter.

Merge order

Worth landing this after #9076. On master today this PR alone turns a gadget payload into a worker kill: the filter's InvalidClassException propagates out of SerializableSerializer.read and up through DeserializingConnectionCallback.recv() into Utils.handleUncaughtException. With #9076 in first, InvalidClassException extends IOException, so a filtered payload is dropped and counted instead — which is the behaviour this PR's documentation implies but doesn't yet get on its own.

[L1nq0]

Thanks for the thorough review, and for saying up front that the mechanism is the shape you'd have asked for. Taking the points in order:

1. Empty default: agreed, keeping option (2) pure. The defaults.yaml entry goes away, the Config javadoc keeps documenting the key, and the recommended pattern moves into docs/SECURITY.md as a copy-paste block operators adopt deliberately.

2. Deny-list completeness: agreed on the false-confidence read; a twelve-entry list of the names we happened to think of is not a security boundary. The SECURITY.md block will fold in the entries you listed (BadAttributeValueExpException, AnnotationInvocationHandler, com.sun.jndi, java.rmi, clojure, commons-fileupload, bsh, org.python, org.jboss) alongside the original set, carry an explicit not-exhaustive note, and be worded per JEP-290's allow-list-first guidance: a starting point for someone who has consciously enabled the fallback and knows their payload classes, not a boundary.

3. Factory scope: I'd rather close the gap than footnote it, so the plan is to move the installation into SerializationFactory.getKryo(), right after the kryoFactory.getKryo(conf) call, so a custom IKryoFactory gets the filter instead of silently ignoring the key; the Config javadoc will state where the filter applies. If that turns out to fight the extension point in some way I'm not seeing yet, I'll say so on this thread and fall back to documenting the DefaultKryoFactory-only scope explicitly.

4. Submit-time validation: agreed, a nimbus rejection at submit is strictly better than a supervisor restart loop. I'll add a ConfigValidation validator that runs the pattern through ObjectInputFilter.Config.createFilter, and getJavaSerializationFilter will throw IllegalArgumentException instead of RuntimeException.

5. Doc scoping: agreed. The prose in Serialization.md and SECURITY.md will name the tuple fallback bridge specifically and state that DefaultStateSerializer, the escape hatch for pre-kryo state, is not covered by this filter.

Smaller things, all accepted: the test fixtures stop squatting third-party package names; coverage keeps the end-to-end round-trips on Storm-owned classes with topology-scoped patterns, and checks the shipped pattern by implementing ObjectInputFilter.FilterInfo in the test and calling checkInput directly. setJavaSerializationFilter becomes a constructor parameter instead of a public setter. The maxbytes allocation stays as is, noted.

Merge order: agreed, and thanks for laying out the interaction. Once #9076 lands I'll rebase this on it, so what reviewers see is the composed behaviour: a filtered payload dropped and counted rather than killing the worker.

[reiabreu]

Hey folks, I'll try to provide some feedback over the weekend. Thank you

[reiabreu]

hey!
@rzo1 already covered all important points and it's a +1 on his review.

I did a sweep with the help of an LLM and this is what is outstanding:

1. Add maxarray (and maxdepth/maxrefs) to the sample pattern, not just maxbytes. maxbytes only counts bytes already read, so one big array still gets allocated before the limit kicks in. The class deny-list does nothing for this.
2. Say that this filter replaces a JVM-wide -Djdk.serialFilter, it does not add to it. The default factory just returns the new filter, so a filter set in worker.childopts is quietly dropped on this path. Either merge with the existing one via ObjectInputFilter.merge(...), or document it.
3. Factory scope (rzo1 3rd point): fix it by documenting, not by moving the install into SerializationFactory.getKryo(). The filter can only go where the ObjectInputStream is opened, which is inside SerializableSerializer. That works for the default factory because it's our code and it builds that serializer. A custom topology.kryo.factory is a black box: SerializationFactory only gets back a finished Kryo, with no way to reach its fallback path. So a generic install can't cover custom factories. Best to just state in the Config javadoc and both docs that the filter covers the default factory's fallback bridge, and a custom factory is on its own.
4. Minor: maxbytes is per object, not per tuple (a new ObjectInputStream is opened per value). Worth saying so in the docs.

[L1nq0]

@reiabreu Thanks for the sweep; all four points are in the revision, pushed as 54e4926.

• maxarray (and maxdepth/maxrefs): added to the sample pattern in docs/SECURITY.md, now maxdepth=64;maxrefs=2097152;maxarray=1048576;maxbytes=10485760. maxarray sits well below maxbytes for the reason you describe: the array is checked when it is created, before its contents are read, so the byte limit alone would miss one big array. The docs next to the block now cover both properties, per-object counting and the overshoot case.

• Merge vs replace: I took the merge branch of your either/or. The install now reads the stream's current filter and combines the two via ObjectInputFilter.merge, so a JVM-wide -Djdk.serialFilter set in worker.childopts is no longer quietly dropped on this path: if either filter rejects, the merged one rejects. GGraziadei raised the same point inline, with a JDK 25 check that matches.

• Factory scope: your read on the extension point settled the question I'd left open on rzo1's thread (I had planned to try moving the install into SerializationFactory.getKryo()). You're right that it can't work generically: the factory only ever hands back a finished Kryo, with no way to reach a third-party factory's fallback path. Went with the pre-announced fallback: the Config javadoc and both docs now state the filter covers the default factory's fallback bridge, a custom topology.kryo.factory is on its own, and DefaultStateSerializer is not covered.

• Per-object maxbytes: documented as such, in Serialization.md, SECURITY.md and the Config javadoc, together with the best-effort caveat above.

Beyond your list, the same revision also took GGraziadei's inline points (constructor-injected filter instead of a public setter, and a length guard in SerializableSerializer.read() so the attacker-sized buffer is no longer allocated from an unvalidated length field on the tuple path), and rzo1's earlier points (empty default, submit-time validation of the pattern, Storm-owned test classes).

[rzo1]

lgtm

[rzo1]

Thanks for the contribution @L1nq0 - thanks for the reviews!