feat: wait for secure relay sync (v2.3.75)

@@ -0,0 +1,302 @@1+package com.peachpasswords.sync2+3+import com.peachpasswords.vault.VaultCrypto4+import kotlinx.coroutines.CompletableDeferred5+import kotlinx.coroutines.CoroutineScope6+import kotlinx.coroutines.Dispatchers7+import kotlinx.coroutines.Job8+import kotlinx.coroutines.channels.Channel9+import kotlinx.coroutines.delay10+import kotlinx.coroutines.isActive11+import kotlinx.coroutines.launch12+import kotlinx.coroutines.withContext13+import kotlinx.coroutines.withTimeout14+import okhttp3.OkHttpClient15+import okhttp3.Request16+import okhttp3.Response17+import okhttp3.WebSocket18+import okhttp3.WebSocketListener19+import okio.ByteString20+import okio.ByteString.Companion.toByteString21+import org.json.JSONObject22+import java.security.MessageDigest23+import java.util.concurrent.TimeUnit24+import javax.inject.Inject25+import javax.inject.Singleton26+27+private const val RECONCILE_INTERVAL_MS = 60_000L28+private const val RETRY_DELAY_MS = 1_000L29+private const val RELAY_CONNECT_TIMEOUT_MS = 10_000L30+private const val RELAY_INBOX_CAPACITY = 831+32+/**33+ * Keeps one outbound waiting socket for each paired peer's current and previous34+ * rendezvous windows. The Cloudflare relay is only a splice: once a browser35+ * arrives, this class runs the same selector + Noise KK + PAIR_IDENTITY + sync36+ * responder protocol as the LAN listener.37+ */38+@Singleton39+class CloudRelayReconnectListener @Inject constructor(40+    private val deviceStore: PairedDeviceStore,41+    private val relaySettings: RelaySettingsStore,42+) {43+    private var supervisor: Job? = null44+    private val watchers = mutableMapOf<String, Job>()45+    private var retainedPrivateKey: ByteArray? = null46+    private var retainedPublicKey: ByteArray? = null47+48+    fun start(49+        scope: CoroutineScope,50+        ourStaticPrivateKey: ByteArray,51+        ourStaticPublicKey: ByteArray,52+        localIdentity: LocalChirpIdentity,53+        vaultFactory: () -> ChirpSyncVaultAdapter?,54+    ) {55+        if (supervisor?.isActive == true) return56+        val privateKey = ourStaticPrivateKey.copyOf()57+        val publicKey = ourStaticPublicKey.copyOf()58+        retainedPrivateKey = privateKey59+        retainedPublicKey = publicKey60+        supervisor = scope.launch(Dispatchers.IO) {61+            try {62+                while (isActive) {63+                    reconcile(64+                        this,65+                        privateKey,66+                        publicKey,67+                        localIdentity,68+                        vaultFactory,69+                    )70+                    delay(RECONCILE_INTERVAL_MS)71+                }72+            } finally {73+                watchers.values.forEach(Job::cancel)74+                watchers.clear()75+            }76+        }77+    }78+79+    fun stop() {80+        supervisor?.cancel()81+        supervisor = null82+        watchers.values.forEach(Job::cancel)83+        watchers.clear()84+        retainedPrivateKey?.let(VaultCrypto::zero)85+        retainedPublicKey?.let(VaultCrypto::zero)86+        retainedPrivateKey = null87+        retainedPublicKey = null88+    }89+90+    private fun reconcile(91+        scope: CoroutineScope,92+        privateKey: ByteArray,93+        publicKey: ByteArray,94+        localIdentity: LocalChirpIdentity,95+        vaultFactory: () -> ChirpSyncVaultAdapter?,96+    ) {97+        if (relaySettings.isLanOnly()) {98+            watchers.values.forEach(Job::cancel)99+            watchers.clear()100+            return101+        }102+103+        val now = System.currentTimeMillis() / 1000L104+        val desired = buildMap {105+            for (device in deviceStore.getAll()) {106+                val secret = device.rendezvousSecretHex107+                    ?.let { runCatching { Rendezvous.secretFromHex(it) }.getOrNull() }108+                    ?: continue109+                val relayUrl = device.relayUrl110+                    ?.let { runCatching { Rendezvous.normalizeRelayUrl(it) }.getOrNull() }111+                    ?: relaySettings.relayUrl()112+                for (rendezvousId in Rendezvous.waitingIds(secret, now)) {113+                    val key = "${device.deviceId}|$relayUrl|$rendezvousId"114+                    put(key, RelayWaitTarget(device.deviceId, relayUrl, rendezvousId))115+                }116+            }117+        }118+119+        for ((key, job) in watchers.toMap()) {120+            if (key !in desired) {121+                job.cancel()122+                watchers.remove(key)123+            }124+        }125+        for ((key, target) in desired) {126+            if (watchers[key]?.isActive == true) continue127+            watchers[key] = scope.launch {128+                while (isActive) {129+                    runCatching {130+                        val known = deviceStore.get(target.deviceId)131+                            ?: throw IllegalStateException("relay peer was unpaired")132+                        val socket = OkHttpRelaySocket.connect(133+                            Rendezvous.buildUrl(target.relayUrl, target.rendezvousId),134+                        )135+                        try {136+                            runCloudRelayResponderSession(137+                                socket = socket,138+                                known = known,139+                                registeredRendezvousId = target.rendezvousId,140+                                ourStaticPrivateKey = privateKey,141+                                ourStaticPublicKey = publicKey,142+                                localIdentity = localIdentity,143+                                vaultFactory = vaultFactory,144+                                onSyncComplete = {145+                                    deviceStore.updateLastSync(known.deviceId)146+                                },147+                            )148+                        } finally {149+                            socket.close()150+                        }151+                    }152+                    delay(RETRY_DELAY_MS)153+                }154+            }155+        }156+    }157+}158+159+private data class RelayWaitTarget(160+    val deviceId: String,161+    val relayUrl: String,162+    val rendezvousId: String,163+)164+165+internal interface RelayBinarySocket {166+    fun send(frame: ByteArray)167+    suspend fun receive(): ByteArray168+    fun close(code: Int = 1000, reason: String = "relay session complete")169+}170+171+/**172+ * One relayed reconnect session. The registered rendezvous capability selects173+ * local state; KK still performs all authentication and has no NK fallback.174+ */175+internal suspend fun runCloudRelayResponderSession(176+    socket: RelayBinarySocket,177+    known: PairedDevice,178+    registeredRendezvousId: String,179+    ourStaticPrivateKey: ByteArray,180+    ourStaticPublicKey: ByteArray,181+    localIdentity: LocalChirpIdentity,182+    vaultFactory: () -> ChirpSyncVaultAdapter?,183+    onSyncComplete: () -> Unit,184+) {185+    val selector = JSONObject(String(socket.receive(), Charsets.UTF_8))186+    require(selector.optString("type") == "PEACH_RECONNECT") {187+        "expected PEACH_RECONNECT"188+    }189+    val presentedId = selector.optString("rendezvous_id")190+    require(191+        MessageDigest.isEqual(192+            registeredRendezvousId.toByteArray(Charsets.US_ASCII),193+            presentedId.toByteArray(Charsets.US_ASCII),194+        ),195+    ) { "reconnect capability does not match relay room" }196+    val secret = known.rendezvousSecretHex197+        ?.let { runCatching { Rendezvous.secretFromHex(it) }.getOrNull() }198+        ?: throw IllegalArgumentException("known peer missing rendezvous secret")199+    require(200+        Rendezvous.waitingIds(secret, System.currentTimeMillis() / 1000L).any {201+            MessageDigest.isEqual(202+                it.toByteArray(Charsets.US_ASCII),203+                registeredRendezvousId.toByteArray(Charsets.US_ASCII),204+            )205+        },206+    ) { "relay room is outside the accepted clock-skew windows" }207+208+    val initiatorKeyHex = known.longTermPublicKeyHex209+        ?: throw IllegalArgumentException("known peer missing static key")210+    val initiatorKey = hexToPubkey(initiatorKeyHex)211+    val session = noiseKKResponder(212+        ourStaticPrivateKey,213+        ourStaticPublicKey,214+        initiatorKey,215+        socket::send,216+        socket::receive,217+    )218+    val theirIdentity = parsePairIdentityFrame(219+        session.decrypt(socket.receive()),220+    )221+    require(identityMatchesAuthenticatedPeer(known, theirIdentity, initiatorKey)) {222+        "peer identity does not match authenticated KK key"223+    }224+    socket.send(session.encrypt(encodePairIdentityFrame(localIdentity)))225+226+    val vault = vaultFactory() ?: return227+    vault.bindAuthenticatedPeer(known)228+    val outcome = runChirpSyncResponder(229+        ChirpSyncSession(230+            remoteStaticPublicKey = initiatorKey.copyOf(),231+            send = { plaintext -> socket.send(session.encrypt(plaintext)) },232+            receive = { session.decrypt(socket.receive()) },233+            close = { socket.close() },234+        ),235+        vault,236+    )237+    if (outcome.status == SyncOutcomeStatus.COMPLETED) {238+        onSyncComplete()239+    }240+}241+242+private object OkHttpRelaySocket {243+    private val client = OkHttpClient.Builder()244+        .readTimeout(0, TimeUnit.MILLISECONDS)245+        .build()246+247+    suspend fun connect(url: String): RelayBinarySocket = withContext(Dispatchers.IO) {248+        val opened = CompletableDeferred<WebSocket>()249+        val incoming = Channel<ByteArray>(RELAY_INBOX_CAPACITY)250+        val listener = object : WebSocketListener() {251+            override fun onOpen(webSocket: WebSocket, response: Response) {252+                opened.complete(webSocket)253+            }254+255+            override fun onMessage(webSocket: WebSocket, bytes: ByteString) {256+                if (bytes.size > MAX_SYNC_FRAME_BYTES) {257+                    webSocket.close(1009, "relay frame too large")258+                    incoming.close(IllegalStateException("relay_frame_too_large"))259+                    return260+                }261+                if (!incoming.trySend(bytes.toByteArray()).isSuccess) {262+                    webSocket.close(1008, "relay receive queue full")263+                    incoming.close(IllegalStateException("relay_receive_queue_full"))264+                }265+            }266+267+            override fun onMessage(webSocket: WebSocket, text: String) {268+                webSocket.close(1003, "binary frames required")269+                incoming.close(IllegalStateException("relay_text_frame_rejected"))270+            }271+272+            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {273+                incoming.close(IllegalStateException("relay_closed_$code"))274+            }275+276+            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {277+                opened.completeExceptionally(t)278+                incoming.close(t)279+            }280+        }281+        val webSocket = client.newWebSocket(Request.Builder().url(url).build(), listener)282+        val connected = try {283+            withTimeout(RELAY_CONNECT_TIMEOUT_MS) { opened.await() }284+        } catch (error: Throwable) {285+            webSocket.cancel()286+            throw error287+        }288+        object : RelayBinarySocket {289+            override fun send(frame: ByteArray) {290+                require(frame.size <= MAX_SYNC_FRAME_BYTES) { "relay_frame_too_large" }291+                check(connected.send(frame.toByteString())) { "relay_send_failed" }292+            }293+294+            override suspend fun receive(): ByteArray = incoming.receive()295+296+            override fun close(code: Int, reason: String) {297+                connected.close(code, reason)298+                incoming.close()299+            }300+        }301+    }302+}

