CVE-2016-10750

@@ -12,6 +12,7 @@ Hazelcast IMDG Enterprise Enhancements:
 * Fine-Grained Control over WAN Replication Events: Introduced the new event type LOADED and the class EntryLoadedListener through which all the loads can be listened to. Also introduced the new configuration element persist-wan-replicated-data to specify whether to persist an incoming event over WAN replication or not. See the Configuring Consumer section.
 * License Enforcements and Warnings: Introduced a license monitor daemon that warns about expirations and instructs about the next steps. These information are also available through REST/JMX APIs. See the License Information section.
 * Delta WAN Synchronization: Introduced WAN anti-entropy mechanism utilizing Merkle Tree to synchronize only the affected subsets for Maps, improving network and heap utilization. See the Delta WAN Synchronization section.
+* Support for Adding/Removing WAN Publishers in a Running Cluster: Introduced the ability to add a newly defined WAN Publisher configuration which defines the replication information for new target clusters and then to assign this configuration to a map or cache. The replication to these target clusters resumes as if it has been defined in the configuration. See the Dynamically Adding WAN Publishers section.
 
 Hazelcast IMDG Open Source Enhancements:
 

@@ -526,6 +526,13 @@
                 </xs:complexType>
             </xs:element>
             <xs:element name="check-class-def-errors" type="xs:boolean" minOccurs="0" maxOccurs="1" default="true"/>
+            <xs:element name="java-serialization-filter" type="java-serialization-filter" minOccurs="0" maxOccurs="1">
+                <xs:annotation>
+                    <xs:documentation>
+                        Basic protection against untrusted deserialization based on class/package blacklisting and whitelisting.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
         </xs:all>
     </xs:complexType>
     <xs:complexType name="serialization-factory">
@@ -553,6 +560,44 @@
             </xs:extension>
         </xs:simpleContent>
     </xs:complexType>
+    <xs:complexType name="java-serialization-filter">
+        <xs:all>
+            <xs:element name="blacklist" type="filter-list" minOccurs="0" maxOccurs="1">
+                <xs:annotation>
+                    <xs:documentation>
+                        Blacklisted classes and packages, which are not allowed to be deserialized.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+            <xs:element name="whitelist" type="filter-list" minOccurs="0" maxOccurs="1">
+                <xs:annotation>
+                    <xs:documentation>
+                        Whitelisted classes and packages, which are allowed to be deserialized. If the list is empty
+                        (no class or package name provided) then all classes are allowed.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+        </xs:all>
+    </xs:complexType>
+    <xs:complexType name="filter-list">
+        <xs:choice minOccurs="0" maxOccurs="unbounded">
+            <xs:element name="class" type="xs:string" minOccurs="0" maxOccurs="unbounded">
+                <xs:annotation>
+                    <xs:documentation>
+                        Name of a class to be included in the list.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+            <xs:element name="package" type="xs:string" minOccurs="0" maxOccurs="unbounded">
+                <xs:annotation>
+                    <xs:documentation>
+                        Name of a package to be included in the list.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+        </xs:choice>
+    </xs:complexType>
+
     <xs:complexType name="socket-interceptor">
         <xs:all>
             <xs:element name="class-name" type="xs:string" minOccurs="0" maxOccurs="1"/>

@@ -134,6 +134,17 @@
                         class-name="com.hazelcast.examples.SerializerFactory"/>
         </serializers>
         <check-class-def-errors>true</check-class-def-errors>
+        <java-serialization-filter>
+            <blacklist>
+                <class>com.acme.app.BeanComparator</class>
+            </blacklist>
+            <whitelist>
+                <class>java.lang.String</class>
+                <class>example.Foo</class>
+                <package>com.acme.app</package>
+                <package>com.acme.app.subpkg</package>
+            </whitelist>
+        </java-serialization-filter>
     </serialization>
 
     <native-memory enabled="false" allocator-type="POOLED">

@@ -17,12 +17,14 @@
 package com.hazelcast.spring;
 
 import com.hazelcast.config.AbstractXmlConfigHelper;
+import com.hazelcast.config.ClassFilter;
 import com.hazelcast.config.DiscoveryConfig;
 import com.hazelcast.config.DiscoveryStrategyConfig;
 import com.hazelcast.config.EvictionConfig;
 import com.hazelcast.config.EvictionPolicy;
 import com.hazelcast.config.GlobalSerializerConfig;
 import com.hazelcast.config.InvalidConfigurationException;
+import com.hazelcast.config.JavaSerializationFilterConfig;
 import com.hazelcast.config.NearCachePreloaderConfig;
 import com.hazelcast.config.SerializationConfig;
 import com.hazelcast.config.SerializerConfig;
@@ -35,6 +37,7 @@
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.support.ManagedMap;
+import org.springframework.beans.factory.support.ManagedSet;
 import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.w3c.dom.NamedNodeMap;
@@ -306,6 +309,8 @@ protected void handleSerialization(Node node) {
                     handlePortableFactories(child, serializationConfigBuilder);
                 } else if ("serializers".equals(nodeName)) {
                     handleSerializers(child, serializationConfigBuilder);
+                } else if ("java-serialization-filter".equals(nodeName)) {
+                    handleJavaSerializationFilter(child, serializationConfigBuilder);
                 }
             }
             configBuilder.addPropertyValue("serializationConfig", beanDefinition);
@@ -511,5 +516,36 @@ private void handleDiscoveryStrategy(Node node, ManagedList<BeanDefinition> disc
             }
             discoveryStrategyConfigs.add(discoveryStrategyConfigBuilder.getBeanDefinition());
         }
+
+        protected void handleJavaSerializationFilter(final Node node, BeanDefinitionBuilder serializationConfigBuilder) {
+            BeanDefinitionBuilder filterConfigBuilder = createBeanBuilder(JavaSerializationFilterConfig.class);
+            for (Node child : childElements(node)) {
+                String name = cleanNodeName(child);
+                if ("blacklist".equals(name)) {
+                    filterConfigBuilder.addPropertyValue("blacklist", createFilterListBean(child));
+                } else if ("whitelist".equals(name)) {
+                    filterConfigBuilder.addPropertyValue("whitelist", createFilterListBean(child));
+                }
+            }
+            serializationConfigBuilder.addPropertyValue("javaSerializationFilterConfig",
+                    filterConfigBuilder.getBeanDefinition());
+        }
+
+        private AbstractBeanDefinition createFilterListBean(Node node) {
+            BeanDefinitionBuilder filterListBuilder = createBeanBuilder(ClassFilter.class);
+            ManagedSet<String> classes = new ManagedSet<String>();
+            ManagedSet<String> packages = new ManagedSet<String>();
+            for (Node child : childElements(node)) {
+                String name = cleanNodeName(child);
+                if ("class".equals(name)) {
+                    classes.add(getTextContent(child));
+                } else if ("package".equals(name)) {
+                    packages.add(getTextContent(child));
+                }
+            }
+            filterListBuilder.addPropertyValue("classes", classes);
+            filterListBuilder.addPropertyValue("packages", packages);
+            return filterListBuilder.getBeanDefinition();
+        }
     }
 }

@@ -1753,6 +1753,46 @@
         </xs:complexType>
     </xs:element>
 
