RelayAdoptionHost.kt

1package com.peachpasswords.sync23import android.util.Log4import java.security.MessageDigest5import javax.inject.Inject6import javax.inject.Singleton7import kotlinx.coroutines.CoroutineScope8import kotlinx.coroutines.Dispatchers9import kotlinx.coroutines.Job10import kotlinx.coroutines.channels.awaitClose11import kotlinx.coroutines.flow.Flow12import kotlinx.coroutines.flow.callbackFlow13import kotlinx.coroutines.launch14import kotlinx.coroutines.withTimeout15import kotlinx.coroutines.withTimeoutOrNull16import org.json.JSONObject1718sealed interface RelayAdoptionEvent {19    /** The source is already occupying the non-buffering room; it is safe to show this. */20    data class InvitationReady(val invitation: String) : RelayAdoptionEvent21    data class AwaitingPeerConfirmation(22        val sas: String,23        val deviceName: String,24    ) : RelayAdoptionEvent25    data class Adopted(val deviceName: String) : RelayAdoptionEvent26    data object Expired : RelayAdoptionEvent27    data class Rejected(val reason: String) : RelayAdoptionEvent28}2930@Singleton31class RelayAdoptionHost @Inject constructor(32    private val relaySettings: RelaySettingsStore,33) {34    private var job: Job? = null35    private var socket: RelayBinarySocket? = null3637    fun start(38        scope: CoroutineScope,39        ourStaticPrivateKey: ByteArray,40        ourStaticPublicKey: ByteArray,41        localIdentity: LocalDeviceIdentity,42        sourceDeps: AdoptionSourceDeps,43        windowMs: Long = PAIRING_WINDOW_MS,44    ): Flow<RelayAdoptionEvent> = callbackFlow {45        // Pairing is also how this mesh's relay reaches a new peer: the address46        // in the invitation is the one the receiver will dial for ever after.47        // Handing out Peach's relay while the user has asked for a private one48        // would push the same lie one layer down, into the mesh.49        val relayUrl = when (val decision = relaySettings.relayInForce()) {50            is RelayDecision.Refuse -> {51                trySend(RelayAdoptionEvent.Rejected(decision.reason.wire()))52                close()53                return@callbackFlow54            }5556            is RelayDecision.Dial -> decision.relayUrl57        }58        val token = Rendezvous.newPairingToken()59        val invitation = createPairingInvitation(relayUrl, token)6061        job = scope.launch(Dispatchers.IO) {62            try {63                val connected = OkHttpRelaySocket.connect(64                    Rendezvous.buildUrl(relayUrl, token),65                )66                socket = connected67                // The relay buffers nothing, so the invitation is not surfaced68                // until the source WebSocket is open and waiting in the room.69                trySend(RelayAdoptionEvent.InvitationReady(invitation))7071                val adopted = withTimeout(windowMs) {72                    runRelayAdoptionSourceSession(73                        socket = connected,74                        expectedToken = token,75                        relayUrl = relayUrl,76                        ourStaticPrivateKey = ourStaticPrivateKey,77                        ourStaticPublicKey = ourStaticPublicKey,78                        localIdentity = localIdentity,79                        sourceDeps = sourceDeps,80                        confirm = { sas, request ->81                            trySend(82                                RelayAdoptionEvent.AwaitingPeerConfirmation(83                                    sas = sas,84                                    deviceName = request.deviceName.ifBlank { "New device" },85                                ),86                            )87                            // Showing a fresh invitation from an unlocked vault88                            // is source-side authorization. The scanning device's89                            // explicit SAS confirmation is the single commit gate.90                            true91                        },92                    )93                }94                if (adopted == null) {95                    trySend(RelayAdoptionEvent.Rejected("adoption_rejected"))96                } else {97                    trySend(98                        RelayAdoptionEvent.Adopted(99                            adopted.deviceName?.takeIf(String::isNotBlank) ?: "New device",100                        ),101                    )102                }103            } catch (error: kotlinx.coroutines.TimeoutCancellationException) {104                trySend(RelayAdoptionEvent.Expired)105            } catch (error: kotlinx.coroutines.CancellationException) {106                // stop() cancels this job when the user leaves the screen or107                // restarts pairing. That is not a pairing failure — reporting it108                // as one raced the new session and replaced the freshly shown109                // code with "Couldn't pair: StandaloneCoroutine was cancelled".110                // Cancellation must propagate, never be swallowed.111                throw error112            } catch (error: Throwable) {113                trySend(RelayAdoptionEvent.Rejected(error.message ?: "relay_adoption_failed"))114            } finally {115                socket?.close()116                socket = null117                close()118            }119        }120121        awaitClose { stop() }122    }123124    fun stop() {125        job?.cancel()126        job = null127        socket?.close()128        socket = null129    }130}131132/**133 * Source half of first-contact adoption through an opaque relay splice.134 *135 * The relay token selects only a room. Authentication and confidentiality136 * begin with Noise NK; the root grant remains behind both SAS decisions in137 * [grantAdoption].138 */139internal suspend fun runRelayAdoptionSourceSession(140    socket: RelayBinarySocket,141    expectedToken: String,142    relayUrl: String,143    ourStaticPrivateKey: ByteArray,144    ourStaticPublicKey: ByteArray,145    localIdentity: LocalDeviceIdentity,146    sourceDeps: AdoptionSourceDeps,147    confirm: suspend (sas: String, request: AdoptionFrame.Request) -> Boolean,148): PairedDevice? {149    val hello = JSONObject(String(socket.receive(), Charsets.UTF_8))150    require(hello.optString("type") == "PEACH_HELLO") { "expected PEACH_HELLO" }151    require(152        MessageDigest.isEqual(153            expectedToken.toByteArray(Charsets.US_ASCII),154            hello.optString("token").toByteArray(Charsets.US_ASCII),155        ),156    ) { "pairing token mismatch" }157158    socket.send(159        JSONObject()160            .put("type", "PEACH_HELLO_ACK")161            .put(162                "ephemeralPublicKeyHex",163                ourStaticPublicKey.joinToString("") { "%02x".format(it) },164            )165            .toString()166            .toByteArray(Charsets.UTF_8),167    )168169    val noise = noiseNKResponder(170        ourStaticPrivateKey,171        ourStaticPublicKey,172        socket::send,173        socket::receive,174    )175    try {176        val channel = object : PairingTransport {177            override suspend fun send(msg: ByteArray) = socket.send(noise.encrypt(msg))178            override suspend fun receive(): ByteArray = noise.decrypt(socket.receive())179            override suspend fun close() = socket.close()180        }181182        val request = receiveFrame(channel)183        require(isAdoptRequest(request)) { "expected ADOPT_REQUEST" }184        request as AdoptionFrame.Request185        if (!confirm(SASDerivation.derive(noise.handshakeHash), request)) return null186187        val pairState = AdoptionPairState(188            rendezvousSecretHex = Rendezvous.bytesToHex(189                Rendezvous.deriveSecret(noise.handshakeHash),190            ),191            relayUrl = Rendezvous.normalizeRelayUrl(relayUrl),192        )193        val boundDeps = object : AdoptionSourceDeps {194            override val sourceIdentity: LocalDeviceIdentity = localIdentity195            override val pairState: AdoptionPairState = pairState196            override suspend fun rootKeyHex(): String = sourceDeps.rootKeyHex()197            override suspend fun vaultId(): String = sourceDeps.vaultId()198            override suspend fun license(): String? = sourceDeps.license()199            override suspend fun applyLicense(license: String) {200                sourceDeps.applyLicense(license)201            }202            override suspend fun onConsent(request: AdoptionFrame.Request): Boolean = true203            override suspend fun upsertPairedDevice(device: PairedDevice) {204                sourceDeps.upsertPairedDevice(device)205            }206            override suspend fun vaultSnapshot() = sourceDeps.vaultSnapshot()207            override suspend fun outgoingDeviceManifest() = sourceDeps.outgoingDeviceManifest()208            override suspend fun markSynced(deviceId: String) = sourceDeps.markSynced(deviceId)209            override suspend fun duressSlotJson() = sourceDeps.duressSlotJson()210        }211        val adopted = grantAdoption(channel, request, boundDeps) ?: return null212        // If the receiver advertised snapshot support, send the vault inline.213        if (request.snapshotCapable == true) {214            try {215                val snapshot = boundDeps.vaultSnapshot()216                if (snapshot != null) {217                    val duressSlot = runCatching { boundDeps.duressSlotJson() }.getOrNull()218                    sendVaultSnapshot(channel, snapshot.first, snapshot.second, duressSlot)219                    val applied = receiveFrame(channel)220                    if (applied is AdoptionFrame.SnapshotApplied) {221                        val manifest = boundDeps.outgoingDeviceManifest()222                        if (manifest != null) {223                            sendFrame(channel, AdoptionFrame.DeviceManifestFrame(manifest))224                            withTimeoutOrNull(5_000) { receiveFrame(channel) }225                        }226                        boundDeps.markSynced(adopted.deviceId)227                    }228                }229            } catch (error: Exception) {230                if (error is SnapshotDuressFrameTooLargeException) throw error231                if (error is kotlinx.coroutines.CancellationException) throw error232                Log.w("PeachAdopt", "source inline snapshot send failed, receiver falls back to sync", error)233            }234        }235        return adopted236    } finally {237        noise.close()238    }239}