RelayAdoptionHost.kt

1package com.peachpasswords.sync23import java.security.MessageDigest4import javax.inject.Inject5import javax.inject.Singleton6import kotlinx.coroutines.CoroutineScope7import kotlinx.coroutines.CompletableDeferred8import 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    private val generation: SyncGenerationPort,34    private val deviceStore: PairedDeviceRegistry,35) {36    private var job: Job? = null37    private var socket: RelayBinarySocket? = null38    private val consent = SourceConsentGate()3940    /**41     * The vault owner's answer to the source-side SAS prompt. §10.2 requires a42     * human decision on the device that hands over the root — without this the43     * grant was sent as soon as the code had been drawn (PB-26-116).44     */45    fun confirmSas(matches: Boolean) = consent.resolve(matches)4647    fun start(48        scope: CoroutineScope,49        ourStaticPrivateKey: ByteArray,50        ourStaticPublicKey: ByteArray,51        localIdentity: LocalDeviceIdentity,52        sourceDeps: AdoptionSourceV2Deps,53        windowMs: Long = PAIRING_WINDOW_MS,54    ): Flow<RelayAdoptionEvent> = callbackFlow {55        // Pairing is also how this mesh's relay reaches a new peer: the address56        // in the invitation is the one the receiver will dial for ever after.57        // Handing out Peach's relay while the user has asked for a private one58        // would push the same lie one layer down, into the mesh.59        val relayUrl = when (val decision = relaySettings.relayInForce()) {60            is RelayDecision.Refuse -> {61                trySend(RelayAdoptionEvent.Rejected(decision.reason.wire()))62                close()63                return@callbackFlow64            }6566            is RelayDecision.Dial -> decision.relayUrl67        }68        val token = Rendezvous.newPairingToken()69        val invitation = createPairingInvitation(70            relayUrl,71            token,72            noiseStaticPublicKeyHex = Rendezvous.bytesToHex(ourStaticPublicKey),73        )7475        job = scope.launch(Dispatchers.IO) {76            try {77                val connected = OkHttpRelaySocket.connectPairing(78                    Rendezvous.buildUrl(relayUrl, token),79                )80                socket = connected81                // The relay buffers nothing, so the invitation is not surfaced82                // until the source WebSocket is open and waiting in the room.83                trySend(RelayAdoptionEvent.InvitationReady(invitation))8485                var peerOutdated = false86                val adopted = withTimeout(windowMs) {87                    runRelayAdoptionSourceSession(88                        socket = connected,89                        expectedToken = token,90                        relayUrl = relayUrl,91                        ourStaticPrivateKey = ourStaticPrivateKey,92                        ourStaticPublicKey = ourStaticPublicKey,93                        localIdentity = localIdentity,94                        sourceDeps = sourceDeps,95                        // §11.6: once this device has reset its generation, a v196                        // peer is repair-required exactly as an outdated sync97                        // peer is. The invitation still goes out; the refusal98                        // happens at the first frame.99                        postReset = { generation.isPostReset(System.currentTimeMillis()) },100                        onPeerOutdated = { peerOutdated = true },101                        confirm = { sas, request ->102                            trySend(103                                RelayAdoptionEvent.AwaitingPeerConfirmation(104                                    sas = sas,105                                    deviceName = capDeviceName(request.deviceName.ifBlank { "New device" }),106                                ),107                            )108                            // §10.2: the device that owns the vault decides.109                            // Cancellation (screen exit, stop(), window expiry)110                            // ends the await without a grant, which is the111                            // fail-closed outcome.112                            consent.await()113                        },114                    )115                }116                if (peerOutdated) {117                    trySend(RelayAdoptionEvent.Rejected(PEER_PROTOCOL_OUTDATED))118                } else if (adopted == null) {119                    trySend(RelayAdoptionEvent.Rejected("adoption_rejected"))120                } else {121                    onDirectV2PairingEstablished(122                        generation,123                        hasDirectDevice = deviceStore.getAll().any { it.trustSource == "direct" },124                        nowMs = System.currentTimeMillis(),125                    )126                    trySend(127                        RelayAdoptionEvent.Adopted(128                            capDeviceName(adopted.deviceName?.takeIf(String::isNotBlank) ?: "New device"),129                        ),130                    )131                }132            } catch (error: kotlinx.coroutines.TimeoutCancellationException) {133                trySend(RelayAdoptionEvent.Expired)134            } catch (error: kotlinx.coroutines.CancellationException) {135                // stop() cancels this job when the user leaves the screen or136                // restarts pairing. That is not a pairing failure — reporting it137                // as one raced the new session and replaced the freshly shown138                // code with "Couldn't pair: StandaloneCoroutine was cancelled".139                // Cancellation must propagate, never be swallowed.140                throw error141            } catch (error: Throwable) {142                trySend(RelayAdoptionEvent.Rejected(error.message ?: "relay_adoption_failed"))143            } finally {144                socket?.close()145                socket = null146                close()147            }148        }149150        awaitClose { stop() }151    }152153    fun stop() {154        // Leaving the screen is a denial, not a silent timeout: an unanswered155        // source prompt must never leave the session waiting on a grant.156        consent.resolve(false)157        job?.cancel()158        job = null159        socket?.close()160        socket = null161    }162}163164/**165 * The source-side consent gate.166 *167 * §10.2 puts a human decision on the device that owns the vault: the root must168 * not be read until someone there approves the SAS the other device is showing.169 * Extracted as a seam so the await/resolve contract is provable without a relay170 * socket — a callback that returned true immediately is exactly the defect this171 * gate exists to prevent (PB-26-116).172 */173internal class SourceConsentGate {174    private val lock = Any()175    private var pending: CompletableDeferred<Boolean>? = null176177    /** Suspend until [resolve]. A gate with no pending await resolves nothing. */178    suspend fun await(): Boolean {179        val deferred = synchronized(lock) {180            CompletableDeferred<Boolean>().also { pending = it }181        }182        return try {183            deferred.await()184        } finally {185            synchronized(lock) { if (pending === deferred) pending = null }186        }187    }188189    /** Answer the live prompt. A second call, or one with no prompt, is a no-op. */190    fun resolve(matches: Boolean) {191        val deferred = synchronized(lock) { pending.also { pending = null } } ?: return192        deferred.complete(matches)193    }194}195196/**197 * Source half of first-contact adoption through an opaque relay splice.198 *199 * The relay token selects only a room. Authentication and confidentiality200 * begin with Noise NK; the root grant remains behind both SAS decisions in201 * [grantAdoption].202 */203internal suspend fun runRelayAdoptionSourceSession(204    socket: RelayBinarySocket,205    expectedToken: String,206    relayUrl: String,207    ourStaticPrivateKey: ByteArray,208    ourStaticPublicKey: ByteArray,209    localIdentity: LocalDeviceIdentity,210    sourceDeps: AdoptionSourceV2Deps,211    confirm: suspend (sas: String, request: AdoptionFrame.Request) -> Boolean,212    postReset: () -> Boolean = { false },213    onPeerOutdated: () -> Unit = {},214): PairedDevice? {215    val hello = JSONObject(String(socket.receive(), Charsets.UTF_8))216    require(hello.optString("type") == "PEACH_HELLO") { "expected PEACH_HELLO" }217    require(218        MessageDigest.isEqual(219            expectedToken.toByteArray(Charsets.US_ASCII),220            hello.optString("token").toByteArray(Charsets.US_ASCII),221        ),222    ) { "pairing token mismatch" }223224    socket.send(225        JSONObject()226            .put("type", "PEACH_HELLO_ACK")227            .put(228                "ephemeralPublicKeyHex",229                ourStaticPublicKey.joinToString("") { "%02x".format(it) },230            )231            .toString()232            .toByteArray(Charsets.UTF_8),233    )234235    val noise = noiseNKResponder(236        ourStaticPrivateKey,237        ourStaticPublicKey,238        socket::send,239        socket::receive,240    )241    try {242        val channel = object : PairingTransport {243            override suspend fun send(msg: ByteArray) = socket.send(noise.encrypt(msg))244            override suspend fun receive(): ByteArray = noise.decrypt(socket.receive())245            override suspend fun close() = socket.close()246        }247248        val request = receiveFrame(channel)249        require(request is AdoptionFrame.Request || request is AdoptionFrame.RequestV2) { "expected ADOPT_REQUEST" }250        // §11.6: refused before any SAS prompt and before any grant. §10.3251        // freezes the wire reason set to the two v1 peers already understand, so252        // this is a local refusal rather than a new ADOPT_REJECT reason.253        if (request is AdoptionFrame.Request && !allowLegacyV1(postReset())) {254            onPeerOutdated()255            return null256        }257        if (request is AdoptionFrame.Request && !confirm(SASDerivation.derive(noise.handshakeHash), request)) return null258259        val pairState = AdoptionPairState(260            rendezvousSecretHex = Rendezvous.bytesToHex(261                Rendezvous.deriveSecret(noise.handshakeHash),262            ),263            relayUrl = Rendezvous.normalizeRelayUrl(relayUrl),264        )265        // §10.5: the whole forwarding surface lives in one binder so a dropped266        // callback is caught by AdoptionAdapterCoverageTest rather than by a peer.267        val boundDeps = bindSourceSessionDeps(268            sourceDeps = sourceDeps,269            sourceIdentity = localIdentity,270            pairState = pairState,271            sas = SASDerivation.derive(noise.handshakeHash),272            snapshotCapable = sourceDeps.snapshotCapable(),273            confirm = confirm,274        )275        val adopted = when (request) {276            is AdoptionFrame.RequestV2 -> grantAdoptionV2(channel, request, boundDeps)277            is AdoptionFrame.Request -> grantAdoption(channel, request, boundDeps)278            else -> null279        } ?: return null280        // If the receiver advertised snapshot support, send the vault inline.281        val receiverWantsSnapshot = when (request) {282            is AdoptionFrame.Request -> request.snapshotCapable == true283            is AdoptionFrame.RequestV2 -> request.snapshotCapable == true284            else -> false285        }286        if (receiverWantsSnapshot) {287            try {288                val snapshot = boundDeps.vaultSnapshot()289                if (snapshot != null) {290                    val duressSlot = runCatching { boundDeps.duressSlotJson() }.getOrNull()291                    sendVaultSnapshot(channel, snapshot.first, snapshot.second, duressSlot)292                    val applied = receiveFrame(channel)293                    if (applied is AdoptionFrame.SnapshotApplied) {294                        val manifest = boundDeps.outgoingDeviceManifest()295                        if (manifest != null) {296                            sendFrame(channel, AdoptionFrame.DeviceManifestFrame(manifest))297                            withTimeoutOrNull(5_000) { receiveFrame(channel) }298                        }299                        boundDeps.markSynced(adopted.deviceId)300                    }301                }302            } catch (error: Exception) {303                if (error is SnapshotDuressFrameTooLargeException) throw error304                if (error is kotlinx.coroutines.CancellationException) throw error305                PeachSyncLog.w("adoption: snapshot send failed — reason=${relayLogReason(error)}")306            }307        }308        return adopted309    } finally {310        noise.close()311    }312}