Don't crash processing fragmented web sockets messages (#5983)

* Don't crash processing fragmented web sockets messages

Closes: https://github.com/square/okhttp/issues/5965

* Update okhttp-testing-support/src/main/kotlin/okhttp3/TestUtil.kt

Co-Authored-By: Jake Wharton <jakew@google.com>

* Update okhttp-testing-support/src/main/kotlin/okhttp3/TestUtil.kt

Co-Authored-By: Jake Wharton <jakew@google.com>

* Update okhttp-testing-support/src/main/kotlin/okhttp3/TestUtil.kt

Co-Authored-By: Jake Wharton <jakew@google.com>

Co-authored-by: Jake Wharton <jakew@google.com>

build.gradle

@@ -17,7 +17,7 @@ buildscript {
       'junit': '4.13',
       'kotlin': '1.3.71',
       'moshi': '1.9.2',
-      'okio': '2.5.0',
+      'okio': '2.6.0',
       'ktlint': '0.36.0',
       'picocli': '4.2.0',
       'openjsse': '1.1.0'

okhttp-testing-support/src/main/kotlin/okhttp3/TestUtil.kt

@@ -21,6 +21,7 @@ import java.net.InetSocketAddress
 import java.net.UnknownHostException
 import java.util.Arrays
 import okhttp3.internal.http2.Header
+import okio.Buffer
 import org.junit.Assume.assumeFalse
 import org.junit.Assume.assumeNoException
 
@@ -43,6 +44,31 @@ object TestUtil {
     return String(array)
   }
 
+  /**
+   * Okio buffers are internally implemented as a linked list of arrays. Usually this implementation
+   * detail is invisible to the caller, but subtle use of certain APIs may depend on these internal
+   * structures.
+   *
+   * We make such subtle calls in [okhttp3.internal.ws.MessageInflater] because we try to read a
+   * compressed stream that is terminated in a web socket frame even though the DEFLATE stream is
+   * not terminated.
+   *
+   * Use this method to create a degenerate Okio Buffer where each byte is in a separate segment of
+   * the internal list.
+   */
+  @JvmStatic
+  fun fragmentBuffer(buffer: Buffer): Buffer {
+    // Write each byte into a new buffer, then clone it so that the segments are shared.
+    // Shared segments cannot be compacted so we'll get a long chain of short segments.
+    val result = Buffer()
+    while (!buffer.exhausted()) {
+      val box = Buffer()
+      box.write(buffer, 1)
+      result.write(box.copy(), 1)
+    }
+    return result
+  }
+
   tailrec fun File.isDescendentOf(directory: File): Boolean {
     val parentFile = parentFile ?: return false
     if (parentFile == directory) return true

okhttp/src/main/kotlin/okhttp3/internal/ws/MessageInflater.kt

@@ -35,14 +35,6 @@ class MessageInflater(
   fun inflate(buffer: Buffer) {
     require(deflatedBytes.size == 0L)
 
-    // Handle the empty message special case. The compressed empty message is one byte, '0x00'. We
-    // can't use the normal flow here because inflaterSource.read() throws EOFException if the
-    // deflated stream isn't complete but there's no bytes to return.
-    if (buffer.size == 1L && buffer[0L] == 0.toByte()) {
-      buffer.skip(1L)
-      return
-    }
-
     if (noContextTakeover) {
       inflater.reset()
     }
@@ -55,7 +47,7 @@ class MessageInflater(
     // We cannot read all, as the source does not close.
     // Instead, we ensure that all bytes from source have been processed by inflater.
     do {
-      inflaterSource.read(buffer, Long.MAX_VALUE)
+      inflaterSource.readOrInflate(buffer, Long.MAX_VALUE)
     } while (inflater.bytesRead < totalBytesToRead)
   }
 

okhttp/src/test/java/okhttp3/internal/ws/MessageDeflaterInflaterTest.kt

@@ -16,6 +16,7 @@
 package okhttp3.internal.ws
 
 import java.io.EOFException
+import okhttp3.TestUtil.fragmentBuffer
 import okio.Buffer
 import okio.ByteString
 import okio.ByteString.Companion.decodeHex
@@ -116,15 +117,15 @@ internal class MessageDeflaterInflaterTest {
     }
   }
 
-  @Test fun `inflate empty buffer`() {
+  /**
+   * Test for an [EOFException] caused by mishandling of fragmented buffers in web socket
+   * compression. https://github.com/square/okhttp/issues/5965
+   */
+  @Test fun `inflate golden value in buffer that has been fragmented`() {
     val inflater = MessageInflater(false)
-
+    val buffer = fragmentBuffer(Buffer().write("f248cdc9c957c8cc4bcb492cc9cccf530400".decodeHex()))
-    try {
+    inflater.inflate(buffer)
-      inflater.inflate(Buffer())
+    assertThat(buffer.readUtf8()).isEqualTo("Hello inflation!")
-      fail()
-    } catch (e: EOFException) {
-      assertThat(e.message).isEqualTo("source exhausted prematurely")
-    }
   }
 
   private fun MessageDeflater.deflate(byteString: ByteString): ByteString {