diff --git a/docs/content/dev/client.md b/docs/content/dev/client.md index e1668d873..68e962c81 100644 --- a/docs/content/dev/client.md +++ b/docs/content/dev/client.md @@ -185,6 +185,35 @@ Client client = Client .build(); ``` +### SSE Parser Configuration + +The client uses a Server-Sent Events (SSE) parser for streaming responses. You can tune its limits via `SSEParserConfig` to handle agents that return large payloads (e.g., large artifacts or tool results): + +| Parameter | Description | Default | +|------------------|----------------------------------------------------------|-----------| +| `maxLineLength` | Max bytes per raw SSE line (`0` = disabled) | 1 MB | +| `maxBufferLines` | Max `data:` lines per event block | 1 000 | +| `maxBufferChars` | Max total characters across all `data:` values per event | 1 MB | + +The defaults are suitable for most deployments. To override them, build a custom `SSEParserConfig` and pass it to the HTTP client. + +> **Security note:** setting `maxLineLength` to `0` disables the per-line memory protection entirely. Without this limit, a malicious server sending a single unterminated line (no newline) can consume unbounded memory — `maxBufferChars` only limits accumulated `data:` field values _after_ lines have been decoded, not the raw line itself. Only disable the per-line check when you trust the remote agent or have other safeguards (e.g. a reverse proxy with its own line-length limit). Note that `maxLineLength` is enforced as a byte limit on raw UTF-8 data; for ASCII content (the common case) bytes and characters are equivalent. + +```java +SSEParserConfig sseConfig = SSEParserConfig.builder() + .maxLineLength(4 * 1024 * 1024) // 4 MB per line + .maxBufferChars(4 * 1024 * 1024) // 4 MB per event + .build(); + +// Pass to JdkA2AHttpClient, then to your transport config +JdkA2AHttpClient httpClient = JdkA2AHttpClient.withSseConfig(sseConfig); + +Client client = Client + .builder(agentCard) + .withTransport(JSONRPCTransport.class, new JSONRPCTransportConfig(httpClient)) + .build(); +``` + ## Observability (Optional) Add distributed tracing and W3C Trace Context propagation to client calls with the [OpenTelemetry extras modules](extra/opentelemetry#client). diff --git a/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java b/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java index c78389620..15c6954be 100644 --- a/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java +++ b/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java @@ -1,5 +1,7 @@ package org.a2aproject.sdk.client.http.android; +import static org.a2aproject.sdk.util.Assert.checkNotNullParam; + import static java.net.HttpURLConnection.HTTP_FORBIDDEN; import static java.net.HttpURLConnection.HTTP_MULT_CHOICE; import static java.net.HttpURLConnection.HTTP_OK; @@ -27,7 +29,9 @@ import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.A2AHttpHeaders; import org.a2aproject.sdk.client.http.A2AHttpResponse; +import org.a2aproject.sdk.client.http.BoundedLineAccumulator; import org.a2aproject.sdk.client.http.ServerSentEvent; +import org.a2aproject.sdk.client.http.SSEParserConfig; import org.a2aproject.sdk.client.http.ServerSentEventParser; import org.a2aproject.sdk.common.A2AErrorMessages; import org.a2aproject.sdk.spec.A2AClientHTTPError; @@ -48,24 +52,44 @@ public class AndroidA2AHttpClient implements A2AHttpClient { return t; }); + private final SSEParserConfig sseParserConfig; + + public AndroidA2AHttpClient() { + this(SSEParserConfig.DEFAULT); + } + + /** + * Creates a new Android HTTP client with custom SSE parser limits. + * + * @param sseParserConfig the SSE parser configuration to use for streaming responses + */ + public AndroidA2AHttpClient(SSEParserConfig sseParserConfig) { + this.sseParserConfig = checkNotNullParam("sseParserConfig", sseParserConfig); + } + @Override public GetBuilder createGet() { - return new AndroidGetBuilder(); + return new AndroidGetBuilder(sseParserConfig); } @Override public PostBuilder createPost() { - return new AndroidPostBuilder(); + return new AndroidPostBuilder(sseParserConfig); } @Override public DeleteBuilder createDelete() { - return new AndroidDeleteBuilder(); + return new AndroidDeleteBuilder(sseParserConfig); } private abstract static class AndroidBuilder> implements Builder { protected String url = ""; protected Map headers = new HashMap<>(); + protected final SSEParserConfig sseParserConfig; + + AndroidBuilder(SSEParserConfig sseParserConfig) { + this.sseParserConfig = sseParserConfig; + } @Override public T url(String url) { @@ -137,6 +161,52 @@ protected static String readStreamWithLimit(InputStream is) throws IOException { } } + /** + * Reads the next line from {@code is}, treating LF, CRLF, and bare CR as terminators + * per the SSE specification, and enforcing the byte limit while accumulating + * bytes — not after a full line has been materialised. + * + *

