fix(sync2): bound every response, live message and LAN frame read

.github/nightly-notes/2026-11-02-fix-sync2-response-limits.md

@@ -0,0 +1,2 @@
+## Fixed
+- Linked devices stop reading an answer from a sync server, WebDAV folder or nearby device that is larger than the protocol allows

app/src/main/java/io/github/aedev/flow/sync/transport/BoundedListener.kt

@@ -0,0 +1,49 @@
+package io.github.aedev.flow.sync.transport
+
+import okhttp3.Response
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import okio.ByteString
+import okio.utf8Size
+
+/**
+ * Hands [delegate] only messages of at most [limit] bytes. OkHttp reads a whole message before it calls back and has
+ * no limit of its own, so a bigger one fails the socket instead of being read on.
+ */
+class BoundedListener(
+    private val limit: Long,
+    private val delegate: WebSocketListener,
+) : WebSocketListener() {
+    override fun onOpen(
+        webSocket: WebSocket,
+        response: Response,
+    ) = delegate.onOpen(webSocket, response)
+
+    override fun onMessage(
+        webSocket: WebSocket,
+        bytes: ByteString,
+    ) = if (bytes.size > limit) webSocket.cancel() else delegate.onMessage(webSocket, bytes)
+
+    override fun onMessage(
+        webSocket: WebSocket,
+        text: String,
+    ) = if (text.utf8Size() > limit) webSocket.cancel() else delegate.onMessage(webSocket, text)
+
+    override fun onClosing(
+        webSocket: WebSocket,
+        code: Int,
+        reason: String,
+    ) = delegate.onClosing(webSocket, code, reason)
+
+    override fun onClosed(
+        webSocket: WebSocket,
+        code: Int,
+        reason: String,
+    ) = delegate.onClosed(webSocket, code, reason)
+
+    override fun onFailure(
+        webSocket: WebSocket,
+        t: Throwable,
+        response: Response?,
+    ) = delegate.onFailure(webSocket, t, response)
+}

app/src/main/java/io/github/aedev/flow/sync/transport/WsClient.kt

