From e5e250db853c097ae7dae00d615263cb23248b14 Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:54:04 +0200 Subject: [PATCH] fix(sync2): bound every response, live message and LAN frame read --- .../2026-11-02-fix-sync2-response-limits.md | 2 + .../flow/sync/transport/BoundedListener.kt | 49 ++++++++++++ .../aedev/flow/sync/transport/WsClient.kt | 78 ++++++++++--------- .../aedev/flow/sync/transport/WsServer.kt | 3 + .../io/github/aedev/flow/sync2/lan/LanLink.kt | 7 ++ .../io/github/aedev/flow/sync2/lan/LanSync.kt | 4 +- .../github/aedev/flow/sync2/linking/Pipes.kt | 5 +- .../aedev/flow/sync2/store/ServerStore.kt | 23 ++++-- .../aedev/flow/sync2/store/Transport.kt | 11 +++ .../aedev/flow/sync2/store/WebDavStore.kt | 5 +- .../sync/transport/BoundedListenerTest.kt | 39 ++++++++++ .../aedev/flow/sync/transport/WsClientTest.kt | 29 +++++++ .../aedev/flow/sync/transport/WsServerTest.kt | 29 +++++++ .../io/github/aedev/flow/sync2/LanLinkTest.kt | 41 ++++++++++ .../aedev/flow/sync2/ServerStoreTest.kt | 17 ++++ .../aedev/flow/sync2/WebDavStoreTest.kt | 21 +++++ 16 files changed, 314 insertions(+), 49 deletions(-) create mode 100644 .github/nightly-notes/2026-11-02-fix-sync2-response-limits.md create mode 100644 app/src/main/java/io/github/aedev/flow/sync/transport/BoundedListener.kt create mode 100644 app/src/test/java/io/github/aedev/flow/sync/transport/BoundedListenerTest.kt create mode 100644 app/src/test/java/io/github/aedev/flow/sync/transport/WsClientTest.kt create mode 100644 app/src/test/java/io/github/aedev/flow/sync/transport/WsServerTest.kt create mode 100644 app/src/test/java/io/github/aedev/flow/sync2/LanLinkTest.kt diff --git a/.github/nightly-notes/2026-11-02-fix-sync2-response-limits.md b/.github/nightly-notes/2026-11-02-fix-sync2-response-limits.md new file mode 100644 index 0000000000..8de37f1ae7 --- /dev/null +++ b/.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 diff --git a/app/src/main/java/io/github/aedev/flow/sync/transport/BoundedListener.kt b/app/src/main/java/io/github/aedev/flow/sync/transport/BoundedListener.kt new file mode 100644 index 0000000000..1cd599db31 --- /dev/null +++ b/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) +} diff --git a/app/src/main/java/io/github/aedev/flow/sync/transport/WsClient.kt b/app/src/main/java/io/github/aedev/flow/sync/transport/WsClient.kt index b48ba9d306..e28c06c243 100644 --- a/app/src/main/java/io/github/aedev/flow/sync/transport/WsClient.kt +++ b/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() } } } diff --git a/app/src/main/java/io/github/aedev/flow/sync/transport/WsServer.kt b/app/src/main/java/io/github/aedev/flow/sync/transport/WsServer.kt index 2d59b80935..1693caeea7 100644 --- a/app/src/main/java/io/github/aedev/flow/sync/transport/WsServer.kt +++ b/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) { diff --git a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanLink.kt b/app/src/main/java/io/github/aedev/flow/sync2/lan/LanLink.kt index 3aefeda053..a6ef09f0a9 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanLink.kt +++ b/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 } diff --git a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanSync.kt b/app/src/main/java/io/github/aedev/flow/sync2/lan/LanSync.kt index e1b33c31cc..ce8b4ad378 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanSync.kt +++ b/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. */ diff --git a/app/src/main/java/io/github/aedev/flow/sync2/linking/Pipes.kt b/app/src/main/java/io/github/aedev/flow/sync2/linking/Pipes.kt index f278992684..3890b78af5 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/linking/Pipes.kt +++ b/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): 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() } diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/ServerStore.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/ServerStore.kt index 4e20aea278..a2f26fdee8 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/ServerStore.kt +++ b/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 = emptyMap(), + limit: Long = MAX_RESPONSE, ): Response { for (renew in listOf(false, true)) { val request = @@ -168,7 +179,7 @@ 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() } @@ -176,7 +187,7 @@ class ServerStore private constructor( } 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. */ diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/Transport.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/Transport.kt index 2996d18b7b..cce59de09b 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/Transport.kt +++ b/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 -> diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/WebDavStore.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/WebDavStore.kt index ea5eaa06ed..10f0f2acd0 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/WebDavStore.kt +++ b/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 = emptyMap(), + limit: Long = MAX_RESPONSE, ): Pair { 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 diff --git a/app/src/test/java/io/github/aedev/flow/sync/transport/BoundedListenerTest.kt b/app/src/test/java/io/github/aedev/flow/sync/transport/BoundedListenerTest.kt new file mode 100644 index 0000000000..09480ba3b4 --- /dev/null +++ b/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(relaxed = true) + private val delegate = mockk(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()) } + verify(exactly = 0) { delegate.onMessage(socket, any()) } + } +} diff --git a/app/src/test/java/io/github/aedev/flow/sync/transport/WsClientTest.kt b/app/src/test/java/io/github/aedev/flow/sync/transport/WsClientTest.kt new file mode 100644 index 0000000000..6362573b3f --- /dev/null +++ b/app/src/test/java/io/github/aedev/flow/sync/transport/WsClientTest.kt @@ -0,0 +1,29 @@ +package io.github.aedev.flow.sync.transport + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.runBlocking +import okhttp3.OkHttpClient +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertNull +import org.junit.Test + +class WsClientTest { + @Test + fun `the client closes on a message past its limit`() = + runBlocking { + val accepted = CompletableDeferred() + val server = WsServer("/limit") { accepted.complete(it) } + try { + val port = server.start() + val client = WsClient(OkHttpClient(), maxMessage = 1024).connect("ws://127.0.0.1:$port/limit") + val connection = accepted.await() + connection.send(ByteArray(1024) { 2 }) + assertArrayEquals(ByteArray(1024) { 2 }, client.receive()) + connection.send(ByteArray(1025)) + assertNull(client.receive()) + connection.close() + } finally { + server.stop() + } + } +} diff --git a/app/src/test/java/io/github/aedev/flow/sync/transport/WsServerTest.kt b/app/src/test/java/io/github/aedev/flow/sync/transport/WsServerTest.kt new file mode 100644 index 0000000000..e8242835e8 --- /dev/null +++ b/app/src/test/java/io/github/aedev/flow/sync/transport/WsServerTest.kt @@ -0,0 +1,29 @@ +package io.github.aedev.flow.sync.transport + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.runBlocking +import okhttp3.OkHttpClient +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertNull +import org.junit.Test + +class WsServerTest { + @Test + fun `a frame past the limit closes the connection`() = + runBlocking { + val accepted = CompletableDeferred() + val server = WsServer("/limit", maxFrame = 1024) { accepted.complete(it) } + try { + val port = server.start() + val client = WsClient(OkHttpClient()).connect("ws://127.0.0.1:$port/limit") + client.send(ByteArray(1024) { 1 }) + val connection = accepted.await() + assertArrayEquals(ByteArray(1024) { 1 }, connection.receive()) + client.send(ByteArray(1025)) + assertNull(connection.receive()) + client.close() + } finally { + server.stop() + } + } +} diff --git a/app/src/test/java/io/github/aedev/flow/sync2/LanLinkTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/LanLinkTest.kt new file mode 100644 index 0000000000..139a63fc00 --- /dev/null +++ b/app/src/test/java/io/github/aedev/flow/sync2/LanLinkTest.kt @@ -0,0 +1,41 @@ +package io.github.aedev.flow.sync2 + +import io.github.aedev.flow.sync.transport.SyncConnection +import io.github.aedev.flow.sync2.crypto.ProtocolException +import io.github.aedev.flow.sync2.crypto.encodePayload +import io.github.aedev.flow.sync2.crypto.lanFrameAad +import io.github.aedev.flow.sync2.crypto.seal +import io.github.aedev.flow.sync2.crypto.u64 +import io.github.aedev.flow.sync2.lan.LanLink +import io.github.aedev.flow.sync2.lan.MAX_LAN_PAYLOAD +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Test + +class LanLinkTest { + private val key = ByteArray(32) { 1 } + + /** A link whose other side sent one frame carrying [size] bytes of content. */ + private fun receiving(size: Int) = + LanLink( + object : SyncConnection { + override suspend fun send(bytes: ByteArray) = Unit + + override suspend fun receive() = u64(0) + seal(key, lanFrameAad(0), encodePayload(ByteArray(size))) + + override fun close() = Unit + }, + peer = "peer", + peerPort = null, + sendKey = key, + receiveKey = key, + ) + + @Test + fun `a frame whose payload inflates past the limit is refused`() = + runTest { + assertEquals(MAX_LAN_PAYLOAD, receiving(MAX_LAN_PAYLOAD).receive()?.size) + val error = runCatching { receiving(MAX_LAN_PAYLOAD + 1).receive() }.exceptionOrNull() + assertEquals("length", (error as ProtocolException).error.code) + } +} diff --git a/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt index 5f74f0fd54..f4ba67efa1 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt @@ -16,6 +16,7 @@ import io.ktor.server.request.path import io.ktor.server.request.receiveText import io.ktor.server.response.header import io.ktor.server.response.respond +import io.ktor.server.response.respondBytes import kotlinx.coroutines.test.runTest import kotlinx.serialization.json.Json import kotlinx.serialization.json.jsonObject @@ -141,6 +142,22 @@ class ServerStoreTest { assertTrue("$error", error is LinkedException.ServerHasGroup) } + @Test + fun answers_past_their_limit_are_refused() = + runTest { + val object4MiB = 4 * 1024 * 1024 + serve { + when (request.path()) { + "/v1/objects/fits" -> respondBytes(ByteArray(object4MiB)) + "/v1/objects/big" -> respondBytes(ByteArray(object4MiB + 1)) + else -> respondJson(200, """{"objects":[],"cursor":"1","pad":"${"x".repeat(16 * 1024 * 1024)}"}""") + } + } + assertEquals(object4MiB, store().getObject("fits")?.size) + assertTrue(runCatching { store().getObject("big") }.exceptionOrNull() is StoreException.Malformed) + assertTrue(runCatching { store().changes(0) }.exceptionOrNull() is StoreException.Malformed) + } + @Test fun the_status_says_when_each_device_was_last_seen() = runTest { diff --git a/app/src/test/java/io/github/aedev/flow/sync2/WebDavStoreTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/WebDavStoreTest.kt index 3f4ddf0b73..46b7b1bab1 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/WebDavStoreTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/WebDavStoreTest.kt @@ -4,6 +4,11 @@ import android.app.Application import androidx.test.ext.junit.runners.AndroidJUnit4 import io.github.aedev.flow.sync2.store.StoreException import io.github.aedev.flow.sync2.store.WebDavStore +import io.ktor.http.HttpMethod +import io.ktor.http.HttpStatusCode +import io.ktor.server.request.httpMethod +import io.ktor.server.response.respondBytes +import io.ktor.server.response.respondText import kotlinx.coroutines.test.runTest import org.junit.After import org.junit.Assert.assertArrayEquals @@ -93,6 +98,22 @@ class WebDavStoreTest { assertNull(plain.tag("dev")) } + @Test + fun answers_past_their_limit_are_refused() = + runTest { + StandInServer { + if (request.httpMethod == HttpMethod.Get) { + respondBytes(ByteArray(4 * 1024 * 1024 + 1)) + } else { + respondText(" ".repeat(16 * 1024 * 1024 + 1), status = HttpStatusCode.MultiStatus) + } + }.use { standIn -> + val dav = WebDavStore.connect(standIn.url, "u", "p") + assertTrue(runCatching { dav.get("dev/x.o") }.exceptionOrNull() is StoreException.Malformed) + assertTrue(runCatching { dav.list("dev/") }.exceptionOrNull() is StoreException.Malformed) + } + } + @Test fun plain_http_is_refused_except_to_private_addresses() { val error = runCatching { WebDavStore.connect("http://example.com/dav/", "u", "p") }.exceptionOrNull()