diff --git a/bitchat/Services/BLE/BLEFragmentAssemblyBuffer.swift b/bitchat/Services/BLE/BLEFragmentAssemblyBuffer.swift index 31656cb4..cea38a5d 100644 --- a/bitchat/Services/BLE/BLEFragmentAssemblyBuffer.swift +++ b/bitchat/Services/BLE/BLEFragmentAssemblyBuffer.swift @@ -53,19 +53,61 @@ struct BLEFragmentHeader: Equatable { } } +/// Reassembles inbound fragment streams keyed by `(sender, fragmentID)`. +/// +/// Fragments are unauthenticated and bypass the packet deduplicator, so any +/// peer in range can address a stream it does not own. The buffer treats the +/// first accepted header as authoritative for the stream's shape and refuses +/// to let a later fragment restate it or rewrite an index already held. +/// +/// This bounds the damage; it does not eliminate it. An injected fragment +/// that reaches an index before the honest one does *is* the first accepted +/// value for that index, and one such packet is enough to corrupt the +/// reassembled payload. What first-wins buys is cost against premature +/// completion or full replacement: the attacker must supply the stream's +/// pinned total and win the race at every index it needs to control. Closing +/// the gap needs authenticated fragment envelopes or a sender-key-bound +/// stream ID — the receiver has nothing to distinguish the variants by. struct BLEFragmentAssemblyBuffer { + enum ConflictReason: Equatable { + case total(expected: Int, actual: Int) + case originalType(expected: UInt8, actual: UInt8) + case broadcastScope(expected: Bool, actual: Bool) + case fragmentData(index: Int) + } + enum AppendResult: Equatable { case stored(header: BLEFragmentHeader, started: Bool) case complete(header: BLEFragmentHeader, reassembledData: Data, started: Bool) case oversized(header: BLEFragmentHeader, projectedSize: Int, limit: Int, started: Bool) + case conflicting(header: BLEFragmentHeader, reason: ConflictReason) } private struct Metadata { let total: Int + let originalType: UInt8 let timestamp: Date let isBroadcast: Bool var lastFragmentAt: Date var lastResyncRequestAt: Date? + /// Set when a fragment would have pushed the stream past its size + /// limit. The assembly is kept (see `append`) but cannot complete on + /// what it holds, so it stops consuming REQUEST_SYNC filter slots + /// until a fragment actually stores and clears the mark. + var exceededBudget: Bool = false + + func conflictReason(for header: BLEFragmentHeader) -> ConflictReason? { + if header.total != total { + return .total(expected: total, actual: header.total) + } + if header.originalType != originalType { + return .originalType(expected: originalType, actual: header.originalType) + } + if header.isBroadcastFragment != isBroadcast { + return .broadcastScope(expected: isBroadcast, actual: header.isBroadcastFragment) + } + return nil + } } private var fragmentsByKey: [BLEFragmentKey: [Int: Data]] = [:] @@ -95,15 +137,76 @@ struct BLEFragmentAssemblyBuffer { maxInFlightAssemblies: Int, now: Date = Date() ) -> AppendResult { - let started = startAssemblyIfNeeded(for: header, maxInFlightAssemblies: maxInFlightAssemblies, now: now) + // Reject a fragment too large to ever be stored before touching the + // table: starting an assembly evicts the oldest in-flight stream to + // make room, and a fragment headed for rejection must not cost a + // legitimate stream its slot. This is reachable with one packet — a + // compressed fragment may legally inflate to more than the whole + // budget, since `BinaryProtocol` caps decompression at 50000:1. + if metadataByKey[header.key] == nil { + let limit = Self.assemblyLimit(for: header.originalType) + if header.fragmentData.count > limit { + return .oversized( + header: header, + projectedSize: header.fragmentData.count, + limit: limit, + started: false + ) + } + } + + let assembly = prepareAssembly( + for: header, + maxInFlightAssemblies: maxInFlightAssemblies, + now: now + ) + let metadata = assembly.metadata + let started = assembly.started + + // A fragment ID identifies one immutable stream from a sender. Keep + // the first accepted header authoritative so a collision or injected + // fragment cannot shorten the stream, switch its size policy, or + // change whether it participates in broadcast gossip recovery. + if let reason = metadata.conflictReason(for: header) { + return .conflicting(header: header, reason: reason) + } + + let existingFragment = fragmentsByKey[header.key]?[header.index] + if let existingFragment, existingFragment != header.fragmentData { + // Duplicate delivery is expected in a mesh, but the same stream + // index must always carry the same bytes. Preserve first-wins + // state instead of letting a conflicting duplicate poison it. + // This protects indices already held — an index still empty when + // the injected fragment arrives is filled by whichever copy wins + // the race, and nothing here can tell them apart. + return .conflicting(header: header, reason: .fragmentData(index: header.index)) + } let currentSize = fragmentsByKey[header.key]?.values.reduce(0) { $0 + $1.count } ?? 0 - let limit = Self.assemblyLimit(for: header.originalType) - let projectedSize = currentSize + header.fragmentData.count + let limit = Self.assemblyLimit(for: metadata.originalType) + let projectedSize = currentSize + (existingFragment == nil ? header.fragmentData.count : 0) guard projectedSize <= limit else { - fragmentsByKey.removeValue(forKey: header.key) - metadataByKey.removeValue(forKey: header.key) + // An incoming fragment must never destroy state it did not + // create: an injected fragment at an unused index would + // otherwise be enough to wipe a legitimate in-flight stream. + // Nothing above the limit is ever stored, so keeping the + // assembly stays inside the same memory bound — a genuinely + // oversized stream just never completes and `removeExpired` + // reaps it. Only an assembly this fragment itself started has + // nothing worth preserving. + if started { + fragmentsByKey.removeValue(forKey: header.key) + metadataByKey.removeValue(forKey: header.key) + } else { + // A retained assembly that hit its ceiling cannot complete on + // what it holds, so stop it drawing REQUEST_SYNC retries it + // cannot use. This is provisional, not a verdict on the + // stream: the fragment that tripped the ceiling may be the + // injected one, so any later fragment that does store clears + // the mark and restores recovery. + metadataByKey[header.key]?.exceededBudget = true + } return .oversized(header: header, projectedSize: projectedSize, limit: limit, started: started) } @@ -111,18 +214,21 @@ struct BLEFragmentAssemblyBuffer { // bypass the packet deduplicator, so relayed duplicates of an // already-held index must not keep suppressing the targeted // REQUEST_SYNC for a stalled stream. - let isNewIndex = fragmentsByKey[header.key]?[header.index] == nil + let isNewIndex = existingFragment == nil fragmentsByKey[header.key]?[header.index] = header.fragmentData if isNewIndex { metadataByKey[header.key]?.lastFragmentAt = now + // Real progress: whatever tripped the ceiling earlier did not stop + // this stream, so it is eligible for recovery again. + metadataByKey[header.key]?.exceededBudget = false } guard let fragments = fragmentsByKey[header.key], - fragments.count == header.total else { + fragments.count == metadata.total else { return .stored(header: header, started: started) } - let reassembled = (0.. Bool { - guard fragmentsByKey[header.key] == nil else { return false } + ) -> (metadata: Metadata, started: Bool) { + if let metadata = metadataByKey[header.key] { + return (metadata, false) + } if fragmentsByKey.count >= maxInFlightAssemblies, let oldest = metadataByKey.min(by: { $0.value.timestamp < $1.value.timestamp })?.key { @@ -147,14 +255,16 @@ struct BLEFragmentAssemblyBuffer { metadataByKey.removeValue(forKey: oldest) } - fragmentsByKey[header.key] = [:] - metadataByKey[header.key] = Metadata( + let metadata = Metadata( total: header.total, + originalType: header.originalType, timestamp: now, isBroadcast: header.isBroadcastFragment, lastFragmentAt: now ) - return true + fragmentsByKey[header.key] = [:] + metadataByKey[header.key] = metadata + return (metadata, true) } /// Fragment stream IDs (8-byte, big-endian) of incomplete broadcast @@ -166,7 +276,11 @@ struct BLEFragmentAssemblyBuffer { /// oldest-stall first; overflow streams stay unmarked and eligible for /// the next pass. Directed reassemblies are excluded: peers only archive /// broadcast fragments for gossip sync, so a targeted request cannot - /// recover them. + /// recover them. Streams that have tripped their size budget are excluded + /// too — they are retained rather than discarded, but cannot complete on + /// what they hold, so requesting more would only burn filter slots. That + /// exclusion lifts as soon as a fragment stores, since the fragment that + /// tripped the budget may have been an injected one. mutating func stalledBroadcastFragmentIDs( stalledAfter: TimeInterval, retryAfter: TimeInterval, @@ -175,6 +289,7 @@ struct BLEFragmentAssemblyBuffer { var candidates: [(key: BLEFragmentKey, lastFragmentAt: Date)] = [] for (key, metadata) in metadataByKey { guard metadata.isBroadcast, + !metadata.exceededBudget, let fragments = fragmentsByKey[key], fragments.count < metadata.total, now.timeIntervalSince(metadata.lastFragmentAt) >= stalledAfter else { continue } diff --git a/bitchat/Services/BLE/BLEFragmentHandler.swift b/bitchat/Services/BLE/BLEFragmentHandler.swift index 9d913151..8f8c69f4 100644 --- a/bitchat/Services/BLE/BLEFragmentHandler.swift +++ b/bitchat/Services/BLE/BLEFragmentHandler.swift @@ -48,6 +48,13 @@ final class BLEFragmentHandler { return } + // Archiving and relay stay deliberately independent of our local + // assembly state. A fragment we reject as conflicting may well be the + // honest one — we cannot tell, only that it lost the race here — so + // suppressing it would censor it for every downstream peer, including + // those that never saw the injected variant, and drop it from + // REQUEST_SYNC recovery. Local first-wins protects this node's + // reassembly; it is not evidence for the rest of the mesh. if header.isBroadcastFragment { env.trackPacketSeen(packet) } @@ -94,7 +101,24 @@ final class BLEFragmentHandler { case let .oversized(header, projectedSize, limit, started): logStartedIfNeeded(header: header, started: started) SecureLogger.warning( - "🚫 Fragment assembly exceeds size limit (\(projectedSize) bytes > \(limit)), evicting. Type=\(header.originalType) Index=\(header.index)/\(header.total)", + "🚫 Fragment rejected: assembly would exceed size limit (\(projectedSize) bytes > \(limit)). Type=\(header.originalType) Index=\(header.index)/\(header.total)", + category: .security + ) + + case let .conflicting(header, reason): + let detail: String + switch reason { + case let .total(expected, actual): + detail = "total expected=\(expected) actual=\(actual)" + case let .originalType(expected, actual): + detail = "type expected=\(expected) actual=\(actual)" + case let .broadcastScope(expected, actual): + detail = "broadcast expected=\(expected) actual=\(actual)" + case let .fragmentData(index): + detail = "different bytes for index=\(index)" + } + SecureLogger.warning( + "🚫 Conflicting fragment ignored. Stream=\(header.idLogString) \(detail)", category: .security ) } diff --git a/bitchatTests/Services/BLEFragmentAssemblyBufferTests.swift b/bitchatTests/Services/BLEFragmentAssemblyBufferTests.swift index ed4da0a1..c30d280b 100644 --- a/bitchatTests/Services/BLEFragmentAssemblyBufferTests.swift +++ b/bitchatTests/Services/BLEFragmentAssemblyBufferTests.swift @@ -52,6 +52,174 @@ struct BLEFragmentAssemblyBufferTests { } } + @Test + func conflictingTotalIsRejectedWithoutDiscardingAssembly() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragmentID = Data(repeating: 0x21, count: 8) + let first = try makeHeader( + fragmentID: fragmentID, + index: 0, + total: 3, + fragmentData: Data([0x01]) + ) + let conflicting = try makeHeader( + fragmentID: fragmentID, + index: 1, + total: 2, + fragmentData: Data([0xEE]) + ) + let second = try makeHeader( + fragmentID: fragmentID, + index: 1, + total: 3, + fragmentData: Data([0x02]) + ) + let third = try makeHeader( + fragmentID: fragmentID, + index: 2, + total: 3, + fragmentData: Data([0x03]) + ) + + _ = buffer.append(first, maxInFlightAssemblies: 8) + let conflict = buffer.append(conflicting, maxInFlightAssemblies: 8) + + #expect( + conflict == .conflicting( + header: conflicting, + reason: .total(expected: 3, actual: 2) + ) + ) + _ = buffer.append(second, maxInFlightAssemblies: 8) + + if case let .complete(_, data, _) = buffer.append(third, maxInFlightAssemblies: 8) { + #expect(data == Data([0x01, 0x02, 0x03])) + } else { + Issue.record("Expected the original assembly to survive a conflicting total") + } + } + + @Test + func conflictingOriginalTypeIsRejected() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragmentID = Data(repeating: 0x22, count: 8) + let first = try makeHeader( + fragmentID: fragmentID, + index: 0, + total: 2, + fragmentData: Data([0x01]) + ) + let conflicting = try makeHeader( + fragmentID: fragmentID, + index: 1, + total: 2, + originalType: MessageType.noiseEncrypted.rawValue, + fragmentData: Data([0xEE]) + ) + + _ = buffer.append(first, maxInFlightAssemblies: 8) + + #expect( + buffer.append(conflicting, maxInFlightAssemblies: 8) == .conflicting( + header: conflicting, + reason: .originalType( + expected: MessageType.message.rawValue, + actual: MessageType.noiseEncrypted.rawValue + ) + ) + ) + } + + @Test + func conflictingRecipientScopeIsRejected() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragmentID = Data(repeating: 0x23, count: 8) + let broadcast = try makeHeader( + fragmentID: fragmentID, + index: 0, + total: 2, + fragmentData: Data([0x01]) + ) + let directed = try makeHeader( + fragmentID: fragmentID, + index: 1, + total: 2, + fragmentData: Data([0x02]), + recipientID: Data(hexString: "0102030405060708") + ) + + _ = buffer.append(broadcast, maxInFlightAssemblies: 8) + + #expect( + buffer.append(directed, maxInFlightAssemblies: 8) == .conflicting( + header: directed, + reason: .broadcastScope(expected: true, actual: false) + ) + ) + } + + @Test + func conflictingDuplicateDataCannotOverwriteFirstFragment() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragmentID = Data(repeating: 0x24, count: 8) + let first = try makeHeader( + fragmentID: fragmentID, + index: 0, + total: 2, + fragmentData: Data([0x01]) + ) + let conflicting = try makeHeader( + fragmentID: fragmentID, + index: 0, + total: 2, + fragmentData: Data([0xEE]) + ) + let second = try makeHeader( + fragmentID: fragmentID, + index: 1, + total: 2, + fragmentData: Data([0x02]) + ) + + _ = buffer.append(first, maxInFlightAssemblies: 8) + + #expect( + buffer.append(conflicting, maxInFlightAssemblies: 8) == .conflicting( + header: conflicting, + reason: .fragmentData(index: 0) + ) + ) + + if case let .complete(_, data, _) = buffer.append(second, maxInFlightAssemblies: 8) { + #expect(data == Data([0x01, 0x02])) + } else { + Issue.record("Expected first-wins fragment data to complete normally") + } + } + + @Test + func exactDuplicateDoesNotCountTwiceTowardSizeLimit() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragmentID = Data(repeating: 0x25, count: 8) + let fragment = try makeHeader( + fragmentID: fragmentID, + index: 0, + total: 2, + fragmentData: Data( + repeating: 0x01, + count: FileTransferLimits.maxPayloadBytes / 2 + 1 + ) + ) + + _ = buffer.append(fragment, maxInFlightAssemblies: 8) + + if case let .stored(_, started) = buffer.append(fragment, maxInFlightAssemblies: 8) { + #expect(!started) + } else { + Issue.record("Expected an exact duplicate to keep the assembly unchanged") + } + } + @Test func appendEvictsOldestAssemblyWhenCapIsReached() throws { var buffer = BLEFragmentAssemblyBuffer() @@ -81,23 +249,32 @@ struct BLEFragmentAssemblyBufferTests { } @Test - func appendOversizedAssemblyDropsPartialState() throws { + func oversizedFragmentIsRejectedWithoutDiscardingAssembly() throws { var buffer = BLEFragmentAssemblyBuffer() let fragmentID = Data(repeating: 0x05, count: 8) - let first = try #require(BLEFragmentHeader(packet: makeFragmentPacket( + let headroom = 10 + let first = try makeHeader( fragmentID: fragmentID, index: 0, total: 2, - originalType: MessageType.message.rawValue, - fragmentData: Data(repeating: 0x01, count: FileTransferLimits.maxPayloadBytes) - ))) - let oversized = try #require(BLEFragmentHeader(packet: makeFragmentPacket( + fragmentData: Data( + repeating: 0x01, + count: FileTransferLimits.maxPayloadBytes - headroom + ) + ) + // An injected fragment at an unused index, sized to blow the budget. + let oversized = try makeHeader( fragmentID: fragmentID, index: 1, total: 2, - originalType: MessageType.message.rawValue, - fragmentData: Data([0x02]) - ))) + fragmentData: Data(repeating: 0xEE, count: headroom + 1) + ) + let legitimate = try makeHeader( + fragmentID: fragmentID, + index: 1, + total: 2, + fragmentData: Data(repeating: 0x02, count: headroom) + ) _ = buffer.append(first, maxInFlightAssemblies: 8) let result = buffer.append(oversized, maxInFlightAssemblies: 8) @@ -107,13 +284,189 @@ struct BLEFragmentAssemblyBufferTests { #expect(limit == FileTransferLimits.maxPayloadBytes) #expect(!started) } else { - Issue.record("Expected oversized fragment assembly to be evicted") + Issue.record("Expected the oversized fragment to be rejected") } - if case let .stored(_, started) = buffer.append(oversized, maxInFlightAssemblies: 8) { - #expect(started) + // The stream a spoofed fragment tried to blow up still completes. + if case let .complete(_, data, _) = buffer.append(legitimate, maxInFlightAssemblies: 8) { + #expect(data.count == FileTransferLimits.maxPayloadBytes) + #expect(data.suffix(headroom) == Data(repeating: 0x02, count: headroom)) } else { - Issue.record("Expected later fragment to start a clean assembly") + Issue.record("Expected the original assembly to survive an oversized fragment") + } + } + + @Test + func oversizedFirstFragmentIsRejectedWithoutClaimingASlot() throws { + var buffer = BLEFragmentAssemblyBuffer() + // A single fragment over the budget can never be stored, so it must be + // turned away before an assembly is started for it — starting one + // evicts the oldest in-flight stream to make room. A compressed + // fragment can reach this size in one packet. + let victimID = Data(repeating: 0x07, count: 8) + let victimFirst = try makeHeader( + fragmentID: victimID, + index: 0, + total: 2, + fragmentData: Data([0x01]) + ) + let victimSecond = try makeHeader( + fragmentID: victimID, + index: 1, + total: 2, + fragmentData: Data([0x02]) + ) + let oversized = try makeHeader( + fragmentID: Data(repeating: 0x06, count: 8), + index: 0, + total: 2, + fragmentData: Data( + repeating: 0xEE, + count: FileTransferLimits.maxPayloadBytes + 1 + ) + ) + + _ = buffer.append(victimFirst, maxInFlightAssemblies: 1) + + if case let .oversized(_, projectedSize, limit, started) = buffer.append(oversized, maxInFlightAssemblies: 1) { + #expect(projectedSize == FileTransferLimits.maxPayloadBytes + 1) + #expect(limit == FileTransferLimits.maxPayloadBytes) + #expect(!started) + } else { + Issue.record("Expected a single over-budget fragment to be rejected") + } + + // The assembly holding the only in-flight slot was never evicted. + if case let .complete(_, data, _) = buffer.append(victimSecond, maxInFlightAssemblies: 1) { + #expect(data == Data([0x01, 0x02])) + } else { + Issue.record("Expected a rejected fragment to leave the in-flight slot alone") + } + } + + @Test + func overBudgetStreamStopsDrawingResyncRequests() throws { + var buffer = BLEFragmentAssemblyBuffer() + let t0 = Date(timeIntervalSince1970: 100) + let headroom = 10 + let starvedID = Data(repeating: 0x08, count: 8) + let healthyID = Data(repeating: 0x09, count: 8) + + let starvedFirst = try makeHeader( + fragmentID: starvedID, + index: 0, + total: 3, + fragmentData: Data( + repeating: 0x01, + count: FileTransferLimits.maxPayloadBytes - headroom + ) + ) + let starvedOversized = try makeHeader( + fragmentID: starvedID, + index: 1, + total: 3, + fragmentData: Data(repeating: 0xEE, count: headroom + 1) + ) + let healthyFirst = try makeHeader( + fragmentID: healthyID, + index: 0, + total: 2, + fragmentData: Data([0x01]) + ) + + _ = buffer.append(starvedFirst, maxInFlightAssemblies: 8, now: t0) + _ = buffer.append(healthyFirst, maxInFlightAssemblies: 8, now: t0) + _ = buffer.append(starvedOversized, maxInFlightAssemblies: 8, now: t0) + + // The starved stream is retained so a spoofed fragment cannot destroy + // it, but it cannot complete on what it holds — only the healthy + // stream is worth a REQUEST_SYNC slot. + let stalled = buffer.stalledBroadcastFragmentIDs( + stalledAfter: 5, + retryAfter: 10, + now: t0.addingTimeInterval(6) + ) + #expect(stalled == [healthyID]) + + // The fragment that tripped the ceiling may have been the injected + // one. A fragment that does store proves the stream is still moving, + // so recovery must come back rather than stay suppressed for the + // assembly's whole lifetime. + let starvedProgress = try makeHeader( + fragmentID: starvedID, + index: 1, + total: 3, + fragmentData: Data(repeating: 0x02, count: headroom) + ) + _ = buffer.append(starvedProgress, maxInFlightAssemblies: 8, now: t0.addingTimeInterval(7)) + + let recovered = buffer.stalledBroadcastFragmentIDs( + stalledAfter: 5, + retryAfter: 10, + now: t0.addingTimeInterval(20) + ) + #expect(recovered.contains(starvedID)) + } + + @Test + func streamsFromDifferentSendersSharingAFragmentIDStayIsolated() throws { + var buffer = BLEFragmentAssemblyBuffer() + // Fragment IDs are only unique per sender, and a colliding ID must not + // make two honest senders reject each other — `BLEFragmentKey` carries + // the sender for exactly this reason. Dropping it would turn the + // first-wins checks below into a mutual-rejection DoS. + let sharedID = Data(repeating: 0x0A, count: 8) + let alice = Data([0xA0, 0xA1, 0xA2, 0xA3, 0xA4, 0xA5, 0xA6, 0xA7]) + let bob = Data([0xB0, 0xB1, 0xB2, 0xB3, 0xB4, 0xB5, 0xB6, 0xB7]) + + // Same ID, different senders, and every header field disagrees. + let aliceFirst = try makeHeader( + fragmentID: sharedID, + index: 0, + total: 2, + fragmentData: Data([0x01]), + senderID: alice + ) + let bobFirst = try makeHeader( + fragmentID: sharedID, + index: 0, + total: 3, + originalType: MessageType.noiseEncrypted.rawValue, + fragmentData: Data([0xB1]), + senderID: bob + ) + let aliceSecond = try makeHeader( + fragmentID: sharedID, + index: 1, + total: 2, + fragmentData: Data([0x02]), + senderID: alice + ) + let bobRest = try (1...2).map { index in + try makeHeader( + fragmentID: sharedID, + index: index, + total: 3, + originalType: MessageType.noiseEncrypted.rawValue, + fragmentData: Data([UInt8(0xB1 + index)]), + senderID: bob + ) + } + + _ = buffer.append(aliceFirst, maxInFlightAssemblies: 8) + #expect(!isConflicting(buffer.append(bobFirst, maxInFlightAssemblies: 8))) + #expect(!isConflicting(buffer.append(bobRest[0], maxInFlightAssemblies: 8))) + + if case let .complete(_, data, _) = buffer.append(aliceSecond, maxInFlightAssemblies: 8) { + #expect(data == Data([0x01, 0x02])) + } else { + Issue.record("Expected Alice's stream to complete independently") + } + + if case let .complete(_, data, _) = buffer.append(bobRest[1], maxInFlightAssemblies: 8) { + #expect(data == Data([0xB1, 0xB2, 0xB3])) + } else { + Issue.record("Expected Bob's stream to complete independently") } } @@ -241,6 +594,45 @@ struct BLEFragmentAssemblyBufferTests { #expect(stalled == [fragmentID]) } + @Test + func conflictingFragmentsDoNotResetStallClock() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragmentID = Data(repeating: 0x26, count: 8) + let first = try makeHeader( + fragmentID: fragmentID, + index: 0, + total: 2, + fragmentData: Data([0x01]) + ) + let conflicting = try makeHeader( + fragmentID: fragmentID, + index: 0, + total: 2, + fragmentData: Data([0xEE]) + ) + + let t0 = Date(timeIntervalSince1970: 100) + _ = buffer.append(first, maxInFlightAssemblies: 8, now: t0) + + // A rejected fragment brings no progress, so a flood of them must not + // keep a stalled stream looking "fresh" and suppress its REQUEST_SYNC. + for offset in [3.0, 5.0] { + let result = buffer.append( + conflicting, + maxInFlightAssemblies: 8, + now: t0.addingTimeInterval(offset) + ) + #expect(result == .conflicting(header: conflicting, reason: .fragmentData(index: 0))) + } + + let stalled = buffer.stalledBroadcastFragmentIDs( + stalledAfter: 5, + retryAfter: 10, + now: t0.addingTimeInterval(6) + ) + #expect(stalled == [fragmentID]) + } + @Test func overflowStalledStreamsRotateAcrossPasses() throws { var buffer = BLEFragmentAssemblyBuffer() @@ -338,6 +730,31 @@ struct BLEFragmentAssemblyBufferTests { } } + private func isConflicting(_ result: BLEFragmentAssemblyBuffer.AppendResult) -> Bool { + if case .conflicting = result { return true } + return false + } + + private func makeHeader( + fragmentID: Data, + index: Int, + total: Int, + originalType: UInt8 = MessageType.message.rawValue, + fragmentData: Data, + senderID: Data = Data([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77]), + recipientID: Data? = nil + ) throws -> BLEFragmentHeader { + try #require(BLEFragmentHeader(packet: makeFragmentPacket( + fragmentID: fragmentID, + index: index, + total: total, + originalType: originalType, + fragmentData: fragmentData, + senderID: senderID, + recipientID: recipientID + ))) + } + private func makeFragmentPacket( fragmentID: Data, index: Int, diff --git a/bitchatTests/Services/BLEFragmentHandlerTests.swift b/bitchatTests/Services/BLEFragmentHandlerTests.swift index e48a74ed..90e2bfa0 100644 --- a/bitchatTests/Services/BLEFragmentHandlerTests.swift +++ b/bitchatTests/Services/BLEFragmentHandlerTests.swift @@ -111,6 +111,27 @@ struct BLEFragmentHandlerTests { #expect(recorder.appendedHeaders.count == 1) } + @Test + func conflictingFragmentIsNotReassembledButStillArchivedForSync() { + let recorder = Recorder() + recorder.appendResult = { header in + .conflicting(header: header, reason: .fragmentData(index: header.index)) + } + let handler = makeHandler(recorder: recorder) + let packet = makeFragmentPacket(sender: remotePeerID, index: 0, total: 2) + + handler.handle(packet, from: remotePeerID) + + // Losing the race locally is not evidence the fragment is forged, so + // archiving stays independent of assembly state: withholding it would + // censor a possibly-honest copy from peers that never saw the + // conflict, and from REQUEST_SYNC recovery. + #expect(recorder.appendedHeaders.count == 1) + #expect(recorder.trackedPackets.count == 1) + #expect(recorder.ingressChecks.isEmpty) + #expect(recorder.reinjectedPackets.isEmpty) + } + @Test func completedReassemblyReinjectsAcceptedPacketWithZeroTTL() throws { let innerSender = PeerID(str: "99AABBCCDDEEFF00")