@@ -17,10 +17,11 @@ import kotlin.coroutines.resumeWithException
 /**
  * Scan/join role transport: an OkHttp WebSocket **client**. The injected
  * [OkHttpClient] is rebuilt with no call/read timeout (a long-lived WS would otherwise be killed
- * by the app's 60 s `callTimeout`) plus a keepalive ping.
+ * by the app's 60 s `callTimeout`) plus a keepalive ping. A message over [maxMessage] bytes fails the connection.
  */
 class WsClient(
     private val baseClient: OkHttpClient,
+    private val maxMessage: Long = Long.MAX_VALUE,
 ) {
     suspend fun connect(url: String): SyncConnection =
         suspendCancellableCoroutine { cont ->
@@ -39,46 +40,49 @@ class WsClient(
             val ws =
                 client.newWebSocket(
                     request,
-                    object : WebSocketListener() {
-                        override fun onOpen(
-                            webSocket: WebSocket,
-                            response: Response,
-                        ) {
-                            if (cont.isActive) cont.resume(WsConnection(webSocket, incoming))
-                        }
+                    BoundedListener(
+                        maxMessage,
+                        object : WebSocketListener() {
+                            override fun onOpen(
+                                webSocket: WebSocket,
+                                response: Response,
+                            ) {
+                                if (cont.isActive) cont.resume(WsConnection(webSocket, incoming))
+                            }
 
-                        override fun onMessage(
-                            webSocket: WebSocket,
-                            bytes: ByteString,
-                        ) {
-                            incoming.trySend(bytes.toByteArray())
-                        }
+                            override fun onMessage(
+                                webSocket: WebSocket,
+                                bytes: ByteString,
+                            ) {
+                                incoming.trySend(bytes.toByteArray())
+                            }
 
-                        override fun onClosing(
-                            webSocket: WebSocket,
-                            code: Int,
-                            reason: String,
-                        ) {
-                            incoming.close()
-                        }
+                            override fun onClosing(
+                                webSocket: WebSocket,
+                                code: Int,
+                                reason: String,
+                            ) {
+                                incoming.close()
+                            }
 
-                        override fun onClosed(
-                            webSocket: WebSocket,
-                            code: Int,
-                            reason: String,
-                        ) {
-                            incoming.close()
-                        }
+                            override fun onClosed(
+                                webSocket: WebSocket,
+                                code: Int,
+                                reason: String,
+                            ) {
+                                incoming.close()
+                            }
 
-                        override fun onFailure(
-                            webSocket: WebSocket,
-                            t: Throwable,
-                            response: Response?,
-                        ) {
-                            incoming.close(t)
-                            if (cont.isActive) cont.resumeWithException(t)
-                        }
-                    },
+                            override fun onFailure(
+                                webSocket: WebSocket,
+                                t: Throwable,
+                                response: Response?,
+                            ) {
+                                incoming.close(t)
+                                if (cont.isActive) cont.resumeWithException(t)
+                            }
+                        },
+                    ),
                 )
             cont.invokeOnCancellation { runCatching { ws.cancel() } }
         }

app/src/main/java/io/github/aedev/flow/sync/transport/WsServer.kt

@@ -20,9 +20,11 @@ import kotlinx.coroutines.launch
  * (OS-assigned ephemeral port, read back for the QR). CIO is pure-Kotlin/coroutine so it runs on
  * Android. v1 handles a single 1↔1 session per server instance. [path] is FLOW-SYNC/1's unless
  * FLOW-SYNC/2 serves on its own; with [onConnection], every connection goes to it instead of [awaitConnection].
+ * A frame over [maxFrame] bytes closes the connection.
  */
 class WsServer(
     private val path: String = QrPath.PATH,
+    private val maxFrame: Long = Long.MAX_VALUE,
     private val onConnection: (suspend (SyncConnection) -> Unit)? = null,
 ) {
     private var server: EmbeddedServer<*, *>? = null
@@ -37,6 +39,7 @@ class WsServer(
                     // vanished without closing is only noticed when our pings go unanswered.
                     pingPeriodMillis = PING_PERIOD_MS
                     timeoutMillis = PING_TIMEOUT_MS
+                    maxFrameSize = maxFrame
                 }
                 routing {
                     webSocket(path) {

app/src/main/java/io/github/aedev/flow/sync2/lan/LanLink.kt

@@ -37,6 +37,12 @@ import java.nio.ByteBuffer
 
 private const val NONCE_SIZE = 32
 
+/** The most a LAN frame's payload may hold (spec/live.md, Session). */
+const val MAX_LAN_PAYLOAD = 4 * 1024 * 1024
+
+/** Room for a payload's gzip, padding (at most 1/32 at this size), box and `seq`. */
+const val MAX_LAN_FRAME = MAX_LAN_PAYLOAD + MAX_LAN_PAYLOAD / 16L
+
 /**
  * A LAN session between two members (spec/live.md, Session), once the hello has checked both: each
  * frame is `u64(seq) ‖ Box(K, "FLOW-SYNC/2 lan" ‖ u64(seq), payload)`, counted strictly in each direction.
@@ -65,6 +71,7 @@ class LanLink internal constructor(
         val frame = connection.receive() ?: return null
         if (frame.size < 8 || ByteBuffer.wrap(frame, 0, 8).long != received) reject(ProtocolError.REPLAY)
         val content = decodePayload(open(receiveKey, lanFrameAad(received), frame.copyOfRange(8, frame.size)))
+        if (content.size > MAX_LAN_PAYLOAD) reject(ProtocolError.LENGTH)
         received++
         return content
     }

app/src/main/java/io/github/aedev/flow/sync2/lan/LanSync.kt

@@ -211,7 +211,7 @@ class LanSync
             coroutineScope {
                 val session = open()
                 val lanKey = purposeKey(session.currentKey(), session.group, session.state.epoch, Purpose.LAN)
-                val server = WsServer(SESSION_PATH) { connection -> accept(connection) }
+                val server = WsServer(SESSION_PATH, MAX_LAN_FRAME) { connection -> accept(connection) }
                 val discovery = LanDiscovery(context)
                 val name = "Flow-" + SyncBytes.toHex(SyncCrypto.randomBytes(2))
                 names += name
@@ -296,7 +296,7 @@ class LanSync
 
         private suspend fun connect(address: String): SyncConnection? =
             withTimeoutOrNull(CONNECT_TIMEOUT_MS) {
-                runCatching { WsClient(OkHttpClient()).connect("ws://$address$SESSION_PATH") }.getOrNull()
+                runCatching { WsClient(OkHttpClient(), MAX_LAN_FRAME).connect("ws://$address$SESSION_PATH") }.getOrNull()
             }
 
         /** A member that dialed this one is remembered where its hello says it listens. */

app/src/main/java/io/github/aedev/flow/sync2/linking/Pipes.kt

@@ -8,6 +8,7 @@ import io.github.aedev.flow.sync.transport.WsServer
 import io.github.aedev.flow.sync2.data.LanAddressRow
 import io.github.aedev.flow.sync2.data.LinkedDatabase
 import io.github.aedev.flow.sync2.engine.LinkedException
+import io.github.aedev.flow.sync2.lan.MAX_LAN_FRAME
 import io.github.aedev.flow.sync2.store.Mailbox
 import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.delay
@@ -160,14 +161,14 @@ suspend fun listen(): Listening? {
             .filter(LocalNetworkAddress::isLocal)
             .take(LAN_ADDRESSES)
     if (addresses.isEmpty()) return null
-    val server = WsServer(PAIR_PATH)
+    val server = WsServer(PAIR_PATH, MAX_LAN_FRAME)
     val port = server.start()
     return Listening(server, addresses.map { LanEndpoint(it, port) })
 }
 
 /** Dials the addresses a code lists, in order, each for a few seconds. */
 internal suspend fun dial(addresses: List<LanEndpoint>): LanPipe? {
-    val client = WsClient(OkHttpClient())
+    val client = WsClient(OkHttpClient(), MAX_LAN_FRAME)
     for (address in addresses.filter { LocalNetworkAddress.isLocal(it.a) }) {
         val host = LocalNetworkAddress.urlHost(address.a)
         withTimeoutOrNull(CONNECT_TIMEOUT_MS) { runCatching { client.connect("ws://$host:${address.p}$PAIR_PATH") }.getOrNull() }

app/src/main/java/io/github/aedev/flow/sync2/store/ServerStore.kt

@@ -2,6 +2,7 @@ package io.github.aedev.flow.sync2.store
 
 import io.github.aedev.flow.sync.crypto.SyncBytes
 import io.github.aedev.flow.sync.qr.LocalNetworkAddress
+import io.github.aedev.flow.sync.transport.BoundedListener
 import io.github.aedev.flow.sync2.crypto.MemberSecrets
 import io.github.aedev.flow.sync2.crypto.sign
 import io.github.aedev.flow.sync2.crypto.tag
@@ -45,6 +46,9 @@ import java.util.concurrent.TimeUnit
 /** Refresh a token this long before it expires. */
 private const val TOKEN_MARGIN_MS = 60_000L
 private const val LIVE_PING_S = 15L
+
+/** The most a live message or a server notice may hold (spec/store.md, Untrusted input). */
+private const val MAX_LIVE_MESSAGE = 64L * 1024
 private val JSON = "application/json".toMediaType()
 private val BYTES = "application/octet-stream".toMediaType()
 
@@ -85,12 +89,18 @@ class ServerStore private constructor(
 
     private fun url(path: String): HttpUrl = base.resolve(path) ?: throw StoreException.Malformed()
 
-    /** The whole body is read here, on the IO dispatcher, so callers never read the network from their own thread. */
-    private suspend fun send(request: Request): Response =
+    /**
+     * The whole body, up to [limit] bytes, is read here, on the IO dispatcher, so callers never read the network from
+     * their own thread.
+     */
+    private suspend fun send(
+        request: Request,
+        limit: Long = MAX_RESPONSE,
+    ): Response =
         withContext(Dispatchers.IO) {
             try {
                 client.newCall(request).execute().use { response ->
-                    response.newBuilder().body(response.body.bytes().toResponseBody(response.body.contentType())).build()
+                    response.newBuilder().body(response.body.bytesUpTo(limit).toResponseBody(response.body.contentType())).build()
                 }
             } catch (e: IOException) {
                 throw StoreException.Network(e)
@@ -158,6 +168,7 @@ class ServerStore private constructor(
         path: String,
         body: RequestBody? = null,
         headers: Map<String, String> = emptyMap(),
+        limit: Long = MAX_RESPONSE,
     ): Response {
         for (renew in listOf(false, true)) {
             val request =
@@ -168,15 +179,15 @@ class ServerStore private constructor(
                     .header("Authorization", "Bearer ${token(renew)}")
                     .apply { headers.forEach { (name, value) -> header(name, value) } }
                     .build()
-            val response = send(request)
+            val response = send(request, limit)
             if (response.code != 401 || renew) return response
             response.close()
         }
         error("the loop returns on its second pass")
     }
 
     private suspend fun getBytes(path: String): ByteArray? =
-        authorized("GET", path).use { response ->
+        authorized("GET", path, limit = MAX_OBJECT_SIZE.toLong()).use { response ->
             if (response.code == 404) null else check(response).body.bytes()
         }
 
@@ -270,7 +281,7 @@ class ServerStore private constructor(
             .readTimeout(0, TimeUnit.SECONDS)
             .callTimeout(0, TimeUnit.SECONDS)
             .build()
-            .newWebSocket(request, listener)
+            .newWebSocket(request, BoundedListener(MAX_LIVE_MESSAGE, listener))
     }
 
     /** Join requests no member has answered yet. */

app/src/main/java/io/github/aedev/flow/sync2/store/Transport.kt

@@ -4,9 +4,20 @@ import io.github.aedev.flow.sync.qr.LocalNetworkAddress
 import okhttp3.Dns
 import okhttp3.HttpUrl
 import okhttp3.OkHttpClient
+import okhttp3.ResponseBody
 import java.net.UnknownHostException
 import java.util.concurrent.TimeUnit
 
+/** The most a response other than an object or a blob may hold (spec/store.md, Untrusted input). */
+internal const val MAX_RESPONSE = 16L * 1024 * 1024
+
+/** Reads a body of at most [limit] bytes, failing as soon as more arrive. */
+internal fun ResponseBody.bytesUpTo(limit: Long): ByteArray {
+    val source = source()
+    if (source.request(limit + 1)) throw StoreException.Malformed()
+    return source.readByteArray()
+}
+
 /** A `.local` name may resolve only to local addresses, checked on every connection. */
 private val LOCAL_ONLY_DNS =
     Dns { host ->

app/src/main/java/io/github/aedev/flow/sync2/store/WebDavStore.kt

@@ -71,6 +71,7 @@ class WebDavStore private constructor(
         path: String,
         body: RequestBody? = null,
         headers: Map<String, String> = emptyMap(),
+        limit: Long = MAX_RESPONSE,
     ): Pair<Int, ByteArray> {
         if (account in refused) throw StoreException.LoginRefused()
         val request =
@@ -84,7 +85,7 @@ class WebDavStore private constructor(
         val (code, bytes) =
             withContext(Dispatchers.IO) {
                 try {
-                    client.newCall(request).execute().use { it.code to it.body.bytes() }
+                    client.newCall(request).execute().use { it.code to it.body.bytesUpTo(limit) }
                 } catch (e: IOException) {
                     throw StoreException.Network(e)
                 }
@@ -99,7 +100,7 @@ class WebDavStore private constructor(
     private fun refusal(code: Int): Nothing = throw StoreException.Refused(code, "webdav")
 
     override suspend fun get(path: String): ByteArray? {
-        val (code, bytes) = send("GET", path)
+        val (code, bytes) = send("GET", path, limit = MAX_OBJECT_SIZE.toLong())
         return when (code) {
             in 200..299 -> bytes
             404 -> null

app/src/test/java/io/github/aedev/flow/sync/transport/BoundedListenerTest.kt

@@ -0,0 +1,39 @@
+package io.github.aedev.flow.sync.transport
+
+import io.mockk.mockk
+import io.mockk.unmockkAll
+import io.mockk.verify
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import okio.ByteString.Companion.toByteString
+import org.junit.After
+import org.junit.Test
+
+class BoundedListenerTest {
+    private val socket = mockk<WebSocket>(relaxed = true)
+    private val delegate = mockk<WebSocketListener>(relaxed = true)
+    private val listener = BoundedListener(4, delegate)
+
+    @After
+    fun tearDown() {
+        unmockkAll()
+    }
+
+    @Test
+    fun `messages within the limit reach the listener`() {
+        listener.onMessage(socket, ByteArray(4).toByteString())
+        listener.onMessage(socket, "abcd")
+        verify(exactly = 1) { delegate.onMessage(socket, ByteArray(4).toByteString()) }
+        verify(exactly = 1) { delegate.onMessage(socket, "abcd") }
+        verify(exactly = 0) { socket.cancel() }
+    }
+
+    @Test
+    fun `a message past the limit fails the socket and never reaches the listener`() {
+        listener.onMessage(socket, ByteArray(5).toByteString())
+        listener.onMessage(socket, "abcé")
+        verify(exactly = 2) { socket.cancel() }
+        verify(exactly = 0) { delegate.onMessage(socket, any<okio.ByteString>()) }
+        verify(exactly = 0) { delegate.onMessage(socket, any<String>()) }
+    }
+}