fix: validate resource URIs and cap subscriptions per session

Author: ez-lbz

mcp-core/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java

@@ -92,6 +92,12 @@ public class McpAsyncServer {
 
 	private static final Logger logger = LoggerFactory.getLogger(McpAsyncServer.class);
 
+	/**
+	 * Maximum number of resource URIs a single session may subscribe to. Prevents a
+	 * client from growing the resource subscriptions map without bound.
+	 */
+	private static final int MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION = 1024;
+
 	private final McpServerTransportProviderBase mcpTransportProvider;
 
 	private final McpJsonMapper jsonMapper;
@@ -758,6 +764,25 @@ private McpRequestHandler<Object> resourcesSubscribeRequestHandler() {
 					});
 			String uri = subscribeRequest.uri();
 			String sessionId = exchange.sessionId();
+
+			// Validate the URI against the registered resources and resource templates
+			// before accepting the subscription.
+			if (this.findResourceSpecification(uri).isEmpty() && this.findResourceTemplateSpecification(uri).isEmpty()) {
+				return Mono.error(RESOURCE_NOT_FOUND.apply(uri));
+			}
+
+			// Cap the number of resource subscriptions per session to prevent a client
+			// from growing the subscriptions map without bound.
+			long sessionSubscriptions = this.resourceSubscriptions.values()
+				.stream()
+				.filter(subscribedSessions -> subscribedSessions.contains(sessionId))
+				.count();
+			if (sessionSubscriptions >= MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION) {
+				return Mono.error(McpError.builder(McpSchema.ErrorCodes.INVALID_PARAMS)
+					.message("Resource subscription limit exceeded for session: " + sessionId)
+					.build());
+			}
+
 			this.resourceSubscriptions.computeIfAbsent(uri, k -> Collections.newSetFromMap(new ConcurrentHashMap<>()))
 				.add(sessionId);
 			logger.debug("Session {} subscribed to resource URI: {}", sessionId, uri);

mcp-test/src/test/java/io/modelcontextprotocol/server/ResourceSubscriptionTests.java

@@ -4,13 +4,16 @@
 
 package io.modelcontextprotocol.server;
 
+import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 
 import io.modelcontextprotocol.MockMcpServerTransport;
 import io.modelcontextprotocol.MockMcpServerTransportProvider;
 import io.modelcontextprotocol.spec.McpSchema;
 import io.modelcontextprotocol.spec.ProtocolVersions;
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -59,6 +62,31 @@ private static McpSchema.JSONRPCRequest unsubscribeRequest(String uri) {
 				McpSchema.UnsubscribeRequest.builder(uri).build());
 	}
 
+	private static McpAsyncServer buildServerWithResource(MockMcpServerTransportProvider transportProvider) {
+		McpSchema.Resource resource = McpSchema.Resource.builder(RESOURCE_URI, "Test Resource")
+			.mimeType("text/plain")
+			.build();
+		return McpServer.async(transportProvider)
+			.serverInfo(SERVER_INFO)
+			.capabilities(McpSchema.ServerCapabilities.builder().resources(true, false).build())
+			.resources(new McpServerFeatures.AsyncResourceSpecification(resource,
+					(exchange, request) -> Mono.just(new McpSchema.ReadResourceResult(List.of()))))
+			.build();
+	}
+
+	private static McpAsyncServer buildServerWithResourceTemplate(MockMcpServerTransportProvider transportProvider) {
+		McpSchema.ResourceTemplate resourceTemplate = McpSchema.ResourceTemplate
+			.builder("test://resource/{id}", "Test Resource Template")
+			.mimeType("text/plain")
+			.build();
+		return McpServer.async(transportProvider)
+			.serverInfo(SERVER_INFO)
+			.capabilities(McpSchema.ServerCapabilities.builder().resources(true, false).build())
+			.resourceTemplates(new McpServerFeatures.AsyncResourceTemplateSpecification(resourceTemplate,
+					(exchange, request) -> Mono.just(new McpSchema.ReadResourceResult(List.of()))))
+			.build();
+	}
+
 	@Test
 	void notifyResourcesUpdated_noSubscribers_completesEmpty() {
 		MockMcpServerTransport transport = new MockMcpServerTransport();
@@ -166,4 +194,69 @@ void notifyResourcesUpdated_afterSessionClose_doesNotNotifySession() {
 		server.closeGracefully().block();
 	}
 
+	@Test
+	void subscribeToRegisteredUri_succeeds() {
+		MockMcpServerTransport transport = new MockMcpServerTransport();
+		MockMcpServerTransportProvider transportProvider = new MockMcpServerTransportProvider(transport);
+		McpAsyncServer server = buildServerWithResource(transportProvider);
+
+		transportProvider.simulateIncomingMessage(initRequest());
+		transportProvider.simulateIncomingMessage(initializedNotification());
+		transport.clearSentMessages();
+		transportProvider.simulateIncomingMessage(subscribeRequest(RESOURCE_URI));
+
+		McpSchema.JSONRPCMessage sent = transport.getLastSentMessage();
+		assertThat(sent).isInstanceOf(McpSchema.JSONRPCResponse.class);
+		McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sent;
+		assertThat(response.error()).as("subscribing to a registered resource must succeed").isNull();
+		assertThat(response.result()).isEqualTo(Map.of());
+
+		server.closeGracefully().block();
+	}
+
+	@Test
+	void subscribeToUnregisteredUri_returnsResourceNotFound() {
+		MockMcpServerTransport transport = new MockMcpServerTransport();
+		MockMcpServerTransportProvider transportProvider = new MockMcpServerTransportProvider(transport);
+		McpAsyncServer server = buildServerWithResource(transportProvider);
+
+		transportProvider.simulateIncomingMessage(initRequest());
+		transportProvider.simulateIncomingMessage(initializedNotification());
+		transport.clearSentMessages();
+		transportProvider.simulateIncomingMessage(subscribeRequest("test://unknown/uri"));
+
+		McpSchema.JSONRPCMessage sent = transport.getLastSentMessage();
+		assertThat(sent).isInstanceOf(McpSchema.JSONRPCResponse.class);
+		McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sent;
+		assertThat(response.error()).isNotNull();
+		assertThat(response.error().code()).isEqualTo(McpSchema.ErrorCodes.RESOURCE_NOT_FOUND);
+
+		server.closeGracefully().block();
+	}
+
+	@Test
+	void subscribeBeyondPerSessionLimit_isRejected() {
+		MockMcpServerTransport transport = new MockMcpServerTransport();
+		MockMcpServerTransportProvider transportProvider = new MockMcpServerTransportProvider(transport);
+		McpAsyncServer server = buildServerWithResourceTemplate(transportProvider);
+
+		transportProvider.simulateIncomingMessage(initRequest());
+		transportProvider.simulateIncomingMessage(initializedNotification());
+		transport.clearSentMessages();
+
+		for (int i = 0; i < 1024; i++) {
+			transportProvider.simulateIncomingMessage(subscribeRequest("test://resource/" + i));
+		}
+
+		transportProvider.simulateIncomingMessage(subscribeRequest("test://resource/1024"));
+
+		McpSchema.JSONRPCMessage sent = transport.getLastSentMessage();
+		assertThat(sent).isInstanceOf(McpSchema.JSONRPCResponse.class);
+		McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sent;
+		assertThat(response.error()).isNotNull();
+		assertThat(response.error().code()).isEqualTo(McpSchema.ErrorCodes.INVALID_PARAMS);
+
+		server.closeGracefully().block();
+	}
+
 }