RelayAdoptionHost.kt

1package com.peachpasswords.sync23import java.security.MessageDigest4import javax.inject.Inject5import javax.inject.Singleton6import kotlinx.coroutines.CoroutineScope7import kotlinx.coroutines.Dispatchers8import kotlinx.coroutines.Job9import kotlinx.coroutines.channels.awaitClose10import kotlinx.coroutines.flow.Flow11import kotlinx.coroutines.flow.callbackFlow12import kotlinx.coroutines.launch13import kotlinx.coroutines.withTimeout14import org.json.JSONObject1516sealed interface RelayAdoptionEvent {17    /** The source is already occupying the non-buffering room; it is safe to show this. */18    data class InvitationReady(val invitation: String) : RelayAdoptionEvent19    data class AwaitingConfirmation(20        val sas: String,21        val deviceName: String,22        val session: PairingSession,23    ) : RelayAdoptionEvent24    data class Adopted(val deviceName: String) : RelayAdoptionEvent25    data object Expired : RelayAdoptionEvent26    data class Rejected(val reason: String) : RelayAdoptionEvent27}2829@Singleton30class RelayAdoptionHost @Inject constructor(31    private val relaySettings: RelaySettingsStore,32) {33    private var job: Job? = null34    private var socket: RelayBinarySocket? = null3536    fun start(37        scope: CoroutineScope,38        ourStaticPrivateKey: ByteArray,39        ourStaticPublicKey: ByteArray,40        localIdentity: LocalDeviceIdentity,41        sourceDeps: AdoptionSourceDeps,42        windowMs: Long = PAIRING_WINDOW_MS,43    ): Flow<RelayAdoptionEvent> = callbackFlow {44        val relayUrl = relaySettings.relayUrl()45        val token = Rendezvous.newPairingToken()46        val invitation = createPairingInvitation(relayUrl, token)4748        job = scope.launch(Dispatchers.IO) {49            try {50                val connected = OkHttpRelaySocket.connect(51                    Rendezvous.buildUrl(relayUrl, token),52                )53                socket = connected54                // The relay buffers nothing, so the invitation is not surfaced55                // until the source WebSocket is open and waiting in the room.56                trySend(RelayAdoptionEvent.InvitationReady(invitation))5758                val adopted = withTimeout(windowMs) {59                    runRelayAdoptionSourceSession(60                        socket = connected,61                        expectedToken = token,62                        relayUrl = relayUrl,63                        ourStaticPrivateKey = ourStaticPrivateKey,64                        ourStaticPublicKey = ourStaticPublicKey,65                        localIdentity = localIdentity,66                        sourceDeps = sourceDeps,67                        confirm = { sas, request ->68                            val pending = PairingSession()69                            trySend(70                                RelayAdoptionEvent.AwaitingConfirmation(71                                    sas = sas,72                                    deviceName = request.deviceName.ifBlank { "New device" },73                                    session = pending,74                                ),75                            )76                            pending.await()77                        },78                    )79                }80                if (adopted == null) {81                    trySend(RelayAdoptionEvent.Rejected("adoption_rejected"))82                } else {83                    trySend(84                        RelayAdoptionEvent.Adopted(85                            adopted.deviceName?.takeIf(String::isNotBlank) ?: "New device",86                        ),87                    )88                }89            } catch (error: kotlinx.coroutines.TimeoutCancellationException) {90                trySend(RelayAdoptionEvent.Expired)91            } catch (error: kotlinx.coroutines.CancellationException) {92                // stop() cancels this job when the user leaves the screen or93                // restarts pairing. That is not a pairing failure — reporting it94                // as one raced the new session and replaced the freshly shown95                // code with "Couldn't pair: StandaloneCoroutine was cancelled".96                // Cancellation must propagate, never be swallowed.97                throw error98            } catch (error: Throwable) {99                trySend(RelayAdoptionEvent.Rejected(error.message ?: "relay_adoption_failed"))100            } finally {101                socket?.close()102                socket = null103                close()104            }105        }106107        awaitClose { stop() }108    }109110    fun stop() {111        job?.cancel()112        job = null113        socket?.close()114        socket = null115    }116}117118/**119 * Source half of first-contact adoption through an opaque relay splice.120 *121 * The relay token selects only a room. Authentication and confidentiality122 * begin with Noise NK; the root grant remains behind both SAS decisions in123 * [grantAdoption].124 */125internal suspend fun runRelayAdoptionSourceSession(126    socket: RelayBinarySocket,127    expectedToken: String,128    relayUrl: String,129    ourStaticPrivateKey: ByteArray,130    ourStaticPublicKey: ByteArray,131    localIdentity: LocalDeviceIdentity,132    sourceDeps: AdoptionSourceDeps,133    confirm: suspend (sas: String, request: AdoptionFrame.Request) -> Boolean,134): PairedDevice? {135    val hello = JSONObject(String(socket.receive(), Charsets.UTF_8))136    require(hello.optString("type") == "PEACH_HELLO") { "expected PEACH_HELLO" }137    require(138        MessageDigest.isEqual(139            expectedToken.toByteArray(Charsets.US_ASCII),140            hello.optString("token").toByteArray(Charsets.US_ASCII),141        ),142    ) { "pairing token mismatch" }143144    socket.send(145        JSONObject()146            .put("type", "PEACH_HELLO_ACK")147            .put(148                "ephemeralPublicKeyHex",149                ourStaticPublicKey.joinToString("") { "%02x".format(it) },150            )151            .toString()152            .toByteArray(Charsets.UTF_8),153    )154155    val noise = noiseNKResponder(156        ourStaticPrivateKey,157        ourStaticPublicKey,158        socket::send,159        socket::receive,160    )161    try {162        val channel = object : PairingTransport {163            override suspend fun send(msg: ByteArray) = socket.send(noise.encrypt(msg))164            override suspend fun receive(): ByteArray = noise.decrypt(socket.receive())165            override suspend fun close() = socket.close()166        }167168        val request = receiveFrame(channel)169        require(isAdoptRequest(request)) { "expected ADOPT_REQUEST" }170        request as AdoptionFrame.Request171        if (!confirm(SASDerivation.derive(noise.handshakeHash), request)) return null172173        val pairState = AdoptionPairState(174            rendezvousSecretHex = Rendezvous.bytesToHex(175                Rendezvous.deriveSecret(noise.handshakeHash),176            ),177            relayUrl = Rendezvous.normalizeRelayUrl(relayUrl),178        )179        val boundDeps = object : AdoptionSourceDeps {180            override val sourceIdentity: LocalDeviceIdentity = localIdentity181            override val pairState: AdoptionPairState = pairState182            override suspend fun rootKeyHex(): String = sourceDeps.rootKeyHex()183            override suspend fun vaultId(): String = sourceDeps.vaultId()184            override suspend fun license(): String? = sourceDeps.license()185            override suspend fun onConsent(request: AdoptionFrame.Request): Boolean = true186            override suspend fun upsertPairedDevice(device: PairedDevice) {187                sourceDeps.upsertPairedDevice(device)188            }189        }190        return grantAdoption(channel, request, boundDeps)191    } finally {192        noise.close()193    }194}