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 kotlinx.coroutines.withTimeoutOrNull15import org.json.JSONObject1617sealed interface RelayAdoptionEvent {18    /** The source is already occupying the non-buffering room; it is safe to show this. */19    data class InvitationReady(val invitation: String) : RelayAdoptionEvent20    data class AwaitingPeerConfirmation(21        val sas: String,22        val deviceName: String,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        // Pairing is also how this mesh's relay reaches a new peer: the address45        // in the invitation is the one the receiver will dial for ever after.46        // Handing out Peach's relay while the user has asked for a private one47        // would push the same lie one layer down, into the mesh.48        val relayUrl = when (val decision = relaySettings.relayInForce()) {49            is RelayDecision.Refuse -> {50                trySend(RelayAdoptionEvent.Rejected(decision.reason.wire()))51                close()52                return@callbackFlow53            }5455            is RelayDecision.Dial -> decision.relayUrl56        }57        val token = Rendezvous.newPairingToken()58        val invitation = createPairingInvitation(relayUrl, token)5960        job = scope.launch(Dispatchers.IO) {61            try {62                val connected = OkHttpRelaySocket.connect(63                    Rendezvous.buildUrl(relayUrl, token),64                )65                socket = connected66                // The relay buffers nothing, so the invitation is not surfaced67                // until the source WebSocket is open and waiting in the room.68                trySend(RelayAdoptionEvent.InvitationReady(invitation))6970                val adopted = withTimeout(windowMs) {71                    runRelayAdoptionSourceSession(72                        socket = connected,73                        expectedToken = token,74                        relayUrl = relayUrl,75                        ourStaticPrivateKey = ourStaticPrivateKey,76                        ourStaticPublicKey = ourStaticPublicKey,77                        localIdentity = localIdentity,78                        sourceDeps = sourceDeps,79                        confirm = { sas, request ->80                            trySend(81                                RelayAdoptionEvent.AwaitingPeerConfirmation(82                                    sas = sas,83                                    deviceName = request.deviceName.ifBlank { "New device" },84                                ),85                            )86                            // Showing a fresh invitation from an unlocked vault87                            // is source-side authorization. The scanning device's88                            // explicit SAS confirmation is the single commit gate.89                            true90                        },91                    )92                }93                if (adopted == null) {94                    trySend(RelayAdoptionEvent.Rejected("adoption_rejected"))95                } else {96                    trySend(97                        RelayAdoptionEvent.Adopted(98                            adopted.deviceName?.takeIf(String::isNotBlank) ?: "New device",99                        ),100                    )101                }102            } catch (error: kotlinx.coroutines.TimeoutCancellationException) {103                trySend(RelayAdoptionEvent.Expired)104            } catch (error: kotlinx.coroutines.CancellationException) {105                // stop() cancels this job when the user leaves the screen or106                // restarts pairing. That is not a pairing failure — reporting it107                // as one raced the new session and replaced the freshly shown108                // code with "Couldn't pair: StandaloneCoroutine was cancelled".109                // Cancellation must propagate, never be swallowed.110                throw error111            } catch (error: Throwable) {112                trySend(RelayAdoptionEvent.Rejected(error.message ?: "relay_adoption_failed"))113            } finally {114                socket?.close()115                socket = null116                close()117            }118        }119120        awaitClose { stop() }121    }122123    fun stop() {124        job?.cancel()125        job = null126        socket?.close()127        socket = null128    }129}130131/**132 * Source half of first-contact adoption through an opaque relay splice.133 *134 * The relay token selects only a room. Authentication and confidentiality135 * begin with Noise NK; the root grant remains behind both SAS decisions in136 * [grantAdoption].137 */138internal suspend fun runRelayAdoptionSourceSession(139    socket: RelayBinarySocket,140    expectedToken: String,141    relayUrl: String,142    ourStaticPrivateKey: ByteArray,143    ourStaticPublicKey: ByteArray,144    localIdentity: LocalDeviceIdentity,145    sourceDeps: AdoptionSourceDeps,146    confirm: suspend (sas: String, request: AdoptionFrame.Request) -> Boolean,147): PairedDevice? {148    val hello = JSONObject(String(socket.receive(), Charsets.UTF_8))149    require(hello.optString("type") == "PEACH_HELLO") { "expected PEACH_HELLO" }150    require(151        MessageDigest.isEqual(152            expectedToken.toByteArray(Charsets.US_ASCII),153            hello.optString("token").toByteArray(Charsets.US_ASCII),154        ),155    ) { "pairing token mismatch" }156157    socket.send(158        JSONObject()159            .put("type", "PEACH_HELLO_ACK")160            .put(161                "ephemeralPublicKeyHex",162                ourStaticPublicKey.joinToString("") { "%02x".format(it) },163            )164            .toString()165            .toByteArray(Charsets.UTF_8),166    )167168    val noise = noiseNKResponder(169        ourStaticPrivateKey,170        ourStaticPublicKey,171        socket::send,172        socket::receive,173    )174    try {175        val channel = object : PairingTransport {176            override suspend fun send(msg: ByteArray) = socket.send(noise.encrypt(msg))177            override suspend fun receive(): ByteArray = noise.decrypt(socket.receive())178            override suspend fun close() = socket.close()179        }180181        val request = receiveFrame(channel)182        require(isAdoptRequest(request)) { "expected ADOPT_REQUEST" }183        request as AdoptionFrame.Request184        if (!confirm(SASDerivation.derive(noise.handshakeHash), request)) return null185186        val pairState = AdoptionPairState(187            rendezvousSecretHex = Rendezvous.bytesToHex(188                Rendezvous.deriveSecret(noise.handshakeHash),189            ),190            relayUrl = Rendezvous.normalizeRelayUrl(relayUrl),191        )192        val boundDeps = object : AdoptionSourceDeps {193            override val sourceIdentity: LocalDeviceIdentity = localIdentity194            override val pairState: AdoptionPairState = pairState195            override suspend fun rootKeyHex(): String = sourceDeps.rootKeyHex()196            override suspend fun vaultId(): String = sourceDeps.vaultId()197            override suspend fun license(): String? = sourceDeps.license()198            override suspend fun applyLicense(license: String) {199                sourceDeps.applyLicense(license)200            }201            override suspend fun onConsent(request: AdoptionFrame.Request): Boolean = true202            override suspend fun upsertPairedDevice(device: PairedDevice) {203                sourceDeps.upsertPairedDevice(device)204            }205            override suspend fun vaultSnapshot() = sourceDeps.vaultSnapshot()206            override suspend fun outgoingDeviceManifest() = sourceDeps.outgoingDeviceManifest()207            override suspend fun markSynced(deviceId: String) = sourceDeps.markSynced(deviceId)208            override suspend fun duressSlotJson() = sourceDeps.duressSlotJson()209        }210        val adopted = grantAdoption(channel, request, boundDeps) ?: return null211        // If the receiver advertised snapshot support, send the vault inline.212        if (request.snapshotCapable == true) {213            try {214                val snapshot = boundDeps.vaultSnapshot()215                if (snapshot != null) {216                    val duressSlot = runCatching { boundDeps.duressSlotJson() }.getOrNull()217                    sendVaultSnapshot(channel, snapshot.first, snapshot.second, duressSlot)218                    val applied = receiveFrame(channel)219                    if (applied is AdoptionFrame.SnapshotApplied) {220                        val manifest = boundDeps.outgoingDeviceManifest()221                        if (manifest != null) {222                            sendFrame(channel, AdoptionFrame.DeviceManifestFrame(manifest))223                            withTimeoutOrNull(5_000) { receiveFrame(channel) }224                        }225                        boundDeps.markSynced(adopted.deviceId)226                    }227                }228            } catch (error: Exception) {229                if (error is SnapshotDuressFrameTooLargeException) throw error230                // Fallback: receiver uses the sync pipeline.231            }232        }233        return adopted234    } finally {235        noise.close()236    }237}