+    <xs:element name="java-serialization-filter">
+        <xs:complexType>
+            <xs:sequence>
+                <xs:element name="blacklist" type="filter-list" minOccurs="0" maxOccurs="1">
+                    <xs:annotation>
+                        <xs:documentation>
+                            Blacklist used for deserialization class filtering.
+                        </xs:documentation>
+                    </xs:annotation>
+                </xs:element>
+                <xs:element name="whitelist" type="filter-list" minOccurs="0" maxOccurs="1">
+                    <xs:annotation>
+                        <xs:documentation>
+                            Blacklist used for deserialization class filtering.
+                        </xs:documentation>
+                    </xs:annotation>
+                </xs:element>
+            </xs:sequence>
+        </xs:complexType>
+    </xs:element>
+
+    <xs:complexType name="filter-list">
+        <xs:choice minOccurs="0" maxOccurs="unbounded">
+            <xs:element name="class" type="xs:string" minOccurs="0" maxOccurs="unbounded">
+                <xs:annotation>
+                    <xs:documentation>
+                        Name of a class to be included in the list.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+            <xs:element name="package" type="xs:string" minOccurs="0" maxOccurs="unbounded">
+                <xs:annotation>
+                    <xs:documentation>
+                        Name of a package to be included in the list.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+        </xs:choice>
+    </xs:complexType>
+
     <xs:complexType name="serialization-factory">
         <xs:attributeGroup ref="class-or-bean-name"/>
         <xs:attribute name="factory-id" type="xs:string" use="required"/>
@@ -2926,6 +2966,7 @@
             <xs:element ref="data-serializable-factories" minOccurs="0" maxOccurs="1"/>
             <xs:element ref="portable-factories" minOccurs="0" maxOccurs="1"/>
             <xs:element ref="serializers" minOccurs="0" maxOccurs="1"/>
+            <xs:element ref="java-serialization-filter" minOccurs="0" maxOccurs="1"/>
         </xs:sequence>
         <xs:attribute name="use-native-byte-order" use="optional" type="xs:string" default="false"/>
         <xs:attribute name="byte-order" use="optional" default="BIG_ENDIAN">

@@ -23,6 +23,7 @@
 import com.hazelcast.config.CacheDeserializedValues;
 import com.hazelcast.config.CacheSimpleConfig;
 import com.hazelcast.config.CardinalityEstimatorConfig;
+import com.hazelcast.config.ClassFilter;
 import com.hazelcast.config.Config;
 import com.hazelcast.config.CountDownLatchConfig;
 import com.hazelcast.config.DiscoveryConfig;
@@ -40,6 +41,7 @@
 import com.hazelcast.config.IcmpFailureDetectorConfig;
 import com.hazelcast.config.InMemoryFormat;
 import com.hazelcast.config.ItemListenerConfig;
+import com.hazelcast.config.JavaSerializationFilterConfig;
 import com.hazelcast.config.ListConfig;
 import com.hazelcast.config.ListenerConfig;
 import com.hazelcast.config.LockConfig;
@@ -1281,4 +1283,25 @@ public void testExplicitPortCountConfiguration() {
 
         assertEquals(42, portCount);
     }
+
+    @Test
+    public void testJavaSerializationFilterConfig() {
+        JavaSerializationFilterConfig filterConfig = config.getSerializationConfig().getJavaSerializationFilterConfig();
+        assertNotNull(filterConfig);
+        
+        ClassFilter blacklist = filterConfig.getBlacklist();
+        assertNotNull(blacklist);
+        assertEquals(1, blacklist.getClasses().size());
+        assertTrue(blacklist.getClasses().contains("com.acme.app.BeanComparator"));
+        assertEquals(0, blacklist.getPackages().size());
+        
+        ClassFilter whitelist = filterConfig.getWhitelist();
+        assertNotNull(whitelist);
+        assertEquals(2, whitelist.getClasses().size());
+        assertTrue(whitelist.getClasses().contains("java.lang.String"));
+        assertTrue(whitelist.getClasses().contains("example.Foo"));
+        assertEquals(2, whitelist.getPackages().size());
+        assertTrue(whitelist.getPackages().contains("com.acme.app"));
+        assertTrue(whitelist.getPackages().contains("com.acme.app.subpkg"));
+    }
 }

@@ -551,6 +551,17 @@
                     <hz:serializer type-class="com.hazelcast.spring.serialization.DummySerializableObject2"
                                    implementation="dummySerializer"/>
                 </hz:serializers>
+                <hz:java-serialization-filter>
+                    <hz:blacklist>
+                        <hz:class>com.acme.app.BeanComparator</hz:class>
+                    </hz:blacklist>
+                    <hz:whitelist>
+                        <hz:class>java.lang.String</hz:class>
+                        <hz:class>example.Foo</hz:class>
+                        <hz:package>com.acme.app</hz:package>
+                        <hz:package>com.acme.app.subpkg</hz:package>
+                    </hz:whitelist>
+                </hz:java-serialization-filter>
             </hz:serialization>
 
             <hz:native-memory enabled="false" allocator-type="POOLED" metadata-space-percentage="10.2"

@@ -449,6 +449,8 @@ protected SerializationConfig parseSerialization(final Node node) {
                 fillPortableFactories(child, serializationConfig);
             } else if ("serializers".equals(name)) {
                 fillSerializers(child, serializationConfig);
+            } else if ("java-serialization-filter".equals(name)) {
+                fillJavaSerializationFilter(child, serializationConfig);
             }
         }
         return serializationConfig;
@@ -507,6 +509,34 @@ protected void fillSerializers(final Node node, SerializationConfig serializatio
         }
     }
 
