From 2db48a34c0cae3968ddd3e037f3ac27057d66d50 Mon Sep 17 00:00:00 2001 From: callebtc <93376500+callebtc@users.noreply.github.com> Date: Thu, 11 Sep 2025 22:35:41 +0200 Subject: [PATCH] wip --- bitchat/Models/RequestSyncPacket.swift | 63 ++++++++ bitchat/Protocols/BitchatProtocol.swift | 2 + bitchat/Services/BLEService.swift | 66 ++++++++- bitchat/Sync/GossipSyncManager.swift | 168 ++++++++++++++++++++++ bitchat/Sync/PacketIdUtil.swift | 22 +++ bitchat/Sync/SeenPacketsBloomFilter.swift | 116 +++++++++++++++ 6 files changed, 436 insertions(+), 1 deletion(-) create mode 100644 bitchat/Models/RequestSyncPacket.swift create mode 100644 bitchat/Sync/GossipSyncManager.swift create mode 100644 bitchat/Sync/PacketIdUtil.swift create mode 100644 bitchat/Sync/SeenPacketsBloomFilter.swift diff --git a/bitchat/Models/RequestSyncPacket.swift b/bitchat/Models/RequestSyncPacket.swift new file mode 100644 index 00000000..bdc62c81 --- /dev/null +++ b/bitchat/Models/RequestSyncPacket.swift @@ -0,0 +1,63 @@ +import Foundation + +// REQUEST_SYNC payload TLV (type, length16, value) +// - 0x01: mBytes (uint16) +// - 0x02: k (uint8) +// - 0x03: bloom filter bits (opaque byte array of length mBytes) +struct RequestSyncPacket { + let mBytes: Int + let k: Int + let bits: Data + + func encode() -> Data { + var out = Data() + func putTLV(_ t: UInt8, _ v: Data) { + out.append(t) + let len = UInt16(v.count) + out.append(UInt8((len >> 8) & 0xFF)) + out.append(UInt8(len & 0xFF)) + out.append(v) + } + // mBytes + var mb = UInt16(mBytes) + let mbData = withUnsafeBytes(of: &mb.bigEndian) { Data($0) } + putTLV(0x01, mbData) + // k + putTLV(0x02, Data([UInt8(k & 0xFF)])) + // bits + putTLV(0x03, bits) + return out + } + + static func decode(from data: Data) -> RequestSyncPacket? { + var off = 0 + var mBytes: Int? = nil + var k: Int? = nil + var bits: Data? = nil + + while off + 3 <= data.count { + let t = Int(data[off]); off += 1 + guard off + 2 <= data.count else { return nil } + let len = (Int(data[off]) << 8) | Int(data[off+1]); off += 2 + guard off + len <= data.count else { return nil } + let v = data.subdata(in: off..<(off+len)); off += len + switch t { + case 0x01: + if v.count == 2 { + let mb = (Int(v[0]) << 8) | Int(v[1]) + mBytes = mb + } + case 0x02: + if v.count == 1 { k = Int(v[0]) } + case 0x03: + bits = v + default: + break // forward compatible; ignore unknown TLVs + } + } + + guard let mb = mBytes, let kk = k, let bb = bits, mb == bb.count else { return nil } + return RequestSyncPacket(mBytes: mb, k: kk, bits: bb) + } +} + diff --git a/bitchat/Protocols/BitchatProtocol.swift b/bitchat/Protocols/BitchatProtocol.swift index 0bc55536..c5ef654a 100644 --- a/bitchat/Protocols/BitchatProtocol.swift +++ b/bitchat/Protocols/BitchatProtocol.swift @@ -125,6 +125,7 @@ enum MessageType: UInt8 { case announce = 0x01 // "I'm here" with nickname case message = 0x02 // Public chat message case leave = 0x03 // "I'm leaving" + case requestSync = 0x21 // Bloom filter-based sync request (local-only) // Noise encryption case noiseHandshake = 0x10 // Handshake (init or response determined by payload) @@ -138,6 +139,7 @@ enum MessageType: UInt8 { case .announce: return "announce" case .message: return "message" case .leave: return "leave" + case .requestSync: return "requestSync" case .noiseHandshake: return "noiseHandshake" case .noiseEncrypted: return "noiseEncrypted" case .fragment: return "fragment" diff --git a/bitchat/Services/BLEService.swift b/bitchat/Services/BLEService.swift index 6d44a94c..04f128b7 100644 --- a/bitchat/Services/BLEService.swift +++ b/bitchat/Services/BLEService.swift @@ -137,6 +137,9 @@ final class BLEService: NSObject { private var pendingDirectedRelays: [String: [String: (packet: BitchatPacket, enqueuedAt: Date)]] = [:] // Debounce for 'reconnected' logs private var lastReconnectLogAt: [String: Date] = [:] + + // MARK: - Gossip Sync + private var gossipSyncManager: GossipSyncManager? // MARK: - Maintenance Timer @@ -395,9 +398,15 @@ final class BLEService: NSObject { } timer.resume() maintenanceTimer = timer - + // Publish initial empty state requestPeerDataPublish() + + // Initialize gossip sync manager + let sync = GossipSyncManager(myPeerID: myPeerID) + sync.delegate = self + sync.start() + self.gossipSyncManager = sync } func setNickname(_ nickname: String) { @@ -772,6 +781,8 @@ final class BLEService: NSObject { self.messageDeduplicator.markProcessed(dedupID) // Call synchronously since we're already on background queue self.broadcastPacket(signedPacket) + // Track our own broadcast for sync + self.gossipSyncManager?.onPublicPacketSeen(signedPacket) } } } @@ -1130,6 +1141,12 @@ final class BLEService: NSObject { } } + // Directed send helper (unicast to a specific peerID) without altering packet contents + private func sendPacketDirected(_ packet: BitchatPacket, to peerID: String) { + guard let data = packet.toBinaryData(padding: false) else { return } + sendOnAllLinks(packet: packet, data: data, pad: false, directedOnlyPeer: peerID) + } + // MARK: - Directed store-and-forward private func spoolDirectedPacket(_ packet: BitchatPacket, recipientPeerID: String) { let msgID = makeMessageID(for: packet) @@ -1380,6 +1397,9 @@ final class BLEService: NSObject { case .message: handleMessage(packet, from: senderID) + case .requestSync: + handleRequestSync(packet, from: senderID) + case .noiseHandshake: handleNoiseHandshake(packet, from: senderID) @@ -1580,12 +1600,17 @@ final class BLEService: NSObject { // Only notify of connection for new or reconnected peers when it is a direct announce if (packet.ttl == self.messageTTL) && (isNewPeer || isReconnectedPeer) { self.delegate?.didConnectToPeer(peerID) + // Schedule initial unicast sync to this peer + self.gossipSyncManager?.scheduleInitialSyncToPeer(peerID, delaySeconds: 5.0) } self.requestPeerDataPublish() self.delegate?.didUpdatePeerList(currentPeerIDs) } + // Track for sync (include our own and others' announces) + gossipSyncManager?.onPublicPacketSeen(packet) + // Send announce back for bidirectional discovery (only once per peer) let announceBackID = "announce-back-\(peerID)" let shouldSendBack = !messageDeduplicator.contains(announceBackID) @@ -1607,6 +1632,15 @@ final class BLEService: NSObject { } } } + + // Handle REQUEST_SYNC: decode payload and respond with missing packets via sync manager + private func handleRequestSync(_ packet: BitchatPacket, from peerID: String) { + guard let req = RequestSyncPacket.decode(from: packet.payload) else { + SecureLogger.warning("⚠️ Malformed REQUEST_SYNC from \(peerID)", category: .session) + return + } + gossipSyncManager?.handleRequestSync(fromPeerID: peerID, request: req) + } // Mention parsing moved to ChatViewModel @@ -1647,6 +1681,11 @@ final class BLEService: NSObject { } } + // Track broadcast messages (recipientID == nil indicates broadcast) + if packet.recipientID == nil && packet.type == MessageType.message.rawValue { + gossipSyncManager?.onPublicPacketSeen(packet) + } + guard accepted else { SecureLogger.warning("🚫 Dropping public message from unverified or unknown peer \(peerID.prefix(8))…", category: .security) return @@ -1858,6 +1897,8 @@ final class BLEService: NSObject { self?.broadcastPacket(signedPacket) } } + // Ensure our own announce is included in sync state + gossipSyncManager?.onPublicPacketSeen(signedPacket) } func sendDeliveryAck(for messageID: String, to peerID: String) { @@ -2246,6 +2287,29 @@ final class BLEService: NSObject { } } +// MARK: - GossipSyncManager Delegate +extension BLEService: GossipSyncManager.Delegate { + func sendPacket(_ packet: BitchatPacket) { + if DispatchQueue.getSpecific(key: messageQueueKey) != nil { + broadcastPacket(packet) + } else { + messageQueue.async { [weak self] in self?.broadcastPacket(packet) } + } + } + + func sendPacket(to peerID: String, packet: BitchatPacket) { + if DispatchQueue.getSpecific(key: messageQueueKey) != nil { + sendPacketDirected(packet, to: peerID) + } else { + messageQueue.async { [weak self] in self?.sendPacketDirected(packet, to: peerID) } + } + } + + func signPacketForBroadcast(_ packet: BitchatPacket) -> BitchatPacket { + return noiseService.signPacket(packet) ?? packet + } +} + // MARK: - CBCentralManagerDelegate extension BLEService: CBCentralManagerDelegate { diff --git a/bitchat/Sync/GossipSyncManager.swift b/bitchat/Sync/GossipSyncManager.swift new file mode 100644 index 00000000..cd04d1a8 --- /dev/null +++ b/bitchat/Sync/GossipSyncManager.swift @@ -0,0 +1,168 @@ +import Foundation + +// Gossip-based sync manager using rotating Bloom filters +final class GossipSyncManager { + protocol Delegate: AnyObject { + func sendPacket(_ packet: BitchatPacket) + func sendPacket(to peerID: String, packet: BitchatPacket) + func signPacketForBroadcast(_ packet: BitchatPacket) -> BitchatPacket + } + + struct Config { + var seenCapacity: Int = 100 // recent broadcast messages kept + var bloomMaxBytes: Int = 256 // up to 256 bytes + var bloomTargetFpr: Double = 0.01 // 1% + } + + private let myPeerID: String + private let config: Config + weak var delegate: Delegate? + + // Bloom filter + private let bloom: SeenPacketsBloomFilter + + // Storage: broadcast messages (ordered by insert), and latest announce per sender + private var messages: [String: BitchatPacket] = [:] // idHex -> packet + private var messageOrder: [String] = [] + private var latestAnnouncementByPeer: [String: (id: String, packet: BitchatPacket)] = [:] + + // Timer + private var periodicTimer: DispatchSourceTimer? + private let queue = DispatchQueue(label: "mesh.sync", qos: .utility) + + init(myPeerID: String, config: Config = Config()) { + self.myPeerID = myPeerID + self.config = config + self.bloom = SeenPacketsBloomFilter(maxBytes: config.bloomMaxBytes, targetFpr: config.bloomTargetFpr) + } + + func start() { + stop() + let timer = DispatchSource.makeTimerSource(queue: queue) + timer.schedule(deadline: .now() + 30.0, repeating: 30.0, leeway: .seconds(1)) + timer.setEventHandler { [weak self] in self?.sendRequestSync() } + timer.resume() + periodicTimer = timer + } + + func stop() { + periodicTimer?.cancel(); periodicTimer = nil + } + + func scheduleInitialSyncToPeer(_ peerID: String, delaySeconds: TimeInterval = 5.0) { + queue.asyncAfter(deadline: .now() + delaySeconds) { [weak self] in + self?.sendRequestSync(to: peerID) + } + } + + func onPublicPacketSeen(_ packet: BitchatPacket) { + let mt = MessageType(rawValue: packet.type) + let isBroadcastMessage = (mt == .message && packet.recipientID == nil) + let isAnnounce = (mt == .announce) + guard isBroadcastMessage || isAnnounce else { return } + + let idBytes = PacketIdUtil.computeId(packet) + bloom.add(idBytes) + let idHex = idBytes.hexEncodedString() + + if isBroadcastMessage { + if messages[idHex] == nil { + messages[idHex] = packet + messageOrder.append(idHex) + // Enforce capacity + let cap = max(1, config.seenCapacity) + while messageOrder.count > cap { + let victim = messageOrder.removeFirst() + messages.removeValue(forKey: victim) + } + } + } else if isAnnounce { + let sender = packet.senderID.hexEncodedString() + latestAnnouncementByPeer[sender] = (id: idHex, packet: packet) + } + } + + private func sendRequestSync() { + let snap = bloom.snapshotActive() + let payload = RequestSyncPacket(mBytes: snap.mBytes, k: snap.k, bits: snap.bits).encode() + let pkt = BitchatPacket( + type: MessageType.requestSync.rawValue, + senderID: Data(hexString: myPeerID) ?? Data(), + recipientID: nil, // broadcast + timestamp: UInt64(Date().timeIntervalSince1970 * 1000), + payload: payload, + signature: nil, + ttl: 0 // local-only + ) + let signed = delegate?.signPacketForBroadcast(pkt) ?? pkt + delegate?.sendPacket(signed) + } + + private func sendRequestSync(to peerID: String) { + let snap = bloom.snapshotActive() + let payload = RequestSyncPacket(mBytes: snap.mBytes, k: snap.k, bits: snap.bits).encode() + var recipient = Data() + var temp = peerID + while temp.count >= 2 && recipient.count < 8 { + let hexByte = String(temp.prefix(2)) + if let b = UInt8(hexByte, radix: 16) { recipient.append(b) } + temp = String(temp.dropFirst(2)) + } + let pkt = BitchatPacket( + type: MessageType.requestSync.rawValue, + senderID: Data(hexString: myPeerID) ?? Data(), + recipientID: recipient, + timestamp: UInt64(Date().timeIntervalSince1970 * 1000), + payload: payload, + signature: nil, + ttl: 0 // local-only + ) + let signed = delegate?.signPacketForBroadcast(pkt) ?? pkt + delegate?.sendPacket(to: peerID, packet: signed) + } + + func handleRequestSync(fromPeerID: String, request: RequestSyncPacket) { + // Build membership checker from provided parameters + let mBits = request.mBytes * 8 + let k = request.k + func mightContain(_ id: Data) -> Bool { + // Same hashing as local bloom; compute indices, check MSB-first bits in request.bits + var h1: UInt64 = 1469598103934665603 + var h2: UInt64 = 0x27d4eb2f165667c5 + for b in id { h1 = (h1 ^ UInt64(b)) &* 1099511628211; h2 = (h2 ^ UInt64(b)) &* 0x100000001B3 } + for i in 0..> (7 - bitIndex)) & 1) == 1 + if !bit { return false } + } + return true + } + + // 1) Announcements: send latest per peer if requester lacks them + for (_, pair) in latestAnnouncementByPeer { + let (idHex, pkt) = pair + let idBytes = Data(hexString: idHex) ?? Data() + if !mightContain(idBytes) { + var toSend = pkt + toSend.ttl = 0 + delegate?.sendPacket(to: fromPeerID, packet: toSend) + } + } + + // 2) Broadcast messages: send all missing + let toSendMsgs = messageOrder.compactMap { messages[$0] } + for pkt in toSendMsgs { + let idBytes = PacketIdUtil.computeId(pkt) + if !mightContain(idBytes) { + var toSend = pkt + toSend.ttl = 0 + delegate?.sendPacket(to: fromPeerID, packet: toSend) + } + } + } +} + diff --git a/bitchat/Sync/PacketIdUtil.swift b/bitchat/Sync/PacketIdUtil.swift new file mode 100644 index 00000000..3688dfbc --- /dev/null +++ b/bitchat/Sync/PacketIdUtil.swift @@ -0,0 +1,22 @@ +import Foundation +import CryptoKit + +// Deterministic packet ID used for sync Bloom membership +// ID = first 16 bytes of SHA-256 over: [type | senderID | timestamp | payload] +enum PacketIdUtil { + static func computeId(_ packet: BitchatPacket) -> Data { + var hasher = SHA256() + hasher.update(data: Data([packet.type])) + hasher.update(data: packet.senderID) + var tsBE = packet.timestamp.bigEndian + withUnsafeBytes(of: &tsBE) { raw in hasher.update(data: Data(raw)) } + hasher.update(data: packet.payload) + let digest = hasher.finalize() + return Data(digest.prefix(16)) + } + + static func computeIdHex(_ packet: BitchatPacket) -> String { + return computeId(packet).hexEncodedString() + } +} + diff --git a/bitchat/Sync/SeenPacketsBloomFilter.swift b/bitchat/Sync/SeenPacketsBloomFilter.swift new file mode 100644 index 00000000..0a474ced --- /dev/null +++ b/bitchat/Sync/SeenPacketsBloomFilter.swift @@ -0,0 +1,116 @@ +import Foundation + +// Rotating Bloom filter for recently seen packet IDs (16-byte IDs) +final class SeenPacketsBloomFilter { + struct Snapshot { let mBytes: Int; let k: Int; let bits: Data } + + private struct Filter { var mBits: Int; var k: Int; var bits: [UInt8]; var count: Int } + + private let maxBytes: Int + private let targetFpr: Double + private let seed: UInt64 = 0x27d4eb2f165667c5 + + private let mBits: Int + private let kOptimal: Int + private let capacityOptimal: Int + + private var active: Filter + private var standby: Filter + private var usingStandby: Bool = false + private let lock = NSLock() + + init(maxBytes: Int = 256, targetFpr: Double = 0.01) { + self.maxBytes = max(1, maxBytes) + self.targetFpr = targetFpr + self.mBits = max(8, self.maxBytes * 8) + let (k, n) = SeenPacketsBloomFilter.deriveParams(mBits: self.mBits, fpr: targetFpr) + self.kOptimal = max(1, k) + self.capacityOptimal = max(1, n) + self.active = Filter(mBits: self.mBits, k: self.kOptimal, bits: [UInt8](repeating: 0, count: self.maxBytes), count: 0) + self.standby = Filter(mBits: self.mBits, k: self.kOptimal, bits: [UInt8](repeating: 0, count: self.maxBytes), count: 0) + } + + private static func deriveParams(mBits: Int, fpr: Double) -> (Int, Int) { + // n ≈ -(m (ln 2)^2) / ln p ; k ≈ (m/n) ln 2 + let ln2 = log(2.0) + let n = max(1, Int(Double(-mBits) * ln2 * ln2 / log(fpr))) + let k = max(1, Int(ceil((Double(mBits) / Double(n)) * ln2))) + return (k, n) + } + + private func indicesFor(id: Data, mBits: Int, k: Int) -> [Int] { + var h1: UInt64 = 1469598103934665603 // FNV-1a 64-bit offset + var h2: UInt64 = seed + for b in id { // treat as unsigned bytes + h1 = (h1 ^ UInt64(b)) &* 1099511628211 + h2 = (h2 ^ UInt64(b)) &* 0x100000001B3 + } + var result = [Int]() + result.reserveCapacity(k) + for i in 0..= startStandbyAt { + standby = Filter(mBits: mBits, k: kOptimal, bits: [UInt8](repeating: 0, count: maxBytes), count: 0) + usingStandby = true + } + insert(into: &active, id: id) + if usingStandby { insert(into: &standby, id: id) } + if active.count >= capacityOptimal { + active = standby + standby = Filter(mBits: mBits, k: kOptimal, bits: [UInt8](repeating: 0, count: maxBytes), count: 0) + usingStandby = false + } + } + + private func insert(into filter: inout Filter, id: Data) { + let idxs = indicesFor(id: id, mBits: filter.mBits, k: filter.k) + for i in idxs { + let byteIndex = i / 8 + let bitIndex = i % 8 + filter.bits[byteIndex] = UInt8(Int(filter.bits[byteIndex]) | (1 << (7 - bitIndex))) + } + filter.count &+= 1 + } + + func mightContain(_ id: Data) -> Bool { + lock.lock(); defer { lock.unlock() } + let a = active + let idx = indicesFor(id: id, mBits: a.mBits, k: a.k) + var inActive = true + for i in idx { + let byteIndex = i / 8 + let bitIndex = i % 8 + let set = ((Int(a.bits[byteIndex]) >> (7 - bitIndex)) & 1) == 1 + if !set { inActive = false; break } + } + if inActive { return true } + if usingStandby { + let s = standby + let idx2 = indicesFor(id: id, mBits: s.mBits, k: s.k) + for i in idx2 { + let byteIndex = i / 8 + let bitIndex = i % 8 + let set = ((Int(s.bits[byteIndex]) >> (7 - bitIndex)) & 1) == 1 + if !set { return false } + } + return true + } + return false + } + + func snapshotActive() -> Snapshot { + lock.lock(); defer { lock.unlock() } + let a = active + return Snapshot(mBytes: a.bits.count, k: a.k, bits: Data(a.bits)) + } +} +