Bound desktop bitfields by authenticated piece counts

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.

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" }
+    }
+  }
+}

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) {

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)

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
       }
     }

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<Int>()
+    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<PeerMessage.Bitfield>(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<IllegalArgumentException> {
+        PeerWire.decode(encoded.copyOfRange(4, encoded.size))
+      }
+      assertFailsWith<IllegalArgumentException> {
+        PeerWire.encode(PeerMessage.Have(count), pieceCount = count)
+      }
+      if (count % 8 != 0) {
+        bits[bits.lastIndex] = 3
+        assertFailsWith<IllegalArgumentException> {
+          PeerWire.encode(PeerMessage.Bitfield(bits), pieceCount = count)
+        }
+      }
+    }
+    assertFailsWith<IllegalArgumentException> { 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<IllegalArgumentException> {
+        PeerWire(connection, pieceCount = 1_000_000).read()
+      }
+      assertEquals(listOf(4, 1), connection.reads)
+    }
+    assertFailsWith<IllegalArgumentException> {
+      PeerWire.encode(PeerMessage.Unknown(24, ByteArray(100_000)), pieceCount = 1_000_000)
+    }
+    assertFailsWith<IllegalArgumentException> {
+      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<PeerMessage.Bitfield>(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<IllegalArgumentException> { rejected.read() }
+    assertEquals(listOf(4, 1), wrong.reads)
+    assertEquals(0, budget.allocated)
+    rejected.close()
+  }
+}

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,15 +122,15 @@ 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) {
         assertFailsWith<IllegalArgumentException> { transport.read() }
       } else {
         assertFailsWith<IllegalStateException> { 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()