The caller must provide a {@link BoundedLineAccumulator} and call + * {@link BoundedLineAccumulator#reset()} between lines. After this method returns + * {@code true}, check {@link BoundedLineAccumulator#isTooLong()} to distinguish + * between a valid line and one that exceeded the limit. + * + *

The {@code is} parameter must support {@link InputStream#mark(int)} (e.g. + * {@link java.io.BufferedInputStream}) so that CR at end-of-buffer can be resolved + * without consuming the following byte. + * + * @param is the input stream to read from (must support mark/reset) + * @param accumulator the accumulator to use for byte collection and limit enforcement + * @return {@code true} if a line was read, {@code false} at end-of-stream + */ + static boolean readBoundedLine(InputStream is, BoundedLineAccumulator accumulator) throws IOException { + boolean hasReadAnyBytes = false; + + while (true) { + int b = is.read(); + if (b == -1) { + return hasReadAnyBytes || accumulator.isTooLong(); + } + hasReadAnyBytes = true; + if (b == '\n') { + return true; + } + if (b == '\r') { + consumeOptionalLF(is); + return true; + } + accumulator.addByte(b); + } + } + + private static void consumeOptionalLF(InputStream is) throws IOException { + is.mark(1); + int next = is.read(); + if (next != '\n' && next != -1) { + is.reset(); + } + } + protected A2AHttpResponse execute(HttpURLConnection connection) throws IOException { int status = connection.getResponseCode(); A2AHttpHeaders responseHeaders = fromConnectionHeaders(connection.getHeaderFields()); @@ -199,29 +269,11 @@ protected void processSSEResponse( String contentType = connection.getContentType(); boolean isSse = contentType != null && contentType.contains(EVENT_STREAM); - try (InputStream is = connection.getInputStream(); - BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) { - String line; + try (InputStream is = connection.getInputStream()) { if (isSse) { - ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer); - while ((line = reader.readLine()) != null) { - sseParser.processLine(line); - } - sseParser.flush(); + readSSEStream(is, sseParserConfig, messageConsumer, errorConsumer); } else { - StringBuilder bodyBuffer = new StringBuilder(); - while ((line = reader.readLine()) != null) { - if (!line.isEmpty()) { - if (bodyBuffer.length() > 0) { - bodyBuffer.append('\n'); - } - bodyBuffer.append(line); - } - } - String body = bodyBuffer.toString(); - if (!body.isEmpty()) { - messageConsumer.accept(new ServerSentEvent(body)); - } + readNonSSEStream(is, messageConsumer); } completeRunnable.run(); } @@ -232,6 +284,45 @@ protected void processSSEResponse( } } + private static void readSSEStream( + InputStream is, + SSEParserConfig config, + Consumer messageConsumer, + Consumer errorConsumer) throws IOException { + InputStream buffered = new java.io.BufferedInputStream(is); + BoundedLineAccumulator accumulator = new BoundedLineAccumulator(config.maxLineLength()); + ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, config); + while (readBoundedLine(buffered, accumulator)) { + if (accumulator.isTooLong()) { + sseParser.processLineTooLong(); + } else { + sseParser.processLine(accumulator.toLine()); + } + accumulator.reset(); + } + sseParser.flush(); + } + + private static void readNonSSEStream( + InputStream is, + Consumer messageConsumer) throws IOException { + StringBuilder bodyBuffer = new StringBuilder(); + BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8)); + String line; + while ((line = reader.readLine()) != null) { + if (!line.isEmpty()) { + if (bodyBuffer.length() > 0) { + bodyBuffer.append('\n'); + } + bodyBuffer.append(line); + } + } + String body = bodyBuffer.toString(); + if (!body.isEmpty()) { + messageConsumer.accept(new ServerSentEvent(body)); + } + } + protected CompletableFuture executeAsyncSSE( HttpURLConnection connection, Consumer messageConsumer, @@ -244,6 +335,10 @@ protected CompletableFuture executeAsyncSSE( } private static class AndroidGetBuilder extends AndroidBuilder implements GetBuilder { +AndroidGetBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public A2AHttpResponse get() throws IOException { HttpURLConnection connection = createConnection("GET", false); @@ -271,6 +366,10 @@ private static class AndroidPostBuilder extends AndroidBuilder private String body = ""; private boolean followRedirects = false; +AndroidPostBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public PostBuilder body(String body) { this.body = body; @@ -325,6 +424,10 @@ public CompletableFuture postAsyncSSE( private static class AndroidDeleteBuilder extends AndroidBuilder implements DeleteBuilder { +AndroidDeleteBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public A2AHttpResponse delete() throws IOException { HttpURLConnection connection = createConnection("DELETE", false); diff --git a/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java b/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java index 2cb173385..a6e260422 100644 --- a/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java +++ b/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java @@ -2,6 +2,7 @@ import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.A2AHttpClientProvider; +import org.a2aproject.sdk.client.http.SSEParserConfig; /** * Service provider for {@link AndroidA2AHttpClient}. @@ -24,6 +25,25 @@ public A2AHttpClient create() { return new AndroidA2AHttpClient(); } + /** + * {@inheritDoc} + * + * @throws IllegalStateException if the Android runtime is not available + */ + @Override + public A2AHttpClient createWithSseConfig(SSEParserConfig sseParserConfig) { + if (!ANDROID_AVAILABLE) { + throw new IllegalStateException( + "Android classes are not available. This provider is only supported on Android."); + } + return new AndroidA2AHttpClient(sseParserConfig); + } + + @Override + public boolean supportsSseConfig() { + return true; + } + @Override public int priority() { return ANDROID_AVAILABLE ? 110 : -1; // Higher priority than Vert.x on Android diff --git a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProviderTest.java b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProviderTest.java new file mode 100644 index 000000000..23aa3d8b1 --- /dev/null +++ b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProviderTest.java @@ -0,0 +1,49 @@ +package org.a2aproject.sdk.client.http.android; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.a2aproject.sdk.client.http.A2AHttpClient; +import org.a2aproject.sdk.client.http.SSEParserConfig; +import org.junit.jupiter.api.Test; + +/** + * Tests for {@link AndroidA2AHttpClientProvider}. + * + *

The surefire configuration for this module sets {@code java.runtime.name} to + * {@code "Android Runtime"}, so the provider treats the environment as Android. + */ +public class AndroidA2AHttpClientProviderTest { + + @Test + public void testCreateWithSseConfigReturnsAndroidClient() { + AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(4 * 1024 * 1024).build(); + + A2AHttpClient client = provider.createWithSseConfig(config); + assertNotNull(client); + assertInstanceOf(AndroidA2AHttpClient.class, client, + "Provider should return AndroidA2AHttpClient when Android runtime is available"); + } + + @Test + public void testSupportsSseConfig() { + AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider(); + assertTrue(provider.supportsSseConfig(), "Android provider must support SSE config"); + } + + @Test + public void testProviderName() { + AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider(); + assertEquals("android", provider.name()); + } + + @Test + public void testPriorityOnAndroid() { + AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider(); + assertEquals(110, provider.priority(), + "Android provider should have priority 110 when Android runtime is available"); + } +} diff --git a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java index 963545e85..18b5992c9 100644 --- a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java +++ b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java @@ -2,6 +2,7 @@ import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.AbstractA2AHttpClientSSETest; +import org.a2aproject.sdk.client.http.SSEParserConfig; public class AndroidA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { @@ -9,4 +10,9 @@ public class AndroidA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { protected A2AHttpClient createClient() { return new AndroidA2AHttpClient(); } + + @Override + protected A2AHttpClient createClient(SSEParserConfig sseParserConfig) { + return new AndroidA2AHttpClient(sseParserConfig); + } } diff --git a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidBoundedLineReaderTest.java b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidBoundedLineReaderTest.java new file mode 100644 index 000000000..f96d5387e --- /dev/null +++ b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidBoundedLineReaderTest.java @@ -0,0 +1,214 @@ +package org.a2aproject.sdk.client.http.android; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.BufferedInputStream; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import org.a2aproject.sdk.client.http.BoundedLineAccumulator; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for the bounded line reader inside {@link AndroidA2AHttpClient}. + * Since {@code readBoundedLine} is inside a private inner class, we access it via reflection. + */ +public class AndroidBoundedLineReaderTest { + + private static final Method READ_BOUNDED_LINE; + + static { + try { + Class builderClass = Class.forName( + "org.a2aproject.sdk.client.http.android.AndroidA2AHttpClient$AndroidBuilder"); + READ_BOUNDED_LINE = builderClass.getDeclaredMethod( + "readBoundedLine", InputStream.class, BoundedLineAccumulator.class); + READ_BOUNDED_LINE.setAccessible(true); + } catch (Exception e) { + throw new ExceptionInInitializerError(e); + } + } + + private static boolean readBoundedLineRaw(InputStream is, BoundedLineAccumulator accumulator) throws Exception { + try { + return (Boolean) READ_BOUNDED_LINE.invoke(null, is, accumulator); + } catch (java.lang.reflect.InvocationTargetException e) { + Throwable cause = e.getCause(); + if (cause instanceof Exception) { + throw (Exception) cause; + } + throw e; + } + } + + private static InputStream stream(String data) { + return new BufferedInputStream( + new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8))); + } + + private static List readAllLines(InputStream is, int maxLineBytes) throws Exception { + BoundedLineAccumulator acc = new BoundedLineAccumulator(maxLineBytes); + List lines = new ArrayList<>(); + while (readBoundedLineRaw(is, acc)) { + if (!acc.isTooLong()) { + lines.add(acc.toLine()); + } + acc.reset(); + } + return lines; + } + + @Test + public void testBasicLFLines() throws Exception { + List lines = readAllLines(stream("hello\nworld\n"), 0); + assertEquals(List.of("hello", "world"), lines); + } + + @Test + public void testCRLFLines() throws Exception { + List lines = readAllLines(stream("hello\r\nworld\r\n"), 0); + assertEquals(List.of("hello", "world"), lines); + } + + @Test + public void testBareCRLines() throws Exception { + List lines = readAllLines(stream("hello\rworld\r"), 0); + assertEquals(List.of("hello", "world"), lines); + } + + @Test + public void testMixedTerminators() throws Exception { + List lines = readAllLines(stream("LF\nCRLF\r\nCR\rend"), 0); + assertEquals(List.of("LF", "CRLF", "CR", "end"), lines); + } + + @Test + public void testStreamEndWithoutTerminator() throws Exception { + List lines = readAllLines(stream("no-newline"), 0); + assertEquals(List.of("no-newline"), lines); + } + + @Test + public void testEmptyStream() throws Exception { + List lines = readAllLines(stream(""), 0); + assertEquals(List.of(), lines); + } + + @Test + public void testEmptyLines() throws Exception { + List lines = readAllLines(stream("\n\n\n"), 0); + assertEquals(List.of("", "", ""), lines); + } + + @Test + public void testReturnsFalseAtEndOfStream() throws Exception { + InputStream is = stream("one\n"); + BoundedLineAccumulator acc = new BoundedLineAccumulator(0); + assertTrue(readBoundedLineRaw(is, acc)); + assertEquals("one", acc.toLine()); + acc.reset(); + assertFalse(readBoundedLineRaw(is, acc)); + } + + @Test + public void testLineTooLongIsFlagged() throws Exception { + InputStream is = stream("short\n" + "x".repeat(20) + "\nok\n"); + BoundedLineAccumulator acc = new BoundedLineAccumulator(10); + + assertTrue(readBoundedLineRaw(is, acc)); + assertFalse(acc.isTooLong()); + assertEquals("short", acc.toLine()); + acc.reset(); + + assertTrue(readBoundedLineRaw(is, acc)); + assertTrue(acc.isTooLong(), "Too-long line should be flagged"); + acc.reset(); + + assertTrue(readBoundedLineRaw(is, acc)); + assertFalse(acc.isTooLong()); + assertEquals("ok", acc.toLine()); + } + + @Test + public void testLineTooLongAtStreamEnd() throws Exception { + InputStream is = stream("ok\n" + "x".repeat(20)); + BoundedLineAccumulator acc = new BoundedLineAccumulator(10); + + assertTrue(readBoundedLineRaw(is, acc)); + assertFalse(acc.isTooLong()); + assertEquals("ok", acc.toLine()); + acc.reset(); + + assertTrue(readBoundedLineRaw(is, acc)); + assertTrue(acc.isTooLong(), "Too-long line at stream end should be flagged"); + } + + @Test + public void testLineExactlyAtLimit() throws Exception { + List lines = readAllLines(stream("1234567890\n"), 10); + assertEquals(1, lines.size()); + assertEquals("1234567890", lines.get(0)); + } + + @Test + public void testLineOneOverLimit() throws Exception { + InputStream is = stream("12345678901\n"); + BoundedLineAccumulator acc = new BoundedLineAccumulator(10); + assertTrue(readBoundedLineRaw(is, acc)); + assertTrue(acc.isTooLong(), "Line one byte over limit should be flagged"); + } + + @Test + public void testMaxLineBytesZeroDisablesLimit() throws Exception { + String longLine = "x".repeat(10_000); + List lines = readAllLines(stream(longLine + "\n"), 0); + assertEquals(1, lines.size()); + assertEquals(longLine, lines.get(0)); + } + + @Test + public void testLineTooLongWithCRLF() throws Exception { + InputStream is = stream("x".repeat(20) + "\r\nok\n"); + BoundedLineAccumulator acc = new BoundedLineAccumulator(10); + + assertTrue(readBoundedLineRaw(is, acc)); + assertTrue(acc.isTooLong()); + acc.reset(); + + assertTrue(readBoundedLineRaw(is, acc)); + assertFalse(acc.isTooLong()); + assertEquals("ok", acc.toLine()); + } + + @Test + public void testLineTooLongWithBareCR() throws Exception { + InputStream is = stream("x".repeat(20) + "\rok\n"); + BoundedLineAccumulator acc = new BoundedLineAccumulator(10); + + assertTrue(readBoundedLineRaw(is, acc)); + assertTrue(acc.isTooLong()); + acc.reset(); + + assertTrue(readBoundedLineRaw(is, acc)); + assertFalse(acc.isTooLong()); + assertEquals("ok", acc.toLine()); + } + + @Test + public void testUtf8MultiByteCharacters() throws Exception { + List lines = readAllLines(stream("héllo wörld café\n"), 0); + assertEquals(1, lines.size()); + assertEquals("héllo wörld café", lines.get(0)); + } + + @Test + public void testCRFollowedByEOFDoesNotHang() throws Exception { + List lines = readAllLines(stream("hello\r"), 0); + assertEquals(List.of("hello"), lines); + } +} diff --git a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClient.java b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClient.java index fbe147f87..e57d05e92 100644 --- a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClient.java +++ b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClient.java @@ -22,6 +22,7 @@ import org.a2aproject.sdk.client.http.A2AHttpHeaders; import org.a2aproject.sdk.client.http.A2AHttpResponse; import org.a2aproject.sdk.client.http.ServerSentEvent; +import org.a2aproject.sdk.client.http.SSEParserConfig; import org.a2aproject.sdk.client.http.ServerSentEventParser; import org.a2aproject.sdk.common.A2AErrorMessages; import org.a2aproject.sdk.spec.A2AClientHTTPError; @@ -48,24 +49,39 @@ class OkHttpA2AHttpClient implements A2AHttpClient { return t; }); + private final SSEParserConfig sseParserConfig; + + OkHttpA2AHttpClient() { + this(SSEParserConfig.DEFAULT); + } + + OkHttpA2AHttpClient(SSEParserConfig sseParserConfig) { + this.sseParserConfig = sseParserConfig; + } + @Override public GetBuilder createGet() { - return new OkHttpGetBuilder(); + return new OkHttpGetBuilder(sseParserConfig); } @Override public PostBuilder createPost() { - return new OkHttpPostBuilder(); + return new OkHttpPostBuilder(sseParserConfig); } @Override public DeleteBuilder createDelete() { - return new OkHttpDeleteBuilder(); + return new OkHttpDeleteBuilder(sseParserConfig); } private abstract static class OkHttpBuilder> implements Builder { protected String url = ""; protected final Map headers = new HashMap<>(); + protected final SSEParserConfig sseParserConfig; + + OkHttpBuilder(SSEParserConfig sseParserConfig) { + this.sseParserConfig = sseParserConfig; + } @Override public T url(String url) { @@ -189,7 +205,7 @@ private void parseResponseBody( new InputStreamReader(body.byteStream(), StandardCharsets.UTF_8))) { String line; if (isSse) { - ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer); + ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, sseParserConfig); while ((line = reader.readLine()) != null) { sseParser.processLine(line); } @@ -214,6 +230,10 @@ private void parseResponseBody( } private static class OkHttpGetBuilder extends OkHttpBuilder implements GetBuilder { +OkHttpGetBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public A2AHttpResponse get() throws IOException { OkHttpClient client = buildClient(false); @@ -248,6 +268,10 @@ private static class OkHttpPostBuilder extends OkHttpBuilder implem private String body = ""; private boolean followRedirects = false; +OkHttpPostBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public PostBuilder body(String body) { this.body = body; @@ -295,6 +319,10 @@ public CompletableFuture postAsyncSSE( } private static class OkHttpDeleteBuilder extends OkHttpBuilder implements DeleteBuilder { +OkHttpDeleteBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public A2AHttpResponse delete() throws IOException { OkHttpClient client = buildClient(false); diff --git a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClientSSETest.java b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClientSSETest.java index c664ca1d3..c672a518a 100644 --- a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClientSSETest.java +++ b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClientSSETest.java @@ -2,6 +2,7 @@ import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.AbstractA2AHttpClientSSETest; +import org.a2aproject.sdk.client.http.SSEParserConfig; public class OkHttpA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { @@ -9,4 +10,9 @@ public class OkHttpA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { protected A2AHttpClient createClient() { return new OkHttpA2AHttpClient(); } + + @Override + protected A2AHttpClient createClient(SSEParserConfig sseParserConfig) { + return new OkHttpA2AHttpClient(sseParserConfig); + } } diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java index 52931720c..95b45fbc6 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java @@ -85,6 +85,54 @@ public static A2AHttpClient create() { .orElseThrow(() -> new IllegalStateException("No A2AHttpClientProvider could be instantiated")); } + /** + * Creates a new A2AHttpClient instance with the given {@link SSEParserConfig} using the + * highest available priority provider that honours it. + * + *

Only providers that return {@code true} from {@link A2AHttpClientProvider#supportsSseConfig()} + * are considered. This prevents a higher-priority provider (e.g. Vert.x at priority 100 or CDI + * at priority 200) from silently ignoring the supplied configuration. If no + * SSE-config-aware provider is available the method falls back to any available provider and + * logs a warning. + * + * @param sseParserConfig the SSE parser configuration to apply + * @return a new A2AHttpClient instance + * @throws IllegalStateException if no provider found or all providers failed to instantiate + */ + public static A2AHttpClient createWithSseConfig(SSEParserConfig sseParserConfig) { + if (sseParserConfig == null) { + return create(); + } + // Prefer providers that explicitly support SSE config to avoid silent config loss. + List sseAwareProviders = PROVIDERS.stream() + .filter(p -> { + if (!p.supportsSseConfig()) { + LOGGER.warning(() -> "Provider " + p.name() + + " skipped because it does not support SSEParserConfig"); + return false; + } + return true; + }) + .toList(); + List candidates = sseAwareProviders.isEmpty() ? PROVIDERS : sseAwareProviders; + if (sseAwareProviders.isEmpty()) { + LOGGER.warning("No A2AHttpClientProvider supports SSEParserConfig; " + + "the supplied configuration may be ignored. " + + "Consider using JdkA2AHttpClient.withSseConfig() directly."); + } + return candidates.stream() + .flatMap(p -> { + try { + return Stream.of(p.createWithSseConfig(sseParserConfig)); + } catch (Exception e) { + LOGGER.log(Level.WARNING, e, () -> "Provider " + p.name() + " skipped"); + return Stream.empty(); + } + }) + .findFirst() + .orElseThrow(() -> new IllegalStateException("No A2AHttpClientProvider could be instantiated")); + } + /** * Creates a new A2AHttpClient instance using a specific provider by name. * diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientProvider.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientProvider.java index 9e8061360..f3078c51f 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientProvider.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientProvider.java @@ -23,6 +23,38 @@ public interface A2AHttpClientProvider { */ A2AHttpClient create(); + /** + * Creates a new instance of an A2AHttpClient with the given {@link SSEParserConfig}. + * + *

Providers that support SSE parser configuration should override both this method + * and {@link #supportsSseConfig()} to return {@code true}. + * The default implementation ignores {@code sseParserConfig} and delegates to {@link #create()}. + * + * @param sseParserConfig the SSE parser configuration to apply + * @return a new A2AHttpClient instance + */ + default A2AHttpClient createWithSseConfig(SSEParserConfig sseParserConfig) { + return create(); + } + + /** + * Returns {@code true} if this provider honours the {@link SSEParserConfig} passed to + * {@link #createWithSseConfig(SSEParserConfig)}. + * + *

Providers that override {@link #createWithSseConfig} to actually apply the + * configuration must also override this method and return {@code true}; the + * {@link A2AHttpClientFactory#createWithSseConfig} method uses this flag to skip + * providers that would silently ignore the supplied configuration. + * + *

The default is {@code false}, matching the default no-op implementation of + * {@link #createWithSseConfig}. + * + * @return {@code true} if this provider applies the given {@link SSEParserConfig} + */ + default boolean supportsSseConfig() { + return false; + } + /** * Returns the priority of this provider. Higher priority providers are * tried first; the first one whose {@link #create()} succeeds is used. diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/BoundedLineAccumulator.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/BoundedLineAccumulator.java new file mode 100644 index 000000000..1c58b0161 --- /dev/null +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/BoundedLineAccumulator.java @@ -0,0 +1,108 @@ +package org.a2aproject.sdk.client.http; + +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; + +/** + * Accumulates raw bytes for a single line while enforcing a configurable byte limit. + * + *

Used by both the blocking ({@code readBoundedLine}) and reactive + * ({@code BoundedLineBodySubscriber}) bounded-line readers to centralise the + * accumulation logic and limit enforcement. + * + *

When {@code maxLineBytes <= 0} the per-line byte cap is disabled. + */ +public final class BoundedLineAccumulator { + + static final int INITIAL_BUFFER_CAPACITY = 256; + + private final int maxLineBytes; + private final ByteArrayOutputStream buffer; + private boolean tooLong; + + /** + * Creates a new accumulator with the given per-line byte limit. + * + * @param maxLineBytes the maximum number of bytes per line ({@code 0} = disabled) + */ + public BoundedLineAccumulator(int maxLineBytes) { + this.maxLineBytes = maxLineBytes; + this.buffer = new ByteArrayOutputStream(INITIAL_BUFFER_CAPACITY); + } + + /** + * Adds a single byte to the accumulator. + * + *

If the limit has already been exceeded, the byte is silently discarded. + * If adding this byte would exceed the limit, the accumulator transitions to + * the "too long" state and discards all previously accumulated bytes. + * + * @param b the byte to add (only the low 8 bits are used) + */ + public void addByte(int b) { + if (tooLong) { + return; + } + if (maxLineBytes > 0 && buffer.size() >= maxLineBytes) { + tooLong = true; + buffer.reset(); + return; + } + buffer.write(b); + } + + /** + * Adds a chunk of bytes to the accumulator. + * + *

If the accumulated byte count plus the chunk length would exceed the limit, + * the accumulator transitions to the "too long" state and discards all + * previously accumulated bytes. The chunk is not written. + * + * @param chunk the byte array to add from + * @param off the start offset within the array + * @param len the number of bytes to add + */ + public void addChunk(byte[] chunk, int off, int len) { + if (tooLong) { + return; + } + if (maxLineBytes > 0 && buffer.size() + len > maxLineBytes) { + tooLong = true; + buffer.reset(); + return; + } + buffer.write(chunk, off, len); + } + + /** + * Returns {@code true} if the accumulated bytes have exceeded the configured limit. + */ + public boolean isTooLong() { + return tooLong; + } + + /** + * Returns the number of bytes currently accumulated. + */ + public int size() { + return buffer.size(); + } + + /** + * Assembles the accumulated bytes into a UTF-8 string. + * + * @return the decoded line + */ + public String toLine() { + return buffer.toString(StandardCharsets.UTF_8); + } + + /** + * Resets the accumulator for the next line, clearing all accumulated bytes + * and the "too long" flag. + */ + public void reset() { + buffer.reset(); + tooLong = false; + } +} diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java index 8168b7f3b..3842755f7 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java @@ -59,6 +59,7 @@ public class JdkA2AHttpClient implements A2AHttpClient { private final HttpClient httpClient; + private final SSEParserConfig sseParserConfig; private volatile @Nullable HttpClient noRedirectClient; /** @@ -81,7 +82,25 @@ public JdkA2AHttpClient() { this(HttpClient.newBuilder() .version(HttpClient.Version.HTTP_2) .followRedirects(HttpClient.Redirect.NEVER) - .build()); + .build(), SSEParserConfig.DEFAULT); + } + + /** + * Creates a new JDK-based HTTP client with secure defaults and custom SSE parser limits. + * + *

Named factory method avoids overload ambiguity with {@link #JdkA2AHttpClient(HttpClient)}. + * + * @param sseParserConfig the SSE parser configuration to use for streaming responses + * @return a new JdkA2AHttpClient with the given SSE parser configuration + * @throws IllegalArgumentException if {@code sseParserConfig} is {@code null} + */ + public static JdkA2AHttpClient withSseConfig(SSEParserConfig sseParserConfig) { + return new JdkA2AHttpClient( + HttpClient.newBuilder() + .version(HttpClient.Version.HTTP_2) + .followRedirects(HttpClient.Redirect.NEVER) + .build(), + sseParserConfig); } /** @@ -100,7 +119,20 @@ public JdkA2AHttpClient() { * @throws IllegalArgumentException if {@code httpClient} is {@code null} */ public JdkA2AHttpClient(HttpClient httpClient) { + this(httpClient, SSEParserConfig.DEFAULT); + } + + /** + * Creates a new JDK-based HTTP client using a caller-provided JDK {@link HttpClient} + * and custom SSE parser limits. + * + * @param httpClient the JDK HTTP client to delegate requests to + * @param sseParserConfig the SSE parser configuration to use for streaming responses + * @throws IllegalArgumentException if {@code httpClient} or {@code sseParserConfig} is {@code null} + */ + public JdkA2AHttpClient(HttpClient httpClient, SSEParserConfig sseParserConfig) { this.httpClient = checkNotNullParam("httpClient", httpClient); + this.sseParserConfig = checkNotNullParam("sseParserConfig", sseParserConfig); } @Override @@ -182,7 +214,7 @@ protected CompletableFuture asyncRequest( Consumer errorConsumer, Runnable completeRunnable ) { - ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer); + ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, sseParserConfig); AtomicBoolean useSseParser = new AtomicBoolean(false); AtomicBoolean errorNotified = new AtomicBoolean(false); StringBuilder nonSseBodyBuffer = new StringBuilder(); @@ -277,6 +309,12 @@ public void onComplete() { boolean isSse = JdkHttpResponse.success(responseInfo.statusCode()) && contentType.contains(EVENT_STREAM); useSseParser.set(isSse); + if (isSse) { + // Use a bounded byte-level subscriber so that the maxLineLength limit is + // enforced *before* a full line is materialised in memory, preventing memory + // exhaustion from a malicious server that never sends a newline. + return new BoundedLineBodySubscriber(subscriber, sseParserConfig.maxLineLength(), sseParser); + } return BodyHandlers.fromLineSubscriber(subscriber).apply(responseInfo); }; @@ -465,5 +503,154 @@ public A2AHttpHeaders headers() { return A2AHttpHeaders.of(response.headers().map()); } } + + /** + * A {@link HttpResponse.BodySubscriber} that splits the raw byte stream into lines and + * delivers each complete line to a {@link Flow.Subscriber}{@code }. + * + *

Unlike {@link java.net.http.HttpResponse.BodyHandlers#fromLineSubscriber}, this + * implementation enforces {@code maxLineBytes} while accumulating bytes, so a + * malicious unterminated line cannot exhaust heap before the limit is applied. + * + *

When {@code maxLineBytes <= 0} the per-line byte cap is disabled, matching the + * semantics of {@link SSEParserConfig#maxLineLength()} == 0. + */ + private static final class BoundedLineBodySubscriber implements HttpResponse.BodySubscriber { + + private final Flow.Subscriber lineSubscriber; + private final ServerSentEventParser sseParser; + private final CompletableFuture result = new CompletableFuture<>(); + private final BoundedLineAccumulator accumulator; + /** True when a bare CR was the last byte of the previous chunk (may start a CRLF pair). */ + private boolean pendingCR = false; + + BoundedLineBodySubscriber(Flow.Subscriber lineSubscriber, int maxLineBytes, + ServerSentEventParser sseParser) { + this.lineSubscriber = lineSubscriber; + this.sseParser = sseParser; + this.accumulator = new BoundedLineAccumulator(maxLineBytes); + } + + @Override + public CompletableFuture getBody() { + return result; + } + + @Override + public void onSubscribe(Flow.Subscription subscription) { + // Request all data from the HTTP body publisher up-front — the JDK HTTP + // layer is not back-pressure sensitive for this use-case, and mirroring what + // BodyHandlers.fromLineSubscriber does keeps behaviour consistent. + subscription.request(Long.MAX_VALUE); + // Give the line subscriber a no-op subscription: it cannot cancel the HTTP + // connection from the line level, and back-pressure is already handled above. + lineSubscriber.onSubscribe(new Flow.Subscription() { + @Override public void request(long n) { /* no-op */ } + @Override public void cancel() { /* no-op */ } + }); + } + + private record TerminatorScan(int lineLen, int resumePos, boolean crAtBoundary) {} + + @Override + public void onNext(List items) { + for (ByteBuffer buf : items) { + processBuffer(buf); + } + } + + private void processBuffer(ByteBuffer buf) { + while (buf.hasRemaining()) { + if (consumePendingCR(buf)) { + break; + } + + int start = buf.position(); + TerminatorScan scan = scanForTerminator(buf, start); + + if (scan == null) { + accumulateRemainingBytes(buf, start); + } else { + pendingCR = scan.crAtBoundary(); + buf.position(scan.resumePos()); + deliverCompleteLine(buf, start, scan.lineLen()); + } + } + } + + private boolean consumePendingCR(ByteBuffer buf) { + if (!pendingCR) { + return false; + } + pendingCR = false; + if (buf.get(buf.position()) == '\n') { + buf.position(buf.position() + 1); + return !buf.hasRemaining(); + } + return false; + } + + private static @Nullable TerminatorScan scanForTerminator(ByteBuffer buf, int start) { + int end = buf.limit(); + for (int i = start; i < end; i++) { + byte b = buf.get(i); + if (b == '\n') { + return new TerminatorScan(i - start, i + 1, false); + } + if (b == '\r') { + int lineLen = i - start; + if (i + 1 < end) { + int resumePos = buf.get(i + 1) == '\n' ? i + 2 : i + 1; + return new TerminatorScan(lineLen, resumePos, false); + } + return new TerminatorScan(lineLen, i + 1, true); + } + } + return null; + } + + private void accumulateRemainingBytes(ByteBuffer buf, int start) { + int len = buf.limit() - start; + byte[] chunk = new byte[len]; + buf.get(start, chunk); + accumulator.addChunk(chunk, 0, len); + buf.position(buf.limit()); + } + + private void deliverCompleteLine(ByteBuffer buf, int start, int lineLen) { + if (accumulator.isTooLong()) { + sseParser.processLineTooLong(); + } else { + byte[] chunk = new byte[lineLen]; + buf.get(start, chunk, 0, lineLen); + accumulator.addChunk(chunk, 0, lineLen); + if (accumulator.isTooLong()) { + sseParser.processLineTooLong(); + } else { + lineSubscriber.onNext(accumulator.toLine()); + } + } + accumulator.reset(); + } + + @Override + public void onError(Throwable throwable) { + lineSubscriber.onError(throwable); + result.completeExceptionally(throwable); + } + + @Override + public void onComplete() { + // Flush any pending state as a final line (stream ended without a trailing LF) + if (accumulator.isTooLong()) { + sseParser.processLineTooLong(); + } else if (accumulator.size() > 0) { + lineSubscriber.onNext(accumulator.toLine()); + } + accumulator.reset(); + lineSubscriber.onComplete(); + result.complete(null); + } + } } diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientProvider.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientProvider.java index 4b53d41f6..31fe0b520 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientProvider.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientProvider.java @@ -15,6 +15,19 @@ public A2AHttpClient create() { return new JdkA2AHttpClient(); } + /** + * {@inheritDoc} + */ + @Override + public A2AHttpClient createWithSseConfig(SSEParserConfig sseParserConfig) { + return JdkA2AHttpClient.withSseConfig(sseParserConfig); + } + + @Override + public boolean supportsSseConfig() { + return true; + } + @Override public int priority() { return 0; // Lowest priority - fallback diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/SSEParserConfig.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/SSEParserConfig.java new file mode 100644 index 000000000..28538a7d7 --- /dev/null +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/SSEParserConfig.java @@ -0,0 +1,96 @@ +package org.a2aproject.sdk.client.http; + +/** + * Configuration for {@link ServerSentEventParser} limits. + * + *

All limits have safe defaults suitable for most deployments. Use the {@link Builder} + * to override individual values -- for example, to raise the per-event character budget + * for agents that return large artifacts. + * + *

Enforcement semantics: {@code maxLineLength} is enforced at the + * transport layer as a byte limit on raw UTF-8 data, before lines are decoded + * to characters. For ASCII-only content (the common case for SSE/JSON-RPC), bytes and + * characters are equivalent. For multi-byte UTF-8 content, the byte-level check is + * stricter — a line of {@code maxLineLength} multi-byte characters may exceed the limit. + * This is intentional: the byte-level enforcement prevents memory exhaustion from a + * malicious unterminated line before it is fully materialised. + * + *

Security note: setting {@code maxLineLength} to {@code 0} disables + * the per-line memory protection entirely. Without this limit, a malicious server sending + * a single unterminated line (no newline) can consume unbounded memory — {@code maxBufferChars} + * only limits accumulated {@code data:} field values after lines have been decoded, + * not the raw line itself. Only disable the per-line check when you trust the remote agent + * or have other safeguards (e.g. a reverse proxy that enforces its own line-length limit). + * + * @param maxLineLength max bytes per raw SSE line (0 = disabled) + * @param maxBufferLines max {@code data:} lines per event block + * @param maxBufferChars max total characters across all {@code data:} values per event + */ +public record SSEParserConfig(int maxLineLength, int maxBufferLines, int maxBufferChars) { + + /** + * Default configuration: 1 MB per-line and per-event character limit, 1 000 data lines per event. + */ + public static final SSEParserConfig DEFAULT = new SSEParserConfig(1024 * 1024, 1000, 1024 * 1024); + + public SSEParserConfig { + if (maxLineLength < 0) { + throw new IllegalArgumentException("maxLineLength must be >= 0, got " + maxLineLength); + } + if (maxBufferLines <= 0) { + throw new IllegalArgumentException("maxBufferLines must be > 0, got " + maxBufferLines); + } + if (maxBufferChars <= 0) { + throw new IllegalArgumentException("maxBufferChars must be > 0, got " + maxBufferChars); + } + } + + /** + * Returns a new {@link Builder} initialized with the {@link #DEFAULT} values. + */ + public static Builder builder() { + return new Builder(); + } + + public static final class Builder { + private int maxLineLength = DEFAULT.maxLineLength; + private int maxBufferLines = DEFAULT.maxBufferLines; + private int maxBufferChars = DEFAULT.maxBufferChars; + + Builder() { + } + + /** + * Sets the maximum number of bytes allowed in a single raw SSE line. + * Set to {@code 0} to disable the per-line check. + * + *

Security note: disabling this check removes the first line of + * defence against memory exhaustion from oversized SSE lines. Ensure + * {@code maxBufferChars} is set to an acceptable upper bound when disabling. + */ + public Builder maxLineLength(int maxLineLength) { + this.maxLineLength = maxLineLength; + return this; + } + + /** + * Sets the maximum number of {@code data:} lines allowed in a single event block. + */ + public Builder maxBufferLines(int maxBufferLines) { + this.maxBufferLines = maxBufferLines; + return this; + } + + /** + * Sets the maximum total characters across all {@code data:} values in a single event. + */ + public Builder maxBufferChars(int maxBufferChars) { + this.maxBufferChars = maxBufferChars; + return this; + } + + public SSEParserConfig build() { + return new SSEParserConfig(maxLineLength, maxBufferLines, maxBufferChars); + } + } +} diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/ServerSentEventParser.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/ServerSentEventParser.java index ea08fc97f..bb12a23fc 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/ServerSentEventParser.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/ServerSentEventParser.java @@ -15,9 +15,9 @@ public class ServerSentEventParser { private static final Logger LOGGER = Logger.getLogger(ServerSentEventParser.class.getName()); - private static final int MAX_BUFFER_SIZE = 1000; - private static final int MAX_BUFFER_CHARS = 1024 * 1024; // 1 MB (Java chars, so up to 2 MB in UTF-16; actual UTF-8 bytes may differ) - private static final int MAX_LINE_LENGTH = 65536; // 64 KB + private final int maxLineLength; + private final int maxBufferLines; + private final int maxBufferChars; private final Consumer eventConsumer; private final @Nullable Consumer errorConsumer; @@ -35,12 +35,33 @@ public class ServerSentEventParser { private boolean skippingCurrentEvent = false; public ServerSentEventParser(Consumer eventConsumer) { - this(eventConsumer, null); + this(eventConsumer, null, SSEParserConfig.DEFAULT); } public ServerSentEventParser(Consumer eventConsumer, @Nullable Consumer errorConsumer) { + this(eventConsumer, errorConsumer, SSEParserConfig.DEFAULT); + } + + public ServerSentEventParser(Consumer eventConsumer, @Nullable Consumer errorConsumer, + SSEParserConfig config) { this.eventConsumer = eventConsumer; this.errorConsumer = errorConsumer; + this.maxLineLength = config.maxLineLength(); + this.maxBufferLines = config.maxBufferLines(); + this.maxBufferChars = config.maxBufferChars(); + } + + /** + * Signals that the transport layer detected a line exceeding the configured limit + * before materialising the full line in memory. This marks the current event + * block as corrupt (same as if {@link #processLine} had received an oversized string) + * and reports the error, without requiring the caller to allocate a placeholder string. + */ + public void processLineTooLong() { + handleError(new IllegalArgumentException("Line exceeds maximum length of " + maxLineLength + " characters")); + skippingCurrentEvent = true; + dataBuffer.clear(); + dataBufferChars = 0; } /** @@ -54,8 +75,8 @@ public void processLine(@Nullable String line) { } // Check line length to prevent DoS; corrupt the current event so it is not dispatched - if (line.length() > MAX_LINE_LENGTH) { - handleError(new IllegalArgumentException("Line exceeds maximum length of " + MAX_LINE_LENGTH + " characters")); + if (maxLineLength > 0 && line.length() > maxLineLength) { + handleError(new IllegalArgumentException("Line exceeds maximum length of " + maxLineLength + " characters")); skippingCurrentEvent = true; dataBuffer.clear(); dataBufferChars = 0; @@ -99,16 +120,16 @@ private void processField(String field, String value) { switch (field) { case "data" -> { // Check line count to prevent DoS; corrupt and skip the rest of this event block - if (dataBuffer.size() >= MAX_BUFFER_SIZE) { - handleError(new IllegalStateException("SSE data buffer exceeded maximum size of " + MAX_BUFFER_SIZE + " lines")); + if (dataBuffer.size() >= maxBufferLines) { + handleError(new IllegalStateException("SSE data buffer exceeded maximum size of " + maxBufferLines + " lines")); skippingCurrentEvent = true; dataBuffer.clear(); dataBufferChars = 0; return; } // Check total char count to prevent OOM on large streams - if (dataBufferChars + value.length() > MAX_BUFFER_CHARS) { - handleError(new IllegalStateException("SSE data buffer exceeded maximum size of " + MAX_BUFFER_CHARS + " chars")); + if (dataBufferChars + value.length() > maxBufferChars) { + handleError(new IllegalStateException("SSE data buffer exceeded maximum size of " + maxBufferChars + " chars")); skippingCurrentEvent = true; dataBuffer.clear(); dataBufferChars = 0; @@ -146,9 +167,14 @@ private void processField(String field, String value) { } private void dispatchEvent() { - // Per SSE spec: update lastEventId before checking data, so ID-only events (e.g. heartbeats) are tracked - if (currentEventId != null) { + // Per SSE spec §9.2.6: copy currentEventId → lastEventId at dispatch, but only for blocks that + // were not skipped. A corrupt/oversized block must not advance the reconnect cursor even if its + // id: field was parsed before the violation was detected. + if (!skippingCurrentEvent && currentEventId != null) { lastEventId = currentEventId; + } else if (skippingCurrentEvent) { + // Roll back the event ID buffer so the skipped block's id: cannot leak into subsequent events. + currentEventId = lastEventId; } String data = String.join("\n", dataBuffer); diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java index 814b01c33..fdb34bd66 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java @@ -1,6 +1,9 @@ package org.a2aproject.sdk.client.http; -import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import org.junit.jupiter.api.Test; @@ -61,6 +64,35 @@ public void testCreateWithNullProviderNameThrows() { ); } + @Test + public void testCreateWithSseConfigNullDelegatesToCreate() { + A2AHttpClient client = A2AHttpClientFactory.createWithSseConfig(null); + assertNotNull(client); + assertInstanceOf(JdkA2AHttpClient.class, client); + } + + @Test + public void testCreateWithSseConfigReturnsJdkClient() { + SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(4 * 1024 * 1024).build(); + A2AHttpClient client = A2AHttpClientFactory.createWithSseConfig(config); + assertNotNull(client); + assertInstanceOf(JdkA2AHttpClient.class, client, + "Factory should return JdkA2AHttpClient with custom SSEParserConfig"); + } + + @Test + public void testCreateWithSseConfigAppliesConfig() throws Exception { + // Verify the config is actually applied by creating a client with a restrictive + // maxLineLength and checking that it rejects oversized SSE lines. + SSEParserConfig restrictive = SSEParserConfig.builder().maxLineLength(50).build(); + A2AHttpClient client = A2AHttpClientFactory.createWithSseConfig(restrictive); + assertNotNull(client); + // If the factory silently ignored the config, it would use the 1 MB default. + // We can't easily verify SSE behaviour without a server, but we at least confirm + // the client was created successfully with a non-default config. + assertInstanceOf(JdkA2AHttpClient.class, client); + } + @Test public void testCreateWithEmptyProviderNameThrows() { assertThrows( diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientProviderTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientProviderTest.java index 525b0b505..0ca3adc6e 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientProviderTest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientProviderTest.java @@ -14,6 +14,15 @@ public void testJdkProviderCreatesClient() { assertInstanceOf(JdkA2AHttpClient.class, client); } + @Test + public void testJdkProviderCreatesClientWithSseParserConfig() { + JdkA2AHttpClientProvider provider = new JdkA2AHttpClientProvider(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(4 * 1024 * 1024).build(); + A2AHttpClient client = provider.createWithSseConfig(config); + assertNotNull(client); + assertInstanceOf(JdkA2AHttpClient.class, client); + } + @Test public void testJdkProviderPriority() { JdkA2AHttpClientProvider provider = new JdkA2AHttpClientProvider(); @@ -25,4 +34,27 @@ public void testJdkProviderName() { JdkA2AHttpClientProvider provider = new JdkA2AHttpClientProvider(); assertEquals("jdk", provider.name(), "JDK provider name should be 'jdk'"); } + + @Test + public void testJdkProviderSupportsSseConfig() { + JdkA2AHttpClientProvider provider = new JdkA2AHttpClientProvider(); + assertTrue(provider.supportsSseConfig(), "JDK provider must support SSE config"); + } + + @Test + public void testDefaultProviderDoesNotSupportSseConfig() { + A2AHttpClientProvider defaultProvider = new A2AHttpClientProvider() { + @Override + public A2AHttpClient create() { + return new JdkA2AHttpClient(); + } + + @Override + public String name() { + return "test"; + } + }; + assertFalse(defaultProvider.supportsSseConfig(), + "Default interface implementation must return false"); + } } diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/AbstractA2AHttpClientSSETest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/AbstractA2AHttpClientSSETest.java index 63644fd1b..8b57a967b 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/AbstractA2AHttpClientSSETest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/AbstractA2AHttpClientSSETest.java @@ -5,10 +5,12 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; import static org.mockserver.model.HttpRequest.request; import static org.mockserver.model.HttpResponse.response; import org.a2aproject.sdk.common.A2AErrorMessages; +import org.jspecify.annotations.Nullable; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -29,6 +31,15 @@ public abstract class AbstractA2AHttpClientSSETest { protected abstract A2AHttpClient createClient(); + /** + * Creates a client with a custom {@link SSEParserConfig}. + * Returns {@code null} if the implementation does not support SSEParserConfig, + * in which case the SSE config integration tests are skipped. + */ + protected @Nullable A2AHttpClient createClient(SSEParserConfig sseParserConfig) { + return null; + } + @BeforeEach public void setup() { mockServer = ClientAndServer.startClientAndServer(0); @@ -377,4 +388,178 @@ public void testPostSSETypedEvents() throws Exception { assertEquals("99", events.get(0).id()); assertEquals("done", events.get(0).data()); } + + @Test + public void testCustomSSEParserConfigRejectsOversizedLine() throws Exception { + A2AHttpClient configClient = createClient(SSEParserConfig.builder().maxLineLength(50).build()); + assumeTrue(configClient != null, "Implementation does not support SSEParserConfig"); + + // 60-char payload exceeds the 50-char per-line limit + String oversizedPayload = "x".repeat(60); + mockServer + .when(request().withMethod("POST").withPath("/sse")) + .respond(response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody("data: " + oversizedPayload + "\n\n")); + + CountDownLatch latch = new CountDownLatch(1); + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + + configClient.createPost() + .url(getBaseUrl() + "/sse") + .body("{}") + .postAsyncSSE( + events::add, + e -> { + error.set(e); + latch.countDown(); + }, + latch::countDown + ); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertNotNull(error.get(), "Custom maxLineLength should reject oversized SSE line"); + assertEquals(0, events.size(), "Oversized event must not be dispatched"); + } + + @Test + public void testCustomSSEParserConfigAcceptsLineWithinLimit() throws Exception { + A2AHttpClient configClient = createClient(SSEParserConfig.builder().maxLineLength(200).build()); + assumeTrue(configClient != null, "Implementation does not support SSEParserConfig"); + + // 100-char payload is within the 200-char per-line limit + String payload = "x".repeat(100); + mockServer + .when(request().withMethod("POST").withPath("/sse")) + .respond(response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody("data: " + payload + "\n\n")); + + CountDownLatch latch = new CountDownLatch(1); + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + + configClient.createPost() + .url(getBaseUrl() + "/sse") + .body("{}") + .postAsyncSSE( + events::add, + error::set, + latch::countDown + ); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertNull(error.get(), "Payload within limit should not trigger an error"); + assertEquals(1, events.size()); + assertEquals(payload, events.get(0).data()); + } + + @Test + public void testSSEWithCRLFLineEndings() throws Exception { + mockServer + .when(request().withMethod("GET").withPath("/sse")) + .respond(response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody("data: first\r\n\r\ndata: second\r\n\r\n")); + + CountDownLatch latch = new CountDownLatch(1); + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + + client.createGet() + .url(getBaseUrl() + "/sse") + .getAsyncSSE(events::add, error::set, latch::countDown); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertNull(error.get(), "CRLF line endings should be handled correctly"); + assertEquals(2, events.size()); + assertEquals("first", events.get(0).data()); + assertEquals("second", events.get(1).data()); + } + + @Test + public void testSSEWithBareCRLineEndings() throws Exception { + mockServer + .when(request().withMethod("GET").withPath("/sse")) + .respond(response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody("data: first\r\rdata: second\r\r")); + + CountDownLatch latch = new CountDownLatch(1); + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + + client.createGet() + .url(getBaseUrl() + "/sse") + .getAsyncSSE(events::add, error::set, latch::countDown); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertNull(error.get(), "Bare CR line endings should be handled correctly"); + assertEquals(2, events.size()); + assertEquals("first", events.get(0).data()); + assertEquals("second", events.get(1).data()); + } + + @Test + public void testSSEWithMixedLineEndings() throws Exception { + // Mix of LF, CRLF, and bare CR terminators + mockServer + .when(request().withMethod("GET").withPath("/sse")) + .respond(response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody("data: lf\n\ndata: crlf\r\n\r\ndata: cr\r\r")); + + CountDownLatch latch = new CountDownLatch(1); + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + + client.createGet() + .url(getBaseUrl() + "/sse") + .getAsyncSSE(events::add, error::set, latch::countDown); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertNull(error.get(), "Mixed line endings should be handled correctly"); + assertEquals(3, events.size()); + assertEquals("lf", events.get(0).data()); + assertEquals("crlf", events.get(1).data()); + assertEquals("cr", events.get(2).data()); + } + + @Test + public void testSSEOversizedLineRejectedBeforeMemoryExhaustion() throws Exception { + A2AHttpClient configClient = createClient(SSEParserConfig.builder().maxLineLength(50).build()); + assumeTrue(configClient != null, "Implementation does not support SSEParserConfig"); + + // Valid event followed by an oversized line in the same event block. + // The oversized event must be dropped, but the parser should recover. + String oversized = "x".repeat(60); + String body = "data: good\n\ndata: " + oversized + "\n\ndata: recovered\n\n"; + + mockServer + .when(request().withMethod("GET").withPath("/sse")) + .respond(response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody(body)); + + CountDownLatch latch = new CountDownLatch(1); + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + + configClient.createGet() + .url(getBaseUrl() + "/sse") + .getAsyncSSE(events::add, errors::add, latch::countDown); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertTrue(errors.size() >= 1, "Error should be reported for oversized line"); + assertEquals(2, events.size(), "Good and recovered events should be dispatched"); + assertEquals("good", events.get(0).data()); + assertEquals("recovered", events.get(1).data()); + } } diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/BoundedLineBodySubscriberTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/BoundedLineBodySubscriberTest.java new file mode 100644 index 000000000..336279d92 --- /dev/null +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/BoundedLineBodySubscriberTest.java @@ -0,0 +1,349 @@ +package org.a2aproject.sdk.client.http; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.lang.reflect.Constructor; +import java.net.http.HttpResponse; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Flow; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for the bounded line-splitting body subscriber inside {@link JdkA2AHttpClient}. + * Since the subscriber is a private inner class, we instantiate it via reflection to test + * its byte-level line splitting and length enforcement in isolation. + */ +public class BoundedLineBodySubscriberTest { + + @SuppressWarnings("unchecked") + private static HttpResponse.BodySubscriber createSubscriber( + Flow.Subscriber lineSubscriber, int maxLineBytes, + ServerSentEventParser sseParser) throws Exception { + Class clazz = Class.forName( + "org.a2aproject.sdk.client.http.JdkA2AHttpClient$BoundedLineBodySubscriber"); + Constructor ctor = clazz.getDeclaredConstructor( + Flow.Subscriber.class, int.class, ServerSentEventParser.class); + ctor.setAccessible(true); + return (HttpResponse.BodySubscriber) ctor.newInstance(lineSubscriber, maxLineBytes, sseParser); + } + + private static List toBuffers(String data) { + return List.of(ByteBuffer.wrap(data.getBytes(StandardCharsets.UTF_8))); + } + + private static void subscribe(HttpResponse.BodySubscriber subscriber) { + subscriber.onSubscribe(new Flow.Subscription() { + @Override + public void request(long n) { /* no-op in tests */ } + + @Override + public void cancel() { /* no-op in tests */ } + }); + } + + private static final class RecordingSubscriber implements Flow.Subscriber { + final List lines = new ArrayList<>(); + final AtomicBoolean completed = new AtomicBoolean(false); + final AtomicReference error = new AtomicReference<>(); + + @Override + public void onSubscribe(Flow.Subscription subscription) { + subscription.request(Long.MAX_VALUE); + } + + @Override + public void onNext(String item) { + lines.add(item); + } + + @Override + public void onError(Throwable throwable) { + error.set(throwable); + } + + @Override + public void onComplete() { + completed.set(true); + } + } + + private static ServerSentEventParser createParser(List errors) { + return new ServerSentEventParser( + event -> { /* discard events */ }, + errors::add, + SSEParserConfig.DEFAULT); + } + + @Test + public void testBasicLFLineSplitting() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("hello\nworld\n")); + subscriber.onComplete(); + + assertEquals(List.of("hello", "world"), rec.lines); + assertTrue(rec.completed.get()); + assertTrue(errors.isEmpty()); + } + + @Test + public void testCRLFLineSplitting() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("hello\r\nworld\r\n")); + subscriber.onComplete(); + + assertEquals(List.of("hello", "world"), rec.lines); + assertTrue(errors.isEmpty()); + } + + @Test + public void testBareCRLineSplitting() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("hello\rworld\r")); + subscriber.onComplete(); + + assertEquals(List.of("hello", "world"), rec.lines); + assertTrue(errors.isEmpty()); + } + + @Test + public void testMixedLineTerminators() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("LF\nCRLF\r\nCR\rend")); + subscriber.onComplete(); + + assertEquals(List.of("LF", "CRLF", "CR", "end"), rec.lines); + assertTrue(errors.isEmpty()); + } + + @Test + public void testCRLFSplitAcrossChunks() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("hello\r")); + subscriber.onNext(toBuffers("\nworld\n")); + subscriber.onComplete(); + + assertEquals(List.of("hello", "world"), rec.lines); + assertTrue(errors.isEmpty()); + } + + @Test + public void testBareCRAtChunkBoundaryFollowedByNonLF() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("hello\r")); + subscriber.onNext(toBuffers("world\n")); + subscriber.onComplete(); + + assertEquals(List.of("hello", "world"), rec.lines); + assertTrue(errors.isEmpty()); + } + + @Test + public void testLineSplitAcrossMultipleChunks() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("hel")); + subscriber.onNext(toBuffers("lo wo")); + subscriber.onNext(toBuffers("rld\n")); + subscriber.onComplete(); + + assertEquals(List.of("hello world"), rec.lines); + assertTrue(errors.isEmpty()); + } + + @Test + public void testStreamEndWithoutTrailingTerminator() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("no-newline")); + subscriber.onComplete(); + + assertEquals(List.of("no-newline"), rec.lines); + assertTrue(errors.isEmpty()); + } + + @Test + public void testEmptyLines() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("\n\n\n")); + subscriber.onComplete(); + + assertEquals(List.of("", "", ""), rec.lines); + assertTrue(errors.isEmpty()); + } + + @Test + public void testLineTooLongCallsParser() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 10, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("short\n" + "x".repeat(20) + "\nok\n")); + subscriber.onComplete(); + + assertEquals(List.of("short", "ok"), rec.lines, + "Too-long line should not appear in line subscriber output"); + assertEquals(1, errors.size(), "One error should be reported for too-long line"); + assertTrue(errors.get(0).getMessage().contains("maximum length")); + } + + @Test + public void testLineTooLongAcrossChunks() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 10, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("12345")); + subscriber.onNext(toBuffers("67890AB\n")); + subscriber.onComplete(); + + assertTrue(rec.lines.isEmpty(), "Too-long line should not appear"); + assertEquals(1, errors.size()); + } + + @Test + public void testLineTooLongAtStreamEnd() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 10, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("data: ok\n" + "x".repeat(20))); + subscriber.onComplete(); + + assertEquals(List.of("data: ok"), rec.lines); + assertEquals(1, errors.size(), + "Too-long line at stream end must trigger processLineTooLong"); + } + + @Test + public void testLineExactlyAtLimit() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 10, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("1234567890\n")); + subscriber.onComplete(); + + assertEquals(1, rec.lines.size()); + assertEquals("1234567890", rec.lines.get(0), "Line exactly at limit should be accepted"); + assertTrue(errors.isEmpty()); + } + + @Test + public void testLineOneOverLimit() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 10, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("12345678901\n")); + subscriber.onComplete(); + + assertTrue(rec.lines.isEmpty(), "Line over limit should not appear"); + assertEquals(1, errors.size()); + } + + @Test + public void testMaxLineBytesZeroDisablesLimit() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + String longLine = "x".repeat(10_000); + subscriber.onNext(toBuffers(longLine + "\n")); + subscriber.onComplete(); + + assertEquals(1, rec.lines.size()); + assertEquals(longLine, rec.lines.get(0)); + assertTrue(errors.isEmpty()); + } + + @Test + public void testUtf8MultiByteCharacters() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + subscriber.onNext(toBuffers("héllo wörld café\n")); + subscriber.onComplete(); + + assertEquals(1, rec.lines.size()); + assertEquals("héllo wörld café", rec.lines.get(0)); + assertTrue(errors.isEmpty()); + } + + @Test + public void testMultipleBuffersInSingleOnNext() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + List buffers = List.of( + ByteBuffer.wrap("first\n".getBytes(StandardCharsets.UTF_8)), + ByteBuffer.wrap("second\n".getBytes(StandardCharsets.UTF_8))); + subscriber.onNext(buffers); + subscriber.onComplete(); + + assertEquals(List.of("first", "second"), rec.lines); + assertTrue(errors.isEmpty()); + } + + @Test + public void testErrorPropagation() throws Exception { + RecordingSubscriber rec = new RecordingSubscriber(); + List errors = new ArrayList<>(); + var subscriber = createSubscriber(rec, 0, createParser(errors)); + subscribe(subscriber); + + RuntimeException expected = new RuntimeException("test error"); + subscriber.onError(expected); + + assertEquals(expected, rec.error.get()); + } +} diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientSSETest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientSSETest.java new file mode 100644 index 000000000..51b4a4efe --- /dev/null +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientSSETest.java @@ -0,0 +1,14 @@ +package org.a2aproject.sdk.client.http; + +public class JdkA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { + + @Override + protected A2AHttpClient createClient() { + return new JdkA2AHttpClient(); + } + + @Override + protected A2AHttpClient createClient(SSEParserConfig sseParserConfig) { + return JdkA2AHttpClient.withSseConfig(sseParserConfig); + } +} diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientTest.java index 0a8780119..a517a1ba1 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientTest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientTest.java @@ -94,7 +94,7 @@ public void testConstructorUsesProvidedHttpClient() throws Exception { @Test public void testConstructorRejectsNullHttpClient() { - assertThrows(IllegalArgumentException.class, () -> new JdkA2AHttpClient(null), "foo"); + assertThrows(IllegalArgumentException.class, () -> new JdkA2AHttpClient((java.net.http.HttpClient) null), "foo"); } @Test diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/ServerSentEventParserTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/ServerSentEventParserTest.java index 0492d1152..5774e4b77 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/ServerSentEventParserTest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/ServerSentEventParserTest.java @@ -2,8 +2,10 @@ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import java.util.ArrayList; import java.util.List; @@ -370,11 +372,12 @@ public void testErrorConsumerCalledForNullLine() { public void testErrorConsumerCalledForLineTooLong() { List events = new ArrayList<>(); AtomicReference error = new AtomicReference<>(); - ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set); + SSEParserConfig config = SSEParserConfig.builder().maxLineLength(1000).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); // Oversized line mid-event: the whole event block is discarded parser.processLine("data: before overflow"); - String longLine = "data: " + "x".repeat(65537); + String longLine = "data: " + "x".repeat(1001); parser.processLine(longLine); // Subsequent lines in the same block are skipped parser.processLine("data: should be skipped"); @@ -425,8 +428,7 @@ public void testErrorConsumerCalledForBufferByteOverflow() { AtomicReference error = new AtomicReference<>(); ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set); - // Value is 65530 chars so the full line ("data: " + value = 65536) stays within the per-line - // limit; 17 such lines (17 * 65530 = 1,114,010 bytes) exceed the 1MB buffer byte limit. + // Each value is 65530 chars; 17 such lines (17 * 65530 = 1,114,010 chars) exceed the 1 MB buffer char limit. String bigValue = "x".repeat(65530); for (int i = 0; i < 17; i++) { parser.processLine("data: " + bigValue); @@ -510,4 +512,327 @@ public void testCRLFLineTerminatorsPreservedInValue() { assertEquals(1, events.size()); assertEquals("value\r", events.get(0).data()); } + + @Test + public void testLargeJsonRpcResponseRejectedByOldLimit() { + // Reproducer: a 70 KB payload exceeds the old 64 KB per-line limit but fits within the + // new 1 MB default, proving the limit raise fixes real-world large JSON-RPC responses. + String hugeJson = "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":\"" + "x".repeat(70_000) + "\"}"; + + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig oldLimit = SSEParserConfig.builder().maxLineLength(65536).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, oldLimit); + + parser.processLine("data: " + hugeJson); + parser.processLine(""); + + assertEquals(0, events.size(), "Event must be rejected under the old 64 KB limit"); + assertEquals(1, errors.size()); + assertInstanceOf(IllegalArgumentException.class, errors.get(0)); + + // Same payload split across two data: lines parses fine under the old limit + events.clear(); + errors.clear(); + String half1 = hugeJson.substring(0, hugeJson.length() / 2); + String half2 = hugeJson.substring(hugeJson.length() / 2); + parser.processLine("data: " + half1); + parser.processLine("data: " + half2); + parser.processLine(""); + + assertEquals(0, errors.size(), "Split payload should not trigger any error"); + assertEquals(1, events.size()); + assertEquals(half1 + "\n" + half2, events.get(0).data()); + } + + @Test + public void testLargeJsonRpcResponseAcceptedByDefault() { + // With the raised 1 MB default the same payload parses on a single line + String hugeJson = "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":\"" + "x".repeat(70_000) + "\"}"; + + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add); + + parser.processLine("data: " + hugeJson); + parser.processLine(""); + + assertEquals(0, errors.size(), "70 KB line should be accepted with default 1 MB limit"); + assertEquals(1, events.size()); + assertEquals(hugeJson, events.get(0).data()); + } + + @Test + public void testLargeSingleLineEventAcceptedByDefault() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set); + + // 200 KB single data: line -- previously rejected at 64 KB, now accepted with 1 MB default + String largeJson = "{\"result\":\"" + "x".repeat(200_000) + "\"}"; + parser.processLine("data: " + largeJson); + parser.processLine(""); + + assertNull(error.get(), "200 KB line should be accepted with default 1 MB limit"); + assertEquals(1, events.size()); + assertEquals(largeJson, events.get(0).data()); + } + + @Test + public void testDisabledLineLengthCheck() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + SSEParserConfig config = SSEParserConfig.builder() + .maxLineLength(0) + .maxBufferChars(2 * 1024 * 1024) + .build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); + + // With maxLineLength=0 (disabled), even very large lines are accepted + String hugeLine = "data: " + "x".repeat(1_500_000); + parser.processLine(hugeLine); + parser.processLine(""); + + assertNull(error.get(), "Line length check should be disabled when maxLineLength=0"); + assertEquals(1, events.size()); + assertEquals("x".repeat(1_500_000), events.get(0).data()); + } + + @Test + public void testCustomBufferLineLimit() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + SSEParserConfig config = SSEParserConfig.builder() + .maxBufferLines(5) + .build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); + + for (int i = 0; i < 5; i++) { + parser.processLine("data: line" + i); + } + assertNull(error.get(), "No error expected at exactly the limit"); + + parser.processLine("data: overflow"); + assertNotNull(error.get(), "errorConsumer should be called when custom buffer line limit exceeded"); + parser.processLine(""); + assertEquals(0, events.size(), "Corrupted event block must not be dispatched"); + } + + @Test + public void testCustomBufferCharLimit() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + SSEParserConfig config = SSEParserConfig.builder() + .maxBufferChars(100) + .build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); + + parser.processLine("data: " + "x".repeat(101)); + assertNotNull(error.get(), "errorConsumer should be called when custom buffer char limit exceeded"); + parser.processLine(""); + assertEquals(0, events.size(), "Corrupted event block must not be dispatched"); + } + + @Test + public void testParserRecoveryAfterCustomLimitViolation() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + SSEParserConfig config = SSEParserConfig.builder() + .maxBufferLines(2) + .build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); + + parser.processLine("data: line0"); + parser.processLine("data: line1"); + parser.processLine("data: overflow"); + assertNotNull(error.get(), "errorConsumer should be called when custom buffer line limit exceeded"); + parser.processLine(""); + assertEquals(0, events.size(), "Corrupted event block must not be dispatched"); + + error.set(null); + parser.processLine("data: ok"); + parser.processLine(""); + assertNull(error.get(), "No error expected after recovery"); + assertEquals(1, events.size(), "Parser should recover after custom limit violation"); + } + + @Test + public void testSSEParserConfigDefaults() { + SSEParserConfig config = SSEParserConfig.DEFAULT; + assertEquals(1024 * 1024, config.maxLineLength()); + assertEquals(1000, config.maxBufferLines()); + assertEquals(1024 * 1024, config.maxBufferChars()); + } + + @Test + public void testSSEParserConfigBuilder() { + SSEParserConfig config = SSEParserConfig.builder() + .maxLineLength(500_000) + .maxBufferLines(2000) + .maxBufferChars(4 * 1024 * 1024) + .build(); + assertEquals(500_000, config.maxLineLength()); + assertEquals(2000, config.maxBufferLines()); + assertEquals(4 * 1024 * 1024, config.maxBufferChars()); + } + + @Test + public void testSSEParserConfigValidation() { + assertDoesNotThrow(() -> SSEParserConfig.builder().maxLineLength(0).build(), + "maxLineLength=0 (disabled) should be allowed"); + + assertThrows(IllegalArgumentException.class, + () -> SSEParserConfig.builder().maxLineLength(-1).build()); + + assertThrows(IllegalArgumentException.class, + () -> SSEParserConfig.builder().maxBufferLines(0).build()); + + assertThrows(IllegalArgumentException.class, + () -> SSEParserConfig.builder().maxBufferChars(0).build()); + } + + // --- lastEventId / skipping interaction --- + + @Test + public void testLastEventIdNotAdvancedWhenBlockSkippedByLineTooLong() { + // Regression: id: in a corrupt block must not update the reconnect cursor. + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + // Limit long enough for "id: good-id" (11) and "data: ok" (8), but shorter than the oversized data line. + SSEParserConfig config = SSEParserConfig.builder().maxLineLength(50).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + // Good event that sets lastEventId to "good-id" + parser.processLine("id: good-id"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals(1, events.size(), "Good event should be dispatched"); + assertEquals("good-id", parser.getLastEventId(), "lastEventId should be set by good event"); + + // Corrupt block: id: comes before the oversized line + parser.processLine("id: bad-id"); + parser.processLine("data: " + "x".repeat(51)); // triggers skip + parser.processLine(""); // end of corrupt block + assertEquals(1, events.size(), "Corrupt block must not be dispatched"); + assertEquals("good-id", parser.getLastEventId(), "lastEventId must not advance for a skipped block"); + } + + @Test + public void testLastEventIdNotAdvancedWhenBlockSkippedByBufferLineOverflow() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferLines(2).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + parser.processLine("id: good-id"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good-id", parser.getLastEventId(), "lastEventId should be set by good event"); + + parser.processLine("id: bad-id"); + parser.processLine("data: line0"); + parser.processLine("data: line1"); + parser.processLine("data: overflow"); // triggers skip + parser.processLine(""); + assertEquals(1, events.size(), "Only the first good event should be dispatched"); + assertEquals("good-id", parser.getLastEventId(), "lastEventId must not advance for a skipped block"); + } + + @Test + public void testSkippedBlockIdDoesNotPoisonNextEventByLineTooLong() { + // Regression: after a skipped block, currentEventId must be rolled back so a subsequent + // event without an id: field does not inherit the skipped block's id. + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxLineLength(50).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + // Good event + parser.processLine("id: good"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good", parser.getLastEventId(), "lastEventId should be set by good event"); + + // Corrupt block with a different id + parser.processLine("id: bad"); + parser.processLine("data: " + "x".repeat(51)); + parser.processLine(""); + + // Next valid event has no id: field + parser.processLine("data: next-valid-event"); + parser.processLine(""); + + assertEquals(2, events.size()); + assertEquals("good", events.get(1).id(), "Next event must carry the pre-skip id, not the skipped block's id"); + assertEquals("good", parser.getLastEventId(), "lastEventId must not be poisoned by the skipped block"); + } + + @Test + public void testSkippedBlockIdDoesNotPoisonNextEventByBufferOverflow() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferLines(2).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + parser.processLine("id: good"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good", parser.getLastEventId(), "lastEventId should be set by good event"); + + parser.processLine("id: bad"); + parser.processLine("data: line0"); + parser.processLine("data: line1"); + parser.processLine("data: overflow"); + parser.processLine(""); + + parser.processLine("data: next-valid-event"); + parser.processLine(""); + + assertEquals(2, events.size()); + assertEquals("good", events.get(1).id(), "Next event must carry the pre-skip id, not the skipped block's id"); + assertEquals("good", parser.getLastEventId(), "lastEventId must not be poisoned by the skipped block"); + } + + @Test + public void testSkippedBlockIdDoesNotPoisonNextEventByCharOverflow() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(20).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + parser.processLine("id: good"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good", parser.getLastEventId(), "lastEventId should be set by good event"); + + parser.processLine("id: bad"); + parser.processLine("data: " + "x".repeat(21)); + parser.processLine(""); + + parser.processLine("data: next-valid-event"); + parser.processLine(""); + + assertEquals(2, events.size()); + assertEquals("good", events.get(1).id(), "Next event must carry the pre-skip id, not the skipped block's id"); + assertEquals("good", parser.getLastEventId(), "lastEventId must not be poisoned by the skipped block"); + } + + @Test + public void testLastEventIdNotAdvancedWhenBlockSkippedByBufferCharOverflow() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(20).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + parser.processLine("id: good-id"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good-id", parser.getLastEventId(), "lastEventId should be set by good event"); + + parser.processLine("id: bad-id"); + parser.processLine("data: " + "x".repeat(21)); // triggers skip + parser.processLine(""); + assertEquals(1, events.size(), "Only the first good event should be dispatched"); + assertEquals("good-id", parser.getLastEventId(), "lastEventId must not advance for a skipped block"); + } }