INT-1926: Option to disallow arbitrary routing

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

Add an option to mapping routers to disable falling back to the
channel key as the channel name.

spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java

@@ -109,6 +109,21 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
 		return _this();
 	}
 
+	/**
+	 * By default, if a resolved channel key does not exist in the channel map, the key
+	 * itself is used as the channel name, which we will attempt to resolve to a channel.
+	 * Invoke this method to disable this feature. This could be useful to prevent
+	 * malicious actors from generating a message that could cause the message to be
+	 * routed to an unexpected channel, such as one upstream of the router, which would
+	 * cause a stack overflow.
+	 * @return the router spec.
+	 * @since 5.2
+	 */
+	public RouterSpec<K, R> noChannelKeyFallback() {
+		this.handler.setChannelKeyFallback(false);
+		return _this();
+	}
+
 	/**
 	 * @param key the key.
 	 * @param channelName the channelName.

spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java

@@ -74,6 +74,8 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter
 
 	private boolean resolutionRequired = true;
 
+	private boolean channelKeyFallback = true;
+
 	private volatile Map<String, String> channelMappings = new LinkedHashMap<>();
 
 
@@ -116,6 +118,20 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter
 		this.resolutionRequired = resolutionRequired;
 	}
 
+	/**
+	 * When true (default), if a resolved channel key does not exist in the channel map,
+	 * the key itself is used as the channel name, which we will attempt to resolve to a
+	 * channel. Set to false to disable this feature. This could be useful to prevent
+	 * malicious actors from generating a message that could cause the message to be
+	 * routed to an unexpected channel, such as one upstream of the router, which would
+	 * cause a stack overflow.
+	 * @param channelKeyFallback false to disable the fall back.
+	 * @since 5.2
+	 */
+	public void setChannelKeyFallback(boolean channelKeyFallback) {
+		this.channelKeyFallback = channelKeyFallback;
+	}
+
 	/**
 	 * Set a limit for how many dynamic channels are retained (for reporting purposes).
 	 * When the limit is exceeded, the oldest channel is discarded.
@@ -241,23 +257,25 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter
 
 		// if the channelMappings contains a mapping, we'll use the mapped value
 		// otherwise, the String-based channelKey itself will be used as the channel name
-		String channelName = channelKey;
+		String channelName = this.channelKeyFallback ? channelKey : null;
 		boolean mapped = false;
 		if (this.channelMappings.containsKey(channelKey)) {
 			channelName = this.channelMappings.get(channelKey);
 			mapped = true;
 		}
-		if (this.prefix != null) {
-			channelName = this.prefix + channelName;
-		}
-		if (this.suffix != null) {
-			channelName = channelName + this.suffix;
-		}
-		MessageChannel channel = resolveChannelForName(channelName, message);
-		if (channel != null) {
-			channels.add(channel);
-			if (!mapped && this.dynamicChannels.get(channelName) == null) {
-				this.dynamicChannels.put(channelName, channel);
+		if (channelName != null) {
+			if (this.prefix != null) {
+				channelName = this.prefix + channelName;
+			}
+			if (this.suffix != null) {
+				channelName = channelName + this.suffix;
+			}
+			MessageChannel channel = resolveChannelForName(channelName, message);
+			if (channel != null) {
+				channels.add(channel);
+				if (!mapped && this.dynamicChannels.get(channelName) == null) {
+					this.dynamicChannels.put(channelName, channel);
+				}
 			}
 		}
 	}

spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/NoFallbackAllowedTests.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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.springframework.integration.dsl.routers;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 5.2
+ *
+ */
+@SpringJUnitConfig
+public class NoFallbackAllowedTests {
+
+	@Test
+	void noStackOverflow(@Autowired Config config) {
+		config.flow()
+				.getInputChannel()
+				.send(new GenericMessage<>("foo", Collections.singletonMap("whereTo", "flow.input")));
+		assertThat(config.queue().receive(0)).isNotNull();
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class Config {
+
+		@Bean
+		public IntegrationFlow flow() {
+			return f -> f.route("headers.whereTo", r -> r
+					.noChannelKeyFallback()
+					.defaultOutputChannel(queue()));
+		}
+
+		@Bean
+		public QueueChannel queue() {
+			return new QueueChannel();
+		}
+
+	}
+
+}