From 0a7c112b8bf7c516cf2c0d97b48f26bb4aa71f53 Mon Sep 17 00:00:00 2001 From: Paul Hitt Date: Sun, 30 Aug 2026 11:51:44 -0400 Subject: [PATCH] Test the inbound path with real ciphertext MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `poll` was at 0%. The tests that cover it — the Live* suites — are opt-in via FN_LIVE_FLEET and skip in an ordinary run, so the single most important path in the client contributed nothing to a normal build. That is worse than having no test, because the file listing says otherwise. Nothing here is stubbed below the transport: the peer is a real DeviceIdentity, the envelopes are real ciphertext from a real ratchet, and only the relay's HTTP is a mock. A decrypt that succeeds here succeeded for the same reason it does in the field. The three-chain rule gets the test its comment asks for. This used to be `existingSession(fromDevice) ?: respond(...)`, which reads as a fallback and is not one: with a stored session it was the only thing tried, decrypt threw for a peer on a different chain, and the catch swallowed it. That happens whenever both sides run X3DH before either has processed the other's prekey — two people opening the same chat at once, which is ordinary — so the test stores one chain and then has the peer start a fresh one. The C1 receive-side gate is asserted from both sides: a handshake offering an unpinned key while claiming a pinned contact is dropped, and the same handshake from the pinned key is accepted. Without the gate, anyone could hand us a session with their own key and every later message would render as that contact. Also that one unreadable envelope does not discard the rest of the page AND that the watermark still passes it, or a single corrupt message is retried for ever and the inbox never drains; and that a failed poll surfaces rather than being reported as an empty inbox, since "no new messages" and "we could not ask" are different states. ⚠️ One test needed two distinct user ids: two identities claiming one user is exactly what C1 refuses, so writing it with the default sender makes the second message vanish — correctly, and confusingly. Conversations 25.4% -> 43.0%, client 63.2% -> 68.9% lines. Co-Authored-By: Claude Opus 4.8 --- .../flamenet/client/ConversationsPollTest.kt | 295 ++++++++++++++++++ 1 file changed, 295 insertions(+) create mode 100644 client/src/test/kotlin/io/flamenet/client/ConversationsPollTest.kt diff --git a/client/src/test/kotlin/io/flamenet/client/ConversationsPollTest.kt b/client/src/test/kotlin/io/flamenet/client/ConversationsPollTest.kt new file mode 100644 index 0000000..66f3aa1 --- /dev/null +++ b/client/src/test/kotlin/io/flamenet/client/ConversationsPollTest.kt @@ -0,0 +1,295 @@ +package io.flamenet.client + +import io.flamenet.e2e.DeviceIdentity +import io.flamenet.e2e.E2ESession +import io.flamenet.e2e.PeerBundle +import io.flamenet.e2e.Vault +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import org.json.JSONObject +import javax.crypto.KeyGenerator +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotEquals +import kotlin.test.assertTrue + +/** + * The inbound path, with real ciphertext. + * + * 🔴 **`poll` was at 0%.** The tests that cover it — the `Live*` suites — are + * opt-in via `FN_LIVE_FLEET` and skip in an ordinary run, so the single most + * important path in the client contributed nothing to a normal build. That is + * worse than having no test, because the file listing says it is covered. + * + * Nothing here is stubbed below the transport: the peer is a real + * `DeviceIdentity`, the envelopes are real ciphertext from a real ratchet, and + * only the relay's HTTP is a mock. A decrypt that succeeds here succeeded for + * the same reason it does in the field. + */ +class ConversationsPollTest { + + private class MapStorage : Vault.Storage { + val records = LinkedHashMap() + override fun read(name: String) = records[name] + override fun write(name: String, bytes: ByteArray) { records[name] = bytes } + override fun delete(name: String) { records.remove(name) } + override fun names() = records.keys.toList() + } + + private lateinit var server: MockWebServer + private lateinit var vault: Vault + private lateinit var me: DeviceIdentity + + @BeforeTest fun start() { + server = MockWebServer().also { it.start() } + vault = Vault(MapStorage(), + KeyGenerator.getInstance("AES").apply { init(256) }.generateKey()) + me = DeviceIdentity.create() + } + + @AfterTest fun stop() = server.shutdown() + + private fun conversations() = + Conversations(me, vault, RelayClient(server.url("/").toString(), "a", "b")) + + private fun json(body: String) = MockResponse() + .setResponseCode(200).setHeader("Content-Type", "application/json").setBody(body) + + /** My own bundle, as a peer would fetch it. */ + private fun myBundle(): PeerBundle { + val opk = me.generateOneTimePreKeys(1).first() + return PeerBundle.fromWire(me.registrationWire().put("opk", + JSONObject().put("id", opk.id).put("pub", Wire.b64(opk.publicKey)))) + } + + /** One inbox page. */ + private fun inbox(vararg messages: String) = + json("""{"messages":[${messages.joinToString(",")}]}""") + + private fun envelopeJson(id: Long, fromDevice: String, payload: String, + fromUser: Long = 7) = + """{"id":$id,"from_user":$fromUser,"from_device":"$fromDevice","payload":"$payload"}""" + + /** A peer who has never written to us before. */ + private fun stranger(): Pair { + val peer = DeviceIdentity.create() + return peer to E2ESession.initiate(peer, myBundle()) + } + + // MARK: - The ordinary case + + @Test + fun `a first message from a new peer opens a session and decrypts`() { + val (peer, session) = stranger() + val envelope = session.encrypt("hello there") + server.enqueue(inbox(envelopeJson(1, peer.deviceId, envelope.toPayload()))) + + val received = conversations().poll() + + assertEquals(1, received.size) + assertEquals("hello there", received.single().text) + assertEquals(peer.deviceId, received.single().fromDevice) + } + + /** + * 🔴 An advancing watermark is what stops the same message arriving for + * ever. It must move even though the relay decides the ids. + */ + @Test + fun `the watermark advances past what was read`() { + val (peer, session) = stranger() + server.enqueue(inbox( + envelopeJson(11, peer.deviceId, session.encrypt("one").toPayload()), + envelopeJson(12, peer.deviceId, session.encrypt("two").toPayload()))) + + conversations().poll() + assertEquals(12, vault.inboxWatermark) + + // And the next poll asks for what comes after it. + server.enqueue(inbox()) + conversations().poll() + val asked = server.takeRequest().let { server.takeRequest() } + assertTrue(asked.path!!.contains("after=12"), "asked ${asked.path}") + } + + /** + * 🔴 **Three chains to try, not one.** This used to be + * `existingSession(fromDevice) ?: respond(...)`, which reads as a fallback + * and is not one: when a stored session existed it was the ONLY thing + * tried, decrypt threw for a peer on a different chain, and the catch + * swallowed it — so `respond` was never reached. + * + * That happens whenever both sides run X3DH before either has processed the + * other's prekey: two people opening the same chat at once, which is + * ordinary behaviour. Each adopts the other's chain, and from then on every + * message arrives on a chain the receiver has thrown away. + */ + @Test + fun `a peer who started a new chain is still readable`() { + val peer = DeviceIdentity.create() + + // An old session with this peer, which we have stored. + val old = E2ESession.initiate(peer, myBundle()) + val firstEnvelope = old.encrypt("from the old chain") + server.enqueue(inbox(envelopeJson(1, peer.deviceId, firstEnvelope.toPayload()))) + assertEquals("from the old chain", conversations().poll().single().text) + + // The peer restarts and initiates a brand new chain to the same device. + val fresh = E2ESession.initiate(peer, myBundle()) + server.enqueue(inbox(envelopeJson(2, peer.deviceId, + fresh.encrypt("from the new chain").toPayload()))) + + val received = conversations().poll() + assertEquals(1, received.size, "the new chain was dropped") + assertEquals("from the new chain", received.single().text) + } + + /** + * A message that cannot be opened must not take the rest of the page with + * it. One corrupt envelope from one peer would otherwise stall every + * conversation behind it. + */ + @Test + fun `an unreadable message does not discard the rest of the batch`() { + val (peer, session) = stranger() + server.enqueue(inbox( + envelopeJson(1, "garbage-device", "not-a-real-payload"), + envelopeJson(2, peer.deviceId, session.encrypt("still delivered").toPayload()))) + + val received = conversations().poll() + assertEquals(1, received.size) + assertEquals("still delivered", received.single().text) + // ⚠️ And the watermark passes BOTH, or the unreadable one is retried + // for ever and the inbox never drains. + assertEquals(2, vault.inboxWatermark) + } + + @Test + fun `an empty inbox is not an error`() { + server.enqueue(inbox()) + assertEquals(0, conversations().poll().size) + } + + /** + * Two peers in one page are each opened on their own chain. + * + * ⚠️ They must be different USERS. Two identities claiming one user id is + * exactly what the C1 gate refuses, so writing this with the default sender + * makes the second message vanish — correctly, and confusingly. + */ + @Test + fun `messages from two peers are both delivered`() { + val (alice, aliceSession) = stranger() + val (bob, bobSession) = stranger() + server.enqueue(inbox( + envelopeJson(1, alice.deviceId, aliceSession.encrypt("from alice").toPayload(), + fromUser = 7), + envelopeJson(2, bob.deviceId, bobSession.encrypt("from bob").toPayload(), + fromUser = 8))) + + val texts = conversations().poll().map { it.text }.toSet() + assertEquals(setOf("from alice", "from bob"), texts) + } + + /** + * The session is persisted, or every message would be a fresh handshake and + * the ratchet would never advance. + */ + @Test + fun `the opened session is stored for the next message`() { + val (peer, session) = stranger() + server.enqueue(inbox(envelopeJson(1, peer.deviceId, + session.encrypt("first").toPayload()))) + conversations().poll() + + // A second message on the SAME chain, read by a fresh Conversations + // over the same vault. + server.enqueue(inbox(envelopeJson(2, peer.deviceId, + session.encrypt("second").toPayload()))) + assertEquals("second", conversations().poll().single().text) + } + + /** Out-of-order arrival is ordinary on a phone changing networks. */ + @Test + fun `messages arriving out of order still decrypt`() { + val (peer, session) = stranger() + val first = session.encrypt("one") + val second = session.encrypt("two") + val third = session.encrypt("three") + + server.enqueue(inbox( + envelopeJson(1, peer.deviceId, first.toPayload()), + envelopeJson(3, peer.deviceId, third.toPayload()), + envelopeJson(2, peer.deviceId, second.toPayload()))) + + val texts = conversations().poll().map { it.text } + assertEquals(setOf("one", "two", "three"), texts.toSet()) + } + + /** + * 🔴 **The receive-side identity gate (audit C1).** Building a session from + * a prekey envelope means accepting whatever identity key it offers, so it + * must pass the SAME pin check the send path applies. Without it, anyone + * could hand us a handshake with their own key while claiming to be a + * contact we have already pinned, and every later message from that session + * would render as that contact. + */ + @Test + fun `a handshake from an unpinned key claiming a pinned contact is refused`() { + val realContact = DeviceIdentity.create() + // We have pinned the real contact's identity for user 7. + vault.saveTrust(7, Vault.Trust( + identities = listOf(Wire.b64(realContact.identitySigning.publicKey)), + verified = true)) + + // An impostor initiates, claiming to be user 7. + val impostor = DeviceIdentity.create() + val session = E2ESession.initiate(impostor, myBundle()) + server.enqueue(inbox(envelopeJson( + 1, impostor.deviceId, session.encrypt("trust me").toPayload(), fromUser = 7))) + + val received = conversations().poll() + assertEquals(0, received.size, "an unpinned key was accepted as a pinned contact") + } + + /** The same handshake from the pinned key itself is accepted. */ + @Test + fun `a handshake from the pinned key is accepted`() { + val contact = DeviceIdentity.create() + vault.saveTrust(7, Vault.Trust( + identities = listOf(Wire.b64(contact.identitySigning.publicKey)), + verified = true)) + + val session = E2ESession.initiate(contact, myBundle()) + server.enqueue(inbox(envelopeJson( + 1, contact.deviceId, session.encrypt("it is me").toPayload(), fromUser = 7))) + + assertEquals("it is me", conversations().poll().single().text) + } + + /** + * A relay that cannot be reached must surface rather than silently + * reporting an empty inbox — "no new messages" and "we could not ask" are + * different states and a client that conflates them looks broken to nobody. + */ + @Test + fun `a failed poll is not reported as an empty inbox`() { + server.enqueue(MockResponse().setResponseCode(503).setBody("down")) + var threw = false + try { conversations().poll() } catch (_: Exception) { threw = true } + assertTrue(threw, "a 503 was reported as an empty inbox") + // The watermark did not move on a failure. + assertEquals(0, vault.inboxWatermark) + } + + /** The device polls its OWN inbox, by its own id. */ + @Test + fun `the poll names this device`() { + server.enqueue(inbox()) + conversations().poll() + val path = server.takeRequest().path!! + assertTrue(path.contains("device_id=${me.deviceId}"), path) + } +}