PairingFrames.kt
1package com.peachpasswords.sync23import kotlinx.coroutines.Dispatchers4import kotlinx.coroutines.channels.Channel5import kotlinx.coroutines.withContext6import java.io.DataInputStream7import java.io.EOFException8import java.io.IOException9import java.io.InputStream10import java.io.OutputStream1112/**13 * Length-delimited frame IO for a pairing/sync connection.14 *15 * Originally declared alongside the Chirp audio pairing that has since been16 * removed, but it was never audio-specific — it is just "send me a frame, give17 * me a frame", which the Noise NK pipeline and the adoption flow both build on.18 * It lives here now, with the frame helpers that use it.19 */20interface PairingTransport {21 suspend fun send(msg: ByteArray)22 suspend fun receive(): ByteArray23 suspend fun close()24}2526/**27 * Length-prefixed framing for the C7 TCP transport, shared by [PairingServer]28 * and [PairingTransport] so neither duplicates the wire format.29 *30 * Frame = [4-byte big-endian length][payload]. This is the framing the existing31 * PairingFrames.runPairingInitiator/Responder already assume implicitly when they32 * call `transport.send(bytes)` / `transport.receive()` — the WebSocket transport33 * gets framing for free from the protocol; a raw TCP socket does not, so we add34 * it here.35 *36 * A frame length cap guards against a hostile/garbled peer asking us to allocate37 * gigabytes before the Noise layer has authenticated anything.38 */39const val MAX_PAIRING_FRAME_BYTES: Int = 1 shl 20 // 1 MiB4041/**42 * PB-26-205: relay pairing/adoption inbox backstop, mirrored from the43 * browser (`pairing-dial.ts`). The 1 MiB frame ceiling clears the 768 KiB44 * adoption frame ceiling with headroom; the 12 MiB / 256-message inbox45 * clears the 10 MiB aggregate adoption budget plus handshake/control46 * overhead. Both clients agree on accept/reject at these boundaries.47 */48const val MAX_PAIRING_INBOX_BYTES: Int = 12 * 1024 * 102449const val MAX_PAIRING_INBOX_MESSAGES: Int = 2565051/** Terminal pairing error: an IOException so it lands in the ERRORED outcome like a transport failure. */52class PairingResourceLimitException(53 message: String = "pairing_resource_limit",54) : IOException(message)5556/**57 * Bounded relay inbox for one pairing/adoption session. The size check runs58 * before [read] performs the allocate/copy, so an oversized frame never59 * materializes. On any violation the channel closes with60 * [PairingResourceLimitException] — every parked and future reader gets that61 * one terminal error — [onLimitExceeded] fires once (the owner closes the62 * socket with status 1009), and later frames are ignored.63 */64internal class RelayPairingInbox(private val onLimitExceeded: () -> Unit) {65 private val channel = Channel<ByteArray>(Channel.UNLIMITED)66 private var queuedBytes = 067 private var queuedMessages = 068 private var terminal: Throwable? = null6970 @Synchronized71 fun onFrame(size: Int, read: () -> ByteArray): Boolean {72 if (terminal != null) return false73 if (74 size > MAX_PAIRING_FRAME_BYTES ||75 queuedMessages + 1 > MAX_PAIRING_INBOX_MESSAGES ||76 queuedBytes + size > MAX_PAIRING_INBOX_BYTES77 ) {78 tripLimit()79 return false80 }81 val frame = read()82 queuedBytes += frame.size83 queuedMessages += 184 channel.trySend(frame)85 return true86 }8788 /** Trips the limit for a non-frame violation (e.g. a text frame). */89 @Synchronized90 fun reject(): Boolean {91 if (terminal != null) return false92 tripLimit()93 return false94 }9596 /** Benign socket close/failure: fails parked readers without a limit close. */97 @Synchronized98 fun close(cause: Throwable) {99 if (terminal != null) return100 terminal = cause101 channel.close(cause)102 }103104 private fun tripLimit() {105 terminal = PairingResourceLimitException()106 // A closed channel still delivers buffered elements first, so drop107 // the queued refs for parity with the browser's cleared inbox.108 while (channel.tryReceive().isSuccess) { /* drop */ }109 queuedBytes = 0110 queuedMessages = 0111 channel.close(terminal)112 onLimitExceeded()113 }114115 suspend fun receive(): ByteArray {116 val frame = channel.receive()117 synchronized(this) {118 queuedMessages -= 1119 queuedBytes -= frame.size120 }121 return frame122 }123}124125/** Blocking write of a single length-prefixed frame. */126fun writeFrame(out: OutputStream, payload: ByteArray) {127 require(payload.size <= MAX_PAIRING_FRAME_BYTES) { "frame too large: ${payload.size}" }128 val header = ByteArray(4)129 header[0] = (payload.size ushr 24).toByte()130 header[1] = (payload.size ushr 16).toByte()131 header[2] = (payload.size ushr 8).toByte()132 header[3] = payload.size.toByte()133 out.write(header)134 out.write(payload)135 out.flush()136}137138/** Blocking read of a single length-prefixed frame. Throws on EOF / oversize. */139fun readFrame(input: InputStream): ByteArray {140 val data = if (input is DataInputStream) input else DataInputStream(input)141 val length = try {142 data.readInt()143 } catch (e: EOFException) {144 throw java.io.IOException("peer closed before frame header", e)145 }146 require(length in 0..MAX_PAIRING_FRAME_BYTES) { "invalid frame length: $length" }147 val buf = ByteArray(length)148 data.readFully(buf)149 return buf150}151152/**153 * Adapts a blocking TCP socket pair to the suspend send/receive shape the REAL154 * crypto stack expects:155 * - noiseNKInitiator / noiseNKResponder take `send: suspend (ByteArray)->Unit`156 * and `receive: suspend ()->ByteArray` (see NoiseNK.kt).157 * - PairingFrames's [PairingTransport] interface is `send/receive/close`.158 *159 * Both the Noise handshake messages and the post-handshake encrypted payloads160 * travel as length-prefixed frames over the same socket, so one [SocketFrameIO]161 * serves the whole session. All IO hops to Dispatchers.IO.162 *163 * This implements the EXISTING [PairingTransport] interface (PairingFrames.kt),164 * which is the contract `runPairingInitiator` / `runPairingResponder` consume —165 * so a TCP socket can drive the already-written handshake unchanged.166 */167class SocketFrameIO(168 private val input: InputStream,169 private val output: OutputStream,170 private val onClose: suspend () -> Unit = {},171) : PairingTransport {172 private val buffered = DataInputStream(input)173174 /**175 * Synchronous frame write. [SyncSession.send] is non-suspending (see176 * SyncTransport.kt), so the sync-exchange bridge needs a blocking send177 * that does NOT wrap a coroutine. Socket writes are bounded by soTimeout.178 */179 fun sendBlocking(msg: ByteArray) {180 writeFrame(output, msg)181 }182183 override suspend fun send(msg: ByteArray) = withContext(Dispatchers.IO) {184 writeFrame(output, msg)185 }186187 override suspend fun receive(): ByteArray = withContext(Dispatchers.IO) {188 readFrame(buffered)189 }190191 override suspend fun close() {192 onClose()193 }194}195196/**197 * Convenience wrappers for the post-handshake encrypted exchange. The Noise198 * [NoiseNKSession] already owns the cipher (encrypt/decrypt); these just splice199 * it onto the frame layer so call sites read clearly.200 *201 * NOTE: the existing PairingFrames handshake calls `session.encrypt` /202 * `session.decrypt` directly around `transport.send` / `transport.receive`; the203 * post-pairing SYNC exchange (runSyncInitiator/Responder) does the same via204 * a [SyncSession]. These helpers exist for the raw-frame paths in205 * PairingServer/PairingTransport that don't go through a SyncSession.206 */207suspend fun PairingTransport.sendEncrypted(session: NoiseNKSession, plaintext: ByteArray) {208 send(session.encrypt(plaintext))209}210211suspend fun PairingTransport.receiveEncrypted(session: NoiseNKSession): ByteArray {212 return session.decrypt(receive())213}