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: LocalChirpIdentity,41        sourceDeps: AdoptionSourceDeps,42        windowMs: Long = CodePairingServer.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: Throwable) {92                trySend(RelayAdoptionEvent.Rejected(error.message ?: "relay_adoption_failed"))93            } finally {94                socket?.close()95                socket = null96                close()97            }98        }99100        awaitClose { stop() }101    }102103    fun stop() {104        job?.cancel()105        job = null106        socket?.close()107        socket = null108    }109}110111/**112 * Source half of first-contact adoption through an opaque relay splice.113 *114 * The relay token selects only a room. Authentication and confidentiality115 * begin with Noise NK; the root grant remains behind both SAS decisions in116 * [grantAdoption].117 */118internal suspend fun runRelayAdoptionSourceSession(119    socket: RelayBinarySocket,120    expectedToken: String,121    relayUrl: String,122    ourStaticPrivateKey: ByteArray,123    ourStaticPublicKey: ByteArray,124    localIdentity: LocalChirpIdentity,125    sourceDeps: AdoptionSourceDeps,126    confirm: suspend (sas: String, request: AdoptionFrame.Request) -> Boolean,127): PairedDevice? {128    val hello = JSONObject(String(socket.receive(), Charsets.UTF_8))129    require(hello.optString("type") == "PEACH_HELLO") { "expected PEACH_HELLO" }130    require(131        MessageDigest.isEqual(132            expectedToken.toByteArray(Charsets.US_ASCII),133            hello.optString("token").toByteArray(Charsets.US_ASCII),134        ),135    ) { "pairing token mismatch" }136137    socket.send(138        JSONObject()139            .put("type", "PEACH_HELLO_ACK")140            .put(141                "ephemeralPublicKeyHex",142                ourStaticPublicKey.joinToString("") { "%02x".format(it) },143            )144            .toString()145            .toByteArray(Charsets.UTF_8),146    )147148    val noise = noiseNKResponder(149        ourStaticPrivateKey,150        ourStaticPublicKey,151        socket::send,152        socket::receive,153    )154    val channel = object : PairingTransport {155        override suspend fun send(msg: ByteArray) = socket.send(noise.encrypt(msg))156        override suspend fun receive(): ByteArray = noise.decrypt(socket.receive())157        override suspend fun close() = socket.close()158    }159160    val request = receiveFrame(channel)161    require(isAdoptRequest(request)) { "expected ADOPT_REQUEST" }162    request as AdoptionFrame.Request163    if (!confirm(SASDerivation.derive(noise.handshakeHash), request)) return null164165    val pairState = AdoptionPairState(166        rendezvousSecretHex = Rendezvous.bytesToHex(167            Rendezvous.deriveSecret(noise.handshakeHash),168        ),169        relayUrl = Rendezvous.normalizeRelayUrl(relayUrl),170    )171    val boundDeps = object : AdoptionSourceDeps {172        override val sourceIdentity: LocalChirpIdentity = localIdentity173        override val pairState: AdoptionPairState = pairState174        override suspend fun rootKeyHex(): String = sourceDeps.rootKeyHex()175        override suspend fun vaultId(): String = sourceDeps.vaultId()176        override suspend fun license(): String? = sourceDeps.license()177        override suspend fun onConsent(request: AdoptionFrame.Request): Boolean = true178        override suspend fun upsertPairedDevice(device: PairedDevice) {179            sourceDeps.upsertPairedDevice(device)180        }181    }182    return grantAdoption(channel, request, boundDeps)183}