@@ -0,0 +1,149 @@1+package com.peachpasswords.sync2+3+import kotlinx.coroutines.Dispatchers4+import kotlinx.coroutines.async5+import kotlinx.coroutines.channels.Channel6+import kotlinx.coroutines.runBlocking7+import org.bouncycastle.crypto.params.X25519PrivateKeyParameters8+import org.junit.Assert.assertEquals9+import org.junit.Assert.assertFalse10+import org.junit.Assert.assertTrue11+import org.junit.Test12+import java.security.SecureRandom13+14+class CloudRelayReconnectListenerTest {15+    private data class KeyPair(val publicKey: ByteArray, val privateKey: ByteArray)16+17+    private fun keys(): KeyPair {18+        val privateKey = X25519PrivateKeyParameters(SecureRandom())19+        return KeyPair(privateKey.generatePublicKey().encoded, privateKey.encoded)20+    }21+22+    private fun socketPair(): Pair<RelayBinarySocket, RelayBinarySocket> {23+        val leftInbox = Channel<ByteArray>(Channel.UNLIMITED)24+        val rightInbox = Channel<ByteArray>(Channel.UNLIMITED)25+        fun socket(inbox: Channel<ByteArray>, peer: Channel<ByteArray>) =26+            object : RelayBinarySocket {27+                override fun send(frame: ByteArray) {28+                    check(peer.trySend(frame.copyOf()).isSuccess)29+                }30+31+                override suspend fun receive(): ByteArray = inbox.receive()32+33+                override fun close(code: Int, reason: String) {34+                    inbox.close()35+                }36+            }37+        return socket(leftInbox, rightInbox) to socket(rightInbox, leftInbox)38+    }39+40+    @Test41+    fun `relayed responder completes the same selector KK and identity wire`() = runBlocking {42+        val browserKeys = keys()43+        val phoneKeys = keys()44+        val secretHex = "61".repeat(32)45+        val rendezvousId = Rendezvous.waitingIds(46+            Rendezvous.secretFromHex(secretHex),47+            System.currentTimeMillis() / 1000L,48+        ).first()49+        val known = PairedDevice(50+            deviceId = "browser",51+            deviceName = "Firefox",52+            deviceType = "extension",53+            longTermPublicKeyHex = pubkeyToHex(browserKeys.publicKey),54+            rendezvousSecretHex = secretHex,55+            pairedAt = 1L,56+        )57+        val phoneIdentity = LocalChirpIdentity(58+            deviceId = "phone",59+            deviceName = "Pixel",60+            deviceType = DeviceType.ANDROID,61+            longTermPublicKeyHex = pubkeyToHex(phoneKeys.publicKey),62+            signingPublicKeyHex = "22".repeat(32),63+        )64+        val browserIdentity = LocalChirpIdentity(65+            deviceId = "browser",66+            deviceName = "Firefox",67+            deviceType = DeviceType.EXTENSION,68+            longTermPublicKeyHex = pubkeyToHex(browserKeys.publicKey),69+            signingPublicKeyHex = "11".repeat(32),70+        )71+        val (browserSocket, phoneSocket) = socketPair()72+        var syncMarked = false73+74+        val responder = async(Dispatchers.Default) {75+            runCloudRelayResponderSession(76+                socket = phoneSocket,77+                known = known,78+                registeredRendezvousId = rendezvousId,79+                ourStaticPrivateKey = phoneKeys.privateKey,80+                ourStaticPublicKey = phoneKeys.publicKey,81+                localIdentity = phoneIdentity,82+                vaultFactory = { null },83+                onSyncComplete = { syncMarked = true },84+            )85+        }86+        browserSocket.send(87+            """{"type":"PEACH_RECONNECT","rendezvous_id":"$rendezvousId"}"""88+                .toByteArray(),89+        )90+        val session = noiseKKInitiator(91+            browserKeys.privateKey,92+            browserKeys.publicKey,93+            phoneKeys.publicKey,94+            browserSocket::send,95+            browserSocket::receive,96+        )97+        browserSocket.send(session.encrypt(encodePairIdentityFrame(browserIdentity)))98+        val theirIdentity = parsePairIdentityFrame(99+            session.decrypt(browserSocket.receive()),100+        )101+102+        responder.await()103+        assertEquals(phoneIdentity.deviceId, theirIdentity.deviceId)104+        assertEquals(phoneIdentity.longTermPublicKeyHex, theirIdentity.longTermPublicKeyHex)105+        assertFalse(syncMarked)106+    }107+108+    @Test109+    fun `selector that does not name the registered relay room fails before KK`() = runBlocking {110+        val phoneKeys = keys()111+        val secretHex = "62".repeat(32)112+        val rendezvousId = Rendezvous.waitingIds(113+            Rendezvous.secretFromHex(secretHex),114+            System.currentTimeMillis() / 1000L,115+        ).first()116+        val known = PairedDevice(117+            deviceId = "browser",118+            longTermPublicKeyHex = pubkeyToHex(keys().publicKey),119+            rendezvousSecretHex = secretHex,120+            pairedAt = 1L,121+        )122+        val (sender, receiver) = socketPair()123+        sender.send(124+            """{"type":"PEACH_RECONNECT","rendezvous_id":"${"Z".repeat(43)}"}"""125+                .toByteArray(),126+        )127+128+        val failure = runCatching {129+            runCloudRelayResponderSession(130+                socket = receiver,131+                known = known,132+                registeredRendezvousId = rendezvousId,133+                ourStaticPrivateKey = phoneKeys.privateKey,134+                ourStaticPublicKey = phoneKeys.publicKey,135+                localIdentity = LocalChirpIdentity(136+                    deviceId = "phone",137+                    deviceName = "Pixel",138+                    deviceType = DeviceType.ANDROID,139+                    longTermPublicKeyHex = pubkeyToHex(phoneKeys.publicKey),140+                    signingPublicKeyHex = "22".repeat(32),141+                ),142+                vaultFactory = { null },143+                onSyncComplete = {},144+            )145+        }.exceptionOrNull()146+147+        assertTrue(failure?.message?.contains("does not match relay room") == true)148+    }149+}