diff --git a/docs/design/torrent-v2-integrity.md b/docs/design/torrent-v2-integrity.md index 0b83ab683..877fe67a0 100644 --- a/docs/design/torrent-v2-integrity.md +++ b/docs/design/torrent-v2-integrity.md @@ -343,6 +343,21 @@ boundaries. They do not perform v2 handshake negotiation or independent-client i The full runtime actor, handshake routing, timer wiring, reply serving and payload integration are still required before this adapter becomes a supported torrent capability. -The inherited generic 64 KiB frame ceiling is sufficient for these bounded hash messages but not -for a desktop-profile one-million-piece bitfield (125,000 bytes plus its message ID). Runtime -integration must make ordinary frame bounds profile-aware; this slice does not relax that target. +Trusted piece-count frame bounds now permit the desktop-profile one-million-piece bitfield +(125,000 bytes plus its message ID). Unrelated messages retain the ordinary 64 KiB ceiling. + + +### Piece-count-aware frame bounds + +`PeerFrameLimits` derives exact bitfield size from a known piece count up to one million. Generic +peer decoding defaults to the existing 64 KiB ceiling when no count is known. With a known count, +only bitfield frames can exceed that ceiling; extended, unknown and hash messages retain it. +Both peer readers inspect the ID and exact bitfield length before reading the remaining body. +Spare bits and piece indices are validated against the count, and availability state supports the +same one-million-piece ceiling. V1 metadata supplies its own count; v2 callers supply the authenticated +layout count without synthesizing SHA-1 hashes. + +The hash transport reserves all frame/decode credit for the larger body before reading it and +retains that credit through dispatch. This removes the wire-size obstacle to the desktop target. +Aggregate session/peer admission, complete v2 runtime integration and measured production resource +gates still need verification; accepting a large bitfield alone does not prove those gates. diff --git a/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerFrameLimits.kt b/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerFrameLimits.kt new file mode 100644 index 000000000..c630b56cd --- /dev/null +++ b/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerFrameLimits.kt @@ -0,0 +1,33 @@ +package com.linroid.ketch.torrent + +/** Trusted piece-count bounds extend only bitfields beyond the ordinary peer-frame ceiling. */ +internal class PeerFrameLimits(val pieceCount: Int? = null) { + init { require(pieceCount == null || pieceCount in 0..1_000_000) } + val bitfieldBytes: Int? = pieceCount?.let { (it + 7) / 8 } + val maximum: Int = maxOf(PeerWire.MAX_FRAME_SIZE, (bitfieldBytes ?: 0) + 1) + + fun validateSize(size: Int) { + require(size in 0..maximum) { "Peer frame exceeds limit" } + } + + /** Called after the one-byte ID, before allocating/reading the remaining body. */ + fun validateType(size: Int, id: Int) { + validateSize(size) + require(size <= PeerWire.MAX_FRAME_SIZE || id == 5) { "Oversized non-bitfield frame" } + if (id == 5 && bitfieldBytes != null) { + require(size == bitfieldBytes + 1) { "Wrong bitfield size" } + } + } + + fun validateBitfield(bytes: ByteArray) { + if (pieceCount == null) { + require(bytes.size < PeerWire.MAX_FRAME_SIZE) + return + } + require(bytes.size == bitfieldBytes) { "Wrong bitfield size" } + val spare = (8 - pieceCount % 8) % 8 + if (spare != 0) { + require(bytes.last().toInt() and ((1 shl spare) - 1) == 0) { "Nonzero bitfield padding" } + } + } +} diff --git a/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerHashTransport.kt b/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerHashTransport.kt index 77795ce9b..2d9c3e4dd 100644 --- a/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerHashTransport.kt +++ b/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerHashTransport.kt @@ -15,6 +15,7 @@ internal class PeerHashTransport( private val exchange: PeerHashExchange, private val frames: TorrentBufferBudget, private val timeoutMs: Long = 180_000, + private val pieceCount: Int? = null, ) { class Frame internal constructor( val message: PeerMessage, @@ -29,6 +30,7 @@ internal class PeerHashTransport( data class Verified(val result: PeerHashExchange.Verified) : Event } + private val limits = PeerFrameLimits(pieceCount) private val reads = Mutex() private var closed = false @@ -63,10 +65,16 @@ internal class PeerHashTransport( try { val message = withTimeout(timeoutMs) { val size = Buffer().write(connection.readExactly(4)).readInt() - require(size in 0..PeerWire.MAX_FRAME_SIZE) { "Peer frame exceeds limit" } + limits.validateSize(size) + val id = if (size == 0) null else connection.readExactly(1).single().toInt() and 255 + if (id != null) limits.validateType(size, id) // Covers raw body, generic decode, hash decode and dispatch copies, plus small headers. lease = checkNotNull(frames.reserve(size * 4 + 512)) { "Peer frame budget exhausted" } - PeerWire.decode(connection.readExactly(size)) + if (id == null) PeerMessage.KeepAlive else { + val payload = Buffer().writeByte(id).write(connection.readExactly(size - 1)) + .readByteArray() + PeerWire.decode(payload, pieceCount = pieceCount) + } } Frame(message, checkNotNull(lease)).also { lease = null } } catch (error: Throwable) { diff --git a/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerProtocolState.kt b/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerProtocolState.kt index 66e9547be..e0afe3b6b 100644 --- a/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerProtocolState.kt +++ b/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerProtocolState.kt @@ -3,7 +3,7 @@ package com.linroid.ketch.torrent /** Per-connection availability and request ownership, independent of the swarm scheduler. */ internal class PeerProtocolState(pieceCount: Int, private val maxPending: Int = 32) { init { - require(pieceCount in 0..209_715 && maxPending in 1..256) + require(pieceCount in 0..1_000_000 && maxPending in 1..256) } val available: BooleanArray = BooleanArray(pieceCount) diff --git a/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerWire.kt b/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerWire.kt index a587342ec..6f61375ef 100644 --- a/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerWire.kt +++ b/library/torrent/src/commonMain/kotlin/com/linroid/ketch/torrent/PeerWire.kt @@ -29,6 +29,7 @@ internal class PeerWire( private val connection: TorrentConnection, private val metadata: TorrentMetadata? = null, private val idleTimeoutMs: Long = 180_000, + private val pieceCount: Int? = metadata?.let { it.pieceHashes.size / 20 }, ) { suspend fun handshake(local: PeerHandshake): PeerHandshake = withTimeout(10_000) { connection.write(encodeHandshake(local)) @@ -37,12 +38,17 @@ internal class PeerWire( suspend fun read(): PeerMessage = withTimeout(idleTimeoutMs) { val size = Buffer().write(connection.readExactly(4)).readInt() - require(size in 0..MAX_FRAME_SIZE) { "Peer frame exceeds limit" } - decode(connection.readExactly(size), metadata) + val limits = PeerFrameLimits(pieceCount) + limits.validateSize(size) + if (size == 0) return@withTimeout PeerMessage.KeepAlive + val id = connection.readExactly(1).single().toInt() and 255 + limits.validateType(size, id) + val payload = Buffer().writeByte(id).write(connection.readExactly(size - 1)).readByteArray() + decode(payload, metadata, pieceCount) } suspend fun send(message: PeerMessage) = withTimeout(idleTimeoutMs) { - connection.write(encode(message, metadata)) + connection.write(encode(message, metadata, pieceCount)) } companion object { @@ -72,11 +78,17 @@ internal class PeerWire( ) } - fun decode(payload: ByteArray, metadata: TorrentMetadata? = null): PeerMessage { - require(payload.size <= MAX_FRAME_SIZE) + fun decode( + payload: ByteArray, + metadata: TorrentMetadata? = null, + pieceCount: Int? = metadata?.let { it.pieceHashes.size / 20 }, + ): PeerMessage { + val limits = PeerFrameLimits(pieceCount) + limits.validateSize(payload.size) if (payload.isEmpty()) return PeerMessage.KeepAlive val input = Buffer().write(payload) val id = input.readByte().toInt() and 255 + limits.validateType(payload.size, id) fun exact(size: Int) = require(input.size == size.toLong()) { "Invalid peer message length" } val message = when (id) { in 0..3 -> { @@ -110,12 +122,16 @@ internal class PeerWire( } else -> PeerMessage.Unknown(id, input.readByteArray()) } - validate(message, metadata) + validate(message, metadata, pieceCount) return message } - fun encode(message: PeerMessage, metadata: TorrentMetadata? = null): ByteArray { - validate(message, metadata) + fun encode( + message: PeerMessage, + metadata: TorrentMetadata? = null, + pieceCount: Int? = metadata?.let { it.pieceHashes.size / 20 }, + ): ByteArray { + validate(message, metadata, pieceCount) val out = Buffer() when (message) { PeerMessage.KeepAlive -> Unit @@ -132,11 +148,21 @@ internal class PeerWire( is PeerMessage.Extended -> out.writeByte(20).writeByte(message.id).write(message.payload) is PeerMessage.Unknown -> out.writeByte(message.id).write(message.payload) } - require(out.size <= MAX_FRAME_SIZE) + PeerFrameLimits(pieceCount).validateSize(out.size.toInt()) return Buffer().writeInt(out.size.toInt()).write(out.readByteArray()).readByteArray() } - private fun validate(message: PeerMessage, metadata: TorrentMetadata?) { + private fun validate(message: PeerMessage, metadata: TorrentMetadata?, pieceCount: Int?) { + val limits = PeerFrameLimits(pieceCount) + require(metadata == null || pieceCount == metadata.pieceHashes.size / 20) + val index = when (message) { + is PeerMessage.Have -> message.index + is PeerMessage.Request -> message.index + is PeerMessage.Cancel -> message.index + is PeerMessage.Piece -> message.index + else -> null + } + if (index != null && pieceCount != null) require(index in 0 until pieceCount) when (message) { is PeerMessage.Have -> validateIndex(message.index, metadata) is PeerMessage.Request -> @@ -145,22 +171,12 @@ internal class PeerWire( validateBlock(message.index, message.begin, message.length, metadata) is PeerMessage.Piece -> validateBlock(message.index, message.begin, message.bytes.size, metadata) - is PeerMessage.Bitfield -> { - require(message.bytes.size < MAX_FRAME_SIZE) - if (metadata != null) { - val count = metadata.pieceHashes.size / 20 - require(message.bytes.size == (count + 7) / 8) { "Wrong bitfield size" } - val spare = (8 - count % 8) % 8 - if (spare != 0) { - require(message.bytes.last().toInt() and ((1 shl spare) - 1) == 0) { - "Nonzero bitfield padding" - } - } - } - } + is PeerMessage.Bitfield -> limits.validateBitfield(message.bytes) is PeerMessage.Port -> require(message.port in 1..65535) - is PeerMessage.Extended -> require(message.id in 0..255) - is PeerMessage.Unknown -> require(message.id in 0..255) + is PeerMessage.Extended -> require(message.id in 0..255 && + message.payload.size <= MAX_FRAME_SIZE - 2) + is PeerMessage.Unknown -> require(message.id in 0..255 && + message.payload.size <= MAX_FRAME_SIZE - 1) else -> Unit } } diff --git a/library/torrent/src/commonTest/kotlin/com/linroid/ketch/torrent/PeerFrameLimitsTest.kt b/library/torrent/src/commonTest/kotlin/com/linroid/ketch/torrent/PeerFrameLimitsTest.kt new file mode 100644 index 000000000..abb36174e --- /dev/null +++ b/library/torrent/src/commonTest/kotlin/com/linroid/ketch/torrent/PeerFrameLimitsTest.kt @@ -0,0 +1,92 @@ +package com.linroid.ketch.torrent + +import kotlinx.coroutines.test.runTest +import okio.Buffer +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertIs +import kotlin.test.assertTrue + +class PeerFrameLimitsTest { + private class Connection(val input: Buffer) : TorrentConnection { + override val remote = PeerEndpoint("127.0.0.1", 1) + val reads = mutableListOf() + override suspend fun readExactly(size: Int): ByteArray { + reads += size + return input.readByteArray(size.toLong()) + } + override suspend fun write(bytes: ByteArray) = Unit + override fun close() = Unit + } + + @Test + fun desktopBitfieldsReachTheLastPieceWithExactSizeAndPadding() = runTest { + for (count in listOf(999_999, 1_000_000)) { + val bits = ByteArray((count + 7) / 8) + bits[0] = 0x80.toByte() + bits[bits.lastIndex] = if (count % 8 == 0) 1 else 2 + val encoded = PeerWire.encode(PeerMessage.Bitfield(bits), pieceCount = count) + val connection = Connection(Buffer().write(encoded)) + val decoded = assertIs(PeerWire(connection, pieceCount = count).read()) + assertContentEquals(bits, decoded.bytes) + val state = PeerProtocolState(count) + state.received(decoded) + assertTrue(state.available[0] && state.available[count - 1]) + assertEquals(listOf(4, 1, bits.size), connection.reads) + assertFailsWith { + PeerWire.decode(encoded.copyOfRange(4, encoded.size)) + } + assertFailsWith { + PeerWire.encode(PeerMessage.Have(count), pieceCount = count) + } + if (count % 8 != 0) { + bits[bits.lastIndex] = 3 + assertFailsWith { + PeerWire.encode(PeerMessage.Bitfield(bits), pieceCount = count) + } + } + } + assertFailsWith { PeerFrameLimits(1_000_001) } + } + + @Test + fun unrelatedOversizedFramesAndWrongBitfieldsFailBeforeBodyReads() = runTest { + for ((size, id) in listOf(125_001 to 20, 125_001 to 22, 125_000 to 5)) { + val connection = Connection(Buffer().writeInt(size).writeByte(id)) + assertFailsWith { + PeerWire(connection, pieceCount = 1_000_000).read() + } + assertEquals(listOf(4, 1), connection.reads) + } + assertFailsWith { + PeerWire.encode(PeerMessage.Unknown(24, ByteArray(100_000)), pieceCount = 1_000_000) + } + assertFailsWith { + PeerWire.encode(PeerMessage.Extended(0, ByteArray(100_000)), pieceCount = 1_000_000) + } + } + + @Test + fun transportAdmitsAndRetainsTheEntireDesktopBitfield() = runTest { + val bits = ByteArray(125_000) + val encoded = PeerWire.encode(PeerMessage.Bitfield(bits), pieceCount = 1_000_000) + val connection = Connection(Buffer().write(encoded)) + val budget = TorrentBufferBudget(600_000) + val exchange = PeerHashExchange(TorrentBufferBudget(32_768), { null }) + val transport = PeerHashTransport(connection, exchange, budget, pieceCount = 1_000_000) + val frame = transport.read() + try { + assertContentEquals(bits, assertIs(frame.message).bytes) + assertTrue(budget.allocated >= 4 * 125_001) + } finally { frame.close(); transport.close() } + assertEquals(0, budget.allocated) + val wrong = Connection(Buffer().writeInt(125_001).writeByte(22)) + val rejected = PeerHashTransport(wrong, exchange, budget, pieceCount = 1_000_000) + assertFailsWith { rejected.read() } + assertEquals(listOf(4, 1), wrong.reads) + assertEquals(0, budget.allocated) + rejected.close() + } +} diff --git a/library/torrent/src/commonTest/kotlin/com/linroid/ketch/torrent/PeerHashTransportTest.kt b/library/torrent/src/commonTest/kotlin/com/linroid/ketch/torrent/PeerHashTransportTest.kt index 8217b7bee..f74303180 100644 --- a/library/torrent/src/commonTest/kotlin/com/linroid/ketch/torrent/PeerHashTransportTest.kt +++ b/library/torrent/src/commonTest/kotlin/com/linroid/ketch/torrent/PeerHashTransportTest.kt @@ -36,7 +36,7 @@ class PeerHashTransportTest { var bodyDelay = 0L override suspend fun readExactly(size: Int): ByteArray { sizes += size - if (sizes.size > 1) delay(bodyDelay) + if (sizes.size > 2) delay(bodyDelay) return incoming.readByteArray(size.toLong()) } override suspend fun write(bytes: ByteArray) { @@ -122,7 +122,7 @@ class PeerHashTransportTest { @Test fun rejectsOversizedOrUnadmittedBodiesBeforeReadingThem() = runTest { for (size in listOf(Int.MAX_VALUE, 128)) { - val connection = Connection(Buffer().writeInt(size)) + val connection = Connection(Buffer().writeInt(size).writeByte(0)) val budget = TorrentBufferBudget(600) val transport = PeerHashTransport(connection, exchange(TorrentBufferBudget(32_768)), budget) if (size == Int.MAX_VALUE) { @@ -130,7 +130,7 @@ class PeerHashTransportTest { } else { assertFailsWith { transport.read() } } - assertEquals(listOf(4), connection.sizes) + assertEquals(if (size == Int.MAX_VALUE) listOf(4) else listOf(4, 1), connection.sizes) assertTrue(connection.closed) assertEquals(0, budget.allocated) transport.close()