diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java b/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java index b19904de6..1e88aff75 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java @@ -13,7 +13,6 @@ import java.util.concurrent.CompletionStage; import java.util.concurrent.Flow; import java.util.concurrent.atomic.AtomicReference; -import java.util.regex.Pattern; import org.reactivestreams.FlowAdapters; import org.reactivestreams.Subscription; @@ -147,21 +146,6 @@ static BodyHandler boundedStringBodyHandler(int maxSize) { static class SseLineSubscriber extends BaseSubscriber { - /** - * Pattern to extract data content from SSE "data:" lines. - */ - private static final Pattern EVENT_DATA_PATTERN = Pattern.compile("^data:(.+)$", Pattern.MULTILINE); - - /** - * Pattern to extract event ID from SSE "id:" lines. - */ - private static final Pattern EVENT_ID_PATTERN = Pattern.compile("^id:(.+)$", Pattern.MULTILINE); - - /** - * Pattern to extract event type from SSE "event:" lines. - */ - private static final Pattern EVENT_TYPE_PATTERN = Pattern.compile("^event:(.+)$", Pattern.MULTILINE); - /** * The sink for emitting parsed response events. */ @@ -227,6 +211,31 @@ protected void hookOnSubscribe(Subscription subscription) { }); } + /** + * Extracts the value of an SSE field from a line, per the + * SSE specification: the characters after the colon with a single leading + * space removed. + * + *

+ * A value may legally contain U+2028 (LINE SEPARATOR), U+2029 (PARAGRAPH + * SEPARATOR) and U+0085 (NEXT LINE). Those are not SSE line terminators, so + * extracting the value with a {@code MULTILINE} regex instead of this method + * silently truncates it there. + * @param line the SSE line, already stripped of its terminator by the line + * subscriber + * @param field the field prefix, e.g. {@code "data:"} + * @return the field value with a single leading space removed, never truncated + * @see #hookOnNext(String) + */ + private static String fieldValue(String line, String field) { + String value = line.substring(field.length()); + if (value.startsWith(" ")) { + value = value.substring(1); + } + return value; + } + @Override protected void hookOnNext(String line) { if (line.isEmpty()) { @@ -241,35 +250,25 @@ protected void hookOnNext(String line) { } else { if (line.startsWith("data:")) { - var matcher = EVENT_DATA_PATTERN.matcher(line); - if (matcher.find()) { - String data = matcher.group(1).trim(); - // Measured before appending, so that an event carrying exactly - // maxSize of data is accepted: the trailing separator below is - // stripped again before the event is emitted. - if (this.eventBuilder.length() + data.length() > this.maxSize) { - upstream().cancel(); - this.sink.error( - new McpTransportException("Inbound SSE event exceeds the maximum allowed size of " - + this.maxSize + " bytes")); - return; - } - this.eventBuilder.append(data).append("\n"); + String data = fieldValue(line, "data:"); + // Measured before appending, so that an event carrying exactly + // maxSize of data is accepted: the trailing separator below is + // stripped again before the event is emitted. + if (this.eventBuilder.length() + data.length() > this.maxSize) { + upstream().cancel(); + this.sink.error(new McpTransportException( + "Inbound SSE event exceeds the maximum allowed size of " + this.maxSize + " bytes")); + return; } + this.eventBuilder.append(data).append("\n"); upstream().request(1); } else if (line.startsWith("id:")) { - var matcher = EVENT_ID_PATTERN.matcher(line); - if (matcher.find()) { - this.currentEventId.set(matcher.group(1).trim()); - } + this.currentEventId.set(fieldValue(line, "id:")); upstream().request(1); } else if (line.startsWith("event:")) { - var matcher = EVENT_TYPE_PATTERN.matcher(line); - if (matcher.find()) { - this.currentEventType.set(matcher.group(1).trim()); - } + this.currentEventType.set(fieldValue(line, "event:")); upstream().request(1); } else if (line.startsWith(":")) { diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/client/transport/ResponseSubscribersTest.java b/mcp-core/src/test/java/io/modelcontextprotocol/client/transport/ResponseSubscribersTest.java new file mode 100644 index 000000000..ae88cc2e1 --- /dev/null +++ b/mcp-core/src/test/java/io/modelcontextprotocol/client/transport/ResponseSubscribersTest.java @@ -0,0 +1,120 @@ +/* + * Copyright 2024-2026 the original author or authors. + */ + +package io.modelcontextprotocol.client.transport; + +import java.net.http.HttpClient; +import java.net.http.HttpHeaders; +import java.net.http.HttpResponse; +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for {@link ResponseSubscribers.SseLineSubscriber}. + * + *

+ * Verifies that SSE field values are extracted per the + * WHATWG HTML Living Standard §9.2.6: the field value is everything after the colon + * minus a single leading space. In particular, U+2028 (LINE SEPARATOR), U+2029 (PARAGRAPH + * SEPARATOR) and U+0085 (NEXT LINE) are legal inside a field value and must not truncate + * it — they are not SSE line terminators. + * + * @see #1136 + */ +class ResponseSubscribersTest { + + private static final HttpResponse.ResponseInfo RESPONSE_INFO = new HttpResponse.ResponseInfo() { + + @Override + public int statusCode() { + return 200; + } + + @Override + public HttpHeaders headers() { + return HttpHeaders.of(Map.of(), (name, value) -> true); + } + + @Override + public HttpClient.Version version() { + return HttpClient.Version.HTTP_1_1; + } + + }; + + private static List parse(List lines) { + return Flux.create(sink -> Flux.fromIterable(lines) + .subscribe(new ResponseSubscribers.SseLineSubscriber(RESPONSE_INFO, sink, Integer.MAX_VALUE))) + .map(event -> ((ResponseSubscribers.SseResponseEvent) event).sseEvent()) + .collectList() + .block(); + } + + /** + * A {@code data:} payload containing U+2028, U+2029 or U+0085 must survive parsing + * intact. A MULTILINE regex used to truncate the value at those characters, because + * the Java regex engine treats them as line terminators. + */ + @Test + void shouldNotTruncateDataAtUnicodeLineSeparators() { + List separators = List.of("\u2028", "\u2029", "\u0085"); + + for (String separator : separators) { + String payload = "{\"text\":\"a" + separator + "b\"}"; + List events = parse(List.of("data: " + payload, "")); + + assertThat(events).as("payload with U+%04X", (int) separator.charAt(0)).hasSize(1); + assertThat(events.get(0).data()).isEqualTo(payload); + } + } + + @Test + void shouldPreserveVerticalTabInData() { + String payload = "{\"text\":\"a\u000Bb\"}"; + + List events = parse(List.of("data: " + payload, "")); + + assertThat(events).hasSize(1); + assertThat(events.get(0).data()).isEqualTo(payload); + } + + @Test + void shouldStripOnlySingleLeadingSpacePerDataLine() { + // The leading space after the colon is stripped per line; any further + // whitespace is part of the value. The whole-event trim only affects the + // first/last line, so the interior line keeps its second space. + List events = parse(List.of("data: first", "data: second", "")); + + assertThat(events).hasSize(1); + assertThat(events.get(0).data()).isEqualTo("first\n second"); + } + + @Test + void shouldJoinMultipleDataLinesWithNewline() { + List events = parse(List.of("data: first", "data: second", "")); + + assertThat(events).hasSize(1); + assertThat(events.get(0).data()).isEqualTo("first\nsecond"); + } + + @Test + void shouldCaptureEventIdAndTypeWithUnicodeValue() { + List events = parse( + List.of("event: message\u2028tail", "id: 42\u2028tail", "data: body", "")); + + assertThat(events).hasSize(1); + assertThat(events.get(0).event()).isEqualTo("message\u2028tail"); + assertThat(events.get(0).id()).isEqualTo("42\u2028tail"); + assertThat(events.get(0).data()).isEqualTo("body"); + } + +}