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) + } +}