From c2850f76813e8570c8ab04c3539773cd8af26ff7 Mon Sep 17 00:00:00 2001 From: unknown Date: Thu, 22 Oct 2026 04:00:42 +0500 Subject: [PATCH] [WAYANG-835] Implement ObjectInputFilter allowlist for legacy object deserialization --- .../operators/ObjectFileSerialization.java | 174 ++++++++++++++++- .../ObjectFileSerializationTest.java | 179 ++++++++++++++++++ .../flink/compiler/criterion/WayangValue.java | 91 +++++---- .../compiler/criterion/WayangValueTest.java | 124 ++++++++++++ 4 files changed, 528 insertions(+), 40 deletions(-) create mode 100644 wayang-platforms/wayang-flink/src/test/java/org/apache/wayang/flink/compiler/criterion/WayangValueTest.java diff --git a/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/ObjectFileSerialization.java b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/ObjectFileSerialization.java index 46d2c152e..fe8a16cac 100644 --- a/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/ObjectFileSerialization.java +++ b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/ObjectFileSerialization.java @@ -25,14 +25,22 @@ import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.ObjectInputFilter; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.util.ArrayList; import java.util.Arrays; +import java.util.BitSet; import java.util.Collection; import java.util.Collections; +import java.util.Currency; +import java.util.Date; import java.util.List; +import java.util.Locale; +import java.util.Map; import java.util.Objects; +import java.util.Optional; +import java.util.UUID; /** * Utility methods that convert between Java objects and the on-disk representation used by {@link ObjectFileSink}. @@ -88,7 +96,7 @@ public static List deserializeChunk(byte[] payload, case JSON: return deserializeJson(payload, elementType); case LEGACY_JAVA_SERIALIZATION: - return deserializeLegacy(payload); + return deserializeLegacy(payload, elementType); default: throw new IllegalArgumentException("Unknown serialization mode: " + mode); } @@ -104,9 +112,10 @@ private static byte[] serializeLegacy(Object[] chunk, int validLength) throws IO } } - private static List deserializeLegacy(byte[] payload) throws IOException, ClassNotFoundException { + private static List deserializeLegacy(byte[] payload, Class elementType) throws IOException, ClassNotFoundException { try (ByteArrayInputStream bis = new ByteArrayInputStream(payload); ObjectInputStream ois = new ObjectInputStream(bis)) { + ois.setObjectInputFilter(createObjectInputFilter(elementType)); Object tmp = ois.readObject(); if (tmp == null) { return Collections.emptyList(); @@ -121,6 +130,167 @@ private static List deserializeLegacy(byte[] payload) throws IOException } } + /** + * Creates an {@link ObjectInputFilter} allowlist for legacy Java deserialization. + * + * @param elementType the expected element type (optional, may be {@code null}) + * @return an {@link ObjectInputFilter} that permits only expected and safe classes + */ + public static ObjectInputFilter createObjectInputFilter(Class elementType) { + ObjectInputFilter filter = filterInfo -> { + Class clazz = filterInfo.serialClass(); + if (clazz == null) { + return ObjectInputFilter.Status.UNDECIDED; + } + if (isAllowedClass(clazz, elementType)) { + return ObjectInputFilter.Status.ALLOWED; + } + return ObjectInputFilter.Status.REJECTED; + }; + + ObjectInputFilter globalFilter = ObjectInputFilter.Config.getSerialFilter(); + if (globalFilter != null) { + return ObjectInputFilter.merge(globalFilter, filter); + } + return filter; + } + + private static boolean isAllowedClass(Class clazz, Class elementType) { + Class current = clazz; + while (current.isArray()) { + current = current.getComponentType(); + } + if (current.isPrimitive()) { + return true; + } + + // Standard base object and scalar types + if (current == Object.class || + current == String.class || + current == Boolean.class || + current == Character.class || + current == Class.class || + isStandardNumber(current) || + Enum.class.isAssignableFrom(current) || + isStandardCharSequence(current)) { + return true; + } + + // Standard math + if (current.getName().startsWith("java.math.")) { + return true; + } + + // Standard date and time types + if (current.getName().startsWith("java.time.") || + (Date.class.isAssignableFrom(current) && + (current.getName().startsWith("java.util.") || current.getName().startsWith("java.sql.")))) { + return true; + } + + // Standard collections, maps, and common utilities + if (((Collection.class.isAssignableFrom(current) || + Map.class.isAssignableFrom(current) || + Map.Entry.class.isAssignableFrom(current)) && + current.getName().startsWith("java.util.")) || + current.getName().equals("java.util.CollSer") || + current.getName().startsWith("java.util.Collections$") || + current.getName().startsWith("java.util.Arrays$") || + current.getName().startsWith("java.util.ImmutableCollections$") || + current == UUID.class || + current == BitSet.class || + current == Locale.class || + current == Currency.class || + current == Optional.class) { + return true; + } + + // Wayang internal types (excluding custom Number, Collection, and Map subclasses) + if (current.getName().startsWith("org.apache.wayang.") && + !Number.class.isAssignableFrom(current) && + !Collection.class.isAssignableFrom(current) && + !Map.class.isAssignableFrom(current)) { + return true; + } + + // Scala tuples and basic collections (for Spark / Scala platform compatibility) + if (current.getName().startsWith("scala.Tuple") || + current.getName().startsWith("scala.collection.")) { + return true; + } + + // Expected element type if specified + if (elementType != null && !Object.class.equals(elementType)) { + Class targetComponent = elementType; + while (targetComponent.isArray()) { + targetComponent = targetComponent.getComponentType(); + } + if (targetComponent.isAssignableFrom(current)) { + return true; + } + if (current.getName().startsWith(targetComponent.getName() + "$")) { + return true; + } + Package pkg = targetComponent.getPackage(); + if (pkg != null && current.getName().startsWith(pkg.getName() + ".")) { + return true; + } + } + + // Configurable system property allowlist (comma/semicolon-separated patterns) + String configuredAllowlist = System.getProperty("wayang.basic.serialization.allowlist", + System.getProperty("wayang.serialization.allowlist")); + if (configuredAllowlist != null && !configuredAllowlist.trim().isEmpty()) { + String[] patterns = configuredAllowlist.split("[,;]"); + for (String pattern : patterns) { + if (matchesPattern(pattern.trim(), current.getName())) { + return true; + } + } + } + + return false; + } + + private static boolean isStandardNumber(Class clazz) { + return clazz == Number.class || + clazz == Byte.class || + clazz == Short.class || + clazz == Integer.class || + clazz == Long.class || + clazz == Float.class || + clazz == Double.class; + } + + private static boolean isStandardCharSequence(Class clazz) { + return clazz == String.class || + clazz == StringBuilder.class || + clazz == StringBuffer.class || + clazz == CharSequence.class; + } + + private static boolean matchesPattern(String pattern, String className) { + if (pattern.isEmpty()) { + return false; + } + if (pattern.equals("*") || pattern.equals("**")) { + return true; + } + if (pattern.endsWith(".**")) { + String prefix = pattern.substring(0, pattern.length() - 3); + return className.equals(prefix) || className.startsWith(prefix + "."); + } + if (pattern.endsWith(".*")) { + String prefix = pattern.substring(0, pattern.length() - 2); + if (className.startsWith(prefix + ".")) { + String remainder = className.substring(prefix.length() + 1); + return !remainder.contains("."); + } + return false; + } + return pattern.equals(className); + } + private static byte[] serializeJson(Object[] chunk, int validLength) throws IOException { Object[] payload = Arrays.copyOf(chunk, validLength); return OBJECT_MAPPER.writeValueAsBytes(payload); diff --git a/wayang-commons/wayang-basic/src/test/java/org/apache/wayang/basic/operators/ObjectFileSerializationTest.java b/wayang-commons/wayang-basic/src/test/java/org/apache/wayang/basic/operators/ObjectFileSerializationTest.java index 0cc3a830d..b581bb307 100644 --- a/wayang-commons/wayang-basic/src/test/java/org/apache/wayang/basic/operators/ObjectFileSerializationTest.java +++ b/wayang-commons/wayang-basic/src/test/java/org/apache/wayang/basic/operators/ObjectFileSerializationTest.java @@ -67,6 +67,144 @@ public void legacyRoundtrip() throws Exception { Assert.assertEquals(42, deserialized.value); } + @Test(expected = java.io.InvalidClassException.class) + public void legacyDeserializationRejectsDisallowedClass() throws Exception { + java.util.concurrent.atomic.AtomicBoolean disallowed = new java.util.concurrent.atomic.AtomicBoolean(true); + Object[] chunk = new Object[]{disallowed}; + + byte[] bytes = ObjectFileSerialization.serializeChunk( + chunk, + chunk.length, + ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION); + + // Deserializing with an unrelated elementType (or null) should be rejected by the filter + ObjectFileSerialization.deserializeChunk( + bytes, + ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION, + String.class); + } + + @Test + public void legacyDeserializationAllowedViaElementType() throws Exception { + java.util.concurrent.atomic.AtomicBoolean payload = new java.util.concurrent.atomic.AtomicBoolean(true); + Object[] chunk = new Object[]{payload}; + + byte[] bytes = ObjectFileSerialization.serializeChunk( + chunk, + chunk.length, + ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION); + + // When elementType explicitly specifies AtomicBoolean, it should be permitted + List result = ObjectFileSerialization.deserializeChunk( + bytes, + ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION, + java.util.concurrent.atomic.AtomicBoolean.class); + + Assert.assertEquals(1, result.size()); + Assert.assertTrue(((java.util.concurrent.atomic.AtomicBoolean) result.get(0)).get()); + } + + @Test + public void legacyDeserializationAllowedViaSystemProperty() throws Exception { + java.util.concurrent.atomic.AtomicBoolean payload = new java.util.concurrent.atomic.AtomicBoolean(true); + Object[] chunk = new Object[]{payload}; + + byte[] bytes = ObjectFileSerialization.serializeChunk( + chunk, + chunk.length, + ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION); + + String propKey = "wayang.basic.serialization.allowlist"; + String originalVal = System.getProperty(propKey); + try { + System.setProperty(propKey, "java.util.concurrent.atomic.**"); + List result = ObjectFileSerialization.deserializeChunk( + bytes, + ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION, + String.class); + Assert.assertEquals(1, result.size()); + Assert.assertTrue(((java.util.concurrent.atomic.AtomicBoolean) result.get(0)).get()); + } finally { + if (originalVal != null) { + System.setProperty(propKey, originalVal); + } else { + System.clearProperty(propKey); + } + } + } + + @Test + public void legacyDeserializationRejectsCustomNumberSubclass() throws Exception { + CustomNumberPayload.hookExecuted = false; + CustomNumberPayload payload = new CustomNumberPayload(123); + byte[] bytes = ObjectFileSerialization.serializeChunk( + new Object[]{payload}, 1, ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION); + + try { + ObjectFileSerialization.deserializeChunk( + bytes, ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION, String.class); + Assert.fail("Expected InvalidClassException"); + } catch (java.io.InvalidClassException e) { + // Expected + } + Assert.assertFalse("Custom readObject() hook should not execute", CustomNumberPayload.hookExecuted); + } + + @Test + public void legacyDeserializationRejectsCustomCollectionSubclass() throws Exception { + CustomCollectionPayload.hookExecuted = false; + CustomCollectionPayload payload = new CustomCollectionPayload<>(); + payload.add("item"); + byte[] bytes = ObjectFileSerialization.serializeChunk( + new Object[]{payload}, 1, ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION); + + try { + ObjectFileSerialization.deserializeChunk( + bytes, ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION, String.class); + Assert.fail("Expected InvalidClassException"); + } catch (java.io.InvalidClassException e) { + // Expected + } + Assert.assertFalse("Custom readObject() hook should not execute", CustomCollectionPayload.hookExecuted); + } + + @Test + public void legacyDeserializationRejectsCustomMapSubclass() throws Exception { + CustomMapPayload.hookExecuted = false; + CustomMapPayload payload = new CustomMapPayload<>(); + payload.put("k", "v"); + byte[] bytes = ObjectFileSerialization.serializeChunk( + new Object[]{payload}, 1, ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION); + + try { + ObjectFileSerialization.deserializeChunk( + bytes, ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION, String.class); + Assert.fail("Expected InvalidClassException"); + } catch (java.io.InvalidClassException e) { + // Expected + } + Assert.assertFalse("Custom readObject() hook should not execute", CustomMapPayload.hookExecuted); + } + + @Test + public void legacyDeserializationAllowsImmutableCollections() throws Exception { + List list = List.of("alpha", "beta"); + java.util.Set set = java.util.Set.of("gamma", "delta"); + java.util.Map map = java.util.Map.of("key1", 10, "key2", 20); + + Object[] chunk = new Object[]{list, set, map}; + byte[] bytes = ObjectFileSerialization.serializeChunk( + chunk, chunk.length, ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION); + + List result = ObjectFileSerialization.deserializeChunk( + bytes, ObjectFileSerializationMode.LEGACY_JAVA_SERIALIZATION, Object.class); + + Assert.assertEquals(3, result.size()); + Assert.assertEquals(list, result.get(0)); + Assert.assertEquals(set, result.get(1)); + Assert.assertEquals(map, result.get(2)); + } + private static class SerializablePayload implements Serializable { final String text; final int value; @@ -76,4 +214,45 @@ private SerializablePayload(String text, int value) { this.value = value; } } + + public static class CustomNumberPayload extends Number implements Serializable { + public static boolean hookExecuted = false; + private final int val; + + public CustomNumberPayload(int val) { + this.val = val; + } + + @Override + public int intValue() { return val; } + @Override + public long longValue() { return val; } + @Override + public float floatValue() { return val; } + @Override + public double doubleValue() { return val; } + + private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException { + hookExecuted = true; + in.defaultReadObject(); + } + } + + public static class CustomCollectionPayload extends java.util.ArrayList implements Serializable { + public static boolean hookExecuted = false; + + private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException { + hookExecuted = true; + in.defaultReadObject(); + } + } + + public static class CustomMapPayload extends java.util.HashMap implements Serializable { + public static boolean hookExecuted = false; + + private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException { + hookExecuted = true; + in.defaultReadObject(); + } + } } diff --git a/wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/compiler/criterion/WayangValue.java b/wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/compiler/criterion/WayangValue.java index 2fc0d6b5f..8a9bf8356 100644 --- a/wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/compiler/criterion/WayangValue.java +++ b/wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/compiler/criterion/WayangValue.java @@ -26,10 +26,9 @@ import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; -import java.io.ObjectInput; import java.io.ObjectInputStream; -import java.io.ObjectOutput; import java.io.ObjectOutputStream; +import org.apache.wayang.basic.operators.ObjectFileSerialization; /** * Implementation of {@link Value} of flink for use in Wayang */ @@ -42,12 +41,36 @@ public class WayangValue implements Value { private T data; + private Class type; + public WayangValue(){ this.data = null; + this.type = null; + } + + public WayangValue(Class type){ + this.data = null; + this.type = type; } public WayangValue(T element){ this.data = element; + if (element != null) { + this.type = (Class) element.getClass(); + } + } + + public WayangValue(T element, Class type){ + this.data = element; + this.type = type; + } + + public Class getType() { + return this.type; + } + + public void setType(Class type) { + this.type = type; } @Override @@ -85,51 +108,43 @@ public void read(DataInputView dataInputView) throws IOException { dataInputView.read(array); - this.data = convertToObject(array); + try { + this.data = convertToObject(array); + } catch (org.apache.wayang.core.api.exception.WayangException e) { + if (e.getCause() instanceof IOException) { + throw (IOException) e.getCause(); + } + throw new IOException("Failed to deserialize WayangValue payload", e); + } } - public byte[] convertToByte(){ - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ObjectOutput out = null; - byte[] result = null; - try { - out = new ObjectOutputStream(bos); + public byte[] convertToByte() { + try (ByteArrayOutputStream bos = new ByteArrayOutputStream(); + ObjectOutputStream out = new ObjectOutputStream(bos)) { out.writeObject(this.data); out.flush(); - result = bos.toByteArray(); + return bos.toByteArray(); } catch (IOException e) { - e.printStackTrace(); - } finally { - try { - bos.close(); - } catch (IOException ex) { - ex.printStackTrace(); - } + throw new org.apache.wayang.core.api.exception.WayangException("Failed to serialize WayangValue payload", e); } - return result; } - public T convertToObject(byte[] array){ - ByteArrayInputStream bis = new ByteArrayInputStream(array); - ObjectInput in = null; - T object = null; - try { - in = new ObjectInputStream(bis); - object = (T) in.readObject(); - } catch (IOException e) { - e.printStackTrace(); - } catch (ClassNotFoundException e) { - e.printStackTrace(); - } finally { - try { - if (in != null) { - in.close(); - } - } catch (IOException ex) { - ex.printStackTrace(); - } + public T convertToObject(byte[] array) { + return convertToObject(array, this.type != null ? this.type : (this.data != null ? this.data.getClass() : null)); + } + + public T convertToObject(byte[] array, Class expectedType) { + if (array == null || array.length == 0) { + return null; + } + Class targetType = expectedType != null ? expectedType : (this.type != null ? this.type : (this.data != null ? this.data.getClass() : null)); + try (ByteArrayInputStream bis = new ByteArrayInputStream(array); + ObjectInputStream in = new ObjectInputStream(bis)) { + in.setObjectInputFilter(ObjectFileSerialization.createObjectInputFilter(targetType)); + return (T) in.readObject(); + } catch (IOException | ClassNotFoundException e) { + throw new org.apache.wayang.core.api.exception.WayangException("Failed to deserialize WayangValue payload", e); } - return object; } public String toString(){ diff --git a/wayang-platforms/wayang-flink/src/test/java/org/apache/wayang/flink/compiler/criterion/WayangValueTest.java b/wayang-platforms/wayang-flink/src/test/java/org/apache/wayang/flink/compiler/criterion/WayangValueTest.java new file mode 100644 index 000000000..d594ff714 --- /dev/null +++ b/wayang-platforms/wayang-flink/src/test/java/org/apache/wayang/flink/compiler/criterion/WayangValueTest.java @@ -0,0 +1,124 @@ +/* + * 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.wayang.flink.compiler.criterion; + +import org.apache.flink.core.memory.DataInputViewStreamWrapper; +import org.apache.flink.core.memory.DataOutputViewStreamWrapper; +import org.apache.wayang.core.api.exception.WayangException; +import org.junit.Assert; +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.InvalidClassException; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.concurrent.atomic.AtomicBoolean; + +public class WayangValueTest { + + @Test + public void testRoundtripAllowedType() { + WayangValue value = new WayangValue<>("hello"); + byte[] bytes = value.convertToByte(); + Assert.assertNotNull(bytes); + + WayangValue target = new WayangValue<>(); + String deserialized = target.convertToObject(bytes); + Assert.assertEquals("hello", deserialized); + } + + @Test + public void testFreshWayangValueWithCustomPayloadTypeConstructor() { + WayangValue source = new WayangValue<>(new CustomPayload("custom-constructor")); + byte[] bytes = source.convertToByte(); + Assert.assertNotNull(bytes); + + WayangValue fresh = new WayangValue<>(CustomPayload.class); + CustomPayload deserialized = fresh.convertToObject(bytes); + Assert.assertNotNull(deserialized); + Assert.assertEquals("custom-constructor", deserialized.msg); + } + + @Test + public void testFreshWayangValueWithSetType() { + WayangValue source = new WayangValue<>(new CustomPayload("custom-setter")); + byte[] bytes = source.convertToByte(); + Assert.assertNotNull(bytes); + + WayangValue fresh = new WayangValue<>(); + fresh.setType(CustomPayload.class); + CustomPayload deserialized = fresh.convertToObject(bytes); + Assert.assertNotNull(deserialized); + Assert.assertEquals("custom-setter", deserialized.msg); + } + + @Test + public void testFreshWayangValueWithExplicitExpectedTypeInConvert() { + WayangValue source = new WayangValue<>(new CustomPayload("custom-explicit")); + byte[] bytes = source.convertToByte(); + Assert.assertNotNull(bytes); + + WayangValue fresh = new WayangValue<>(); + CustomPayload deserialized = fresh.convertToObject(bytes, CustomPayload.class); + Assert.assertNotNull(deserialized); + Assert.assertEquals("custom-explicit", deserialized.msg); + } + + @Test + public void testFlinkReadWriteRoundtripWithCustomPayload() throws Exception { + WayangValue source = new WayangValue<>(new CustomPayload("flink-stream"), CustomPayload.class); + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + DataOutputViewStreamWrapper outView = new DataOutputViewStreamWrapper(bos); + source.write(outView); + + ByteArrayInputStream bis = new ByteArrayInputStream(bos.toByteArray()); + DataInputViewStreamWrapper inView = new DataInputViewStreamWrapper(bis); + WayangValue fresh = new WayangValue<>(CustomPayload.class); + fresh.read(inView); + Assert.assertNotNull(fresh.get()); + Assert.assertEquals("flink-stream", fresh.get().msg); + } + + @Test + public void testRejectsDisallowedTypeThrowsException() throws Exception { + AtomicBoolean disallowed = new AtomicBoolean(true); + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bos)) { + out.writeObject(disallowed); + out.flush(); + } + + WayangValue target = new WayangValue<>(); + try { + target.convertToObject(bos.toByteArray()); + Assert.fail("Expected WayangException due to disallowed class"); + } catch (WayangException e) { + Assert.assertTrue(e.getCause() instanceof InvalidClassException); + } + } + + public static class CustomPayload implements Serializable { + final String msg; + + public CustomPayload(String msg) { + this.msg = msg; + } + } +}