Skip to content

fix: stop truncating SSE field values containing U+2028/U+2029/U+0085 #1142

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lzcGeek wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: stop truncating SSE field values containing U+2028/U+2029/U+0085 #1142

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 52 additions & 42 deletions mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -147,21 +146,6 @@ static BodyHandler<String> boundedStringBodyHandler(int maxSize) {

static class SseLineSubscriber extends BaseSubscriber<String> {

/**
* 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.
*/
Expand Down Expand Up @@ -227,49 +211,75 @@ protected void hookOnSubscribe(Subscription subscription) {
});
}

/**
* Extracts the value of an SSE field from a line, per the <a href=
* "https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation">
* SSE specification</a>: the characters after the colon with a single leading
* space removed.
*
* <p>
* 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;
}

/**
* Returns the buffered data lines joined by the separators the data: handler
* appended, with only the final separator removed. Trimming the whole buffer
* instead would also strip significant leading/trailing whitespace from the first
* and last data lines, which the SSE field rules explicitly preserve.
*/
private String concatenatedDataLines() {
String buffered = this.eventBuilder.toString();
return buffered.substring(0, buffered.length() - 1);
}

@Override
protected void hookOnNext(String line) {
if (line.isEmpty()) {
// Empty line means end of event
if (this.eventBuilder.length() > 0) {
String eventData = this.eventBuilder.toString();
SseEvent sseEvent = new SseEvent(currentEventId.get(), currentEventType.get(), eventData.trim());
String eventData = concatenatedDataLines();
SseEvent sseEvent = new SseEvent(currentEventId.get(), currentEventType.get(), eventData);

this.sink.next(new SseResponseEvent(responseInfo, sseEvent));
this.eventBuilder.setLength(0);
}
}
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(":")) {
Expand All @@ -290,8 +300,8 @@ else if (line.startsWith(":")) {
@Override
protected void hookOnComplete() {
if (this.eventBuilder.length() > 0) {
String eventData = this.eventBuilder.toString();
SseEvent sseEvent = new SseEvent(currentEventId.get(), currentEventType.get(), eventData.trim());
String eventData = concatenatedDataLines();
SseEvent sseEvent = new SseEvent(currentEventId.get(), currentEventType.get(), eventData);
this.sink.next(new SseResponseEvent(responseInfo, sseEvent));
}
this.sink.complete();
Expand Down
140 changes: 140 additions & 0 deletions mcp-core/src/test/java/io/modelcontextprotocol/client/transport/ResponseSubscribersTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,140 @@
/*
* 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}.
*
* <p>
* Verifies that SSE field values are extracted per the <a href=
* "https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation">
* WHATWG HTML Living Standard §9.2.6</a>: 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 <a href="https://github.com/modelcontextprotocol/java-sdk/issues/1136">#1136</a>
*/
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<ResponseSubscribers.SseEvent> parse(List<String> lines) {
return Flux.<ResponseSubscribers.ResponseEvent>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<String> separators = List.of("\u2028", "\u2029", "\u0085");

for (String separator : separators) {
String payload = "{\"text\":\"a" + separator + "b\"}";
List<ResponseSubscribers.SseEvent> 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<ResponseSubscribers.SseEvent> 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.
List<ResponseSubscribers.SseEvent> events = parse(List.of("data: first", "data: second", ""));

assertThat(events).hasSize(1);
assertThat(events.get(0).data()).isEqualTo("first\n second");
}

/**
* Only the separator the {@code data:} handler appended after the last line may be
* removed when the event is dispatched; trimming the whole buffer would also strip
* significant whitespace from the first and last data lines.
*/
@Test
void shouldNotTrimSignificantWhitespaceOfSingleDataLine() {
List<ResponseSubscribers.SseEvent> events = parse(List.of("data: padded ", ""));

assertThat(events).hasSize(1);
assertThat(events.get(0).data()).isEqualTo(" padded ");
}

@Test
void shouldPreserveLeadingAndTrailingWhitespaceOfFirstAndLastDataLines() {
List<ResponseSubscribers.SseEvent> events = parse(List.of("data: first", "data: last ", ""));

assertThat(events).hasSize(1);
assertThat(events.get(0).data()).isEqualTo(" first\nlast ");
}

@Test
void shouldJoinMultipleDataLinesWithNewline() {
List<ResponseSubscribers.SseEvent> events = parse(List.of("data: first", "data: second", ""));

assertThat(events).hasSize(1);
assertThat(events.get(0).data()).isEqualTo("first\nsecond");
}

@Test
void shouldCaptureEventIdAndTypeWithUnicodeValue() {
List<ResponseSubscribers.SseEvent> 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");
}

}