CrossClientSyncProcessTest.kt

1package com.peachpasswords.sync23import java.io.BufferedReader4import kotlinx.serialization.json.Json5import java.io.BufferedWriter6import java.io.File7import java.io.InputStreamReader8import java.io.OutputStreamWriter9import java.net.InetSocketAddress10import java.net.ServerSocket11import java.net.Socket12import java.util.Base6413import kotlin.system.exitProcess14import kotlinx.coroutines.runBlocking15import org.junit.Assert.assertEquals16import org.junit.Assert.assertNotNull17import org.junit.Test1819/**20 * Process endpoint for scripts/check-cross-client-sync.mjs. The socket carries21 * one base64-encoded, production-padded SyncMessage frame per line; the22 * transcript is kept separately so the parent process can assert every ordered23 * wire type.24 *25 * Scenario shape is supplied entirely by the parent — this endpoint reads knobs26 * and does not know what a "scenario" is. Keeping the semantics in the harness27 * is what lets both clients be configured identically and lets a reviewer read28 * the matrix in one place.29 */30class CrossClientSyncProcessTest {31    @Test32    fun exchange_adoption_v2_with_browser_process() {33        val role = configured("PEACH_ADOPTION_HARNESS_ROLE") ?: return34        val portFile = File(requireConfigured("PEACH_SYNC_HARNESS_PORT_FILE"))35        val transcriptFile = File(requireConfigured("PEACH_SYNC_HARNESS_TRANSCRIPT_FILE"))36        val raw = if (role == "source") {37            ServerSocket(0).also { portFile.writeText(it.localPort.toString()) }38                .accept().also { portFile.delete() }39        } else {40            Socket().also {41                it.connect(InetSocketAddress("127.0.0.1", requireConfigured("PEACH_SYNC_HARNESS_PORT").toInt()), 5_000)42            }43        }44        raw.use { socket ->45            val relay = LineRelaySocket(socket)46            val root = ByteArray(32) { (it + 1).toByte() }47            val token = "AAECAwQFBgcICQoLDA0ODw"48            val keys = localSyncKeypair(randomLocalIdentityCredentials("android-source"))49            val identity = harnessIdentity("android", keys.publicKey)50            var pending = false51            var rootAfterPending = false52            var promoted = false53            var collisionChecks = 054            var rotations = 055            val adopted = runBlocking {56                if (role == "source") {57                    runRelayAdoptionSourceSession(58                        relay, token, "wss://relay.example.test", keys.secretKey, keys.publicKey, identity,59                        object : AdoptionSourceV2Deps {60                            override val sourceIdentity = identity61                            override suspend fun rootKeyHex(): String {62                                rootAfterPending = pending63                                return Rendezvous.bytesToHex(root)64                            }65                            override suspend fun vaultId() = computeVaultId(root)66                            override suspend fun license() = "PPK-cross-client"67                            override suspend fun onConsent(request: AdoptionFrame.Request) = true68                            override suspend fun onConsent(request: AdoptionFrame.RequestV2) = true69                            override suspend fun persistPendingIdentity(device: PairedDevice, nowMs: Long): Boolean {70                                pending = true71                                return true72                            }73                            override suspend fun completePendingIdentity(device: PairedDevice): Boolean {74                                promoted = true75                                return true76                            }77                            override suspend fun upsertPairedDevice(device: PairedDevice) = Unit78                            override fun hasDeviceIdentityCollision(device: PairedDevice) = collisionChecks++ == 079                            override suspend fun rotateSourceIdentity() = identity80                            override fun snapshotCapable() = false81                        },82                        confirm = { _, _ -> true },83                    )84                } else {85                    runRelayAdoptionReceiverSession(86                        relay,87                        PairingInvitation("wss://relay.example.test", token, noiseStaticPublicKeyHex = requireConfigured("PEACH_ADOPTION_HARNESS_STATIC")),88                        harnessReceiverDeps(identity, root) { rotations += 1 },89                        confirm = { true },90                    )91                }92            }93            transcriptFile.writeText(94                "{\"status\":\"${if (adopted != null) "completed" else "rejected"}\"," +95                    "\"pendingBeforeRoot\":$rootAfterPending,\"promotedAfterApplied\":$promoted,\"rotations\":$rotations}",96            )97            assertNotNull(adopted)98            if (role == "source") {99                assertEquals(true, rootAfterPending)100                assertEquals(true, promoted)101            }102        }103    }104105    private fun harnessIdentity(name: String, publicKey: ByteArray) = LocalDeviceIdentity(106        "$name-id", name, DeviceType.ANDROID, Rendezvous.bytesToHex(publicKey), "cd".repeat(32), 49142,107        "wss://relay.example.test",108    )109110    private fun harnessReceiverDeps(111        identity: LocalDeviceIdentity,112        root: ByteArray,113        onRotate: () -> Unit,114    ) = object : AdoptionReceiverV2Deps {115        override val deviceName = "android"116        override val deviceType = DeviceType.ANDROID117        override fun generateDeviceId() = identity.deviceId118        override fun localIdentity() = identity119        override suspend fun rotateLocalIdentity() = identity.copy(deviceId = "android-rotated").also { onRotate() }120        override suspend fun confirmSas() = true121        override suspend fun validateSource(prepare: AdoptionFrame.Prepare) = Unit122        override suspend fun hasSourceIdentityCollision(prepare: AdoptionFrame.Prepare) = false123        override suspend fun createVaultFromRootKey(rootKeyHex: String) = computeVaultId(hexToPubkey(rootKeyHex))124        override suspend fun deriveLocalIdentity() = identity125        override suspend fun upsertPairedDevice(device: PairedDevice) = Unit126        override fun onError(error: String) = throw AssertionError(error)127        override suspend fun license() = "PPK-cross-client"128    }129130    private class LineRelaySocket(socket: Socket) : RelayBinarySocket {131        private val input = BufferedReader(InputStreamReader(socket.getInputStream(), Charsets.UTF_8))132        private val output = BufferedWriter(OutputStreamWriter(socket.getOutputStream(), Charsets.UTF_8))133        override fun send(frame: ByteArray) {134            output.write(Base64.getEncoder().encodeToString(frame))135            output.newLine()136            output.flush()137        }138        override suspend fun receive(): ByteArray = Base64.getDecoder().decode(input.readLine() ?: error("cross-client peer closed"))139        override fun close(code: Int, reason: String) = Unit140    }141142    @Test143    fun exchange_production_frames_with_browser_process() {144        val mode = configured("PEACH_SYNC_HARNESS_MODE") ?: return145        val portFile = File(requireConfigured("PEACH_SYNC_HARNESS_PORT_FILE"))146        val transcriptFile = File(requireConfigured("PEACH_SYNC_HARNESS_TRANSCRIPT_FILE"))147        val config = HarnessConfig.read(if (mode == "initiator") "android" else "browser", ::configured)148        val socket = if (mode == "initiator") {149            ServerSocket(0).also {150                portFile.writeText(it.localPort.toString())151            }.accept().also { portFile.delete() }152        } else {153            Socket().also {154                it.connect(InetSocketAddress("127.0.0.1", requireConfigured("PEACH_SYNC_HARNESS_PORT").toInt()), 5_000)155            }156        }157158        socket.use { raw ->159            val input = BufferedReader(InputStreamReader(raw.getInputStream(), Charsets.UTF_8))160            val output = BufferedWriter(OutputStreamWriter(raw.getOutputStream(), Charsets.UTF_8))161            val transcript = mutableListOf<String>()162            val vault = ProcessVault(config)163            // Simulated process death: the transcript is on disk before the164            // process goes, so the parent can still say what the peer had seen.165            val die: () -> Nothing = {166                transcriptFile.writeText(renderTranscript("DIED", null, vault, transcript))167                exitProcess(7)168            }169            val session = SyncSession(170                remoteStaticPublicKey = ByteArray(32) { 7 },171                send = { bytes ->172                    val message = decodeSyncMessage(bytes)173                    val wireMessage = if (config.mismatch && message is SyncMessage.Manifest) {174                        SyncMessage.Request(emptyList())175                    } else {176                        message177                    }178                    transcript += "send:${messageType(wireMessage)}"179                    if (config.stopAfter == messageType(wireMessage)) die()180                    val padded = SyncPadding.pad(wireMessage.encodeToBytes())181                    output.write(Base64.getEncoder().encodeToString(padded))182                    output.newLine()183                    output.flush()184                },185                receive = {186                    val line = input.readLine() ?: error("cross-client peer closed")187                    val payload = SyncPadding.unpad(Base64.getDecoder().decode(line))188                    transcript += "recv:${messageType(decodeSyncMessage(payload))}"189                    payload190                },191                close = {},192            )193            val outcome = runBlocking {194                if (mode == "initiator") {195                    runSyncInitiator(session, vault, nowMs = System.currentTimeMillis())196                } else {197                    runSyncResponder(session, vault, nowMs = System.currentTimeMillis())198                }199            }200            // A clean stop is only deterministic if our socket is still up when201            // the peer writes its last frame, so hold it open until the peer202            // goes away rather than exiting the instant the round ends.203            if (config.drain) {204                runCatching {205                    raw.soTimeout = 2_000206                    while (input.readLine() != null) Unit207                }208            }209            transcriptFile.writeText(renderTranscript(outcome.status.name, outcome, vault, transcript))210            assertEquals(211                "Android process sync outcome",212                config.expectStatus,213                outcome.status,214            )215        }216    }217218    private fun craftedManifest(config: HarnessConfig): SignedDeviceManifest? =219        config.signedManifestJson?.let {220            json.decodeFromString(SignedDeviceManifest.serializer(), it)221        }222223    /**224     * The production verdict on the peer's manifest. The trusted key stands in225     * for this device's stored issuer key, and every known issuer resolves to226     * it, so a revocation signed by another key fails closed.227     */228    private fun manifestVerdict(config: HarnessConfig, manifest: SignedDeviceManifest, nowMs: Long): String {229        val trustedHex = config.trustedPubkeyHex ?: return "not-checked"230        val accepted = verifyDeviceManifest(231            manifest = manifest,232            issuerSigningPublicKey = hexToPubkey(trustedHex),233            resolveIssuerSigningKeys = { setOf(trustedHex.lowercase()) },234            nowMs = nowMs,235        )236        return if (accepted != null) "accepted" else "rejected"237    }238239    private fun renderTranscript(240        status: String,241        outcome: SyncOutcome?,242        vault: ProcessVault,243        transcript: List<String>,244    ): String = buildString {245        append("{\"status\":\"").append(status).append("\"")246        outcome?.error?.let { append(",\"error\":\"").append(it.replace("\"", "\\\"")).append("\"") }247        append(",\"entriesSent\":").append(outcome?.entriesSent ?: 0)248        append(",\"entriesReceived\":").append(outcome?.entriesReceived ?: 0)249        append(",\"applied\":").append(vault.appliedCount)250        append(",\"staged\":").append(vault.stagedCount)251        append(",\"manifestAccepted\":").append(252            vault.manifestVerdict?.let { it == "accepted" } ?: "null",253        )254        append(",\"events\":[")255        transcript.forEachIndexed { index, event ->256            if (index > 0) append(',')257            append('"').append(event).append('"')258        }259        append("]}")260    }261262    private val json = Json { ignoreUnknownKeys = true }263264    private fun configured(name: String): String? =265        System.getenv(name) ?: System.getProperty(name.lowercase().replace('_', '.'))266267    private fun requireConfigured(name: String): String =268        configured(name) ?: error("missing $name")269270    private fun messageType(message: SyncMessage): String = when (message) {271        is SyncMessage.Hello -> "SYNC_HELLO"272        is SyncMessage.Manifest -> "SYNC_MANIFEST"273        is SyncMessage.Request -> "SYNC_REQUEST"274        is SyncMessage.Entries -> "SYNC_ENTRIES"275        is SyncMessage.Tombstones -> "SYNC_TOMBSTONES"276        is SyncMessage.DeviceManifest -> "SYNC_DEVICE_MANIFEST"277        is SyncMessage.Done -> "SYNC_DONE"278        is SyncMessage.Error -> "SYNC_ERROR"279    }280281    /** Every knob the matrix can set, in one place. Defaults reproduce a plain round. */282    private class HarnessConfig(283        val id: String,284        val parity: String,285        val localVersion: Long,286        val vaultId: String,287        val manifestSupport: Boolean,288        val entryCount: Int,289        val expectStatus: SyncOutcomeStatus,290        val drain: Boolean,291        val mismatch: Boolean,292        val locked: Boolean,293        val stopAfter: String?,294        /**295         * A manifest the harness signed: the wire shape IS the scenario, and one296         * signer keeps both clients verifying the same bytes rather than each297         * signing its own.298         */299        val signedManifestJson: String?,300        val trustedPubkeyHex: String?,301    ) {302        companion object {303            fun read(id: String, configured: (String) -> String?) = HarnessConfig(304                id = id,305                parity = configured("PEACH_SYNC_HARNESS_MTIME_PARITY") ?: "",306                localVersion = configured("PEACH_SYNC_HARNESS_LOCAL_VERSION")?.toLongOrNull() ?: 7L,307                vaultId = configured("PEACH_SYNC_HARNESS_VAULT_ID") ?: "cross-client-vault",308                manifestSupport = configured("PEACH_SYNC_HARNESS_MANIFEST_VERSION") != "3",309                entryCount = configured("PEACH_SYNC_HARNESS_ENTRY_COUNT")?.toIntOrNull() ?: 1,310                expectStatus = SyncOutcomeStatus.valueOf(311                    configured("PEACH_SYNC_HARNESS_EXPECT_STATUS") ?: "COMPLETED",312                ),313                drain = configured("PEACH_SYNC_HARNESS_DRAIN") == "1",314                mismatch = configured("PEACH_SYNC_HARNESS_MISMATCH") == "message-order",315                // A locked session may hold the transport key and stage316                // ciphertext, but it never authors or applies a device manifest317                // and never merges entries into the live vault (PB-26-112 C).318                locked = configured("PEACH_SYNC_HARNESS_LOCKED") == "1",319                stopAfter = configured("PEACH_SYNC_HARNESS_STOP_AFTER")?.takeIf(String::isNotEmpty),320                signedManifestJson = configured("PEACH_SYNC_HARNESS_SIGNED_MANIFEST")?.takeIf(String::isNotEmpty),321                trustedPubkeyHex = configured("PEACH_SYNC_HARNESS_TRUSTED_PUBKEY")?.takeIf(String::isNotEmpty),322            )323        }324    }325326    private inner class ProcessVault(private val config: HarnessConfig) : SyncVaultAdapter {327        private val id = config.id328        // Entries are shared ids so a diff can actually flow. Parity decides329        // which side is newer for a given index: the same parity on both sides330        // means equal data versions, opposite parities means entries flow both331        // ways. Must stay identical to the Browser endpoint's rule.332        private val ids = (0 until config.entryCount).map { "shared-entry-$it" }333        private val entries = linkedMapOf<String, Long>().apply {334            ids.forEachIndexed { index, entryId -> put(entryId, mtimeFor(index)) }335        }336        private val blobs = linkedMapOf<String, String>().apply {337            ids.forEachIndexed { index, entryId -> put(entryId, "$id-frame-${index % 8}") }338        }339340        private fun mtimeFor(index: Int): Long {341            if (config.parity.isEmpty()) return 10L342            return if (index % 2 == config.parity.toInt()) 15L else 10L343        }344        private val tombstones = emptyMap<String, Long>()345        private val manifest = SignedDeviceManifest(346            DeviceManifestBody(DEVICE_MANIFEST_VERSION, id, 1_700_000_000_000L, emptyList()),347            "11".repeat(64),348        )349350        /** The production verdict on the peer's manifest, or null if none arrived. */351        var manifestVerdict: String? = null352            private set353354        private val nowMs = System.currentTimeMillis()355        private val crafted: SignedDeviceManifest? = craftedManifest(config)356357        /** Entries this side merged into its live vault. */358        var appliedCount = 0359            private set360361        /** Entries this side only staged because the vault is locked. */362        var stagedCount = 0363            private set364365        private val staged = linkedMapOf<String, String>()366367        override suspend fun computeVaultId() = config.vaultId368        override suspend fun getOurDeviceId() = id369        override suspend fun getLocalVersion() = config.localVersion370        override suspend fun buildEntryManifest() = entries.toMap()371        override suspend fun buildTombstoneManifest() = tombstones372373        override suspend fun getEncryptedEntries(entryIds: List<String>) =374            entryIds.mapNotNull { key -> blobs[key]?.let { key to it } }.toMap()375376        override suspend fun applyEntries(entries: Map<String, String>, remoteManifest: Map<String, Long>) {377            if (config.locked) {378                staged.putAll(entries)379                stagedCount += entries.size380                return381            }382            appliedCount += entries.size383            this.entries.putAll(entries.mapValues { remoteManifest[it.key] ?: 0L })384            blobs.putAll(entries)385        }386387        override suspend fun applyTombstones(tombstones: Map<String, Long>) = Unit388389        override suspend fun buildOutgoingDeviceManifest() =390            if (!config.manifestSupport || config.locked) null else crafted ?: manifest391392        override suspend fun applyIncomingDeviceManifest(393            manifest: SignedDeviceManifest,394            authenticatedRemoteStaticPublicKey: ByteArray,395        ) {396            manifestVerdict = manifestVerdict(config, manifest, nowMs)397        }398399        override val supportsDeviceManifest = config.manifestSupport && !config.locked400    }401}