+    protected void fillJavaSerializationFilter(final Node node, SerializationConfig serializationConfig) {
+        JavaSerializationFilterConfig filterConfig = new JavaSerializationFilterConfig();
+        serializationConfig.setJavaSerializationFilterConfig(filterConfig);
+        for (Node child : childElements(node)) {
+            final String name = cleanNodeName(child);
+            if ("blacklist".equals(name)) {
+                ClassFilter list = parseClassFilterList(child);
+                filterConfig.setBlacklist(list);
+            } else if ("whitelist".equals(name)) {
+                ClassFilter list = parseClassFilterList(child);
+                filterConfig.setWhitelist(list);
+            }
+        }
+    }
+
+    private ClassFilter parseClassFilterList(Node node) {
+        ClassFilter list = new ClassFilter();
+        for (Node child : childElements(node)) {
+            final String name = cleanNodeName(child);
+            if ("class".equals(name)) {
+                list.addClasses(getTextContent(child));
+            } else if ("package".equals(name)) {
+                list.addPackages(getTextContent(child));
+            }
+        }
+        return list;
+    }
+
     protected void fillNativeMemoryConfig(Node node, NativeMemoryConfig nativeMemoryConfig) {
         final NamedNodeMap atts = node.getAttributes();
         final Node enabledNode = atts.getNamedItem("enabled");

@@ -0,0 +1,159 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.config;
+
+import static com.hazelcast.util.Preconditions.checkNotNull;
+import static java.util.Collections.unmodifiableSet;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.hazelcast.logging.ILogger;
+import com.hazelcast.logging.Logger;
+
+/**
+ * Holds blacklist and whitelist configuration in java deserialization configuration.
+ */
+public class ClassFilter {
+
+    private static final String PROPERTY_CLASSNAME_LIMIT = "hazelcast.serialization.filter.classname.limit";
+    private static final int CLASSNAME_LIMIT = Integer.getInteger(PROPERTY_CLASSNAME_LIMIT, 10000);
+    private static final ILogger LOGGER = Logger.getLogger(ClassFilter.class);
+
+    private final Set<String> classes = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+    private final Set<String> packages = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+
+    private AtomicBoolean warningLogged = new AtomicBoolean();
+
+    /**
+     * Returns unmodifiable set of class names.
+     */
+    public Set<String> getClasses() {
+        return unmodifiableSet(classes);
+    }
+
+    /**
+     * Returns unmodifiable set of package names.
+     */
+    public Set<String> getPackages() {
+        return unmodifiableSet(packages);
+    }
+
+    public ClassFilter addClasses(String... names) {
+        checkNotNull(names);
+        for (String name : names) {
+            classes.add(name);
+        }
+        return this;
+    }
+
+    public ClassFilter setClasses(Collection<String> names) {
+        checkNotNull(names);
+        classes.clear();
+        classes.addAll(names);
+        return this;
+    }
+
+    public ClassFilter addPackages(String... names) {
+        checkNotNull(names);
+        for (String name : names) {
+            packages.add(name);
+        }
+        return this;
+    }
+
+    public ClassFilter setPackages(Collection<String> names) {
+        checkNotNull(names);
+        packages.clear();
+        packages.addAll(names);
+        return this;
+    }
+
+    public boolean isEmpty() {
+        return classes.isEmpty() && packages.isEmpty();
+    }
+
+    public boolean isListed(String className) {
+        if (classes.contains(className)) {
+            return true;
+        }
+        if (!packages.isEmpty()) {
+            int dotPosition = className.lastIndexOf(".");
+            if (dotPosition > 0) {
+                // String packageName = ;
+                return checkPackage(className, className.substring(0, dotPosition));
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Checks if given class name is listed by package. If it's listed, then performance optimization is used and classname is
+     * added directly to {@code classes} collection.
+     *
+     * @param className Class name to be checked.
+     * @param packageName Package name of the checked class.
+     * @return {@code true} iff class is listed by-package
+     */
+    private boolean checkPackage(String className, String packageName) {
+        if (packages.contains(packageName)) {
+            if (classes.size() < CLASSNAME_LIMIT) {
+                // performance optimization
+                classes.add(className);
+            } else if (warningLogged.compareAndSet(false, true)) {
+                LOGGER.warning(String.format(
+                        "The class names collection size reached its limit. Optimizations for package names checks "
+                                + "will not optimize next usages. You can control the class names collection size limit by "
+                                + "setting system property '%s'. Actual value is %d.",
+                        PROPERTY_CLASSNAME_LIMIT, CLASSNAME_LIMIT));
+            }
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public int hashCode() {
+        final int prime = 31;
+        int result = 1;
+        result = prime * result + ((classes == null) ? 0 : classes.hashCode());
+        result = prime * result + ((packages == null) ? 0 : packages.hashCode());
+        return result;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+        ClassFilter other = (ClassFilter) obj;
+        return ((classes == null && other.classes == null) || (classes != null && classes.equals(other.classes)))
+                && ((packages == null && other.packages == null) || (packages != null && packages.equals(other.packages)));
+    }
+
+    @Override
+    public String toString() {
+        return "ClassFilter{classes=" + classes + ", packages=" + packages + "}";
+    }
+
+}

@@ -400,8 +400,15 @@ private static void serializationXmlGenerator(XmlGenerator gen, Config config) {
             }
             gen.close();
         }
-        gen.node("check-class-def-errors", c.isCheckClassDefErrors())
-                .close();
+        gen.node("check-class-def-errors", c.isCheckClassDefErrors());
+        JavaSerializationFilterConfig javaSerializationFilterConfig = c.getJavaSerializationFilterConfig();
+        if (javaSerializationFilterConfig != null) {
+            gen.open("java-serialization-filter");
+            appendFilterList(gen, "blacklist", javaSerializationFilterConfig.getBlacklist());
+            appendFilterList(gen, "whitelist", javaSerializationFilterConfig.getWhitelist());
+            gen.close();
+        }
+        gen.close();
     }
 
     private static String classNameOrClass(String className, Class clazz) {
@@ -1398,6 +1405,20 @@ private static void appendSerializationFactory(XmlGenerator gen, String elementN
         }
     }
 
+    private static void appendFilterList(XmlGenerator gen, String listName, ClassFilter classFilterList) {
+        if (classFilterList.isEmpty()) {
+            return;
+        }
+        gen.open(listName);
+        for (String className : classFilterList.getClasses()) {
+            gen.node("class", className);
+        }
+        for (String packageName : classFilterList.getPackages()) {
+            gen.node("package", packageName);
+        }
+        gen.close();
+    }
+
     private static final class XmlGenerator {
 
         private final StringBuilder xml;

@@ -0,0 +1,90 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.config;
+
+/**
+ * Configuration for Serialization Filter.
+ */
+public class JavaSerializationFilterConfig {
+
+    private volatile ClassFilter blacklist;
+    private volatile ClassFilter whitelist;
+
+    public ClassFilter getBlacklist() {
+        if (blacklist == null) {
+            blacklist = new ClassFilter();
+            // default blacklist - some well-known vulnerable classes/packages
+            blacklist.addClasses(
+                    "com.sun.org.apache.xalan.internal.xsltc.trax.TemplatesImpl",
+                    "bsh.XThis",
+                    "org.apache.commons.beanutils.BeanComparator",
+                    "org.codehaus.groovy.runtime.ConvertedClosure",
+                    "org.codehaus.groovy.runtime.MethodClosure",
+                    "org.springframework.beans.factory.ObjectFactory",
+                    "com.sun.org.apache.xalan.internal.xsltc.trax.TemplatesImpl")
+            .addPackages(
+                    "org.apache.commons.collections.functors",
+                    "org.apache.commons.collections4.functors");
+        }
+        return blacklist;
+    }
+
+    public JavaSerializationFilterConfig setBlacklist(ClassFilter blackList) {
+        this.blacklist = blackList;
+        return this;
+    }
+
+    public ClassFilter getWhitelist() {
+        if (whitelist == null) {
+            whitelist = new ClassFilter();
+        }
+        return whitelist;
+    }
+
+    public JavaSerializationFilterConfig setWhitelist(ClassFilter whiteList) {
+        this.whitelist = whiteList;
+        return this;
+    }
+
+    @Override
+    public int hashCode() {
+        final int prime = 31;
+        int result = 1;
+        result = prime * result + ((blacklist == null) ? 0 : blacklist.hashCode());
+        result = prime * result + ((whitelist == null) ? 0 : whitelist.hashCode());
+        return result;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+        JavaSerializationFilterConfig other = (JavaSerializationFilterConfig) obj;
+        return ((blacklist == null && other.blacklist == null) || (blacklist != null && blacklist.equals(other.blacklist)))
+                && ((whitelist == null && other.whitelist == null) || (whitelist != null && whitelist.equals(other.whitelist)));
+    }
+
+    @Override
+    public String toString() {
+        return "JavaSerializationFilterConfig{ blacklist=" + blacklist + ", whitelist=" + whitelist + "}";
+    }
+
+}

@@ -63,6 +63,8 @@ public class SerializationConfig {
 
     private Set<ClassDefinition> classDefinitions;
 
+    private JavaSerializationFilterConfig javaSerializationFilterConfig;
+
     public SerializationConfig() {
     }
 
@@ -442,6 +444,23 @@ public SerializationConfig setAllowUnsafe(boolean allowUnsafe) {
         return this;
     }
 
+    /**
+     * @return the javaSerializationFilterConfig
+     */
+    public JavaSerializationFilterConfig getJavaSerializationFilterConfig() {
+        return javaSerializationFilterConfig;
+    }
+
+    /**
+     * Allows to configure deserialization protection filter.
+     *
+     * @param javaSerializationFilterConfig the javaSerializationFilterConfig to set (may be {@code null})
+     */
+    public SerializationConfig setJavaSerializationFilterConfig(JavaSerializationFilterConfig javaSerializationFilterConfig) {
+        this.javaSerializationFilterConfig = javaSerializationFilterConfig;
+        return this;
+    }
+
     @Override
     public String toString() {
         return "SerializationConfig{"
@@ -456,6 +475,7 @@ public String toString() {
                 + ", classDefinitions=" + classDefinitions
                 + ", byteOrder=" + byteOrder
                 + ", useNativeByteOrder=" + useNativeByteOrder
+                + ", javaSerializationFilterConfig=" + javaSerializationFilterConfig
                 + '}';
     }
 }

@@ -88,19 +88,15 @@ public abstract class AbstractSerializationService implements InternalSerializat
     private final byte version;
     private final ILogger logger = Logger.getLogger(InternalSerializationService.class);
 
-    AbstractSerializationService(InputOutputFactory inputOutputFactory, byte version, ClassLoader classLoader,
-                                 ManagedContext managedContext, PartitioningStrategy globalPartitionStrategy,
-                                 int initialOutputBufferSize,
-                                 BufferPoolFactory bufferPoolFactory,
-                                 Supplier<RuntimeException> notActiveExceptionSupplier) {
-        this.inputOutputFactory = inputOutputFactory;
-        this.version = version;
-        this.classLoader = classLoader;
-        this.managedContext = managedContext;
-        this.globalPartitioningStrategy = globalPartitionStrategy;
-        this.outputBufferSize = initialOutputBufferSize;
-        this.bufferPoolThreadLocal = new BufferPoolThreadLocal(this, bufferPoolFactory,
-                notActiveExceptionSupplier);
+    AbstractSerializationService(Builder<?> builder) {
+        this.inputOutputFactory = builder.inputOutputFactory;
+        this.version = builder.version;
+        this.classLoader = builder.classLoader;
+        this.managedContext = builder.managedContext;
+        this.globalPartitioningStrategy = builder.globalPartitionStrategy;
+        this.outputBufferSize = builder.initialOutputBufferSize;
+        this.bufferPoolThreadLocal = new BufferPoolThreadLocal(this, builder.bufferPoolFactory,
+                builder.notActiveExceptionSupplier);
         this.nullSerializerAdapter = createSerializerAdapter(new ConstantSerializers.NullSerializer(), this);
     }
 
@@ -564,4 +560,64 @@ private SerializerAdapter lookupJavaSerializer(Class type) {
         }
         return null;
     }
+
+    public abstract static class Builder<T extends Builder<T>> {
+        private InputOutputFactory inputOutputFactory;
+        private byte version;
+        private ClassLoader classLoader;
+        private ManagedContext managedContext;
+        private PartitioningStrategy globalPartitionStrategy;
+        private int initialOutputBufferSize;
+        private BufferPoolFactory bufferPoolFactory;
+        private Supplier<RuntimeException> notActiveExceptionSupplier;
+
+        protected Builder() {
+        }
+
+        protected abstract T self();
+
+        public final T withInputOutputFactory(InputOutputFactory inputOutputFactory) {
+            this.inputOutputFactory = inputOutputFactory;
+            return self();
+        }
+
+        public final T withVersion(byte version) {
+            this.version = version;
+            return self();
+        }
+
+        public final T withClassLoader(ClassLoader classLoader) {
+            this.classLoader = classLoader;
+            return self();
+        }
+
+        public ClassLoader getClassLoader() {
+            return classLoader;
+        }
+
+        public final T withManagedContext(ManagedContext managedContext) {
+            this.managedContext = managedContext;
+            return self();
+        }
+
+        public final T withGlobalPartitionStrategy(PartitioningStrategy globalPartitionStrategy) {
+            this.globalPartitionStrategy = globalPartitionStrategy;
+            return self();
+        }
+
+        public final T withInitialOutputBufferSize(int initialOutputBufferSize) {
+            this.initialOutputBufferSize = initialOutputBufferSize;
+            return self();
+        }
+
+        public final T withBufferPoolFactory(BufferPoolFactory bufferPoolFactory) {
+            this.bufferPoolFactory = bufferPoolFactory;
+            return self();
+        }
+
+        public final T withNotActiveExceptionSupplier(Supplier<RuntimeException> notActiveExceptionSupplier) {
+            this.notActiveExceptionSupplier = notActiveExceptionSupplier;
+            return self();
+        }
+    }
 }

@@ -17,6 +17,7 @@
 package com.hazelcast.internal.serialization.impl;
 
 import com.hazelcast.config.GlobalSerializerConfig;
+import com.hazelcast.config.JavaSerializationFilterConfig;
 import com.hazelcast.config.SerializationConfig;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.HazelcastInstanceAware;
@@ -29,6 +30,8 @@
 import com.hazelcast.internal.serialization.SerializationServiceBuilder;
 import com.hazelcast.internal.serialization.impl.bufferpool.BufferPoolFactoryImpl;
 import com.hazelcast.nio.ClassLoaderUtil;
+import com.hazelcast.nio.ClassNameFilter;
+import com.hazelcast.nio.SerializationClassNameFilter;
 import com.hazelcast.nio.serialization.ClassDefinition;
 import com.hazelcast.nio.serialization.DataSerializableFactory;
 import com.hazelcast.nio.serialization.HazelcastSerializationException;
@@ -87,6 +90,8 @@ public class DefaultSerializationServiceBuilder implements SerializationServiceB
 
     protected Supplier<RuntimeException> notActiveExceptionSupplier;
 
+    protected ClassNameFilter classNameFilter;
+
     @Override
     public SerializationServiceBuilder setVersion(byte version) {
         byte maxVersion = BuildInfoProvider.getBuildInfo().getSerializationVersion();
@@ -125,6 +130,8 @@ public SerializationServiceBuilder setConfig(SerializationConfig config) {
         enableCompression = config.isEnableCompression();
         enableSharedObject = config.isEnableSharedObject();
         allowUnsafe = config.isAllowUnsafe();
+        JavaSerializationFilterConfig filterConfig = config.getJavaSerializationFilterConfig();
+        classNameFilter = filterConfig == null ? null : new SerializationClassNameFilter(filterConfig);
         return this;
     }
 
@@ -272,10 +279,22 @@ protected InternalSerializationService createSerializationService(InputOutputFac
                                                                       Supplier<RuntimeException> notActiveExceptionSupplier) {
         switch (version) {
             case 1:
-                SerializationServiceV1 serializationServiceV1 = new SerializationServiceV1(inputOutputFactory, version,
-                        portableVersion, classLoader, dataSerializableFactories, portableFactories, managedContext,
-                        partitioningStrategy, initialOutputBufferSize, new BufferPoolFactoryImpl(), enableCompression,
-                        enableSharedObject, notActiveExceptionSupplier);
+                SerializationServiceV1 serializationServiceV1 = SerializationServiceV1.builder()
+                    .withInputOutputFactory(inputOutputFactory)
+                    .withVersion(version)
+                    .withPortableVersion(portableVersion)
+                    .withClassLoader(classLoader)
+                    .withDataSerializableFactories(dataSerializableFactories)
+                    .withPortableFactories(portableFactories)
+                    .withManagedContext(managedContext)
+                    .withGlobalPartitionStrategy(partitioningStrategy)
+                    .withInitialOutputBufferSize(initialOutputBufferSize)
+                    .withBufferPoolFactory(new BufferPoolFactoryImpl())
+                    .withEnableCompression(enableCompression)
+                    .withEnableSharedObject(enableSharedObject)
+                    .withNotActiveExceptionSupplier(notActiveExceptionSupplier)
+                    .withClassNameFilter(classNameFilter)
+                    .build();
                 serializationServiceV1.registerClassDefinitions(classDefinitions, checkClassDefErrors);
                 return serializationServiceV1;
 

@@ -18,6 +18,7 @@
 
 import com.hazelcast.nio.BufferObjectDataInput;
 import com.hazelcast.nio.ClassLoaderUtil;
+import com.hazelcast.nio.ClassNameFilter;
 import com.hazelcast.nio.ObjectDataInput;
 import com.hazelcast.nio.ObjectDataOutput;
 import com.hazelcast.nio.serialization.HazelcastSerializationException;
@@ -53,10 +54,12 @@ public static final class JavaSerializer extends SingletonSerializer<Object> {
 
         private final boolean shared;
         private final boolean gzipEnabled;
+        private final ClassNameFilter classFilter;
 
-        public JavaSerializer(boolean shared, boolean gzipEnabled) {
+        public JavaSerializer(boolean shared, boolean gzipEnabled, ClassNameFilter classFilter) {
             this.shared = shared;
             this.gzipEnabled = gzipEnabled;
+            this.classFilter = classFilter;
         }
 
         @Override
@@ -74,7 +77,7 @@ public Object read(final ObjectDataInput in) throws IOException {
 
         private Object read(InputStream in, ClassLoader classLoader) throws IOException {
             try {
-                ObjectInputStream objectInputStream = newObjectInputStream(classLoader, in);
+                ObjectInputStream objectInputStream = newObjectInputStream(classLoader, classFilter, in);
                 if (shared) {
                     return objectInputStream.readObject();
                 }
@@ -130,9 +133,11 @@ private void writeGzipped(OutputStream out, Object obj) throws IOException {
     public static final class ExternalizableSerializer extends SingletonSerializer<Externalizable> {
 
         private final boolean gzipEnabled;
+        private final ClassNameFilter classFilter;
 
-        public ExternalizableSerializer(boolean gzipEnabled) {
+        public ExternalizableSerializer(boolean gzipEnabled, ClassNameFilter classFilter) {
             this.gzipEnabled = gzipEnabled;
+            this.classFilter = classFilter;
         }
 
         @Override
@@ -167,7 +172,7 @@ private Externalizable readGzipped(InputStream in, String className, ClassLoader
 
         private Externalizable read(InputStream in, String className, ClassLoader classLoader) throws Exception {
             Externalizable ds = ClassLoaderUtil.newInstance(classLoader, className);
-            ObjectInputStream objectInputStream = newObjectInputStream(classLoader, in);
+            ObjectInputStream objectInputStream = newObjectInputStream(classLoader, classFilter, in);
             ds.readExternal(objectInputStream);
             return ds;
         }

@@ -16,15 +16,13 @@
 
 package com.hazelcast.internal.serialization.impl;
 
-import com.hazelcast.core.ManagedContext;
 import com.hazelcast.core.PartitioningStrategy;
-import com.hazelcast.internal.serialization.InputOutputFactory;
 import com.hazelcast.internal.serialization.PortableContext;
 import com.hazelcast.internal.serialization.impl.ConstantSerializers.BooleanSerializer;
 import com.hazelcast.internal.serialization.impl.ConstantSerializers.ByteSerializer;
 import com.hazelcast.internal.serialization.impl.ConstantSerializers.StringArraySerializer;
-import com.hazelcast.internal.serialization.impl.bufferpool.BufferPoolFactory;
 import com.hazelcast.nio.BufferObjectDataInput;
+import com.hazelcast.nio.ClassNameFilter;
 import com.hazelcast.nio.ObjectDataInput;
 import com.hazelcast.nio.serialization.ClassDefinition;
 import com.hazelcast.nio.serialization.Data;
@@ -37,7 +35,6 @@
 import com.hazelcast.nio.serialization.Portable;
 import com.hazelcast.nio.serialization.PortableFactory;
 import com.hazelcast.nio.serialization.PortableReader;
-import com.hazelcast.util.function.Supplier;
 
 import java.io.Externalizable;
 import java.io.IOException;
@@ -46,6 +43,7 @@
 import java.math.BigInteger;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Date;
 import java.util.LinkedList;
 import java.util.Map;
@@ -86,28 +84,23 @@ public class SerializationServiceV1 extends AbstractSerializationService {
     private final PortableContextImpl portableContext;
     private final PortableSerializer portableSerializer;
 
-     SerializationServiceV1(InputOutputFactory inputOutputFactory, byte version, int portableVersion, ClassLoader classLoader,
-            Map<Integer, ? extends DataSerializableFactory> dataSerializableFactories,
-            Map<Integer, ? extends PortableFactory> portableFactories, ManagedContext managedContext,
-            PartitioningStrategy globalPartitionStrategy, int initialOutputBufferSize, BufferPoolFactory bufferPoolFactory,
-            boolean enableCompression, boolean enableSharedObject, Supplier<RuntimeException> notActiveExceptionSupplier) {
-        super(inputOutputFactory, version, classLoader, managedContext, globalPartitionStrategy, initialOutputBufferSize,
-                bufferPoolFactory, notActiveExceptionSupplier);
-
-        PortableHookLoader loader = new PortableHookLoader(portableFactories, classLoader);
-        portableContext = new PortableContextImpl(this, portableVersion);
+    SerializationServiceV1(AbstractBuilder<?> builder) {
+        super(builder);
+        PortableHookLoader loader = new PortableHookLoader(builder.portableFactories, builder.getClassLoader());
+        portableContext = new PortableContextImpl(this, builder.portableVersion);
         for (ClassDefinition cd : loader.getDefinitions()) {
             portableContext.registerClassDefinition(cd);
         }
 
         dataSerializerAdapter = createSerializerAdapter(
-                new DataSerializableSerializer(dataSerializableFactories, classLoader), this);
+                new DataSerializableSerializer(builder.dataSerializableFactories, builder.getClassLoader()), this);
         portableSerializer = new PortableSerializer(portableContext, loader.getFactories());
         portableSerializerAdapter = createSerializerAdapter(portableSerializer, this);
 
-        javaSerializerAdapter = createSerializerAdapter(new JavaSerializer(enableSharedObject, enableCompression), this);
+        javaSerializerAdapter = createSerializerAdapter(
+                new JavaSerializer(builder.enableSharedObject, builder.enableCompression, builder.classNameFilter), this);
         javaExternalizableAdapter = createSerializerAdapter(
-                new JavaDefaultSerializers.ExternalizableSerializer(enableCompression), this);
+                new JavaDefaultSerializers.ExternalizableSerializer(builder.enableCompression, builder.classNameFilter), this);
         registerConstantSerializers();
         registerJavaTypeSerializers();
     }
@@ -248,10 +241,77 @@ public ObjectDataInput initDataSerializableInputAndSkipTheHeader(Data data) thro
         return input;
     }
 
+    public static Builder builder() {
+        return new Builder();
+    }
+
     private void skipBytesSafely(ObjectDataInput input, int count) throws IOException {
         if (input.skipBytes(count) != count) {
             throw new HazelcastSerializationException("Malformed serialization format");
         }
     }
 
+    public abstract static class AbstractBuilder<T extends AbstractBuilder<T>> extends AbstractSerializationService.Builder<T> {
+
+        private int portableVersion;
+        private Map<Integer, ? extends DataSerializableFactory> dataSerializableFactories = Collections.emptyMap();
+        private Map<Integer, ? extends PortableFactory> portableFactories = Collections.emptyMap();
+        private boolean enableCompression;
+        private boolean enableSharedObject;
+        private ClassNameFilter classNameFilter;
+
+        protected AbstractBuilder() {
+        }
+
+        public final T withPortableVersion(int portableVersion) {
+            this.portableVersion = portableVersion;
+            return self();
+        }
+
+        public final T withDataSerializableFactories(
+                Map<Integer, ? extends DataSerializableFactory> dataSerializableFactories) {
+            this.dataSerializableFactories = dataSerializableFactories;
+            return self();
+        }
+
+        public Map<Integer, ? extends DataSerializableFactory> getDataSerializableFactories() {
+            return dataSerializableFactories;
+        }
+
+        public final T withPortableFactories(Map<Integer, ? extends PortableFactory> portableFactories) {
+            this.portableFactories = portableFactories;
+            return self();
+        }
+
+        public final T withEnableCompression(boolean enableCompression) {
+            this.enableCompression = enableCompression;
+            return self();
+        }
+
+        public final T withEnableSharedObject(boolean enableSharedObject) {
+            this.enableSharedObject = enableSharedObject;
+            return self();
+        }
+
+        public final T withClassNameFilter(ClassNameFilter classNameFilter) {
+            this.classNameFilter = classNameFilter;
+            return self();
+        }
+    }
+
+    public static final class Builder extends AbstractBuilder<Builder> {
+
+        protected Builder() {
+        }
+
+        @Override
+        protected Builder self() {
+            return this;
+        }
+
+        public SerializationServiceV1 build() {
+            return new SerializationServiceV1(this);
+        }
+
+    }
 }

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.nio;
+
+/**
+ * Allows to intercept class resolution during deserialization based on classname. It's used as a validation mechanism in
+ * Look-ahead ObjectInputStream.
+ */
+public interface ClassNameFilter {
+
+    /**
+     * Called from {@link java.io.ObjectInputStream#resolveClass}. When the deserialization should not be allowed the method
+     * throws a {@link RuntimeException}.
+     *
+     * @param className name of class to be deserialized
+     */
+    void filter(String className);
+}

@@ -161,8 +161,9 @@ public static void readFully(InputStream in, byte[] buffer) throws IOException {
         }
     }
 
-    public static ObjectInputStream newObjectInputStream(final ClassLoader classLoader, InputStream in) throws IOException {
-        return new ClassLoaderAwareObjectInputStream(classLoader, in);
+    public static ObjectInputStream newObjectInputStream(final ClassLoader classLoader, ClassNameFilter classFilter,
+            InputStream in) throws IOException {
+        return new ClassLoaderAwareObjectInputStream(classLoader, classFilter, in);
     }
 
     public static OutputStream newOutputStream(final ByteBuffer dst) {
@@ -627,15 +628,22 @@ public static String toDebugString(String name, ByteBuffer byteBuffer) {
     private static final class ClassLoaderAwareObjectInputStream extends ObjectInputStream {
 
         private final ClassLoader classLoader;
+        private final ClassNameFilter classFilter;
 
-        private ClassLoaderAwareObjectInputStream(final ClassLoader classLoader, final InputStream in) throws IOException {
+        private ClassLoaderAwareObjectInputStream(final ClassLoader classLoader, ClassNameFilter classFilter,
+                final InputStream in) throws IOException {
             super(in);
             this.classLoader = classLoader;
+            this.classFilter = classFilter;
         }
 
         @Override
         protected Class<?> resolveClass(ObjectStreamClass desc) throws ClassNotFoundException {
-            return ClassLoaderUtil.loadClass(classLoader, desc.getName());
+            String name = desc.getName();
+            if (classFilter != null) {
+                classFilter.filter(name);
+            }
+            return ClassLoaderUtil.loadClass(classLoader, name);
         }
 
         @Override

@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.nio;
+
+import static java.lang.String.format;
+
+import com.hazelcast.config.ClassFilter;
+import com.hazelcast.config.JavaSerializationFilterConfig;
+import com.hazelcast.util.Preconditions;
+
+/**
+ * Implementation of basic protection against untrusted deserialization. It holds blacklist and whitelist with classnames and
+ * package names.
+ *
+ * @see #filter(String)
+ */
+public final class SerializationClassNameFilter implements ClassNameFilter {
+
+    private static final String DESERIALIZATION_ERROR = "Resolving class %s is not allowed.";
+
+    private final ClassFilter blacklist;
+    private final ClassFilter whitelist;
+
+    public SerializationClassNameFilter(JavaSerializationFilterConfig config) {
+        Preconditions.checkNotNull(config, "JavaSerializationFilterConfig has to be provided");
+        blacklist = config.getBlacklist();
+        whitelist = config.getWhitelist();
+    }
+
+    /**
+     * Throws {@link SecurityException} if the given class name appears on the blacklist or does not appear on a non-empty
+     * whitelist.
+     *
+     * @param className class name to check
+     * @throws SecurityException if the classname is not allowed for deserialization
+     */
+    public void filter(String className) throws SecurityException {
+        if (blacklist.isListed(className) || (!whitelist.isEmpty() && !whitelist.isListed(className))) {
+            throw new SecurityException(format(DESERIALIZATION_ERROR, className));
+        }
+    }
+}

@@ -3100,6 +3100,13 @@
                     </xs:documentation>
                 </xs:annotation>
             </xs:element>
+            <xs:element name="java-serialization-filter" type="java-serialization-filter" minOccurs="0" maxOccurs="1">
+                <xs:annotation>
+                    <xs:documentation>
+                        Basic protection against untrusted deserialization based on class/package blacklisting and whitelisting.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
         </xs:all>
     </xs:complexType>
     <xs:complexType name="serialization-factory">
@@ -3139,6 +3146,43 @@
             </xs:extension>
         </xs:simpleContent>
     </xs:complexType>
+    <xs:complexType name="java-serialization-filter">
+        <xs:all>
+            <xs:element name="blacklist" type="filter-list" minOccurs="0" maxOccurs="1">
+                <xs:annotation>
+                    <xs:documentation>
+                        Blacklisted classes and packages, which are not allowed to be deserialized.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+            <xs:element name="whitelist" type="filter-list" minOccurs="0" maxOccurs="1">
+                <xs:annotation>
+                    <xs:documentation>
+                        Whitelisted classes and packages, which are allowed to be deserialized. If the list is empty
+                        (no class or package name provided) then all classes are allowed.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+        </xs:all>
+    </xs:complexType>
+    <xs:complexType name="filter-list">
+        <xs:choice minOccurs="0" maxOccurs="unbounded">
+            <xs:element name="class" type="xs:string" minOccurs="0" maxOccurs="unbounded">
+                <xs:annotation>
+                    <xs:documentation>
+                        Name of a class to be included in the list.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+            <xs:element name="package" type="xs:string" minOccurs="0" maxOccurs="unbounded">
+                <xs:annotation>
+                    <xs:documentation>
+                        Name of a package to be included in the list.
+                    </xs:documentation>
+                </xs:annotation>
+            </xs:element>
+        </xs:choice>
+    </xs:complexType>
 
     <xs:complexType name="socket-interceptor">
         <xs:all>

@@ -1784,6 +1784,8 @@ https://hazelcast.org/documentation/.
         * <check-class-def-errors>:
             If set to true, serialization system will check the class definitions error at the start of serialization process
             and throw a Serialization Exception with the error definition.
+        * <java-serialization-filter>:
+            Allows to configure blacklisting and whitelisting for deserialized classes when Java serialization is used.
     -->
     <serialization>
         <portable-version>0</portable-version>
@@ -1802,6 +1804,17 @@ https://hazelcast.org/documentation/.
                         class-name="com.hazelcast.examples.SerializerFactory"/>
         </serializers>
         <check-class-def-errors>true</check-class-def-errors>
+        <java-serialization-filter>
+            <blacklist>
+                <class>com.acme.app.BeanComparator</class>
+            </blacklist>
+            <whitelist>
+                <class>java.lang.String</class>
+                <class>example.Foo</class>
+                <package>com.acme.app</package>
+                <package>com.acme.app.subpkg</package>
+            </whitelist>
+        </java-serialization-filter>
     </serialization>
     <!--
         ===== HAZELCAST HIGH-DENSITY MEMORY STORE CONFIGURATION =====

@@ -0,0 +1,133 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.cluster;
+
+import static org.junit.Assert.assertFalse;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.net.DatagramPacket;
+import java.net.InetAddress;
+import java.net.MulticastSocket;
+import java.nio.ByteBuffer;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import com.hazelcast.config.Config;
+import com.hazelcast.config.JavaSerializationFilterConfig;
+import com.hazelcast.config.JoinConfig;
+import com.hazelcast.config.MulticastConfig;
+import com.hazelcast.config.NetworkConfig;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.instance.HazelcastInstanceFactory;
+import com.hazelcast.internal.serialization.impl.SerializationConstants;
+import com.hazelcast.nio.IOUtil;
+import com.hazelcast.nio.Packet;
+import com.hazelcast.test.annotation.QuickTest;
+
+/**
+ * Tests if deserialization blacklisting works for MutlicastService.
+ */
+@Category(QuickTest.class)
+public class MulticastDeserializationTest {
+
+    private static final int MULTICAST_PORT = 53535;
+    private static final String MULTICAST_GROUP = "224.0.0.219";
+    // TTL==0 : Restricted to the same host. Won't be output by any interface.
+    private static final int MULTICAST_TTL = 0;
+
+    @Before
+    @After
+    public void killAllHazelcastInstances() throws IOException {
+        HazelcastInstanceFactory.terminateAll();
+    }
+
+    /**
+     * <pre>
+     * Given: Multicast is configured.
+     * When: DatagramPacket with a correct Packet comes. The Packet references Java serializer and the serialized object is not a Join message.
+     * Then: The object from the Packet is not deserialized.
+     * </pre>
+     */
+    @Test
+    public void test() throws Exception {
+        Config config = new Config();
+        JavaSerializationFilterConfig javaSerializationFilterConfig = new JavaSerializationFilterConfig();
+        javaSerializationFilterConfig.getBlacklist().addClasses(TestDeserialized.class.getName());
+        config.getSerializationConfig().setJavaSerializationFilterConfig(javaSerializationFilterConfig);
+        NetworkConfig networkConfig = config.getNetworkConfig();
+        JoinConfig join = networkConfig.getJoin();
+        join.getTcpIpConfig().setEnabled(false);
+        MulticastConfig multicastConfig = join.getMulticastConfig();
+        multicastConfig.setMulticastPort(MULTICAST_PORT);
+        multicastConfig.setMulticastGroup(MULTICAST_GROUP);
+        multicastConfig.setMulticastTimeToLive(MULTICAST_TTL);
+        multicastConfig.setEnabled(true);
+
+        Hazelcast.newHazelcastInstance(config);
+        sendJoinDatagram(new TestDeserialized());
+        Thread.sleep(500L);
+        assertFalse("Untrusted deserialization is possible", TestDeserialized.IS_DESERIALIZED);
+    }
+
+    private void sendJoinDatagram(Object object) throws IOException {
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        ObjectOutputStream oos = new ObjectOutputStream(bos);
+        try {
+            oos.writeObject(object);
+        } finally {
+            IOUtil.closeResource(oos);
+        }
+        byte[] data = bos.toByteArray();
+        MulticastSocket multicastSocket = null;
+        try {
+            multicastSocket = new MulticastSocket(MULTICAST_PORT);
+            multicastSocket.setTimeToLive(MULTICAST_TTL);
+            InetAddress group = InetAddress.getByName(MULTICAST_GROUP);
+            multicastSocket.joinGroup(group);
+            int msgSize = data.length;
+
+            ByteBuffer bbuf = ByteBuffer.allocate(1 + 4 + msgSize);
+            bbuf.put(Packet.VERSION);
+            bbuf.putInt(SerializationConstants.JAVA_DEFAULT_TYPE_SERIALIZABLE);
+            bbuf.put(data);
+            byte[] packetData = bbuf.array();
+            DatagramPacket packet = new DatagramPacket(packetData, packetData.length, group, MULTICAST_PORT);
+            multicastSocket.send(packet);
+            multicastSocket.leaveGroup(group);
+        } finally {
+            IOUtil.closeResource(multicastSocket);
+        }
+    }
+
+    public static class TestDeserialized implements Serializable {
+        private static final long serialVersionUID = 1L;
+        public static volatile boolean IS_DESERIALIZED = false;
+
+        private void writeObject(java.io.ObjectOutputStream out) throws IOException {
+        }
+
+        private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
+            IS_DESERIALIZED = true;
+        }
+    }
+}

@@ -1058,7 +1058,8 @@ && nullSafeEqual(c1.isUseNativeByteOrder(), c2.isUseNativeByteOrder())
                     && nullSafeEqual(c1.getByteOrder(), c2.getByteOrder())
                     && nullSafeEqual(c1.isEnableCompression(), c2.isEnableCompression())
                     && nullSafeEqual(c1.isEnableSharedObject(), c2.isEnableSharedObject())
-                    && nullSafeEqual(c1.isAllowUnsafe(), c2.isAllowUnsafe());
+                    && nullSafeEqual(c1.isAllowUnsafe(), c2.isAllowUnsafe())
+                    && nullSafeEqual(c1.getJavaSerializationFilterConfig(), c2.getJavaSerializationFilterConfig());
         }
 
         private static boolean isCompatible(GlobalSerializerConfig c1, GlobalSerializerConfig c2) {

@@ -360,6 +360,11 @@ public void testSerializationConfig() {
                 .setClassName("SerializerClass")
                 .setTypeClassName("TypeClass");
 
+        JavaSerializationFilterConfig filterConfig = new JavaSerializationFilterConfig();
+        filterConfig.getBlacklist().addClasses("example.Class1", "acme.Test").addPackages("org.infinitban");
+        filterConfig.getWhitelist().addClasses("WhiteOne", "WhiteTwo").addPackages("com.hazelcast", "test.package");
+                
+
         SerializationConfig expectedConfig = new SerializationConfig()
                 .setAllowUnsafe(true)
                 .setPortableVersion(2)
@@ -369,6 +374,7 @@ public void testSerializationConfig() {
                 .setEnableCompression(true)
                 .setEnableSharedObject(true)
                 .setGlobalSerializerConfig(globalSerializerConfig)
+                .setJavaSerializationFilterConfig(filterConfig)
                 .addDataSerializableFactoryClass(10, "SerializableFactory")
                 .addPortableFactoryClass(10, "PortableFactory")
                 .addSerializerConfig(serializerConfig);
@@ -388,6 +394,7 @@ public void testSerializationConfig() {
         assertEquals(expectedConfig.getDataSerializableFactoryClasses(), actualConfig.getDataSerializableFactoryClasses());
         assertEquals(expectedConfig.getPortableFactoryClasses(), actualConfig.getPortableFactoryClasses());
         assertEquals(expectedConfig.getSerializerConfigs(), actualConfig.getSerializerConfigs());
+        assertEquals(expectedConfig.getJavaSerializationFilterConfig(), actualConfig.getJavaSerializationFilterConfig());
     }
 
     @Test

@@ -27,6 +27,8 @@
 import com.hazelcast.test.HazelcastTestSupport;
 import com.hazelcast.test.annotation.QuickTest;
 import com.hazelcast.topic.TopicOverloadPolicy;
+
+import org.hamcrest.CoreMatchers;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 import org.junit.runner.RunWith;
@@ -60,6 +62,7 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -2382,6 +2385,38 @@ public void testGlobalSerializer() {
         assertTrue(globalSerializerConfig.isOverrideJavaSerialization());
     }
 
+    @Test
+    public void testJavaSerializationFilter() {
+        String xml = HAZELCAST_START_TAG
+                + "  <serialization>\n"
+                + "      <java-serialization-filter>\n"
+                + "          <whitelist>\n"
+                + "              <class>java.lang.String</class>\n"
+                + "              <class>example.Foo</class>\n"
+                + "              <package>com.acme.app</package>\n"
+                + "              <package>com.acme.app.subpkg</package>\n"
+                + "          </whitelist>\n"
+                + "          <blacklist>\n"
+                + "              <class>com.acme.app.BeanComparator</class>\n"
+                + "          </blacklist>\n"
+                + "      </java-serialization-filter>\n"
+                + "  </serialization>\n"
+                + HAZELCAST_END_TAG;
+        
+        Config config = new InMemoryXmlConfig(xml);
+        JavaSerializationFilterConfig javaSerializationFilterConfig = config.getSerializationConfig().getJavaSerializationFilterConfig();
+        assertNotNull(javaSerializationFilterConfig);
+        ClassFilter blackList = javaSerializationFilterConfig.getBlacklist();
+        assertNotNull(blackList);
+        ClassFilter whiteList = javaSerializationFilterConfig.getWhitelist();
+        assertNotNull(whiteList);
+        assertTrue(whiteList.getClasses().contains("java.lang.String"));
+        assertTrue(whiteList.getClasses().contains("example.Foo"));
+        assertTrue(whiteList.getPackages().contains("com.acme.app"));
+        assertTrue(whiteList.getPackages().contains("com.acme.app.subpkg"));
+        assertTrue(blackList.getClasses().contains("com.acme.app.BeanComparator"));
+    }
+
     @Test
     public void testHotRestart() {
         String dir = "/mnt/hot-restart-root/";

@@ -0,0 +1,127 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.nio;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import com.hazelcast.config.JavaSerializationFilterConfig;
+import com.hazelcast.test.HazelcastParallelClassRunner;
+import com.hazelcast.test.annotation.QuickTest;
+
+/**
+ * Unit tests for {@link SerializationClassNameFilter}.
+ */
+@RunWith(HazelcastParallelClassRunner.class)
+@Category(QuickTest.class)
+public class SerializationClassNameFilterTest {
+
+    /**
+     * <pre>
+     * Given: Neither whitelist nor blacklist is configured.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called.
+     * Then: no exception is thrown
+     * </pre>
+     */
+    @Test
+    public void testNoList() {
+        JavaSerializationFilterConfig config = new JavaSerializationFilterConfig();
+        new SerializationClassNameFilter(config).filter("java.lang.Object");
+    }
+
+    /**
+     * <pre>
+     * Given: Default blacklist is used.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a class name which is included in the default blacklist.
+     * Then: {@link SecurityException} is thrown
+     * </pre>
+     */
+    @Test(expected = SecurityException.class)
+    public void testDefaultBlacklist() throws ClassNotFoundException {
+        new SerializationClassNameFilter(new JavaSerializationFilterConfig()).filter("bsh.XThis");
+    }
+
+    /**
+     * <pre>
+     * Given: Default blacklist is used.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a class in package which is included in the default blacklist.
+     * Then: {@link SecurityException} is thrown
+     * </pre>
+     */
+    @Test(expected = SecurityException.class)
+    public void testPackageDefaultBlacklisted() throws ClassNotFoundException {
+        new SerializationClassNameFilter(new JavaSerializationFilterConfig()).filter("org.apache.commons.collections.functors.Test");
+    }
+
+    /**
+     * <pre>
+     * Given: Whitelist is set.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a whitelisted class.
+     * Then: no exception is thrown
+     * </pre>
+     */
+    @Test
+    public void testClassInWhitelist() throws ClassNotFoundException {
+        JavaSerializationFilterConfig config = new JavaSerializationFilterConfig();
+        config.getWhitelist().addClasses("java.lang.Test1", "java.lang.Test2", "java.lang.Test3");
+        new SerializationClassNameFilter(config).filter("java.lang.Test2");
+    }
+
+    /**
+     * <pre>
+     * Given: Whitelist is set.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a class which has whitelisted package.
+     * Then: no exception is thrown
+     * </pre>
+     */
+    @Test
+    public void testPackageInWhitelist() throws ClassNotFoundException {
+        JavaSerializationFilterConfig config = new JavaSerializationFilterConfig();
+        config.getWhitelist().addPackages("com.whitelisted");
+        new SerializationClassNameFilter(config).filter("com.whitelisted.Test2");
+    }
+
+    /**
+     * <pre>
+     * Given: Whitelist is set.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a not whitelisted class.
+     * Then: {@link SecurityException} is thrown
+     * </pre>
+     */
+    @Test(expected = SecurityException.class)
+    public void testClassNotInWhitelist() throws ClassNotFoundException {
+        JavaSerializationFilterConfig config = new JavaSerializationFilterConfig();
+        config.getWhitelist().addClasses("java.lang.Test1", "java.lang.Test2", "java.lang.Test3");
+        new SerializationClassNameFilter(config).filter("java.lang.Test4");
+    }
+
+    /**
+     * <pre>
+     * Given: Blacklist and Whitelist are set.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a class which is whitelisted and blacklisted together.
+     * Then: {@link SecurityException} is thrown
+     * </pre>
+     */
+    @Test(expected = SecurityException.class)
+    public void testWhitelistedAndBlacklisted() throws ClassNotFoundException {
+        JavaSerializationFilterConfig config = new JavaSerializationFilterConfig();
+        config.getWhitelist().addClasses("java.lang.Test1", "java.lang.Test2", "java.lang.Test3");
+        config.getBlacklist().addClasses("java.lang.Test3", "java.lang.Test2", "java.lang.Test1");
+        new SerializationClassNameFilter(config).filter("java.lang.Test1");
+    }
+}
\ No newline at end of file

@@ -85,7 +85,7 @@ public void testGlobalSerializer_withOverrideJavaSerializable() {
         globalSerializerConfig.setOverrideJavaSerialization(true);
         final AtomicInteger writeCounter = new AtomicInteger();
         final AtomicInteger readCounter = new AtomicInteger();
-        final JavaSerializer javaSerializer = new JavaSerializer(true, false);
+        final JavaSerializer javaSerializer = new JavaSerializer(true, false, null);
         SerializationConfig serializationConfig = new SerializationConfig().setGlobalSerializerConfig(
                 globalSerializerConfig.setImplementation(new StreamSerializer<Object>() {
                     @Override

@@ -639,6 +639,17 @@
                         class-name="com.hazelcast.examples.SerializerFactory"/>
         </serializers>
         <check-class-def-errors>false</check-class-def-errors>
+        <java-serialization-filter>
+            <blacklist>
+                <class>com.acme.app.BeanComparator</class>
+            </blacklist>
+            <whitelist>
+                <class>java.lang.String</class>
+                <class>example.Foo</class>
+                <package>com.acme.app</package>
+                <package>com.acme.app.subpkg</package>
+            </whitelist>
+        </java-serialization-filter>
     </serialization>
 
     <native-memory allocator-type="STANDARD" enabled="true">