Merge 3f523d4031b1a087ade0e61226fcc3f4a9038af7 into 948d6a85b9f254760c475a64f02fe8899e4f3d7f

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..<header.total).reduce(into: Data()) { data, index in
+        let reassembled = (0..<metadata.total).reduce(into: Data()) { data, index in
             if let fragment = fragments[index] {
                 data.append(fragment)
             }
@@ -134,12 +240,14 @@ struct BLEFragmentAssemblyBuffer {
         return .complete(header: header, reassembledData: reassembled, started: started)
     }
 
-    private mutating func startAssemblyIfNeeded(
+    private mutating func prepareAssembly(
         for header: BLEFragmentHeader,
         maxInFlightAssemblies: Int,
         now: Date
-    ) -> 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 }

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

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,

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