diff --git a/bitchat/Services/BLEService.swift b/bitchat/Services/BLEService.swift index 3d15a24a..a54bd43c 100644 --- a/bitchat/Services/BLEService.swift +++ b/bitchat/Services/BLEService.swift @@ -7,6 +7,78 @@ import CryptoKit import UIKit #endif +struct NotificationStreamAssembler { + private var buffer = Data() + + mutating func append(_ chunk: Data) -> (frames: [Data], droppedPrefixes: [UInt8], reset: Bool) { + guard !chunk.isEmpty else { return ([], [], false) } + + buffer.append(chunk) + + var frames: [Data] = [] + var dropped: [UInt8] = [] + var reset = false + let maxFrameLength = TransportConfig.blePendingWriteBufferCapBytes + + let minHeaderBytes = 14 // version + type + ttl + timestamp(8) + flags + length(2) + let minFramePrefix = minHeaderBytes + BinaryProtocol.senderIDSize + + while buffer.count >= minFramePrefix { + guard let first = buffer.first else { break } + if first != 1 { + dropped.append(buffer.removeFirst()) + continue + } + + guard buffer.count >= minHeaderBytes else { break } + + let headerBytes = Array(buffer.prefix(minFramePrefix)) + guard headerBytes.count == minFramePrefix else { break } + + let flags = headerBytes[11] + let hasRecipient = (flags & BinaryProtocol.Flags.hasRecipient) != 0 + let hasSignature = (flags & BinaryProtocol.Flags.hasSignature) != 0 + let payloadLen = (Int(headerBytes[12]) << 8) | Int(headerBytes[13]) + + var frameLength = minFramePrefix + payloadLen + if hasRecipient { frameLength += BinaryProtocol.recipientIDSize } + if hasSignature { frameLength += BinaryProtocol.signatureSize } + + guard frameLength > 0, frameLength <= maxFrameLength else { + buffer.removeAll() + reset = true + break + } + + if buffer.count < frameLength { + // Check if a new frame start exists within the incomplete buffer; if so, drop leading partial bytes. + if let nextStart = buffer.dropFirst().firstIndex(of: 1) { + let dropCount = buffer.distance(from: buffer.startIndex, to: nextStart) + if dropCount > 0 { + buffer.removeFirst(dropCount) + dropped.append(1) // treat as dropped partial start + } + } + break + } + + let frame = Data(buffer.prefix(frameLength)) + frames.append(frame) + buffer.removeFirst(frameLength) + } + + if !buffer.isEmpty, buffer.allSatisfy({ $0 == 0 }) { + buffer.removeAll(keepingCapacity: false) + } + + return (frames, dropped, reset) + } + + mutating func reset() { + buffer.removeAll(keepingCapacity: false) + } +} + /// BLEService — Bluetooth Mesh Transport /// - Emits events exclusively via `BitchatDelegate` for UI. /// - ChatViewModel must consume delegate callbacks (`didReceivePublicMessage`, `didReceiveNoisePayload`). @@ -40,6 +112,7 @@ final class BLEService: NSObject { var isConnecting: Bool = false var isConnected: Bool = false var lastConnectionAttempt: Date? = nil + var assembler = NotificationStreamAssembler() } private var peripherals: [String: PeripheralState] = [:] // UUID -> PeripheralState private var peerToPeripheralUUID: [String: String] = [:] // PeerID -> Peripheral UUID @@ -2515,7 +2588,8 @@ extension BLEService: CBCentralManagerDelegate { peerID: nil, isConnecting: true, isConnected: false, - lastConnectionAttempt: Date() + lastConnectionAttempt: Date(), + assembler: NotificationStreamAssembler() ) peripheral.delegate = self @@ -2564,7 +2638,9 @@ func centralManager(_ central: CBCentralManager, didConnect peripheral: CBPeriph characteristic: nil, peerID: nil, isConnecting: false, - isConnected: true + isConnected: true, + lastConnectionAttempt: nil, + assembler: NotificationStreamAssembler() ) } @@ -2702,7 +2778,8 @@ extension BLEService { peerID: nil, isConnecting: true, isConnected: false, - lastConnectionAttempt: Date() + lastConnectionAttempt: Date(), + assembler: NotificationStreamAssembler() ) peripheral.delegate = self let options: [String: Any] = [ @@ -2839,53 +2916,72 @@ extension BLEService: CBPeripheralDelegate { return } - guard let data = characteristic.value else { + guard let data = characteristic.value, !data.isEmpty else { SecureLogger.warning("⚠️ No data in notification", category: .session) return } - - // Received BLE notification - - // Process directly on main thread to avoid deadlocks (matches original implementation) - guard let packet = BinaryProtocol.decode(data) else { - // Avoid dumping entire payload; log size and short prefix for diagnostics - let prefix = data.prefix(16).map { String(format: "%02x", $0) }.joined(separator: " ") - SecureLogger.error("❌ Failed to decode notification packet (len=\(data.count), prefix=\(prefix))", category: .session) - return - } - - // Use the packet's senderID as the peer identifier - let senderID = packet.senderID.hexEncodedString() - // Only log non-announce packets - if packet.type != MessageType.announce.rawValue { - SecureLogger.debug("📦 Decoded notification packet type: \(packet.type) from sender: \(senderID)", category: .session) + + bufferNotificationChunk(data, from: peripheral) } - + + private func bufferNotificationChunk(_ chunk: Data, from peripheral: CBPeripheral) { let peripheralUUID = peripheral.identifier.uuidString - - // Update mapping ONLY for announce packets that come directly from the peer (not relayed) + + var state = peripherals[peripheralUUID] ?? PeripheralState( + peripheral: peripheral, + characteristic: nil, + peerID: nil, + isConnecting: false, + isConnected: peripheral.state == .connected, + lastConnectionAttempt: nil, + assembler: NotificationStreamAssembler() + ) + + var assembler = state.assembler + let result = assembler.append(chunk) + state.assembler = assembler + peripherals[peripheralUUID] = state + + for byte in result.droppedPrefixes { + SecureLogger.warning("⚠️ Dropping byte from BLE stream (unexpected prefix \(String(format: "%02x", byte)))", category: .session) + } + + if result.reset { + SecureLogger.error("❌ Invalid BLE frame length; reset notification stream", category: .session) + } + + for frame in result.frames { + guard let packet = BinaryProtocol.decode(frame) else { + let prefix = frame.prefix(16).map { String(format: "%02x", $0) }.joined(separator: " ") + SecureLogger.error("❌ Failed to decode assembled notification frame (len=\(frame.count), prefix=\(prefix))", category: .session) + continue + } + processNotificationPacket(packet, from: peripheral, peripheralUUID: peripheralUUID) + } + } + + private func processNotificationPacket(_ packet: BitchatPacket, from peripheral: CBPeripheral, peripheralUUID: String) { + let senderID = packet.senderID.hexEncodedString() + + if packet.type != MessageType.announce.rawValue { + SecureLogger.debug("📦 Decoded notification packet type: \(packet.type) from sender: \(senderID)", category: .session) + } + if packet.type == MessageType.announce.rawValue { - // Only update mapping if this is a direct announce (TTL == messageTTL means not relayed) if packet.ttl == messageTTL { if var state = peripherals[peripheralUUID] { state.peerID = senderID peripherals[peripheralUUID] = state } peerToPeripheralUUID[senderID] = peripheralUUID - // Mapping update - direct announce from peer } - // Record ingress link for last-hop suppression and process + let msgID = makeMessageID(for: packet) collectionsQueue.async(flags: .barrier) { [weak self] in self?.ingressByMessageID[msgID] = (.peripheral(peripheralUUID), Date()) } - // Process the announce packet regardless of whether we updated the mapping handleReceivedPacket(packet, from: senderID) } else { - // For non-announce packets, DO NOT update mappings - // These could be relayed packets from other peers - // Always use the packet's original senderID - // Record ingress link for last-hop suppression and process let msgID = makeMessageID(for: packet) collectionsQueue.async(flags: .barrier) { [weak self] in self?.ingressByMessageID[msgID] = (.peripheral(peripheralUUID), Date()) diff --git a/bitchat/Sync/GCSFilter.swift b/bitchat/Sync/GCSFilter.swift index d05856fc..d3c4d9fb 100644 --- a/bitchat/Sync/GCSFilter.swift +++ b/bitchat/Sync/GCSFilter.swift @@ -4,7 +4,7 @@ import CryptoKit // Golomb-Coded Set (GCS) filter utilities for sync. // Hashing: // - Packet ID is 16 bytes (see PacketIdUtil). For GCS mapping, use h64 = first 8 bytes of SHA-256 over the 16-byte ID. -// - Map to [0, M) via (h64 % M). +// - Map to [1, M) by computing (h64 % M) and remapping 0 -> 1 to avoid zero-length deltas. // Encoding (v1): // - Sort mapped values ascending; encode deltas (first is v0, then vi - v{i-1}) as positive integers x >= 1. // - Golomb-Rice with parameter P: q = (x - 1) >> P encoded as unary (q ones then a zero), then write P-bit remainder r = (x - 1) & ((1< Params { let p = deriveP(targetFpr: targetFpr) + guard !ids.isEmpty else { + return Params(p: p, m: 1, data: Data()) + } + let cap = estimateMaxElements(sizeBytes: maxBytes, p: p) - let n = min(ids.count, cap) - let selected = Array(ids.prefix(n)) - // Map to [0, M) - let mInit = UInt32(n << p) - var mapped = selected.map { id16 -> UInt64 in - let h = h64(id16) - return UInt64(h % UInt64(max(1, mInit))) - }.sorted() + let selected = Array(ids.prefix(cap)) + let range = max(1, hashRange(count: selected.count, p: p)) + let modulo = UInt64(range) + + var mapped = selected + .map { h64($0) } + .map { mapHash($0, modulo: modulo) } + .sorted() + mapped = normalizeMappedValues(mapped, modulo: modulo) + + if mapped.isEmpty { + return Params(p: p, m: range, data: Data()) + } + var encoded = encode(sorted: mapped, p: p) - var trimmedN = n - // Trim if over budget - while encoded.count > maxBytes && trimmedN > 0 { - trimmedN = (trimmedN * 9) / 10 // drop ~10% - mapped = Array(mapped.prefix(trimmedN)) + var trimmedCount = mapped.count + + while encoded.count > maxBytes && trimmedCount > 0 { + if trimmedCount == 1 { + mapped.removeAll() + encoded = Data() + break + } + trimmedCount = max(1, (trimmedCount * 9) / 10) + mapped = Array(mapped.prefix(trimmedCount)) encoded = encode(sorted: mapped, p: p) } - let finalM = UInt32(max(1, trimmedN << p)) - return Params(p: p, m: finalM, data: encoded) + + return Params(p: p, m: range, data: encoded) } static func decodeToSortedSet(p: Int, m: UInt32, data: Data) -> [UInt64] { @@ -77,6 +92,12 @@ enum GCSFilter { return false } + static func bucket(for id: Data, modulus m: UInt32) -> UInt64 { + let modulo = UInt64(max(1, m)) + guard modulo > 1 else { return 0 } + return mapHash(h64(id), modulo: modulo) + } + private static func h64(_ id16: Data) -> UInt64 { var hasher = SHA256() hasher.update(data: id16) @@ -88,6 +109,39 @@ enum GCSFilter { return x & 0x7fff_ffff_ffff_ffff } + private static func hashRange(count: Int, p: Int) -> UInt32 { + guard count > 0 else { return 1 } + if p >= 64 { return UInt32.max } + let multiplier = UInt64(1) << UInt64(p) + let (product, overflow) = UInt64(count).multipliedReportingOverflow(by: multiplier) + if overflow { return UInt32.max } + if product == 0 { return 1 } + return product > UInt64(UInt32.max) ? UInt32.max : UInt32(product) + } + + private static func mapHash(_ hash: UInt64, modulo: UInt64) -> UInt64 { + guard modulo > 1 else { return 0 } + let value = hash % modulo + if value == 0 { return 1 } + return value + } + + private static func normalizeMappedValues(_ values: [UInt64], modulo: UInt64) -> [UInt64] { + guard modulo > 1 else { return [] } + guard !values.isEmpty else { return [] } + var result: [UInt64] = [] + result.reserveCapacity(values.count) + var last: UInt64 = 0 + for value in values { + let normalized = min(value, modulo - 1) + if normalized > last { + result.append(normalized) + last = normalized + } + } + return result + } + private static func encode(sorted: [UInt64], p: Int) -> Data { let writer = BitWriter() var prev: UInt64 = 0 diff --git a/bitchat/Sync/GossipSyncManager.swift b/bitchat/Sync/GossipSyncManager.swift index 52157c89..9bf25433 100644 --- a/bitchat/Sync/GossipSyncManager.swift +++ b/bitchat/Sync/GossipSyncManager.swift @@ -1,5 +1,4 @@ import Foundation -import CryptoKit // Gossip-based sync manager using on-demand GCS filters final class GossipSyncManager { @@ -53,6 +52,12 @@ final class GossipSyncManager { } func onPublicPacketSeen(_ packet: BitchatPacket) { + queue.async { [weak self] in + self?._onPublicPacketSeen(packet) + } + } + + private func _onPublicPacketSeen(_ packet: BitchatPacket) { let mt = MessageType(rawValue: packet.type) let isBroadcastRecipient: Bool = { guard let r = packet.recipientID else { return true } @@ -119,18 +124,17 @@ final class GossipSyncManager { } func handleRequestSync(fromPeerID: String, request: RequestSyncPacket) { + queue.async { [weak self] in + self?._handleRequestSync(fromPeerID: fromPeerID, request: request) + } + } + + private func _handleRequestSync(fromPeerID: String, request: RequestSyncPacket) { // Decode GCS into sorted set and prepare membership checker let sorted = GCSFilter.decodeToSortedSet(p: request.p, m: request.m, data: request.data) func mightContain(_ id: Data) -> Bool { - var hasher = SHA256() - hasher.update(data: id) // 16-byte PacketId - let digest = hasher.finalize() - let db = Data(digest) - var x: UInt64 = 0 - let take = min(8, db.count) - for i in 0.. String? in diff --git a/bitchat/Utils/InputValidator.swift b/bitchat/Utils/InputValidator.swift index c2323cfc..aa05c077 100644 --- a/bitchat/Utils/InputValidator.swift +++ b/bitchat/Utils/InputValidator.swift @@ -8,7 +8,9 @@ struct InputValidator { struct Limits { static let maxNicknameLength = 50 - static let maxMessageLength = 10_000 + // BinaryProtocol caps payload length at UInt16.max (65_535). Leave headroom + // for headers/padding by limiting user content to 60_000 bytes. + static let maxMessageLength = 60_000 static let maxReasonLength = 200 static let maxPeerIDLength = 64 static let hexPeerIDLength = 16 // 8 bytes = 16 hex chars diff --git a/bitchatTests/GCSFilterTests.swift b/bitchatTests/GCSFilterTests.swift new file mode 100644 index 00000000..57229bbc --- /dev/null +++ b/bitchatTests/GCSFilterTests.swift @@ -0,0 +1,22 @@ +import XCTest +@testable import bitchat + +final class GCSFilterTests: XCTestCase { + func testBuildFilterWithDuplicateIdsProducesStableEncoding() { + let id = Data(repeating: 0xAB, count: 16) + let ids = Array(repeating: id, count: 64) + + let params = GCSFilter.buildFilter(ids: ids, maxBytes: 128, targetFpr: 0.01) + XCTAssertGreaterThanOrEqual(params.m, 1) + + let decoded = GCSFilter.decodeToSortedSet(p: params.p, m: params.m, data: params.data) + XCTAssertLessThanOrEqual(decoded.count, 1) + } + + func testBucketAvoidsZeroCandidate() { + let id = Data(repeating: 0x01, count: 16) + let bucket = GCSFilter.bucket(for: id, modulus: 2) + XCTAssertNotEqual(bucket, 0) + XCTAssertLessThan(bucket, 2) + } +} diff --git a/bitchatTests/GossipSyncManagerTests.swift b/bitchatTests/GossipSyncManagerTests.swift new file mode 100644 index 00000000..e2227d32 --- /dev/null +++ b/bitchatTests/GossipSyncManagerTests.swift @@ -0,0 +1,70 @@ +import Foundation +import XCTest +@testable import bitchat + +final class GossipSyncManagerTests: XCTestCase { + func testConcurrentPacketIntakeAndSyncRequest() { + let manager = GossipSyncManager(myPeerID: "0102030405060708") + let delegate = RecordingDelegate() + let sendExpectation = expectation(description: "sync request sent") + delegate.onSend = { sendExpectation.fulfill() } + manager.delegate = delegate + + let iterations = 200 + let group = DispatchGroup() + + for i in 0.. Void)? + private(set) var lastPacket: BitchatPacket? + private let lock = NSLock() + + func sendPacket(_ packet: BitchatPacket) { + lock.lock() + lastPacket = packet + lock.unlock() + onSend?() + } + + func sendPacket(to peerID: String, packet: BitchatPacket) { + sendPacket(packet) + } + + func signPacketForBroadcast(_ packet: BitchatPacket) -> BitchatPacket { + packet + } +} diff --git a/bitchatTests/NotificationStreamAssemblerTests.swift b/bitchatTests/NotificationStreamAssemblerTests.swift new file mode 100644 index 00000000..87c34a5f --- /dev/null +++ b/bitchatTests/NotificationStreamAssemblerTests.swift @@ -0,0 +1,111 @@ +import XCTest +@testable import bitchat + +final class NotificationStreamAssemblerTests: XCTestCase { + private func makePacket(timestamp: UInt64 = 0x0102030405) -> BitchatPacket { + let sender = Data([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77]) + return BitchatPacket( + type: MessageType.message.rawValue, + senderID: sender, + recipientID: nil, + timestamp: timestamp, + payload: Data([0xDE, 0xAD, 0xBE, 0xEF]), + signature: nil, + ttl: 3 + ) + } + + func testAssemblesSingleFrameAcrossChunks() { + var assembler = NotificationStreamAssembler() + let packet = makePacket() + guard let frame = packet.toBinaryData(padding: false) else { + return XCTFail("Failed to encode packet") + } + XCTAssertNotNil(BinaryProtocol.decode(frame)) + let payloadLen = (Int(frame[12]) << 8) | Int(frame[13]) + XCTAssertEqual(payloadLen, packet.payload.count) + + let splitIndex = min(20, max(1, frame.count / 2)) + let first = frame.prefix(splitIndex) + let second = frame.suffix(from: splitIndex) + XCTAssertEqual(first.count + second.count, frame.count) + + var result = assembler.append(first) + XCTAssertTrue(result.frames.isEmpty) + XCTAssertTrue(result.droppedPrefixes.isEmpty) + XCTAssertFalse(result.reset) + + result = assembler.append(second) + XCTAssertEqual(result.frames.count, 1) + XCTAssertTrue(result.droppedPrefixes.isEmpty) + XCTAssertFalse(result.reset) + + guard let frameData = result.frames.first else { + return XCTFail("Missing frame data") + } + if frameData.count != frame.count { + XCTFail("Frame size mismatch: expected \(frame.count) got \(frameData.count)\nframe=\(Array(frame))\nassembled=\(Array(frameData))") + return + } + guard let decoded = BinaryProtocol.decode(frameData) else { + return XCTFail("Failed to decode frame") + } + XCTAssertEqual(decoded.type, packet.type) + XCTAssertEqual(decoded.payload, packet.payload) + XCTAssertEqual(decoded.senderID, packet.senderID) + XCTAssertEqual(decoded.timestamp, packet.timestamp) + + var directAssembler = NotificationStreamAssembler() + let directResult = directAssembler.append(frame) + XCTAssertEqual(directResult.frames.first?.count, frame.count) + } + + func testAssemblesMultipleFramesSequentially() { + var assembler = NotificationStreamAssembler() + let packet1 = makePacket(timestamp: 0xABC) + let packet2 = makePacket(timestamp: 0xDEF) + + guard let frame1 = packet1.toBinaryData(padding: false), + let frame2 = packet2.toBinaryData(padding: false) else { + return XCTFail("Failed to encode packets") + } + + var combined = Data() + combined.append(frame1) + combined.append(frame2) + let firstChunk = combined.prefix(20) + let secondChunk = combined.suffix(from: 20) + + var result = assembler.append(firstChunk) + XCTAssertTrue(result.frames.isEmpty) + + result = assembler.append(secondChunk) + XCTAssertEqual(result.frames.count, 2) + guard let decoded1 = BinaryProtocol.decode(result.frames[0]), + let decoded2 = BinaryProtocol.decode(result.frames[1]) else { + return XCTFail("Failed to decode frames") + } + XCTAssertEqual(decoded1.timestamp, packet1.timestamp) + XCTAssertEqual(decoded2.timestamp, packet2.timestamp) + } + + func testDropsInvalidPrefixByte() { + var assembler = NotificationStreamAssembler() + let packet = makePacket(timestamp: 0xF00) + guard let frame = packet.toBinaryData(padding: false) else { + return XCTFail("Failed to encode packet") + } + var noisyFrame = Data([0x00]) + noisyFrame.append(frame) + + let result = assembler.append(noisyFrame) + XCTAssertEqual(result.droppedPrefixes, [0x00]) + XCTAssertEqual(result.frames.count, 1) + XCTAssertFalse(result.reset) + + guard let decoded = BinaryProtocol.decode(result.frames[0]) else { + return XCTFail("Failed to decode frame after drop") + } + XCTAssertEqual(decoded.timestamp, packet.timestamp) + } +}