INT-4342: White List for Payload Deserializer

JIRA: https://jira.spring.io/browse/INT-4342

Use similar code to Spring AMQP to add white list support for Integration's
use of the `DeserializingMessageConverter`; introduce the `WhiteListDeserializingMessageConverter`.

Polishing

Missed this change in PR.

Fix XSD attribute

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java

@@ -38,7 +38,6 @@ import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
-import org.springframework.core.serializer.support.DeserializingConverter;
 import org.springframework.core.serializer.support.SerializingConverter;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -51,6 +50,7 @@ import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.integration.store.SimpleMessageGroupFactory;
+import org.springframework.integration.support.converter.WhiteListDeserializingConverter;
 import org.springframework.integration.transaction.TransactionSynchronizationFactory;
 import org.springframework.integration.util.UUIDConverter;
 import org.springframework.jdbc.core.JdbcOperations;
@@ -137,7 +137,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
 
 	private volatile JdbcTemplate jdbcTemplate;
 
-	private volatile DeserializingConverter deserializer;
+	private volatile WhiteListDeserializingConverter deserializer;
 
 	private volatile SerializingConverter serializer;
 
@@ -159,7 +159,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
 	 * Convenient constructor for configuration use.
 	 */
 	public JdbcChannelMessageStore() {
-		this.deserializer = new DeserializingConverter();
+		this.deserializer = new WhiteListDeserializingConverter();
 		this.serializer = new SerializingConverter();
 	}
 
@@ -202,7 +202,18 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
 	 */
 	@SuppressWarnings({"unchecked", "rawtypes"})
 	public void setDeserializer(Deserializer<? extends Message<?>> deserializer) {
-		this.deserializer = new DeserializingConverter((Deserializer) deserializer);
+		this.deserializer = new WhiteListDeserializingConverter((Deserializer) deserializer);
+	}
+
+	/**
+	 * Add patterns for packages/classes that are allowed to be deserialized. A class can
+	 * be fully qualified or a wildcard '*' is allowed at the beginning or end of the
+	 * class name. Examples: {@code com.foo.*}, {@code *.MyClass}.
+	 * @param patterns the patterns.
+	 * @since 4.2.13
+	 */
+	public void addWhiteListPatterns(String... patterns) {
+		this.deserializer.addWhiteListPatterns(patterns);
 	}
 
 	/**

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java

@@ -37,7 +37,6 @@ import org.apache.commons.logging.LogFactory;
 
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
-import org.springframework.core.serializer.support.DeserializingConverter;
 import org.springframework.core.serializer.support.SerializingConverter;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.integration.store.AbstractMessageGroupStore;
@@ -45,6 +44,7 @@ import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageMetadata;
 import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.SimpleMessageGroup;
+import org.springframework.integration.support.converter.WhiteListDeserializingConverter;
 import org.springframework.integration.util.UUIDConverter;
 import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.core.JdbcTemplate;
@@ -189,7 +189,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 
 	private final JdbcOperations jdbcTemplate;
 
-	private volatile DeserializingConverter deserializer;
+	private volatile WhiteListDeserializingConverter deserializer;
 
 	private volatile SerializingConverter serializer;
 
@@ -213,7 +213,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 	public JdbcMessageStore(JdbcOperations jdbcOperations) {
 		Assert.notNull(jdbcOperations, "'dataSource' must not be null");
 		this.jdbcTemplate = jdbcOperations;
-		this.deserializer = new DeserializingConverter();
+		this.deserializer = new WhiteListDeserializingConverter();
 		this.serializer = new SerializingConverter();
 	}
 
@@ -265,7 +265,18 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 	 */
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	public void setDeserializer(Deserializer<? extends Message<?>> deserializer) {
-		this.deserializer = new DeserializingConverter((Deserializer) deserializer);
+		this.deserializer = new WhiteListDeserializingConverter((Deserializer) deserializer);
+	}
+
+	/**
+	 * Add patterns for packages/classes that are allowed to be deserialized. A class can
+	 * be fully qualified or a wildcard '*' is allowed at the beginning or end of the
+	 * class name. Examples: {@code com.foo.*}, {@code *.MyClass}.
+	 * @param patterns the patterns.
+	 * @since 4.2.13
+	 */
+	public void addWhiteListPatterns(String... patterns) {
+		this.deserializer.addWhiteListPatterns(patterns);
 	}
 
 	@Override

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MessageRowMapper.java

@@ -19,7 +19,7 @@ package org.springframework.integration.jdbc.store.channel;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 
-import org.springframework.core.serializer.support.DeserializingConverter;
+import org.springframework.integration.support.converter.WhiteListDeserializingConverter;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.support.lob.LobHandler;
 import org.springframework.messaging.Message;
@@ -30,20 +30,22 @@ import org.springframework.messaging.Message;
  * that select clause ordering is unimportant.
  *
  * @author Gunnar Hillert
+ * @author Gary Russell
  * @since 2.2
  *
  */
 public class MessageRowMapper implements RowMapper<Message<?>> {
 
-	private final DeserializingConverter deserializer;
+	private final WhiteListDeserializingConverter deserializer;
 
 	private final LobHandler lobHandler;
 
-	public MessageRowMapper(DeserializingConverter deserializer, LobHandler lobHandler) {
+	public MessageRowMapper(WhiteListDeserializingConverter deserializer, LobHandler lobHandler) {
 		this.deserializer = deserializer;
 		this.lobHandler = lobHandler;
 	}
 
+	@Override
 	public Message<?> mapRow(ResultSet rs, int rowNum) throws SQLException {
 		return (Message<?>) this.deserializer.convert(this.lobHandler.getBlobAsBytes(rs, "MESSAGE_BYTES"));
 	}