mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-25 02:25:20 +00:00
Fix BLE stream crashes and gossip sync races (#663)
* Handle long BLE packets safely * Keep BLE stream aligned after partial drops --------- Co-authored-by: jack <jackjackbits@users.noreply.github.com>
This commit is contained in:
@@ -7,6 +7,78 @@ import CryptoKit
|
|||||||
import UIKit
|
import UIKit
|
||||||
#endif
|
#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
|
/// BLEService — Bluetooth Mesh Transport
|
||||||
/// - Emits events exclusively via `BitchatDelegate` for UI.
|
/// - Emits events exclusively via `BitchatDelegate` for UI.
|
||||||
/// - ChatViewModel must consume delegate callbacks (`didReceivePublicMessage`, `didReceiveNoisePayload`).
|
/// - ChatViewModel must consume delegate callbacks (`didReceivePublicMessage`, `didReceiveNoisePayload`).
|
||||||
@@ -40,6 +112,7 @@ final class BLEService: NSObject {
|
|||||||
var isConnecting: Bool = false
|
var isConnecting: Bool = false
|
||||||
var isConnected: Bool = false
|
var isConnected: Bool = false
|
||||||
var lastConnectionAttempt: Date? = nil
|
var lastConnectionAttempt: Date? = nil
|
||||||
|
var assembler = NotificationStreamAssembler()
|
||||||
}
|
}
|
||||||
private var peripherals: [String: PeripheralState] = [:] // UUID -> PeripheralState
|
private var peripherals: [String: PeripheralState] = [:] // UUID -> PeripheralState
|
||||||
private var peerToPeripheralUUID: [String: String] = [:] // PeerID -> Peripheral UUID
|
private var peerToPeripheralUUID: [String: String] = [:] // PeerID -> Peripheral UUID
|
||||||
@@ -2515,7 +2588,8 @@ extension BLEService: CBCentralManagerDelegate {
|
|||||||
peerID: nil,
|
peerID: nil,
|
||||||
isConnecting: true,
|
isConnecting: true,
|
||||||
isConnected: false,
|
isConnected: false,
|
||||||
lastConnectionAttempt: Date()
|
lastConnectionAttempt: Date(),
|
||||||
|
assembler: NotificationStreamAssembler()
|
||||||
)
|
)
|
||||||
peripheral.delegate = self
|
peripheral.delegate = self
|
||||||
|
|
||||||
@@ -2564,7 +2638,9 @@ func centralManager(_ central: CBCentralManager, didConnect peripheral: CBPeriph
|
|||||||
characteristic: nil,
|
characteristic: nil,
|
||||||
peerID: nil,
|
peerID: nil,
|
||||||
isConnecting: false,
|
isConnecting: false,
|
||||||
isConnected: true
|
isConnected: true,
|
||||||
|
lastConnectionAttempt: nil,
|
||||||
|
assembler: NotificationStreamAssembler()
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2702,7 +2778,8 @@ extension BLEService {
|
|||||||
peerID: nil,
|
peerID: nil,
|
||||||
isConnecting: true,
|
isConnecting: true,
|
||||||
isConnected: false,
|
isConnected: false,
|
||||||
lastConnectionAttempt: Date()
|
lastConnectionAttempt: Date(),
|
||||||
|
assembler: NotificationStreamAssembler()
|
||||||
)
|
)
|
||||||
peripheral.delegate = self
|
peripheral.delegate = self
|
||||||
let options: [String: Any] = [
|
let options: [String: Any] = [
|
||||||
@@ -2839,53 +2916,72 @@ extension BLEService: CBPeripheralDelegate {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
guard let data = characteristic.value else {
|
guard let data = characteristic.value, !data.isEmpty else {
|
||||||
SecureLogger.warning("⚠️ No data in notification", category: .session)
|
SecureLogger.warning("⚠️ No data in notification", category: .session)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Received BLE notification
|
bufferNotificationChunk(data, from: peripheral)
|
||||||
|
|
||||||
// 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)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private func bufferNotificationChunk(_ chunk: Data, from peripheral: CBPeripheral) {
|
||||||
let peripheralUUID = peripheral.identifier.uuidString
|
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 {
|
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 packet.ttl == messageTTL {
|
||||||
if var state = peripherals[peripheralUUID] {
|
if var state = peripherals[peripheralUUID] {
|
||||||
state.peerID = senderID
|
state.peerID = senderID
|
||||||
peripherals[peripheralUUID] = state
|
peripherals[peripheralUUID] = state
|
||||||
}
|
}
|
||||||
peerToPeripheralUUID[senderID] = peripheralUUID
|
peerToPeripheralUUID[senderID] = peripheralUUID
|
||||||
// Mapping update - direct announce from peer
|
|
||||||
}
|
}
|
||||||
// Record ingress link for last-hop suppression and process
|
|
||||||
let msgID = makeMessageID(for: packet)
|
let msgID = makeMessageID(for: packet)
|
||||||
collectionsQueue.async(flags: .barrier) { [weak self] in
|
collectionsQueue.async(flags: .barrier) { [weak self] in
|
||||||
self?.ingressByMessageID[msgID] = (.peripheral(peripheralUUID), Date())
|
self?.ingressByMessageID[msgID] = (.peripheral(peripheralUUID), Date())
|
||||||
}
|
}
|
||||||
// Process the announce packet regardless of whether we updated the mapping
|
|
||||||
handleReceivedPacket(packet, from: senderID)
|
handleReceivedPacket(packet, from: senderID)
|
||||||
} else {
|
} 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)
|
let msgID = makeMessageID(for: packet)
|
||||||
collectionsQueue.async(flags: .barrier) { [weak self] in
|
collectionsQueue.async(flags: .barrier) { [weak self] in
|
||||||
self?.ingressByMessageID[msgID] = (.peripheral(peripheralUUID), Date())
|
self?.ingressByMessageID[msgID] = (.peripheral(peripheralUUID), Date())
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ import CryptoKit
|
|||||||
// Golomb-Coded Set (GCS) filter utilities for sync.
|
// Golomb-Coded Set (GCS) filter utilities for sync.
|
||||||
// Hashing:
|
// Hashing:
|
||||||
// - Packet ID is 16 bytes (see PacketIdUtil). For GCS mapping, use h64 = first 8 bytes of SHA-256 over the 16-byte ID.
|
// - 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):
|
// Encoding (v1):
|
||||||
// - Sort mapped values ascending; encode deltas (first is v0, then vi - v{i-1}) as positive integers x >= 1.
|
// - 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<<P)-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<<P)-1).
|
||||||
@@ -29,25 +29,40 @@ enum GCSFilter {
|
|||||||
|
|
||||||
static func buildFilter(ids: [Data], maxBytes: Int, targetFpr: Double) -> Params {
|
static func buildFilter(ids: [Data], maxBytes: Int, targetFpr: Double) -> Params {
|
||||||
let p = deriveP(targetFpr: targetFpr)
|
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 cap = estimateMaxElements(sizeBytes: maxBytes, p: p)
|
||||||
let n = min(ids.count, cap)
|
let selected = Array(ids.prefix(cap))
|
||||||
let selected = Array(ids.prefix(n))
|
let range = max(1, hashRange(count: selected.count, p: p))
|
||||||
// Map to [0, M)
|
let modulo = UInt64(range)
|
||||||
let mInit = UInt32(n << p)
|
|
||||||
var mapped = selected.map { id16 -> UInt64 in
|
var mapped = selected
|
||||||
let h = h64(id16)
|
.map { h64($0) }
|
||||||
return UInt64(h % UInt64(max(1, mInit)))
|
.map { mapHash($0, modulo: modulo) }
|
||||||
}.sorted()
|
.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 encoded = encode(sorted: mapped, p: p)
|
||||||
var trimmedN = n
|
var trimmedCount = mapped.count
|
||||||
// Trim if over budget
|
|
||||||
while encoded.count > maxBytes && trimmedN > 0 {
|
while encoded.count > maxBytes && trimmedCount > 0 {
|
||||||
trimmedN = (trimmedN * 9) / 10 // drop ~10%
|
if trimmedCount == 1 {
|
||||||
mapped = Array(mapped.prefix(trimmedN))
|
mapped.removeAll()
|
||||||
|
encoded = Data()
|
||||||
|
break
|
||||||
|
}
|
||||||
|
trimmedCount = max(1, (trimmedCount * 9) / 10)
|
||||||
|
mapped = Array(mapped.prefix(trimmedCount))
|
||||||
encoded = encode(sorted: mapped, p: p)
|
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] {
|
static func decodeToSortedSet(p: Int, m: UInt32, data: Data) -> [UInt64] {
|
||||||
@@ -77,6 +92,12 @@ enum GCSFilter {
|
|||||||
return false
|
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 {
|
private static func h64(_ id16: Data) -> UInt64 {
|
||||||
var hasher = SHA256()
|
var hasher = SHA256()
|
||||||
hasher.update(data: id16)
|
hasher.update(data: id16)
|
||||||
@@ -88,6 +109,39 @@ enum GCSFilter {
|
|||||||
return x & 0x7fff_ffff_ffff_ffff
|
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 {
|
private static func encode(sorted: [UInt64], p: Int) -> Data {
|
||||||
let writer = BitWriter()
|
let writer = BitWriter()
|
||||||
var prev: UInt64 = 0
|
var prev: UInt64 = 0
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
import Foundation
|
import Foundation
|
||||||
import CryptoKit
|
|
||||||
|
|
||||||
// Gossip-based sync manager using on-demand GCS filters
|
// Gossip-based sync manager using on-demand GCS filters
|
||||||
final class GossipSyncManager {
|
final class GossipSyncManager {
|
||||||
@@ -53,6 +52,12 @@ final class GossipSyncManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func onPublicPacketSeen(_ packet: BitchatPacket) {
|
func onPublicPacketSeen(_ packet: BitchatPacket) {
|
||||||
|
queue.async { [weak self] in
|
||||||
|
self?._onPublicPacketSeen(packet)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private func _onPublicPacketSeen(_ packet: BitchatPacket) {
|
||||||
let mt = MessageType(rawValue: packet.type)
|
let mt = MessageType(rawValue: packet.type)
|
||||||
let isBroadcastRecipient: Bool = {
|
let isBroadcastRecipient: Bool = {
|
||||||
guard let r = packet.recipientID else { return true }
|
guard let r = packet.recipientID else { return true }
|
||||||
@@ -119,18 +124,17 @@ final class GossipSyncManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func handleRequestSync(fromPeerID: String, request: RequestSyncPacket) {
|
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
|
// Decode GCS into sorted set and prepare membership checker
|
||||||
let sorted = GCSFilter.decodeToSortedSet(p: request.p, m: request.m, data: request.data)
|
let sorted = GCSFilter.decodeToSortedSet(p: request.p, m: request.m, data: request.data)
|
||||||
func mightContain(_ id: Data) -> Bool {
|
func mightContain(_ id: Data) -> Bool {
|
||||||
var hasher = SHA256()
|
let bucket = GCSFilter.bucket(for: id, modulus: request.m)
|
||||||
hasher.update(data: id) // 16-byte PacketId
|
return GCSFilter.contains(sortedValues: sorted, candidate: bucket)
|
||||||
let digest = hasher.finalize()
|
|
||||||
let db = Data(digest)
|
|
||||||
var x: UInt64 = 0
|
|
||||||
let take = min(8, db.count)
|
|
||||||
for i in 0..<take { x = (x << 8) | UInt64(db[i]) }
|
|
||||||
let v = (x & 0x7fff_ffff_ffff_ffff) % UInt64(request.m)
|
|
||||||
return GCSFilter.contains(sortedValues: sorted, candidate: v)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 1) Announcements: send latest per peer if requester lacks them
|
// 1) Announcements: send latest per peer if requester lacks them
|
||||||
@@ -182,9 +186,15 @@ final class GossipSyncManager {
|
|||||||
|
|
||||||
// Explicit removal hook for LEAVE/stale peer
|
// Explicit removal hook for LEAVE/stale peer
|
||||||
func removeAnnouncementForPeer(_ peerID: String) {
|
func removeAnnouncementForPeer(_ peerID: String) {
|
||||||
|
queue.async { [weak self] in
|
||||||
|
self?._removeAnnouncementForPeer(peerID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private func _removeAnnouncementForPeer(_ peerID: String) {
|
||||||
let normalizedPeerID = peerID.lowercased()
|
let normalizedPeerID = peerID.lowercased()
|
||||||
_ = latestAnnouncementByPeer.removeValue(forKey: normalizedPeerID)
|
_ = latestAnnouncementByPeer.removeValue(forKey: normalizedPeerID)
|
||||||
|
|
||||||
// Remove messages from this peer
|
// Remove messages from this peer
|
||||||
// Collect IDs to remove first to avoid concurrent modification
|
// Collect IDs to remove first to avoid concurrent modification
|
||||||
let messageIdsToRemove = messages.compactMap { (id, message) -> String? in
|
let messageIdsToRemove = messages.compactMap { (id, message) -> String? in
|
||||||
|
|||||||
@@ -8,7 +8,9 @@ struct InputValidator {
|
|||||||
|
|
||||||
struct Limits {
|
struct Limits {
|
||||||
static let maxNicknameLength = 50
|
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 maxReasonLength = 200
|
||||||
static let maxPeerIDLength = 64
|
static let maxPeerIDLength = 64
|
||||||
static let hexPeerIDLength = 16 // 8 bytes = 16 hex chars
|
static let hexPeerIDLength = 16 // 8 bytes = 16 hex chars
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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..<iterations {
|
||||||
|
group.enter()
|
||||||
|
DispatchQueue.global(qos: .userInitiated).async {
|
||||||
|
let packet = BitchatPacket(
|
||||||
|
type: MessageType.message.rawValue,
|
||||||
|
senderID: Data(hexString: "1122334455667788") ?? Data(),
|
||||||
|
recipientID: nil,
|
||||||
|
timestamp: 1_000_000 + UInt64(i),
|
||||||
|
payload: Data([UInt8(truncatingIfNeeded: i)]),
|
||||||
|
signature: nil,
|
||||||
|
ttl: 1
|
||||||
|
)
|
||||||
|
manager.onPublicPacketSeen(packet)
|
||||||
|
Thread.sleep(forTimeInterval: 0.001)
|
||||||
|
group.leave()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
DispatchQueue.global(qos: .userInitiated).asyncAfter(deadline: .now() + 0.002) {
|
||||||
|
manager.scheduleInitialSyncToPeer("FFFFFFFFFFFFFFFF", delaySeconds: 0.0)
|
||||||
|
}
|
||||||
|
|
||||||
|
group.wait()
|
||||||
|
wait(for: [sendExpectation], timeout: 2.0)
|
||||||
|
|
||||||
|
guard let lastPacket = delegate.lastPacket else {
|
||||||
|
XCTFail("Expected sync packet to be sent")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
XCTAssertEqual(lastPacket.type, MessageType.requestSync.rawValue)
|
||||||
|
XCTAssertNotNil(RequestSyncPacket.decode(from: lastPacket.payload))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private final class RecordingDelegate: GossipSyncManager.Delegate {
|
||||||
|
var onSend: (() -> 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
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user