From 2108d718bb9a86ba613b11c923bc255195b27a73 Mon Sep 17 00:00:00 2001 From: Brady Wied Date: Mon, 29 Jun 2026 14:04:19 -0600 Subject: [PATCH] Fix issues with chunked transfer encoding (#1) Problems 1. When handling chunked transfer encoding, if we can't read a byte from the underlying pushback stream, we end up an infinite loop. We also don't handle offset/dLen edge cases well and do not deliberately handle chunk sizes larger than Integer.MAX_VALUE. 2. Old IPR file style (modern IJ crashes) 3. Pre Savant 2.2.0 style IML 4. IJ's TestNG run config template could not properly run tests 5. GHA did not run tests on PRs and had old Savant version. Solutions 1. When handling chunked transfer encoding, if we can't read a byte from the underlying pushback stream, then bail. Also simplify and cover other edge cases (length of zero, length/offset bigger than destination buffer, and conflation of index/length). 2. Convert to the .idea directory 3. IML update. 4. Standard run config 5. Have GHA run tests on PRs and update Savant version. --- .github/workflows/test.yml | 25 +- .idea/.gitignore | 23 + .idea/compiler.xml | 8 + .idea/copyright/Apache_v2.xml | 6 + .idea/copyright/profiles_settings.xml | 7 + .../_kts_definition_dependencies.xml | 13 + .idea/misc.xml | 6 + .idea/modules.xml | 11 + .../_template__of_TestNG.xml | 24 + .idea/vcs.xml | 6 + build.savant | 2 +- java-http.iml | 43 +- java-http.ipr | 1432 ----------------- load-tests/self/self.iml | 3 +- load-tests/tomcat/tomcat.iml | 3 +- pom.xml | 2 +- .../http/io/ChunkedInputStream.java | 634 ++++---- .../java/io/fusionauth/http/ChunkedTest.java | 791 ++++----- .../http/io/ChunkedInputStreamTest.java | 697 +++++--- 19 files changed, 1342 insertions(+), 2394 deletions(-) create mode 100644 .idea/.gitignore create mode 100644 .idea/compiler.xml create mode 100644 .idea/copyright/Apache_v2.xml create mode 100644 .idea/copyright/profiles_settings.xml create mode 100644 .idea/libraries/_kts_definition_dependencies.xml create mode 100644 .idea/misc.xml create mode 100644 .idea/modules.xml create mode 100644 .idea/runConfigurations/_template__of_TestNG.xml create mode 100644 .idea/vcs.xml delete mode 100644 java-http.ipr diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index ec6cc260..97f7b000 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -4,17 +4,18 @@ name: test on: push: branches: [ main ] + pull_request: # Allows you to run this workflow manually from the Actions tab workflow_dispatch: jobs: test: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 steps: - - uses: actions/checkout@v4 - - uses: actions/setup-java@v4 + - uses: actions/checkout@v6 + - uses: actions/setup-java@v5 with: distribution: 'temurin' java-version: | @@ -22,17 +23,19 @@ jobs: 21 - name: Install Savant Build run: | - mkdir -p ~/dev/savant + curl -O https://repository.savantbuild.org/org/savantbuild/savant-core/2.2.0/savant-2.2.0.tar.gz + tar xzvf savant-2.2.0.tar.gz + savant-2.2.0/bin/sb --version + SAVANT_PATH=$(realpath -s "./savant-2.2.0/bin") + echo "${SAVANT_PATH}" >> $GITHUB_PATH mkdir -p ~/.savant/plugins - cd ~/dev/savant - curl -fSL https://github.com/savant-build/savant-core/releases/download/2.0.0/savant-2.0.0.tar.gz > savant.tar.gz - tar -xzf savant.tar.gz - ln -s savant-2.0.0 current - rm savant.tar.gz - cat < ~/.savant/plugins/org.savantbuild.plugin.java.properties + # For now, using the JDK that comes on the GHA runner + cat << EOF > ~/.savant/plugins/org.savantbuild.plugin.java.properties 17=${JAVA_HOME_17_X64} 21=${JAVA_HOME_21_X64} EOF + echo "~/.savant/plugins/org.savantbuild.plugin.java.properties" + cat ~/.savant/plugins/org.savantbuild.plugin.java.properties shell: bash - name: Run the build run: | @@ -42,7 +45,7 @@ jobs: shell: bash - name: Archive TestNG reports if: failure() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v6 with: name: testng-reports path: build/test-reports diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 00000000..c5038861 --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,23 @@ +# Default ignored files +/shelf/ +/workspace.xml +# Ignored default folder with query files +/queries/ +# Datasource local storage ignored files +/dataSources.local.xml +/dataSources/data_sources_history.xml +# Editor-based HTTP Client requests +/httpRequests/ + +# Random generated files +gbrowser_project.xml +db-forest-config.xml +copilot.data.migration.ask2agent.xml +git_toolbox_prj.xml + +# User-specific settings +/tasks.xml +/usage.statistics.xml +/dictionaries/ +/sonarlint/ +/caches/ diff --git a/.idea/compiler.xml b/.idea/compiler.xml new file mode 100644 index 00000000..4a78e563 --- /dev/null +++ b/.idea/compiler.xml @@ -0,0 +1,8 @@ + + + + + + \ No newline at end of file diff --git a/.idea/copyright/Apache_v2.xml b/.idea/copyright/Apache_v2.xml new file mode 100644 index 00000000..d89943da --- /dev/null +++ b/.idea/copyright/Apache_v2.xml @@ -0,0 +1,6 @@ + + + + diff --git a/.idea/copyright/profiles_settings.xml b/.idea/copyright/profiles_settings.xml new file mode 100644 index 00000000..796e79bf --- /dev/null +++ b/.idea/copyright/profiles_settings.xml @@ -0,0 +1,7 @@ + + + + + + \ No newline at end of file diff --git a/.idea/libraries/_kts_definition_dependencies.xml b/.idea/libraries/_kts_definition_dependencies.xml new file mode 100644 index 00000000..d8fa45d2 --- /dev/null +++ b/.idea/libraries/_kts_definition_dependencies.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 00000000..3fc8108b --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 00000000..7ed0c1bd --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,11 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/runConfigurations/_template__of_TestNG.xml b/.idea/runConfigurations/_template__of_TestNG.xml new file mode 100644 index 00000000..660e2a36 --- /dev/null +++ b/.idea/runConfigurations/_template__of_TestNG.xml @@ -0,0 +1,24 @@ + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 00000000..94a25f7f --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/build.savant b/build.savant index ce02e360..fc07bd8c 100644 --- a/build.savant +++ b/build.savant @@ -18,7 +18,7 @@ restifyVersion = "4.2.1" slf4jVersion = "2.0.17" testngVersion = "7.11.0" -project(group: "io.fusionauth", name: "java-http", version: "1.4.1", licenses: ["ApacheV2_0"]) { +project(group: "io.fusionauth", name: "java-http", version: "1.4.2", licenses: ["ApacheV2_0"]) { workflow { fetch { // Dependency resolution order: diff --git a/java-http.iml b/java-http.iml index 3575a77f..696ab1f6 100644 --- a/java-http.iml +++ b/java-http.iml @@ -17,113 +17,112 @@ - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - - + \ No newline at end of file diff --git a/java-http.ipr b/java-http.ipr deleted file mode 100644 index 7b0b5133..00000000 --- a/java-http.ipr +++ /dev/null @@ -1,1432 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/load-tests/self/self.iml b/load-tests/self/self.iml index b712131a..0170a0b6 100644 --- a/load-tests/self/self.iml +++ b/load-tests/self/self.iml @@ -19,5 +19,4 @@ - - + \ No newline at end of file diff --git a/load-tests/tomcat/tomcat.iml b/load-tests/tomcat/tomcat.iml index 2d0342d7..3c100bc9 100644 --- a/load-tests/tomcat/tomcat.iml +++ b/load-tests/tomcat/tomcat.iml @@ -21,5 +21,4 @@ - - + \ No newline at end of file diff --git a/pom.xml b/pom.xml index a32143b3..9b1a6a2c 100644 --- a/pom.xml +++ b/pom.xml @@ -2,7 +2,7 @@ 4.0.0 io.fusionauth java-http - 1.4.0 + 1.4.2 jar Java HTTP library (client and server) diff --git a/src/main/java/io/fusionauth/http/io/ChunkedInputStream.java b/src/main/java/io/fusionauth/http/io/ChunkedInputStream.java index 4f02199f..7b56aa9e 100644 --- a/src/main/java/io/fusionauth/http/io/ChunkedInputStream.java +++ b/src/main/java/io/fusionauth/http/io/ChunkedInputStream.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved + * Copyright (c) 2022-2026, FusionAuth, All Rights Reserved * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -15,11 +15,13 @@ */ package io.fusionauth.http.io; +import io.fusionauth.http.ParseException; +import io.fusionauth.http.util.HTTPTools; + import java.io.IOException; import java.io.InputStream; +import java.util.Objects; -import io.fusionauth.http.ParseException; -import io.fusionauth.http.util.HTTPTools; import static io.fusionauth.http.util.HTTPTools.makeParseException; /** @@ -28,336 +30,346 @@ * @author Brian Pontarelli */ public class ChunkedInputStream extends InputStream { - private final byte[] b1 = new byte[1]; - - private final byte[] buffer; + private final byte[] b1 = new byte[1]; - private final PushbackInputStream delegate; + private final byte[] delegateBuffer; - private final StringBuilder headerSizeHex = new StringBuilder(); + private final PushbackInputStream delegate; - private int bufferIndex; + private final StringBuilder chunkSizeBuilder = new StringBuilder(); - private int bufferLength; + private int delegateBufferIndex; - private int chunkBytesRead; + private int delegateBufferLength; - private int chunkBytesRemaining; + private int chunkBytesRead; - private int chunkSize; + private int chunkBytesRemaining; - private ChunkedBodyState state = ChunkedBodyState.ChunkSize; + private int chunkSize; - public ChunkedInputStream(PushbackInputStream delegate, int bufferSize) { - this.delegate = delegate; - this.buffer = new byte[bufferSize]; - } + private ChunkedBodyState state = ChunkedBodyState.ChunkSize; - @Override - public int read(byte[] destination, int dOff, int dLen) throws IOException { - int dIndex = dOff; - while (dIndex < dLen) { - if (state == ChunkedInputStream.ChunkedBodyState.Complete) { - pushBackOverReadBytes(); - break; - } - - // Read some more if we are out of bytes - if (bufferIndex >= bufferLength) { - bufferIndex = 0; - bufferLength = delegate.read(buffer); - } - - // Process the buffer - while (bufferIndex < bufferLength) { - ChunkedBodyState nextState; - try { - nextState = state.next(buffer[bufferIndex], chunkSize, chunkBytesRead); - } catch (ParseException e) { - // This allows us to add the index to the exception. Useful for debugging. - e.setIndex(bufferIndex); - throw e; - } + public ChunkedInputStream(PushbackInputStream delegate, int bufferSize) { + this.delegate = delegate; + this.delegateBuffer = new byte[bufferSize]; + } - // We have reached the end of the encoded payload. Push back any additional bytes read. - if (state == ChunkedBodyState.Complete) { - state = nextState; - bufferIndex++; - pushBackOverReadBytes(); - break; + @Override + public int read(byte[] destination, int dOff, int dLen) throws IOException { + if (dLen == 0) { + return 0; } - - // Capture the character to calculate the next chunk size - if (nextState == ChunkedBodyState.ChunkSize) { - headerSizeHex.appendCodePoint(buffer[bufferIndex]); - state = nextState; - bufferIndex++; - continue; + Objects.checkFromIndexSize(dOff, dLen, destination.length); + + int dEndIndex = dLen + dOff; + int dCurrentIndex = dOff; + while (dCurrentIndex < dEndIndex) { + if (state == ChunkedInputStream.ChunkedBodyState.Complete) { + pushBackOverReadBytes(); + break; + } + + // Read some more if we are out of bytes + if (delegateBufferIndex >= delegateBufferLength) { + delegateBufferIndex = 0; + delegateBufferLength = delegate.read(delegateBuffer); + // nothing left to read from the delegate stream. This is either an incomplete chunk or + // the client failed to send a terminating/terminal chunk of 0\r\n\r\n + // Per RFC 9112 section 8, this is an 'incomplete' message body + if (delegateBufferLength == -1) { + return -1; + } + } + + // Process the buffer + while (delegateBufferIndex < delegateBufferLength && dCurrentIndex < dEndIndex) { + ChunkedBodyState nextState; + try { + nextState = state.next(delegateBuffer[delegateBufferIndex], chunkSize, chunkBytesRead); + } catch (ParseException e) { + // This allows us to add the index to the exception. Useful for debugging. + e.setIndex(delegateBufferIndex); + throw e; + } + + // We have reached the end of the encoded payload. Push back any additional bytes read. + if (state == ChunkedBodyState.Complete) { + state = nextState; + delegateBufferIndex++; + pushBackOverReadBytes(); + break; + } + + // Capture the character to calculate the next chunk size + if (nextState == ChunkedBodyState.ChunkSize) { + chunkSizeBuilder.appendCodePoint(delegateBuffer[delegateBufferIndex]); + state = nextState; + delegateBufferIndex++; + continue; + } + + // We have found the chunk, this means we can now convert the captured chunk size bytes and then try and read the chunk. + if (state != ChunkedBodyState.Chunk && nextState == ChunkedBodyState.Chunk) { + if (chunkSizeBuilder.isEmpty()) { + throw new ChunkException("Chunk size is missing"); + } + + // This is the start of a chunk, so set the size and counter and reset the size hex string + long chunkSizeLong = Long.parseLong(chunkSizeBuilder, 0, chunkSizeBuilder.length(), 16); + if (chunkSizeLong > Integer.MAX_VALUE) { + throw new ChunkException("Chunk size is too large"); + } + + chunkSize = (int) chunkSizeLong; + chunkBytesRead = 0; + chunkBytesRemaining = chunkSize; + chunkSizeBuilder.delete(0, chunkSizeBuilder.length()); + + // A chunk size of 0 indicates this is the terminating chunk. Continue and we will expect the state machine + // to process the final CRLF and hit the Complete state. + if (chunkSize == 0) { + state = nextState; + continue; + } + } + + int lengthToCopy; + if (chunkBytesRemaining > 0) { + int remainingDelegateBufferBytes = delegateBufferLength - delegateBufferIndex; + // we've got: + // 1) chunkBytesRemaining - what's left in the chunk + // 2) remainingInBuffer - what's left in the buffer the chunk is in + // 3) dEndIndex - dCurrentIndex - the what's left of the total we've been asked to copy + // we have to take the minimum of all of that + lengthToCopy = Math.min(Math.min(chunkBytesRemaining, remainingDelegateBufferBytes), dEndIndex - dCurrentIndex); + } else { + // Nothing to do, continue to the next state. + state = nextState; + delegateBufferIndex++; + continue; + } + + // Copy 'lengthToCopy' to the destination buffer + System.arraycopy(delegateBuffer, delegateBufferIndex, destination, dCurrentIndex, lengthToCopy); + delegateBufferIndex += lengthToCopy; + chunkBytesRead += lengthToCopy; + chunkBytesRemaining -= lengthToCopy; + dCurrentIndex += lengthToCopy; + state = nextState; + } } - // We have found the chunk, this means we can now convert the captured chunk size bytes and then try and read the chunk. - if (state != ChunkedBodyState.Chunk && nextState == ChunkedBodyState.Chunk) { - if (headerSizeHex.isEmpty()) { - throw new ChunkException("Chunk size is missing"); - } - - // This is the start of a chunk, so set the size and counter and reset the size hex string - chunkSize = (int) Long.parseLong(headerSizeHex, 0, headerSizeHex.length(), 16); - - chunkBytesRead = 0; - chunkBytesRemaining = chunkSize; - headerSizeHex.delete(0, headerSizeHex.length()); - - // A chunk size of 0 indicates this is the terminating chunk. Continue and we will expect the state machine - // to process the final CRLF and hit the Complete state. - if (chunkSize == 0) { - state = nextState; - continue; - } - } + int total = dCurrentIndex - dOff; + return total == 0 ? -1 : total; + } - int lengthToCopy; - if (chunkBytesRemaining > 0) { - int remainingInBuffer = bufferLength - bufferIndex; - lengthToCopy = Math.min(Math.min(chunkBytesRemaining, remainingInBuffer), dLen - dIndex); // That's an ugly baby! - - // This means we don't have room in the destination buffer - if (lengthToCopy == 0) { - state = nextState; - bufferIndex++; - return dIndex - dOff; - } - } else { - // Nothing to do, continue to the next state. - state = nextState; - bufferIndex++; - continue; + @Override + public int read() throws IOException { + var read = read(b1); + if (read <= 0) { + return read; } - // Copy 'lengthToCopy' to the destination buffer - System.arraycopy(buffer, bufferIndex, destination, dIndex, lengthToCopy); - bufferIndex += lengthToCopy; - chunkBytesRead += lengthToCopy; - chunkBytesRemaining -= lengthToCopy; - dIndex += lengthToCopy; - state = nextState; - - // If we have bytes to copy, we are at the correct location in the state machine, If we don't have room, break. - // - This will break the while loop, and return at the end of this method the total bytes we have written to the destination buffer. - if (dIndex == dLen) { - break; - } - } + return b1[0] & 0xFF; } - int total = dIndex - dOff; - return total == 0 ? -1 : total; - } + // I'm not sure what the need for this is as of now. None of the existing code paths use it. could be + // useful for HTTP/2 in the future + private void pushBackOverReadBytes() { + int leftOver = delegateBufferLength - delegateBufferIndex; + if (leftOver > 0) { + delegate.push(delegateBuffer, delegateBufferIndex, leftOver); - @Override - public int read() throws IOException { - var read = read(b1); - if (read <= 0) { - return read; + // Move the pointer to the end of the buffer, We have used up the bytes by pushing them back. + delegateBufferIndex = delegateBufferLength; + } } - return b1[0] & 0xFF; - } - private void pushBackOverReadBytes() { - int leftOver = bufferLength - bufferIndex; - if (leftOver > 0) { - delegate.push(buffer, bufferIndex, leftOver); - - // Move the pointer to the end of the buffer, We have used up the bytes by pushing them back. - bufferIndex = bufferLength; + public enum ChunkedBodyState { + ChunkExtensionStart { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\r') { + return ChunkExtensionCR; + } else if (HTTPTools.isTokenCharacter(ch)) { + return ChunkExtensionName; + } + + throw makeParseException(ch, this); + } + }, + ChunkExtensionName { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\r') { + return ChunkExtensionCR; + } else if (ch == '=') { + return ChunkExtensionValueSep; + } else if (ch == ';') { + return ChunkExtensionStart; + } else if (HTTPTools.isTokenCharacter(ch)) { + return ChunkExtensionName; + } + + throw makeParseException(ch, this); + } + }, + ChunkExtensionValueSep { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\r') { + return ChunkExtensionCR; + } else if (ch == ';') { + return ChunkExtensionStart; + } else if (HTTPTools.isTokenCharacter(ch)) { + return ChunkExtensionValue; + } + + throw makeParseException(ch, this); + } + }, + ChunkExtensionValue { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\r') { + return ChunkExtensionCR; + } else if (ch == ';') { + return ChunkExtensionStart; + } else if (HTTPTools.isTokenCharacter(ch)) { + return ChunkExtensionValue; + } + + throw makeParseException(ch, this); + } + }, + ChunkExtensionCR { + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\n') { + return ChunkExtensionLF; + } + + throw makeParseException(ch, this); + } + }, + ChunkExtensionLF { + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + return Chunk; + } + }, + ChunkSize { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\r') { + return ChunkSizeCR; + } else if (ch == ';') { + return ChunkExtensionStart; + } else if (HTTPTools.isHexadecimalCharacter(ch)) { + return ChunkSize; + } + + throw makeParseException(ch, this); + } + }, + + ChunkSizeCR { + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\n') { + return ChunkSizeLF; + } + + throw makeParseException(ch, this); + } + }, + + ChunkSizeLF { + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + return Chunk; + } + }, + Chunk { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (length == 0) { + // Following the final 0 length chunk, trailers are optional. + if (HTTPTools.isURICharacter(ch)) { + return Trailer; + } else { + return Complete; + } + + } else if (bytesRead == length && ch == '\r') { + return ChunkCR; + } else if (bytesRead < length) { + return Chunk; + } + + throw makeParseException(ch, this); + } + }, + + ChunkCR { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\n') { + return length == 0 ? Complete : ChunkLF; + } + + throw makeParseException(ch, this); + } + }, + + ChunkLF { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (length == 0) { + return Complete; + } else if (HTTPTools.isHexadecimalCharacter(ch)) { + return ChunkSize; + } + + throw makeParseException(ch, this); + } + }, + + Complete { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + return Complete; + } + }, + Trailer { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\r') { + return TrailerCR; + } else { + return Trailer; + } + } + }, + TrailerCR { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (ch == '\n') { + return TrailerLF; + } + + throw makeParseException(ch, this); + } + }, + TrailerLF { + @Override + public ChunkedBodyState next(byte ch, long length, long bytesRead) { + if (HTTPTools.isURICharacter(ch)) { + return Trailer; + } else { + return Complete; + } + } + }; + + public abstract ChunkedInputStream.ChunkedBodyState next(byte ch, long length, long bytesRead); } - } - - - public enum ChunkedBodyState { - ChunkExtensionStart { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\r') { - return ChunkExtensionCR; - } else if (HTTPTools.isTokenCharacter(ch)) { - return ChunkExtensionName; - } - - throw makeParseException(ch, this); - } - }, - ChunkExtensionName { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\r') { - return ChunkExtensionCR; - } else if (ch == '=') { - return ChunkExtensionValueSep; - } else if (ch == ';') { - return ChunkExtensionStart; - } else if (HTTPTools.isTokenCharacter(ch)) { - return ChunkExtensionName; - } - - throw makeParseException(ch, this); - } - }, - ChunkExtensionValueSep { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\r') { - return ChunkExtensionCR; - } else if (ch == ';') { - return ChunkExtensionStart; - } else if (HTTPTools.isTokenCharacter(ch)) { - return ChunkExtensionValue; - } - - throw makeParseException(ch, this); - } - }, - ChunkExtensionValue { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\r') { - return ChunkExtensionCR; - } else if (ch == ';') { - return ChunkExtensionStart; - } else if (HTTPTools.isTokenCharacter(ch)) { - return ChunkExtensionValue; - } - - throw makeParseException(ch, this); - } - }, - ChunkExtensionCR { - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\n') { - return ChunkExtensionLF; - } - - throw makeParseException(ch, this); - } - }, - ChunkExtensionLF { - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - return Chunk; - } - }, - ChunkSize { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\r') { - return ChunkSizeCR; - } else if (ch == ';') { - return ChunkExtensionStart; - } else if (HTTPTools.isHexadecimalCharacter(ch)) { - return ChunkSize; - } - - throw makeParseException(ch, this); - } - }, - - ChunkSizeCR { - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\n') { - return ChunkSizeLF; - } - - throw makeParseException(ch, this); - } - }, - - ChunkSizeLF { - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - return Chunk; - } - }, - Chunk { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (length == 0) { - // Following the final 0 length chunk, trailers are optional. - if (HTTPTools.isURICharacter(ch)) { - return Trailer; - } else { - return Complete; - } - - } else if (bytesRead == length && ch == '\r') { - return ChunkCR; - } else if (bytesRead < length) { - return Chunk; - } - - throw makeParseException(ch, this); - } - }, - - ChunkCR { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\n') { - return length == 0 ? Complete : ChunkLF; - } - - throw makeParseException(ch, this); - } - }, - - ChunkLF { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (length == 0) { - return Complete; - } else if (HTTPTools.isHexadecimalCharacter(ch)) { - return ChunkSize; - } - - throw makeParseException(ch, this); - } - }, - - Complete { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - return Complete; - } - }, - Trailer { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\r') { - return TrailerCR; - } else { - return Trailer; - } - } - }, - TrailerCR { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (ch == '\n') { - return TrailerLF; - } - - throw makeParseException(ch, this); - } - }, - TrailerLF { - @Override - public ChunkedBodyState next(byte ch, long length, long bytesRead) { - if (HTTPTools.isURICharacter(ch)) { - return Trailer; - } else { - return Complete; - } - } - }; - - public abstract ChunkedInputStream.ChunkedBodyState next(byte ch, long length, long bytesRead); - } } diff --git a/src/test/java/io/fusionauth/http/ChunkedTest.java b/src/test/java/io/fusionauth/http/ChunkedTest.java index a2a51cd4..da636d63 100644 --- a/src/test/java/io/fusionauth/http/ChunkedTest.java +++ b/src/test/java/io/fusionauth/http/ChunkedTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved + * Copyright (c) 2022-2026, FusionAuth, All Rights Reserved * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -15,11 +15,17 @@ */ package io.fusionauth.http; -import java.io.ByteArrayInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.io.Writer; +import com.inversoft.net.ssl.SSLTools; +import com.inversoft.rest.RESTClient; +import com.inversoft.rest.TextResponseHandler; +import io.fusionauth.http.HTTPValues.Headers; +import io.fusionauth.http.server.CountingInstrumenter; +import io.fusionauth.http.server.HTTPHandler; +import io.fusionauth.http.server.HTTPServer; +import org.testng.annotations.Test; + +import java.io.*; +import java.net.Socket; import java.net.URI; import java.net.http.HttpRequest; import java.net.http.HttpRequest.BodyPublishers; @@ -29,17 +35,7 @@ import java.nio.file.Path; import java.nio.file.Paths; -import com.inversoft.net.ssl.SSLTools; -import com.inversoft.rest.RESTClient; -import com.inversoft.rest.TextResponseHandler; -import io.fusionauth.http.HTTPValues.Headers; -import io.fusionauth.http.server.CountingInstrumenter; -import io.fusionauth.http.server.HTTPHandler; -import io.fusionauth.http.server.HTTPServer; -import org.testng.annotations.Test; -import static org.testng.Assert.assertEquals; -import static org.testng.Assert.assertTrue; -import static org.testng.Assert.fail; +import static org.testng.Assert.*; /** * Tests various chunked Transfer-Encoding capabilities of the HTTP server. @@ -47,377 +43,412 @@ * @author Brian Pontarelli */ public class ChunkedTest extends BaseTest { - public static final String ExpectedResponse = "{\"version\":\"42\"}"; - - public static final String RequestBody = "{\"message\":\"Hello World\"}"; - - @Test(dataProvider = "schemes") - public void chunkedRequest(String scheme) throws Exception { - // Use a large chunked request - String responseBody = "These pretzels are making me thirsty. ".repeat(16_000); - byte[] responseBodyBytes = responseBody.getBytes(StandardCharsets.UTF_8); - - HTTPHandler handler = (req, res) -> { - assertTrue(req.isChunked()); - - try { - byte[] body = req.getInputStream().readAllBytes(); - assertEquals(body, responseBodyBytes); - } catch (IOException e) { - fail("Unable to parse body", e); - } - - byte[] responseBytes = ExpectedResponse.getBytes(StandardCharsets.UTF_8); - res.setHeader(Headers.ContentType, "application/json"); - res.setHeader("Content-Length", responseBytes.length + ""); - res.setStatus(200); - - try { - OutputStream outputStream = res.getOutputStream(); - outputStream.write(responseBytes); - } catch (IOException e) { - throw new RuntimeException(e); - } - }; - - CountingInstrumenter instrumenter = new CountingInstrumenter(); - try (var client = makeClient(scheme, null); HTTPServer ignore = makeServer(scheme, handler, instrumenter).start()) { - URI uri = makeURI(scheme, ""); - - // This ensures we are not just passing the test because we interrupt the InputStream and cause it to not hang. - for (int i = 0; i < 1_000; i++) { - var response = client.send(HttpRequest.newBuilder() - .uri(uri) - .header(Headers.ContentType, "text/plain") - .POST(BodyPublishers.ofInputStream(() -> - new ByteArrayInputStream(responseBodyBytes))) - .build(), - r -> BodySubscribers.ofString(StandardCharsets.UTF_8) - ); - - assertEquals(response.statusCode(), 200); - assertEquals(response.body(), ExpectedResponse); - assertEquals(instrumenter.getChunkedRequests(), (i + 1)); - } + public static final String ExpectedResponse = "{\"version\":\"42\"}"; + + public static final String RequestBody = "{\"message\":\"Hello World\"}"; + + @Test(dataProvider = "schemes") + public void chunkedRequest(String scheme) throws Exception { + // Use a large chunked request + String responseBody = "These pretzels are making me thirsty. ".repeat(16_000); + byte[] responseBodyBytes = responseBody.getBytes(StandardCharsets.UTF_8); + + HTTPHandler handler = (req, res) -> { + assertTrue(req.isChunked()); + + try { + byte[] body = req.getInputStream().readAllBytes(); + assertEquals(body, responseBodyBytes); + } catch (IOException e) { + fail("Unable to parse body", e); + } + + byte[] responseBytes = ExpectedResponse.getBytes(StandardCharsets.UTF_8); + res.setHeader(Headers.ContentType, "application/json"); + res.setHeader("Content-Length", responseBytes.length + ""); + res.setStatus(200); + + try { + OutputStream outputStream = res.getOutputStream(); + outputStream.write(responseBytes); + } catch (IOException e) { + throw new RuntimeException(e); + } + }; + + CountingInstrumenter instrumenter = new CountingInstrumenter(); + try (var client = makeClient(scheme, null); HTTPServer ignore = makeServer(scheme, handler, instrumenter).start()) { + URI uri = makeURI(scheme, ""); + + // This ensures we are not just passing the test because we interrupt the InputStream and cause it to not hang. + for (int i = 0; i < 1_000; i++) { + var response = client.send(HttpRequest.newBuilder() + .uri(uri) + .header(Headers.ContentType, "text/plain") + .POST(BodyPublishers.ofInputStream(() -> + new ByteArrayInputStream(responseBodyBytes))) + .build(), + r -> BodySubscribers.ofString(StandardCharsets.UTF_8) + ); + + assertEquals(response.statusCode(), 200); + assertEquals(response.body(), ExpectedResponse); + assertEquals(instrumenter.getChunkedRequests(), (i + 1)); + } + } } - } - - @Test(dataProvider = "schemes") - public void chunkedRequest_doNotReadTheInputStream(String scheme) throws Exception { - // Use a large chunked request - String responseBody = "These pretzels are making me thirsty. ".repeat(16_000); - byte[] responseBodyBytes = responseBody.getBytes(StandardCharsets.UTF_8); - - HTTPHandler handler = (req, res) -> { - assertTrue(req.isChunked()); - - // - // Nobody read the InputStream, the InputStream has gone bad! - // - William Lichter (Can't Hardly Wait) - - // By not reading the InputStream, the server will have to drain it before writing the response. - - byte[] responseBytes = ExpectedResponse.getBytes(StandardCharsets.UTF_8); - res.setHeader(Headers.ContentType, "application/json"); - res.setHeader("Content-Length", responseBytes.length + ""); - res.setStatus(200); - - try { - OutputStream outputStream = res.getOutputStream(); - outputStream.write(responseBytes); - } catch (IOException e) { - throw new RuntimeException(e); - } - }; - - CountingInstrumenter instrumenter = new CountingInstrumenter(); - try (var client = makeClient(scheme, null); HTTPServer ignore = makeServer(scheme, handler, instrumenter) - // Ensure we can drain the entire body. In practice, the number of bytes drained will be larger than the body - // because it is encoded using Transfer-Encoding: chunked which contains other data. So double it for good measure. - .withMaximumBytesToDrain(responseBodyBytes.length * 2) - .start()) { - URI uri = makeURI(scheme, ""); - - // This ensures we are not just passing the test because we interrupt the InputStream and cause it to not hang. - int iterations = 50; - for (int i = 0; i < iterations; i++) { - var response = client.send(HttpRequest.newBuilder() - .uri(uri) - .header(Headers.ContentType, "text/plain") - // Note that using a InputStream based publisher will caues the JDK to - // enable Transfer-Encoding: chunked - .POST(BodyPublishers.ofInputStream(() -> - new ByteArrayInputStream(responseBodyBytes))) - .build(), - r -> BodySubscribers.ofString(StandardCharsets.UTF_8) - ); - - assertEquals(response.statusCode(), 200); - assertEquals(response.body(), ExpectedResponse); - } - - // Expect that we counted the correct number of chunked requests. - long chunkedRequests = instrumenter.getChunkedRequests(); - assertEquals(chunkedRequests, iterations); + + @Test(dataProvider = "schemes") + public void chunkedRequest_doNotReadTheInputStream(String scheme) throws Exception { + // Use a large chunked request + String responseBody = "These pretzels are making me thirsty. ".repeat(16_000); + byte[] responseBodyBytes = responseBody.getBytes(StandardCharsets.UTF_8); + + HTTPHandler handler = (req, res) -> { + assertTrue(req.isChunked()); + + // + // Nobody read the InputStream, the InputStream has gone bad! + // - William Lichter (Can't Hardly Wait) + + // By not reading the InputStream, the server will have to drain it before writing the response. + + byte[] responseBytes = ExpectedResponse.getBytes(StandardCharsets.UTF_8); + res.setHeader(Headers.ContentType, "application/json"); + res.setHeader("Content-Length", responseBytes.length + ""); + res.setStatus(200); + + try { + OutputStream outputStream = res.getOutputStream(); + outputStream.write(responseBytes); + } catch (IOException e) { + throw new RuntimeException(e); + } + }; + + CountingInstrumenter instrumenter = new CountingInstrumenter(); + try (var client = makeClient(scheme, null); HTTPServer ignore = makeServer(scheme, handler, instrumenter) + // Ensure we can drain the entire body. In practice, the number of bytes drained will be larger than the body + // because it is encoded using Transfer-Encoding: chunked which contains other data. So double it for good measure. + .withMaximumBytesToDrain(responseBodyBytes.length * 2) + .start()) { + URI uri = makeURI(scheme, ""); + + // This ensures we are not just passing the test because we interrupt the InputStream and cause it to not hang. + int iterations = 50; + for (int i = 0; i < iterations; i++) { + var response = client.send(HttpRequest.newBuilder() + .uri(uri) + .header(Headers.ContentType, "text/plain") + // Note that using a InputStream based publisher will caues the JDK to + // enable Transfer-Encoding: chunked + .POST(BodyPublishers.ofInputStream(() -> + new ByteArrayInputStream(responseBodyBytes))) + .build(), + r -> BodySubscribers.ofString(StandardCharsets.UTF_8) + ); + + assertEquals(response.statusCode(), 200); + assertEquals(response.body(), ExpectedResponse); + } + + // Expect that we counted the correct number of chunked requests. + long chunkedRequests = instrumenter.getChunkedRequests(); + assertEquals(chunkedRequests, iterations); + } } - } - - @Test(dataProvider = "schemes") - public void chunkedResponse(String scheme) throws Exception { - HTTPHandler handler = (req, res) -> { - res.setHeader(Headers.ContentType, "text/plain"); - res.setStatus(200); - - try { - OutputStream outputStream = res.getOutputStream(); - outputStream.write(ExpectedResponse.getBytes()); - outputStream.close(); - } catch (IOException e) { - throw new RuntimeException(e); - } - }; - - CountingInstrumenter instrumenter = new CountingInstrumenter(); - try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler, instrumenter).start()) { - URI uri = makeURI(scheme, ""); - var response = client.send( - HttpRequest.newBuilder().uri(uri).header(Headers.ContentType, "application/json").GET().build(), - r -> BodySubscribers.ofString(StandardCharsets.UTF_8) - ); - - assertEquals(response.statusCode(), 200); - assertEquals(response.body(), ExpectedResponse); - assertEquals(instrumenter.getChunkedResponses(), 1); + + @Test(dataProvider = "schemes") + public void chunkedResponse(String scheme) throws Exception { + HTTPHandler handler = (req, res) -> { + res.setHeader(Headers.ContentType, "text/plain"); + res.setStatus(200); + + try { + OutputStream outputStream = res.getOutputStream(); + outputStream.write(ExpectedResponse.getBytes()); + outputStream.close(); + } catch (IOException e) { + throw new RuntimeException(e); + } + }; + + CountingInstrumenter instrumenter = new CountingInstrumenter(); + try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler, instrumenter).start()) { + URI uri = makeURI(scheme, ""); + var response = client.send( + HttpRequest.newBuilder().uri(uri).header(Headers.ContentType, "application/json").GET().build(), + r -> BodySubscribers.ofString(StandardCharsets.UTF_8) + ); + + assertEquals(response.statusCode(), 200); + assertEquals(response.body(), ExpectedResponse); + assertEquals(instrumenter.getChunkedResponses(), 1); + } } - } - - @Test(dataProvider = "schemes") - public void chunkedResponseRestify(String scheme) { - String html = """ - Success! - parm=some values - theRest=some other values - """; - HTTPHandler handler = (req, res) -> { - res.setHeader(Headers.ContentType, "text/html; charset=UTF-8"); - res.setHeader(Headers.CacheControl, "no-cache"); - res.setStatus(200); - - try { - Writer writer = res.getWriter(); - writer.write(html); - } catch (IOException e) { - throw new RuntimeException(e); - } - }; - - CountingInstrumenter instrumenter = new CountingInstrumenter(); - try (HTTPServer ignore = makeServer(scheme, handler, instrumenter).start()) { - SSLTools.disableSSLValidation(); - URI uri = makeURI(scheme, ""); - var response = new RESTClient<>(String.class, String.class).url(uri.toString()) - .get() - .successResponseHandler(new TextResponseHandler()) - .errorResponseHandler(new TextResponseHandler()) - .go(); - assertEquals(response.status, 200); - assertEquals(response.successResponse, html); - assertEquals(instrumenter.getChunkedResponses(), 1); - } finally { - SSLTools.enableSSLValidation(); + + @Test(dataProvider = "schemes") + public void chunkedResponseRestify(String scheme) { + String html = """ + Success! + parm=some values + theRest=some other values + """; + HTTPHandler handler = (req, res) -> { + res.setHeader(Headers.ContentType, "text/html; charset=UTF-8"); + res.setHeader(Headers.CacheControl, "no-cache"); + res.setStatus(200); + + try { + Writer writer = res.getWriter(); + writer.write(html); + } catch (IOException e) { + throw new RuntimeException(e); + } + }; + + CountingInstrumenter instrumenter = new CountingInstrumenter(); + try (HTTPServer ignore = makeServer(scheme, handler, instrumenter).start()) { + SSLTools.disableSSLValidation(); + URI uri = makeURI(scheme, ""); + var response = new RESTClient<>(String.class, String.class).url(uri.toString()) + .get() + .successResponseHandler(new TextResponseHandler()) + .errorResponseHandler(new TextResponseHandler()) + .go(); + assertEquals(response.status, 200); + assertEquals(response.successResponse, html); + assertEquals(instrumenter.getChunkedResponses(), 1); + } finally { + SSLTools.enableSSLValidation(); + } } - } - - @Test(dataProvider = "schemes") - public void chunkedResponseStreamingFile(String scheme) throws Exception { - Path file = Paths.get("src/test/java/io/fusionauth/http/ChunkedTest.java"); - HTTPHandler handler = (req, res) -> { - res.setHeader(Headers.ContentType, "text/plain"); - res.setStatus(200); - - try (InputStream is = Files.newInputStream(file)) { - OutputStream outputStream = res.getOutputStream(); - is.transferTo(outputStream); - outputStream.close(); - } catch (IOException e) { - throw new RuntimeException(e); - } - }; - - CountingInstrumenter instrumenter = new CountingInstrumenter(); - try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler, instrumenter).start()) { - URI uri = makeURI(scheme, ""); - var response = client.send( - HttpRequest.newBuilder().uri(uri).header(Headers.ContentType, "application/json").GET().build(), - r -> BodySubscribers.ofString(StandardCharsets.UTF_8) - ); - - assertEquals(response.statusCode(), 200); - assertEquals(response.body(), Files.readString(file)); - assertEquals(instrumenter.getChunkedResponses(), 1); + + @Test(dataProvider = "schemes") + public void chunkedResponseStreamingFile(String scheme) throws Exception { + Path file = Paths.get("src/test/java/io/fusionauth/http/ChunkedTest.java"); + HTTPHandler handler = (req, res) -> { + res.setHeader(Headers.ContentType, "text/plain"); + res.setStatus(200); + + try (InputStream is = Files.newInputStream(file)) { + OutputStream outputStream = res.getOutputStream(); + is.transferTo(outputStream); + outputStream.close(); + } catch (IOException e) { + throw new RuntimeException(e); + } + }; + + CountingInstrumenter instrumenter = new CountingInstrumenter(); + try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler, instrumenter).start()) { + URI uri = makeURI(scheme, ""); + var response = client.send( + HttpRequest.newBuilder().uri(uri).header(Headers.ContentType, "application/json").GET().build(), + r -> BodySubscribers.ofString(StandardCharsets.UTF_8) + ); + + assertEquals(response.statusCode(), 200); + assertEquals(response.body(), Files.readString(file)); + assertEquals(instrumenter.getChunkedResponses(), 1); + } } - } - - @Test(dataProvider = "schemes") - public void chunkedResponseWriter(String scheme) throws Exception { - String html = """ - Success! - parm=some values - theRest=some other values - """; - HTTPHandler handler = (req, res) -> { - res.setHeader(Headers.ContentType, "text/html; charset=UTF-8"); - res.setHeader(Headers.CacheControl, "no-cache"); - res.setStatus(200); - - try { - Writer writer = res.getWriter(); - writer.write(html); - } catch (IOException e) { - throw new RuntimeException(e); - } - }; - - CountingInstrumenter instrumenter = new CountingInstrumenter(); - try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler, instrumenter).start()) { - URI uri = makeURI(scheme, ""); - var response = client.send( - HttpRequest.newBuilder().uri(uri).GET().build(), - r -> BodySubscribers.ofString(StandardCharsets.UTF_8) - ); - assertEquals(response.statusCode(), 200); - assertEquals(response.body(), html); - assertEquals(instrumenter.getChunkedResponses(), 1); + + @Test(dataProvider = "schemes") + public void chunkedResponseWriter(String scheme) throws Exception { + String html = """ + Success! + parm=some values + theRest=some other values + """; + HTTPHandler handler = (req, res) -> { + res.setHeader(Headers.ContentType, "text/html; charset=UTF-8"); + res.setHeader(Headers.CacheControl, "no-cache"); + res.setStatus(200); + + try { + Writer writer = res.getWriter(); + writer.write(html); + } catch (IOException e) { + throw new RuntimeException(e); + } + }; + + CountingInstrumenter instrumenter = new CountingInstrumenter(); + try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler, instrumenter).start()) { + URI uri = makeURI(scheme, ""); + var response = client.send( + HttpRequest.newBuilder().uri(uri).GET().build(), + r -> BodySubscribers.ofString(StandardCharsets.UTF_8) + ); + assertEquals(response.statusCode(), 200); + assertEquals(response.body(), html); + assertEquals(instrumenter.getChunkedResponses(), 1); + } + } + + @Test(dataProvider = "schemes", groups = "performance") + public void performanceChunked(String scheme) throws Exception { + verbose = true; + String responseBody = "These pretzels are making me thirsty. ".repeat(16_000); + byte[] responseBodyBytes = responseBody.getBytes(StandardCharsets.UTF_8); + + HTTPHandler handler = (req, res) -> { + res.setHeader(Headers.ContentType, "text/plain"); + res.setStatus(200); + + try { + OutputStream outputStream = res.getOutputStream(); + outputStream.write(responseBodyBytes); + outputStream.close(); + } catch (IOException e) { + throw new RuntimeException(e); + } + }; + + int iterations = 15_000; + CountingInstrumenter instrumenter = new CountingInstrumenter(); + try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler, instrumenter) + // Note default max requests per connection is 100k, so we shouldn't be closing the connections based upon the total requests. + .start()) { + URI uri = makeURI(scheme, ""); + long start = System.currentTimeMillis(); + long lastLog = start; + + for (int i = 0; i < iterations; i++) { + var response = client.send( + HttpRequest.newBuilder().uri(uri).GET().build(), + r -> BodySubscribers.ofString(StandardCharsets.UTF_8) + ); + + assertEquals(response.statusCode(), 200); + assertEquals(response.body(), responseBody); + + if (System.currentTimeMillis() - lastLog > 5_000) { + long now = System.currentTimeMillis(); + double currentAverage = (now - start) / (double) i; + printf("Chunked Performance: Iterations [%,d] Response body [%,d] bytes. Running average is [%f] ms.\n", i, responseBodyBytes.length, currentAverage); + lastLog = System.currentTimeMillis(); + } + } + + long end = System.currentTimeMillis(); + double average = (end - start) / (double) iterations; + printf("Chunked Performance: Iterations [%,d] Response body [%,d] bytes. Final average is [%f] ms.\n", iterations, responseBodyBytes.length, average); + + // HTTP + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.563467] ms. + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.572667] ms + // - With updates to ChunkedInputStream to read 1 byte at a time until we find the chunkSize + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.567000] ms. + // - With updates to loop on processChunk + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.587400] ms. + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.588800] ms. + // - With only a larger buffer size + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.560933] ms. + + // HTTPS + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.998800] ms. + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [1.025333] ms. + // - With updates to ChunkedInputStream to read 1 byte at a time until we find the chunkSize + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [1.047000] ms. + // - With updates to loop on processChunk + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.984133] ms. + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.997933] ms. + // - With only a larger buffer size + // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [1.000400] ms. + } + + // We are using keep-alive, so expect 1 connection, and the total requests accepted, and chunked responses should equal the iteration count. + // - This assertion does seem to fail every once in a while, and it will be 2 instead of 1. My guess is that this is ok - we don't always know + // how an HTTP client is going to work, and it may decide to cycle a connection even if the server didn't force it. We are using the JDK + // REST client which does seem to be fairly predictable, but for example, using a REST client that uses HttpURLConnection is much less + // predictable, but fast. 😀 + // + // - Going to call this a pass if we have one or two connections. + assertTrue(instrumenter.getConnections() == 1 || instrumenter.getConnections() == 2); + assertEquals(instrumenter.getChunkedResponses(), iterations); + assertEquals(instrumenter.getAcceptedRequests(), iterations); } - } - - @Test(dataProvider = "schemes", groups = "performance") - public void performanceChunked(String scheme) throws Exception { - verbose = true; - String responseBody = "These pretzels are making me thirsty. ".repeat(16_000); - byte[] responseBodyBytes = responseBody.getBytes(StandardCharsets.UTF_8); - - HTTPHandler handler = (req, res) -> { - res.setHeader(Headers.ContentType, "text/plain"); - res.setStatus(200); - - try { - OutputStream outputStream = res.getOutputStream(); - outputStream.write(responseBodyBytes); - outputStream.close(); - } catch (IOException e) { - throw new RuntimeException(e); - } - }; - - int iterations = 15_000; - CountingInstrumenter instrumenter = new CountingInstrumenter(); - try (var client = makeClient(scheme, null); var ignore = makeServer(scheme, handler, instrumenter) - // Note default max requests per connection is 100k, so we shouldn't be closing the connections based upon the total requests. - .start()) { - URI uri = makeURI(scheme, ""); - long start = System.currentTimeMillis(); - long lastLog = start; - - for (int i = 0; i < iterations; i++) { - var response = client.send( - HttpRequest.newBuilder().uri(uri).GET().build(), - r -> BodySubscribers.ofString(StandardCharsets.UTF_8) - ); - - assertEquals(response.statusCode(), 200); - assertEquals(response.body(), responseBody); - - if (System.currentTimeMillis() - lastLog > 5_000) { - long now = System.currentTimeMillis(); - double currentAverage = (now - start) / (double) i; - printf("Chunked Performance: Iterations [%,d] Response body [%,d] bytes. Running average is [%f] ms.\n", i, responseBodyBytes.length, currentAverage); - lastLog = System.currentTimeMillis(); + + @Test + public void chunkedRequest_integerOverflowChunkSize() throws Exception { + // Use case: Larger chunk size than we support + + HTTPHandler handler = (req, res) -> { + // try and read the request + req.getInputStream().readAllBytes(); + }; + + try (HTTPServer ignore = makeServer("http", handler, new CountingInstrumenter()).start()) { + // Chunk size of Integer.MAX_VALUE + 1 (0x80000000) + long chunkSize = (long) Integer.MAX_VALUE + 1; + String request = "POST /api/system/version HTTP/1.1\r\n" + + "Host: localhost:4242\r\n" + + "Transfer-Encoding: chunked\r\n" + + "Content-Type: text/plain\r\n" + + "\r\n" + + Long.toHexString(chunkSize) + "\r\n" + + "AB\r\n" + + "0\r\n" + + "\r\n"; + + try (Socket socket = new Socket("127.0.0.1", 4242); + OutputStream os = socket.getOutputStream(); + InputStream is = socket.getInputStream()) { + os.write(request.getBytes(StandardCharsets.UTF_8)); + os.flush(); + byte[] response = is.readAllBytes(); + String responseString = new String(response, StandardCharsets.UTF_8); + assertEquals(responseString, + "HTTP/1.1 500 \r\nconnection: close\r\ncontent-length: 0\r\n\r\n"); + } } - } - - long end = System.currentTimeMillis(); - double average = (end - start) / (double) iterations; - printf("Chunked Performance: Iterations [%,d] Response body [%,d] bytes. Final average is [%f] ms.\n", iterations, responseBodyBytes.length, average); - - // HTTP - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.563467] ms. - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.572667] ms - // - With updates to ChunkedInputStream to read 1 byte at a time until we find the chunkSize - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.567000] ms. - // - With updates to loop on processChunk - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.587400] ms. - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.588800] ms. - // - With only a larger buffer size - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.560933] ms. - - // HTTPS - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.998800] ms. - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [1.025333] ms. - // - With updates to ChunkedInputStream to read 1 byte at a time until we find the chunkSize - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [1.047000] ms. - // - With updates to loop on processChunk - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.984133] ms. - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [0.997933] ms. - // - With only a larger buffer size - // Chunked Performance: Iterations [15,000] Response body [608,000] bytes. Final average is [1.000400] ms. } - // We are using keep-alive, so expect 1 connection, and the total requests accepted, and chunked responses should equal the iteration count. - // - This assertion does seem to fail every once in a while, and it will be 2 instead of 1. My guess is that this is ok - we don't always know - // how an HTTP client is going to work, and it may decide to cycle a connection even if the server didn't force it. We are using the JDK - // REST client which does seem to be fairly predictable, but for example, using a REST client that uses HttpURLConnection is much less - // predictable, but fast. 😀 - // - // - Going to call this a pass if we have one or two connections. - assertTrue(instrumenter.getConnections() == 1 || instrumenter.getConnections() == 2); - assertEquals(instrumenter.getChunkedResponses(), iterations); - assertEquals(instrumenter.getAcceptedRequests(), iterations); - } - - @Test(dataProvider = "schemes") - public void smallChunkedRequest(String scheme) throws Exception { - // Use a very small chunked request. This is testing various buffer sizes. - - HTTPHandler handler = (req, res) -> { - assertTrue(req.isChunked()); - - try { - byte[] body = req.getInputStream().readAllBytes(); - assertEquals(new String(body), RequestBody); - } catch (IOException e) { - fail("Unable to parse body", e); - } - - byte[] responseBytes = ExpectedResponse.getBytes(StandardCharsets.UTF_8); - res.setHeader(Headers.ContentType, "application/json"); - res.setHeader("Content-Length", responseBytes.length + ""); - res.setStatus(200); - - try { - OutputStream outputStream = res.getOutputStream(); - outputStream.write(responseBytes); - } catch (IOException e) { - throw new RuntimeException(e); - } - }; - - CountingInstrumenter instrumenter = new CountingInstrumenter(); - try (var client = makeClient(scheme, null); HTTPServer ignore = makeServer(scheme, handler, instrumenter).start()) { - URI uri = makeURI(scheme, ""); - var response = client.send( - HttpRequest.newBuilder() - .uri(uri) - .header(Headers.ContentType, "application/json") - .POST(BodyPublishers.ofInputStream(() -> - new ByteArrayInputStream(RequestBody.getBytes()))) - .build(), - r -> BodySubscribers.ofString(StandardCharsets.UTF_8) - ); - - assertEquals(response.statusCode(), 200); - assertEquals(response.body(), ExpectedResponse); - assertEquals(instrumenter.getChunkedRequests(), 1); + @Test(dataProvider = "schemes") + public void smallChunkedRequest(String scheme) throws Exception { + // Use a very small chunked request. This is testing various buffer sizes. + + HTTPHandler handler = (req, res) -> { + assertTrue(req.isChunked()); + + try { + byte[] body = req.getInputStream().readAllBytes(); + assertEquals(new String(body), RequestBody); + } catch (IOException e) { + fail("Unable to parse body", e); + } + + byte[] responseBytes = ExpectedResponse.getBytes(StandardCharsets.UTF_8); + res.setHeader(Headers.ContentType, "application/json"); + res.setHeader("Content-Length", responseBytes.length + ""); + res.setStatus(200); + + try { + OutputStream outputStream = res.getOutputStream(); + outputStream.write(responseBytes); + } catch (IOException e) { + throw new RuntimeException(e); + } + }; + + CountingInstrumenter instrumenter = new CountingInstrumenter(); + try (var client = makeClient(scheme, null); HTTPServer ignore = makeServer(scheme, handler, instrumenter).start()) { + URI uri = makeURI(scheme, ""); + var response = client.send( + HttpRequest.newBuilder() + .uri(uri) + .header(Headers.ContentType, "application/json") + .POST(BodyPublishers.ofInputStream(() -> + new ByteArrayInputStream(RequestBody.getBytes()))) + .build(), + r -> BodySubscribers.ofString(StandardCharsets.UTF_8) + ); + + assertEquals(response.statusCode(), 200); + assertEquals(response.body(), ExpectedResponse); + assertEquals(instrumenter.getChunkedRequests(), 1); + } } - } } diff --git a/src/test/java/io/fusionauth/http/io/ChunkedInputStreamTest.java b/src/test/java/io/fusionauth/http/io/ChunkedInputStreamTest.java index 3605e99f..febe8616 100644 --- a/src/test/java/io/fusionauth/http/io/ChunkedInputStreamTest.java +++ b/src/test/java/io/fusionauth/http/io/ChunkedInputStreamTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved + * Copyright (c) 2022-2026, FusionAuth, All Rights Reserved * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -15,268 +15,501 @@ */ package io.fusionauth.http.io; +import io.fusionauth.http.util.ThrowingFunction; +import org.testng.annotations.Test; + import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; import java.nio.charset.StandardCharsets; -import io.fusionauth.http.util.ThrowingFunction; -import org.testng.annotations.Test; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.fail; /** * @author Brian Pontarelli */ @Test public class ChunkedInputStreamTest { - @SuppressWarnings("GrazieInspection") - @Test - public void chunkExtensions() throws Exception { - // Test extensions - // - We do not support these, but we need to be able to ignore them w/out puking. - // - // ;foo=bar Single extension - // ;foo= Single extension, no value - // ;foo Single extension, no value, no equals - // ;foo;bar Two extensions, no values, no equals - // ;foo;bar= Two extensions, no values - // ;foo;bar=baz Two extensions, one value, one equals - // ;foo=;bar=baz Two extensions, one value, one equals - // ;foo=bar;bar=baz Two extensions, two values - // ; No extension, only a separator. Not sure if this is valid, but we should be able to ignore it. - // 0;foo=bar;bar Extensions on the final 0 chunk - withBody( - """ - 3;foo=bar\r - Hi \r - 4;foo=\r - mom!\r - 3;foo\r - Lo\r - 2;foo;bar\r - ok\r - 1;foo;bar=\r - \r - 1;foo;bar=baz\r - n\r - 2;foo=bar;baz\r - o \r - 3;foo=bar;bar=baz\r - ext\r - 2;\r - en\r - 4\r - sion\r - 2;\r - s!\r - 0;foo=bar;bar\r - \r - """) - .assertResult("Hi mom! Look no extensions!") - .assertLeftOverBytes(0); - } - - @Test - public void multipleChunks() throws Exception { - withBody(""" - A\r - 1234567890\r - 14\r - 12345678901234567890\r - 1E\r - 123456789012345678901234567890\r - 0\r - \r - """ - ).assertResult("123456789012345678901234567890123456789012345678901234567890") - .assertLeftOverBytes(0) - .assertNextRead(ChunkedInputStream::read, -1); - } - - @Test - public void ok() throws Exception { - withBody( - """ - 3\r - Hi \r - 4\r - mom!\r - 0\r - \r - """ - ).assertResult("Hi mom!") - .assertLeftOverBytes(0); - } - - @Test - public void partialChunks() throws IOException { - var buf = new byte[1024]; - var inputStream = new ChunkedInputStream(withParts( - """ - A\r - 12345678""", - """ - 90\r - 14\r - 12345678901234567890\r - 1E\r - 123456789012345678901234567890\r - 0\r - \r - """), 1024); - // All chunks will be read on the first attempt because the buffer is large enough - assertEquals(inputStream.read(buf), 60); - var result = new String(buf, 0, 60); - assertEquals(result, "123456789012345678901234567890123456789012345678901234567890"); - assertEquals(inputStream.read(), -1); - } - - @Test - public void partialHeader() throws IOException { - var buf = new byte[1024]; - var inputStream = new ChunkedInputStream(withParts( - """ - A\r - 1234567890\r - 14""", - """ - \r - 12345678901234567890\r - 0\r - \r - """), 1024); - - // All chunks will be read on the first attempt because the buffer is large enough - assertEquals(inputStream.read(buf), 30); - var result = new String(buf, 0, 30); - assertEquals(result, "123456789012345678901234567890"); - assertEquals(inputStream.read(buf), -1); - } - - @Test - public void trailers() throws Exception { - // It isn't clear if any HTTP server actually users or supports trailers. But, the spec indicates we should at least ignore them. - // - https://www.rfc-editor.org/rfc/rfc2616.html#section-3.6.1 - withBody( - """ - 30\r - There is no fate but what we make for ourselves.\r - 12\r - - - Sarah Connor - \r - 0\r - Judgement-Day: August 29, 1997 2:14 AM EDT\r - \r - """) - .assertResult(""" - There is no fate but what we make for ourselves. - - Sarah Connor - """) - // If we correctly read to the end of the InputStream we should not have any bytes left over in the PushbackInputStream - .assertLeftOverBytes(0); - } - - private Builder withBody(String body) { - return new Builder().withBody(body); - } - - private PushbackInputStream withParts(String... parts) { - return new PushbackInputStream(new PieceMealInputStream(parts), null); - } - - @SuppressWarnings("UnusedReturnValue") - private static class Builder { - public String body; - - public ChunkedInputStream chunkedInputStream; - - public PushbackInputStream pushbackInputStream; - - /** - * Used to ensure the parser worked correctly and was able to read to the end of the encoded body. - * - * @param expected the number of expected bytes that were over-read. - * @return this. - */ - public Builder assertLeftOverBytes(int expected) throws IOException { - int actual = pushbackInputStream.getAvailableBufferedBytesRemaining(); - if (actual != expected) { - if (actual > 0) { - byte[] leftOverBytes = new byte[actual]; - int leftOverRead = pushbackInputStream.read(leftOverBytes); - // No reason to think these would not be equal... but they better be. - assertEquals(leftOverBytes.length, leftOverRead); - assertEquals(actual, expected, "\nHere is what was left over in the buffer\n[" + new String(leftOverBytes) + "]"); + @SuppressWarnings("GrazieInspection") + @Test + public void chunkExtensions() throws Exception { + // Test extensions + // - We do not support these, but we need to be able to ignore them w/out puking. + // + // ;foo=bar Single extension + // ;foo= Single extension, no value + // ;foo Single extension, no value, no equals + // ;foo;bar Two extensions, no values, no equals + // ;foo;bar= Two extensions, no values + // ;foo;bar=baz Two extensions, one value, one equals + // ;foo=;bar=baz Two extensions, one value, one equals + // ;foo=bar;bar=baz Two extensions, two values + // ; No extension, only a separator. Not sure if this is valid, but we should be able to ignore it. + // 0;foo=bar;bar Extensions on the final 0 chunk + withBody( + """ + 3;foo=bar\r + Hi \r + 4;foo=\r + mom!\r + 3;foo\r + Lo\r + 2;foo;bar\r + ok\r + 1;foo;bar=\r + \r + 1;foo;bar=baz\r + n\r + 2;foo=bar;baz\r + o \r + 3;foo=bar;bar=baz\r + ext\r + 2;\r + en\r + 4\r + sion\r + 2;\r + s!\r + 0;foo=bar;bar\r + \r + """) + .assertResult("Hi mom! Look no extensions!") + .assertLeftOverBytes(0); + } + + @Test + public void multipleChunks() throws Exception { + withBody(""" + A\r + 1234567890\r + 14\r + 12345678901234567890\r + 1E\r + 123456789012345678901234567890\r + 0\r + \r + """ + ).assertResult("123456789012345678901234567890123456789012345678901234567890") + .assertLeftOverBytes(0) + .assertNextRead(ChunkedInputStream::read, -1); + } + + @Test + public void ok() throws Exception { + withBody( + """ + 3\r + Hi \r + 4\r + mom!\r + 0\r + \r + """ + ).assertResult("Hi mom!") + .assertLeftOverBytes(0); + } + + @Test + public void no_chunks_can_be_read() throws IOException { + // Use case: After reading 1 byte (the non terminated chunk size), no carriage return is received and we are + // unable to read anything from the underlying pushback input stream. We should bail. + + // arrange + String body = "3"; + PushbackInputStream pushbackInputStream = new PushbackInputStream(new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), + null); + ChunkedInputStream chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); + + // act + byte[] result = chunkedInputStream.readAllBytes(); + + // assert + assertEquals(result.length, + 0); + } + + @Test + public void zero_length() throws IOException { + // arrange + String body = ""; + PushbackInputStream pushbackInputStream = new PushbackInputStream(new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), + null); + ChunkedInputStream chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); + byte[] buffer = new byte[0]; + + // act + int result = chunkedInputStream.read(buffer, 0, 0); + + // assert + assertEquals(result, 0, + "contract says if len is zero, no bytes are read and zero is returned"); + } + + @Test + public void non_zero_offset() throws IOException { + // Use case: We have 2 chunks, a 10 byte chunk and a 5 byte chunk. If we ask for 15 bytes + // we should read 10 bytes from the first chunk and 5 from the second chunk. + // note that the offset should be independent of where the data is being read from. it + // only should affect the destination buffer. + + // arrange + // two chunks: 10 bytes ("ABCDEFGHIJ") + 5 bytes ("KLMNO") = 15 bytes total + String body = "a\r\nABCDEFGHIJ\r\n5\r\nKLMNO\r\n0\r\n\r\n"; + PushbackInputStream pushbackInputStream = new PushbackInputStream( + new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), null); + ChunkedInputStream chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); + + byte[] dest = new byte[25]; + + // act + int read = chunkedInputStream.read(dest, 10, 15); + + // assert + assertEquals(read, + 15, + "We asked for 15 bytes, and those exist because the buffer from 'body' is 30 bytes, therefore we should get 15 bytes"); + assertEquals(new String(dest, 10, read, StandardCharsets.UTF_8), + "ABCDEFGHIJKLMNO", + "dest[10..24] must contain both chunks"); + } + + @Test + public void bufferOverrun_zero_offset() throws IOException { + // Use case: dLen is bigger than the buffer size with zero offset + + // arrange + // one 20-byte chunk + String body = "14\r\nABCDEFGHIJKLMNOPQRST\r\n0\r\n\r\n"; + PushbackInputStream pushbackInputStream = new PushbackInputStream( + new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), null); + ChunkedInputStream chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); + + byte[] destinationBuffer = new byte[10]; + + // act + try { + chunkedInputStream.read(destinationBuffer, 0, 20); + fail("expected an exception"); + } catch (IndexOutOfBoundsException e) { + assertEquals(e.getMessage(), + "Range [0, 0 + 20) out of bounds for length 10"); } - } + } + + @Test + public void bufferOverrun_nonzero_offset() throws IOException { + // Use case: dLen is bigger than the buffer size with non-zero offset + + // arrange + // one 20-byte chunk + String body = "14\r\nABCDEFGHIJKLMNOPQRST\r\n0\r\n\r\n"; + PushbackInputStream pushbackInputStream = new PushbackInputStream( + new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), null); + ChunkedInputStream chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); - return this; + byte[] destinationBuffer = new byte[10]; + + // act + try { + chunkedInputStream.read(destinationBuffer, 1, 19); + fail("expected an exception"); + } catch (IndexOutOfBoundsException e) { + assertEquals(e.getMessage(), + "Range [1, 1 + 19) out of bounds for length 10"); + } } - public Builder assertNextRead(ThrowingFunction function, int expected) throws Exception { - var result = function.apply(chunkedInputStream); - assertEquals(result, expected); - return this; + @Test + public void chunk_larger_than_32_bytes() throws IOException { + // arrange + // 80000000 is Integer.MAX_VALUE + 1 in hex + String body = "80000000\r\nABC\r\n0\r\n\r\n"; + PushbackInputStream pushbackInputStream = new PushbackInputStream( + new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), null); + ChunkedInputStream chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); + byte[] dest = new byte[5]; + + // act + assert + try { + chunkedInputStream.read(dest, 0, 5); + fail("expected an exception"); + } catch (ChunkException e) { + assertEquals(e.getMessage(), + "Chunk size is too large"); + } } - public Builder assertResult(String expected) throws IOException { - pushbackInputStream = new PushbackInputStream(new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), null); - chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); + @Test + public void cross_chunk_read_does_not_overflow_destination() throws IOException { + // Use case: 2 chunks, 8 hex bytes total. the first chunk is 3 bytes. The second one is 5. + // If we ready 5 bytes, we'll need the first chunk and the 2nd chunk. + + // arrange + String body = "3\r\nABC\r\n5\r\nDEFGH\r\n0\r\n\r\n"; + PushbackInputStream pushbackInputStream = new PushbackInputStream( + new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), null); + ChunkedInputStream chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); - String actual = new String(chunkedInputStream.readAllBytes(), StandardCharsets.UTF_8); - assertEquals(actual, expected); - return this; + byte[] dest = new byte[5]; + + // act + // 5 bytes requested, crossing the chunk boundary (3 from chunk 1, 2 from chunk 2) + int read = chunkedInputStream.read(dest, 0, 5); + + // assert + assertEquals(read, 5); + assertEquals(new String(dest, 0, 5, StandardCharsets.UTF_8), "ABCDE"); } - public Builder withBody(String body) { - this.body = body; - return this; + @Test + public void destination_smaller_than_chunk() throws IOException { + // Use case: The destination buffer is smaller than a single chunk. + + // arrange + // one 20-byte chunk + String body = "14\r\nABCDEFGHIJKLMNOPQRST\r\n0\r\n\r\n"; + PushbackInputStream pushbackInputStream = new PushbackInputStream( + new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), null); + ChunkedInputStream chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); + + byte[] destinationBuffer = new byte[10]; + + // act – first read: We read 10 of the 20 bytes + int read1 = chunkedInputStream.read(destinationBuffer, 0, 10); + + // assert + assertEquals(read1, 10, "We asked for 10 bytes"); + StringBuilder stringBuilder = new StringBuilder(); + String string = new String(destinationBuffer, 0, 10, StandardCharsets.UTF_8); + assertEquals(string, "ABCDEFGHIJ", + "first read must yield the first half of the chunk"); + stringBuilder.append(string); + + // act – second read: read 5 more of the 20 bytes total + int read2 = chunkedInputStream.read(destinationBuffer, 0, 5); + + // assert + assertEquals(read2, 5, "second read must return the remaining chunk bytes"); + + // now read the final 5 bytes in at offset 5 + int read3 = chunkedInputStream.read(destinationBuffer, 5, 5); + assertEquals(read3, 5); + string = new String(destinationBuffer, 0, 10, StandardCharsets.UTF_8); + assertEquals(string, "KLMNOPQRST", + "2nd and 3rd read must yield the second half of the chunk"); + stringBuilder.append(string); + + // act – body is exhausted + int read4 = chunkedInputStream.read(destinationBuffer, 0, 10); + + // assert + assertEquals(read4, -1, "third read must signal end of chunked body"); + assertEquals(stringBuilder.toString(), + "ABCDEFGHIJKLMNOPQRST", + "Altogether, we should read the entire thing"); } - } - private static class PieceMealInputStream extends InputStream { - private final byte[][] parts; + @Test + public void incompleteRequest() throws IOException { + // Use case: After reading 1 chunk successfully, no further chunk size is received and we are + // unable to read anything from the underlying pushback input stream. This should be an + // incomplete request per RFC 9112 section 8. - private int partsIndex; + // arrange + String body = """ + 3\r + Hi \r + """; - private int subPartIndex = 0; + PushbackInputStream pushbackInputStream = new PushbackInputStream(new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), + null); + ChunkedInputStream chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); - public PieceMealInputStream(String... parts) { - this.parts = new byte[parts.length][]; - for (int i = 0; i < parts.length; i++) { - String part = parts[i]; - this.parts[i] = part.getBytes(); - } + // act + byte[] result = chunkedInputStream.readAllBytes(); + + // assert + assertEquals(result.length, + 0); } - @Override - public int read() { - throw new IllegalStateException("Unexpected call to read()"); + @Test + public void partialChunks() throws IOException { + var buf = new byte[1024]; + var inputStream = new ChunkedInputStream(withParts( + """ + A\r + 12345678""", + """ + 90\r + 14\r + 12345678901234567890\r + 1E\r + 123456789012345678901234567890\r + 0\r + \r + """), 1024); + // All chunks will be read on the first attempt because the buffer is large enough + assertEquals(inputStream.read(buf), 60); + var result = new String(buf, 0, 60); + assertEquals(result, "123456789012345678901234567890123456789012345678901234567890"); + assertEquals(inputStream.read(), -1); } - @Override - public int read(byte[] b, int off, int len) { - if (partsIndex >= parts.length) { - return -1; - } - - // We may only read part way through one of the parts. - // If we didn't read all the way through, use the subPartIndex - int read = Math.min(parts[partsIndex].length - subPartIndex, b.length); - System.arraycopy(parts[partsIndex], 0, b, 0, read); - if (read < parts[partsIndex].length - subPartIndex) { - subPartIndex = read; - } else { - partsIndex++; - } - - return read; + @Test + public void partialHeader() throws IOException { + var buf = new byte[1024]; + var inputStream = new ChunkedInputStream(withParts( + """ + A\r + 1234567890\r + 14""", + """ + \r + 12345678901234567890\r + 0\r + \r + """), 1024); + + // All chunks will be read on the first attempt because the buffer is large enough + assertEquals(inputStream.read(buf), 30); + var result = new String(buf, 0, 30); + assertEquals(result, "123456789012345678901234567890"); + assertEquals(inputStream.read(buf), -1); } - @Override - public int read(byte[] b) { - throw new IllegalStateException("Unexpected call to read(byte[] b)"); + @Test + public void trailers() throws Exception { + // It isn't clear if any HTTP server actually users or supports trailers. But, the spec indicates we should at least ignore them. + // - https://www.rfc-editor.org/rfc/rfc2616.html#section-3.6.1 + withBody( + """ + 30\r + There is no fate but what we make for ourselves.\r + 12\r + + - Sarah Connor + \r + 0\r + Judgement-Day: August 29, 1997 2:14 AM EDT\r + \r + """) + .assertResult(""" + There is no fate but what we make for ourselves. + - Sarah Connor + """) + // If we correctly read to the end of the InputStream we should not have any bytes left over in the PushbackInputStream + .assertLeftOverBytes(0); + } + + private Builder withBody(String body) { + return new Builder().withBody(body); + } + + private PushbackInputStream withParts(String... parts) { + return new PushbackInputStream(new PieceMealInputStream(parts), null); + } + + @SuppressWarnings("UnusedReturnValue") + private static class Builder { + public String body; + + public ChunkedInputStream chunkedInputStream; + + public PushbackInputStream pushbackInputStream; + + /** + * Used to ensure the parser worked correctly and was able to read to the end of the encoded body. + * + * @param expected the number of expected bytes that were over-read. + * @return this. + */ + public Builder assertLeftOverBytes(int expected) throws IOException { + int actual = pushbackInputStream.getAvailableBufferedBytesRemaining(); + if (actual != expected) { + if (actual > 0) { + byte[] leftOverBytes = new byte[actual]; + int leftOverRead = pushbackInputStream.read(leftOverBytes); + // No reason to think these would not be equal... but they better be. + assertEquals(leftOverBytes.length, leftOverRead); + assertEquals(actual, expected, "\nHere is what was left over in the buffer\n[" + new String(leftOverBytes) + "]"); + } + } + + return this; + } + + public Builder assertNextRead(ThrowingFunction function, int expected) throws Exception { + var result = function.apply(chunkedInputStream); + assertEquals(result, expected); + return this; + } + + public Builder assertResult(String expected) throws IOException { + pushbackInputStream = new PushbackInputStream(new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), null); + chunkedInputStream = new ChunkedInputStream(pushbackInputStream, 2048); + + String actual = new String(chunkedInputStream.readAllBytes(), StandardCharsets.UTF_8); + assertEquals(actual, expected); + return this; + } + + public Builder withBody(String body) { + this.body = body; + return this; + } + } + + private static class PieceMealInputStream extends InputStream { + private final byte[][] parts; + + private int partsIndex; + + private int subPartIndex = 0; + + public PieceMealInputStream(String... parts) { + this.parts = new byte[parts.length][]; + for (int i = 0; i < parts.length; i++) { + String part = parts[i]; + this.parts[i] = part.getBytes(); + } + } + + @Override + public int read() { + throw new IllegalStateException("Unexpected call to read()"); + } + + @Override + public int read(byte[] b, int off, int len) { + if (partsIndex >= parts.length) { + return -1; + } + if (len == 0) { + return 0; + } + + // We may only read part way through one of the parts. + // If we didn't read all the way through, use the subPartIndex + byte[] part = parts[partsIndex]; + int remainingInPart = part.length - subPartIndex; + // whichever is smaller, what's left in the actual array or what we were asked to read + int toRead = Math.min(remainingInPart, len); + System.arraycopy(part, subPartIndex, b, off, toRead); + subPartIndex += toRead; + + if (subPartIndex >= part.length) { + partsIndex++; + subPartIndex = 0; + } + + return toRead; + } + + @Override + public int read(byte[] b) { + throw new IllegalStateException("Unexpected call to read(byte[] b)"); + } } - } }