=htp #16063 fail on data truncation for GZIP coding

akka-http-tests/src/test/scala/akka/http/coding/DecoderSpec.scala

@@ -35,5 +35,6 @@ class DecoderSpec extends WordSpec with CodecSpecSupport {
 
   case object DummyDecompressor extends Decompressor {
     def decompress(buffer: ByteString): ByteString = buffer ++ ByteString("compressed")
+    def finish(): ByteString = ByteString.empty
   }
 }

akka-http-tests/src/test/scala/akka/http/coding/GzipSpec.scala

@@ -51,6 +51,10 @@ class GzipSpec extends WordSpec with CodecSpecSupport {
       val ex = the[DataFormatException] thrownBy ourGunzip(corruptGzipContent)
       ex.getMessage should equal("invalid literal/length code")
     }
+    "throw an error on truncated input" in {
+      val ex = the[ZipException] thrownBy ourGunzip(streamGzip(smallTextBytes).dropRight(5))
+      ex.getMessage should equal("Truncated GZIP stream")
+    }
     "throw early if header is corrupt" in {
       val ex = the[ZipException] thrownBy ourGunzip(ByteString(0, 1, 2, 3, 4))
       ex.getMessage should equal("Not in GZIP format")
@@ -100,7 +104,7 @@ class GzipSpec extends WordSpec with CodecSpecSupport {
 
   def gzip(s: String) = ourGzip(ByteString(s, "UTF8"))
   def ourGzip(bytes: ByteString): ByteString = Gzip.newCompressor.compressAndFinish(bytes)
-  def ourGunzip(bytes: ByteString): ByteString = Gzip.newDecompressor.decompress(bytes)
+  def ourGunzip(bytes: ByteString): ByteString = Gzip.newDecompressor.decompressAndFinish(bytes)
 
   lazy val corruptGzipContent = {
     val content = gzip("Hello").toArray

akka-http-tests/src/test/scala/akka/http/server/directives/CodingDirectivesSpec.scala

@@ -67,6 +67,14 @@ class CodingDirectivesSpec extends RoutingSpec {
         responseAs[String] shouldEqual "The request's encoding is corrupt:\nNot in GZIP format"
       }
     }
+    "reject truncated gzip request content" in {
+      Post("/", helloGzipped.dropRight(2)) ~> `Content-Encoding`(gzip) ~> {
+        decodeRequest(Gzip) { echoRequestContent }
+      } ~> check {
+        status shouldEqual BadRequest
+        responseAs[String] shouldEqual "The request's encoding is corrupt:\nTruncated GZIP stream"
+      }
+    }
     "reject requests with content encoded with 'deflate'" in {
       Post("/", "Hello") ~> `Content-Encoding`(deflate) ~> {
         decodeRequest(Gzip) { completeOk }

akka-http/src/main/scala/akka/http/coding/Decoder.scala

@@ -27,7 +27,7 @@ trait Decoder {
     val decompressor = newDecompressor
 
     def decodeChunk(bytes: ByteString): ByteString = decompressor.decompress(bytes)
-    def finish(): ByteString = ByteString.empty
+    def finish(): ByteString = decompressor.finish()
 
     StreamUtils.byteStringTransformer(decodeChunk, finish)
   }
@@ -36,5 +36,11 @@ trait Decoder {
 /** A stateful object representing ongoing decompression. */
 abstract class Decompressor {
   /** Decompress the buffer and return decompressed data. */
-  def decompress(buffer: ByteString): ByteString
+  def decompress(input: ByteString): ByteString
+
+  /** Flushes potential remaining data from any internal buffers and may report on truncation errors */
+  def finish(): ByteString
+
+  /** Combines decompress and finish */
+  def decompressAndFinish(input: ByteString): ByteString = decompress(input) ++ finish()
 }

akka-http/src/main/scala/akka/http/coding/Deflate.scala

@@ -91,10 +91,10 @@ class DeflateCompressor extends Compressor {
 class DeflateDecompressor extends Decompressor {
   protected lazy val inflater = new Inflater()
 
-  def decompress(buffer: ByteString): ByteString =
+  def decompress(input: ByteString): ByteString =
     try {
-      inflater.setInput(buffer.toArray)
-      drain(new Array[Byte](buffer.length * 2))
+      inflater.setInput(input.toArray)
+      drain(new Array[Byte](input.length * 2))
     } catch {
       case e: DataFormatException ⇒
         throw new ZipException(e.getMessage.toOption getOrElse "Invalid ZLIB data format")
@@ -106,4 +106,9 @@ class DeflateDecompressor extends Decompressor {
     else if (inflater.needsDictionary) throw new ZipException("ZLIB dictionary missing")
     else result
   }
+
+  def finish(): ByteString = {
+    inflater.end()
+    ByteString.empty
+  }
 }

akka-http/src/main/scala/akka/http/coding/Gzip.scala

@@ -63,10 +63,16 @@ class GzipCompressor extends DeflateCompressor {
 class GzipDecompressor extends DeflateDecompressor {
   override protected lazy val inflater = new Inflater(true) // disable ZLIB headers
   override def decompress(input: ByteString): ByteString = DecompressionStateMachine.run(input)
+  override def finish(): ByteString =
+    if (DecompressionStateMachine.isFinished) ByteString.empty
+    else fail("Truncated GZIP stream")
 
   import GzipDecompressor._
+
   object DecompressionStateMachine extends StateMachine {
-    def initialState = readHeaders
+    def isFinished: Boolean = currentState == finished
+
+    def initialState = finished
 
     private def readHeaders(data: ByteString): Action =
       try {
@@ -106,12 +112,13 @@ class GzipDecompressor extends DeflateDecompressor {
 
         inflater.reset()
         checkSum.reset()
-        ContinueWith(initialState, remainingData) // start over to support multiple concatenated gzip streams
+        ContinueWith(finished, remainingData) // start over to support multiple concatenated gzip streams
       } catch {
         case ByteReader.NeedMoreData ⇒ SuspendAndRetryWithMoreData
       }
 
-    private def fail(msg: String) = throw new ZipException(msg)
+    lazy val finished: ByteString ⇒ Action =
+      data ⇒ if (data.nonEmpty) ContinueWith(readHeaders, data) else SuspendAndRetryWithMoreData
 
     private def crc16(data: ByteString) = {
       val crc = new CRC32
@@ -119,6 +126,8 @@ class GzipDecompressor extends DeflateDecompressor {
       crc.getValue.toInt & 0xFFFF
     }
   }
+
+  private def fail(msg: String) = throw new ZipException(msg)
 }
 
 /** INTERNAL API */
@@ -176,6 +185,7 @@ private[http] object GzipDecompressor {
     def initialState: State
 
     private[this] var state: State = initialState
+    def currentState: State = state
 
     /** Run the state machine with the current input */
     final def run(input: ByteString): ByteString = {
@@ -191,7 +201,8 @@ private[http] object GzipDecompressor {
           case EmitAndSuspend(output) ⇒ result ++ output
           case ContinueWith(next, remainingInput) ⇒
             state = next
-            rec(remainingInput, result)
+            if (remainingInput.nonEmpty) rec(remainingInput, result)
+            else result
           case EmitAndContinueWith(output, next, remainingInput) ⇒
             state = next
             rec(remainingInput, result ++ output)

akka-http/src/main/scala/akka/http/coding/NoCoding.scala

@@ -35,4 +35,5 @@ object NoCodingCompressor extends Compressor {
 }
 object NoCodingDecompressor extends Decompressor {
   def decompress(input: ByteString): ByteString = input
+  def finish(): ByteString = ByteString.empty
 }