fix(sync): read a payload's gzip as exactly one member with no flags

app/src/main/java/io/github/aedev/flow/sync2/crypto/Payloads.kt

@@ -1,14 +1,18 @@
 package io.github.aedev.flow.sync2.crypto
 
 import java.io.ByteArrayOutputStream
-import java.io.IOException
-import java.io.InputStream
 import java.nio.ByteBuffer
-import java.util.zip.GZIPInputStream
+import java.nio.ByteOrder
+import java.util.zip.CRC32
+import java.util.zip.DataFormatException
 import java.util.zip.GZIPOutputStream
+import java.util.zip.Inflater
 
 private const val MIN_SIZE = 256
 private const val MAX_INFLATED = 64 * 1024 * 1024
+private const val GZIP_HEADER = 10
+private const val GZIP_TRAILER = 8
+private val GZIP_START = byteArrayOf(0x1f, 0x8b.toByte(), 8, 0)
 
 /** PADMÉ with a 256-byte floor. */
 fun paddedSize(n: Int): Int {
@@ -30,23 +34,41 @@ fun decodePayload(payload: ByteArray): ByteArray {
     val length = ByteBuffer.wrap(payload, 0, 4).int
     if (length < 0 || length > payload.size - 4) reject(ProtocolError.PAYLOAD)
     if ((4 + length until payload.size).any { payload[it] != 0.toByte() }) reject(ProtocolError.PAYLOAD)
-    return try {
-        inflate(payload.inputStream(4, length))
-    } catch (_: IOException) {
-        reject(ProtocolError.PAYLOAD)
-    }
+    return inflate(payload, 4, length)
 }
 
-private fun inflate(body: InputStream): ByteArray {
-    val out = ByteArrayOutputStream()
-    val buffer = ByteArray(8192)
-    GZIPInputStream(body).use { input ->
-        while (true) {
-            val read = input.read(buffer)
-            if (read < 0) break
+/**
+ * The one gzip member at [start], [length] bytes long (spec/crypto.md, Payload). No library reads gzip this strictly,
+ * so the fixed header and trailer are read here; inflating and the CRC stay in the library.
+ */
+private fun inflate(
+    payload: ByteArray,
+    start: Int,
+    length: Int,
+): ByteArray {
+    if (length < GZIP_HEADER + GZIP_TRAILER) reject(ProtocolError.PAYLOAD)
+    if (GZIP_START.indices.any { payload[start + it] != GZIP_START[it] }) reject(ProtocolError.PAYLOAD)
+    val inflater = Inflater(true)
+    try {
+        // The trailer stays in the input: zlib may want a byte past the stream, and what's left shows where it ended.
+        inflater.setInput(payload, start + GZIP_HEADER, length - GZIP_HEADER)
+        val out = ByteArrayOutputStream()
+        val crc = CRC32()
+        val buffer = ByteArray(8192)
+        while (!inflater.finished()) {
+            val read = inflater.inflate(buffer)
+            if (read == 0 && !inflater.finished()) reject(ProtocolError.PAYLOAD)
+            crc.update(buffer, 0, read)
             out.write(buffer, 0, read)
             if (out.size() > MAX_INFLATED) reject(ProtocolError.PAYLOAD)
         }
+        if (inflater.remaining != GZIP_TRAILER) reject(ProtocolError.PAYLOAD)
+        val trailer = ByteBuffer.wrap(payload, start + length - GZIP_TRAILER, GZIP_TRAILER).order(ByteOrder.LITTLE_ENDIAN)
+        if (trailer.int != crc.value.toInt() || trailer.int != out.size()) reject(ProtocolError.PAYLOAD)
+        return out.toByteArray()
+    } catch (_: DataFormatException) {
+        reject(ProtocolError.PAYLOAD)
+    } finally {
+        inflater.end()
     }
-    return out.toByteArray()
 }

app/src/test/java/io/github/aedev/flow/sync2/PayloadTest.kt

@@ -0,0 +1,74 @@
+package io.github.aedev.flow.sync2
+
+import io.github.aedev.flow.sync2.crypto.ProtocolError
+import io.github.aedev.flow.sync2.crypto.ProtocolException
+import io.github.aedev.flow.sync2.crypto.decodePayload
+import io.github.aedev.flow.sync2.crypto.encodePayload
+import org.junit.Assert.assertArrayEquals
+import org.junit.Assert.assertEquals
+import org.junit.Test
+import java.io.ByteArrayOutputStream
+import java.nio.ByteBuffer
+import java.util.Random
+import java.util.zip.Deflater
+import java.util.zip.DeflaterOutputStream
+
+/** What this device writes reads back, and what it reads stays within its limit (spec/crypto.md, Payload). */
+class PayloadTest {
+    private fun randomBytes(size: Int) = ByteArray(size).also { Random(size.toLong()).nextBytes(it) }
+
+    private val mebibyte = 1024 * 1024
+
+    @Test
+    fun what_the_encoder_writes_reads_back() {
+        for (content in listOf(ByteArray(0), "short".toByteArray(), ByteArray(mebibyte), randomBytes(mebibyte))) {
+            assertArrayEquals(content, decodePayload(encodePayload(content)))
+        }
+    }
+
+    @Test
+    fun the_encoder_writes_one_member_with_no_flags() {
+        val payload = encodePayload("short".toByteArray())
+        assertArrayEquals(byteArrayOf(0x1f, 0x8b.toByte(), 8, 0), payload.copyOfRange(4, 8))
+        val length = ByteBuffer.wrap(payload, 0, 4).int
+        assertEquals("the trailer holds the length", 5, payload[4 + length - 4].toInt())
+    }
+
+    @Test
+    fun content_up_to_the_limit_reads_and_past_it_is_refused() {
+        val limit = 64 * mebibyte
+        assertEquals(limit, decodePayload(encodePayload(ByteArray(limit))).size)
+        val error = runCatching { decodePayload(encodePayload(ByteArray(limit + 1))) }.exceptionOrNull()
+        assertEquals(ProtocolError.PAYLOAD, (error as? ProtocolException)?.error)
+    }
+
+    private fun assertRefused(payload: ByteArray) {
+        val error = runCatching { decodePayload(payload) }.exceptionOrNull()
+        assertEquals("$error", ProtocolError.PAYLOAD, (error as? ProtocolException)?.error)
+    }
+
+    private val header = byteArrayOf(0x1f, 0x8b.toByte(), 8, 0, 0, 0, 0, 0, 0, 0)
+
+    private fun framed(body: ByteArray) = ByteBuffer.allocate(4).putInt(body.size).array() + body
+
+    @Test
+    fun a_body_too_short_for_its_header_and_trailer_is_refused() {
+        val whole = encodePayload("short".toByteArray()).drop(4).toByteArray()
+        for (size in 0 until 18) assertRefused(framed(whole.copyOf(size)))
+    }
+
+    @Test
+    fun a_stream_that_isn_t_deflate_is_refused() {
+        assertRefused(framed(header + ByteArray(8) { -1 } + ByteArray(8)))
+    }
+
+    @Test(timeout = 30_000)
+    fun a_stream_cut_off_before_its_end_is_refused() {
+        val deflated =
+            ByteArrayOutputStream()
+                .also { out ->
+                    DeflaterOutputStream(out, Deflater(9, true)).use { it.write(randomBytes(4096)) }
+                }.toByteArray()
+        assertRefused(framed(header + deflated.copyOf(deflated.size / 2)))
+    }
+}

app/src/test/resources/sync2/README.md

@@ -10,7 +10,7 @@ All byte strings are lowercase hex, except inside roster entries, which use the
 | `derive.json` | `Derive` and `Hash` ([crypto.md](../crypto.md#labelled-derivation)) |
 | `ed25519.json` | Strict verification: valid, tampered, wrong key, non-canonical `S`, small-order key; the eight small-order encodings; a non-canonical key encoding |
 | `box.json` | Box seal and open with fixed salts; tampered ciphertext, tag, commitment and salt, wrong key, wrong AAD, truncation |
-| `payload.json` | PADMÉ sizes; payload decoding; bad length, non-zero padding, corrupt gzip |
+| `payload.json` | PADMÉ sizes; payload decoding, including another encoder's header bytes and a stored block; bad length, non-zero padding, corrupt gzip, more than one member, a bad or short trailer, any header flag, an empty body, bytes around the member |
 | `keys.json` | Member ids, epoch commitment, purpose keys, HPKE wrap with a fixed ephemeral key, wrong-epoch and wrong-member unwraps, history chain, wrap and chain paths |
 | `recovery.json` | Recovery secret, 24 words and derived keys; words with a bad checksum, 23 words, and a valid 12-word phrase |
 | `envelope.json` | Log objects of each type, including a store login, opened in stream order, a live message, a snapshot blob; truncation, magic, suite, group, unknown sender, signature, tampering, replay, gap, a log read from the middle, wrong `prev`, swapped epoch, `seq` above 2<sup>53</sup> − 1, blob hash |